import subprocess import pwd from click.testing import CliRunner from datetime import datetime, timedelta import pytest # pkill py && rm -Rfv ~/.certidude && TRAVIS=1 py.test tests runner = CliRunner() @pytest.fixture(scope='module') def client(): from certidude.api import certidude_app from falcon import testing app = certidude_app() return testing.TestClient(app) def generate_csr(cn=None): from cryptography import x509 from cryptography.hazmat.primitives.asymmetric import rsa, padding from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.backends import default_backend from cryptography.x509.oid import NameOID key = rsa.generate_private_key( public_exponent=65537, key_size=1024, backend=default_backend()) csr = x509.CertificateSigningRequestBuilder() if cn is not None: csr = csr.subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, cn)])) buf = csr.sign(key, hashes.SHA256(), default_backend() ).public_bytes(serialization.Encoding.PEM) return buf def test_cli_setup_authority(): import shutil import os if os.path.exists("/run/certidude/signer.pid"): with open("/run/certidude/signer.pid") as fh: try: os.kill(int(fh.read()), 15) except OSError: pass os.unlink("/run/certidude/signer.pid") if os.path.exists("/run/certidude/server.pid"): with open("/run/certidude/server.pid") as fh: try: os.kill(int(fh.read()), 15) except OSError: pass os.unlink("/run/certidude/server.pid") if os.path.exists("/var/lib/certidude/ca.example.lan"): shutil.rmtree("/var/lib/certidude/ca.example.lan") if os.path.exists("/etc/certidude/server.conf"): os.unlink("/etc/certidude/server.conf") if os.path.exists("/etc/certidude/client.conf"): os.unlink("/etc/certidude/client.conf") # Remove OpenVPN stuff if os.path.exists("/etc/openvpn"): for filename in os.listdir("/etc/openvpn"): if filename.endswith(".conf"): os.unlink(os.path.join("/etc/openvpn", filename)) if os.path.exists("/etc/openvpn/keys"): shutil.rmtree("/etc/openvpn/keys") from certidude.cli import entry_point as cli from certidude import const result = runner.invoke(cli, ['setup', 'authority']) assert not result.exception, result.output from certidude import config, authority assert authority.ca_cert.serial_number >= 0x100000000000000000000000000000000000000 assert authority.ca_cert.serial_number <= 0xfffffffffffffffffffffffffffffffffffffff assert authority.ca_cert.not_valid_before < datetime.now() assert authority.ca_cert.not_valid_after > datetime.now() + timedelta(days=7000) # Start server before any signing operations are performed result = runner.invoke(cli, ['serve', '-f']) assert not result.exception, result.output import requests # Password is bot, users created by Travis usertoken = "Basic dXNlcmJvdDpib3Q=" admintoken = "Basic YWRtaW5ib3Q6Ym90" result = runner.invoke(cli, ['users']) assert not result.exception, result.output # Check that we can retrieve empty CRL r = client().simulate_get("/api/revoked/") assert r.status_code == 200 # Test command line interface result = runner.invoke(cli, ['list', '-srv']) assert not result.exception, result.output # Test CA certificate fetch r = client().simulate_get("/api/certificate") assert r.status_code == 200 assert r.headers.get('content-type') == "application/x-x509-ca-cert" r = requests.get("http://ca.example.lan/api/certificate") assert r.status_code == 200 assert r.headers.get('content-type') == "application/x-x509-ca-cert" # Test request submission buf = generate_csr(cn=u"test") r = client().simulate_post("/api/request/", body=buf) assert r.status_code == 415 # wrong content type r = client().simulate_post("/api/request/", body=buf, headers={"content-type":"application/pkcs10"}) assert r.status_code == 202 # success r = client().simulate_post("/api/request/", body=buf, headers={"content-type":"application/pkcs10"}) assert r.status_code == 202 # already exists, same keypair so it's ok r = client().simulate_post("/api/request/", query_string="wait=1", body=buf, headers={"content-type":"application/pkcs10"}) assert r.status_code == 303 # redirect to long poll r = client().simulate_post("/api/request/", body=generate_csr(cn=u"test"), headers={"content-type":"application/pkcs10"}) assert r.status_code == 409 # duplicate cn, different keypair r = client().simulate_get("/api/request/test/", headers={"Accept":"application/json"}) assert r.status_code == 200 # fetch as JSON ok assert r.headers.get('content-type') == "application/json" r = client().simulate_get("/api/request/test/", headers={"Accept":"application/x-pem-file"}) assert r.status_code == 200 # fetch as PEM ok assert r.headers.get('content-type') == "application/x-pem-file" r = client().simulate_get("/api/request/test/", headers={"Accept":"text/plain"}) assert r.status_code == 415 # not available as plaintext r = client().simulate_get("/api/request/nonexistant/", headers={"Accept":"application/json"}) assert r.status_code == 404 # nonexistant common names r = client().simulate_post("/api/request/", query_string="autosign=1", body=buf, headers={"content-type":"application/pkcs10"}) assert r.status_code == 200 # autosign successful assert r.headers.get('content-type') == "application/x-pem-file" # TODO: submit messed up CSR-s: no CN, empty CN etc # Test command line interface result = runner.invoke(cli, ['list', '-srv']) assert not result.exception, result.output result = runner.invoke(cli, ['sign', 'test', '-o']) assert not result.exception, result.output result = runner.invoke(cli, ['revoke', 'test']) assert not result.exception, result.output authority.generate_ovpn_bundle(u"test2") authority.generate_pkcs12_bundle(u"test3") result = runner.invoke(cli, ['list', '-srv']) assert not result.exception, result.output result = runner.invoke(cli, ['cron']) assert not result.exception, result.output # Test session API call r = client().simulate_get("/api/", headers={"Authorization":usertoken}) assert r.status_code == 200 r = client().simulate_get("/api/", headers={"Authorization":admintoken}) assert r.status_code == 200 r = client().simulate_get("/api/") assert r.status_code == 401 # Test signed certificate API call r = client().simulate_get("/api/signed/nonexistant/") assert r.status_code == 404 r = client().simulate_get("/api/signed/test2/") assert r.status_code == 200 assert r.headers.get('content-type') == "application/x-pem-file" r = client().simulate_get("/api/signed/test2/", headers={"Accept":"application/json"}) assert r.status_code == 200 assert r.headers.get('content-type') == "application/json" r = client().simulate_get("/api/signed/test2/", headers={"Accept":"text/plain"}) assert r.status_code == 415 # Test revocations API call r = client().simulate_get("/api/revoked/") assert r.status_code == 200 assert r.headers.get('content-type') == "application/x-pkcs7-crl" r = requests.get("http://ca.example.lan/api/revoked/") assert r.status_code == 200 assert r.headers.get('content-type') == "application/x-pkcs7-crl" r = client().simulate_get("/api/revoked/", headers={"Accept":"application/x-pem-file"}) assert r.status_code == 200 assert r.headers.get('content-type') == "application/x-pem-file" r = requests.get("http://ca.example.lan/api/revoked/", headers={"Accept":"application/x-pem-file"}) assert r.status_code == 200 assert r.headers.get('content-type') == "application/x-pem-file" r = client().simulate_get("/api/revoked/", headers={"Accept":"text/plain"}) assert r.status_code == 415 r = client().simulate_get("/api/revoked/", query_string="wait=true", headers={"Accept":"application/x-pem-file"}) assert r.status_code == 303 # Test attribute fetching API call r = client().simulate_get("/api/signed/test2/attr/") assert r.status_code == 403 r = client().simulate_get("/api/signed/test2/lease/", headers={"Authorization":admintoken}) assert r.status_code == 404 # Insert lease as if VPN gateway had submitted it path, _, _ = authority.get_signed("test2") from xattr import setxattr setxattr(path, "user.lease.address", b"127.0.0.1") setxattr(path, "user.lease.last_seen", b"random") r = client().simulate_get("/api/signed/test2/attr/") assert r.status_code == 200 # Test lease retrieval r = client().simulate_get("/api/signed/test2/lease/") assert r.status_code == 401 r = client().simulate_get("/api/signed/test2/lease/", headers={"Authorization":usertoken}) assert r.status_code == 403 r = client().simulate_get("/api/signed/test2/lease/", headers={"Authorization":admintoken}) assert r.status_code == 200 assert r.headers.get('content-type') == "application/json; charset=UTF-8" # Tags should not be visible anonymously r = client().simulate_get("/api/signed/test2/tag/") assert r.status_code == 401 r = client().simulate_get("/api/signed/test2/tag/", headers={"Authorization":usertoken}) assert r.status_code == 403 r = client().simulate_get("/api/signed/test2/tag/", headers={"Authorization":admintoken}) assert r.status_code == 200 # Tags can be added only by admin r = client().simulate_post("/api/signed/test2/tag/") assert r.status_code == 401 r = client().simulate_post("/api/signed/test2/tag/", headers={"Authorization":usertoken}) assert r.status_code == 403 r = client().simulate_post("/api/signed/test2/tag/", body="key=other&value=something", headers={"content-type": "application/x-www-form-urlencoded", "Authorization":admintoken}) assert r.status_code == 200 # Tags can be overwritten only by admin r = client().simulate_put("/api/signed/test2/tag/other/") assert r.status_code == 401 r = client().simulate_put("/api/signed/test2/tag/other/", headers={"Authorization":usertoken}) assert r.status_code == 403 r = client().simulate_put("/api/signed/test2/tag/other/", body="value=else", headers={"content-type": "application/x-www-form-urlencoded", "Authorization":admintoken}) assert r.status_code == 200 # Tags can be deleted only by admin r = client().simulate_delete("/api/signed/test2/tag/else/") assert r.status_code == 401 r = client().simulate_delete("/api/signed/test2/tag/else/", headers={"Authorization":usertoken}) assert r.status_code == 403 r = client().simulate_delete("/api/signed/test2/tag/else/", headers={"content-type": "application/x-www-form-urlencoded", "Authorization":admintoken}) assert r.status_code == 200 # Test revocation r = client().simulate_delete("/api/signed/test2/") assert r.status_code == 401 r = client().simulate_delete("/api/signed/test2/", headers={"Authorization":usertoken}) assert r.status_code == 403 r = client().simulate_delete("/api/signed/test2/", headers={"Authorization":admintoken}) assert r.status_code == 200 result = runner.invoke(cli, ['revoke', 'test3']) assert not result.exception, result.output # Test static r = client().simulate_delete("/nonexistant.html") assert r.status_code == 404 r = client().simulate_delete("/index.html") assert r.status_code == 200 # Log can be read only by admin r = client().simulate_get("/api/log/") assert r.status_code == 401 r = client().simulate_get("/api/log/", headers={"Authorization":usertoken}) assert r.status_code == 403 r = client().simulate_get("/api/log/", headers={"Authorization":admintoken}) assert r.status_code == 200 assert r.headers.get('content-type') == "application/json; charset=UTF-8" # Test token mech r = client().simulate_post("/api/token/") assert r.status_code == 404 config.BUNDLE_FORMAT = "ovpn" config.USER_ENROLLMENT_ALLOWED = True r = client().simulate_post("/api/token/") assert r.status_code == 401 # needs auth r = client().simulate_post("/api/token/", headers={"Authorization":usertoken}) assert r.status_code == 403 # regular user forbidden r = client().simulate_post("/api/token/", body="user=userbot", # TODO: test nonexistant user headers={"content-type": "application/x-www-form-urlencoded", "Authorization":admintoken}) assert r.status_code == 200 # token generated by admin r2 = client().simulate_get("/api/token/", query_string="u=userbot&t=1493184342&c=ac9b71421d5741800c5a4905b20c1072594a2df863e60ba836464888786bf2a6", headers={"content-type": "application/x-www-form-urlencoded", "Authorization":admintoken}) assert r2.status_code == 403 # invalid checksum r2 = client().simulate_get("/api/token/", query_string=r.content, headers={"User-Agent":"Mozilla/5.0 (X11; Fedora; Linux x86_64) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/57.0.2987.133 Safari/537.36"}) assert r2.status_code == 200 # token consumed by anyone on Fedora assert r2.headers.get('content-type') == "application/x-openvpn" config.BUNDLE_FORMAT = "p12" # Switch to PKCS#12 r2 = client().simulate_get("/api/token/", query_string=r.content) assert r2.status_code == 200 # token consumed by anyone on unknown device assert r2.headers.get('content-type') == "application/x-pkcs12" result = runner.invoke(cli, ['setup', 'openvpn', 'server', "-cn", "vpn.example.lan", "ca.example.lan"]) assert not result.exception, result.output result = runner.invoke(cli, ['setup', 'openvpn', 'client', "-cn", "roadwarrior1", "ca.example.lan", "vpn.example.lan"]) assert not result.exception, result.output import os if not os.path.exists("/etc/openvpn/keys"): os.makedirs("/etc/openvpn/keys") with open("/etc/certidude/client.conf", "a") as fh: fh.write("insecure = true\n") # pregen dhparam result = runner.invoke(cli, ["request", "--no-wait"]) assert not result.exception, "server responded %s, server logs say %s" % (result.output, open("/var/log/certidude.log").read()) result = runner.invoke(cli, ['sign', 'vpn.example.lan']) assert not result.exception, result.output result = runner.invoke(cli, ["request", "--no-wait"]) assert not result.exception, result.output