Fix expired self-signed cert validation: remove bypass, enable X509_STRICT|POLICY_CHECK, extract numeric error codes from pyOpenSSL
This commit is contained in:
@@ -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)})
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user