Fix OCSP snapshots for orphaned revocations

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: