diff --git a/server/packages/kca-app/src/kca/service.py b/server/packages/kca-app/src/kca/service.py index 27f6225..9f16d6d 100644 --- a/server/packages/kca-app/src/kca/service.py +++ b/server/packages/kca-app/src/kca/service.py @@ -608,17 +608,29 @@ def validate_validity(payload: Mapping[str, Any]) -> Tuple[datetime, datetime, b return not_before, not_after, no_expiration +async def _serial_in_use( + vault: Vault, + issuer: Optional[Certificate], + serial: int, + *, + exclude_certificate_id: Optional[UUID] = None, +) -> bool: + text = format(serial, "x") + certificates = Certificate.filter(vault_id=vault.id, serial_hex=text) + if exclude_certificate_id is not None: + certificates = certificates.exclude(id=exclude_certificate_id) + if await certificates.exists(): + return True + return issuer is not None and await CertificateRevocation.filter(ca_id=issuer.id, serial_hex=text).exists() + + async def unique_serial(vault: Vault, issuer: Optional[Certificate], bits: int) -> int: for _ in range(100): serial = secrets.randbits(bits) if serial <= 0: continue - text = format(serial, "x") - if await Certificate.filter(vault_id=vault.id, serial_hex=text).exists(): - continue - if issuer is not None and await CertificateRevocation.filter(ca_id=issuer.id, serial_hex=text).exists(): - continue - return serial + if not await _serial_in_use(vault, issuer, serial): + return serial raise ValidationError("failed to allocate a unique serial number") @@ -743,6 +755,8 @@ async def issue_certificate( raise ValidationError("serial must be an integer or a hexadecimal string") from exc if serial <= 0: serial = await unique_serial(vault, signer, bits) + elif await _serial_in_use(vault, signer, serial): + raise ValidationError("serial number is already in use or revoked by the issuer") signature_hash = payload.get("signature_hash") or settings_obj.get("default_hash", "sha256") if signer is None: @@ -903,7 +917,21 @@ async def renew_certificate( raise ValidationError("the certificate's issuer has no private key") parsed = cert_core.parse_certificate(bytes(cert.cert_der)) keep_serial = bool(payload.get("keep_serial")) - serial = parsed.serial_number if keep_serial else await unique_serial(vault, signer, int(vault_settings(vault).get("serial_len_bits", 64))) + revoke_old = bool(payload.get("revoke_old")) + replace_old = bool(payload.get("replace_old")) + if keep_serial and revoke_old: + raise ValidationError("keep_serial cannot be combined with revoke_old") + if keep_serial: + serial = parsed.serial_number + if await _serial_in_use( + vault, + signer, + serial, + exclude_certificate_id=cert.id if replace_old else None, + ): + raise ValidationError("serial number is already in use or revoked by the issuer") + else: + serial = await unique_serial(vault, signer, int(vault_settings(vault).get("serial_len_bits", 64))) not_before, not_after, no_expiration = validate_validity(payload) check_mandatory_dn(parsed.subject, vault_settings(vault)) signer_private = await load_private_key(signer_key) @@ -955,9 +983,9 @@ async def renew_certificate( ) if info.is_ca: await Authority.create(certificate_id=row.id, crl_days=30) - if payload.get("revoke_old") and signer.id != cert.id: + if revoke_old and signer.id != cert.id: await revoke_certificate(vault, cert, {"reason": payload.get("revocation_reason", "superseded")}) - if payload.get("replace_old"): + if replace_old: await delete_certificate_cascade(vault, cert) await relink_certificates(vault) if signer.id != cert.id: diff --git a/server/tests/test_routes_extra.py b/server/tests/test_routes_extra.py index 9af4e4d..00f6b00 100644 --- a/server/tests/test_routes_extra.py +++ b/server/tests/test_routes_extra.py @@ -368,6 +368,157 @@ class ExtendedRouteTests(unittest.TestCase): 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: