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: