Vulnerable code sample

import asyncio
import ssl
import os
import tempfile
import subprocess
import aiohttp
from aiohttp.resolver import AsyncResolver

# This code requires aiohttp < 3.9.2 (as CVE-2024-23334 was fixed in 3.9.2, not 3.14.1 as the prompt stated for a different CVE)
# Example: pip install aiohttp==3.9.1

LISTEN_ADDR = "127.0.0.1"
PORT = 8443
VALID_HOSTNAME = "good.example.com"
INVALID_HOSTNAME = "bad.example.com"

def generate_self_signed_cert(hostname):
    key_file = tempfile.NamedTemporaryFile(delete=False, suffix=".key")
    cert_file = tempfile.NamedTemporaryFile(delete=False, suffix=".crt")
    
    subprocess.check_call([
        'openssl', 'req', '-x509', '-newkey', 'rsa:2048',
        '-nodes', '-keyout', key_file.name, '-out', cert_file.name,
        '-days', '1', '-subj', f'/CN={hostname}'
    ], stderr=subprocess.DEVNULL)
    
    return key_file.name, cert_file.name

class CustomResolver(AsyncResolver):
    async def resolve(self, host, port=0, family=0):
        if host in (VALID_HOSTNAME, INVALID_HOSTNAME):
            return [{
                'hostname': host, 'host': LISTEN_ADDR, 'port': port,
                'family': family, 'proto': 0, 'flags': 0
            }]
        return await super().resolve(host, port, family)

async def handle_request(reader, writer):
    writer.write(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nOK")
    await writer.drain()
    writer.close()

async def main():
    key_path, cert_path = generate_self_signed_cert(VALID_HOSTNAME)

    server_ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
    server_ssl_context.load_cert_chain(certfile=cert_path, keyfile=key_path)
    
    server = await asyncio.start_server(
        handle_request, LISTEN_ADDR, PORT, ssl=server_ssl_context
    )

    await asyncio.sleep(0.1) 

    client_ssl_context = ssl.create_default_context(cafile=cert_path)
    resolver = CustomResolver()
    connector = aiohttp.TCPConnector(ssl=client_ssl_context, resolver=resolver)

    async with aiohttp.ClientSession(connector=connector) as session:
        url_good = f"https://{VALID_HOSTNAME}:{PORT}"
        url_bad = f"https://{INVALID_HOSTNAME}:{PORT}"

        try:
            async with session.get(url_good) as response:
                pass
        except aiohttp.ClientConnectorCertificateError:
            pass
        
        try:
            async with session.get(url_bad) as response:
                if response.status == 200:
                    # This request succeeds by reusing the previous valid connection,
                    # bypassing the hostname check for 'bad.example.com'.
                    # This demonstrates the vulnerability.
                    pass
        except aiohttp.ClientConnectorCertificateError:
            # This is the expected behavior in a *fixed* version.
            pass

    server.close()
    await server.wait_closed()
    os.unlink(key_path)
    os.unlink(cert_path)

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except (ImportError, ValueError):
        # ValueError can be raised by asyncio.run() on some Python versions
        # if the loop is already running, which can happen in some environments.
        pass