Patched code sample

This code demonstrates the behavior of the fixed version of pyOpenSSL (version 26.0.0 and later). In vulnerable versions, an unhandled exception in the `set_tlsext_servername_callback` would be ignored, and the connection would be accepted. This script shows the correct, secure behavior where such an exception causes the TLS handshake to be aborted, thus rejecting the connection.

```python
import socket
import threading
import time
import tempfile
import os
from OpenSSL import SSL, crypto

def generate_self_signed_cert():
    """Generates a self-signed cert and key, returning their file paths."""
    key = crypto.PKey()
    key.generate_key(crypto.TYPE_RSA, 2048)

    cert = crypto.X509()
    cert.get_subject().CN = "localhost"
    cert.set_serial_number(1000)
    cert.gmtime_adj_notBefore(0)
    cert.gmtime_adj_notAfter(10 * 365 * 24 * 60 * 60)  # 10 years
    cert.set_issuer(cert.get_subject())
    cert.set_pubkey(key)
    cert.sign(key, 'sha256')

    with tempfile.NamedTemporaryFile(delete=False, suffix=".pem", mode="wb") as cert_file:
        cert_path = cert_file.name
        cert_file.write(crypto.dump_certificate(crypto.FILETYPE_PEM, cert))

    with tempfile.NamedTemporaryFile(delete=False, suffix=".pem", mode="wb") as key_file:
        key_path = key_file.name
        key_file.write(crypto.dump_privatekey(crypto.FILETYPE_PEM, key))

    return cert_path, key_path

def sni_callback_that_raises_exception(connection):
    """
    This callback demonstrates the vulnerability scenario. It raises an
    unhandled exception during SNI processing.

    - Pre-fix (vulnerable): The exception would be ignored, and the connection
      would be accepted, bypassing any security checks intended here.
    - Post-fix (secure, v26.0.0+): The exception correctly causes the TLS
      handshake to be aborted, rejecting the connection.
    """
    print("[Server] SNI callback triggered, now raising an exception.")
    raise ValueError("Simulating a critical error in the SNI callback")

def server_logic(host, port, cert_file, key_file, server_ready_event):
    """Server that uses the problematic SNI callback."""
    context = SSL.Context(SSL.TLS_SERVER_METHOD)
    context.use_privatekey_file(key_file)
    context.use_certificate_file(cert_file)

    # Set the callback that will raise an exception. This is the core of the test.
    context.set_tlsext_servername_callback(sni_callback_that_raises_exception)

    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.bind((host, port))
    sock.listen(1)
    
    server_ready_event.set() # Signal that the server is ready
    print(f"[Server] Listening on {host}:{port}")
    
    try:
        client_sock, addr = sock.accept()
        print(f"[Server] Accepted connection from {addr}")

        ssl_conn = SSL.Connection(context, client_sock)
        
        # In the FIXED version of pyOpenSSL, this handshake will fail because
        # the exception from the SNI callback is now handled by aborting
        # the connection.
        ssl_conn.do_handshake()

        # This line should NOT be reached with a fixed pyOpenSSL version.
        print("\n[Server] !!! VULNERABLE BEHAVIOR !!! Handshake succeeded unexpectedly.")

    except SSL.Error as e:
        # This is the expected, SECURE behavior.
        print(f"\n[Server] >>> CORRECT, FIXED BEHAVIOR <<< Handshake failed as expected due to callback exception.")
        print(f"[Server] SSL.Error: {e}")
        
    finally:
        sock.close()
        print("[Server] Server shut down.")

def client_logic(host, port, ca_cert_file):
    """Client that tries to connect to the server."""
    print("[Client] Attempting to connect...")
    context = SSL.Context(SSL.TLS_CLIENT_METHOD)
    
    # Trust the self-signed certificate for this demonstration
    context.load_verify_locations(ca_cert_file)
    # The callback simply checks if the certificate is OK.
    context.set_verify(SSL.VERIFY_PEER, lambda conn, cert, err, depth, ok: ok)
    
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    ssl_conn = SSL.Connection(context, sock)
    # Set a server name to trigger the SNI callback on the server
    ssl_conn.set_tlsext_host_name(b'example.com') 
    
    try:
        ssl_conn.connect((host, port))
        
        # This handshake should fail because the server will reject the connection.
        ssl_conn.do_handshake()

        # This line should NOT be reached.
        print("\n[Client] !!! VULNERABLE BEHAVIOR !!! Handshake succeeded.")

    except SSL.Error as e:
        # This is the expected result from the client's perspective.
        print(f"\n[Client] >>> CORRECT, FIXED BEHAVIOR <<< Handshake failed as server rejected the connection.")
        print(f"[Client] SSL.Error: {e}")
        
    finally:
        ssl_conn.close()
        print("[Client] Connection closed.")

if __name__ == "__main__":
    HOST, PORT = "127.0.0.1", 8443
    cert_path, key_path = None, None
    
    print("This script demonstrates the fix for CVE-2023-27448 in pyOpenSSL >= 26.0.0.")
    print("Expected outcome: The server and client will report a handshake failure.\n")
    
    try:
        cert_path, key_path = generate_self_signed_cert()
        
        server_ready = threading.Event()
        server_thread = threading.Thread(
            target=server_logic,
            args=(HOST, PORT, cert_path, key_path, server_ready)
        )
        server_thread.daemon = True
        server_thread.start()
        
        if not server_ready.wait(timeout=5):
            raise RuntimeError("Server did not start in time.")
        
        # Give a moment for the server to be fully listening
        time.sleep(0.1)

        client_logic(HOST, PORT, cert_path)

        server_thread.join(timeout=2)
        
    except ImportError:
        print("Error: pyOpenSSL is not installed. Please run 'pip install pyopenssl'.")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")
    finally:
        if cert_path and os.path.exists(cert_path):
            os.remove(cert_path)
        if key_path and os.path.exists(key_path):
            os.remove(key_path)