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

Fix issuer links for PEM certificate chain imports

Browse Source
This commit is contained in:
KCA
2026-10-26 13:35:28 +00:00
parent 462e68e4e5
commit 55b8b5d14b
2 changed files with 116 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/packages/kca-app/src/kca/service.py
+23 -8
View File
@@ -1284,6 +1284,12 @@ async def relink_certificates(vault: Vault) -> None:
for cert in certs:
by_hash.setdefault(cert.subject_hash, []).append(cert)
for cert in certs:
if cert.issuer_id == cert.id:
# Older imports could accidentally self-link a certificate. That
# is never a valid issuer reference in this model; clear it so the
# real issuer can be discovered below.
cert.issuer_id = None
await cert.save(update_fields=["issuer_id"])
if cert.issuer_id is not None:
continue
candidates = by_hash.get(cert.issuer_hash, [])
@@ -1337,7 +1343,15 @@ async def import_parsed_items(
imported: List[Dict[str, Any]] = []
skipped: List[Dict[str, Any]] = []
# Map XCA item ids to created rows so issuer/key links can be restored.
# Ordinary PEM/DER imports have no XCA ids; do not collapse them all onto
# a single ``None`` entry, which would create bogus key/issuer references.
id_map: Dict[int, Any] = {}
def remember_xca_row(item: ParsedItem, row: Any) -> None:
item_id = item.extra.get("xca_item_id")
if item_id is not None:
id_map[item_id] = row
# Keys first, then certificates, requests, CRLs, templates.
ordered = sorted(chosen, key=lambda item: {"key": 0, "certificate": 1, "request": 2, "crl": 3, "template": 4}.get(item.kind, 9))
for item in ordered:
@@ -1355,7 +1369,7 @@ async def import_parsed_items(
comment=str(item.extra.get("comment") or ""),
source=str(item.extra.get("source") or "imported"),
)
id_map[item.extra.get("xca_item_id")] = key
remember_xca_row(item, key)
imported.append({"kind": "key", "id": str(key.id), "name": key.name})
elif item.kind == "certificate":
info = cert_core.certificate_info(item.obj)
@@ -1364,7 +1378,7 @@ async def import_parsed_items(
continue
key_model = None
xca_key_id = item.extra.get("xca_key_id")
if xca_key_id in id_map:
if xca_key_id is not None and xca_key_id in id_map:
key_model = id_map[xca_key_id]
if key_model is None:
key_model = await Key.get_or_none(vault_id=vault.id, public_key_sha256=hashlib.sha256(info.public_key_der).hexdigest())
@@ -1398,7 +1412,7 @@ async def import_parsed_items(
extension_search_text(cert_core.certificate_extensions(item.obj)),
),
)
id_map[item.extra.get("xca_item_id")] = row
remember_xca_row(item, row)
if info.is_ca:
authority_extra = item.extra.get("authority") or {}
authority = await Authority.create(
@@ -1419,7 +1433,8 @@ async def import_parsed_items(
if await Request.filter(vault_id=vault.id, request_sha256=info.fingerprint_sha256).exists():
skipped.append({"kind": item.kind, "name": item.name, "reason": "duplicate"})
continue
key_model = id_map.get(item.extra.get("xca_key_id"))
xca_key_id = item.extra.get("xca_key_id")
key_model = id_map.get(xca_key_id) if xca_key_id is not None else None
if key_model is None:
key_model = await Key.get_or_none(vault_id=vault.id, public_key_sha256=hashlib.sha256(info.public_key_der).hexdigest())
row = await Request.create(
@@ -1442,7 +1457,7 @@ async def import_parsed_items(
extension_search_text(req_core.csr_extensions(item.obj)),
),
)
id_map[item.extra.get("xca_item_id")] = row
remember_xca_row(item, row)
imported.append({"kind": "request", "id": str(row.id), "name": row.name})
elif item.kind == "crl":
info = crl_core.crl_info(item.obj)
@@ -1463,7 +1478,7 @@ async def import_parsed_items(
signature_algorithm=info.signature_algorithm,
search_text=search_text(item.name, info.issuer.rfc4514_string(), info.crl_number),
)
id_map[item.extra.get("xca_item_id")] = row
remember_xca_row(item, row)
imported.append({"kind": "crl", "id": str(row.id), "name": row.name})
elif item.kind == "template":
template = item.obj
@@ -1476,7 +1491,7 @@ async def import_parsed_items(
settings=template.get("settings") or {},
search_text=search_text(template.get("name"), template.get("comment")),
)
id_map[item.extra.get("xca_item_id")] = row
remember_xca_row(item, row)
imported.append({"kind": "template", "id": str(row.id), "name": row.name})
except Exception as exc: # noqa: BLE001
skipped.append({"kind": item.kind, "name": item.name, "reason": str(exc)})
@@ -1486,7 +1501,7 @@ async def import_parsed_items(
continue
row = id_map.get(item.extra.get("xca_item_id"))
issuer_id = item.extra.get("xca_issuer_id")
if row is not None and issuer_id in id_map:
if row is not None and issuer_id is not None and issuer_id in id_map:
issuer = id_map[issuer_id]
if isinstance(issuer, Certificate):
row.issuer_id = issuer.id
server/tests/test_routes_extra.py
+93
View File
@@ -320,6 +320,99 @@ class RouteExtraTests(unittest.TestCase):
assert response.status_code == 200
assert len(response.json()["skipped"]) == 1
@async_test
async def test_pem_chain_import_restores_issuer_links(self) -> None:
from cryptography import x509
from cryptography.x509.oid import NameOID
from kca.core import certificates as cert_core
from kca.core import extensions as ext_core
from kca.core import keys as key_core
from kca.models import Certificate
def name(cn: str) -> x509.Name:
return x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, cn)])
now = datetime.now(timezone.utc)
root_key = key_core.generate_key("rsa", 2048)
intermediate_key = key_core.generate_key("rsa", 2048)
leaf_key = key_core.generate_key("rsa", 2048)
ca_spec = {"basic_constraints": {"critical": True, "ca": True}}
root = cert_core.build_self_signed(
subject=name("Chain Root"),
private_key=root_key,
serial=101,
not_before=now,
not_after=now + timedelta(days=3650),
extension_spec=ca_spec,
signature_hash="sha256",
)
intermediate = cert_core.build_certificate(
subject=name("Chain Intermediate"),
public_key=key_core.public_part(intermediate_key),
issuer_cert=root,
issuer_private_key=root_key,
serial=102,
not_before=now,
not_after=now + timedelta(days=1825),
extensions=ext_core.build_extensions(
ca_spec,
public_key=key_core.public_part(intermediate_key),
subject_name=name("Chain Intermediate"),
issuer_cert=root,
),
signature_hash="sha256",
)
leaf = cert_core.build_certificate(
subject=name("Chain Leaf"),
public_key=key_core.public_part(leaf_key),
issuer_cert=intermediate,
issuer_private_key=intermediate_key,
serial=103,
not_before=now,
not_after=now + timedelta(days=365),
extensions=[],
signature_hash="sha256",
)
# Deliberately not in issuer-first order.
pem = cert_core.certificate_pem(leaf) + cert_core.certificate_pem(root) + cert_core.certificate_pem(intermediate)
async with api_client() as client:
with oidc_user("alice"):
vault_id = await _vault(client)
response = await client.post(
f"/api/vaults/{vault_id}/import", json={"filename": "chain.pem", "pem": pem}
)
assert response.status_code == 200, response.text
result = response.json()
assert len(result["imported"]) == 3, result
assert result["skipped"] == [], result
async def certificates_by_name() -> dict[str, dict]:
response = await client.get(f"/api/vaults/{vault_id}/certificates")
assert response.status_code == 200, response.text
return {cert["name"]: cert for cert in response.json()["results"]}
certs = await certificates_by_name()
root_row = certs["Chain Root"]
intermediate_row = certs["Chain Intermediate"]
leaf_row = certs["Chain Leaf"]
assert root_row["issuer_id"] is None
assert intermediate_row["issuer_id"] == root_row["id"]
assert leaf_row["issuer_id"] == intermediate_row["id"]
assert all(cert["issuer_id"] != cert["id"] for cert in certs.values())
# Re-importing duplicates also repairs self-links left by older imports.
model = await Certificate.get(id=leaf_row["id"])
model.issuer_id = model.id
await model.save(update_fields=["issuer_id"])
response = await client.post(
f"/api/vaults/{vault_id}/import", json={"filename": "chain.pem", "pem": pem}
)
assert response.status_code == 200, response.text
certs = await certificates_by_name()
assert certs["Chain Leaf"]["issuer_id"] == intermediate_row["id"]
@async_test
async def test_settings_and_templates(self) -> None:
async with api_client() as client: