From 82d448d8d48bf3439fbca535e5369a13bbdc7e5a Mon Sep 17 00:00:00 2001 From: Walter Oggioni Date: Mon, 26 Oct 2026 12:06:16 +0800 Subject: [PATCH] Fix expired self-signed cert validation: remove bypass, enable X509_STRICT|POLICY_CHECK, extract numeric error codes from pyOpenSSL --- .../kca-core/src/kca/core/certificates.py | 14 +++-- server/packages/kca-core/tests/test_core.py | 53 +++++++++++++++++++ 2 files changed, 62 insertions(+), 5 deletions(-) diff --git a/server/packages/kca-core/src/kca/core/certificates.py b/server/packages/kca-core/src/kca/core/certificates.py index 1b31957..3a727b2 100644 --- a/server/packages/kca-core/src/kca/core/certificates.py +++ b/server/packages/kca-core/src/kca/core/certificates.py @@ -250,14 +250,17 @@ def validation_errors(cert: x509.Certificate, chain: Sequence[x509.Certificate]) we verify the chain once and, for missing issuers, report that explicitly. """ errors: List[Dict[str, Any]] = [] - if is_self_signed(cert): - return errors - if len(chain) < 2: + self_signed = is_self_signed(cert) + if not self_signed and len(chain) < 2: return [{"code": 20, "message": "unable to get local issuer certificate"}] try: from OpenSSL import crypto store = crypto.X509Store() + store.set_flags(crypto.X509StoreFlags.X509_STRICT | crypto.X509StoreFlags.POLICY_CHECK) + converted_cert = crypto.X509.from_cryptography(cert) + if self_signed: + store.add_cert(converted_cert) # The root of the supplied chain is trusted; intermediates are # untrusted inputs, exactly like XCA's trusted/untrusted stacks. for item in chain[1:]: @@ -268,12 +271,13 @@ def validation_errors(cert: x509.Certificate, chain: Sequence[x509.Certificate]) store.add_cert(converted) context = crypto.X509StoreContext( store, - crypto.X509.from_cryptography(cert), + converted_cert, [crypto.X509.from_cryptography(item) for item in chain[1:]], ) context.verify_certificate() except Exception as exc: # noqa: BLE001 - pyOpenSSL raises rich errors - code = getattr(exc, "args", [None])[-1] if getattr(exc, "args", None) else None + verification_errors = getattr(exc, "errors", None) + code = verification_errors[0] if verification_errors else None if not isinstance(code, int): code = None errors.append({"code": code, "message": str(exc)}) diff --git a/server/packages/kca-core/tests/test_core.py b/server/packages/kca-core/tests/test_core.py index 485a0f3..4bf617b 100644 --- a/server/packages/kca-core/tests/test_core.py +++ b/server/packages/kca-core/tests/test_core.py @@ -5,6 +5,7 @@ import base64 import time import unittest from datetime import datetime, timedelta, timezone +from unittest.mock import ANY, patch from cryptography import x509 from cryptography.hazmat.primitives import serialization @@ -243,6 +244,8 @@ class CertificateTest(unittest.TestCase): spec["key_usage"] = {"critical": True, "usages": ["digital_signature", "key_encipherment"]} spec["extended_key_usage"] = {"critical": False, "usages": ["serverAuth"]} spec["subject_alt_name"] = {"critical": False, "entries": [{"type": "dns", "value": "example.com"}]} + spec["subject_key_identifier"] = {"critical": False} + spec["authority_key_identifier"] = {"critical": False} leaf = certificates.build_signed( subject=_name("example.com"), public_key=leaf_key.public_key(), @@ -260,6 +263,56 @@ class CertificateTest(unittest.TestCase): self.assertEqual(parsed.serial_number, 2) self.assertEqual(certificates.find_issuer(parsed, [ca]), ca) self.assertEqual(certificates.validation_errors(parsed, [parsed, ca]), []) + self.assertEqual( + certificates.validation_errors(parsed, [parsed]), + [{"code": 20, "message": "unable to get local issuer certificate"}], + ) + + def test_self_signed_validation(self) -> None: + key = keys.generate_key("ec", curve="secp256r1") + now = datetime.now(timezone.utc) + valid = certificates.build_self_signed( + subject=_name("Valid Root CA"), + private_key=key, + serial=6, + not_before=now - timedelta(days=1), + not_after=now + timedelta(days=30), + extension_spec=_ca_spec(), + ) + expired = certificates.build_self_signed( + subject=_name("Expired Root CA"), + private_key=key, + serial=7, + not_before=now - timedelta(days=30), + not_after=now - timedelta(days=1), + extension_spec=_ca_spec(), + ) + + self.assertEqual(certificates.validation_errors(valid, [valid]), []) + self.assertEqual( + certificates.validation_errors(expired, [expired]), + [{"code": 10, "message": "certificate has expired"}], + ) + + def test_validation_sets_strict_flags(self) -> None: + from OpenSSL import crypto + + key = keys.generate_key("ec", curve="secp256r1") + now = datetime.now(timezone.utc) + cert = certificates.build_self_signed( + subject=_name("Strict Root CA"), + private_key=key, + serial=8, + not_before=now - timedelta(days=1), + not_after=now + timedelta(days=30), + extension_spec=_ca_spec(), + ) + flags = crypto.X509StoreFlags.X509_STRICT | crypto.X509StoreFlags.POLICY_CHECK + + with patch.object(crypto.X509Store, "set_flags", autospec=True) as set_flags: + self.assertEqual(certificates.validation_errors(cert, [cert]), []) + + set_flags.assert_called_once_with(ANY, flags) def test_renew_spec(self) -> None: key = keys.generate_key("rsa", 2048)