Configurar proxy no HTTPX: autenticação, ambiente e timeouts

Como passar proxy= no HTTPX, autenticar com caracteres especiais, usar HTTP_PROXY e NO_PROXY, e distinguir 407, ProxyError e timeouts.

Configurar proxy no HTTPX: autenticação, ambiente e timeouts

Este guia é para quem precisa encaminhar o HTTPX por um proxy HTTP com usuário e senha. Você monta o parâmetro proxy=, separa a autenticação do proxy da autenticação da origem e lê timeouts sem desligar o TLS. Não distribuímos chaves gratuitas nem vendemos IP. O laboratório abaixo usa só localhost.

01Como passar o proxy no Client HTTPX#

1. Instale o HTTPX na versão pinada e confira também o httpcore. O Client 0.28.1 expõe proxy=, não o dicionário proxies= que o Requests ainda usa. A assinatura instalada é Client(..., proxy=None, mounts=None, trust_env=True, verify=True, timeout=Timeout(timeout=5.0)). Se você copiar um snippet antigo com proxies={"http": "...", "https": "..."}, o construtor recusa o argumento.

A documentação de proxies descreve um único URL de proxy para todo o tráfego HTTP e HTTPS. O salto até o proxy quase sempre começa em http://, mesmo quando o destino final é HTTPS. Isso não é erro de digitação: o cliente ainda fala HTTP com o proxy para pedir CONNECT, e só depois faz o handshake TLS com a origem. Um proxy que escuta em HTTPS no primeiro salto existe, mas não foi o caso testado aqui.

2. Monte a URL com usuário e senha já percent-encoded e passe-a em proxy=. Use trust_env=False quando a linha de comando do processo puder herdar HTTP_PROXY de outro job. O laboratório local devolveu HTTP 200 e o cabeçalho Via: 1.1 httpx-pt-lab só quando o pedido passou pelo encaminhador.

O protocolo do proxy não diz de onde o IP de saída veio. HTTP ou HTTPS descrevem o salto. Residencial, móvel ou datacenter descrevem a origem do endereço. Filtrar tipo de IP é trabalho do catálogo de proxies residenciais ou da página de proxies móveis, não do parâmetro proxy=. Depois de colar o endpoint no cliente, confira o IP de saída no verificador de proxy. A página meu IP mostra o que o navegador enxerga; essa rota não é a rota do processo Python.

02Como autenticar usuário, senha e caracteres especiais#

A mesma página de proxies coloca credenciais na parte userinfo da URL. Caracteres como @, : e # quebram o parse se você concatenar a senha crua. O laboratório usou a senha dummy p@ss:wrd#1 e urllib.parse.quote(..., safe=""), que virou p%40ss%3Awrd%231. Com o userinfo correto, o GET HTTP voltou 200. Sem Authorization de proxy, o mesmo GET HTTP voltou 407 e o cabeçalho Proxy-Authenticate: Basic realm="httpx-pt-lab". Com senha errada, o HTTP também voltou 407. O HTTPX não transforma esse 407 de encaminhamento em exceção: você lê response.status_code.

A autenticação da origem é outro cabeçalho. auth=(usuario, senha) no Client vira Authorization para o site de destino. No laboratório, origem com Basic correto respondeu 200 em /origin-auth; sem esse par respondeu 401. O mesmo par da origem, com senha de proxy errada, continuou 407. Credencial de origem não paga o proxy, e userinfo do proxy não autentica a API de destino.

O HTTPX 0.28.1 exporta NetRCAuth como classe opt-in. Ele não lê ~/.netrc sozinho. O processo filho do laboratório gravou um netrc com as credenciais dummy do proxy, apontou HOME só para aquele diretório e mesmo assim recebeu 407 ao usar proxy= sem userinfo.

03O que muda entre HTTP encaminhado e HTTPS com CONNECT#

No HTTP claro o proxy interpreta o pedido em forma absoluta, abre TCP até a origem e devolve o status que a origem enviou. No HTTPS o cliente manda CONNECT host:porta. Se a autenticação do proxy passa, o proxy responde 200 Connection Established e a partir daí só copia bytes. O TLS acontece entre o HTTPX e a origem. Por isso o laboratório gerou uma CA curta com SAN DNS:localhost e IP:127.0.0.1, assinou o certificado da origem e passou ssl.create_default_context(cafile=...) em verify. verify=False não entra neste guia.

3. Faça o GET HTTPS com o mesmo proxy= e o contexto da CA. O laboratório observou status 200 e "scheme": "https" no JSON da origem. O corpo não carrega Via, porque o CONNECT não reescreve o HTTP da origem.

A falha de autenticação muda de forma. Sem userinfo, ou com senha dummy errada, o CONNECT não devolve um Response 407 para o seu client.get. O httpcore levanta ProxyError: 407 Proxy Authentication Required e o HTTPX mapeia para httpx.ProxyError. É o mesmo código 407 do mundo HTTP, mas a API que você captura no except é outra. Trate os dois caminhos: status_code == 407 no HTTP encaminhado e httpx.ProxyError no túnel.

mounts permite um HTTPTransport(proxy=...) por esquema. A documentação mostra http:// e https:// em dicionários separados e lembra que a URL do proxy, no salto inicial, continua http:// na maior parte dos produtos. O laboratório executou um GET HTTP com mounts e obteve 200. Não medimos ordem de precedência entre proxy= e mounts; não afirme uma ordem que este arquivo não rodou.

SOCKS5 existe no HTTPX só com extra httpx[socks]. Este laboratório não levantou fixture SOCKS e não descreve comportamento SOCKS.

04Como usar HTTPPROXY, NOPROXY e trustenv#

Por omissão o HTTPX lê o ambiente, como descreve a página de variáveis de ambiente. HTTP_PROXY, HTTPS_PROXY e ALL_PROXY aceitam um URL de proxy. NO_PROXY é lista separada por vírgula. trust_env=False no Client ou no httpx.get desliga essa leitura.

4. Rode o pedido em um processo filho com ambiente sanitizado. Não exporte proxy na shell de login. No laboratório o filho recebeu só PATH, HOME temporário, TMPDIR e as variáveis do caso. Com HTTP_PROXY apontando para o encaminhador autenticado, o GET HTTP sem argumento proxy= voltou 200 e Via: 1.1 httpx-pt-lab. Com o mesmo HTTP_PROXY e NO_PROXY=127.0.0.1, o GET voltou 200 com Via vazio: o host local saiu da rota. Com trust_env=False e HTTP_PROXY ainda definido no filho, o GET também foi direto. Isso é o interruptor, não um segundo dicionário de proxy.

Se a sua CI injeta HTTPS_PROXY para o apt ou para o Git, o worker Python herda o mesmo valor. Um Client de teste que deveria falar com 127.0.0.1 acaba tentando autenticar num proxy corporativo. trust_env=False no cliente de teste isola o caso. NO_PROXY isola um host quando o restante do processo ainda precisa do proxy.

Nada disso escolhe geografia nem garante que um site aceite o IP. Ambiente só decide se o socket do HTTPX sai direto ou passa pelo encaminhador.

05Como ajustar timeouts sem confundir connect e read#

O padrão do HTTPX é cinco segundos de inatividade. A página de timeouts quebra isso em connect, read, write e pool. Connect não é só o TCP até o proxy: é o orçamento de estabelecimento e TLS. No httpcore 1.0.9 esse mesmo timeout cobre start_tls (trechos 100 e 138–145). Depois de um HTTP CONNECT, o http_proxy.py ainda usa o connect timeout no TLS da origem (trechos 250–251 e 290–296). Não medimos um caso extra de TLS; o laboratório só observou recusa Darwin (ConnectError Errno 61) versus bind sem listen (ConnectTimeout). Read cobre cada pedaço do corpo. Um endpoint lento que já enviou status 200 e o primeiro byte ainda pode estourar read depois.

5. Abra client.stream, leia o primeiro chunk e continue iterando. O laboratório fez a origem HTTPS escrever X, dar flush, esperar 1,6 s e só então mandar o resto. O cliente usou read=0,35 através do CONNECT. O primeiro byte observado foi X com a resposta ainda em 200; a continuação levantou ReadTimeout 0,35 s depois. O timeout de leitura não apaga o fato de o CONNECT e o TLS terem funcionado.

HTTPTransport(retries=2) não é retry de aplicação. A documentação de transports limita o contador a ConnectError e ConnectTimeout. No laboratório, um 503 da origem com retries=2 gerou um único hit. Não configure retries do transporte esperando que um 503 “esquente” sozinho. Para 503, 429 ou corpo truncado, use a sua política (tenacity ou equivalente), não o transporte.

Porta fechada e porta ocupada sem listen não são o mesmo erro neste Darwin. Contra um porto sem socket, o HTTPX observou ConnectError: [Errno 61] Connection refused em cerca de 2 ms. Contra um socket só com bind e sem listen, observou ConnectTimeout: timed out em 0,80 s com timeout=0,8. Se o seu runbook trata os dois como “proxy recusou”, você mistura cabo arrancado com backlog que nunca aceita.

06Como fazer uma chamada async com AsyncClient#

O HTTPX síncrono e o AsyncClient compartilham proxy=, verify, timeout e trust_env. A diferença está no await. O guia async mostra async with httpx.AsyncClient() as client e r = await client.get(...). Instâncias de transporte explícitas no mundo async usam AsyncHTTPTransport, não HTTPTransport.

6. Execute uma única chamada. O laboratório fez asyncio.run em volta de um async with e um await client.get HTTPS com proxy e CA local. O status foi 200. Isso não mede mil conexões, não compara throughput e não afirma que async é obrigatório para proxy.

Não abra async with dentro do loop quente: a documentação avisa que isso descarta o pool. Um cliente por tarefa de longa vida, fechado com o context manager ou com await client.aclose(), é o padrão descrito.

07Como ler 407, ProxyError, conexão recusada e ConnectTimeout#

7. Separe falha de credencial, falha de socket e falha de leitura. A tabela abaixo resume só o que este laboratório local observou no HTTPX 0.28.1 / httpcore 1.0.9 / Python 3.9.6 em Darwin, com LibreSSL 3.3.6 no openssl e LibreSSL 2.8.3 no módulo ssl. Não é ranking de provedor.

8. Reproduza recusa e timeout de connect em portas locais antes de culpar o fornecedor. O laboratório não enviou tráfego a serviço pago.

Um certificado de origem que não sobe não deve deixar thread servindo. O caso de dependência única do laboratório tentou load_cert_chain em arquivos inexistentes depois de um bind; observou FileNotFoundError, não iniciou thread e fechou o listener. openssl no PATH é pré-requisito explícito para emitir a CA de um dia com SAN de localhost.

Quando o pedido Python já passa, ainda resta conferir o IP de saída com o verificador. Não use a aba do navegador como prova do processo. Protocolo HTTP não implica IP residencial.

08Quando um proxy HTTP residencial documentado encaixa#

O HTTPX aceita qualquer encaminhador HTTP com userinfo. Um produto residencial entra quando você precisa de um IP de origem de ISP de acesso, não porque o Client “prefira” residencial. A ficha pública Oxylabs abre em português neste endereço. O README de Residential Proxies documenta protocolos HTTP, HTTPS, HTTP/3 (MASQUE) e SOCKS5, entrada pr.oxylabs.io:7777 e usuário no formato customer-USERNAME. O guia de integração HTTPX usa exatamente httpx.Client(proxy=proxy_url) com http://<username>:<password>@pr.oxylabs.io:7777. Isso é encaixe de autenticação e protocolo documentado. Não medimos latência, taxa de sucesso nem geografia. Não há segundo fornecedor neste texto para ranking.

Se o seu destino pede IP móvel, a comparação fica na página de proxies móveis. Se pede pool residencial, na página residencial. Em ambos os casos o Client continua o mesmo: proxy= com userinfo citado, verify com CA real, timeout explícito.

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

Crie a sua pasta isolada httpx-lab/tests, grave o Python completo abaixo como httpx-lab/tests/lab.py, use um virtualenv, instale httpx==0.28.1 e httpcore==1.0.9, exija openssl no PATH e rode o script salvo. Resultados, evidências e fontes deste arquivo caem no pai isolado httpx-lab, não no diretório de trabalho arbitrário. Não há pacote privado para baixar.

mkdir -p httpx-lab/tests
cd httpx-lab
python3 -m venv .venv
. .venv/bin/activate
python -m pip install httpx==0.28.1 httpcore==1.0.9
openssl version
# Grave o Python completo abaixo como tests/lab.py.
python tests/lab.py

O laboratório completo abaixo é o arquivo executável. Ele sobe origens HTTP e HTTPS próprias, um proxy com Basic, CONNECT com bytes TLS reais, casos de ambiente em filho sanitizado, um sucesso async, recusa versus ConnectTimeout, corpo lento e a falha de certificado. Não importa o lab de outro artigo. Pré-requisito: openssl no PATH. Credenciais são dummy. Não há tráfego de provedor.

#!/usr/bin/env python3
"""Bounded local HTTPX proxy laboratory (httpx-pt).

Prerequisite: openssl on PATH.
Creates a short-lived localhost SAN CA, HTTP and HTTPS origins, and an
authenticated forwarding HTTP proxy that tunnels CONNECT with real TLS bytes.
Never sets verify=False. Never talks to paid providers. Dummy credentials only.
Child processes receive a sanitized environment (no system/global mutation).
"""
from __future__ import annotations

import argparse
import asyncio
import base64
import json
import os
import select
import shutil
import socket
import ssl
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, Callable, Dict, List, Optional, Tuple
from urllib.parse import quote, urlparse

import httpx
import httpcore

ROOT = Path(__file__).resolve().parent
ARTICLE_ROOT = ROOT.parent
EVIDENCE = ARTICLE_ROOT / "evidence"
RAW_DIR = EVIDENCE / "raw"
SOURCES = ARTICLE_ROOT / "sources"

PROXY_USER = "lab-user"
PROXY_PASS = "p@ss:wrd#1"
ORIGIN_USER = "origem"
ORIGIN_PASS = "segredo-origem"
WRONG_PASS = "senha-errada"

CONNECT_TIMEOUT_S = 0.8
SLOW_DELAY_S = 1.6
SLOW_READ_S = 0.35


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


def redact(text: str) -> str:
    out = text
    for secret in (PROXY_PASS, ORIGIN_PASS, WRONG_PASS, PROXY_USER, ORIGIN_USER):
        out = out.replace(secret, "***")
        out = out.replace(quote(secret, safe=""), "***")
    out = out.replace(quote(PROXY_PASS, safe=""), "***")
    return out


def proxy_url(host: str, port: int, user: str = PROXY_USER, password: str = PROXY_PASS) -> str:
    return "http://{u}:{p}@{h}:{n}".format(
        u=quote(user, safe=""),
        p=quote(password, safe=""),
        h=host,
        n=port,
    )


def require_openssl() -> str:
    path = shutil.which("openssl")
    if not path:
        raise SystemExit("openssl is a prerequisite and was not found on PATH")
    proc = subprocess.run([path, "version"], capture_output=True, text=True, check=True)
    return proc.stdout.strip() or proc.stderr.strip()


def run_openssl(args: List[str], cwd: Path) -> None:
    proc = subprocess.run(["openssl"] + args, cwd=str(cwd), capture_output=True, text=True)
    if proc.returncode != 0:
        raise RuntimeError(redact(proc.stderr or proc.stdout or "openssl failed"))


def make_ca(tmpdir: Path) -> Dict[str, Path]:
    ca_key = tmpdir / "ca.key"
    ca_pem = tmpdir / "ca.pem"
    origin_key = tmpdir / "origin.key"
    origin_csr = tmpdir / "origin.csr"
    origin_crt = tmpdir / "origin.crt"
    ext = tmpdir / "san.ext"
    run_openssl(
        [
            "req",
            "-new",
            "-newkey",
            "rsa:2048",
            "-days",
            "1",
            "-nodes",
            "-x509",
            "-subj",
            "/CN=httpx-pt-lab-ca",
            "-keyout",
            str(ca_key),
            "-out",
            str(ca_pem),
        ],
        tmpdir,
    )
    run_openssl(
        [
            "req",
            "-new",
            "-newkey",
            "rsa:2048",
            "-nodes",
            "-subj",
            "/CN=localhost",
            "-keyout",
            str(origin_key),
            "-out",
            str(origin_csr),
        ],
        tmpdir,
    )
    ext.write_text("subjectAltName=DNS:localhost,IP:127.0.0.1\n", encoding="utf-8")
    run_openssl(
        [
            "x509",
            "-req",
            "-in",
            str(origin_csr),
            "-CA",
            str(ca_pem),
            "-CAkey",
            str(ca_key),
            "-CAcreateserial",
            "-days",
            "1",
            "-extfile",
            str(ext),
            "-out",
            str(origin_crt),
        ],
        tmpdir,
    )
    return {"ca_pem": ca_pem, "origin_key": origin_key, "origin_crt": origin_crt}


def client_ssl_context(ca_pem: Path) -> ssl.SSLContext:
    ctx = ssl.create_default_context(cafile=str(ca_pem))
    ctx.check_hostname = True
    ctx.verify_mode = ssl.CERT_REQUIRED
    return ctx


def basic_ok(header: Optional[str], user: str, password: str) -> bool:
    if not header or not header.lower().startswith("basic "):
        return False
    try:
        raw = base64.b64decode(header.split(" ", 1)[1].strip())
        got_u, got_p = raw.decode("utf-8").split(":", 1)
    except Exception:
        return False
    return got_u == user and got_p == password


class OriginHandler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"
    token = "unset"
    scheme_label = "http"

    def log_message(self, fmt: str, *args: Any) -> None:
        return

    def _json(self, code: int, payload: Dict[str, Any]) -> None:
        body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
        self.send_response(code)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Connection", "close")
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self) -> None:  # noqa: N802
        parsed = urlparse(self.path)
        path = parsed.path
        peer = self.client_address[0]
        if path == "/whoami":
            self._json(
                200,
                {
                    "lab": self.token,
                    "scheme": self.scheme_label,
                    "peer": peer,
                    "path": path,
                    "authorization": "present" if self.headers.get("Authorization") else "absent",
                    "via_proxy_header": self.headers.get("Via", ""),
                },
            )
            return
        if path == "/origin-auth":
            if not basic_ok(self.headers.get("Authorization"), ORIGIN_USER, ORIGIN_PASS):
                self.send_response(401)
                self.send_header("WWW-Authenticate", 'Basic realm="origin"')
                self.send_header("Content-Length", "0")
                self.send_header("Connection", "close")
                self.end_headers()
                return
            self._json(200, {"lab": self.token, "origin_auth": "ok", "scheme": self.scheme_label})
            return
        if path == "/status-503":
            body = b"unavailable"
            self.send_response(503)
            self.send_header("Content-Type", "text/plain")
            self.send_header("Content-Length", str(len(body)))
            self.send_header("Connection", "close")
            self.end_headers()
            self.wfile.write(body)
            return
        if path == "/slow":
            rest = b"SLOW-BODY-REST"
            total = b"X" + rest
            self.send_response(200)
            self.send_header("Content-Type", "text/plain")
            self.send_header("Content-Length", str(len(total)))
            self.send_header("Connection", "close")
            self.end_headers()
            self.wfile.write(b"X")
            self.wfile.flush()
            time.sleep(SLOW_DELAY_S)
            self.wfile.write(rest)
            return
        self._json(404, {"lab": self.token, "error": "not-found"})


class ProxyHandler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"
    token = "unset"

    def log_message(self, fmt: str, *args: Any) -> None:
        return

    def _unauthorized(self) -> None:
        body = b"proxy authentication required"
        self.send_response(407)
        self.send_header("Proxy-Authenticate", 'Basic realm="httpx-pt-lab"')
        self.send_header("Content-Type", "text/plain")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Connection", "close")
        self.end_headers()
        self.wfile.write(body)

    def _authed(self) -> bool:
        return basic_ok(self.headers.get("Proxy-Authorization"), PROXY_USER, PROXY_PASS)

    def do_CONNECT(self) -> None:  # noqa: N802
        if not self._authed():
            self._unauthorized()
            return
        host, _, port_s = self.path.partition(":")
        port = int(port_s or "443")
        upstream = socket.create_connection((host, port), timeout=5)
        self.send_response(200, "Connection Established")
        self.send_header("Connection", "close")
        self.end_headers()
        self._tunnel(self.connection, upstream)

    def do_GET(self) -> None:  # noqa: N802
        self._forward()

    def do_HEAD(self) -> None:  # noqa: N802
        self._forward()

    def _forward(self) -> None:
        if not self._authed():
            self._unauthorized()
            return
        parsed = urlparse(self.path)
        if parsed.scheme != "http" or not parsed.hostname:
            self.send_error(400, "absolute-form HTTP URL required")
            return
        port = parsed.port or 80
        target = "{host}:{port}".format(host=parsed.hostname, port=port)
        path = parsed.path or "/"
        if parsed.query:
            path = path + "?" + parsed.query
        lines = ["{m} {p} HTTP/1.1".format(m=self.command, p=path)]
        skip = {"proxy-authorization", "proxy-connection", "connection", "host"}
        lines.append("Host: {h}".format(h=parsed.netloc))
        lines.append("Via: 1.1 httpx-pt-lab")
        for key, value in self.headers.items():
            if key.lower() in skip:
                continue
            lines.append("{k}: {v}".format(k=key, v=value))
        lines.append("Connection: close")
        lines.append("")
        raw = ("\r\n".join(lines) + "\r\n").encode("iso-8859-1")
        with socket.create_connection((parsed.hostname, port), timeout=5) as upstream:
            upstream.sendall(raw)
            chunks: List[bytes] = []
            while True:
                buf = upstream.recv(65536)
                if not buf:
                    break
                chunks.append(buf)
        blob = b"".join(chunks)
        self.connection.sendall(blob)

    def _tunnel(self, client: socket.socket, upstream: socket.socket) -> None:
        sockets = [client, upstream]
        try:
            while True:
                readable, _, _ = select.select(sockets, [], [], 10)
                if not readable:
                    break
                for sock in readable:
                    other = upstream if sock is client else client
                    data = sock.recv(65536)
                    if not data:
                        return
                    other.sendall(data)
        finally:
            try:
                upstream.close()
            except OSError:
                pass


class LabServer:
    def __init__(self, handler: Callable[..., BaseHTTPRequestHandler], tls: Optional[ssl.SSLContext] = None) -> None:
        self._httpd = ThreadingHTTPServer(("127.0.0.1", 0), handler)
        self._httpd.allow_reuse_address = True
        self._tls = tls
        self._started = False
        if tls is not None:
            self._httpd.socket = tls.wrap_socket(self._httpd.socket, server_side=True)
        self.thread = threading.Thread(target=self._httpd.serve_forever, daemon=True)

    @property
    def port(self) -> int:
        return int(self._httpd.server_address[1])

    def start(self) -> None:
        self.thread.start()
        self._started = True

    def stop(self) -> None:
        if self._started:
            self._httpd.shutdown()
            self.thread.join(timeout=5)
        try:
            self._httpd.server_close()
        except OSError:
            pass
        self._started = False


def capture_call(fn: Callable[[], Any]) -> Dict[str, Any]:
    started = time.time()
    try:
        value = fn()
        return {
            "ok": True,
            "value": value,
            "error_type": None,
            "error": None,
            "elapsed_s": round(time.time() - started, 4),
        }
    except Exception as exc:
        return {
            "ok": False,
            "value": None,
            "error_type": type(exc).__name__,
            "error_module": type(exc).__module__,
            "error": redact(str(exc)),
            "elapsed_s": round(time.time() - started, 4),
            "traceback": redact(traceback.format_exc()),
        }


def json_body(response: httpx.Response) -> Any:
    try:
        return response.json()
    except Exception:
        return {"text": redact(response.text[:300])}


def summarize_response(response: httpx.Response) -> Dict[str, Any]:
    return {
        "status": response.status_code,
        "http_version": response.http_version,
        "body": json_body(response),
        "headers": {k: redact(v) for k, v in list(response.headers.items())[:12]},
    }


def unused_port() -> int:
    probe = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    probe.bind(("127.0.0.1", 0))
    port = int(probe.getsockname()[1])
    probe.close()
    return port


def child_main(case: str) -> int:
    http_origin = os.environ["LAB_HTTP_ORIGIN"]
    https_origin = os.environ["LAB_HTTPS_ORIGIN"]
    proxy_port = int(os.environ["LAB_PROXY_PORT"])
    token = os.environ["LAB_TOKEN"]
    ca_pem = Path(os.environ["LAB_CA_PEM"])
    ctx = client_ssl_context(ca_pem)
    timeout = httpx.Timeout(5.0)
    result: Dict[str, Any]
    if case == "env_route":
        with httpx.Client(timeout=timeout, trust_env=True) as client:
            r = client.get(http_origin + "/whoami")
        result = {"status": r.status_code, "body": json_body(r)}
    elif case == "no_proxy":
        with httpx.Client(timeout=timeout, trust_env=True) as client:
            r = client.get(http_origin + "/whoami")
        result = {"status": r.status_code, "body": json_body(r)}
    elif case == "trust_env_false":
        with httpx.Client(timeout=timeout, trust_env=False) as client:
            r = client.get(http_origin + "/whoami")
        result = {"status": r.status_code, "body": json_body(r)}
    elif case == "netrc":
        with httpx.Client(timeout=timeout, trust_env=True, proxy="http://127.0.0.1:{0}".format(proxy_port)) as client:
            r = client.get(http_origin + "/whoami")
        result = {"status": r.status_code, "text": redact(r.text[:200])}
    elif case == "https_env":
        with httpx.Client(timeout=timeout, trust_env=True, verify=ctx) as client:
            r = client.get(https_origin + "/whoami")
        result = {"status": r.status_code, "body": json_body(r)}
    else:
        result = {"error": "unknown-case", "case": case}
    sys.stdout.write(json.dumps({"token": token, "case": case, "result": result}, ensure_ascii=False))
    sys.stdout.write("\n")
    return 0


def run_child(case: str, env: Dict[str, str]) -> Dict[str, Any]:
    clean = {
        "PATH": os.environ.get("PATH", ""),
        "LANG": "C",
        "LC_ALL": "C",
        "TERM": "dumb",
        "HOME": env.get("HOME", ""),
        "TMPDIR": env.get("TMPDIR", ""),
        "PYTHONUNBUFFERED": "1",
    }
    clean.update(env)
    proc = subprocess.run(
        [sys.executable, str(Path(__file__).resolve()), "--child", case],
        env=clean,
        capture_output=True,
        text=True,
        timeout=30,
    )
    payload: Dict[str, Any] = {
        "returncode": proc.returncode,
        "stdout": redact(proc.stdout),
        "stderr": redact(proc.stderr),
    }
    try:
        payload["parsed"] = json.loads(proc.stdout.strip().splitlines()[-1])
    except Exception:
        payload["parsed"] = None
    return payload


def missing_tls_startup(tmpdir: Path) -> Dict[str, Any]:
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.bind(("127.0.0.1", 0))
    port = int(sock.getsockname()[1])
    started_thread = False
    error_type = None
    error = None
    try:
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        ctx.load_cert_chain(str(tmpdir / "missing-cert.pem"), str(tmpdir / "missing-key.pem"))
        wrapped = ctx.wrap_socket(sock, server_side=True)
        wrapped.close()
    except Exception as exc:
        error_type = type(exc).__name__
        error = redact(str(exc))
    finally:
        try:
            sock.close()
        except OSError:
            pass
    return {
        "port_bound": port,
        "started_serving_thread": started_thread,
        "error_type": error_type,
        "error": error,
        "listener_closed": True,
    }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--child", default="")
    args = parser.parse_args()
    if args.child:
        return child_main(args.child)

    RAW_DIR.mkdir(parents=True, exist_ok=True)
    SOURCES.mkdir(parents=True, exist_ok=True)
    EVIDENCE.mkdir(parents=True, exist_ok=True)

    openssl_version = require_openssl()
    token = "httpx-pt-{0}".format(os.urandom(4).hex())
    cases: List[Dict[str, Any]] = []
    observations: Dict[str, Any] = {}
    tmpdir = Path(tempfile.mkdtemp(prefix="httpx-pt-lab-"))
    http_srv: Optional[LabServer] = None
    https_srv: Optional[LabServer] = None
    proxy_srv: Optional[LabServer] = None
    holder: Optional[socket.socket] = None

    try:
        certs = make_ca(tmpdir)
        shutil.copy(certs["ca_pem"], EVIDENCE / "ca.pem")

        OriginHandler.token = token
        ProxyHandler.token = token

        class HttpOrigin(OriginHandler):
            scheme_label = "http"

        class HttpsOrigin(OriginHandler):
            scheme_label = "https"

        http_srv = LabServer(HttpOrigin)
        http_srv.start()
        server_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        server_ctx.load_cert_chain(str(certs["origin_crt"]), str(certs["origin_key"]))
        https_srv = LabServer(HttpsOrigin, tls=server_ctx)
        https_srv.start()
        proxy_srv = LabServer(ProxyHandler)
        proxy_srv.start()

        http_origin = "http://127.0.0.1:{0}".format(http_srv.port)
        https_origin = "https://127.0.0.1:{0}".format(https_srv.port)
        px = proxy_url("127.0.0.1", proxy_srv.port)
        px_wrong = proxy_url("127.0.0.1", proxy_srv.port, password=WRONG_PASS)
        px_plain = "http://127.0.0.1:{0}".format(proxy_srv.port)
        verify = client_ssl_context(certs["ca_pem"])
        timeout = httpx.Timeout(5.0)

        def add(case_id: str, predicate: str, passed: bool, observation: Any) -> None:
            cases.append(
                {
                    "id": case_id,
                    "predicate": predicate,
                    "passed": bool(passed),
                    "observation": observation,
                }
            )

        # Direct baseline
        with httpx.Client(timeout=timeout, trust_env=False) as client:
            r_http = client.get(http_origin + "/whoami")
        add(
            "direct_http",
            "direct HTTP 200 and lab token match",
            r_http.status_code == 200 and json_body(r_http).get("lab") == token,
            summarize_response(r_http),
        )
        with httpx.Client(timeout=timeout, trust_env=False, verify=verify) as client:
            r_https = client.get(https_origin + "/whoami")
        add(
            "direct_https",
            "direct HTTPS 200 with custom verify context (not verify=False)",
            r_https.status_code == 200 and json_body(r_https).get("scheme") == "https",
            summarize_response(r_https),
        )

        # Proxy HTTP + HTTPS with correct auth
        with httpx.Client(timeout=timeout, trust_env=False, proxy=px) as client:
            r_phttp = client.get(http_origin + "/whoami")
        add(
            "proxy_http_ok",
            "proxy= HTTP forward 200 with Via from lab proxy",
            r_phttp.status_code == 200 and "httpx-pt-lab" in str(json_body(r_phttp).get("via_proxy_header")),
            summarize_response(r_phttp),
        )
        with httpx.Client(timeout=timeout, trust_env=False, proxy=px, verify=verify) as client:
            r_phttps = client.get(https_origin + "/whoami")
        add(
            "proxy_https_connect_ok",
            "proxy= HTTPS CONNECT 200 with verified TLS to origin",
            r_phttps.status_code == 200 and json_body(r_phttps).get("scheme") == "https",
            summarize_response(r_phttps),
        )

        # Missing / wrong / correct proxy auth separately
        with httpx.Client(timeout=timeout, trust_env=False, proxy=px_plain) as client:
            r_miss_http = client.get(http_origin + "/whoami")
        add(
            "proxy_http_missing_auth",
            "HTTP forward without credentials returns 407",
            r_miss_http.status_code == 407,
            summarize_response(r_miss_http),
        )
        miss_https = capture_call(
            lambda: httpx.Client(timeout=timeout, trust_env=False, proxy=px_plain, verify=verify).get(
                https_origin + "/whoami"
            )
        )
        add(
            "proxy_https_missing_auth",
            "HTTPS CONNECT without credentials raises ProxyError (observed class recorded)",
            miss_https["error_type"] == "ProxyError",
            miss_https,
        )
        with httpx.Client(timeout=timeout, trust_env=False, proxy=px_wrong) as client:
            r_wrong_http = client.get(http_origin + "/whoami")
        add(
            "proxy_http_wrong_auth",
            "HTTP forward with wrong password returns 407",
            r_wrong_http.status_code == 407,
            summarize_response(r_wrong_http),
        )
        wrong_https = capture_call(
            lambda: httpx.Client(timeout=timeout, trust_env=False, proxy=px_wrong, verify=verify).get(
                https_origin + "/whoami"
            )
        )
        add(
            "proxy_https_wrong_auth",
            "HTTPS CONNECT with wrong password raises ProxyError",
            wrong_https["error_type"] == "ProxyError",
            wrong_https,
        )
        with httpx.Client(timeout=timeout, trust_env=False, proxy=px) as client:
            r_ok_again = client.get(http_origin + "/whoami")
        add(
            "proxy_http_correct_auth",
            "HTTP forward with quoted special-character dummy password returns 200",
            r_ok_again.status_code == 200,
            summarize_response(r_ok_again),
        )

        # Origin auth distinction
        with httpx.Client(timeout=timeout, trust_env=False, proxy=px, auth=(ORIGIN_USER, ORIGIN_PASS)) as client:
            r_oa = client.get(http_origin + "/origin-auth")
        add(
            "origin_auth_ok",
            "origin BasicAuth succeeds independently of proxy userinfo",
            r_oa.status_code == 200 and json_body(r_oa).get("origin_auth") == "ok",
            summarize_response(r_oa),
        )
        with httpx.Client(timeout=timeout, trust_env=False, proxy=px) as client:
            r_oa_miss = client.get(http_origin + "/origin-auth")
        add(
            "origin_auth_missing",
            "missing origin auth is 401, not 407",
            r_oa_miss.status_code == 401,
            summarize_response(r_oa_miss),
        )
        with httpx.Client(timeout=timeout, trust_env=False, proxy=px_wrong, auth=(ORIGIN_USER, ORIGIN_PASS)) as client:
            r_oa_proxy_bad = client.get(http_origin + "/origin-auth")
        add(
            "origin_auth_does_not_satisfy_proxy",
            "correct origin credentials do not replace proxy 407",
            r_oa_proxy_bad.status_code == 407,
            summarize_response(r_oa_proxy_bad),
        )

        # Mounts: one executed HTTP request, no untested precedence claims
        mounts = {
            "http://": httpx.HTTPTransport(proxy=px),
            "https://": httpx.HTTPTransport(proxy=px, verify=verify),
        }
        with httpx.Client(timeout=timeout, trust_env=False, mounts=mounts, verify=verify) as client:
            r_mount = client.get(http_origin + "/whoami")
        add(
            "mounts_http_executed",
            "Client(mounts=...) HTTP request 200 (scheme routing documented, precedence not claimed)",
            r_mount.status_code == 200,
            summarize_response(r_mount),
        )

        # Environment-only / NO_PROXY / trust_env=False
        child_home = tmpdir / "child-home"
        child_home.mkdir()
        netrc_path = child_home / ".netrc"
        netrc_path.write_text(
            "machine 127.0.0.1 login {u} password {p}\n".format(u=PROXY_USER, p=PROXY_PASS),
            encoding="utf-8",
        )
        os.chmod(netrc_path, 0o600)
        base_env = {
            "HOME": str(child_home),
            "TMPDIR": str(tmpdir),
            "LAB_HTTP_ORIGIN": http_origin,
            "LAB_HTTPS_ORIGIN": https_origin,
            "LAB_PROXY_PORT": str(proxy_srv.port),
            "LAB_TOKEN": token,
            "LAB_CA_PEM": str(certs["ca_pem"]),
        }
        env_route = dict(base_env)
        env_route["HTTP_PROXY"] = px
        env_route["http_proxy"] = px
        child_env = run_child("env_route", env_route)
        via = ((child_env.get("parsed") or {}).get("result") or {}).get("body", {}).get("via_proxy_header", "")
        add(
            "env_only_http_proxy",
            "HTTP_PROXY in sanitized child env routes without proxy= argument",
            (child_env.get("parsed") or {}).get("result", {}).get("status") == 200 and "httpx-pt-lab" in str(via),
            child_env,
        )
        no_proxy_env = dict(env_route)
        no_proxy_env["NO_PROXY"] = "127.0.0.1"
        no_proxy_env["no_proxy"] = "127.0.0.1"
        child_np = run_child("no_proxy", no_proxy_env)
        via_np = ((child_np.get("parsed") or {}).get("result") or {}).get("body", {}).get("via_proxy_header", "")
        add(
            "no_proxy_bypass",
            "NO_PROXY=127.0.0.1 bypasses HTTP_PROXY (empty Via)",
            (child_np.get("parsed") or {}).get("result", {}).get("status") == 200 and via_np == "",
            child_np,
        )
        te_env = dict(env_route)
        child_te = run_child("trust_env_false", te_env)
        via_te = ((child_te.get("parsed") or {}).get("result") or {}).get("body", {}).get("via_proxy_header", "")
        add(
            "trust_env_false",
            "trust_env=False ignores HTTP_PROXY",
            (child_te.get("parsed") or {}).get("result", {}).get("status") == 200 and via_te == "",
            child_te,
        )
        netrc_env = dict(base_env)
        child_netrc = run_child("netrc", netrc_env)
        add(
            "httpx_no_autonetric",
            "HTTPX does not auto-use ~/.netrc for proxy auth (407 without Proxy-Authorization)",
            (child_netrc.get("parsed") or {}).get("result", {}).get("status") == 407,
            child_netrc,
        )

        # AsyncClient one success
        async def async_once() -> int:
            async with httpx.AsyncClient(timeout=timeout, trust_env=False, proxy=px, verify=verify) as client:
                r = await client.get(https_origin + "/whoami")
                return r.status_code

        async_status = asyncio.run(async_once())
        add(
            "async_client_success",
            "AsyncClient async with/await one HTTPS success (no concurrency claim)",
            async_status == 200,
            {"status": async_status},
        )

        # Refused vs bound-nonlistening
        closed = unused_port()
        refused = capture_call(
            lambda: httpx.Client(timeout=httpx.Timeout(CONNECT_TIMEOUT_S), trust_env=False).get(
                "http://127.0.0.1:{0}/".format(closed)
            )
        )
        add(
            "connect_refused",
            "nothing listening yields ConnectError (observed class recorded honestly)",
            refused["error_type"] == "ConnectError",
            refused,
        )
        holder = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        holder.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        holder.bind(("127.0.0.1", 0))
        bound_port = int(holder.getsockname()[1])
        bound_obs = capture_call(
            lambda: httpx.Client(timeout=httpx.Timeout(CONNECT_TIMEOUT_S), trust_env=False).get(
                "http://127.0.0.1:{0}/".format(bound_port)
            )
        )
        add(
            "bound_nonlistening",
            "bound-nonlistening Darwin observed class recorded (distinct from refused)",
            bound_obs["error_type"] in {"ConnectTimeout", "ConnectError"}
            and bound_obs["error_type"] != refused["error_type"],
            bound_obs,
        )

        # Slow body first byte then ReadTimeout while streaming
        slow_timeout = httpx.Timeout(connect=2.0, read=SLOW_READ_S, write=2.0, pool=2.0)

        def slow_stream() -> Dict[str, Any]:
            first = None
            first_at = None
            later_error = None
            with httpx.Client(timeout=slow_timeout, trust_env=False, proxy=px, verify=verify) as client:
                with client.stream("GET", https_origin + "/slow") as response:
                    iterator = response.iter_bytes()
                    first = next(iterator)
                    first_at = time.time()
                    try:
                        leftover = b"".join(iterator)
                        return {
                            "status": response.status_code,
                            "first": first.decode("latin1"),
                            "completed": leftover.decode("latin1"),
                            "later_error": None,
                        }
                    except Exception as exc:
                        later_error = type(exc).__name__
                        return {
                            "status": response.status_code,
                            "first": first.decode("latin1") if first else None,
                            "first_before_timeout": True,
                            "later_error": later_error,
                            "elapsed_after_first_s": round(time.time() - first_at, 4) if first_at else None,
                        }

        slow_obs = slow_stream()
        add(
            "slow_body_first_byte_then_readtimeout",
            "slow HTTPS body through CONNECT: first byte X observed, then ReadTimeout while streaming",
            slow_obs.get("first") == "X" and slow_obs.get("later_error") == "ReadTimeout",
            slow_obs,
        )

        # HTTPTransport retries: ConnectError/ConnectTimeout only, not HTTP 503
        attempts = {"n": 0}

        class CountingOrigin(HttpOrigin):
            def do_GET(self) -> None:  # noqa: N802
                attempts["n"] += 1
                OriginHandler.do_GET(self)

        counting = LabServer(CountingOrigin)
        counting.start()
        try:
            transport = httpx.HTTPTransport(retries=2)
            with httpx.Client(timeout=timeout, trust_env=False, transport=transport) as client:
                r503 = client.get("http://127.0.0.1:{0}/status-503".format(counting.port))
            add(
                "retries_skip_http_503",
                "HTTPTransport(retries=2) does not retry HTTP 503 (single origin hit)",
                r503.status_code == 503 and attempts["n"] == 1,
                {"status": r503.status_code, "origin_hits": attempts["n"]},
            )
        finally:
            counting.stop()

        retry_transport = httpx.HTTPTransport(retries=1)
        retry_refused = capture_call(
            lambda: httpx.Client(
                timeout=httpx.Timeout(CONNECT_TIMEOUT_S),
                trust_env=False,
                transport=retry_transport,
            ).get("http://127.0.0.1:{0}/".format(unused_port()))
        )
        add(
            "retries_connect_error_still_connecterror",
            "retries apply to connect failures; final class remains ConnectError",
            retry_refused["error_type"] == "ConnectError",
            retry_refused,
        )

        # Negative discriminator: lab token, not a public IP service
        add(
            "negative_discriminator_lab_token",
            "origin body contains lab token and not a public my-ip JSON shape",
            json_body(r_phttp).get("lab") == token and "ip" not in json_body(r_phttp),
            {"body": json_body(r_phttp)},
        )

        tls_fail = missing_tls_startup(tmpdir)
        add(
            "missing_tls_startup_failure",
            "HTTPS origin without cert files fails before serving thread; listener closed",
            tls_fail["error_type"] is not None and tls_fail["started_serving_thread"] is False and tls_fail["listener_closed"],
            tls_fail,
        )

        observations = {
            "http_origin": http_origin,
            "https_origin": https_origin,
            "proxy_port": proxy_srv.port,
            "refused_class": refused.get("error_type"),
            "bound_nonlistening_class": bound_obs.get("error_type"),
            "missing_https_auth_class": miss_https.get("error_type"),
            "wrong_https_auth_class": wrong_https.get("error_type"),
        }

        runtime = {
            "python": sys.version,
            "httpx": httpx.__version__,
            "httpcore": httpcore.__version__,
            "openssl_cli": openssl_version,
            "openssl_ssl": ssl.OPENSSL_VERSION,
            "platform": sys.platform,
            "executable": sys.executable,
            "started_at": utcnow(),
            "token_redacted": True,
        }
        (SOURCES / "installed-runtime.json").write_text(
            json.dumps(runtime, indent=2, ensure_ascii=False) + "\n", encoding="utf-8"
        )

        report = {
            "updated_at": utcnow(),
            "runtime": runtime,
            "cases": cases,
            "observations": observations,
            "passed": sum(1 for c in cases if c["passed"]),
            "failed": sum(1 for c in cases if not c["passed"]),
            "total": len(cases),
        }
        (ROOT / "results.json").write_text(json.dumps(report, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
        lines = [
            "httpx-pt lab {0}".format(report["updated_at"]),
            "python {0}".format(sys.version.split()[0]),
            "httpx {0} httpcore {1}".format(httpx.__version__, httpcore.__version__),
            "openssl {0}".format(openssl_version),
            "passed {0}/{1}".format(report["passed"], report["total"]),
            "",
        ]
        for c in cases:
            flag = "PASS" if c["passed"] else "FAIL"
            lines.append("{0} {1}: {2}".format(flag, c["id"], c["predicate"]))
        text = "\n".join(lines) + "\n"
        (ROOT / "results.txt").write_text(text, encoding="utf-8")
        (RAW_DIR / "results-redacted.txt").write_text(redact(text), encoding="utf-8")
        (RAW_DIR / "observations.json").write_text(
            json.dumps(observations, indent=2, ensure_ascii=False) + "\n", encoding="utf-8"
        )
        return 0 if report["failed"] == 0 else 1
    finally:
        if http_srv is not None:
            http_srv.stop()
        if https_srv is not None:
            https_srv.stop()
        if proxy_srv is not None:
            proxy_srv.stop()
        if holder is not None:
            try:
                holder.close()
            except OSError:
                pass
        shutil.rmtree(tmpdir, ignore_errors=True)


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