#!/usr/bin/env python3 """Synthetic-only loopback test. Never put real credentials into this fixture. The wrapper records exactly what /bin/sh passes before exec replaces it with curl. The HTTP endpoint acts as a proxy but never forwards traffic or resolves targets. """ import base64 import datetime import http.server import json import os from pathlib import Path import platform import shlex import shutil import subprocess import sys import tempfile import threading USER = "fixture-user" PASSWORD = "synthetic-only-@:$ space '{{literal}}'" ROOT = Path(__file__).resolve().parent if len(sys.argv) > 1 and sys.argv[1] == "--wrapper": Path(os.environ["ARGV_CAPTURE"]).write_text(json.dumps({ "argv": sys.argv[3:], "synthetic_password_in_environment": os.environ.get("PROXY_PASSWORD") == PASSWORD, })) os.execv(sys.argv[2], [sys.argv[2], *sys.argv[3:]]) class Proxy(http.server.BaseHTTPRequestHandler): def log_message(self, *_): pass def do_GET(self): self.server.observed.append({"method": "GET", "target": self.path, "authorization": self.headers.get("Proxy-Authorization")}) self.server.received.set() self.server.release.wait(3) expected = "Basic " + base64.b64encode((USER + ":" + self.server.password).encode()).decode() valid = self.headers.get("Proxy-Authorization") == expected body = b"synthetic-auth-ok\n" if valid else b"synthetic-auth-rejected\n" self.send_response(200 if valid else 407) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def do_CONNECT(self): self.server.observed.append({"method": "CONNECT", "target": self.path, "authorization": self.headers.get("Proxy-Authorization")}) self.server.received.set() self.server.release.wait(3) self.send_response(407) self.send_header("Proxy-Authenticate", 'Basic realm="synthetic"') self.send_header("Content-Length", "0") self.end_headers() def run_case(name, auth, password=PASSWORD, missing=False, target="http://fixture.invalid/check", fail=True, baseline=False, script=None): server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Proxy) server.observed, server.received, server.release = [], threading.Event(), threading.Event() server.password = password thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() with tempfile.TemporaryDirectory(prefix="curl-synthetic-") as temporary: capture = Path(temporary) / "argv.json" # A minimal environment prevents real credentials or inherited proxy configuration reaching children. env = {"PATH": "/usr/bin:/bin", "PROXY_USERNAME": USER, "PROXY_URL": f"http://127.0.0.1:{server.server_port}", "ARGV_CAPTURE": str(capture)} if not missing: env["PROXY_PASSWORD"] = password executable = " ".join(map(shlex.quote, [sys.executable, str(ROOT / "credential_fixture.py"), "--wrapper", CURL])) transfer = " --connect-timeout 10 --max-time 30 --output /dev/null --write-out 'http=%{http_code} connect=%{http_connect} seconds=%{time_total}\\n' " if baseline else " --connect-timeout 2 --max-time 5 " command = ("exec " + executable + " --disable --silent --show-error " + ("--fail " if fail else "") + "--noproxy '' " '--proxy "$PROXY_URL" ' + auth + transfer + shlex.quote(target)) if script is not None: command = script.replace("curl --disable", "exec " + executable + " --disable", 1).replace("http://proxy.example.invalid:8080", env["PROXY_URL"]) child = subprocess.Popen(["/bin/sh", "-c", command], env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) # Observe only this synthetic curl process; no systemwide process or environment dump. received = server.received.wait(1) listing = subprocess.run(["ps", "-ww", "-p", str(child.pid), "-o", "command="], capture_output=True, text=True).stdout.strip() if received else None environment_listing = subprocess.run(["ps", "eww", "-p", str(child.pid), "-o", "command="], capture_output=True, text=True).stdout.strip() if received else None server.release.set() stdout, stderr = child.communicate(timeout=8) arguments = json.loads(capture.read_text()) if capture.exists() else {"argv": [], "synthetic_password_in_environment": False} result = {"name": name, "shell_command": command, "argv_before_curl_exec": arguments["argv"], "password_in_pre_exec_argv": any(PASSWORD in arg for arg in arguments["argv"]), "password_in_environment": arguments["synthetic_password_in_environment"], "running_curl_ps_command": listing, "running_curl_ps_with_environment": environment_listing, "password_visible_to_own_user_ps_environment": PASSWORD in environment_listing if environment_listing else None, "password_in_running_ps": PASSWORD in listing if listing else None, "returncode": child.returncode, "stdout": stdout, "stderr": stderr, "requests": server.observed} server.shutdown() server.server_close() return result if __name__ == "__main__": CURL = shutil.which("curl") normal = '--proxy-user "$PROXY_USERNAME:$PROXY_PASSWORD"' internal = "--variable %PROXY_USERNAME --variable %PROXY_PASSWORD --expand-proxy-user '{{PROXY_USERNAME}}:{{PROXY_PASSWORD}}'" cases = [run_case("shell-expanded", normal), run_case("curl-internal", internal), run_case("curl-internal-missing-password", internal, missing=True), run_case("curl-internal-empty-password", internal, password=""), run_case("curl-internal-connect-rejection", internal, target="https://example.com/"), run_case("curl-internal-connect-rejection-without-fail", internal, target="https://example.com/", fail=False), run_case("published-baseline-connect-rejection", internal, target="https://example.com/", baseline=True)] checks = { "shell_expansion_passes_password_as_argument": cases[0]["password_in_pre_exec_argv"], "internal_expansion_avoids_password_argument": not cases[1]["password_in_pre_exec_argv"], "both_forms_authenticate_identical_special_characters": all(c["returncode"] == 0 and c["stdout"] == "synthetic-auth-ok\n" for c in cases[:2]), "both_retain_password_in_environment": all(c["password_in_environment"] for c in cases[:2]), "missing_variable_fails_before_network": cases[2]["returncode"] != 0 and not cases[2]["requests"], "empty_variable_is_accepted": cases[3]["returncode"] == 0 and len(cases[3]["requests"]) == 1, "connect_rejection_with_fail_is_22": cases[4]["returncode"] == 22 and len(cases[4]["requests"]) == 1, "connect_rejection_without_fail_is_56": cases[5]["returncode"] == 56 and len(cases[5]["requests"]) == 1, "published_baseline_exits_22_and_reports_connect407": cases[6]["returncode"] == 22 and cases[6]["stdout"].startswith("http=000 connect=407 seconds="), } output = {"timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(), "platform": platform.platform(), "python": sys.version, "curl": subprocess.check_output([CURL, "--version"], text=True), "synthetic_only": True, "network_scope": "127.0.0.1 only; server never forwards or resolves target", "cases": cases, "checks": checks, "passed": sum(checks.values()), "total": len(checks)} (ROOT / "credential-results.json").write_text(json.dumps(output, indent=2) + "\n") print(json.dumps({"passed": output["passed"], "total": output["total"], "checks": checks}, indent=2)) raise SystemExit(0 if all(checks.values()) else 1)