woggioni
aee1025015
- Parse and validate from snapshot JSON, reject future/missing timestamps - Enforce snapshot age vs in cache and stale fallback (no lifetime extension on repeat S3 reads) - OCSP: , , - Remove unused from signer - Publisher heartbeats: , republish clean snapshots past interval, preserve dirty-only behavior and concurrency - Config/env/docker/docs updates for heartbeat Rolling upgrade: enable publisher heartbeats before deploying the age-enforcing responder so existing snapshots remain within the age window.
1201 lines
56 KiB
Python
1201 lines
56 KiB
Python
"""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_publisher_heartbeat_republishes_due_clean_snapshot(self) -> None:
|
|
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, "heartbeat due ca")
|
|
ca = await _cert(client, vault_id, ca_key["id"], "Heartbeat Due CA", ca=True)
|
|
previous_published_at = datetime.now(timezone.utc) - timedelta(seconds=31)
|
|
await OcspSnapshot.create(
|
|
ca_id=ca["id"],
|
|
dirty=False,
|
|
published_at=previous_published_at,
|
|
)
|
|
published = []
|
|
|
|
async def fake_publish(publish_ca, snapshot):
|
|
published.append(snapshot)
|
|
return "key"
|
|
|
|
async def ensure_context():
|
|
return None
|
|
|
|
mixin = ocsp.OcspPublisherMixin(
|
|
1, ensure_context, heartbeat_interval_seconds=30
|
|
)
|
|
with patch.object(ocsp, "publish_snapshot", fake_publish):
|
|
await mixin.run_once()
|
|
|
|
assert len(published) == 1
|
|
row = await OcspSnapshot.get(ca_id=ca["id"])
|
|
assert row.dirty is False
|
|
assert row.published_at is not None
|
|
assert row.published_at != previous_published_at
|
|
|
|
@async_test
|
|
async def test_publisher_heartbeat_skips_recent_clean_snapshot(self) -> None:
|
|
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, "heartbeat recent ca")
|
|
ca = await _cert(client, vault_id, ca_key["id"], "Heartbeat Recent CA", ca=True)
|
|
await OcspSnapshot.create(
|
|
ca_id=ca["id"],
|
|
dirty=False,
|
|
published_at=datetime.now(timezone.utc),
|
|
)
|
|
published = []
|
|
|
|
async def fake_publish(publish_ca, snapshot):
|
|
published.append(snapshot)
|
|
return "key"
|
|
|
|
async def ensure_context():
|
|
return None
|
|
|
|
mixin = ocsp.OcspPublisherMixin(
|
|
1, ensure_context, heartbeat_interval_seconds=30
|
|
)
|
|
with patch.object(ocsp, "publish_snapshot", fake_publish):
|
|
await mixin.run_once()
|
|
|
|
assert published == []
|
|
|
|
@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
|