Add server template apply modes, bulk revocation, explicit DN order and detail tabs

server/packages/kca-app/src/kca/routes/certificates.py

@@ -26,6 +26,7 @@ from ..service import (
     chain_for_certificate,
     delete_certificate_cascade,
     generate_crl,
+    bulk_revoke_certificates,
     issue_certificate,
     load_private_key,
     preview_certificate,
@@ -97,6 +98,25 @@ def register(app: KayaApp, kca: "Kca") -> None:
             return
         await send_json(ctx, 201, await serialize_certificate_detail(cert, vault))
 
+    @app.POST("/api/vaults/${vault_id}/certificates/bulk-revoke")
+    @operation(summary="Revoke several certificates", tags=["certificates"])
+    @require_auth(kca.oidc)
+    async def bulk_revoke(ctx: HttpContext, vault_id: str) -> None:
+        vault = await _vault(kca, ctx, vault_id)
+        if vault is None:
+            return
+        try:
+            body = await read_json(ctx)
+        except JsonRequestError as exc:
+            await send_error(ctx, 400, str(exc))
+            return
+        try:
+            result = await bulk_revoke_certificates(vault, body)
+        except (ValidationError, ValueError) as exc:
+            await send_error(ctx, 400, str(exc))
+            return
+        await send_json(ctx, 200, result)
+
     @app.POST("/api/vaults/${vault_id}/certificates/preview")
     @operation(summary="Preview/validate a certificate payload", tags=["certificates"])
     @require_auth(kca.oidc)

server/packages/kca-app/src/kca/service.py

@@ -73,6 +73,17 @@ def merge_settings(current: Mapping[str, Any], patch: Mapping[str, Any]) -> Dict
     if not (32 <= bits <= 160):
         raise ValidationError("serial_len_bits must be between 32 and 160")
     merged["serial_len_bits"] = bits
+    for key in ("explicit_dn", "mandatory_dn"):
+        entries = merged.get(key) or []
+        if not isinstance(entries, list):
+            raise ValidationError(f"{key} must be a list")
+        for entry in entries:
+            if not isinstance(entry, str) or not entry.strip():
+                raise ValidationError(f"{key} entries must be non-empty strings")
+            try:
+                name_core.resolve_oid(entry)
+            except ValueError as exc:
+                raise ValidationError(f"unknown subject attribute {entry!r}") from exc
     return merged
 
 
@@ -965,6 +976,63 @@ async def revoke_certificate(vault: Vault, cert: Certificate, payload: Mapping[s
     return row
 
 
+async def bulk_revoke_certificates(vault: Vault, payload: Mapping[str, Any]) -> Dict[str, Any]:
+    """Revoke several certificates in one request, reporting per-item failures."""
+    raw_ids = payload.get("certificate_ids") or payload.get("ids") or []
+    if not isinstance(raw_ids, list):
+        raise ValidationError("certificate_ids must be a list")
+    seen: set[str] = set()
+    ids: list[str] = []
+    for raw in raw_ids:
+        text = str(raw)
+        if text in seen:
+            continue
+        seen.add(text)
+        ids.append(text)
+    if not ids:
+        raise ValidationError("select at least one certificate")
+    if len(ids) > 200:
+        raise ValidationError("at most 200 certificates can be revoked at once")
+    reason = str(payload.get("reason") or "unspecified")
+    if reason not in {item["name"] for item in crl_core.CRL_REASONS}:
+        raise ValidationError("unknown revocation reason")
+    invalidity = payload.get("invalidity_date")
+    if payload.get("revoked_at"):
+        parse_datetime(payload["revoked_at"], "revoked_at")
+    if invalidity:
+        parse_datetime(invalidity, "invalidity_date")
+    revocation_payload = {
+        "reason": reason,
+        "revoked_at": payload.get("revoked_at"),
+        "invalidity_date": invalidity,
+    }
+    revoked: list[Dict[str, Any]] = []
+    skipped: list[Dict[str, Any]] = []
+    errors: list[Dict[str, Any]] = []
+    for raw_id in ids:
+        parsed_id = parse_uuid(raw_id)
+        if parsed_id is None:
+            errors.append({"id": raw_id, "error": "invalid certificate id"})
+            continue
+        cert = await Certificate.get_or_none(id=parsed_id, vault_id=vault.id)
+        if cert is None:
+            errors.append({"id": raw_id, "error": "certificate not found"})
+            continue
+        existing = None
+        if cert.issuer_id is not None:
+            existing = await CertificateRevocation.get_or_none(ca_id=cert.issuer_id, serial_hex=cert.serial_hex)
+        if existing is not None:
+            skipped.append({"id": str(cert.id), "serial_hex": cert.serial_hex, "reason": "already revoked"})
+            continue
+        try:
+            row = await revoke_certificate(vault, cert, revocation_payload)
+        except (ValidationError, ValueError) as exc:
+            errors.append({"id": str(cert.id), "serial_hex": cert.serial_hex, "error": str(exc)})
+            continue
+        revoked.append({"id": str(cert.id), "serial_hex": cert.serial_hex, "revocation_id": str(row.id)})
+    return {"revoked": revoked, "skipped": skipped, "errors": errors}
+
+
 async def generate_crl(vault: Vault, ca: Certificate, payload: Mapping[str, Any]) -> Crl:
     if not ca.is_ca:
         raise ValidationError("the selected certificate is not a CA")
@@ -1323,6 +1391,7 @@ def _reason_from_bit(bit: int) -> str:
 __all__ = [
     "DEFAULT_VAULT_SETTINGS",
     "ValidationError",
+    "bulk_revoke_certificates",
     "chain_for_certificate",
     "extension_search_text",
     "delete_certificate_cascade",

server/tests/test_routes_extra.py

@@ -152,6 +152,30 @@ class RouteExtraTests(unittest.TestCase):
                 )
                 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:
@@ -303,6 +327,55 @@ class VaultExportTests(unittest.TestCase):
                     names = archive.namelist()
                     assert "manifest.json" in names
                     assert any(name.startswith("certificates/") for name in names)
-                    assert any(name.startswith("keys/") 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"