Fix cross-vault OCSP snapshot poisoning

server/tests/test_routes_extra.py

@@ -496,8 +496,96 @@ class OcspSnapshotTests(unittest.TestCase):
                 assert len(fake.calls) == 1
                 call = fake.calls[0]
                 assert call["Bucket"] == "test-bucket"
-                assert call["Key"] == f"ocsp/{ca['fingerprint_sha256']}.json"
+                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