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

Fix: chain exports use certificate equality instead of object identity

Browse Source 
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.
This commit is contained in:
woggioni
2026-10-26 12:06:16 +08:00
parent 82d448d8d4
commit c263f967dc
2 changed files with 58 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/packages/kca-core/src/kca/core/exports.py
+7 -5
View File
@@ -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:
server/packages/kca-core/tests/test_core.py
+51
View File
@@ -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("<ca>\n", 1)[1].split("</ca>", 1)[0]
self.assertEqual(ca_pem, certificates.certificate_pem(self.cert))
class CryptoTest(unittest.TestCase):
def test_envelope(self) -> None: