Patched code sample

import datetime
import ssl
import socket
import threading
import tempfile
import os
from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives import serialization

# This code demonstrates that the fix for CVE-2023-49083 (described by the user
# with a placeholder CVE number) is working in a patched version of the
# `cryptography` library, which is used by Python's `ssl` module.
#
# The vulnerability:
# A certificate chain is created as follows:
#   Root CA -> Intermediate CA -> Leaf Certificate
#
# 1. The Intermediate CA has a Name Constraint extension that *excludes* the
#    hostname `bar.example.com`.
# 2. The Leaf Certificate is a wildcard certificate for `*.example.com`.
#
# In a VULNERABLE version, connecting to `bar.example.com` would SUCCEED because
# the name constraint was not checked against the peer hostname, only against
# SANs in lower certificates.
#
# In a FIXED version, connecting to `bar.example.com` must FAIL with a
# certificate verification error, because the peer hostname is now correctly
# checked against the name constraint from the Intermediate CA.
#
# This script sets up this exact scenario and attempts a TLS handshake. It will
# only "succeed" (i.e., print the success message) if the handshake properly
# fails with the expected SSLCertVerificationError, proving the fix is active.


def generate_private_key():
    """Generates an RSA private key."""
    return rsa.generate_private_key(public_exponent=65537, key_size=2048)


def generate_cert_chain():
    """
    Generates the certificate chain described in the vulnerability.
    Returns keys and certs as PEM-encoded bytes.
    """
    # 1. Generate Root CA
    root_key = generate_private_key()
    subject = issuer = x509.Name([
        x509.NameAttribute(NameOID.COMMON_NAME, "Test Root CA")
    ])
    root_cert = (
        x509.CertificateBuilder()
        .subject_name(subject)
        .issuer_name(issuer)
        .public_key(root_key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(datetime.datetime.utcnow())
        .not_valid_after(datetime.datetime.utcnow() + datetime.timedelta(days=1))
        .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True)
        .sign(root_key, hashes.SHA256())
    )

    # 2. Generate Intermediate CA with the critical Name Constraint
    intermediate_key = generate_private_key()
    subject = x509.Name([
        x509.NameAttribute(NameOID.COMMON_NAME, "Test Intermediate CA")
    ])
    # The crucial name constraint excluding bar.example.com
    name_constraints = x509.NameConstraints(
        permitted_subtrees=None,
        excluded_subtrees=[x509.DNSName("bar.example.com")],
    )
    intermediate_cert = (
        x509.CertificateBuilder()
        .subject_name(subject)
        .issuer_name(root_cert.subject)
        .public_key(intermediate_key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(datetime.datetime.utcnow())
        .not_valid_after(datetime.datetime.utcnow() + datetime.timedelta(days=1))
        .add_extension(name_constraints, critical=True)
        .add_extension(x509.BasicConstraints(ca=True, path_length=0), critical=True)
        .sign(root_key, hashes.SHA256())
    )

    # 3. Generate Leaf Certificate for *.example.com
    leaf_key = generate_private_key()
    subject = x509.Name([
        x509.NameAttribute(NameOID.COMMON_NAME, "leaf.example.com")
    ])
    leaf_cert = (
        x509.CertificateBuilder()
        .subject_name(subject)
        .issuer_name(intermediate_cert.subject)
        .public_key(leaf_key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(datetime.datetime.utcnow())
        .not_valid_after(datetime.datetime.utcnow() + datetime.timedelta(days=1))
        .add_extension(
            x509.SubjectAlternativeName([x509.DNSName("*.example.com")]),
            critical=False,
        )
        .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
        .sign(intermediate_key, hashes.SHA256())
    )

    # Serialize all to PEM
    pem_root_key = root_key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption(),
    )
    pem_root_cert = root_cert.public_bytes(serialization.Encoding.PEM)
    pem_intermediate_key = intermediate_key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption(),
    )
    pem_intermediate_cert = intermediate_cert.public_bytes(serialization.Encoding.PEM)
    pem_leaf_key = leaf_key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption(),
    )
    pem_leaf_cert = leaf_cert.public_bytes(serialization.Encoding.PEM)

    return {
        "root_ca": {"key": pem_root_key, "cert": pem_root_cert},
        "intermediate_ca": {"key": pem_intermediate_key, "cert": pem_intermediate_cert},
        "leaf": {"key": pem_leaf_key, "cert": pem_leaf_cert},
    }

def run_server(server_sock, cert_file, key_file):
    """Server thread target."""
    try:
        conn, _ = server_sock.accept()
        conn.do_handshake()
    except ssl.SSLError as e:
        # Server might see an error if client aborts, this is okay.
        pass
    finally:
        server_sock.close()

def main():
    """Main demonstration logic."""
    certs = generate_cert_chain()
    
    # Use temporary files to store certs for ssl.SSLContext
    with tempfile.NamedTemporaryFile(mode='w+', delete=False) as root_ca_file, \
         tempfile.NamedTemporaryFile(mode='w+', delete=False) as full_chain_file, \
         tempfile.NamedTemporaryFile(mode='w+', delete=False) as leaf_key_file:
        
        root_ca_file.write(certs["root_ca"]["cert"].decode())
        root_ca_file_path = root_ca_file.name

        # The server needs to present the full chain (leaf + intermediate)
        full_chain_file.write(certs["leaf"]["cert"].decode())
        full_chain_file.write(certs["intermediate_ca"]["cert"].decode())
        full_chain_file_path = full_chain_file.name

        leaf_key_file.write(certs["leaf"]["key"].decode())
        leaf_key_file_path = leaf_key_file.name

    try:
        # 1. Server Context: uses the leaf key and the full chain
        server_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        server_context.load_cert_chain(certfile=full_chain_file_path, keyfile=leaf_key_file_path)

        # 2. Client Context: trusts our custom Root CA
        client_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        client_context.load_verify_locations(cafile=root_ca_file_path)
        client_context.verify_mode = ssl.CERT_REQUIRED
        client_context.check_hostname = True

        # 3. Set up a local socket pair for communication
        server_sock, client_sock = socket.socketpair()

        # 4. Start the server in a separate thread
        server_thread = threading.Thread(
            target=run_server,
            args=(server_context.wrap_socket(server_sock, server_side=True), full_chain_file_path, leaf_key_file_path)
        )
        server_thread.start()

        # 5. Client attempts to connect, triggering the vulnerability check
        client_exception = None
        try:
            # Here is the critical test: we connect to the excluded name.
            # In a fixed system, this handshake MUST fail.
            hostname_to_test = "bar.example.com"
            print(f"Attempting TLS handshake with peer name '{hostname_to_test}'...")
            print("This name is in the Intermediate CA's 'excludedSubtrees'.")
            
            with client_context.wrap_socket(
                client_sock, server_side=False, server_hostname=hostname_to_test
            ) as s:
                s.do_handshake()
                
        except ssl.SSLCertVerificationError as e:
            client_exception = e
            print("\nSUCCESS: Handshake correctly failed as expected.")
            print(f"Received expected error: {e}")
        except ssl.SSLError as e:
            client_exception = e
            print(f"\nERROR: Handshake failed with an unexpected SSL error: {e}")
        except Exception as e:
            client_exception = e
            print(f"\nERROR: An unexpected exception occurred: {e}")
        finally:
            server_thread.join()

        if client_exception and isinstance(client_exception, ssl.SSLCertVerificationError):
            print("\nDEMONSTRATION COMPLETE: The vulnerability appears to be patched.")
            print("The system correctly blocked a certificate for an excluded DNS name.")
        else:
            print("\nDEMONSTRATION FAILED: The vulnerability may still be present.")
            print("The system incorrectly allowed a certificate for an excluded DNS name.")

    finally:
        # Clean up temporary files
        os.unlink(root_ca_file_path)
        os.unlink(full_chain_file_path)
        os.unlink(leaf_key_file_path)


if __name__ == "__main__":
    main()