From c263f967dc52cb06e10048624b8e1628b9d5fcc5 Mon Sep 17 00:00:00 2001 From: Walter Oggioni Date: Mon, 26 Oct 2026 12:06:16 +0800 Subject: [PATCH] Fix: chain exports use certificate equality instead of object identity MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The chain export formatters in kca-core used identity checks (e.g. , ) to determine whether to exclude the leaf certificate from the chain. This caused duplicates when the chain was built from separately-parsed certificate objects that were cryptographically identical (same DER/fingerprint) but different Python instances. Changed → for value equality across all affected formats: - _pem_chain() : PEM chain export - PKCS#7 chain : serialization_certificates - PKCS#12 chain : serialize_key_and_certificates - OpenVPN : issuer detection - JWK chain : x5c certificate list Also added a regression test (test_chain_exports_exclude_equivalent_leaf) that reproduces the exact issue: a leaf certificate re-parsed from its DER form is treated as a distinct object, so the old -based checks kept both copies in the output. The new -based checks correctly deduplicate, retain the issuer, and handle OpenVPN's CA selection. --- .../packages/kca-core/src/kca/core/exports.py | 12 +++-- server/packages/kca-core/tests/test_core.py | 51 +++++++++++++++++++ 2 files changed, 58 insertions(+), 5 deletions(-) diff --git a/server/packages/kca-core/src/kca/core/exports.py b/server/packages/kca-core/src/kca/core/exports.py index 05febc1..2b0d445 100644 --- a/server/packages/kca-core/src/kca/core/exports.py +++ b/server/packages/kca-core/src/kca/core/exports.py @@ -62,7 +62,7 @@ def _password_encryption(password: Optional[str]) -> serialization.KeySerializat def _pem_chain(cert: x509.Certificate, chain: Sequence[x509.Certificate]) -> bytes: parts = [certificate_pem(cert)] for item in chain: - if item is cert: + if item == cert: continue parts.append(certificate_pem(item)) return "".join(parts).encode("ascii") @@ -100,7 +100,7 @@ def export_certificate( if fmt in ("pkcs7", "pkcs7-chain", "pkcs7-multi"): certs = [cert] if fmt != "pkcs7": - certs.extend(item for item in chain if item is not cert) + certs.extend(item for item in chain if item != cert) data = pkcs7.serialize_certificates(certs, Encoding.DER) return ExportResult(f"{stem}.p7b", "application/pkcs7-mime", data) if fmt == "der": @@ -108,7 +108,7 @@ def export_certificate( if fmt in ("pkcs12", "pfx", "pkcs12-chain"): if private_key is None: raise ValueError("PKCS#12 export requires the private key") - cas = [item for item in chain if item is not cert] if fmt == "pkcs12-chain" else [] + cas = [item for item in chain if item != cert] if fmt == "pkcs12-chain" else [] data = pkcs12.serialize_key_and_certificates( name=stem.encode(), key=private_key, @@ -120,7 +120,7 @@ def export_certificate( if fmt == "openvpn": if private_key is None: raise ValueError("OpenVPN export requires the private key") - ca_cert = next((item for item in reversed(chain) if item is not cert), None) + ca_cert = next((item for item in reversed(chain) if item != cert), None) ca_pem = certificate_pem(ca_cert) if ca_cert is not None else certificate_pem(cert) key_pem = private_pem(private_key).decode("ascii") text = ( @@ -141,7 +141,9 @@ def export_certificate( jwk["x5t"] = _b64u(hashlib.sha1(der).digest()) jwk["x5t#256"] = _b64u(hashlib.sha256(der).digest()) if fmt == "jwk-chain": - jwk["x5c"] = [base64.b64encode(certificate_der(item)).decode("ascii") for item in [cert, *chain]] + chain_certs = [cert] + chain_certs.extend(item for item in chain if item != cert) + jwk["x5c"] = [base64.b64encode(certificate_der(item)).decode("ascii") for item in chain_certs] return ExportResult(f"{stem}.jwk", "application/json", json.dumps(jwk, indent=2).encode("utf-8")) if fmt == "tls-auth": if not tls_auth_key: diff --git a/server/packages/kca-core/tests/test_core.py b/server/packages/kca-core/tests/test_core.py index 4bf617b..b3e1531 100644 --- a/server/packages/kca-core/tests/test_core.py +++ b/server/packages/kca-core/tests/test_core.py @@ -2,6 +2,7 @@ from __future__ import annotations import base64 +import json import time import unittest from datetime import datetime, timedelta, timezone @@ -478,6 +479,56 @@ class ExportImportTest(unittest.TestCase): items = imports.parse_file(result.data, "export.pfx", "secret") self.assertGreaterEqual(len(items), 2) + def test_chain_exports_exclude_equivalent_leaf(self) -> None: + now = datetime.now(timezone.utc) + leaf = certificates.build_signed( + subject=_name("Export Leaf"), + public_key=self.key.public_key(), + issuer_cert=self.cert, + issuer_private_key=self.key, + serial=7, + not_before=now, + not_after=now + timedelta(days=30), + extension_spec=default_extension_spec(), + ) + leaf_copy = certificates.parse_certificate(certificates.certificate_der(leaf)) + self.assertIsNot(leaf_copy, leaf) + self.assertEqual(leaf_copy, leaf) + chain = [leaf_copy, self.cert] + + for fmt in ("pem-chain", "pkcs7-chain", "pkcs12-chain"): + result = exports.export_certificate( + leaf, + fmt, + name="export", + chain=chain, + private_key=self.key, + password="secret", + ) + items = imports.parse_file(result.data, result.filename, "secret") + exported_certs = [item.obj for item in items if item.kind == "certificate"] + self.assertEqual(len(exported_certs), 2, fmt) + self.assertEqual(sum(item == leaf for item in exported_certs), 1, fmt) + self.assertEqual(sum(item == self.cert for item in exported_certs), 1, fmt) + + result = exports.export_certificate(leaf, "jwk-chain", name="export", chain=chain) + x5c = json.loads(result.data)["x5c"] + exported_der = [base64.b64decode(item) for item in x5c] + self.assertEqual( + exported_der, + [certificates.certificate_der(leaf), certificates.certificate_der(self.cert)], + ) + + result = exports.export_certificate( + leaf, + "openvpn", + name="export", + chain=[self.cert, leaf_copy], + private_key=self.key, + ) + ca_pem = result.data.decode("ascii").split("\n", 1)[1].split("", 1)[0] + self.assertEqual(ca_pem, certificates.certificate_pem(self.cert)) + class CryptoTest(unittest.TestCase): def test_envelope(self) -> None: