woggioni/kca
2
0
Fork 0
Code Actions Packages Releases Activity

Reject duplicate serials and make keep_serial renewal safe

Browse Source 
Add shared serial collision checking against existing certificates and
issuer revocations, reuse it for automatic serial allocation, and enforce
it for user-supplied serials so duplicate serials no longer break the
OCSP snapshot.

make keep_serial renewal require replace_old, refuse keep_serial +
revoke_old, and reject serials already revoked by the issuer so revoked
serials cannot be inherited by renewed certificates.
This commit is contained in:
woggioni
2026-10-25 21:23:34 +00:00
parent a8d322776d
commit 4c1bfa4e4d
2 changed files with 188 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/packages/kca-app/src/kca/service.py
+37 -9
View File
@@ -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:
server/tests/test_routes_extra.py
+151
View File
@@ -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: