Proxy no curl: configuração, autenticação e diagnóstico

Configurar proxy HTTP no curl, conferir CONNECT em HTTPS, separar -U de -u e ler recusa e timeout pelos códigos de saída.

Este texto é para quem precisa mandar o curl por um proxy HTTP ou SOCKS5 e conferir o caminho. Os exemplos usam credenciais dummy. A Top Proxy não vende IP e não distribui lista grátis. Um 200 local não prova geo, conta de serviço nem qualidade de provedor.

01Como passar um proxy HTTP no curl#

A binária deste Mac foi curl 8.7.1 com libcurl 8.7.1, SecureTransport e LibreSSL 3.3.6. Os recursos listados incluem SSL, HTTP2 e HTTPS-proxy. Essa última palavra é uma flag de compilação. Não é um hop TLS até o proxy medido neste laboratório. O laboratório usou só um proxy HTTP mais origens locais.

O capítulo HTTP proxy do everything curl (lido em 14 de outubro de 2026) é direto. -x ou --proxy assume HTTP. Sem porta, o curl cai na 1080 por hábito histórico. Um destino https:// não pede https:// na URL do proxy. O proxy fala HTTP. O destino pode ser TLS.

Comece pelo explícito. Não deixe a shell herdar um proxy silencioso e chame isso de teste. curl -q precisa ser a primeira opção. -q desliga a curlrc padrão. Sem isso, um arquivo no home injeta proxy, CA ou senha e você acha que o argv está limpo. O laboratório não mudou HOME. O processo filho copia o ambiente, tira nomes *_proxy sem olhar maiúscula, cabeçalhos CGI HTTP_*, REMOTE_USER, AUTH_TYPE e qualquer OPENSSL_* herdado, e só depois injeta o experimento. -q continua a primeira opção. Sem --netrc e sem -k.

O curl não tem o interruptor trust_env da biblioteca Requests. Pedir trust_env=False no Python também muda CA e netrc, não só a ordem do proxy. Aqui o equivalente é -q, um netrc vazio privado e o ambiente do filho sem CURL_CA_BUNDLE, SSL_CERT_FILE e parentes. Não use isso como desculpa para -k.

1. Leia curl -q --version. Manual online descreve opção que 8.7.1 talvez não tenha.
2. Trate URL do proxy e URL do destino como duas coisas. Esquema do hop e esquema do site não são o mesmo botão.
3. Rode curl -q --proxy http://127.0.0.1:PORTA --noproxy '' http://destino.example/ contra um host que você controla.
4. Abra o verbose. O pedido ao proxy vem em forma absoluta, GET http://destino..., não só o caminho da origem.

--noproxy '' é de propósito. Algumas instalações pulam loopback. Se o caso exige o proxy, esse pulo não pode esconder uma asserção falha. Bypass se testa à parte.

Sem --proxy, o curl bateu na origem HTTP com 200. O proxy de loopback ficou com zero eventos. Isso é o controle direto. Não diz o que o curl faria com http_proxy ainda setado. Isso vem depois.

As linhas abaixo são placeholder de provedor. Não sobem fixture, não baixam nada deste site e não substituem o arquivo do laboratório. PORTA, USUARIO e a senha são do seu proxy.

# PLACEHOLDER, nao e o laboratorio executado
curl -q --proxy "http://127.0.0.1:PORTA" --noproxy '' \
  --proxy-user "USUARIO:SENHA" \
  "http://example.invalid/"

O laboratório completo está no bloco Python no final deste artigo. Não é um download público. Copie o bloco para lab.py. Pin: Pillow==12.1.1. Python 3.14. openssl e cwebp no PATH. O arquivo cria assets, evidence e evidence/raw sozinho antes de copiar a CA.

1. python3.14 -c "import sys,PIL; print(sys.version.split()[0]); print(PIL.__version__)" deve mostrar 3.14.6 e 12.1.1 neste run.
2. openssl version e cwebp -version precisam existir. Aqui foram LibreSSL 3.3.6 e cwebp 1.6.0.
3. Diretório de trabalho: a pasta que contém lab.py. Se o arquivo ficar numa pasta chamada tests, assets e evidence nascem no pai. Caso contrário nascem ao lado de lab.py.
4. python3.14 -m pip install Pillow==12.1.1 e em seguida python3.14 lab.py.
5. A falha suportada de startup HTTPS parcial é CURL_PT_PARTIAL_HTTPS_FAIL=1 python3.14 lab.py e sai 2.

Um 200 no stdout é pouco. O proxy precisa encaminhar a resposta real da origem. Um CONNECT falso sem túnel não conta como proxy HTTP.

02Como um destino HTTPS usa CONNECT#

Muita gente chama isso de HTTPS-proxy porque o site começa com https://. O hop continua HTTP falando CONNECT. O everything curl descreve o túnel. O curl manda CONNECT host:porta. O proxy abre TCP. O TLS corre entre cliente e origem.

Foi o que o laboratório viu. Com dummy correto, o CONNECT chegou na origem HTTPS local. Depois veio status 200 e o corpo com a identidade da origem. http_connect foi 200. O proxy passou bytes TLS. Não inventou um 200 sem túnel.

--cacert vale para o TLS até a origem. Não governa, no silêncio, uma segunda camada TLS até o proxy. --proxy-cacert seria o arquivo de um proxy HTTPS. Este build lista HTTPS-proxy nos recursos. Se o seu hop fala TLS com o proxy, este laboratório não mediu. Não tire isso da linha de Features.

1. Deixe a URL do proxy em http://127.0.0.1:PORTA mesmo quando o destino é https://.
2. Passe a CA da origem com --cacert. -k não é o caso normal.
3. Procure CONNECT host:porta no log do proxy antes de comemorar o 200.
4. Repita o mesmo comando sem --cacert. Você quer falha de TLS depois do CONNECT, não "proxy morto".

O último passo é o que as pessoas pulam. Sem a CA do laboratório, o CONNECT ainda autorizou. O curl saiu com código 60, certificado do par não verificado. O túnel existia. O certificado, não. A página Meu IP mostra a saída do navegador, não deste processo curl. O verificador de proxy diz se um host aceita HTTP ou SOCKS5 antes de você apontar o curl.

A figura é a tabela deste run. Não é capa 16:9 e não é print do verificador.

03Como separar senha do proxy e senha do destino#

-U ou --proxy-user é autenticação do proxy. -u ou --user é autenticação da origem. Quem mistura as duas gasta uma hora no andar errado.

O capítulo Proxy authentication (mesmo dia) trata 407 quando o proxy recusa. O laboratório mostra que 407 não é um único comportamento.

Forward HTTP sem usuário devolveu status 407 e exit 0. Sem --fail, o curl trata 407 como transferência concluída com status feio. Com --fail, o mesmo 407 virou exit 22. Isso é 8.7.1 nesta binária. Não é lei universal de todo 407.

Dummy errada de novo 407, motivo wrong no fixture. Dummy correta, inclusive caracteres @ : / # % via --proxy-user depois do primeiro dois-pontos, devolveu origem 200. Userinfo na URL do proxy precisa de percent-encoding por componente. Senão o parser come a senha num @.

CONNECT sem login foi outro filme. O proxy gravou authReason: missing. O curl saiu 56. http_connect foi 407. Não houve corpo da origem. CONNECT é montagem de túnel. 407 ali é erro de transporte, não HTML.

CONNECT com dummy errada: de novo exit 56, http_connect 407, motivo wrong. Dummy correta: origem 200 depois de CONNECT 200.

A origem não viu Proxy-Authorization. Esse cabeçalho fica no hop do proxy. Basic da origem ficou em /origin-auth. Sem -u, status 401. Com -u e -U juntos, status 200.

1. Senha com caractere especial em --proxy-user depois do primeiro dois-pontos, ou encodada na URL do proxy.
2. -u só para o site, nunca como substituto de -U.
3. Anote 407 HTTP com exit 0 e 407 de CONNECT com exit 56 em colunas diferentes.
4. Apague credenciais do verbose antes de guardar o log. Tire userinfo da URL, senha certa, senha errada e o cabeçalho Basic completo. Não deixe Authorization nem Proxy-Authorization em captura.

Produtos residenciais documentam a mesma separação sem este laboratório ter batido nos pools. Oxylabs Residential Proxies (14 de outubro de 2026) cita HTTP, HTTPS, HTTP/3 e SOCKS5, usuário e senha ou IP na allowlist, entrada pr.oxylabs.io:7777 no formato customer-USERNAME:PASSWORD, e um exemplo curl com -x e -U. Isso é forma documentada, não corrida medida contra o gateway deles. Decodo Residential Quick Start (mesmo dia) mostra gate.decodo.com:7000 e curl -U 'USERNAME:PASSWORD' -x gate.decodo.com:7000. O catálogo ainda publica o produto no caminho smartproxy-com. O nome da documentação é Decodo. O caminho no site não mudou por causa do nome novo.

04Como o curl lê httpproxy, HTTPPROXY e NOPROXY#

O capítulo variáveis de ambiente (14 de outubro de 2026) é o texto que falta na maior parte dos gists. http_proxy em minúsculas manda HTTP pelo proxy. HTTPS_PROXY e ALL_PROXY aceitam maiúsculas. HTTP_PROXY em maiúsculas o curl recusa de propósito. CGI prefixa cabeçalhos com HTTP_. Um Proxy: na requisição viraria HTTP_PROXY no script. Aceitar isso já quebrou software demais.

O laboratório injetou nomes e valores dummy a partir de um filho limpo. Não inferiu rota só pelo status. Dois proxies instrumentados, alpha e bravo, mostraram quem recebeu o GET.

HTTP_PROXY em maiúsculas com NO_PROXY vazio: origem 200, alpha ocioso. O curl ignorou a variável. http_proxy em minúsculas com NO_PROXY='': origem 200 e GET autorizado no alpha. HTTPS_PROXY em maiúsculas no destino HTTPS: CONNECT no alpha e 200 com CA local. ALL_PROXY no HTTP: GET no alpha.

--proxy ganha da variável. Com http_proxy apontando para alpha e --proxy apontando para bravo, só o bravo viu o GET. Alpha ficou zerado. Precedência não se lê num 200 solto.

Bypass foi caso próprio. http_proxy no alpha e NO_PROXY com o hostname do laboratório: 200 direto, alpha ocioso. O contrário, --noproxy '' com a mesma variável, forçou o GET no alpha. Loopback default não pode se passar por política.

1. Limpe o ambiente do processo filho. Não edite o proxy do workstation.
2. Teste http_proxy minúsculo e HTTP_PROXY maiúsculo em corridas separadas.
3. Use dois proxies se a pergunta é quem ganhou, não se o status foi 200.
4. Trate NO_PROXY e --noproxy '' como experimentos, não como efeito colateral.

Cada pedido do laboratório parte dessa cópia limpa e só então recebe os nomes dummy do experimento. Netrc vazio fica num arquivo temporário do próprio run. HOME não muda. Curlrc extra morre no -q.

05Como SOCKS5 e socks5h escolhem quem resolve o nome#

SOCKS no everything curl (mesmo dia) desenha a tabela. SOCKS5 resolve no cliente. socks5h ou --socks5-hostname manda o hostname ao proxy. Um GET SOCKS que dá 200 não localiza o DNS. Precisa do ATYP no handshake.

O fixture SOCKS5 do laboratório gravou o ATYP e mapeou hostname controlado para loopback. --resolve mandou curl-pt.origin para 127.0.0.1. Isso é injeção de resolvedor no próprio curl, não vazamento DNS medido na rede pública. O arquivo hosts da máquina ficou quieto.

--socks5 contra IP: 200, ATYP ipv4 em 127.0.0.1 mapeado. --socks5-hostname contra o hostname de laboratório: 200, ATYP domain com curl-pt.origin. --socks5 contra o mesmo hostname, com --resolve local, mandou ipv4 127.0.0.1. Isso é resolução no cliente, não no proxy.

1. Escolha --socks5 ou --socks5-hostname de propósito. Não misture os dois no mesmo diagnóstico.
2. Olhe ATYP. ipv4 ou ipv6 no proxy significa endereço já resolvido. domain significa nome enviado.
3. Se o nome for de laboratório, diga que o resolvedor foi injetado.
4. SOCKS5 no provedor e tipo de IP residencial ou móvel são eixos diferentes. Um não prova o outro.

Quem precisa de saída móvel ou residencial olha o catálogo, não o ATYP. Proxies móveis e proxies residenciais descrevem origem do IP. Este tutorial descreve o cliente de linha de comando.

06Como ler recusa, --connect-timeout e --max-time#

O capítulo de timeouts (14 de outubro de 2026) separa duas alavancas. --connect-timeout limita o setup. A página oficial inclui a consulta DNS e o handshake TCP, TLS ou QUIC que o pedido pediu. --max-time limita a operação inteira, inclusive no meio da transferência. Os dois podem sair 28. O número sozinho não conta a história. Este curl 8.7.1 não lista HTTP3 nos Features. QUIC não foi medido neste laboratório. Não dá para copiar o trecho QUIC do manual como resultado da binária.

Recusa verdadeira neste laboratório foi um listener de loopback que o processo abriu, escutou e fechou, depois o curl tentou a mesma porta. Exit 7, connection refused. Isso é erro de conexão. Não é timeout de connect. Não use 192.0.2.1 nem buraco público para fingir recusa.

O outro soquete ficou bindado em porta efêmera, sem listen, e ficou assim até o pedido acabar. No Darwin desta máquina o curl não viu RST na hora. Saiu 28 com --connect-timeout 1. É condição local do fixture, timeout de connect nesta porta sem fila. Não é recusa 7. Não é a injeção --resolve. Não é timeout de internet.

/slow-body mandou cabeçalhos completos, o primeiro byte do corpo, sinalizou o evento, e só então esperou além de --max-time 1. O evento do primeiro byte disparou. O curl saiu 28 depois de cerca de um segundo. Isso é timeout de transferência depois do corpo ter começado. Não é timeout de leitura estilo Requests. Não é --connect-timeout. --max-time neste curl é teto total.

1. Anote exit, http_code, http_connect e um pedaço seguro do stderr.
2. Recusa fica exit 7 nesta binária. Timeout de transferência depois do primeiro byte fica 28.
3. Não troque recusa observada por uma tese mais forte de timeout.
4. Se for falar de low-speed, rode um caso longo de verdade. Este laboratório não inventou espera de relógio para isso.

Retries, se um dia você ligar, merecem bound de tentativas e de tempo. Uma flag de shell não prova idempotência do POST.

07Como escolher um endpoint documentado#

Depois que o cliente está sob controle, o próximo passo é um endpoint que o provedor realmente documenta. Protocolo suportado e origem do IP continuam eixos diferentes. Um 200 no laboratório local não diz que o pool residencial do provedor atende o Brasil, nem que a conta passa no login do app.

As páginas públicas Oxylabs e Decodo, ainda listado em smartproxy-com abrem nesses endereços. Isso mostra a ficha no site. Não prova que a conta paga conecta.

Oxylabs documenta pr.oxylabs.io:7777 com usuário no formato customer-USERNAME e senha, ou IP na allowlist. Decodo documenta gate.decodo.com:7000 com usuário e senha ou IP na allowlist, e o exemplo curl usa -U e -x. Nenhum dos dois foi alvo deste laboratório. Trate os exemplos oficiais como forma, não como latência.

Antes de gastar tráfego, confira host, porta e protocolo no verificador. Se a dúvida for "qual IP o navegador mostra", use Meu IP e não misture com o argv do curl.

Oxylabs: proxies residenciais e de datacenter4.0770

A Oxylabs vende proxies residenciais, de datacenter e móveis. Residencial self-serve começa em $6/GB para 5 GB/mês. Pagamentos de comprador: cartões, TED, AliPay, PayPal.

10 países

US$ 0,59

por 1 GB

Decodo (Smartproxy): proxies residenciais e de datacenter4.22106

A Decodo (antes Smartproxy) vende proxies residenciais e de datacenter. Residencial: $4/GB sem plano mensal, ou pacotes mensais a partir de $3.75/GB para 3 GB. Pagamentos: cartões, PayPal, Alipay, Google/Apple Pay, cripto (com limites).

83 países

US$ 0,38

por 1 GB

Se o destino for residencial, comece pela página de proxies residenciais. Se for saída móvel, use proxies móveis. Este artigo não escolhe o pool por você. Só mostra como o curl fala com o hop.

08Fonte completa do laboratório#

O bloco abaixo é o arquivo único. Copie para lab.py. Pillow==12.1.1. Não é um anexo público nem um URL para baixar.

#!/usr/bin/env python3
"""Self-contained curl proxy lab for curl-pt. 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
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_PT_LAB_CHILD"
PARTIAL_HTTPS_FAIL = "CURL_PT_PARTIAL_HTTPS_FAIL"
LAB_HOST = "curl-pt.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"
CURL_BIN = os.environ.get("CURL_BIN", "/usr/bin/curl")

WRONG_PASS = "wrong-pass"
CHILD_TIMEOUT_S = 90
OPENSSL_TIMEOUT_S = 20
CWEBP_TIMEOUT_S = 20

CGI_EXACT = {
    "REQUEST_METHOD",
    "GATEWAY_INTERFACE",
    "CONTENT_TYPE",
    "CONTENT_LENGTH",
    "QUERY_STRING",
    "SCRIPT_NAME",
    "PATH_INFO",
    "SERVER_PROTOCOL",
    "REMOTE_ADDR",
    "REMOTE_HOST",
    "REMOTE_USER",
    "REMOTE_IDENT",
    "AUTH_TYPE",
    "SERVER_NAME",
    "SERVER_PORT",
    "SERVER_SOFTWARE",
    "DOCUMENT_ROOT",
    "PATH_TRANSLATED",
    "REQUESTS_CA_BUNDLE",
    "CURL_CA_BUNDLE",
    "SSL_CERT_FILE",
    "SSL_CERT_DIR",
    "SSLKEYLOGFILE",
    "NETRC",
}


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


def _secret_forms() -> list[str]:
    forms: list[str] = []
    pairs = (
        (PROXY_USER, PROXY_PASS),
        (PROXY_USER, WRONG_PASS),
        (ORIGIN_USER, ORIGIN_PASS),
    )
    for user, pw in pairs:
        forms.extend(
            [
                pw,
                quote(pw, safe=""),
                quote(pw, safe=":/"),
                f"{user}:{pw}",
                f"{quote(user, safe='')}:{quote(pw, safe='')}",
                base64.b64encode(f"{user}:{pw}".encode()).decode(),
            ]
        )
    uniq: list[str] = []
    for item in forms:
        if item and item not in uniq:
            uniq.append(item)
    uniq.sort(key=len, reverse=True)
    return uniq


def redact_text(value: str) -> str:
    out = value
    for secret in _secret_forms():
        out = out.replace(secret, "[redacted]")
    out = re.sub(
        r"(?im)^(?P<prefix>[<>*\s]*)(?P<header>(?:proxy-)?authorization:)\s*.*$",
        lambda m: f"{m.group('prefix')}{m.group('header')} [redacted]",
        out,
    )
    out = re.sub(
        r"(?i)(?P<header>(?:proxy-)?authorization:)\s*\S.*",
        lambda m: f"{m.group('header')} [redacted]",
        out,
    )
    out = re.sub(r"(?i)(https?://)[^/@\s:]+:[^/@\s]+@", r"\1[redacted]@", out)
    return out


def env_key_is_dirty(key: str) -> bool:
    kl = key.lower()
    if key == "HOME":
        return False
    if kl.endswith("_proxy") or kl in {"no_proxy", "noproxy"}:
        return True
    if kl.startswith("openssl_"):
        return True
    if key in CGI_EXACT:
        return True
    if key.startswith("HTTP_"):
        return True
    return False


def clean_env_copy(src: dict[str, str] | None = None) -> dict[str, str]:
    env = dict(src if src is not None else os.environ)
    for key in list(env):
        if env_key_is_dirty(key):
            env.pop(key, None)
    return env


def blob_has_userinfo(text: str) -> bool:
    return re.search(r"(?i)https?://[^/@\s:]+:[^/@\s]+@", text) is not None


def blob_has_basic_token(text: str) -> bool:
    return re.search(r"(?i)(?:proxy-)?authorization:\s*basic\s+[A-Za-z0-9+/=]{8,}", text) is not None


def blob_has_residual_auth(text: str) -> bool:
    if blob_has_userinfo(text) or blob_has_basic_token(text):
        return True
    for match in re.finditer(r"(?im)^[<>*\s]*(?:proxy-)?authorization:\s*(.*)$", text):
        rest = match.group(1).strip()
        if rest and rest != "[redacted]":
            return True
    if re.search(r"(?i)(?:proxy-)?authorization:\s*(?!\[redacted\])\S", text):
        return True
    return False


def client_http_200(headers: str, http_code: str) -> bool:
    if http_code != "200":
        return False
    norm = (headers or "").replace("\r", "")
    return re.search(r"(?im)^HTTP/1\.[01].* 200\b", norm) is not None


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
        self._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
        self._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-pt.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-pt-lab-ca",
        ],
        check=True,
        capture_output=True,
        timeout=OPENSSL_TIMEOUT_S,
    )
    subprocess.run(
        [
            "openssl",
            "req",
            "-newkey",
            "rsa:2048",
            "-nodes",
            "-keyout",
            str(origin_key),
            "-out",
            str(origin_csr),
            "-subj",
            "/CN=localhost",
        ],
        check=True,
        capture_output=True,
        timeout=OPENSSL_TIMEOUT_S,
    )
    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,
        timeout=OPENSSL_TIMEOUT_S,
    )
    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}


class EnvScope:
    def __init__(self, inject: dict[str, str]) -> None:
        self.inject = inject
        self.prev: dict[str, str | None] = {}

    def __enter__(self) -> "EnvScope":
        for k, v in self.inject.items():
            self.prev[k] = os.environ.get(k)
            os.environ[k] = v
        return self

    def __exit__(self, *_exc: object) -> None:
        for k, old in self.prev.items():
            if old is None:
                os.environ.pop(k, None)
            else:
                os.environ[k] = old


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]:
    from PIL import Image, ImageDraw, ImageFont

    ASSETS.mkdir(parents=True, exist_ok=True)
    font_path = None
    for cand in (
        "/System/Library/Fonts/Supplemental/Verdana.ttf",
        "/Library/Fonts/Arial Unicode.ttf",
        "/System/Library/Fonts/Supplemental/Georgia.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 = ["Caso", "Resultado", "Detalhe"]
    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), "Laboratorio curl 8.7.1 em loopback, 14 de outubro de 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)
    subprocess.run(["cwebp", "-q", "78", str(FIGURE_PNG), "-o", str(FIGURE_WEBP)], check=True, capture_output=True, timeout=CWEBP_TIMEOUT_S)
    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 any(s in text for s in (PROXY_PASS, ORIGIN_PASS, WRONG_PASS, f"{PROXY_USER}:{WRONG_PASS}")):
        raise RuntimeError("refusing to write unredacted secret")
    if blob_has_userinfo(text) or blob_has_basic_token(text) or blob_has_residual_auth(text):
        raise RuntimeError("refusing to write unredacted auth header or userinfo")
    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_copy(os.environ)
    env[CHILD_FLAG] = "1"
    env["PYTHONDONTWRITEBYTECODE"] = "1"
    proc = subprocess.run(
        [sys.executable, str(Path(__file__).resolve())],
        env=env,
        timeout=CHILD_TIMEOUT_S,
    )
    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 | None = None
    netrc_path: Path | None = None
    versions: dict[str, Any] = {}
    curl_ver = None
    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 assert_redacted(tag: str, row: dict[str, Any]) -> None:
        blob = " ".join(str(x) for x in row.get("argv") or []) + "\n" + str(row.get("stderr") or "") + "\n" + str(row.get("stdout") or "")
        assert_true(f"{tag}_no_wrong_dummy", WRONG_PASS not in blob, "wrong dummy absent from argv/stderr", kind="correction")
        assert_true(f"{tag}_no_correct_dummy", PROXY_PASS not in blob and ORIGIN_PASS not in blob, "correct dummy absent", kind="correction")
        assert_true(f"{tag}_no_userinfo", not blob_has_userinfo(blob), "auth userinfo redacted", kind="correction")
        assert_true(f"{tag}_no_basic", not blob_has_basic_token(blob), "complete Basic header redacted", kind="correction")

    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 = clean_env_copy(os.environ)
        if netrc_path is not None:
            run_env["NETRC"] = str(netrc_path)
        if env is not None:
            run_env.update(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_text(a) for a in 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", ""),
            "ssl_verify_result": wo_parsed.get("ssl_verify_result", ""),
            "timedOut": timed_out,
            "firstOption": cmd[1],
        }

    try:
        tmp = Path(tempfile.mkdtemp(prefix="curl-pt-lab-"))
        owned["tmp"] = tmp
        netrc_path = tmp / "netrc"
        netrc_path.write_text("")
        netrc_path.chmod(stat.S_IRUSR | stat.S_IWUSR)
        os.environ["NETRC"] = str(netrc_path)
        curl_ver = subprocess.run([CURL_BIN, "-q", "--version"], capture_output=True, text=True, timeout=5)
        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"),
        }
        certs = make_certs(tmp)
        ASSETS.mkdir(parents=True, exist_ok=True)
        EVIDENCE.mkdir(parents=True, exist_ok=True)
        RAW.mkdir(parents=True, exist_ok=True)
        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])
        alpha = HttpForwardProxy("alpha", PROXY_USER, PROXY_PASS)
        owned["alpha"] = alpha
        bravo = HttpForwardProxy("bravo", PROXY_USER, PROXY_PASS)
        owned["bravo"] = bravo
        for proxy in (alpha, bravo):
            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))
            proxy.start()
        socks = Socks5Proxy("socks")
        owned["socks"] = 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))
        socks.start()
        held = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        owned["held"] = held
        held.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        held.bind(("127.0.0.1", 0))
        held_port = int(held.getsockname()[1])
        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-pt.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")
        probe_names = [
            "OPENSSL_CONF",
            "OPENSSL_MODULES",
            "OPENSSL_ENGINES",
            "OPENSSL_CONF_INCLUDE",
            "LDAP_PROXY",
            "ldap_proxy",
            "WSS_PROXY",
            "wss_proxy",
            "REMOTE_USER",
            "AUTH_TYPE",
            "HTTP_COOKIE",
            "HTTP_AUTHORIZATION",
            "HTTP_PROXY",
            "http_proxy",
        ]
        probe = {name: "dummy" for name in probe_names}
        probe["HOME"] = "keep-home"
        cleaned = clean_env_copy(probe)
        dropped = sorted(name for name in probe_names if name not in cleaned)
        assert_true("clean_env_drops_probe_names", dropped == sorted(probe_names), ",".join(dropped), kind="correction")
        assert_true("clean_env_keeps_home_name", cleaned.get("HOME") == "keep-home", "HOME retained without printing inherited env", kind="correction")

        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_redacted("explicit_http_ok", r)
        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}")
        assert_redacted("http_auth_wrong", r)
        assert_true("http_auth_wrong_redacted_marker", "[redacted]" in " ".join(r["argv"]), "argv keeps redaction marker", kind="correction")

        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_redacted("https_connect_ok", r)
        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["http_connect"] == "200" and r.get("ssl_verify_result") in {"0", ""} and r["exit"] == 0, f"http_connect={r['http_connect']} ssl={r.get('ssl_verify_result')}")

        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_nonzero", r["exit"] == 56 and r["http_code"] == "000" and r["http_connect"] == "407" and not r["body"], f"exit={r['exit']} code={r['http_code']} connect={r['http_connect']} body={r['body'][:40]}")
        assert_true("https_auth_missing_reason", any(e["method"] == "CONNECT" and e["authReason"] == "missing" and not e["authorized"] 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_nonzero", r["exit"] == 56 and r["http_code"] == "000" and r["http_connect"] == "407" and not r["body"], f"exit={r['exit']} code={r['http_code']} connect={r['http_connect']} body={r['body'][:40]}")
        assert_true("https_auth_wrong_reason", any(e["method"] == "CONNECT" and e["authReason"] == "wrong" and not e["authorized"] for e in ev), f"events={ev}")
        assert_redacted("https_auth_wrong", r)

        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={"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={"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={"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" and e["authorized"] for e in ev_a) and r["http_code"] == "200", f"events={ev_a}")

        before_a = len(alpha.snapshot())
        r = run_curl(["--verbose", *resolve_args(), http_url], 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" and e["method"] == "GET" and e["authorized"] for e in ev_a) and r["http_code"] == "200", 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={"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) and r["exit"] == 0 and r["http_code"] == "200" and "origin-http" in r["body"], f"bravo={ev_b} code={r['http_code']}")
        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={"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={"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" and e["authorized"] for e in ev_a) and r["exit"] == 0 and r["http_code"] == "200" and "origin-http" in r["body"], f"events={ev_a} code={r['http_code']}")

        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" and e.get("host") == "127.0.0.1" and e.get("mapped") 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 and e.get("mapped") 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_recorded", atyp == "ipv4" and any(e.get("atyp") == "ipv4" and e.get("host") == "127.0.0.1" and e.get("mapped") for e in ev), f"atyp={atyp} events={ev}")

        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_nonzero", r["exit"] == 28, f"exit={r['exit']}")
        assert_true("held_not_called_refusal_unless_7", r["exit"] == 28 and r["exit"] != 7 and "timeout" in r["stderr"].lower(), f"exit={r['exit']} stderr={r['stderr'][:120]}")

        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 and r["body"].startswith("H") and client_http_200(r["headers"], r["http_code"]), f"event={saw_first} body={r['body'][:20]!r} code={r['http_code']} hasStatusLine={bool(re.search(r'(?im)^HTTP/1', (r['headers'] or '').replace(chr(13), '')))}")
        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 and r["body"].startswith("H") and client_http_200(r["headers"], r["http_code"]), "exit 28 after first byte is max-time, not connect-timeout")

        contradiction_slow = {
            "headers": "",
            "http_code": "000",
            "body": r["body"],
            "exit": 28,
            "firstByte": True,
        }
        assert_true(
            "slow_body_rejects_missing_headers",
            not (
                contradiction_slow["firstByte"]
                and contradiction_slow["body"].startswith("H")
                and client_http_200(contradiction_slow["headers"], contradiction_slow["http_code"])
                and contradiction_slow["exit"] == 28
            ),
            "missing HTTP200 status line and code000 rejected",
            kind="replay",
        )
        nine = [
            not ({"http_connect": "599", "ssl_verify_result": "0", "exit": 0}["http_connect"] == "200" and {"http_connect": "599", "ssl_verify_result": "0", "exit": 0}["exit"] == 0 and {"http_connect": "599", "ssl_verify_result": "0", "exit": 0}["http_connect"] == "200"),
            not (124 == 56 and "000" == "000" and "407" == "407" and not ""),
            not (60 == 56 and "000" == "000" and "407" == "407" and not ""),
            not ("domain" == "ipv4"),
            not (124 == 28),
            not (0 == 28 and 0 != 7 and "timeout" in "ok"),
            not (28 == 28 and False and "H".startswith("H")),
            not (True and 1 == 0 and "200" == "200" and "origin-http" in "fail"),
            not (True and 0 == 0 and "407" == "200" and "origin-http" in "proxy-407"),
        ]
        assert_true("original_nine_contradictions_rejected", all(nine) and len(nine) == 9, "nine original contradictory records still rejected", kind="replay")

        token = base64.b64encode(bytes([65, 66, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76])).decode()
        samples = [
            "> Proxy-Authorization: Basic " + token,
            "< Authorization: Basic " + token,
            "* Proxy-Authorization: Basic " + token,
            "Proxy-Authorization: Basic " + token,
        ]
        unlisted_ok = True
        for sample in samples:
            red = redact_text(sample)
            if token in red or blob_has_residual_auth(red) or "[redacted]" not in red:
                unlisted_ok = False
                break
        assert_true("unlisted_auth_header_redacted", unlisted_ok, "unlisted Authorization values fully redacted before persist", kind="correction")

        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),
                "transferRows": sum(1 for c in captured if c.get("name") not in {"curl_version", "origin_proxy_auth_leak"}),
                "metadataRows": sum(1 for c in captured if c.get("name") in {"curl_version", "origin_proxy_auth_leak"}),
                "asserted": len(assertions),
                "historicalAsserted": 53,
                "correction01Asserted": 19,
                "correctionAsserted": sum(1 for a in assertions if a.get("kind") == "correction"),
                "replayAsserted": sum(1 for a in assertions if a.get("kind") == "replay"),
                "core72": 72,
                "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",
                "http3Quic": "unmeasured; this curl 8.7.1 Features line has no HTTP3",
                "resolver": "curl --resolve for curl-pt.origin only",
                "heldBindNoListen": "fixture-induced Darwin connect-timeout exit 28, not refusal 7 and not internet timeout",
                "countReading": "27 capture rows = 25 curl transfers + version/leak metadata; 72 core predicates = 53 historical + 19 correction01; correction02 adds residual-auth and contradiction replays without new transfers",
                "turnTainted": True,
            },
        }
        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
        if tmp is not None:
            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())