"""Additional route-level tests: CSRs, tokens, import/export and templates.""" from __future__ import annotations import base64 import unittest from datetime import datetime, timedelta, timezone from tests.helpers.api import api_client from tests.helpers.asynctest import async_test from tests.helpers.oidc import oidc_user def _subject(cn: str) -> list[dict]: return [{"name": "C", "value": "DE"}, {"name": "O", "value": "KCA"}, {"name": "CN", "value": cn}] def _extensions(ca: bool = False) -> dict: return { "basic_constraints": {"critical": True, "ca": ca, "path_length": None}, "key_usage": {"critical": True, "usages": ["key_cert_sign", "crl_sign"] if ca else ["digital_signature"]}, "subject_key_identifier": {"critical": False}, "authority_key_identifier": {"critical": False}, } async def _vault(client) -> str: response = await client.post("/api/vaults", json={"name": "Test"}) assert response.status_code == 201, response.text return response.json()["id"] async def _key(client, vault_id: str, name: str = "key") -> dict: response = await client.post( f"/api/vaults/{vault_id}/keys", json={"name": name, "key_type": "rsa", "size": 2048} ) assert response.status_code == 201, response.text return response.json() async def _cert(client, vault_id: str, key_id: str, name: str, ca: bool = False, signer_id: str | None = None) -> dict: now = datetime.now(timezone.utc) response = await client.post( f"/api/vaults/{vault_id}/certificates", json={ "name": name, "key_id": key_id, "signer_id": signer_id, "subject": _subject(name), "not_before": now.isoformat(), "not_after": (now + timedelta(days=365)).isoformat(), "extensions": _extensions(ca), }, ) assert response.status_code == 201, response.text return response.json() class KeyGenerationTests(unittest.TestCase): @async_test async def test_shared_generation_is_offloaded_and_bounded(self) -> None: import asyncio import threading from unittest.mock import patch from kca import service active = 0 max_active = 0 worker_threads: set[int] = set() lock = threading.Lock() barrier = threading.Barrier(2) def fake_generate(*args: object) -> object: nonlocal active, max_active with lock: active += 1 max_active = max(max_active, active) worker_threads.add(threading.get_ident()) try: barrier.wait(timeout=5) finally: with lock: active -= 1 return object() with ( patch.object(service, "_keygen_semaphore", asyncio.Semaphore(2)), patch.object(service.key_core, "generate_key", new=fake_generate), ): results = await asyncio.gather(*(service.generate_key("rsa", 8192) for _ in range(6))) self.assertEqual(len(results), 6) self.assertEqual(max_active, 2) self.assertNotIn(threading.get_ident(), worker_threads) @async_test async def test_certificate_generation_is_offloaded(self) -> None: import threading from unittest.mock import patch from kca.core import keys as key_core event_loop_thread = threading.get_ident() generation_threads: list[int] = [] generated_key = key_core.generate_key("rsa", 1024) def fake_generate(*args: object) -> object: generation_threads.append(threading.get_ident()) return generated_key async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) with patch.object(key_core, "generate_key", new=fake_generate): response = await client.post( f"/api/vaults/{vault_id}/certificates/preview", json={ "subject": _subject("preview"), "new_key": {"key_type": "rsa", "size": 8192}, }, ) assert response.status_code == 200, response.text response = await client.post( f"/api/vaults/{vault_id}/certificates/preview", json={"subject": _subject("default preview")}, ) assert response.status_code == 200, response.text response = await client.post( f"/api/vaults/{vault_id}/certificates", json={ "name": "issued", "subject": _subject("issued"), "new_key": {"key_type": "rsa", "size": 8192, "name": "issued key"}, }, ) assert response.status_code == 201, response.text self.assertEqual(len(generation_threads), 3) self.assertNotIn(event_loop_thread, generation_threads) class RouteExtraTests(unittest.TestCase): @async_test async def test_csr_creation_and_signing(self) -> None: async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) ca_key = await _key(client, vault_id, "ca key") ca = await _cert(client, vault_id, ca_key["id"], "Root CA", ca=True) response = await client.post( f"/api/vaults/{vault_id}/requests", json={ "name": "web request", "subject": _subject("web.example.com"), "new_key": {"key_type": "ec", "curve": "secp256r1", "name": "web key"}, "extensions": _extensions(False), }, ) assert response.status_code == 201, response.text request = response.json() assert request["signature_valid"] is True response = await client.post( f"/api/vaults/{vault_id}/certificates", json={ "name": "web cert", "request_id": request["id"], "signer_id": ca["id"], "not_before": datetime.now(timezone.utc).isoformat(), "not_after": (datetime.now(timezone.utc) + timedelta(days=30)).isoformat(), }, ) assert response.status_code == 201, response.text cert = response.json() assert cert["issuer_id"] == ca["id"] response = await client.get(f"/api/vaults/{vault_id}/requests/{request['id']}") assert response.json()["signed"] is True @async_test async def test_invalid_csr_signature_cannot_be_issued(self) -> None: from cryptography import x509 from kca.core import keys as key_core from kca.core import requests as req_core from kca.models import Certificate async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) ca_key = await _key(client, vault_id, "ca key") ca = await _cert(client, vault_id, ca_key["id"], "Root CA", ca=True) key = key_core.generate_key("rsa", 2048) csr = req_core.build_csr( subject=x509.Name( [x509.NameAttribute(x509.oid.NameOID.COMMON_NAME, "invalid.example.com")] ), private_key=key, ) der = bytearray(req_core.csr_der(csr)) der[-1] ^= 1 invalid_der = bytes(der) assert not req_core.verify_csr(req_core.parse_csr(invalid_der)) response = await client.post( f"/api/vaults/{vault_id}/import", json={ "filename": "invalid.csr", "content_base64": base64.b64encode(invalid_der).decode(), }, ) assert response.status_code == 200, response.text imported = response.json()["imported"] assert len(imported) == 1 assert imported[0]["kind"] == "request" request_id = imported[0]["id"] response = await client.get(f"/api/vaults/{vault_id}/requests/{request_id}") assert response.status_code == 200, response.text assert response.json()["signature_valid"] is False now = datetime.now(timezone.utc) response = await client.post( f"/api/vaults/{vault_id}/certificates", json={ "name": "invalid CSR cert", "request_id": request_id, "signer_id": ca["id"], "not_before": now.isoformat(), "not_after": (now + timedelta(days=30)).isoformat(), }, ) assert response.status_code == 400, response.text assert response.json()["error"] == "signing request has an invalid signature" assert await Certificate.filter(request_id=request_id).count() == 0 response = await client.get(f"/api/vaults/{vault_id}/requests/{request_id}") assert response.json()["signed"] is False @async_test async def test_token_authentication(self) -> None: async with api_client() as client: with oidc_user("alice"): response = await client.post("/api/tokens", json={"name": "script"}) assert response.status_code == 201, response.text secret = response.json()["token"] assert secret.startswith("kca_") # Session is no longer used: the bearer token authenticates. response = await client.get("/api/vaults", headers={"authorization": f"Bearer {secret}"}) assert response.status_code == 200 response = await client.post("/api/tokens", headers={"authorization": f"Bearer {secret}"}, json={"name": "x"}) assert response.status_code == 401 @async_test async def test_import_preview_and_commit(self) -> None: async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) key = await _key(client, vault_id, "import key") cert = await _cert(client, vault_id, key["id"], "Import Me", ca=True) export = await client.post( f"/api/vaults/{vault_id}/certificates/{cert['id']}/export", json={"format": "pem"} ) pem = base64.b64decode(export.json()["content_base64"]) response = await client.post( f"/api/vaults/{vault_id}/import", json={"filename": "test.crt", "content_base64": base64.b64encode(pem).decode(), "dry_run": True}, ) assert response.status_code == 200, response.text assert response.json()["items"][0]["duplicate"] is True response = await client.post( f"/api/vaults/{vault_id}/import", json={"filename": "test.crt", "content_base64": base64.b64encode(pem).decode()}, ) assert response.status_code == 200 assert len(response.json()["skipped"]) == 1 @async_test async def test_settings_and_templates(self) -> None: async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) response = await client.get(f"/api/vaults/{vault_id}/settings") assert response.status_code == 200 version = response.json()["version"] response = await client.patch( f"/api/vaults/{vault_id}/settings", json={"settings": {"default_hash": "sha384", "serial_len_bits": 80}, "version": version}, ) assert response.status_code == 200, response.text assert response.json()["settings"]["default_hash"] == "sha384" response = await client.get(f"/api/vaults/{vault_id}/templates") templates = response.json()["results"] assert any(t["id"] == "builtin:ca" for t in templates) response = await client.post( f"/api/vaults/{vault_id}/templates/builtin:ca/apply", json={"target": {"subject": [], "extensions": {}}}, ) assert response.status_code == 200 assert response.json()["extensions"]["basic_constraints"]["ca"] is True response = await client.post( f"/api/vaults/{vault_id}/templates/builtin:tls-server/apply", json={ "target": { "subject": [{"name": "CN", "value": "keep.example.com"}], "extensions": {}, }, "apply_subject": False, "apply_extensions": True, }, ) assert response.status_code == 200 merged = response.json() assert merged["subject"] == [{"name": "CN", "value": "keep.example.com"}] assert "serverAuth" in merged["extensions"]["extended_key_usage"]["usages"] settings_response = await client.get(f"/api/vaults/{vault_id}/settings") response = await client.patch( f"/api/vaults/{vault_id}/settings", json={ "settings": {"explicit_dn": ["CN", "not-a-real-attribute"]}, "version": settings_response.json()["version"], }, ) assert response.status_code == 400 @async_test async def test_revocation_and_crl(self) -> None: async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) ca_key = await _key(client, vault_id, "ca") ca = await _cert(client, vault_id, ca_key["id"], "Root", ca=True) leaf_key = await _key(client, vault_id, "leaf") leaf = await _cert( client, vault_id, leaf_key["id"], "Leaf", signer_id=ca["id"] ) response = await client.post( f"/api/vaults/{vault_id}/certificates/{leaf['id']}/revoke", json={"reason": "keyCompromise"}, ) assert response.status_code == 200, response.text response = await client.get( f"/api/vaults/{vault_id}/certificates/{leaf['id']}" ) assert response.json()["revoked"] is True response = await client.post( f"/api/vaults/{vault_id}/certificates/{ca['id']}/crl", json={"days": 30}, ) assert response.status_code == 201, response.text crl = response.json() assert crl["revoked_count"] == 1 assert crl["crl_number"] == 1 @async_test async def test_key_delete_unlinks(self) -> None: async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) key = await _key(client, vault_id) cert = await _cert(client, vault_id, key["id"], "Unlink") response = await client.delete(f"/api/vaults/{vault_id}/keys/{key['id']}") assert response.status_code == 204 response = await client.get(f"/api/vaults/{vault_id}/certificates/{cert['id']}") assert response.json()["key_id"] is None class ExtendedRouteTests(unittest.TestCase): @async_test async def test_preview_advanced_extensions(self) -> None: async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) key = await _key(client, vault_id) payload = { "name": "preview", "key_id": key["id"], "subject": _subject("preview"), "extensions": { "basic_constraints": {"critical": True, "ca": False, "path_length": None}, "crl_distribution_points": {"critical": False, "urls": ["http://example.com/ca.crl"]}, "authority_information_access": { "critical": False, "descriptions": [{"method": "ocsp", "location": {"type": "uri", "value": "http://ocsp.example.com"}}], }, "raw": [{"oid": "1.2.3.4", "critical": False, "der_hex": "0c03616263"}], }, } response = await client.post(f"/api/vaults/{vault_id}/certificates/preview", json=payload) assert response.status_code == 200, response.text names = [row["name"] for row in response.json()["extensions"]] assert "CRL Distribution Points" in names assert "Authority Information Access" in names @async_test async def test_mandatory_dn_and_no_expiration(self) -> None: async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) settings = (await client.get(f"/api/vaults/{vault_id}/settings")).json() await client.patch( f"/api/vaults/{vault_id}/settings", json={"settings": {"mandatory_dn": ["CN"]}, "version": settings["version"]}, ) key = await _key(client, vault_id) now = datetime.now(timezone.utc) response = await client.post( f"/api/vaults/{vault_id}/certificates", json={ "name": "missing cn", "key_id": key["id"], "subject": [{"name": "O", "value": "KCA"}], "validity": {"amount": 1, "unit": "days"}, }, ) assert response.status_code == 400 assert "mandatory" in response.json()["error"] response = await client.post( f"/api/vaults/{vault_id}/certificates", json={ "name": "forever", "key_id": key["id"], "subject": _subject("forever"), "validity": {"no_expiration": True}, }, ) assert response.status_code == 201, response.text assert response.json()["no_expiration"] is True @async_test async def test_ca_template_and_revocation_edit(self) -> None: async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) ca_key = await _key(client, vault_id, "ca") ca = await _cert(client, vault_id, ca_key["id"], "CA", ca=True) response = await client.patch( f"/api/vaults/{vault_id}/certificates/{ca['id']}/ca", json={"crl_days": 45, "template_id": "builtin:tls-server"}, ) assert response.status_code == 200, response.text assert response.json()["default_template"] == "builtin:tls-server" assert response.json()["crl_days"] == 45 response = await client.post( f"/api/vaults/{vault_id}/certificates/{ca['id']}/revocations", json={"serial_hex": "1234", "reason": "keyCompromise", "invalidity_date": datetime.now(timezone.utc).isoformat()}, ) assert response.status_code == 201, response.text rev = response.json() response = await client.patch( f"/api/vaults/{vault_id}/certificates/{ca['id']}/revocations/{rev['id']}", json={"reason": "superseded", "invalidity_date": None}, ) assert response.status_code == 200, response.text assert response.json()["reason"] == "superseded" class SerialRenewalTests(unittest.TestCase): @async_test async def test_explicit_serial_must_be_unique_and_not_revoked(self) -> None: async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) ca_key = await _key(client, vault_id, "serial ca") ca = await _cert(client, vault_id, ca_key["id"], "Serial CA", ca=True) leaf_key = await _key(client, vault_id, "serial leaf") async def issue(name: str, serial: str): now = datetime.now(timezone.utc) return await client.post( f"/api/vaults/{vault_id}/certificates", json={ "name": name, "key_id": leaf_key["id"], "signer_id": ca["id"], "subject": _subject(name), "serial_hex": serial, "not_before": now.isoformat(), "not_after": (now + timedelta(days=365)).isoformat(), "extensions": _extensions(False), }, ) response = await issue("Serial leaf", "00ab") assert response.status_code == 201, response.text assert response.json()["serial_hex"] == "ab" response = await issue("Duplicate serial", "AB") assert response.status_code == 400, response.text assert "serial" in response.json()["error"] response = await client.post( f"/api/vaults/{vault_id}/certificates/{ca['id']}/revocations", json={"serial_hex": "beef", "reason": "superseded"}, ) assert response.status_code == 201, response.text response = await issue("Revoked serial", "beef") assert response.status_code == 400, response.text assert "serial" in response.json()["error"] response = await client.get(f"/api/vaults/{vault_id}/certificates") assert len(response.json()["results"]) == 2 @async_test async def test_keep_serial_rejects_duplicate_and_revoked_renewals(self) -> None: async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) ca_key = await _key(client, vault_id, "renewal ca") ca = await _cert(client, vault_id, ca_key["id"], "Renewal CA", ca=True) leaf_key = await _key(client, vault_id, "renewal leaf") leaf = await _cert( client, vault_id, leaf_key["id"], "Renewal Leaf", signer_id=ca["id"] ) renew_url = f"/api/vaults/{vault_id}/certificates/{leaf['id']}/renew" for payload in ( {"keep_serial": True, "revoke_old": True, "replace_old": True}, {"keep_serial": True}, ): response = await client.post(renew_url, json=payload) assert response.status_code == 400, response.text assert "serial" in response.json()["error"] response = await client.get(f"/api/vaults/{vault_id}/certificates/{leaf['id']}") assert response.status_code == 200 response = await client.get(f"/api/vaults/{vault_id}/certificates") assert len(response.json()["results"]) == 2 response = await client.get( f"/api/vaults/{vault_id}/certificates/{ca['id']}/revocations" ) assert response.json()["results"] == [] response = await client.post( f"/api/vaults/{vault_id}/certificates/{leaf['id']}/revoke", json={"reason": "superseded"}, ) assert response.status_code == 200, response.text response = await client.post( renew_url, json={"keep_serial": True, "replace_old": True} ) assert response.status_code == 400, response.text assert "serial" in response.json()["error"] response = await client.get(f"/api/vaults/{vault_id}/certificates/{leaf['id']}") assert response.status_code == 200 assert response.json()["revoked"] is True response = await client.get(f"/api/vaults/{vault_id}/certificates") assert len(response.json()["results"]) == 2 @async_test async def test_safe_serial_renewals_have_distinct_ocsp_status(self) -> None: async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) ca_key = await _key(client, vault_id, "ocsp renewal ca") ca = await _cert(client, vault_id, ca_key["id"], "OCSP Renewal CA", ca=True) leaf_key = await _key(client, vault_id, "ocsp renewal leaf") leaf = await _cert( client, vault_id, leaf_key["id"], "Kept Serial Leaf", signer_id=ca["id"] ) kept_serial = leaf["serial_hex"] response = await client.post( f"/api/vaults/{vault_id}/certificates/{leaf['id']}/renew", json={ "keep_serial": True, "replace_old": True, "validity": {"amount": 30, "unit": "days"}, }, ) assert response.status_code == 201, response.text renewed = response.json() assert renewed["serial_hex"] == kept_serial response = await client.get(f"/api/vaults/{vault_id}/certificates/{leaf['id']}") assert response.status_code == 404 response = await client.get( f"/api/vaults/{vault_id}/certificates/{ca['id']}/ocsp-status" ) entries = response.json()["snapshot"]["entries"] assert len(entries) == 1 assert entries[0]["serial_hex"] == kept_serial.lower() assert entries[0]["status"] == "valid" old_leaf = await _cert( client, vault_id, leaf_key["id"], "New Serial Leaf", signer_id=ca["id"] ) response = await client.post( f"/api/vaults/{vault_id}/certificates/{old_leaf['id']}/renew", json={"revoke_old": True, "replace_old": True}, ) assert response.status_code == 201, response.text new_leaf = response.json() assert new_leaf["serial_hex"] != old_leaf["serial_hex"] assert new_leaf["revoked"] is False response = await client.get( f"/api/vaults/{vault_id}/certificates/{ca['id']}/ocsp-status" ) entries = { entry["serial_hex"]: entry for entry in response.json()["snapshot"]["entries"] } assert len(entries) == 3 assert entries[kept_serial.lower()]["status"] == "valid" assert entries[old_leaf["serial_hex"].lower()]["status"] == "revoked" assert entries[new_leaf["serial_hex"].lower()]["status"] == "valid" class VaultExportTests(unittest.TestCase): @async_test async def test_vault_zip_export(self) -> None: import io import zipfile async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) key = await _key(client, vault_id, "zip key") cert = await _cert(client, vault_id, key["id"], "Zip Cert", ca=True) response = await client.post(f"/api/vaults/{vault_id}/export", json={"password": "secret"}) assert response.status_code == 200, response.text payload = response.json() data = base64.b64decode(payload["content_base64"]) with zipfile.ZipFile(io.BytesIO(data)) as archive: names = archive.namelist() assert "manifest.json" in names assert any(name.startswith("certificates/") for name in names) key_files = [name for name in names if name.startswith("keys/") and name.endswith(".key")] assert key_files assert b"BEGIN ENCRYPTED PRIVATE KEY" in archive.read(key_files[0]) manifest = archive.read("manifest.json") assert b"Zip Cert" in manifest class BulkRevocationTests(unittest.TestCase): @async_test async def test_bulk_revoke_reports_skips_and_errors(self) -> None: async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) ca_key = await _key(client, vault_id, "bulk ca") ca = await _cert(client, vault_id, ca_key["id"], "Bulk CA", ca=True) leaf1_key = await _key(client, vault_id, "leaf1") leaf1 = await _cert(client, vault_id, leaf1_key["id"], "Leaf 1", signer_id=ca["id"]) leaf2_key = await _key(client, vault_id, "leaf2") leaf2 = await _cert(client, vault_id, leaf2_key["id"], "Leaf 2", signer_id=ca["id"]) self_key = await _key(client, vault_id, "self") self_signed = await _cert(client, vault_id, self_key["id"], "Self") response = await client.post( f"/api/vaults/{vault_id}/certificates/bulk-revoke", json={ "certificate_ids": [leaf1["id"], leaf2["id"], self_signed["id"]], "reason": "keyCompromise", }, ) assert response.status_code == 200, response.text result = response.json() assert len(result["revoked"]) == 2 assert len(result["errors"]) == 1 assert result["errors"][0]["id"] == self_signed["id"] response = await client.get(f"/api/vaults/{vault_id}/certificates/{leaf1['id']}") assert response.json()["revoked"] is True response = await client.post( f"/api/vaults/{vault_id}/certificates/bulk-revoke", json={"certificate_ids": [leaf1["id"], leaf2["id"]], "reason": "superseded"}, ) assert response.status_code == 200, response.text assert len(response.json()["skipped"]) == 2 response = await client.post( f"/api/vaults/{vault_id}/certificates/bulk-revoke", json={"certificate_ids": [], "reason": "superseded"}, ) assert response.status_code == 400 response = await client.post( f"/api/vaults/{vault_id}/certificates/bulk-revoke", json={"certificate_ids": ["not-a-uuid"], "reason": "superseded"}, ) assert response.status_code == 200 assert response.json()["errors"][0]["error"] == "invalid certificate id" @async_test async def test_bulk_revoke_validation_and_isolation(self) -> None: async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) response = await client.post( f"/api/vaults/{vault_id}/certificates/bulk-revoke", json={"certificate_ids": ["00000000-0000-0000-0000-000000000000"], "reason": "superseded", "invalidity_date": "not-a-date"}, ) assert response.status_code == 400 response = await client.post( f"/api/vaults/{vault_id}/certificates/bulk-revoke", json={"certificate_ids": [str(i) for i in range(201)], "reason": "superseded"}, ) assert response.status_code == 400 with oidc_user("bob"): response = await client.post( f"/api/vaults/{vault_id}/certificates/bulk-revoke", json={"certificate_ids": ["00000000-0000-0000-0000-000000000000"], "reason": "superseded"}, ) assert response.status_code == 404 class OcspSnapshotTests(unittest.TestCase): @async_test async def test_snapshot_contains_status_and_reason(self) -> None: async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) ca_key = await _key(client, vault_id, "ocsp ca") ca = await _cert(client, vault_id, ca_key["id"], "OCSP CA", ca=True) leaf_key = await _key(client, vault_id, "ocsp leaf") leaf = await _cert(client, vault_id, leaf_key["id"], "OCSP Leaf", signer_id=ca["id"]) await client.post( f"/api/vaults/{vault_id}/certificates/{leaf['id']}/revoke", json={"reason": "keyCompromise"}, ) response = await client.get( f"/api/vaults/{vault_id}/certificates/{ca['id']}/ocsp-status" ) assert response.status_code == 200, response.text payload = response.json() assert payload["enabled"] is False snapshot = payload["snapshot"] assert snapshot["schema"] == "kca-ocsp-status/v1" assert snapshot["issuer"]["sha256"] == ca["fingerprint_sha256"] entries = {entry["serial_hex"]: entry for entry in snapshot["entries"]} assert len(snapshot["entries"]) == 1 assert entries[leaf["serial_hex"]]["status"] == "revoked" assert entries[leaf["serial_hex"]]["reason"] == "keyCompromise" assert entries[leaf["serial_hex"]]["revoked_at"] assert entries[leaf["serial_hex"]]["not_after"] is not None @async_test async def test_snapshot_includes_manual_revocation_without_certificate(self) -> None: async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) ca_key = await _key(client, vault_id, "manual ocsp ca") ca = await _cert(client, vault_id, ca_key["id"], "Manual OCSP CA", ca=True) response = await client.post( f"/api/vaults/{vault_id}/certificates/{ca['id']}/revocations", json={ "serial_hex": "AB:CD", "reason": "keyCompromise", "invalidity_date": datetime.now(timezone.utc).isoformat(), }, ) assert response.status_code == 201, response.text response = await client.get( f"/api/vaults/{vault_id}/certificates/{ca['id']}/ocsp-status" ) assert response.status_code == 200, response.text entries = response.json()["snapshot"]["entries"] assert len(entries) == 1 entry = entries[0] assert entry["serial_hex"] == "abcd" assert entry["status"] == "revoked" assert entry["reason"] == "keyCompromise" assert entry["revoked_at"] assert entry["invalidity_date"] assert entry["not_after"] is None @async_test async def test_snapshot_includes_crl_imported_revocation_without_certificate(self) -> None: from kca.models import CertificateRevocation, Crl async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) ca_key = await _key(client, vault_id, "crl ocsp ca") ca = await _cert(client, vault_id, ca_key["id"], "CRL OCSP CA", ca=True) response = await client.post( f"/api/vaults/{vault_id}/certificates/{ca['id']}/revocations", json={"serial_hex": "1234abcd", "reason": "superseded"}, ) assert response.status_code == 201, response.text response = await client.post( f"/api/vaults/{vault_id}/certificates/{ca['id']}/crl", json={"days": 30}, ) assert response.status_code == 201, response.text crl = await Crl.get(id=response.json()["id"]) crl_der = bytes(crl.crl_der) await crl.delete() await CertificateRevocation.filter(ca_id=ca["id"]).delete() response = await client.post( f"/api/vaults/{vault_id}/import", json={ "filename": "unknown-serial.crl", "content_base64": base64.b64encode(crl_der).decode(), }, ) assert response.status_code == 200, response.text assert [item["kind"] for item in response.json()["imported"]] == ["crl"] response = await client.get( f"/api/vaults/{vault_id}/certificates/{ca['id']}/ocsp-status" ) assert response.status_code == 200, response.text entries = response.json()["snapshot"]["entries"] assert len(entries) == 1, entries assert entries[0]["serial_hex"] == "1234abcd" assert entries[0]["status"] == "revoked" assert entries[0]["reason"] == "superseded" assert entries[0]["revoked_at"] assert entries[0]["not_after"] is None @async_test async def test_snapshot_retains_revocation_after_certificate_delete(self) -> None: async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) ca_key = await _key(client, vault_id, "deleted ocsp ca") ca = await _cert(client, vault_id, ca_key["id"], "Deleted OCSP CA", ca=True) leaf_key = await _key(client, vault_id, "deleted ocsp leaf") leaf = await _cert( client, vault_id, leaf_key["id"], "Deleted OCSP Leaf", signer_id=ca["id"] ) response = await client.post( f"/api/vaults/{vault_id}/certificates/{leaf['id']}/revoke", json={"reason": "cessationOfOperation"}, ) assert response.status_code == 200, response.text response = await client.delete( f"/api/vaults/{vault_id}/certificates/{leaf['id']}" ) assert response.status_code == 204, response.text response = await client.get( f"/api/vaults/{vault_id}/certificates/{ca['id']}/ocsp-status" ) assert response.status_code == 200, response.text entries = response.json()["snapshot"]["entries"] assert len(entries) == 1 assert entries[0]["serial_hex"] == leaf["serial_hex"].lower() assert entries[0]["status"] == "revoked" assert entries[0]["reason"] == "cessationOfOperation" assert entries[0]["revoked_at"] assert entries[0]["not_after"] is None @async_test async def test_revocation_marks_snapshot_dirty(self) -> None: from dataclasses import replace from unittest.mock import patch from kca import ocsp from kca.models import OcspSnapshot async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) ca_key = await _key(client, vault_id, "dirty ca") ca = await _cert(client, vault_id, ca_key["id"], "Dirty CA", ca=True) leaf_key = await _key(client, vault_id, "dirty leaf") leaf = await _cert(client, vault_id, leaf_key["id"], "Dirty Leaf", signer_id=ca["id"]) configured = replace(ocsp.settings, ocsp_s3_bucket="test-bucket") with patch.object(ocsp, "settings", configured): response = await client.post( f"/api/vaults/{vault_id}/certificates/{leaf['id']}/revoke", json={"reason": "superseded"}, ) assert response.status_code == 200, response.text row = await OcspSnapshot.get(ca_id=ca["id"]) assert row.dirty is True @async_test async def test_existing_revocation_marks_snapshot_dirty(self) -> None: from dataclasses import replace from unittest.mock import patch from kca import ocsp from kca.models import OcspSnapshot async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) ca_key = await _key(client, vault_id, "existing dirty ca") ca = await _cert(client, vault_id, ca_key["id"], "Existing Dirty CA", ca=True) leaf_key = await _key(client, vault_id, "existing dirty leaf") leaf = await _cert( client, vault_id, leaf_key["id"], "Existing Dirty Leaf", signer_id=ca["id"] ) configured = replace(ocsp.settings, ocsp_s3_bucket="test-bucket") with patch.object(ocsp, "settings", configured): response = await client.post( f"/api/vaults/{vault_id}/certificates/{leaf['id']}/revoke", json={"reason": "superseded"}, ) assert response.status_code == 200, response.text await OcspSnapshot.filter(ca_id=ca["id"]).update(dirty=False) response = await client.post( f"/api/vaults/{vault_id}/certificates/{leaf['id']}/revoke", json={"reason": "superseded"}, ) assert response.status_code == 200, response.text row = await OcspSnapshot.get(ca_id=ca["id"]) assert row.dirty is True @async_test async def test_manual_revocation_changes_mark_snapshot_dirty(self) -> None: from dataclasses import replace from unittest.mock import patch from kca import ocsp from kca.models import OcspSnapshot async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) ca_key = await _key(client, vault_id, "manual dirty ca") ca = await _cert(client, vault_id, ca_key["id"], "Manual Dirty CA", ca=True) configured = replace(ocsp.settings, ocsp_s3_bucket="test-bucket") with patch.object(ocsp, "settings", configured): await OcspSnapshot.create(ca_id=ca["id"], dirty=False) response = await client.post( f"/api/vaults/{vault_id}/certificates/{ca['id']}/revocations", json={"serial_hex": "1234", "reason": "keyCompromise"}, ) assert response.status_code == 201, response.text revocation = response.json() row = await OcspSnapshot.get(ca_id=ca["id"]) assert row.dirty is True await OcspSnapshot.filter(ca_id=ca["id"]).update(dirty=False) response = await client.patch( f"/api/vaults/{vault_id}/certificates/{ca['id']}/revocations/{revocation['id']}", json={"reason": "superseded"}, ) assert response.status_code == 200, response.text row = await OcspSnapshot.get(ca_id=ca["id"]) assert row.dirty is True await OcspSnapshot.filter(ca_id=ca["id"]).update(dirty=False) response = await client.delete( f"/api/vaults/{vault_id}/certificates/{ca['id']}/revocations/{revocation['id']}" ) assert response.status_code == 204, response.text row = await OcspSnapshot.get(ca_id=ca["id"]) assert row.dirty is True @async_test async def test_publisher_preserves_changes_during_publication(self) -> None: from dataclasses import replace from unittest.mock import patch from kca import ocsp from kca.models import CertificateRevocation, OcspSnapshot async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) ca_key = await _key(client, vault_id, "concurrent publish ca") ca = await _cert(client, vault_id, ca_key["id"], "Concurrent Publish CA", ca=True) leaf_key = await _key(client, vault_id, "concurrent publish leaf") leaf = await _cert( client, vault_id, leaf_key["id"], "Concurrent Publish Leaf", signer_id=ca["id"] ) published = [] async def fake_publish(publish_ca, snapshot): published.append(snapshot) if len(published) == 1: await CertificateRevocation.create( ca_id=publish_ca.id, serial_hex=leaf["serial_hex"], revoked_at=datetime.now(timezone.utc), reason="keyCompromise", ) await ocsp.mark_ocsp_dirty(publish_ca.id) return ocsp.object_key(str(publish_ca.id)) configured = replace(ocsp.settings, ocsp_s3_bucket="test-bucket") with patch.object(ocsp, "settings", configured), patch.object( ocsp, "publish_snapshot", fake_publish ): await ocsp.mark_ocsp_dirty(ca["id"]) async def ensure_context(): return None mixin = ocsp.OcspPublisherMixin(1, ensure_context) await mixin.run_once() row = await OcspSnapshot.get(ca_id=ca["id"]) assert row.dirty is True await mixin.run_once() assert len(published) == 2 first_entries = {entry["serial_hex"]: entry for entry in published[0]["entries"]} second_entries = {entry["serial_hex"]: entry for entry in published[1]["entries"]} serial = leaf["serial_hex"].lower() assert first_entries[serial]["status"] == "valid" assert second_entries[serial]["status"] == "revoked" assert second_entries[serial]["reason"] == "keyCompromise" row = await OcspSnapshot.get(ca_id=ca["id"]) assert row.dirty is False @async_test async def test_publisher_marks_dirty_and_uploads(self) -> None: from dataclasses import replace from unittest.mock import patch from kca import ocsp from kca.models import OcspSnapshot class FakeS3: def __init__(self): self.calls = [] def put_object(self, **kwargs): self.calls.append(kwargs) fake = FakeS3() async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) ca_key = await _key(client, vault_id, "ocsp publish ca") ca = await _cert(client, vault_id, ca_key["id"], "OCSP Publish CA", ca=True) configured = replace( ocsp.settings, ocsp_s3_bucket="test-bucket", ocsp_s3_prefix="ocsp", ) with patch.object(ocsp, "settings", configured), patch.object( ocsp, "_s3_client", lambda: fake ): await ocsp.mark_ocsp_dirty(ca["id"]) row = await OcspSnapshot.get(ca_id=ca["id"]) assert row.dirty is True async def ensure_context(): return None mixin = ocsp.OcspPublisherMixin(1, ensure_context) await mixin.run_once() assert len(fake.calls) == 1 call = fake.calls[0] assert call["Bucket"] == "test-bucket" assert call["Key"] == f"ocsp/{ca['id']}.json" assert b"kca-ocsp-status/v1" in call["Body"] row = await OcspSnapshot.get(ca_id=ca["id"]) assert row.dirty is False assert row.published_at is not None @async_test async def test_imported_ca_cannot_overwrite_other_vault_snapshot(self) -> None: import json from dataclasses import replace from unittest.mock import patch from kca import ocsp from kca.models import Certificate, OcspSnapshot class FakeS3: def __init__(self): self.objects = {} def put_object(self, **kwargs): self.objects[kwargs["Key"]] = kwargs["Body"] fake = FakeS3() async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) ca_key = await _key(client, vault_id, "target ca") ca = await _cert(client, vault_id, ca_key["id"], "Target CA", ca=True) leaf_key = await _key(client, vault_id, "target leaf") leaf = await _cert( client, vault_id, leaf_key["id"], "Target Leaf", signer_id=ca["id"] ) response = await client.post( f"/api/vaults/{vault_id}/certificates/{leaf['id']}/revoke", json={"reason": "keyCompromise"}, ) assert response.status_code == 200, response.text exported = await client.post( f"/api/vaults/{vault_id}/certificates/{ca['id']}/export", json={"format": "pem"}, ) assert exported.status_code == 200, exported.text public_ca = exported.json()["content_base64"] configured = replace( ocsp.settings, ocsp_s3_bucket="test-bucket", ocsp_s3_prefix="ocsp", ) with patch.object(ocsp, "settings", configured), patch.object( ocsp, "_s3_client", lambda: fake ): target = await Certificate.get(id=ca["id"]) target_key = await ocsp.publish_snapshot(target) with oidc_user("bob"): attacker_vault_id = await _vault(client) payload = {"filename": "target-ca.crt", "content_base64": public_ca} response = await client.post( f"/api/vaults/{attacker_vault_id}/import", json=payload ) assert response.status_code == 200, response.text assert len(response.json()["imported"]) == 1 imported = await Certificate.get( vault_id=attacker_vault_id, fingerprint_sha256=ca["fingerprint_sha256"], ) assert imported.id != target.id assert imported.fingerprint_sha256 == target.fingerprint_sha256 row = await OcspSnapshot.get(ca_id=imported.id) assert row.dirty is True row.dirty = False await row.save(update_fields=["dirty", "updated_at"]) response = await client.post( f"/api/vaults/{attacker_vault_id}/import", json=payload ) assert response.status_code == 200, response.text assert len(response.json()["skipped"]) == 1 row = await OcspSnapshot.get(ca_id=imported.id) assert row.dirty is True imported_key = await ocsp.publish_snapshot(imported) assert target_key == f"ocsp/{ca['id']}.json" assert imported_key == f"ocsp/{imported.id}.json" assert imported_key != target_key assert set(fake.objects) == {target_key, imported_key} target_snapshot = json.loads(fake.objects[target_key]) target_entries = {entry["serial_hex"]: entry for entry in target_snapshot["entries"]} assert target_entries[leaf["serial_hex"].lower()]["status"] == "revoked" imported_snapshot = json.loads(fake.objects[imported_key]) imported_serials = {entry["serial_hex"] for entry in imported_snapshot["entries"]} assert leaf["serial_hex"].lower() not in imported_serials