#!/usr/bin/env python3 """Local-only route comparison: curl 8.x versus Python Requests. Starts a marker target and a marker proxy only on 127.0.0.1 ephemeral ports. The proxy returns its own marker and never forwards. Every child process gets an allowlisted process environment and an empty temporary HOME. The parent environment is not copied, and curl skips its default curlrc. Native platform proxy settings are neither changed nor isolated: results cover this tested machine configuration only, and the Requests child fails closed before resolving or connecting to any non-loopback host. This fixture covers HTTP destinations and HTTP proxies only. It deliberately makes no HTTPS, CONNECT, SOCKS, DNS, authentication, performance, or real-exit claim. Run it with an interpreter that already has `requests` installed: python3 curl-python-proxy-route-checks.py """ import http.server import json import os import shutil import socket import subprocess import sys import tempfile import threading from contextlib import contextmanager TARGET_MARKER = "TARGET_LOOPBACK_MARKER" PROXY_MARKER = "PROXY_LOOPBACK_MARKER" TIMEOUT_SECONDS = 4 REQUEST_LINES = [] REQUEST_LINES_LOCK = threading.Lock() class MarkerHandler(http.server.BaseHTTPRequestHandler): role = "target" def do_GET(self): # noqa: N802 - handler API spelling with REQUEST_LINES_LOCK: REQUEST_LINES.append({"role": self.role, "requestline": self.requestline}) body = (TARGET_MARKER if self.role == "target" else PROXY_MARKER).encode() self.send_response(200) self.send_header("Content-Type", "text/plain") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def log_message(self, _format, *_args): pass def server(role): handler = type("%sHandler" % role.title(), (MarkerHandler,), {"role": role}) httpd = http.server.ThreadingHTTPServer(("127.0.0.1", 0), handler) thread = threading.Thread(target=httpd.serve_forever, daemon=True) thread.start() return httpd def only_loopback(host, allowed_ports): if host != "127.0.0.1" or int(allowed_ports) <= 0: raise RuntimeError("fixture refuses a non-loopback endpoint") @contextmanager def refuse_nonloopback_connects(allowed_ports): """Fail closed before unexpected DNS or network use in the Requests child.""" original_connect = socket.socket.connect original_getaddrinfo = socket.getaddrinfo def guarded_getaddrinfo(host, *args, **kwargs): if host not in ("127.0.0.1", None): raise RuntimeError("fixture blocked non-loopback name resolution") return original_getaddrinfo(host, *args, **kwargs) def guarded_connect(sock, address): host, port = address[:2] if host != "127.0.0.1" or port not in allowed_ports: raise RuntimeError("fixture blocked non-loopback connection") return original_connect(sock, address) socket.getaddrinfo = guarded_getaddrinfo socket.socket.connect = guarded_connect try: yield finally: socket.getaddrinfo = original_getaddrinfo socket.socket.connect = original_connect def child_env(home, updates): # Intentionally does not inherit os.environ. env = {"HOME": home, "PATH": os.defpath, "LC_ALL": "C"} env.update(updates) return env def curl_request(curl_bin, url, home, env_updates, extra_args=()): result = subprocess.run( [curl_bin, "-q", "--silent", "--show-error", "--fail", "--max-time", str(TIMEOUT_SECONDS), *extra_args, url], env=child_env(home, env_updates), text=True, capture_output=True, check=False, ) if result.returncode: raise RuntimeError("curl failed with code %d" % result.returncode) return result.stdout.strip() def request_get(url, home, env_updates, target_port, proxy_port, trust_env=True, proxies=None): command = [sys.executable, __file__, "--requests-child", url, str(target_port), str(proxy_port), "1" if trust_env else "0", json.dumps(proxies)] result = subprocess.run(command, env=child_env(home, env_updates), text=True, capture_output=True, check=False) if result.returncode: raise RuntimeError("Requests child failed with code %d: %s" % (result.returncode, result.stderr.strip())) return json.loads(result.stdout)["marker"] def request_child(url, target_port, proxy_port, trust_env, proxies): import requests with refuse_nonloopback_connects({target_port, proxy_port}): with requests.Session() as session: session.trust_env = trust_env response = session.get(url, timeout=TIMEOUT_SECONDS, proxies=proxies, allow_redirects=False) response.raise_for_status() print(json.dumps({"marker": response.text.strip()})) def assert_marker(case, client, actual, expected): if actual != expected: raise AssertionError("%s/%s expected %s, got %s" % (case, client, expected, actual)) def pop_request_line(case, expected_role, expected_target): with REQUEST_LINES_LOCK: if not REQUEST_LINES: raise AssertionError("%s did not reach a local marker server" % case) observed = REQUEST_LINES.pop(0) if observed["role"] != expected_role: raise AssertionError("%s expected %s, got %s" % (case, expected_role, observed["role"])) prefix = "GET %s HTTP/" % expected_target if not observed["requestline"].startswith(prefix): raise AssertionError("%s expected request form %r, got %r" % (case, prefix, observed["requestline"])) return observed["requestline"] def main(): if len(sys.argv) > 1 and sys.argv[1] == "--requests-child": _, _, url, target_port, proxy_port, trust_env, proxies_json = sys.argv request_child(url, int(target_port), int(proxy_port), trust_env == "1", json.loads(proxies_json)) return curl_bin = shutil.which("curl", path=os.defpath) if not curl_bin: raise SystemExit("curl is required") try: import requests except ImportError: raise SystemExit("requests is required; no dependency was installed by this fixture") target = server("target") proxy = server("proxy") target_url = "http://127.0.0.1:%d/route" % target.server_port proxy_url = "http://127.0.0.1:%d" % proxy.server_port only_loopback("127.0.0.1", target.server_port) only_loopback("127.0.0.1", proxy.server_port) rows = [] # Each row is (name, env, curl extra args, Requests trust_env, per-request proxies, # expected curl, expected Requests, ambient parent proxy). cases = [ ("no_proxy_env_in_child", {}, (), True, None, TARGET_MARKER, TARGET_MARKER, False), ("uppercase_HTTP_PROXY", {"HTTP_PROXY": proxy_url}, (), True, None, TARGET_MARKER, PROXY_MARKER, False), ("lowercase_http_proxy", {"http_proxy": proxy_url}, (), True, None, PROXY_MARKER, PROXY_MARKER, False), ("NO_PROXY_bypasses", {"http_proxy": proxy_url, "NO_PROXY": "127.0.0.1"}, (), True, None, TARGET_MARKER, TARGET_MARKER, False), ("requests_trust_env_false", {"http_proxy": proxy_url}, (), False, None, PROXY_MARKER, TARGET_MARKER, False), ("explicit_proxy_over_NO_PROXY", {"NO_PROXY": "127.0.0.1"}, ("--proxy", proxy_url,), True, {"http": proxy_url}, TARGET_MARKER, PROXY_MARKER, False), ("sanitized_child_ignores_parent_proxy", {}, (), True, None, TARGET_MARKER, TARGET_MARKER, True), ] try: with tempfile.TemporaryDirectory(prefix="curl-requests-route-") as home: curl_version = subprocess.check_output([curl_bin, "--version"], env=child_env(home, {}), text=True).splitlines()[0] import urllib3 versions = {"curl": curl_version, "python": sys.version.split()[0], "requests": requests.__version__, "urllib3": urllib3.__version__} with refuse_nonloopback_connects({target.server_port, proxy.server_port}): for name, env_updates, curl_args, trust_env, proxies, expected_curl, expected_requests, ambient_parent_proxy in cases: original_ambient = os.environ.get("http_proxy") if ambient_parent_proxy: os.environ["http_proxy"] = proxy_url try: curl_marker = curl_request(curl_bin, target_url, home, env_updates, curl_args) curl_line = pop_request_line(name + "/curl", "proxy" if curl_marker == PROXY_MARKER else "target", target_url if curl_marker == PROXY_MARKER else "/route") requests_marker = request_get(target_url, home, env_updates, target.server_port, proxy.server_port, trust_env, proxies) requests_line = pop_request_line(name + "/requests", "proxy" if requests_marker == PROXY_MARKER else "target", target_url if requests_marker == PROXY_MARKER else "/route") finally: if ambient_parent_proxy: if original_ambient is None: os.environ.pop("http_proxy", None) else: os.environ["http_proxy"] = original_ambient assert_marker(name, "curl", curl_marker, expected_curl) assert_marker(name, "requests", requests_marker, expected_requests) rows.append({"case": name, "curl": curl_marker, "requests": requests_marker, "curl_request_line": curl_line, "requests_request_line": requests_line}) finally: target.shutdown(); proxy.shutdown() target.server_close(); proxy.server_close() print(json.dumps({"result": "PASS", "scope": "HTTP loopback only", "versions": versions, "cases": rows}, indent=2)) if __name__ == "__main__": main()