diff --git a/server/packages/kca-app/src/kca/ocsp.py b/server/packages/kca-app/src/kca/ocsp.py index 2adef97..50b9c83 100644 --- a/server/packages/kca-app/src/kca/ocsp.py +++ b/server/packages/kca-app/src/kca/ocsp.py @@ -42,17 +42,19 @@ def object_key(ca_id: str) -> str: async def snapshot_for_ca(ca: Certificate) -> Dict[str, Any]: """Build the full OCSP status snapshot for certificates issued by ``ca``.""" certs = await Certificate.filter(vault_id=ca.vault_id, issuer_id=ca.id).order_by("serial_hex") # type: ignore[attr-defined] + certs_by_serial = {cert.serial_hex.lower(): cert for cert in certs} revocations = { row.serial_hex.lower(): row for row in await CertificateRevocation.filter(ca_id=ca.id) } entries = [] - for cert in certs: - revocation = revocations.get(cert.serial_hex.lower()) + for serial in sorted(certs_by_serial.keys() | revocations.keys()): + cert = certs_by_serial.get(serial) + revocation = revocations.get(serial) entry: Dict[str, Any] = { - "serial_hex": cert.serial_hex.lower(), + "serial_hex": serial, "status": "revoked" if revocation is not None else "valid", - "not_after": _iso(cert.not_after), + "not_after": _iso(cert.not_after) if cert is not None else None, } if revocation is not None: entry.update( diff --git a/server/packages/kca-app/src/kca/service.py b/server/packages/kca-app/src/kca/service.py index a637f94..de181c6 100644 --- a/server/packages/kca-app/src/kca/service.py +++ b/server/packages/kca-app/src/kca/service.py @@ -1365,7 +1365,7 @@ async def import_parsed_items( await relink_certificates(vault) # Merge the revocations carried by imported CRLs into their CA, as XCA # does when a CRL is imported. - for crl in await Crl.filter(vault_id=vault.id, issuer_id__not=None): + for crl in await Crl.filter(vault_id=vault.id, issuer_id__isnull=False): ca = await Certificate.get_or_none(id=crl.issuer_id, vault_id=vault.id) if ca is None: continue diff --git a/server/tests/test_routes_extra.py b/server/tests/test_routes_extra.py index 46931a8..e20b642 100644 --- a/server/tests/test_routes_extra.py +++ b/server/tests/test_routes_extra.py @@ -427,9 +427,118 @@ class OcspSnapshotTests(unittest.TestCase): 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: