From 0301cbff6fddaa7d5553812e570e2e3a5ddd6f42 Mon Sep 17 00:00:00 2001 From: Walter Oggioni Date: Tue, 27 Oct 2026 14:09:00 +0000 Subject: [PATCH] Batch certificate summary lookups --- .../kca-app/src/kca/routes/certificates.py | 4 +- server/packages/kca-app/src/kca/service.py | 61 ++++++++++++++-- server/tests/test_routes_extra.py | 71 +++++++++++++++++++ 3 files changed, 129 insertions(+), 7 deletions(-) diff --git a/server/packages/kca-app/src/kca/routes/certificates.py b/server/packages/kca-app/src/kca/routes/certificates.py index a2654ee..7d943cd 100644 --- a/server/packages/kca-app/src/kca/routes/certificates.py +++ b/server/packages/kca-app/src/kca/routes/certificates.py @@ -39,7 +39,7 @@ from ..service import ( save_key_once, search_text, serialize_certificate_detail, - serialize_certificate_summary, + serialize_certificate_summaries, serialize_revocation, serialize_template_summary, serialize_crl_detail, @@ -74,7 +74,7 @@ def register(app: KayaApp, kca: "Kca") -> None: if vault is None: return rows = await Certificate.filter(vault_id=vault.id).order_by("name", "id") - await send_json(ctx, 200, {"results": [await serialize_certificate_summary(c) for c in rows]}) + await send_json(ctx, 200, {"results": await serialize_certificate_summaries(rows)}) @app.POST("/api/vaults/${vault_id}/certificates") @operation(summary="Issue a certificate", tags=["certificates"]) diff --git a/server/packages/kca-app/src/kca/service.py b/server/packages/kca-app/src/kca/service.py index 07ca9c1..5daa144 100644 --- a/server/packages/kca-app/src/kca/service.py +++ b/server/packages/kca-app/src/kca/service.py @@ -220,23 +220,27 @@ async def revoked_for(cert: Certificate) -> Optional[CertificateRevocation]: return await CertificateRevocation.get_or_none(ca_id=cert.issuer_id, serial_hex=cert.serial_hex) -async def serialize_certificate_summary(cert: Certificate) -> Dict[str, Any]: - revoked = await revoked_for(cert) +def _certificate_summary( + cert: Certificate, + revoked: Optional[CertificateRevocation], + has_private_key: bool, +) -> Dict[str, Any]: + parsed = cert_core.parse_certificate(bytes(cert.cert_der)) return { "id": str(cert.id), "name": cert.name, "comment": cert.comment, "source": cert.source, - "subject": name_core.one_line(cert_core.parse_certificate(bytes(cert.cert_der)).subject), + "subject": name_core.one_line(parsed.subject), "issuer_id": str(cert.issuer_id) if cert.issuer_id else None, - "issuer_name": name_core.one_line(cert_core.parse_certificate(bytes(cert.cert_der)).issuer), + "issuer_name": name_core.one_line(parsed.issuer), "serial_hex": cert.serial_hex, "not_before": iso(cert.not_before), "not_after": iso(cert.not_after), "no_expiration": cert.no_expiration, "is_ca": cert.is_ca, "path_length": cert.path_length, - "has_private_key": bool(cert.key_id and await Key.filter(id=cert.key_id, has_private=True).exists()), + "has_private_key": has_private_key, "revoked": revoked is not None, "revoked_at": iso(revoked.revoked_at) if revoked else None, "fingerprint_sha256": cert.fingerprint_sha256, @@ -247,6 +251,52 @@ async def serialize_certificate_summary(cert: Certificate) -> Dict[str, Any]: } +async def serialize_certificate_summary(cert: Certificate) -> Dict[str, Any]: + revoked = await revoked_for(cert) + has_private_key = bool( + cert.key_id and await Key.filter(id=cert.key_id, has_private=True).exists() + ) + return _certificate_summary(cert, revoked, has_private_key) + + +async def serialize_certificate_summaries( + certificates: Sequence[Certificate], +) -> List[Dict[str, Any]]: + """Serialize a certificate list with constant-count revocation/key queries.""" + if not certificates: + return [] + + issuer_ids = {cert.issuer_id for cert in certificates if cert.issuer_id is not None} + serial_hexes = {cert.serial_hex for cert in certificates if cert.issuer_id is not None} + revocations = ( + await CertificateRevocation.filter( + ca_id__in=issuer_ids, + serial_hex__in=serial_hexes, + ) + if issuer_ids + else [] + ) + revoked_by_certificate = { + (revocation.ca_id, revocation.serial_hex): revocation + for revocation in revocations + } + + key_ids = {cert.key_id for cert in certificates if cert.key_id is not None} + private_key_ids = ( + set(await Key.filter(id__in=key_ids, has_private=True).values_list("id", flat=True)) + if key_ids + else set() + ) + return [ + _certificate_summary( + cert, + revoked_by_certificate.get((cert.issuer_id, cert.serial_hex)), + cert.key_id in private_key_ids, + ) + for cert in certificates + ] + + async def chain_for_certificate(cert: Certificate, vault: Vault) -> List[Certificate]: chain = [cert] current = cert @@ -1591,6 +1641,7 @@ __all__ = [ "save_key", "save_key_once", "serialize_certificate_detail", + "serialize_certificate_summaries", "serialize_certificate_summary", "serialize_crl_detail", "serialize_crl_summary", diff --git a/server/tests/test_routes_extra.py b/server/tests/test_routes_extra.py index 49622c0..941b9b0 100644 --- a/server/tests/test_routes_extra.py +++ b/server/tests/test_routes_extra.py @@ -604,6 +604,77 @@ class ExtendedRouteTests(unittest.TestCase): assert response.json()["reason"] == "superseded" +class CertificateListTests(unittest.TestCase): + @async_test + async def test_summaries_batch_revocation_and_private_key_queries(self) -> None: + from unittest.mock import patch + + from kca import service + from kca.models import CertificateRevocation, Key + + async with api_client() as client: + with oidc_user("alice"): + vault_id = await _vault(client) + url = f"/api/vaults/{vault_id}/certificates" + + with ( + patch.object( + CertificateRevocation, + "filter", + wraps=CertificateRevocation.filter, + ) as revocation_filter, + patch.object(Key, "filter", wraps=Key.filter) as key_filter, + ): + response = await client.get(url) + assert response.status_code == 200, response.text + assert response.json()["results"] == [] + revocation_filter.assert_not_called() + key_filter.assert_not_called() + + ca_key = await _key(client, vault_id, "list ca") + ca = await _cert(client, vault_id, ca_key["id"], "List CA", ca=True) + leaf_key = await _key(client, vault_id, "list leaf") + leaf = await _cert( + client, + vault_id, + leaf_key["id"], + "List Leaf", + signer_id=ca["id"], + ) + response = await client.post( + f"/api/vaults/{vault_id}/certificates/{leaf['id']}/revoke", + json={"reason": "superseded"}, + ) + assert response.status_code == 200, response.text + + with ( + patch.object( + CertificateRevocation, + "filter", + wraps=CertificateRevocation.filter, + ) as revocation_filter, + patch.object(Key, "filter", wraps=Key.filter) as key_filter, + patch.object( + service, + "revoked_for", + side_effect=AssertionError("per-certificate revocation query"), + ) as single_revocation, + ): + response = await client.get(url) + + assert response.status_code == 200, response.text + revocation_filter.assert_called_once() + key_filter.assert_called_once() + single_revocation.assert_not_called() + summaries = {row["id"]: row for row in response.json()["results"]} + assert summaries[ca["id"]]["revoked"] is False + assert summaries[ca["id"]]["revoked_at"] is None + assert summaries[ca["id"]]["has_private_key"] is True + assert summaries[leaf["id"]]["revoked"] is True + assert summaries[leaf["id"]]["revoked_at"] is not None + assert summaries[leaf["id"]]["has_private_key"] is True + + class SerialRenewalTests(unittest.TestCase): @async_test async def test_explicit_serial_must_be_unique_and_not_revoked(self) -> None: