Function _forward_proxy_with_server

e2e/python/test_sandbox_policy.py:201–252

Return a closure that starts an HTTP server and sends a forward proxy request. The closure starts a minimal HTTP server on the given port inside the sandbox, then sends a plain HTTP forward proxy request (non-CONNECT) through the sandbox proxy and returns the raw response.

()

def _forward_proxy_with_server():
    """Return a closure that starts an HTTP server and sends a forward proxy request.

    The closure starts a minimal HTTP server on the given port inside the sandbox,
    then sends a plain HTTP forward proxy request (non-CONNECT) through the sandbox
    proxy and returns the raw response.
    """

    def fn(proxy_host, proxy_port, target_host, target_port):
        import socket
        import threading
        import time
        from http.server import BaseHTTPRequestHandler, HTTPServer

        class Handler(BaseHTTPRequestHandler):
            def do_GET(self):
                self.send_response(200)
                body = b"forward-proxy-ok"
                self.send_header("Content-Length", str(len(body)))
                self.end_headers()
                self.wfile.write(body)

            def log_message(self, *args):
                pass  # suppress log output

        srv = HTTPServer(("0.0.0.0", int(target_port)), Handler)
        threading.Thread(target=srv.handle_request, daemon=True).start()
        time.sleep(0.5)

        conn = socket.create_connection((proxy_host, int(proxy_port)), timeout=10)
        try:
            req = (
                f"GET http://{target_host}:{target_port}/test HTTP/1.1\r\n"
                f"Host: {target_host}:{target_port}\r\n\r\n"
            )
            conn.sendall(req.encode())
            data = b""
            conn.settimeout(5)
            try:
                while True:
                    chunk = conn.recv(4096)
                    if not chunk:
                        break
                    data += chunk
            except socket.timeout:
                pass
            return data.decode("latin1")
        finally:
            conn.close()
            srv.server_close()

    return fn


def _forward_proxy_raw():