Proxy mit curl nutzen: Anmeldung, CONNECT und Fehler prüfen

HTTP-Proxy in curl setzen, CONNECT für HTTPS prüfen, Anmeldung von -U und -u trennen und Timeouts an Exit-Codes lesen.

Dieser Text ist für Leute, die curl über einen HTTP- oder SOCKS5-Proxy schicken und den Weg prüfen wollen. Die Beispiele nutzen Dummy-Zugangsdaten. Top Proxy verkauft keine IPs und verteilt keine freien Listen. Ein lokaler 200er ist kein Beweis für Geo, Account oder Anbieterqualität.

01HTTP-Proxy mit curl setzen#

Die eingebaute Binary auf diesem Lauf war curl 8.7.1 mit libcurl 8.7.1, SecureTransport und LibreSSL 3.3.6. Features umfassten SSL, HTTP2 und HTTPS-proxy. Das letzte Wort ist eine Build-Flagge, kein gemessener HTTPS-Proxy-Hop. Das Labor hat nur einen HTTP-Proxy plus lokale Origins benutzt.

everything curl zum HTTP-Proxy (abgerufen am 14. Oktober 2026) sagt klar: -x bzw. --proxy meint standardmäßig einen HTTP-Proxy. Ohne Port nimmt curl 1080. Ein HTTPS-Ziel braucht deshalb kein https:// vor dem Proxy. Der Proxy spricht HTTP. Das Ziel kann trotzdem https:// sein.

Setzen Sie den Proxy explizit, bevor die Shell erbt. curl -q muss die erste Option sein. -q schaltet die Standard-curlrc aus. Sonst kann eine private Datei im Home still Proxys, CAs oder Auth nachladen. Das Labor hat HOME nicht umgebogen. Isolation kam von -q und einem Kindprozess ohne Proxy- und CA-Variablen.

1. Binary und Features mit curl -q --version lesen. Eine neuere Online-Manpage beschreibt Optionen, die 8.7.1 vielleicht nicht hat.
2. Ziel und Proxy als getrennte URLs behandeln. Proxy-Schema und Ziel-Schema sind zwei Knöpfe.
3. curl -q --proxy http://127.0.0.1:PORT --noproxy '' http://ziel.example/ gegen einen Host schicken, den Sie kontrollieren.
4. Verbose lesen. Der Request an den Proxy ist Absolute-Form, also GET http://ziel..., nicht der Origin-Pfad allein.

--noproxy '' ist hier Absicht. Manche Umgebungen umgehen Loopback. Wenn der Fall den Proxy zwingend braucht, darf diese Umgehung den Nachweis nicht verstecken. Einen Bypass testen Sie extra, nicht als Nebeneffekt.

Ohne --proxy traf curl den HTTP-Origin mit Status 200. Der offene Loopback-Proxy zählte null Ereignisse. Das ist die direkte Kontrolle. Sie sagt nichts darüber, was curl täte, wenn http_proxy noch stünde. Das kommt später.

Gegen Ihren eigenen Proxy reicht die Kurzform unten. PORT und die Zugangsdaten kommen von Ihnen, nicht aus diesem Text.

# Kurzform gegen Ihren Proxy, nicht das Loopback-Labor
curl -q --proxy "http://127.0.0.1:PORT" --noproxy '' \
  --proxy-user "USER:PASS" \
  "http://example.invalid/"

Das vollständige Loopback-Labor steht am Ende dieses Artikels als ausführbares Python. Es gibt keinen Download auf dem Katalog und keine Admin-URL. Eine lokale Datei in einem privaten Entwurf ist für Leser nicht erreichbar.

1. Den Python-Block am Ende in eine Datei lab.py speichern.
2. In dasselbe leere Verzeichnis wechseln.
3. python3 lab.py ausführen.

Voraussetzungen: Python 3 und openssl. Die Standard-Binary ist /usr/bin/curl. Eine andere Binary setzen Sie mit CURL_BIN=/pfad/zu/curl python3 lab.py. Pillow und cwebp sind nur für die Abbildung nötig. Ohne sie laufen die Prüfungen weiter und schreiben JSON. Das Skript legt evidence/ und assets/ selbst an.

Ein 200er auf stdout ist zu wenig. Der Proxy muss die Origin-Antwort wirklich weiterreichen. Ein Fake ohne Forwarding wäre kein HTTP-Proxy.

02HTTPS-Ziele über CONNECT schicken#

Leute nennen das oft HTTPS-Proxy, weil die Website https:// trägt. Der Hop davor bleibt ein HTTP-Proxy, der CONNECT spricht. everything curl beschreibt denselben Tunnel: curl schickt CONNECT host:port, der Proxy öffnet TCP, TLS läuft zwischen Client und Origin.

Das Labor hat genau das gesehen. Mit korrekten Dummy-Daten kam CONNECT zum lokalen HTTPS-Origin und danach Origin-Status 200 plus Body mit der Origin-Kennung. http_connect war 200. Der Proxy hat TLS-Bytes durchgereicht, nicht eine gefälschte 200 ohne Tunnel.

--cacert gilt für TLS zum Origin. Es steuert nicht still eine zweite TLS-Schicht zum Proxy. --proxy-cacert wäre die Datei für einen HTTPS-Proxy. Dieser Build listet HTTPS-proxy als Feature. Ob Ihr Hop TLS zum Proxy spricht, hat dieses Labor nicht gemessen. Behaupten Sie das nicht aus der Feature-Zeile.

1. Proxy-URL auf http://127.0.0.1:PORT lassen, auch wenn das Ziel https:// ist.
2. Die Origin-CA mit --cacert übergeben. -k ist kein Normalfall.
3. Im Proxy-Log CONNECT host:port suchen, bevor Sie den 200er feiern.
4. Denselben Lauf ohne --cacert wiederholen. Sie wollen TLS-Fehler nach CONNECT, nicht „Proxy tot“.

Der letzte Schritt ist der, den Leute überspringen. Ohne Labor-CA blieb CONNECT autorisiert. curl beendete mit Exit 60, also Peer-Zertifikat nicht verifiziert. Der Tunnel war da. Das Zertifikat nicht. Eine Browserseite Meine IP zeigt den Browser-Exit, nicht diesen curl-Prozess. Der Proxy-Checker sagt, ob ein Host HTTP oder SOCKS5 annimmt, bevor Sie curl darauf zeigen.

Die Abbildung ist die Ergebnistabelle dieses Laufs, kein 16:9-Cover und kein Screenshot des Checkers.

03Proxy-Anmeldung von Origin-Anmeldung trennen#

-U bzw. --proxy-user ist Proxy-Auth. -u bzw. --user ist Origin-Auth. Wer die beiden mischt, debuggt eine Stunde die falsche Schicht.

everything curl zur Umgebung und Auth bleibt bei HTTP 407, wenn der Proxy ablehnt. Das Labor zeigt, dass 407 nicht ein Verhalten ist.

HTTP-Forwarding ohne User lieferte Status 407 und Exit 0. curl behandelt 407 ohne --fail als erfolgreichen Transfer mit unschönem Status. Mit --fail wurde derselbe 407er zu Exit 22. Das ist 8.7.1 auf dieser Binary, keine universelle Regel für jedes 407.

Falsches Dummy-Passwort war wieder 407, Fixture-Grund wrong. Korrekte Dummy-Daten, inklusive Sonderzeichen @ : / # % über --proxy-user nach dem ersten Doppelpunkt, lieferten Origin 200. Userinfo in der Proxy-URL braucht komponentenweise Percent-Encoding. Sonst frisst der Parser das Passwort an einem @.

CONNECT ohne Anmeldung war anders. Der Proxy schrieb authReason: missing. curl beendete mit Exit 56. http_connect war 407. Es gab keinen Origin-Body. CONNECT ist Tunnelaufbau. 407 dort ist Transportfehler, kein HTML.

Falsche CONNECT-Daten: wieder Exit 56, http_connect 407, Grund wrong. Korrekte Daten: Origin 200 nach CONNECT 200.

Der Origin sah kein Proxy-Authorization. Der Header bleibt am Proxy-Hop. Origin-Basic lag auf /origin-auth: ohne -u Status 401, mit -u und gleichzeitig -U Status 200.

1. Passwort mit Sonderzeichen in --proxy-user nach dem ersten Doppelpunkt oder encodiert in der Proxy-URL.
2. -u nur für die Website, nie als Ersatz für -U.
3. HTTP-407 mit Exit 0 und CONNECT-407 mit Exit 56 getrennt notieren.
4. Verbose-Logs vor dem Speichern schwärzen. Basic-Header gehören nicht in Tickets.

Anbieter für Residential-Proxys dokumentieren dieselbe Trennung, ohne dass dieses Labor ihre Pools getroffen hat. Oxylabs Residential Proxies (14. Oktober 2026) nennt HTTP, HTTPS, HTTP/3 und SOCKS5, User/Passwort oder IP-Freigabe, Einstieg pr.oxylabs.io:7777 als customer-USERNAME:PASSWORD, plus ein curl-Beispiel mit -x und -U. Das ist die dokumentierte Form, kein gemessener Lauf gegen deren Gateway. Decodo Residential Quick Start (gleicher Tag) zeigt gate.decodo.com:7000 und curl -U 'USERNAME:PASSWORD' -x gate.decodo.com:7000. Der Katalog führt das Produkt weiter unter smartproxy-com. Der öffentliche Name der Doku ist Decodo. Der Katalogpfad hat sich dadurch nicht geändert.

04Umgebungsvariablen und Bypass prüfen#

everything curl zu Proxy-Umgebungsvariablen (14. Oktober 2026) ist hier die Quelle, nicht Folklore aus Python. curl liest [scheme]_proxy. http_proxy gilt nur klein. HTTPS_PROXY und ALL_PROXY dürfen groß sein. Ein protokollspezifischer Wert schlägt ALL_PROXY. NO_PROXY ist eine Liste. --noproxy macht dasselbe auf der Kommandozeile.

Großes HTTP_PROXY ignoriert curl extra. CGI setzt aus Request-Headern Variablen mit Präfix HTTP_. Ein Header Proxy: würde sonst zu HTTP_PROXY. Deshalb bleibt nur die Kleinschreibung. Das Labor hat HTTP_PROXY auf den Alpha-Proxy gesetzt. curl ging direkt zum Origin. Alpha blieb leer. Kleines http_proxy mit demselben Dummy-Wert lieferte 200, und nur Alpha sah das GET.

HTTPS_PROXY auf Alpha schickte HTTPS durch CONNECT, wieder 200, wieder CONNECT im Log. ALL_PROXY ohne http_proxy schickte HTTP über Alpha. Ob ein gesetztes http_proxy ein gleichzeitiges ALL_PROXY schlägt, steht in der curl-Doku. Dieses Labor hat die beiden nicht parallel gegen zwei Identitäten laufen lassen. Explizites --proxy auf Bravo bei gesetztem http_proxy=Alpha traf nur Bravo. Status 200 allein hätte die Route nicht verraten. Zwei instrumentierte Identitäten haben sie verraten.

Bypass ist ein eigener Fall. NO_PROXY auf den Labor-Host bei gesetztem http_proxy lieferte 200, Alpha sah nichts. --noproxy '' bei derselben Variable zwang den Proxy wieder. Loopback-Defaults dürfen einen fehlgeschlagenen Proxy nicht als Erfolg tarnen.

Der Kindprozess startete mit einer unveränderlichen, geleerten Umgebung. Proxy-Variablen, CGI-Reste, CURL_CA_BUNDLE, SSL_CERT_FILE, SSLKEYLOGFILE und geerbte OpenSSL-Overrides fehlten dort. Jeder Fall bekam eine Kopie dieser Basis, nicht os.environ. Eine leere private NETRC-Datei lag nur im Temp-Pfad dieser Kopie. --netrc wurde nicht gesetzt. HOME blieb unangetastet. curl hat kein trust_env=False. Wer geerbte CA-Bündel und Proxys aus will, räumt die Variablen und startet mit -q. trust_env=False in Requests ändert außerdem CA- und netrc-Verhalten. Das auf curl zu übertragen wäre falsch.

--resolve auf den Labor-Host nach 127.0.0.1 war injizierte Namenszuordnung in curl selbst. Das ist kein gemessener Leak ins öffentliche DNS und keine Hosts-Datei.

05SOCKS5 und Namensauflösung unterscheiden#

everything curl zu SOCKS (14. Oktober 2026) trennt, wer den Namen auflöst. socks5 / --socks5: curl löst lokal. socks5h / --socks5-hostname: der Hostname geht an den Proxy. IPv6 steht in der Tabelle bei Version 5, nicht bei 4.

Ein erfolgreicher SOCKS-Request lokalisiert DNS nicht. Sie brauchen den Handshake. Das Labor hat ATYP am kontrollierten SOCKS5-Fixture gelesen.

--socks5 gegen eine IPv4-Loopback-URL sendete ATYP ipv4, Host 127.0.0.1, Origin 200. --socks5-hostname gegen den kontrollierten Hostnamen sendete ATYP domain mit genau diesem Namen, wieder 200. --socks5 gegen denselben Host, plus --resolve, sendete ATYP ipv4 auf 127.0.0.1. Die lokale Auflösung kam von --resolve, nicht von einem öffentlichen Resolver. So misst 8.7.1 das, nicht eine Requests-Geschichte über socks5h.

1. Schema oder Flag bewusst wählen. socks5:// in -x ist nicht socks5h://.
2. Den Hostnamen nur dann an den Proxy geben, wenn Sie das wollen.
3. ATYP im Proxy-Log lesen. Status 200 reicht nicht.
4. Unbekanntes Schema als Lücke stehen lassen, nicht als Erfolg.

SOCKS5-Userpass hat dieses Fixture nicht geprüft. HTTP-Basic am HTTP-Proxy ist ein anderer Handshake.

06Timeouts und Verbindungsfehler lesen#

Die aktuelle offizielle Manpage zu --connect-timeout und die gepinnte Optiondatei von curl 8.7.1 (beide 14. Oktober 2026 gelesen): --connect-timeout begrenzt die Verbindungsphase. Die Phase ist fertig, wenn DNS-Lookup und die angeforderten TCP-, TLS- oder QUIC-Handshakes durch sind, nicht schon beim ersten SYN-ACK. Danach gilt die Option nicht mehr. Die gesamte Operation begrenzt --max-time. Das ist kein Read-Timeout wie in Requests. Exit 28 am gehaltenen Bind ohne listen war auf Darwin 24 mit dieser Binary beobachtet, kein allgemeines OS-Versprechen.

Verweigerung und Timeout sind verschiedene Exits. Das Labor hat einen eigenen Listener auf einem ephemeren Loopback-Port gebunden, gehört, dann geschlossen. curl gegen diesen Port endete mit Exit 7. Verbose: Connection refused. Das ist Verbindungsfehler, kein gemessener Connect-Timeout.

Derselbe Rechner, Socket nur gebunden, absichtlich ohne listen, Eigentum bis zum Request-Ende: curl mit --connect-timeout 1 endete mit Exit 28. Verbose: ipv4 connect timeout after 1000ms. Darwin schickt hier oft keinen RST. Das Labor nennt das beobachteten Connect-Timeout am gehaltenen Nicht-Listener, nicht Verweigerung. 192.0.2.1 oder ein totes Gateway wäre die falsche Geschichte gewesen.

/slow-body schickte Header sofort, flushte das erste Body-Byte, signalisierte das, und wartete länger als --max-time 1 auf den Rest. Erstes Byte war da. Exit 28 nach 1,03 s. Das ist Transfer-Timeout nach Body-Start, kein Connect-Timeout und kein Requests-ReadTimeout. Ein Delay vor den Headern hätte das nicht bewiesen.

--retry hat dieser Lauf nicht gezeigt. Eine Flagge in der Manpage ist keine Idempotenz für POST.

07Dokumentierte Endpunkte auswählen#

Protokollsupport und IP-Herkunft sind verschiedene Achsen. Ein Labor-Request beweist nur diese Labor-Route. Mobile Proxys und Residential Proxys sind Katalogseiten zu Herkunft, nicht zu curl-Flags.

Die öffentlichen Karten Oxylabs und Decodo unter smartproxy-com waren an diesem Tag erreichbar. Die Doku oben beschreibt HTTP-Gateways plus User/Passwort oder Freigabe. Das Labor hat keinen bezahlten Einstieg getroffen. Keine Latenz, kein Land, keine Kontoberechtigung.

Wenn curl durch so einen Hop soll, bleiben dieselben Checks: -q zuerst, -x als HTTP-Proxy auch für HTTPS-Ziele, -U getrennt von -u, Verbose auf CONNECT, Exit 7 gegen 28 gegen 56 gegen 60 nicht in einen Satz pressen.

Oxylabs: Residential- und Datacenter-Proxys4.0770

Oxylabs verkauft Residential-, Datacenter- und Mobile-Proxys. Residential Self-Serve startet bei $6/GB für 5 GB/Monat. Käuferzahlungen: Karten, Überweisung, AliPay, PayPal.

10 Länder

0,59 $

pro 1 GB

Decodo (Smartproxy): Residential- und Datacenter-Proxys4.22106

Decodo (ehemals Smartproxy) verkauft Residential- und Datacenter-Proxys. Residential: $4/GB ohne Monatsplan oder monatliche Pakete ab $3.75/GB für 3 GB. Zahlungen: Karten, PayPal, Alipay, Google/Apple Pay, Krypto (mit Einschränkungen).

83 Länder

0,38 $

pro 1 GB

Meine IP bleibt der Browser-Weg. Der Proxy-Checker bleibt der schnelle Protokollblick. Keins von beiden ersetzt ein curl-Trace.

08Das Loopback-Labor selbst ausführen#

Speichern Sie den folgenden Block als lab.py in einem leeren Ordner. Dann python3 lab.py. Standard-curl ist /usr/bin/curl, Override CURL_BIN. Das ist der vollständige Fixture-Quelltext, kein Katalog-Download.

#!/usr/bin/env python3
"""Self-contained curl proxy lab. Loopback fixtures only. One entry point."""

from __future__ import annotations

import base64
import json
import os
import re
import select
import shutil
import socket
import ssl
import stat
import struct
import subprocess
import sys
import tempfile
import threading
import time
import traceback
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any
from urllib.parse import quote, urlsplit

HERE = Path(__file__).resolve().parent
# Package layout: tests/lab.py writes into the article directory.
# Extracted standalone lab.py writes next to itself.
ART_ROOT = HERE.parent if HERE.name == "tests" else HERE
ASSETS = ART_ROOT / "assets"
EVIDENCE = ART_ROOT / "evidence"
RAW = EVIDENCE / "raw"
RESULTS_JSON = HERE / "results.json"
RESULTS_TXT = HERE / "results.txt"
FIGURE_WEBP = ASSETS / "curl-results.webp"
FIGURE_PNG = ASSETS / "curl-results.png"
CA_PUBLIC = EVIDENCE / "ca.pem"

CHILD_FLAG = "CURL_LAB_CHILD"
PARTIAL_HTTPS_FAIL = "CURL_LAB_PARTIAL_HTTPS_FAIL"
LAB_HOST = "curl-lab.origin"
PROXY_USER = "lab-user"
PROXY_PASS = "Aa1@:/#%"
ORIGIN_USER = "origin-user"
ORIGIN_PASS = "origin-lab"
SLOW_BODY = b"H" + b"ELLO-SLOW-BODY!"
SLOW_WAIT_S = 2.4
MAX_TIME_SLOW = "1.0"
CONNECT_TIMEOUT_HELD = "1.0"
# Default is the Darwin system binary, not PATH lookup. Override with CURL_BIN.
CURL_BIN = os.environ.get("CURL_BIN", "/usr/bin/curl")

CLEAR_KEYS = [
    "HTTP_PROXY",
    "HTTPS_PROXY",
    "ALL_PROXY",
    "NO_PROXY",
    "FTP_PROXY",
    "http_proxy",
    "https_proxy",
    "all_proxy",
    "no_proxy",
    "ftp_proxy",
    "REQUESTS_CA_BUNDLE",
    "CURL_CA_BUNDLE",
    "SSL_CERT_FILE",
    "SSL_CERT_DIR",
    "SSLKEYLOGFILE",
    "REQUEST_METHOD",
    "GATEWAY_INTERFACE",
    "CONTENT_TYPE",
    "CONTENT_LENGTH",
    "QUERY_STRING",
    "SCRIPT_NAME",
    "PATH_INFO",
    "SERVER_PROTOCOL",
    "REMOTE_ADDR",
    "SERVER_NAME",
    "SERVER_PORT",
    "NETRC",
    "SOCKS_PROXY",
    "socks_proxy",
    "OPENSSL_CONF",
    "OPENSSL_ENGINES",
    "OPENSSL_MODULES",
    "OPENSSL_TRACE",
    "SSLEAY_CONF",
]

AUTH_FLAGS = {"--proxy-user", "-U", "--user", "-u"}


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()


def redact_text(value: str) -> str:
    out = value
    encoded = quote(PROXY_PASS, safe="")
    b64 = base64.b64encode(f"{PROXY_USER}:{PROXY_PASS}".encode()).decode()
    origin_b64 = base64.b64encode(f"{ORIGIN_USER}:{ORIGIN_PASS}".encode()).decode()
    for secret in (PROXY_PASS, ORIGIN_PASS, encoded, b64, origin_b64):
        if secret:
            out = out.replace(secret, "[redacted]")
    out = re.sub(r"(://)([^/@\s'\"]+):([^@/\s'\"]+)@", r"\1[redacted]:[redacted]@", out)
    out = re.sub(r"(?i)((?:proxy-)?authorization:\s*basic)\s+\S+", r"\1 [redacted-b64]", out)
    return out


def redact_argv(cmd: list[str]) -> list[str]:
    out: list[str] = []
    hide_next = False
    for arg in cmd:
        if hide_next:
            out.append("[redacted-auth]")
            hide_next = False
            continue
        if arg in AUTH_FLAGS:
            hide_next = True
            out.append(arg)
            continue
        out.append(redact_text(arg))
    return out


def clean_env(src: dict[str, str]) -> dict[str, str]:
    env = dict(src)
    for key in list(env):
        lower = key.lower()
        if key in CLEAR_KEYS or lower.endswith("_proxy") or lower.startswith("openssl_"):
            env.pop(key, None)
    env.pop("NETRC", None)
    return env


def json_safe(obj: Any) -> Any:
    if isinstance(obj, str):
        return redact_text(obj)
    if isinstance(obj, list):
        return [json_safe(x) for x in obj]
    if isinstance(obj, dict):
        return {str(k): json_safe(v) for k, v in obj.items()}
    return obj


class OriginHandler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def log_message(self, *_args: object) -> None:
        return

    def _leaked_proxy_auth(self) -> bool:
        return any(k.lower() == "proxy-authorization" for k in self.headers.keys())

    def _mark(self) -> str:
        return str(getattr(self.server, "origin_id", "origin"))

    def _write(self, code: int, body: bytes, extra: dict[str, str] | None = None) -> None:
        leaked = self._leaked_proxy_auth()
        getattr(self.server, "leak_flags").append(leaked)
        self.send_response(code)
        self.send_header("Content-Type", "text/plain; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("X-Origin-Id", self._mark())
        self.send_header("X-Proxy-Auth-Leaked", "1" if leaked else "0")
        self.send_header("Connection", "close")
        if extra:
            for k, v in extra.items():
                self.send_header(k, v)
        self.end_headers()
        self.wfile.write(body)
        self.wfile.flush()

    def do_GET(self) -> None:  # noqa: N802
        path = urlsplit(self.path).path
        shared = getattr(self.server, "shared")
        if path == "/slow-body":
            leaked = self._leaked_proxy_auth()
            getattr(self.server, "leak_flags").append(leaked)
            self.send_response(200)
            self.send_header("Content-Type", "text/plain; charset=utf-8")
            self.send_header("Content-Length", str(len(SLOW_BODY)))
            self.send_header("X-Origin-Id", self._mark())
            self.send_header("X-Slow-Body", "1")
            self.send_header("Connection", "close")
            self.end_headers()
            self.wfile.write(SLOW_BODY[:1])
            self.wfile.flush()
            shared["first_byte"].set()
            shared["slow_continue"].wait(timeout=SLOW_WAIT_S)
            try:
                self.wfile.write(SLOW_BODY[1:])
                self.wfile.flush()
            except OSError:
                pass
            return
        if path == "/origin-auth":
            header = self.headers.get("Authorization", "")
            expected = "Basic " + base64.b64encode(f"{ORIGIN_USER}:{ORIGIN_PASS}".encode()).decode()
            if header != expected:
                self._write(401, b"ORIGIN-401", {"WWW-Authenticate": 'Basic realm="origin"'})
                return
            self._write(200, f"{self._mark()}|origin-auth-ok".encode())
            return
        self._write(200, f"{self._mark()}|{path}".encode())


def stop_origin(server: ThreadingHTTPServer) -> None:
    thread = getattr(server, "_thread", None)
    if thread is not None and thread.is_alive():
        try:
            server.shutdown()
        except Exception:
            pass
        thread.join(timeout=2)
    try:
        server.server_close()
    except Exception:
        pass


def start_origin(origin_id: str, shared: dict[str, Any], tls: dict[str, Path] | None) -> ThreadingHTTPServer:
    server = ThreadingHTTPServer(("127.0.0.1", 0), OriginHandler)
    server.origin_id = origin_id
    server.shared = shared
    server.leak_flags = []
    server.daemon_threads = False
    try:
        if tls is not None:
            ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
            ctx.load_cert_chain(str(tls["cert"]), str(tls["key"]))
            server.socket = ctx.wrap_socket(server.socket, server_side=True)
        thread = threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.2}, name=origin_id)
        thread.start()
        server._thread = thread
        return server
    except Exception:
        stop_origin(server)
        raise


class HttpForwardProxy:
    def __init__(self, name: str, user: str | None, password: str | None) -> None:
        self.name = name
        self.user = user
        self.password = password
        self.allowed: dict[tuple[str, int], tuple[str, int]] = {}
        self.events: list[dict[str, Any]] = []
        self._lock = threading.Lock()
        self._stop = threading.Event()
        self._workers: list[threading.Thread] = []
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.sock.bind(("127.0.0.1", 0))
        self.sock.listen(64)
        self.sock.settimeout(0.25)
        self.port = int(self.sock.getsockname()[1])
        self._thread = threading.Thread(target=self._accept, name=f"proxy-{name}")

    def allow(self, host: str, port: int, dest: tuple[str, int]) -> None:
        self.allowed[(host.lower(), port)] = dest

    def start(self) -> None:
        self._thread.start()

    def snapshot(self) -> list[dict[str, Any]]:
        with self._lock:
            return list(self.events)

    def _record(self, row: dict[str, Any]) -> None:
        with self._lock:
            self.events.append(row)

    def _auth_reason(self, headers: dict[str, str]) -> str:
        if self.user is None:
            return "open"
        raw = headers.get("proxy-authorization", "")
        if not raw:
            return "missing"
        if not raw.lower().startswith("basic "):
            return "wrong"
        try:
            decoded = base64.b64decode(raw.split(" ", 1)[1]).decode("utf-8")
            user, passwd = decoded.split(":", 1)
        except Exception:
            return "wrong"
        if user == self.user and passwd == self.password:
            return "ok"
        return "wrong"

    def _map_dest(self, host: str, port: int) -> tuple[str, int] | None:
        return self.allowed.get((host.lower(), port))

    def _accept(self) -> None:
        while not self._stop.is_set():
            try:
                conn, _addr = self.sock.accept()
            except socket.timeout:
                continue
            except OSError:
                break
            worker = threading.Thread(target=self._handle, args=(conn,), name=f"{self.name}-conn")
            self._workers.append(worker)
            worker.start()

    def _recv_headers(self, conn: socket.socket) -> bytes:
        buf = b""
        conn.settimeout(5)
        while b"\r\n\r\n" not in buf:
            chunk = conn.recv(4096)
            if not chunk:
                break
            buf += chunk
            if len(buf) > 65536:
                break
        return buf

    def _handle(self, conn: socket.socket) -> None:
        upstream = None
        try:
            raw = self._recv_headers(conn)
            if not raw:
                return
            head, _, _rest = raw.partition(b"\r\n\r\n")
            lines = head.decode("iso-8859-1", "replace").split("\r\n")
            if not lines:
                return
            parts = lines[0].split(" ")
            if len(parts) < 2:
                return
            method, target = parts[0], parts[1]
            headers: dict[str, str] = {}
            for line in lines[1:]:
                if ":" in line:
                    k, v = line.split(":", 1)
                    headers[k.strip().lower()] = v.strip()
            reason = self._auth_reason(headers)
            authorized = reason in ("ok", "open")
            self._record(
                {
                    "proxyId": self.name,
                    "method": method,
                    "target": target,
                    "authReason": reason,
                    "authorized": authorized,
                }
            )
            if not authorized:
                body = b"proxy-407"
                msg = (
                    "HTTP/1.1 407 Proxy Authentication Required\r\n"
                    'Proxy-Authenticate: Basic realm="lab"\r\n'
                    "Content-Length: %d\r\n"
                    "Connection: close\r\n\r\n" % len(body)
                ).encode() + body
                conn.sendall(msg)
                return
            if method == "CONNECT":
                host, port_s = target.rsplit(":", 1)
                port = int(port_s)
                mapped = self._map_dest(host, port)
                if mapped is None:
                    conn.sendall(b"HTTP/1.1 403 Forbidden\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
                    return
                upstream = socket.create_connection(mapped, timeout=5)
                conn.sendall(b"HTTP/1.1 200 Connection Established\r\n\r\n")
                self._tunnel(conn, upstream)
                return
            split = urlsplit(target)
            if split.scheme != "http" or not split.hostname:
                conn.sendall(b"HTTP/1.1 400 Bad Request\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
                return
            dest_port = split.port or 80
            mapped = self._map_dest(split.hostname, dest_port)
            if mapped is None:
                conn.sendall(b"HTTP/1.1 403 Forbidden\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
                return
            path = split.path or "/"
            if split.query:
                path += "?" + split.query
            upstream = socket.create_connection(mapped, timeout=5)
            skip = {"proxy-authorization", "proxy-connection"}
            out = [f"{method} {path} HTTP/1.1"]
            host_hdr = f"{split.hostname}:{dest_port}" if dest_port != 80 else split.hostname
            out.append(f"Host: {host_hdr}")
            for k, v in headers.items():
                if k in skip or k == "host":
                    continue
                out.append(f"{k}: {v}")
            out.append("Connection: close")
            upstream.sendall(("\r\n".join(out) + "\r\n\r\n").encode("iso-8859-1"))
            self._tunnel(conn, upstream)
        except Exception:
            pass
        finally:
            for sock in (conn, upstream):
                if sock is None:
                    continue
                try:
                    sock.close()
                except OSError:
                    pass

    def _tunnel(self, a: socket.socket, b: socket.socket) -> None:
        sockets = [a, b]
        try:
            a.settimeout(None)
            b.settimeout(None)
            while True:
                ready, _, _ = select.select(sockets, [], [], 8)
                if not ready:
                    break
                for sock in ready:
                    other = b if sock is a else a
                    data = sock.recv(16384)
                    if not data:
                        return
                    other.sendall(data)
        except OSError:
            return

    def stop(self) -> None:
        self._stop.set()
        try:
            self.sock.close()
        except OSError:
            pass
        thread = getattr(self, "_thread", None)
        if thread is not None and thread.ident is not None:
            thread.join(timeout=2)
        for worker in self._workers:
            worker.join(timeout=1)


class Socks5Proxy:
    def __init__(self, name: str) -> None:
        self.name = name
        self.allowed: dict[tuple[str, int], tuple[str, int]] = {}
        self.events: list[dict[str, Any]] = []
        self._lock = threading.Lock()
        self._stop = threading.Event()
        self._workers: list[threading.Thread] = []
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.sock.bind(("127.0.0.1", 0))
        self.sock.listen(32)
        self.sock.settimeout(0.25)
        self.port = int(self.sock.getsockname()[1])
        self._thread = threading.Thread(target=self._accept, name="socks5")

    def allow(self, host: str, port: int, dest: tuple[str, int]) -> None:
        self.allowed[(host.lower(), port)] = dest

    def start(self) -> None:
        self._thread.start()

    def snapshot(self) -> list[dict[str, Any]]:
        with self._lock:
            return list(self.events)

    def _map(self, host: str, port: int) -> tuple[str, int] | None:
        key = (host.lower(), port)
        if key in self.allowed:
            return self.allowed[key]
        if host in ("127.0.0.1", "::1") :
            for (h, p), dest in self.allowed.items():
                if p == port:
                    return dest
        return None

    def _accept(self) -> None:
        while not self._stop.is_set():
            try:
                conn, _addr = self.sock.accept()
            except socket.timeout:
                continue
            except OSError:
                break
            worker = threading.Thread(target=self._handle, args=(conn,))
            self._workers.append(worker)
            worker.start()

    def _recvn(self, conn: socket.socket, n: int) -> bytes:
        buf = b""
        while len(buf) < n:
            chunk = conn.recv(n - len(buf))
            if not chunk:
                raise OSError("eof")
            buf += chunk
        return buf

    def _handle(self, conn: socket.socket) -> None:
        upstream = None
        try:
            conn.settimeout(5)
            head = self._recvn(conn, 2)
            if head[0] != 5:
                return
            self._recvn(conn, head[1])
            conn.sendall(b"\x05\x00")
            req = self._recvn(conn, 4)
            if req[0] != 5 or req[1] != 1:
                return
            atyp = req[3]
            host = ""
            atyp_name = "other"
            if atyp == 1:
                host = socket.inet_ntoa(self._recvn(conn, 4))
                atyp_name = "ipv4"
            elif atyp == 3:
                ln = self._recvn(conn, 1)[0]
                host = self._recvn(conn, ln).decode("ascii", "replace")
                atyp_name = "domain"
            elif atyp == 4:
                host = socket.inet_ntop(socket.AF_INET6, self._recvn(conn, 16))
                atyp_name = "ipv6"
            else:
                return
            port = struct.unpack("!H", self._recvn(conn, 2))[0]
            mapped = self._map(host, port)
            with self._lock:
                self.events.append(
                    {
                        "proxyId": self.name,
                        "atyp": atyp_name,
                        "host": host,
                        "port": port,
                        "mapped": bool(mapped),
                    }
                )
            if mapped is None:
                conn.sendall(b"\x05\x02\x00\x01\x00\x00\x00\x00\x00\x00")
                return
            upstream = socket.create_connection(mapped, timeout=5)
            conn.sendall(b"\x05\x00\x00\x01\x7f\x00\x00\x01" + struct.pack("!H", mapped[1]))
            self._tunnel(conn, upstream)
        except Exception:
            pass
        finally:
            for sock in (conn, upstream):
                if sock is None:
                    continue
                try:
                    sock.close()
                except OSError:
                    pass

    def _tunnel(self, a: socket.socket, b: socket.socket) -> None:
        sockets = [a, b]
        try:
            a.settimeout(None)
            b.settimeout(None)
            while True:
                ready, _, _ = select.select(sockets, [], [], 8)
                if not ready:
                    break
                for sock in ready:
                    other = b if sock is a else a
                    data = sock.recv(16384)
                    if not data:
                        return
                    other.sendall(data)
        except OSError:
            return

    def stop(self) -> None:
        self._stop.set()
        try:
            self.sock.close()
        except OSError:
            pass
        thread = getattr(self, "_thread", None)
        if thread is not None and thread.ident is not None:
            thread.join(timeout=2)
        for worker in self._workers:
            worker.join(timeout=1)


def make_certs(tmp: Path) -> dict[str, Path]:
    ca_key = tmp / "ca.key"
    ca_pem = tmp / "ca.pem"
    origin_key = tmp / "origin.key"
    origin_csr = tmp / "origin.csr"
    origin_crt = tmp / "origin.crt"
    ext = tmp / "san.cnf"
    ext.write_text(
        "subjectAltName=DNS:localhost,DNS:curl-lab.origin,IP:127.0.0.1\n"
        "basicConstraints=CA:FALSE\n"
    )
    subprocess.run(
        [
            "openssl",
            "req",
            "-x509",
            "-newkey",
            "rsa:2048",
            "-sha256",
            "-days",
            "1",
            "-nodes",
            "-keyout",
            str(ca_key),
            "-out",
            str(ca_pem),
            "-subj",
            "/CN=curl-lab-ca",
        ],
        check=True,
        capture_output=True,
    )
    subprocess.run(
        [
            "openssl",
            "req",
            "-newkey",
            "rsa:2048",
            "-nodes",
            "-keyout",
            str(origin_key),
            "-out",
            str(origin_csr),
            "-subj",
            "/CN=localhost",
        ],
        check=True,
        capture_output=True,
    )
    subprocess.run(
        [
            "openssl",
            "x509",
            "-req",
            "-in",
            str(origin_csr),
            "-CA",
            str(ca_pem),
            "-CAkey",
            str(ca_key),
            "-CAcreateserial",
            "-out",
            str(origin_crt),
            "-days",
            "1",
            "-extfile",
            str(ext),
        ],
        check=True,
        capture_output=True,
    )
    for key in (ca_key, origin_key):
        key.chmod(stat.S_IRUSR | stat.S_IWUSR)
    return {"ca_key": ca_key, "ca_pem": ca_pem, "origin_key": origin_key, "origin_crt": origin_crt}


def parse_writeout(text: str) -> dict[str, str]:
    out: dict[str, str] = {}
    if "---CURL---" not in text:
        return out
    block = text.split("---CURL---", 1)[1]
    for line in block.strip().splitlines():
        if "=" in line:
            k, v = line.split("=", 1)
            out[k.strip()] = v.strip()
    return out


def render_figure(rows: list[dict[str, Any]], versions: dict[str, Any], started: str) -> dict[str, Any]:
    try:
        from PIL import Image, ImageDraw, ImageFont
    except ImportError:
        return {"skipped": True, "reason": "Pillow not installed; assertions still run"}

    ASSETS.mkdir(parents=True, exist_ok=True)
    font_path = None
    for cand in (
        "/System/Library/Fonts/Supplemental/Arial Unicode.ttf",
        "/Library/Fonts/Arial Unicode.ttf",
        "/System/Library/Fonts/Supplemental/Arial.ttf",
        "/System/Library/Fonts/Helvetica.ttc",
    ):
        if Path(cand).exists():
            font_path = cand
            break
    if font_path:
        title_font = ImageFont.truetype(font_path, 22)
        cell_font = ImageFont.truetype(font_path, 15)
        small = ImageFont.truetype(font_path, 13)
    else:
        title_font = ImageFont.load_default()
        cell_font = ImageFont.load_default()
        small = ImageFont.load_default()
    header = ["Fall", "Ergebnis", "Detail"]
    data = []
    for row in rows:
        detail = str(row.get("detail", ""))[:90]
        data.append([str(row.get("name", "")), str(row.get("result", "")), detail])
    col_w = [320, 90, 720]
    row_h = 28
    width = sum(col_w) + 40
    height = 90 + row_h * (len(data) + 1) + 40
    img = Image.new("RGB", (width, height), (15, 23, 42))
    draw = ImageDraw.Draw(img)
    draw.text((20, 16), "curl 8.7.1 Loopback-Labor, 14. Oktober 2026", fill=(196, 181, 253), font=title_font)
    meta = (
        f"{versions.get('curlLine', '')[:80]}  "
        f"Python {versions.get('python')}  {started[:19]}Z"
    )
    draw.text((20, 48), meta, fill=(203, 213, 225), font=small)
    y = 78
    x = 20
    for i, h in enumerate(header):
        draw.rectangle([x, y, x + col_w[i], y + row_h], fill=(88, 28, 135))
        draw.text((x + 8, y + 6), h, fill=(250, 245, 255), font=cell_font)
        x += col_w[i]
    y += row_h
    for idx, cells in enumerate(data):
        bg = (30, 41, 59) if idx % 2 == 0 else (15, 23, 42)
        x = 20
        passed = cells[1] == "pass"
        for i, val in enumerate(cells):
            color = (167, 243, 208) if i == 1 and passed else ((252, 165, 165) if i == 1 else (226, 232, 240))
            draw.rectangle([x, y, x + col_w[i], y + row_h], fill=bg)
            draw.text((x + 8, y + 6), val, fill=color, font=cell_font)
            x += col_w[i]
        y += row_h
    img.save(FIGURE_PNG)
    cwebp = shutil.which("cwebp")
    if not cwebp:
        return {
            "skipped": False,
            "path": str(FIGURE_PNG),
            "bytes": FIGURE_PNG.stat().st_size,
            "dimensions": list(img.size),
            "format": "png",
            "note": "cwebp missing; PNG kept. Assertions still run.",
        }
    subprocess.run([cwebp, "-q", "78", str(FIGURE_PNG), "-o", str(FIGURE_WEBP)], check=True, capture_output=True)
    st = FIGURE_WEBP.stat()
    return {
        "path": str(FIGURE_WEBP),
        "bytes": st.st_size,
        "dimensions": list(img.size),
        "format": "webp",
    }


def write_results(payload: dict[str, Any]) -> None:
    RAW.mkdir(parents=True, exist_ok=True)
    HERE.mkdir(parents=True, exist_ok=True)
    text = json.dumps(json_safe(payload), indent=2, sort_keys=True)
    if PROXY_PASS in text or ORIGIN_PASS in text:
        raise RuntimeError("refusing to write unredacted secret")
    if re.search(r"://(?!\[redacted\]:\[redacted\]@)[^/\s]+:[^@/\s]+@", text):
        raise RuntimeError("refusing to write unredacted URL userinfo")
    if re.search(r"(?i)(?:proxy-)?authorization:\s*basic\s+(?!\[redacted-b64\])\S+", text):
        raise RuntimeError("refusing to write unredacted Authorization")
    RESULTS_JSON.write_text(text)
    (RAW / "results.json").write_text(text)
    counts = payload.get("counts", {})
    lines = [
        f"ok={payload.get('ok')}",
        f"exitStatus={payload.get('exitStatus')}",
        f"ranAt={payload.get('ranAt')}",
        f"finishedAt={payload.get('finishedAt')}",
        f"captured={counts.get('captured')} asserted={counts.get('asserted')} "
        f"passed={counts.get('passed')} failed={counts.get('failed')} injected={counts.get('injected')}",
        f"failedNames={payload.get('failedNames')}",
        "",
        "assertions:",
    ]
    for a in payload.get("assertions", []):
        flag = "PASS" if a.get("passed") else "FAIL"
        lines.append(f"  {flag} {a.get('name')} :: {a.get('detail')}")
    RESULTS_TXT.write_text(redact_text("\n".join(lines) + "\n"))
    (RAW / "results.txt").write_text(RESULTS_TXT.read_text())


def launch_child() -> int:
    env = clean_env(os.environ)
    env[CHILD_FLAG] = "1"
    env["PYTHONDONTWRITEBYTECODE"] = "1"
    proc = subprocess.run([sys.executable, str(Path(__file__).resolve())], env=env)
    return int(proc.returncode)


def run_lab() -> int:
    started = utc_now()
    captured: list[dict[str, Any]] = []
    assertions: list[dict[str, Any]] = []
    injected: list[dict[str, Any]] = []
    owned: dict[str, Any] = {}
    tmp = Path(tempfile.mkdtemp(prefix="curl-lab-"))
    netrc_path = tmp / "netrc"
    netrc_path.write_text("")
    netrc_path.chmod(stat.S_IRUSR | stat.S_IWUSR)
    baseline_env = clean_env(os.environ)
    baseline_env["NETRC"] = str(netrc_path)
    curl_ver = subprocess.run([CURL_BIN, "-q", "--version"], capture_output=True, text=True, timeout=5, env=dict(baseline_env))
    versions = {
        "python": sys.version.split()[0],
        "curl": curl_ver.stdout.splitlines()[0] if curl_ver.stdout else "",
        "curlLine": curl_ver.stdout.splitlines()[0] if curl_ver.stdout else "",
        "curlFeatures": next((ln for ln in curl_ver.stdout.splitlines() if ln.startswith("Features:")), ""),
        "curlProtocols": next((ln for ln in curl_ver.stdout.splitlines() if ln.startswith("Protocols:")), ""),
        "opensslPython": ssl.OPENSSL_VERSION,
        "os": sys.platform,
        "platform": os.uname().sysname + "-" + os.uname().release + "-" + os.uname().machine,
        "interpreter": sys.executable,
        "curlBin": CURL_BIN,
        "homeUnchanged": os.environ.get("HOME"),
    }
    seq = {"n": 0}

    def capture(name: str, **fields: Any) -> None:
        captured.append({"name": name, **json_safe(fields)})

    def assert_true(name: str, cond: bool, detail: str, kind: str = "assert") -> None:
        assertions.append({"name": name, "passed": bool(cond), "detail": redact_text(detail), "kind": kind})

    def run_curl(extra: list[str], timeout: float = 12.0, env: dict[str, str] | None = None) -> dict[str, Any]:
        seq["n"] += 1
        body_path = tmp / f"body-{seq['n']}.txt"
        hdr_path = tmp / f"hdr-{seq['n']}.txt"
        wo = (
            "\n---CURL---\n"
            "http_code=%{http_code}\n"
            "http_connect=%{http_connect}\n"
            "ssl_verify_result=%{ssl_verify_result}\n"
            "errormsg=%{errormsg}\n"
        )
        cmd = [
            CURL_BIN,
            "-q",
            "--silent",
            "--show-error",
            "--globoff",
            "--max-redirs",
            "0",
            "-D",
            str(hdr_path),
            "-o",
            str(body_path),
            "-w",
            wo,
        ]
        if not any(a in ("--max-time", "-m") for a in extra):
            cmd.extend(["--max-time", "8"])
        cmd.extend(extra)
        if cmd[1] != "-q":
            raise RuntimeError("curl -q must be the first option")
        if any(a in ("--netrc", "--netrc-optional", "-n", "--insecure", "-k") for a in cmd):
            raise RuntimeError("forbidden curl option")
        run_env = dict(env) if env is not None else dict(baseline_env)
        try:
            proc = subprocess.run(cmd, env=run_env, capture_output=True, text=True, timeout=timeout)
            timed_out = False
        except subprocess.TimeoutExpired as exc:
            proc_stdout = exc.stdout.decode() if isinstance(exc.stdout, bytes) else (exc.stdout or "")
            proc_stderr = exc.stderr.decode() if isinstance(exc.stderr, bytes) else (exc.stderr or "")
            timed_out = True

            class T:
                returncode = 124
                stdout = proc_stdout
                stderr = proc_stderr

            proc = T()  # type: ignore[assignment]
        body = body_path.read_text(errors="replace") if body_path.exists() else ""
        hdr = hdr_path.read_text(errors="replace") if hdr_path.exists() else ""
        wo_parsed = parse_writeout(proc.stdout)
        verbose = redact_text(proc.stderr)
        argv_pub = redact_argv(cmd)
        return {
            "argv": argv_pub,
            "exit": int(proc.returncode),
            "stdout": redact_text(proc.stdout),
            "stderr": verbose,
            "body": redact_text(body),
            "headers": redact_text(hdr),
            "writeout": {k: redact_text(v) for k, v in wo_parsed.items()},
            "http_code": wo_parsed.get("http_code", ""),
            "http_connect": wo_parsed.get("http_connect", ""),
            "timedOut": timed_out,
            "firstOption": cmd[1],
        }

    try:
        ASSETS.mkdir(parents=True, exist_ok=True)
        EVIDENCE.mkdir(parents=True, exist_ok=True)
        RAW.mkdir(parents=True, exist_ok=True)
        HERE.mkdir(parents=True, exist_ok=True)
        certs = make_certs(tmp)
        shutil.copyfile(certs["ca_pem"], CA_PUBLIC)
        shared = {"first_byte": threading.Event(), "slow_continue": threading.Event()}
        http_origin = start_origin("origin-http", shared, None)
        owned["http_origin"] = http_origin
        if os.environ.get(PARTIAL_HTTPS_FAIL) == "1":
            start_origin(
                "origin-https",
                shared,
                {"cert": certs["origin_crt"], "key": tmp / "missing-origin.key"},
            )
        https_origin = start_origin(
            "origin-https",
            shared,
            {"cert": certs["origin_crt"], "key": certs["origin_key"]},
        )
        owned["https_origin"] = https_origin
        http_port = int(http_origin.server_address[1])
        https_port = int(https_origin.server_address[1])
        def allow_http_https(proxy: HttpForwardProxy) -> None:
            proxy.allow("127.0.0.1", http_port, ("127.0.0.1", http_port))
            proxy.allow("localhost", http_port, ("127.0.0.1", http_port))
            proxy.allow(LAB_HOST, http_port, ("127.0.0.1", http_port))
            proxy.allow("127.0.0.1", https_port, ("127.0.0.1", https_port))
            proxy.allow("localhost", https_port, ("127.0.0.1", https_port))
            proxy.allow(LAB_HOST, https_port, ("127.0.0.1", https_port))

        alpha = HttpForwardProxy("alpha", PROXY_USER, PROXY_PASS)
        allow_http_https(alpha)
        owned["alpha"] = alpha
        alpha.start()
        bravo = HttpForwardProxy("bravo", PROXY_USER, PROXY_PASS)
        allow_http_https(bravo)
        owned["bravo"] = bravo
        bravo.start()
        socks = Socks5Proxy("socks")
        socks.allow("127.0.0.1", http_port, ("127.0.0.1", http_port))
        socks.allow("localhost", http_port, ("127.0.0.1", http_port))
        socks.allow(LAB_HOST, http_port, ("127.0.0.1", http_port))
        owned["socks"] = socks
        socks.start()
        held = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        held.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        held.bind(("127.0.0.1", 0))
        held_port = int(held.getsockname()[1])
        owned["held"] = held
        closed_listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        closed_listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        closed_listener.bind(("127.0.0.1", 0))
        refuse_port = int(closed_listener.getsockname()[1])
        closed_listener.listen(1)
        closed_listener.close()

        http_url = f"http://{LAB_HOST}:{http_port}/"
        https_url = f"https://{LAB_HOST}:{https_port}/"
        http_ip = f"http://127.0.0.1:{http_port}/"
        slow_url = f"http://{LAB_HOST}:{http_port}/slow-body"
        auth_url = f"http://{LAB_HOST}:{http_port}/origin-auth"
        encoded_pass = quote(PROXY_PASS, safe="")
        alpha_auth = f"http://{quote(PROXY_USER, safe='')}:{encoded_pass}@127.0.0.1:{alpha.port}"
        bravo_auth = f"http://{quote(PROXY_USER, safe='')}:{encoded_pass}@127.0.0.1:{bravo.port}"
        alpha_bare = f"http://127.0.0.1:{alpha.port}"
        wrong_auth = f"http://{quote(PROXY_USER, safe='')}:{quote('wrong-pass', safe='')}@127.0.0.1:{alpha.port}"
        resolve_http = f"{LAB_HOST}:{http_port}:127.0.0.1"
        resolve_https = f"{LAB_HOST}:{https_port}:127.0.0.1"
        injected.append(
            {
                "branch": "resolver",
                "label": "curl --resolve maps curl-lab.origin to 127.0.0.1; not public DNS",
            }
        )

        def resolve_args() -> list[str]:
            return ["--resolve", resolve_http, "--resolve", resolve_https]

        def proxy_args(url: str) -> list[str]:
            return ["--proxy", url, "--noproxy", ""]

        capture("curl_version", stdout=curl_ver.stdout.strip(), exit=curl_ver.returncode)
        assert_true("curl_version_ok", curl_ver.returncode == 0 and "libcurl/" in curl_ver.stdout, curl_ver.stdout.splitlines()[0])
        assert_true("curl_has_https", "https" in curl_ver.stdout.lower(), "https protocol listed")

        r = run_curl(["--verbose", *resolve_args(), http_ip])
        capture("direct_http", **r, proxyEvents=len(alpha.snapshot()))
        assert_true("direct_http_200", r["exit"] == 0 and r["http_code"] == "200" and "origin-http" in r["body"], f"exit={r['exit']} code={r['http_code']}")
        assert_true("direct_http_no_proxy", len(alpha.snapshot()) == 0, "alpha idle")
        assert_true("direct_dash_q_first", r["firstOption"] == "-q", r["argv"][1] if r["argv"] else "")

        before = len(alpha.snapshot())
        r = run_curl(["--verbose", *resolve_args(), *proxy_args(alpha_auth), http_url])
        ev = alpha.snapshot()[before:]
        capture("explicit_http_ok", **r, events=ev)
        assert_true("explicit_http_ok_200", r["exit"] == 0 and r["http_code"] == "200" and "origin-http" in r["body"], f"exit={r['exit']} code={r['http_code']}")
        assert_true(
            "explicit_http_used_alpha",
            any(e["method"] == "GET" and e["authorized"] and LAB_HOST in e["target"] for e in ev),
            f"events={ev}",
        )
        assert_true("explicit_http_no_leak", all(flag is False for flag in http_origin.leak_flags), "origin leak flags")

        before = len(alpha.snapshot())
        r = run_curl(["--verbose", *resolve_args(), *proxy_args(alpha_bare), http_url])
        ev = alpha.snapshot()[before:]
        capture("http_auth_missing", **r, events=ev)
        assert_true("http_auth_missing_407", r["http_code"] == "407", f"code={r['http_code']} exit={r['exit']}")
        assert_true("http_auth_missing_exit0", r["exit"] == 0, f"exit={r['exit']} without --fail")
        assert_true("http_auth_missing_reason", any(e["authReason"] == "missing" for e in ev), f"events={ev}")

        r = run_curl(["--fail", *resolve_args(), *proxy_args(alpha_bare), http_url])
        capture("http_auth_missing_fail", **r)
        assert_true("http_auth_missing_fail_22", r["exit"] == 22, f"exit={r['exit']} expected 22")

        before = len(alpha.snapshot())
        r = run_curl(["--verbose", *resolve_args(), *proxy_args(wrong_auth), http_url])
        ev = alpha.snapshot()[before:]
        capture("http_auth_wrong", **r, events=ev)
        assert_true("http_auth_wrong_407", r["http_code"] == "407", f"code={r['http_code']} exit={r['exit']}")
        assert_true("http_auth_wrong_reason", any(e["authReason"] == "wrong" for e in ev), f"events={ev}")

        before = len(alpha.snapshot())
        r = run_curl(
            [
                "--verbose",
                *resolve_args(),
                "--proxy",
                alpha_bare,
                "--proxy-user",
                f"{PROXY_USER}:{PROXY_PASS}",
                "--noproxy",
                "",
                http_url,
            ]
        )
        ev = alpha.snapshot()[before:]
        capture("http_proxy_user_special_chars", **r, events=ev)
        assert_true(
            "http_proxy_user_special_ok",
            r["exit"] == 0 and r["http_code"] == "200" and any(e["authReason"] == "ok" for e in ev),
            f"exit={r['exit']} code={r['http_code']}",
        )

        before = len(alpha.snapshot())
        r = run_curl(
            ["--verbose", *resolve_args(), *proxy_args(alpha_auth), "--cacert", str(certs["ca_pem"]), https_url]
        )
        ev = alpha.snapshot()[before:]
        capture("https_connect_ok", **r, events=ev)
        assert_true("https_connect_200", r["exit"] == 0 and r["http_code"] == "200" and "origin-https" in r["body"], f"exit={r['exit']} body={r['body'][:80]}")
        assert_true(
            "https_connect_method",
            any(e["method"] == "CONNECT" and e["authorized"] and str(https_port) in e["target"] for e in ev),
            f"events={ev}",
        )
        assert_true("https_connect_status", r["exit"] == 0 and r["http_connect"] == "200", f"exit={r['exit']} http_connect={r['http_connect']}")

        before = len(alpha.snapshot())
        r = run_curl(["--verbose", *resolve_args(), *proxy_args(alpha_bare), "--cacert", str(certs["ca_pem"]), https_url])
        ev = alpha.snapshot()[before:]
        capture("https_auth_missing", **r, events=ev)
        assert_true("https_auth_missing_56_407", r["exit"] == 56 and r["http_connect"] == "407" and r["http_code"] == "000" and "origin-https" not in r["body"], f"exit={r['exit']} connect={r['http_connect']} code={r['http_code']}")
        assert_true("https_auth_missing_reason", any(e["method"] == "CONNECT" and e["authReason"] == "missing" for e in ev), f"events={ev}")

        before = len(alpha.snapshot())
        r = run_curl(["--verbose", *resolve_args(), *proxy_args(wrong_auth), "--cacert", str(certs["ca_pem"]), https_url])
        ev = alpha.snapshot()[before:]
        capture("https_auth_wrong", **r, events=ev)
        assert_true("https_auth_wrong_56_407", r["exit"] == 56 and r["http_connect"] == "407" and r["http_code"] == "000" and "origin-https" not in r["body"], f"exit={r['exit']} connect={r['http_connect']} code={r['http_code']}")
        assert_true("https_auth_wrong_reason", any(e["method"] == "CONNECT" and e["authReason"] == "wrong" for e in ev), f"events={ev}")

        before = len(alpha.snapshot())
        r = run_curl(["--verbose", *resolve_args(), *proxy_args(alpha_auth), https_url])
        ev = alpha.snapshot()[before:]
        capture("https_untrusted_ca", **r, events=ev)
        assert_true("https_untrusted_exit60", r["exit"] == 60, f"exit={r['exit']}")
        assert_true(
            "https_untrusted_still_connect",
            any(e["method"] == "CONNECT" and e["authorized"] for e in ev),
            f"events={ev}",
        )

        r = run_curl(["--verbose", *resolve_args(), *proxy_args(alpha_auth), auth_url])
        capture("origin_auth_missing", **r)
        assert_true("origin_auth_401", r["http_code"] == "401", f"code={r['http_code']}")

        r = run_curl(
            [
                "--verbose",
                *resolve_args(),
                *proxy_args(alpha_auth),
                "--user",
                f"{ORIGIN_USER}:{ORIGIN_PASS}",
                auth_url,
            ]
        )
        capture("origin_auth_ok", **r)
        assert_true("origin_auth_200", r["exit"] == 0 and r["http_code"] == "200" and "origin-auth-ok" in r["body"], f"code={r['http_code']}")

        before_a = len(alpha.snapshot())
        before_b = len(bravo.snapshot())
        r = run_curl(["--verbose", *resolve_args(), http_url], env={**baseline_env, "HTTP_PROXY": alpha_auth, "NO_PROXY": ""})
        ev_a = alpha.snapshot()[before_a:]
        ev_b = bravo.snapshot()[before_b:]
        capture("HTTP_PROXY_uppercase_ignored", **r, alpha=ev_a, bravo=ev_b, envNames=["HTTP_PROXY", "NO_PROXY"])
        assert_true("HTTP_PROXY_ignored_direct", r["exit"] == 0 and r["http_code"] == "200", f"exit={r['exit']}")
        assert_true("HTTP_PROXY_alpha_idle", len(ev_a) == 0, f"alpha={ev_a}")

        before_a = len(alpha.snapshot())
        r = run_curl(["--verbose", *resolve_args(), http_url], env={**baseline_env, "http_proxy": alpha_auth, "NO_PROXY": ""})
        ev_a = alpha.snapshot()[before_a:]
        capture("http_proxy_lowercase_used", **r, events=ev_a, envNames=["http_proxy", "NO_PROXY"])
        assert_true("http_proxy_used_200", r["exit"] == 0 and r["http_code"] == "200", f"exit={r['exit']}")
        assert_true("http_proxy_alpha_got_get", any(e["method"] == "GET" and e["authorized"] for e in ev_a), f"events={ev_a}")

        before_a = len(alpha.snapshot())
        r = run_curl(["--verbose", *resolve_args(), "--cacert", str(certs["ca_pem"]), https_url], env={**baseline_env, "HTTPS_PROXY": alpha_auth, "NO_PROXY": ""})
        ev_a = alpha.snapshot()[before_a:]
        capture("HTTPS_PROXY_connect", **r, events=ev_a, envNames=["HTTPS_PROXY", "NO_PROXY"])
        assert_true("HTTPS_PROXY_200", r["exit"] == 0 and r["http_code"] == "200", f"exit={r['exit']}")
        assert_true("HTTPS_PROXY_connect_logged", any(e["method"] == "CONNECT" for e in ev_a), f"events={ev_a}")

        before_a = len(alpha.snapshot())
        r = run_curl(["--verbose", *resolve_args(), http_url], env={**baseline_env, "ALL_PROXY": alpha_auth, "NO_PROXY": ""})
        ev_a = alpha.snapshot()[before_a:]
        capture("ALL_PROXY_used", **r, events=ev_a, envNames=["ALL_PROXY", "NO_PROXY"])
        assert_true("ALL_PROXY_200", r["exit"] == 0 and r["http_code"] == "200", f"code={r['http_code']}")
        assert_true("ALL_PROXY_alpha", any(e["proxyId"] == "alpha" for e in ev_a), f"events={ev_a}")

        before_a = len(alpha.snapshot())
        before_b = len(bravo.snapshot())
        r = run_curl(["--verbose", *resolve_args(), *proxy_args(bravo_auth), http_url], env={**baseline_env, "http_proxy": alpha_auth, "NO_PROXY": ""})
        ev_a = alpha.snapshot()[before_a:]
        ev_b = bravo.snapshot()[before_b:]
        capture("explicit_beats_env", **r, alpha=ev_a, bravo=ev_b)
        assert_true("explicit_beats_env_bravo", any(e["method"] == "GET" and e["authorized"] for e in ev_b), f"bravo={ev_b}")
        assert_true("explicit_beats_env_alpha_idle", len(ev_a) == 0, f"alpha={ev_a}")

        before_a = len(alpha.snapshot())
        r = run_curl(["--verbose", *resolve_args(), http_url], env={**baseline_env, "http_proxy": alpha_auth, "NO_PROXY": LAB_HOST})
        ev_a = alpha.snapshot()[before_a:]
        capture("NO_PROXY_bypass", **r, events=ev_a, envNames=["http_proxy", "NO_PROXY"])
        assert_true("NO_PROXY_direct_200", r["exit"] == 0 and r["http_code"] == "200", f"code={r['http_code']}")
        assert_true("NO_PROXY_alpha_idle", len(ev_a) == 0, f"alpha={ev_a}")

        before_a = len(alpha.snapshot())
        r = run_curl(["--verbose", *resolve_args(), "--noproxy", "", http_url], env={**baseline_env, "http_proxy": alpha_auth})
        ev_a = alpha.snapshot()[before_a:]
        capture("noproxy_empty_forces_proxy", **r, events=ev_a)
        assert_true("noproxy_empty_uses_alpha", any(e["method"] == "GET" for e in ev_a), f"events={ev_a}")

        before = len(socks.snapshot())
        r = run_curl(["--verbose", *resolve_args(), "--socks5", f"127.0.0.1:{socks.port}", http_ip])
        ev = socks.snapshot()[before:]
        capture("socks5_ip", **r, events=ev)
        assert_true("socks5_ip_200", r["exit"] == 0 and r["http_code"] == "200", f"exit={r['exit']}")
        assert_true("socks5_ip_atyp_ipv4", any(e.get("atyp") == "ipv4" for e in ev), f"events={ev}")

        before = len(socks.snapshot())
        r = run_curl(["--verbose", *resolve_args(), "--socks5-hostname", f"127.0.0.1:{socks.port}", http_url])
        ev = socks.snapshot()[before:]
        capture("socks5h_host", **r, events=ev)
        assert_true("socks5h_host_200", r["exit"] == 0 and r["http_code"] == "200", f"exit={r['exit']}")
        assert_true(
            "socks5h_host_atyp_domain",
            any(e.get("atyp") == "domain" and e.get("host") == LAB_HOST for e in ev),
            f"events={ev}",
        )

        before = len(socks.snapshot())
        r = run_curl(["--verbose", *resolve_args(), "--socks5", f"127.0.0.1:{socks.port}", http_url])
        ev = socks.snapshot()[before:]
        atyp = ev[-1]["atyp"] if ev else None
        capture("socks5_host_measured", **r, events=ev, atyp=atyp)
        assert_true("socks5_host_200", r["exit"] == 0 and r["http_code"] == "200", f"exit={r['exit']} atyp={atyp}")
        assert_true("socks5_host_atyp_ipv4", atyp == "ipv4", f"atyp={atyp}")

        r = run_curl(["--verbose", f"http://127.0.0.1:{refuse_port}/"])
        capture("refusal_closed_listener", **r, port=refuse_port)
        assert_true("refusal_exit7", r["exit"] == 7, f"exit={r['exit']} stderr={r['stderr'][:120]}")
        assert_true("refusal_not_timeout", r["exit"] != 28, "connection error is not transfer timeout")

        r = run_curl(
            ["--verbose", "--connect-timeout", CONNECT_TIMEOUT_HELD, "--max-time", "2", f"http://127.0.0.1:{held_port}/"]
        )
        capture("held_bind_no_listen", **r, port=held_port)
        injected.append(
            {
                "branch": "bound_no_listen",
                "label": "Owned bind-without-listen kept during request. Darwin often waits instead of RST.",
                "exit": r["exit"],
                "stderr": r["stderr"][:160],
            }
        )
        assert_true("held_exit28_connect_timeout", r["exit"] == 28, f"exit={r['exit']} not refusal7")
        assert_true("held_not_refusal7", r["exit"] != 7, f"exit={r['exit']}")

        shared["first_byte"].clear()
        shared["slow_continue"].clear()
        t0 = time.monotonic()
        r = run_curl(
            ["--verbose", *resolve_args(), *proxy_args(alpha_auth), "--max-time", MAX_TIME_SLOW, slow_url],
            timeout=8.0,
        )
        elapsed = time.monotonic() - t0
        saw_first = shared["first_byte"].is_set()
        shared["slow_continue"].set()
        capture("slow_body_max_time", **r, firstByteEvent=saw_first, elapsed=elapsed)
        assert_true("slow_body_first_byte", saw_first, "headers and first body byte flushed before wait")
        assert_true("slow_body_exit28", r["exit"] == 28, f"exit={r['exit']} elapsed={elapsed:.3f}")
        assert_true("slow_body_is_transfer_timeout", r["exit"] == 28 and saw_first, "exit 28 after first byte is max-time, not connect-timeout")

        leaked_any = any(http_origin.leak_flags) or any(https_origin.leak_flags)
        capture("origin_proxy_auth_leak", leaked=leaked_any, httpFlags=len(http_origin.leak_flags))
        assert_true("no_proxy_auth_to_origin", leaked_any is False, f"leaked={leaked_any}")

        failed = [a for a in assertions if not a["passed"]]
        figure_rows = [
            {
                "name": a["name"],
                "kind": a.get("kind", "assert"),
                "result": "pass" if a["passed"] else "fail",
                "passed": a["passed"],
                "detail": a.get("detail", ""),
            }
            for a in assertions
        ]
        fig = render_figure(figure_rows, versions, started)
        payload = {
            "ok": not failed,
            "ranAt": started,
            "finishedAt": utc_now(),
            "exitStatus": 0 if not failed else 1,
            "versions": versions,
            "command": [sys.executable, str(Path(__file__).resolve())],
            "counts": {
                "captured": len(captured),
                "transfers": sum(1 for row in captured if "argv" in row),
                "metadataRows": sum(1 for row in captured if "argv" not in row),
                "asserted": len(assertions),
                "passed": sum(1 for a in assertions if a["passed"]),
                "failed": len(failed),
                "injected": len(injected),
            },
            "failedNames": [a["name"] for a in failed],
            "captured": captured,
            "assertions": assertions,
            "injected": injected,
            "figure": fig,
            "notes": {
                "noPasswordsInOutput": True,
                "privateKeysDeleted": True,
                "dashQFirst": True,
                "noInsecure": True,
                "httpsProxyLayer": "unmeasured",
                "schemeOverAllProxy": "documented/unmeasured; http_proxy and ALL_PROXY tested separately, not simultaneously",
                "resolver": "curl --resolve for curl-lab.origin only",
                "envIsolation": "immutable baseline_env copies per curl; no global os.environ writes",
            },
        }
        write_results(payload)
        return 0 if not failed else 1
    except Exception as exc:
        payload = {
            "ok": False,
            "ranAt": started,
            "finishedAt": utc_now(),
            "exitStatus": 2,
            "versions": versions,
            "error": {"type": type(exc).__name__, "message": redact_text(str(exc))},
            "traceback": redact_text(traceback.format_exc()),
            "counts": {
                "captured": len(captured),
                "asserted": len(assertions),
                "passed": sum(1 for a in assertions if a["passed"]),
                "failed": sum(1 for a in assertions if not a["passed"]),
                "injected": len(injected),
            },
            "captured": captured,
            "assertions": assertions,
            "injected": injected,
        }
        if os.environ.get(PARTIAL_HTTPS_FAIL) == "1":
            RAW.mkdir(parents=True, exist_ok=True)
            (RAW / "partial-https-fail.json").write_text(json.dumps(json_safe(payload), indent=2, sort_keys=True))
        else:
            write_results(payload)
        return 2
    finally:
        shared_obj = locals().get("shared")
        if isinstance(shared_obj, dict):
            shared_obj.get("slow_continue", threading.Event()).set()
        for key in ("alpha", "bravo", "socks"):
            obj = owned.get(key)
            if obj is not None:
                try:
                    obj.stop()
                except Exception:
                    pass
        for key in ("http_origin", "https_origin"):
            srv = owned.get(key)
            if srv is not None:
                stop_origin(srv)
        held_sock = owned.get("held")
        if held_sock is not None:
            try:
                held_sock.close()
            except Exception:
                pass
        for name in ("ca.key", "origin.key", "netrc", "ca.srl", "origin.csr", "san.cnf"):
            path = tmp / name
            if path.exists():
                path.unlink()
        try:
            shutil.rmtree(tmp, ignore_errors=True)
        except Exception:
            pass


def main() -> int:
    if os.environ.get(CHILD_FLAG) != "1":
        return launch_child()
    return run_lab()


if __name__ == "__main__":
    sys.exit(main())