first build

tests/conftest.py

@@ -0,0 +1,75 @@
+"""Shared fixtures: settings, a tiny CA, and a profile-signing PKCS#12."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.serialization import pkcs12
+from tests.helpers import issue_test_ca
+
+from cert_portal.config import Settings
+from cert_portal.directory import DirectoryUser
+
+FIXTURES = Path(__file__).parent / "fixtures"
+
+
+@pytest.fixture
+def test_ca():
+    return issue_test_ca()
+
+
+@pytest.fixture
+def signing_p12(tmp_path: Path) -> Path:
+    key, cert = issue_test_ca()
+    blob = pkcs12.serialize_key_and_certificates(
+        name=b"profile-signer",
+        key=key,
+        cert=cert,
+        cas=None,
+        encryption_algorithm=serialization.BestAvailableEncryption(b"signer-secret"),
+    )
+    path = tmp_path / "profile-signing.p12"
+    path.write_bytes(blob)
+    return path
+
+
+@pytest.fixture
+def settings(signing_p12: Path) -> Settings:
+    return Settings(
+        session_secret="test-session-secret-32bytes-min",
+        session_https_only=False,
+        environment="development",
+        docs_enabled=True,
+        public_base_url="http://testserver",
+        oidc_issuer="https://idp.example.test/realms/test",
+        oidc_client_id="cert-portal",
+        oidc_client_secret="secret",
+        oidc_redirect_uri="http://testserver/auth/callback",
+        ldap_url="ldaps://dc.example.test",
+        ldap_bind_dn="CN=portal,DC=example,DC=test",
+        ldap_bind_password="ldap-secret",
+        ldap_base_dn="DC=example,DC=test",
+        certsrv_url="https://ca.example.test/certsrv",
+        certsrv_username="portal",
+        certsrv_password="ca-secret",
+        certsrv_domain="EXAMPLE",
+        certsrv_template="UserWifi",
+        profile_signing_p12_path=signing_p12,
+        profile_signing_p12_password="signer-secret",
+        profile_organization="Test Org",
+        profile_display_name="Test User Certificate",
+        profile_identifier_prefix="com.example.cert-portal",
+    )
+
+
+@pytest.fixture
+def directory_user() -> DirectoryUser:
+    return DirectoryUser(
+        dn="CN=Nick,OU=People,DC=example,DC=test",
+        upn="[email protected]",
+        sid="S-1-5-21-3623811015-3361044348-30300820-1013",
+        display_name="Nick Example",
+        mail="[email protected]",
+    )

tests/fixtures/certfnsh_denied.html

@@ -0,0 +1,10 @@
+<html>
+<head>
+  <title>Microsoft Active Directory Certificate Services</title>
+</head>
+<body>
+  <h1>Microsoft Active Directory Certificate Services</h1>
+  <B>Certificate Request Denied</B>
+  Your certificate request was denied by an administrator.
+</body>
+</html>

tests/fixtures/certfnsh_issued.html

@@ -0,0 +1,12 @@
+<html>
+<head>
+  <title>Microsoft Active Directory Certificate Services</title>
+</head>
+<body>
+  <h1>Microsoft Active Directory Certificate Services</h1>
+  <B> Certificate Issued</B>
+  The certificate you requested was issued to you.
+  <P><A href="certnew.cer?ReqID=18&amp;Enc=bin">DER Encoded</A></P>
+  <P><A href="certnew.cer?ReqID=18&amp;Enc=b64">Base 64 encoded</A></P>
+</body>
+</html>

tests/fixtures/certfnsh_pending.html

@@ -0,0 +1,12 @@
+<html>
+<head>
+  <title>Microsoft Active Directory Certificate Services</title>
+</head>
+<body>
+  <h1>Microsoft Active Directory Certificate Services</h1>
+  <B>Certificate Pending</B>
+  Your certificate request has been received. However, you must wait for an
+  administrator to issue the certificate you requested.
+  <P>Your Request Id is 42.</P>
+</body>
+</html>

tests/fixtures/certfnsh_unknown.html

@@ -0,0 +1,8 @@
+<html>
+<head>
+  <title>Microsoft Active Directory Certificate Services</title>
+</head>
+<body>
+  <p>Unexpected intern error 0x80094011</p>
+</body>
+</html>

tests/helpers.py

@@ -0,0 +1,50 @@
+"""Certificate helpers shared by tests (not pytest fixtures)."""
+
+from __future__ import annotations
+
+import datetime
+
+from cryptography import x509
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.x509.oid import NameOID
+
+
+def _name(cn: str) -> x509.Name:
+    return x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, cn)])
+
+
+def issue_test_ca() -> tuple[rsa.RSAPrivateKey, x509.Certificate]:
+    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    now = datetime.datetime.now(datetime.UTC)
+    cert = (
+        x509.CertificateBuilder()
+        .subject_name(_name("Test CA"))
+        .issuer_name(_name("Test CA"))
+        .public_key(key.public_key())
+        .serial_number(1)
+        .not_valid_before(now - datetime.timedelta(minutes=1))
+        .not_valid_after(now + datetime.timedelta(days=30))
+        .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True)
+        .sign(key, hashes.SHA256())
+    )
+    return key, cert
+
+
+def sign_csr(csr: x509.CertificateSigningRequest, ca_key, ca_cert) -> x509.Certificate:
+    now = datetime.datetime.now(datetime.UTC)
+    builder = (
+        x509.CertificateBuilder()
+        .subject_name(csr.subject)
+        .issuer_name(ca_cert.subject)
+        .public_key(csr.public_key())
+        .serial_number(x509.random_serial_number())
+        .not_valid_before(now - datetime.timedelta(minutes=1))
+        .not_valid_after(now + datetime.timedelta(days=7))
+    )
+    try:
+        san = csr.extensions.get_extension_for_class(x509.SubjectAlternativeName)
+        builder = builder.add_extension(san.value, critical=False)
+    except x509.ExtensionNotFound:
+        pass
+    return builder.sign(ca_key, hashes.SHA256())

tests/test_app.py

@@ -0,0 +1,134 @@
+import json
+from base64 import b64encode
+
+from fastapi.testclient import TestClient
+from itsdangerous import TimestampSigner
+
+from cert_portal.app import create_app
+from cert_portal.auth_oidc import SESSION_USER_KEY, claims_username, session_user_from_token
+from cert_portal.directory import DirectoryUser
+from cert_portal.exceptions import UserNotFoundError
+from cert_portal.issue import IssuedProfile
+from cert_portal.profile import APPLE_CONFIG_TYPE
+
+SESSION_COOKIE = "cert_portal_session"
+
+NICK = {
+    "sub": "abc",
+    "username": "[email protected]",
+    "email": "[email protected]",
+    "name": "Nick",
+}
+
+
+def _client(settings, lookup, issuer=None):
+    app = create_app(settings, lookup_user=lookup, issuer=issuer)
+    return TestClient(app)
+
+
+def _sign_in(client: TestClient, settings, user: dict) -> None:
+    """Starlette 1.x TestClient no longer has session_transaction; sign the cookie."""
+    signer = TimestampSigner(str(settings.session_secret))
+    payload = b64encode(json.dumps({SESSION_USER_KEY: user}).encode("utf-8"))
+    client.cookies.set(SESSION_COOKIE, signer.sign(payload).decode("utf-8"))
+
+
+def test_health(settings, directory_user):
+    client = _client(settings, lambda username: directory_user)
+    response = client.get("/health")
+    assert response.status_code == 200
+    assert response.json()["status"] == "ok"
+
+
+def test_home_unauthenticated_shows_login(settings, directory_user):
+    client = _client(settings, lambda username: directory_user)
+    response = client.get("/")
+    assert response.status_code == 200
+    assert "Sign in with SSO" in response.text
+
+
+def test_help_requires_login(settings, directory_user):
+    client = _client(settings, lambda username: directory_user)
+    response = client.get("/help", follow_redirects=False)
+    assert response.status_code == 302
+    assert response.headers["location"] == "/"
+
+
+def test_home_authenticated_shows_sid(settings, directory_user):
+    client = _client(settings, lambda username: directory_user)
+    _sign_in(client, settings, NICK)
+    response = client.get("/")
+    assert response.status_code == 200
+    assert directory_user.sid in response.text
+    assert "Download certificate profile" in response.text
+
+
+def test_home_directory_error(settings):
+    def lookup(username: str) -> DirectoryUser:
+        raise UserNotFoundError("No Active Directory user matched your sign-in name.")
+
+    client = _client(settings, lookup)
+    _sign_in(
+        client,
+        settings,
+        {
+            "sub": "abc",
+            "username": "[email protected]",
+            "email": "[email protected]",
+            "name": "Missing",
+        },
+    )
+    response = client.get("/")
+    assert "No Active Directory user" in response.text
+
+
+def test_download_returns_signed_mobileconfig(settings, directory_user):
+    class FakeIssuer:
+        def issue(self, user: DirectoryUser) -> IssuedProfile:
+            return IssuedProfile(
+                signed_mobileconfig=b"\x30fake-cms",
+                filename="[email protected]",
+                upn=user.upn,
+                sid=user.sid,
+            )
+
+    client = _client(settings, lambda username: directory_user, issuer=FakeIssuer())
+    _sign_in(client, settings, NICK)
+    response = client.post("/download")
+    assert response.status_code == 200
+    assert response.content == b"\x30fake-cms"
+    assert response.headers["content-type"].startswith(APPLE_CONFIG_TYPE)
+    assert "[email protected]" in response.headers["content-disposition"]
+
+
+def test_download_requires_session(settings, directory_user):
+    client = _client(settings, lambda username: directory_user)
+    response = client.post("/download", follow_redirects=False)
+    assert response.status_code == 302
+
+
+def test_help_authenticated(settings, directory_user):
+    client = _client(settings, lambda username: directory_user)
+    _sign_in(client, settings, NICK)
+    response = client.get("/help")
+    assert response.status_code == 200
+    assert "Private key on the server" in response.text
+
+
+def test_login_oidc_failure_shows_error(settings, directory_user):
+    client = _client(settings, lambda username: directory_user)
+    response = client.get("/login")
+    assert response.status_code == 502
+    assert "Could not start sign-in" in response.text
+
+
+def test_username_claim_fallback():
+    assert claims_username({"email": "[email protected]"}, "email") == "[email protected]"
+    assert claims_username({"preferred_username": "a"}, "email") == "a"
+
+
+def test_session_user_from_token():
+    token = {"userinfo": {"sub": "1", "email": "[email protected]", "name": "A"}}
+    user = session_user_from_token(token, "email")
+    assert user["username"] == "[email protected]"
+    assert user["name"] == "A"

tests/test_certsrv.py

@@ -0,0 +1,36 @@
+from pathlib import Path
+
+from cert_portal.certsrv import parse_certfnsh_html
+
+FIXTURES = Path(__file__).parent / "fixtures"
+
+
+def test_parse_issued_uses_reqid_from_certnew_link():
+    html = (FIXTURES / "certfnsh_issued.html").read_text(encoding="utf-8")
+    parsed = parse_certfnsh_html(html)
+    assert parsed.disposition == "issued"
+    assert parsed.req_id == 18
+
+
+def test_parse_pending():
+    html = (FIXTURES / "certfnsh_pending.html").read_text(encoding="utf-8")
+    parsed = parse_certfnsh_html(html)
+    assert parsed.disposition == "pending"
+    assert parsed.req_id == 42
+
+
+def test_parse_denied():
+    html = (FIXTURES / "certfnsh_denied.html").read_text(encoding="utf-8")
+    parsed = parse_certfnsh_html(html)
+    assert parsed.disposition == "denied"
+    assert parsed.req_id is None
+
+
+def test_parse_unknown_layout():
+    html = (FIXTURES / "certfnsh_unknown.html").read_text(encoding="utf-8")
+    parsed = parse_certfnsh_html(html)
+    assert parsed.disposition == "unknown"
+
+
+def test_empty_html():
+    assert parse_certfnsh_html("").disposition == "unknown"

tests/test_directory.py

@@ -0,0 +1,18 @@
+from cert_portal.directory import decode_object_sid, encode_object_sid
+
+
+def test_sid_round_trip_typical_account():
+    sid = "S-1-5-21-3623811015-3361044348-30300820-1013"
+    assert decode_object_sid(encode_object_sid(sid)) == sid
+
+
+def test_sid_well_known_administrators():
+    sid = "S-1-5-32-544"
+    assert decode_object_sid(encode_object_sid(sid)) == sid
+
+
+def test_sid_revision_and_authority():
+    raw = encode_object_sid("S-1-5-18")
+    assert raw[0] == 1
+    assert raw[1] == 1  # one sub-authority
+    assert decode_object_sid(raw) == "S-1-5-18"

tests/test_enrollment.py

@@ -0,0 +1,33 @@
+from cryptography import x509
+from cryptography.x509 import load_pem_x509_csr
+
+from cert_portal.enrollment import (
+    SID_SAN_URI_PREFIX,
+    UPN_OID,
+    build_csr,
+    generate_rsa_key,
+    sid_san_uri,
+)
+
+
+def test_csr_contains_upn_othername_and_sid_uri():
+    key = generate_rsa_key(2048).key
+    upn = "[email protected]"
+    sid = "S-1-5-21-1-2-3-4"
+    pem = build_csr(key=key, upn=upn, sid=sid)
+    csr = load_pem_x509_csr(pem)
+    assert csr.subject.rfc4514_string() == f"CN={upn}"
+    san = csr.extensions.get_extension_for_class(x509.SubjectAlternativeName).value
+    names = list(san)
+    uris = [n.value for n in names if isinstance(n, x509.UniformResourceIdentifier)]
+    others = [n for n in names if isinstance(n, x509.OtherName)]
+    assert uris == [f"{SID_SAN_URI_PREFIX}{sid}"]
+    assert sid_san_uri(sid) in uris
+    assert any(o.type_id == UPN_OID for o in others)
+    upn_other = next(o for o in others if o.type_id == UPN_OID)
+    # DER UTF8String: tag 0x0c, short length, payload
+    assert upn_other.value[0] == 0x0C
+    length = upn_other.value[1]
+    assert upn_other.value[2 : 2 + length].decode("utf-8") == upn
+    assert b"-----BEGIN CERTIFICATE REQUEST-----" in pem
+    assert pem.startswith(b"-----BEGIN")

tests/test_issue.py

@@ -0,0 +1,43 @@
+from cryptography import x509
+from cryptography.hazmat.primitives.serialization import Encoding
+from cryptography.x509 import load_pem_x509_csr
+from tests.helpers import sign_csr
+
+from cert_portal.certsrv import CertsrvClient
+from cert_portal.enrollment import SID_SAN_URI_PREFIX
+from cert_portal.issue import ProfileIssuer
+
+
+class _FakeCertsrv:
+    def __init__(self, ca_key, ca_cert):
+        self.ca_key = ca_key
+        self.ca_cert = ca_cert
+        self.last_csr = None
+
+    def submit_csr(self, csr_pem, template=None):
+        pem = csr_pem if isinstance(csr_pem, bytes) else csr_pem.encode("ascii")
+        self.last_csr = pem
+        csr = load_pem_x509_csr(pem)
+        cert = sign_csr(csr, self.ca_key, self.ca_cert)
+        return cert.public_bytes(Encoding.PEM)
+
+
+def test_issue_signs_profile_for_user(settings, directory_user, test_ca):
+    ca_key, ca_cert = test_ca
+    fake = _FakeCertsrv(ca_key, ca_cert)
+    issuer = ProfileIssuer(settings, certsrv=fake)
+    issued = issuer.issue(directory_user)
+    assert issued.upn == directory_user.upn
+    assert issued.sid == directory_user.sid
+    assert issued.filename.endswith(".mobileconfig")
+    assert issued.signed_mobileconfig.startswith(b"\x30")
+    assert b"com.apple.security.pkcs12" in issued.signed_mobileconfig
+    csr = load_pem_x509_csr(fake.last_csr)
+    san = csr.extensions.get_extension_for_class(x509.SubjectAlternativeName).value
+    uris = [n.value for n in san if isinstance(n, x509.UniformResourceIdentifier)]
+    assert f"{SID_SAN_URI_PREFIX}{directory_user.sid}" in uris
+
+
+def test_ntlm_username_includes_domain(settings):
+    client = CertsrvClient(settings)
+    assert client._ntlm_user() == "EXAMPLE\\portal"

tests/test_profile.py

@@ -0,0 +1,55 @@
+import plistlib
+
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.serialization import pkcs7
+from tests.helpers import issue_test_ca
+
+from cert_portal.enrollment import export_pkcs12
+from cert_portal.profile import (
+    PKCS12_PAYLOAD_TYPE,
+    build_unsigned_mobileconfig,
+    load_signing_identity,
+    sign_mobileconfig,
+)
+
+
+def test_unsigned_profile_embeds_pkcs12_and_password():
+    key, cert = issue_test_ca()
+    p12, password = export_pkcs12(key=key, certificate=cert, friendly_name="[email protected]")
+    unsigned = build_unsigned_mobileconfig(
+        identity_p12=p12,
+        identity_password=password,
+        display_name="User Certificate",
+        organization="Example",
+        identifier_prefix="com.example.cert-portal",
+        username="[email protected]",
+    )
+    profile = plistlib.loads(unsigned)
+    payloads = profile["PayloadContent"]
+    types = [p["PayloadType"] for p in payloads]
+    assert PKCS12_PAYLOAD_TYPE in types
+    assert "com.apple.security.scep" not in types
+    identity = next(p for p in payloads if p["PayloadType"] == PKCS12_PAYLOAD_TYPE)
+    assert identity["Password"] == password
+    assert identity["PayloadContent"] == p12
+
+
+def test_signed_profile_is_cms_and_contains_plist(signing_p12):
+    key, cert = issue_test_ca()
+    p12, password = export_pkcs12(key=key, certificate=cert)
+    unsigned = build_unsigned_mobileconfig(
+        identity_p12=p12,
+        identity_password=password,
+        display_name="User Certificate",
+        organization="Example",
+        identifier_prefix="com.example.cert-portal",
+        username="[email protected]",
+    )
+    identity = load_signing_identity(signing_p12, "signer-secret")
+    signed = sign_mobileconfig(unsigned, identity)
+    assert signed != unsigned
+    assert signed[:1] == b"\x30"  # DER SEQUENCE
+    assert b"com.apple.security.pkcs12" in signed
+    certs = pkcs7.load_der_pkcs7_certificates(signed)
+    fps = {c.fingerprint(hashes.SHA256()) for c in certs}
+    assert identity.certificate.fingerprint(hashes.SHA256()) in fps