diff --git a/tests/proxy.py b/tests/proxy.py index 094c5345b..ac215e719 100644 --- a/tests/proxy.py +++ b/tests/proxy.py @@ -1,11 +1,12 @@ import contextlib import os +from pathlib import Path import socket import subprocess +import sys +import tempfile import time -import psutil - import pytest from tests.assertions import assert_no_proxy_request, wait_for, wait_for_stdout @@ -30,100 +31,82 @@ def cleanup_proxy_env_vars(): os.environ.pop("https_proxy", None) -def _get_process_tree(proc): - """Return a list of psutil.Process for proc and all its descendants.""" - procs = [proc] - try: - procs.extend(proc.children(recursive=True)) - except (psutil.NoSuchProcess, psutil.AccessDenied): - pass - return procs - - -def _discover_listening_port(process, timeout=10): - """Use psutil to discover which port the process (or any of its children) - is listening on. On Windows, pip-installed mitmdump is a launcher that - spawns Python child processes, so the actual listener lives in a - descendant, not the top-level PID.""" +def _wait_for_output_file(process, output_file, timeout=10): deadline = time.monotonic() + timeout - proc = psutil.Process(process.pid) while time.monotonic() < deadline: if process.poll() is not None: + stdout, _ = process.communicate(timeout=1) raise RuntimeError( - f"mitmdump exited with code {process.returncode} before listening" - ) - # Collect the process and all its children (pip-installed mitmdump on - # Windows spawns child python.exe processes that do the actual work). - tree = _get_process_tree(proc) - - listeners = [] - for p in tree: - try: - listeners.extend( - conn - for conn in p.net_connections(kind="tcp") - if conn.status == psutil.CONN_LISTEN + "test proxy exited with code {} before listening:\n{}".format( + process.returncode, stdout.decode("utf-8", errors="replace") ) - except (psutil.NoSuchProcess, psutil.AccessDenied, OSError): - continue - - if listeners: - assert ( - len(listeners) == 1 - ), f"Expected mitmdump to listen on exactly one port, got: {listeners}" - return listeners[0].laddr.port - time.sleep(0.2) + ) + try: + port = Path(output_file).read_text(encoding="ascii") + except OSError: + port = "" + if port: + return int(port) + time.sleep(0.05) raise TimeoutError( - f"mitmdump (pid {process.pid}) did not start listening within {timeout}s" + f"test proxy (pid {process.pid}) did not start listening within {timeout}s" ) -def start_mitmdump( +def start_proxy( proxy_type, proxy_auth: str = None, listen_host: str = "127.0.0.1", retries: int = 3 ): - """Start mitmdump on a free port. Returns (process, port). - Retries up to `retries` times if mitmdump fails to start listening.""" + """Start the stdlib test proxy on a free port. Returns (process, port).""" + proxy_server = Path(__file__).with_name("proxy_server.py") for attempt in range(1, retries + 1): + output = tempfile.NamedTemporaryFile(delete=False) + output_file = output.name + output.close() + try: + os.unlink(output_file) + except OSError: + pass + proxy_command = [ - "mitmdump", - "--set", - f"listen_host={listen_host}", - "--listen-port", + sys.executable, + "-u", + str(proxy_server), + "--type", + proxy_type, + "--listen-host", + listen_host, + "--port", "0", + "--output", + output_file, ] - if proxy_type == "socks5-proxy": - proxy_command += ["--mode", "socks5"] - if proxy_auth: - proxy_command += ["-v", "--proxyauth", proxy_auth] - - proxy_env = os.environ.copy() - proxy_env["PYTHONUNBUFFERED"] = "1" + proxy_command += ["--proxy-auth", proxy_auth] proxy_process = subprocess.Popen( proxy_command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - env=proxy_env, ) try: - port = _discover_listening_port(proxy_process) + port = _wait_for_output_file(proxy_process, output_file) return proxy_process, port except (TimeoutError, RuntimeError) as e: proxy_process.kill() proxy_process.wait() if attempt < retries: - print(f"mitmdump attempt {attempt}/{retries} failed, retrying: {e}") + print(f"test proxy attempt {attempt}/{retries} failed, retrying: {e}") continue - else: - pytest.fail(str(e)) - except Exception: - proxy_process.terminate() - proxy_process.wait() - raise - pytest.fail("start_mitmdump: all retries exhausted") + pytest.fail(str(e)) + finally: + try: + os.unlink(output_file) + except OSError: + pass + + pytest.fail("start_proxy: all retries exhausted") def proxy_test_finally( @@ -139,7 +122,7 @@ def proxy_test_finally( if proxy_process: try: - # Give mitmdump some time to get a response from the mock server + # Give the proxy some time to get a response from the mock server. assert wait_for( lambda: len(httpserver.log) >= expected_httpserver_logsize, timeout ) diff --git a/tests/proxy_server.py b/tests/proxy_server.py new file mode 100644 index 000000000..2a3d52c0a --- /dev/null +++ b/tests/proxy_server.py @@ -0,0 +1,373 @@ +"""Simple, lightweight, dependency-free mitmdump replacement for proxy tests.""" + +import argparse +import base64 +import select +import socket +import sys +import threading +from urllib.parse import urlsplit + +BUFFER_SIZE = 64 * 1024 +HEADER_LIMIT = 1024 * 1024 +SOCKS_VERSION = 5 + + +class ProxyError(Exception): + pass + + +def _recv_exact(sock, size): + data = b"" + while len(data) < size: + chunk = sock.recv(size - len(data)) + if not chunk: + raise ProxyError("unexpected EOF") + data += chunk + return data + + +def _recv_until(sock, marker, limit=HEADER_LIMIT): + data = b"" + while marker not in data: + chunk = sock.recv(BUFFER_SIZE) + if not chunk: + raise ProxyError("unexpected EOF") + data += chunk + if len(data) > limit: + raise ProxyError("header too large") + return data + + +def _parse_headers(header_lines): + headers = [] + for line in header_lines: + if not line: + continue + name, _, value = line.partition(":") + headers.append((name.strip(), value.strip())) + return headers + + +def _get_header(headers, name): + name = name.lower() + for key, value in headers: + if key.lower() == name: + return value + return None + + +def _send_proxy_auth_required(client, method): + response = ( + b"HTTP/1.1 407 Proxy Authentication Required\r\n" + b'Proxy-Authenticate: Basic realm="sentry-native-test"\r\n' + b"Content-Length: 0\r\n" + b"Connection: close\r\n" + b"\r\n" + ) + client.sendall(response) + print(f"{method} 407 Proxy Authentication Required", flush=True) + + +def _auth_matches(headers, proxy_auth): + if not proxy_auth: + return True + expected = "Basic " + base64.b64encode(proxy_auth.encode("utf-8")).decode("ascii") + return _get_header(headers, "Proxy-Authorization") == expected + + +def _read_http_body(client, headers, initial): + transfer_encoding = _get_header(headers, "Transfer-Encoding") + if transfer_encoding and "chunked" in transfer_encoding.lower(): + return _read_chunked_body(client, initial) + + content_length = _get_header(headers, "Content-Length") + if content_length is None: + return initial + + expected = int(content_length) + body = initial + while len(body) < expected: + chunk = client.recv(min(BUFFER_SIZE, expected - len(body))) + if not chunk: + raise ProxyError("unexpected EOF while reading request body") + body += chunk + return body + + +def _read_chunked_body(client, initial): + body = b"" + pending = initial + while True: + while b"\r\n" not in pending: + chunk = client.recv(BUFFER_SIZE) + if not chunk: + raise ProxyError("unexpected EOF while reading chunk size") + pending += chunk + + line, pending = pending.split(b"\r\n", 1) + body += line + b"\r\n" + size_text = line.split(b";", 1)[0] + chunk_size = int(size_text, 16) + if chunk_size == 0: + while True: + while b"\r\n" not in pending: + chunk = client.recv(BUFFER_SIZE) + if not chunk: + raise ProxyError("unexpected EOF while reading trailers") + pending += chunk + + line, pending = pending.split(b"\r\n", 1) + body += line + b"\r\n" + if not line: + return body + + need = chunk_size + 2 + while len(pending) < need: + chunk = client.recv(BUFFER_SIZE) + if not chunk: + raise ProxyError("unexpected EOF while reading chunk") + pending += chunk + + body += pending[:need] + pending = pending[need:] + + +def _read_http_response(server): + data = _recv_until(server, b"\r\n\r\n") + header, body = data.split(b"\r\n\r\n", 1) + lines = header.decode("iso-8859-1").split("\r\n") + status_line = lines[0] + headers = _parse_headers(lines[1:]) + + transfer_encoding = _get_header(headers, "Transfer-Encoding") + if transfer_encoding and "chunked" in transfer_encoding.lower(): + body = _read_chunked_body(server, body) + return header + b"\r\n\r\n" + body, status_line + + content_length = _get_header(headers, "Content-Length") + if content_length is not None: + expected = int(content_length) + while len(body) < expected: + chunk = server.recv(min(BUFFER_SIZE, expected - len(body))) + if not chunk: + break + body += chunk + else: + while True: + chunk = server.recv(BUFFER_SIZE) + if not chunk: + break + body += chunk + + return header + b"\r\n\r\n" + body, status_line + + +def _forward_http(client, proxy_auth): + data = _recv_until(client, b"\r\n\r\n") + header, initial_body = data.split(b"\r\n\r\n", 1) + lines = header.decode("iso-8859-1").split("\r\n") + method, target, version = lines[0].split(" ", 2) + headers = _parse_headers(lines[1:]) + + if not _auth_matches(headers, proxy_auth): + _send_proxy_auth_required(client, method) + return + + url = urlsplit(target) + if not url.scheme or not url.hostname: + host = _get_header(headers, "Host") + if not host: + raise ProxyError("missing absolute URL and Host header") + target_host, target_port = _split_host_port(host, 80) + path = target + else: + target_host = url.hostname + target_port = url.port or 80 + path = url.path or "/" + if url.query: + path += "?" + url.query + + body = _read_http_body(client, headers, initial_body) + outbound_headers = [ + (name, value) + for name, value in headers + if name.lower() not in {"proxy-authorization", "proxy-connection"} + ] + request = ( + f"{method} {path} {version}\r\n".encode("ascii") + + b"".join( + f"{name}: {value}\r\n".encode("iso-8859-1") + for name, value in outbound_headers + ) + + b"\r\n" + + body + ) + + with socket.create_connection((target_host, target_port), timeout=10) as server: + server.sendall(request) + response, status_line = _read_http_response(server) + client.sendall(response) + + status = " ".join(status_line.split(" ", 2)[1:]) + print(f"{method} {status}", flush=True) + + +def _split_host_port(host, default_port): + if host.startswith("["): + end = host.find("]") + address = host[1:end] + port = int(host[end + 2 :]) if host[end + 1 :].startswith(":") else default_port + return address, port + + if ":" in host: + address, port = host.rsplit(":", 1) + return address, int(port) + return host, default_port + + +def _send_socks_reply(client, status, bind_host="0.0.0.0", bind_port=0): + try: + socket.inet_pton(socket.AF_INET, bind_host) + atyp = 1 + address = socket.inet_aton(bind_host) + except OSError: + atyp = 4 + address = socket.inet_pton(socket.AF_INET6, bind_host) + client.sendall( + bytes([SOCKS_VERSION, status, 0, atyp]) + address + bind_port.to_bytes(2, "big") + ) + + +def _read_socks_address(client): + atyp = _recv_exact(client, 1)[0] + if atyp == 1: + host = socket.inet_ntoa(_recv_exact(client, 4)) + elif atyp == 3: + length = _recv_exact(client, 1)[0] + host = _recv_exact(client, length).decode("idna") + elif atyp == 4: + host = socket.inet_ntop(socket.AF_INET6, _recv_exact(client, 16)) + else: + raise ProxyError(f"unsupported SOCKS address type {atyp}") + port = int.from_bytes(_recv_exact(client, 2), "big") + return host, port + + +def _forward_socks5(client): + version, method_count = _recv_exact(client, 2) + if version != SOCKS_VERSION: + raise ProxyError(f"unsupported SOCKS version {version}") + methods = _recv_exact(client, method_count) + if 0 not in methods: + client.sendall(bytes([SOCKS_VERSION, 0xFF])) + return + client.sendall(bytes([SOCKS_VERSION, 0])) + + version, command, _ = _recv_exact(client, 3) + if version != SOCKS_VERSION or command != 1: + _send_socks_reply(client, 7) + return + + host, port = _read_socks_address(client) + try: + server = socket.create_connection((host, port), timeout=10) + except OSError: + _send_socks_reply(client, 5) + return + + with server: + bind_host, bind_port = server.getsockname()[:2] + _send_socks_reply(client, 0, bind_host, bind_port) + _relay_tunnel(client, server) + + +def _relay_tunnel(client, server): + readable_sockets = [client, server] + request_method = None + response_status = None + logged = False + client_buffer = b"" + server_buffer = b"" + + while readable_sockets: + readable, _, _ = select.select(readable_sockets, [], [], 30) + if not readable: + return + for source in readable: + target = server if source is client else client + data = source.recv(BUFFER_SIZE) + if not data: + readable_sockets.remove(source) + try: + target.shutdown(socket.SHUT_WR) + except OSError: + pass + continue + target.sendall(data) + + if logged: + continue + + if source is client and request_method is None: + client_buffer += data + if b"\r\n" in client_buffer: + line = client_buffer.split(b"\r\n", 1)[0] + request_method = line.decode("iso-8859-1").split(" ", 1)[0] + elif source is server and response_status is None: + server_buffer += data + if b"\r\n" in server_buffer: + line = server_buffer.split(b"\r\n", 1)[0] + parts = line.decode("iso-8859-1").split(" ", 2) + response_status = " ".join(parts[1:]) + + if request_method and response_status: + print(f"{request_method} {response_status}", flush=True) + logged = True + + +def _handle_client(client, proxy_type, proxy_auth): + with client: + try: + if proxy_type == "socks5-proxy": + _forward_socks5(client) + else: + _forward_http(client, proxy_auth) + except Exception as exc: + print(f"proxy error: {exc}", file=sys.stderr, flush=True) + + +def serve(proxy_type, listen_host, port, proxy_auth, output): + family = socket.AF_INET6 if ":" in listen_host else socket.AF_INET + with socket.socket(family, socket.SOCK_STREAM) as listener: + listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + listener.bind((listen_host, port)) + listener.listen() + + with open(output, "w", encoding="ascii") as f: + f.write(str(listener.getsockname()[1])) + + while True: + client, _ = listener.accept() + thread = threading.Thread( + target=_handle_client, + args=(client, proxy_type, proxy_auth), + daemon=True, + ) + thread.start() + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--type", choices=("http-proxy", "socks5-proxy"), required=True) + parser.add_argument("--listen-host", default="127.0.0.1") + parser.add_argument("--port", type=int, default=0) + parser.add_argument("--proxy-auth") + parser.add_argument("--output", required=True) + args = parser.parse_args() + + serve(args.type, args.listen_host, args.port, args.proxy_auth, args.output) + + +if __name__ == "__main__": + main() diff --git a/tests/requirements.txt b/tests/requirements.txt index a75d1d9bc..007213320 100644 --- a/tests/requirements.txt +++ b/tests/requirements.txt @@ -6,8 +6,6 @@ msgpack==1.0.8 pytest-xdist==3.8.0 clang-format==20.1.5 pywin32==308; sys_platform == "win32" -# mitmproxy requires OpenSSL to build on Windows ARM64, skip it there -mitmproxy==12.2.2; platform_machine != "ARM64" psutil==7.1.1 # For E2E tests that call Sentry API requests==2.33.0 diff --git a/tests/test_integration_crashpad.py b/tests/test_integration_crashpad.py index cb2b0aa69..0773639f8 100644 --- a/tests/test_integration_crashpad.py +++ b/tests/test_integration_crashpad.py @@ -1,5 +1,4 @@ import os -import shutil import subprocess import sys import time @@ -24,7 +23,7 @@ from .proxy import ( setup_proxy_env_vars, cleanup_proxy_env_vars, - start_mitmdump, + start_proxy, proxy_test_finally, ) from .assertions import ( @@ -67,7 +66,7 @@ def test_crashpad_capture(cmake, httpserver): def _setup_crashpad_proxy_test(cmake, httpserver, proxy): if proxy: - proxy_process, port = start_mitmdump(proxy) + proxy_process, port = start_proxy(proxy) else: proxy_process, port = None, None @@ -82,9 +81,6 @@ def _setup_crashpad_proxy_test(cmake, httpserver, proxy): def test_crashpad_crash_proxy_env(cmake, httpserver): - if not shutil.which("mitmdump"): - pytest.skip("mitmdump is not installed") - proxy_process = None # store the proxy process to terminate it later try: env, proxy_process, tmp_path, port = _setup_crashpad_proxy_test( @@ -109,9 +105,6 @@ def test_crashpad_crash_proxy_env(cmake, httpserver): def test_crashpad_crash_proxy_env_port_incorrect(cmake, httpserver): - if not shutil.which("mitmdump"): - pytest.skip("mitmdump is not installed") - proxy_process = None # store the proxy process to terminate it later try: env, proxy_process, tmp_path, port = _setup_crashpad_proxy_test( @@ -136,9 +129,6 @@ def test_crashpad_crash_proxy_env_port_incorrect(cmake, httpserver): def test_crashpad_proxy_set_empty(cmake, httpserver): - if not shutil.which("mitmdump"): - pytest.skip("mitmdump is not installed") - proxy_process = None # store the proxy process to terminate it later try: env, proxy_process, tmp_path, port = _setup_crashpad_proxy_test( @@ -166,9 +156,6 @@ def test_crashpad_proxy_set_empty(cmake, httpserver): def test_crashpad_proxy_https_not_http(cmake, httpserver): - if not shutil.which("mitmdump"): - pytest.skip("mitmdump is not installed") - proxy_process = None # store the proxy process to terminate it later # we start the proxy but expect it to remain unused (dsn is http, so shouldn't use https proxy) try: @@ -208,9 +195,6 @@ def test_crashpad_proxy_https_not_http(cmake, httpserver): ) @pytest.mark.parametrize("proxy_running", [True, False]) def test_crashpad_crash_proxy(cmake, httpserver, run_args, proxy_running): - if not shutil.which("mitmdump"): - pytest.skip("mitmdump is not installed") - proxy_process = None # store the proxy process to terminate it later expected_logsize = 0 diff --git a/tests/test_integration_proxy.py b/tests/test_integration_proxy.py index 7dac03a09..2eced91a6 100644 --- a/tests/test_integration_proxy.py +++ b/tests/test_integration_proxy.py @@ -1,5 +1,4 @@ import os -import shutil import sys import pytest @@ -14,7 +13,7 @@ from .conditions import has_http from .proxy import ( closed_port, - start_mitmdump, + start_proxy, proxy_test_finally, ) @@ -25,7 +24,7 @@ def _setup_http_proxy_test( cmake, httpserver, proxy, proxy_auth=None, listen_host="127.0.0.1" ): if proxy: - proxy_process, port = start_mitmdump(proxy, proxy_auth, listen_host=listen_host) + proxy_process, port = start_proxy(proxy, proxy_auth, listen_host=listen_host) else: proxy_process, port = None, None @@ -40,9 +39,6 @@ def _setup_http_proxy_test( def test_proxy_from_env(cmake, httpserver): - if not shutil.which("mitmdump"): - pytest.skip("mitmdump is not installed") - proxy_process = None # store the proxy process to terminate it later try: env, proxy_process, tmp_path, port = _setup_http_proxy_test( @@ -63,9 +59,6 @@ def test_proxy_from_env(cmake, httpserver): def test_proxy_from_env_port_incorrect(cmake, httpserver): - if not shutil.which("mitmdump"): - pytest.skip("mitmdump is not installed") - proxy_process = None # store the proxy process to terminate it later try: env, proxy_process, tmp_path, port = _setup_http_proxy_test( @@ -87,9 +80,6 @@ def test_proxy_from_env_port_incorrect(cmake, httpserver): def test_proxy_auth(cmake, httpserver): - if not shutil.which("mitmdump"): - pytest.skip("mitmdump is not installed") - proxy_process = None # store the proxy process to terminate it later try: env, proxy_process, tmp_path, port = _setup_http_proxy_test( @@ -116,9 +106,6 @@ def test_proxy_auth(cmake, httpserver): def test_proxy_auth_incorrect(cmake, httpserver): - if not shutil.which("mitmdump"): - pytest.skip("mitmdump is not installed") - proxy_process = None # store the proxy process to terminate it later try: env, proxy_process, tmp_path, port = _setup_http_proxy_test( @@ -145,9 +132,6 @@ def test_proxy_auth_incorrect(cmake, httpserver): def test_proxy_ipv6(cmake, httpserver): - if not shutil.which("mitmdump"): - pytest.skip("mitmdump is not installed") - proxy_process = None # store the proxy process to terminate it later try: env, proxy_process, tmp_path, port = _setup_http_proxy_test( @@ -166,9 +150,6 @@ def test_proxy_ipv6(cmake, httpserver): def test_proxy_set_empty(cmake, httpserver): - if not shutil.which("mitmdump"): - pytest.skip("mitmdump is not installed") - proxy_process = None # store the proxy process to terminate it later try: env, proxy_process, tmp_path, port = _setup_http_proxy_test( @@ -190,9 +171,6 @@ def test_proxy_set_empty(cmake, httpserver): def test_proxy_https_not_http(cmake, httpserver): - if not shutil.which("mitmdump"): - pytest.skip("mitmdump is not installed") - proxy_process = None # store the proxy process to terminate it later try: # we start the proxy but expect it to remain unused (dsn is http, so shouldn't use https proxy) @@ -227,9 +205,6 @@ def test_proxy_https_not_http(cmake, httpserver): ) @pytest.mark.parametrize("proxy_running", [True, False]) def test_capture_proxy(cmake, httpserver, run_args, proxy_running): - if not shutil.which("mitmdump"): - pytest.skip("mitmdump is not installed") - proxy_process = None # store the proxy process to terminate it later expected_logsize = 0 diff --git a/tests/test_proxy_server.py b/tests/test_proxy_server.py new file mode 100644 index 000000000..ab3cb0b16 --- /dev/null +++ b/tests/test_proxy_server.py @@ -0,0 +1,220 @@ +import base64 +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +import queue +import socket +import subprocess +import sys +import threading +import time + +import pytest + + +class _RequestHandler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def do_GET(self): + if self.path == "/through-socks": + self.server.response_gate.wait(timeout=5) + self._respond(b"") + + def do_POST(self): + body = self.rfile.read(int(self.headers.get("Content-Length", "0"))) + self._respond(body) + + def _respond(self, body): + self.server.requests.put( + (self.command, self.path, dict(self.headers.items()), body) + ) + self.send_response(200) + self.send_header("Content-Length", "2") + self.send_header("Connection", "close") + self.end_headers() + self.wfile.write(b"OK") + + def log_message(self, format, *args): + pass + + +@pytest.fixture +def target_server(): + server = ThreadingHTTPServer(("127.0.0.1", 0), _RequestHandler) + server.requests = queue.Queue() + server.response_gate = threading.Event() + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + + yield server + + server.shutdown() + server.server_close() + thread.join(timeout=5) + + +@pytest.fixture +def start_proxy(tmp_path): + processes = [] + proxy_server = Path(__file__).with_name("proxy_server.py") + + def start(proxy_type, proxy_auth=None): + output = tmp_path / "proxy-{}.port".format(len(processes)) + command = [ + sys.executable, + "-u", + str(proxy_server), + "--type", + proxy_type, + "--listen-host", + "127.0.0.1", + "--port", + "0", + "--output", + str(output), + ] + if proxy_auth: + command += ["--proxy-auth", proxy_auth] + + process = subprocess.Popen( + command, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) + processes.append(process) + + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + if process.poll() is not None: + stdout, _ = process.communicate(timeout=1) + pytest.fail( + "test proxy exited before listening:\n{}".format( + stdout.decode("utf-8", errors="replace") + ) + ) + try: + port = output.read_text(encoding="ascii") + except OSError: + port = "" + if port: + return int(port) + time.sleep(0.01) + + pytest.fail("test proxy did not start listening") + + yield start + + for process in processes: + if process.poll() is None: + process.terminate() + try: + process.communicate(timeout=5) + except subprocess.TimeoutExpired: + process.kill() + process.communicate(timeout=5) + + +def _exchange(port, request): + with socket.create_connection(("127.0.0.1", port), timeout=5) as client: + client.sendall(request) + response = b"" + while True: + chunk = client.recv(64 * 1024) + if not chunk: + return response + response += chunk + + +def _proxy_request(target_port, proxy_auth=None): + headers = [ + "POST http://127.0.0.1:{}/envelope?test=1 HTTP/1.1".format(target_port), + "Host: 127.0.0.1:{}".format(target_port), + "Content-Length: 4", + "Connection: close", + ] + if proxy_auth: + encoded = base64.b64encode(proxy_auth.encode("utf-8")).decode("ascii") + headers.append("Proxy-Authorization: Basic {}".format(encoded)) + return ("\r\n".join(headers) + "\r\n\r\ndata").encode("ascii") + + +def _recv_exact(sock, size): + data = b"" + while len(data) < size: + chunk = sock.recv(size - len(data)) + assert chunk + data += chunk + return data + + +def test_http_forwarding(start_proxy, target_server): + port = start_proxy("http-proxy") + target_port = target_server.server_address[1] + + response = _exchange(port, _proxy_request(target_port)) + + assert response.startswith(b"HTTP/1.1 200") + assert response.endswith(b"OK") + method, path, _, body = target_server.requests.get(timeout=5) + assert (method, path, body) == ("POST", "/envelope?test=1", b"data") + + +def test_proxy_auth_rejection(start_proxy, target_server): + port = start_proxy("http-proxy", "user:password") + target_port = target_server.server_address[1] + + response = _exchange(port, _proxy_request(target_port, "wrong:password")) + + assert response.startswith(b"HTTP/1.1 407") + assert target_server.requests.empty() + + +def test_proxy_auth_forwarding(start_proxy, target_server): + port = start_proxy("http-proxy", "user:password") + target_port = target_server.server_address[1] + + response = _exchange(port, _proxy_request(target_port, "user:password")) + + assert response.startswith(b"HTTP/1.1 200") + assert response.endswith(b"OK") + _, _, headers, body = target_server.requests.get(timeout=5) + assert "Proxy-Authorization" not in headers + assert body == b"data" + + +def test_socks5_tunneling(start_proxy, target_server): + port = start_proxy("socks5-proxy") + target_port = target_server.server_address[1] + + with socket.create_connection(("127.0.0.1", port), timeout=5) as client: + client.sendall(b"\x05\x01\x00") + assert _recv_exact(client, 2) == b"\x05\x00" + + client.sendall( + b"\x05\x01\x00\x01" + + socket.inet_aton("127.0.0.1") + + target_port.to_bytes(2, "big") + ) + assert _recv_exact(client, 2) == b"\x05\x00" + _recv_exact(client, 8) + + request = ( + "GET /through-socks HTTP/1.1\r\n" + "Host: 127.0.0.1:{}\r\n" + "Connection: close\r\n" + "\r\n" + ).format(target_port) + client.sendall(request.encode("ascii")) + client.shutdown(socket.SHUT_WR) + target_server.response_gate.set() + + response = b"" + while True: + chunk = client.recv(64 * 1024) + if not chunk: + break + response += chunk + + assert response.startswith(b"HTTP/1.1 200") + assert response.endswith(b"OK") + method, path, _, body = target_server.requests.get(timeout=5) + assert (method, path, body) == ("GET", "/through-socks", b"")