Proxy con curl: configuración, autenticación y errores
Configura curl con un proxy HTTP o SOCKS5, separa -U de -u, lee CONNECT, http_proxy y los códigos 407, 7, 28 y 60.
Esta guía es para quien lanza curl por un proxy HTTP o SOCKS5 y necesita separar autenticación, CONNECT y variables de entorno. Las comprobaciones de este texto usaron orígenes en loopback, no una cuenta de proveedor. No hay claves gratis y Top Proxy no vende IP. Un navegador en Mi IP muestra otra ruta.
01Cómo indicar un proxy HTTP en curl
-x / --proxy apunta al proxy. La URL final es el destino. Son esquemas distintos. Un proxy HTTP delante de un sitio HTTPS suele verse así, y no convierte el proxy en un "proxy HTTPS" solo porque el destino lleva https://.
El binario comprobado fue curl 8.7.1 (27 de marzo de 2024), libcurl/8.7.1 con SecureTransport, LibreSSL 3.3.6, nghttp2 y la lista de features que incluye HTTPS-proxy, MultiSSL y AsynchDNS. Un manual en línea puede describir flags que este binario no tiene. Empieza cada invocación con -q para que un curlrc heredado no meta proxy, CA ni usuario. Pasa argumentos en un array, no en un string que el shell expanda.
everything curl, HTTP proxy (consultado el 14 de octubre de 2026) dice que -x / --proxy asume un proxy HTTP y que, sin puerto, el valor histórico es 1080. En verbose curl se conecta al proxy, no al origen, y la línea de petición cambia.
- Fija el binario con
curl -q --version. Esta corrida usó curl 8.7.1. - Arranca con
curl -q. El resto de flags va detrás. - Pon
-x http://127.0.0.1:PUERTOy un destino HTTP que controles. - Lee el verbose y, si puedes, el log del proxy. El estado 200 solo no dice por qué puerto salió el GET.
En el laboratorio del 14 de octubre de 2026, con entorno hijo limpio, la línea
# PLACEHOLDERS. No es el laboratorio ejecutado.
curl -q -sS -v --max-time 8 --noproxy '' \
-x http://127.0.0.1:PUERTO_PROXY \
-U 'USUARIO:CONTRASEÑA' \
http://destino.ejemplo/
conectó al puerto del proxy, no al origen, y envió forma absoluta:
* Connected to 127.0.0.1 (127.0.0.1) port PUERTO_PROXY
> GET http://curl-es-lab.origin:PUERTO_ORIGEN/ HTTP/1.1
> Host: curl-es-lab.origin:PUERTO_ORIGEN
El cuerpo del origen devolvió origin-http|/. Eso prueba esa ruta de laboratorio. No es un exit residencial ni móvil. El GET directo, sin -x, fue a 127.0.0.1 del origen con GET / relativo y el proxy se quedó quieto. Ese control importa. Sin él, un 200 "parece que funciona" y no sabes si el proxy existió.
--noproxy '' en los casos que deben usar proxy evita que un bypass de loopback esconda un aserto fallido. Si el destino es 127.0.0.1 y tu entorno trae NO_PROXY amplio, curl puede ir directo y tú celebras un falso positivo.
El puerto HTTP y el puerto SOCKS de un proveedor casi nunca coinciden. Mezclar -x http:// contra un puerto SOCKS es un error de protocolo. Esta corrida no capturó ese cruce: no hay un caso de reset ni de basura en los 23 resultados. Mira en el panel qué puerto está etiquetado como HTTP CONNECT y cuál como SOCKS5. El checker de proxy sirve para ver si un host responde, no para certificar que tu proceso curl usa esa misma salida.
Protocolo e IP son ejes distintos. HTTP o SOCKS5 no convierten un datacenter en residencial.
02Cómo tunelizar HTTPS con CONNECT
La gente llama "proxy HTTPS" al hecho de abrir https://. El salto del medio sigue siendo, casi siempre, un proxy HTTP que habla CONNECT. everything curl lo describe así: HTTPS pide un túnel, curl manda CONNECT, el proxy abre TCP y el cifrado corre entre cliente y origen. El proxy no debería leer ese TLS si el túnel es real.
No hace falta https:// en la URL del proxy para un destino HTTPS. Distingue TLS hacia el origen de TLS hacia un proxy HTTPS. --cacert en esta prueba controló el origen de laboratorio. Un proxy con TLS propio es otra capa. Este laboratorio no la midió. Trátala como hueco, no como éxito.
- Deja el proxy en
http://127.0.0.1:PUERTOaunque el destino seahttps://. - Confía el origen con
--cacertapuntando a tu CA. No uses-k. - Confirma CONNECT en verbose o en el log del proxy antes de cantar victoria por un 200.
- Repite sin CA. Quieres un fallo TLS que aun así haya hecho CONNECT.
Sin CA, CONNECT salió autorizado y curl cerró con código 60. El túnel existió. El certificado no. Eso no es "el proxy caído". Un 200 en el navegador tampoco sustituye esta ruta: Mi IP habla del navegador.
El verbose del caso bueno, ya redactado, mostró:
* Establish HTTP proxy tunnel to curl-es-lab.origin:PUERTO_TLS
> CONNECT curl-es-lab.origin:PUERTO_TLS HTTP/1.1
y luego el cuerpo origin-https|/. El fixture reenvió bytes TLS de verdad. Un 200 falso de CONNECT sin túnel no habría entregado ese origen.
Si el proxy HTTP exige usuario y CONNECT falla, este curl 8.7.1 no devolvió un HTML 407 con exit 0. Devolvió 56 (failure in receiving network data) con HTTP/1.1 407 en el túnel. HTTP forwarding y CONNECT no son el mismo tipo de error. Anota el código de salida, el estado del proxy y la excepción de transporte por separado. No copies un recetario de 407 de otro cliente.
03Cómo separar -U y -u
-U / --proxy-user es el proxy. -u / --user es el origen. everything curl, autenticación de proxy (14 de octubre de 2026) lo dice sin rodeos: fallar o equivocar las credenciales del proxy produce 407, y eso es independiente de la autenticación del servidor.
La contraseña dummy del laboratorio incluía @, :, /, # y %. Esos caracteres van en -U como argumento aparte, o en userinfo con codificación por componente. No las pongas crudas en una URL que luego pegas en un ticket. Los volcados de esta guía ocultan Basic. No imprimas Proxy-Authorization cuando depuras.
Faltan credenciales, están mal, o están bien. Son tres casos.
En HTTP forwarding, sin -U, el fixture contestó 407 y curl salió 0. El cuerpo fue proxy-407. Verbose vio Proxy-Authenticate: Basic realm="lab". El log del proxy marcó authReason: missing. Con contraseña incorrecta, el mismo 407 y authReason: wrong. Curl no "falló" en el sentido de exit distinto de cero. Si tu script mira solo $?, se traga el 407.
Con --fail, el mismo 407 HTTP salió 22. Eso es este binario, este flag, este 407 de forwarding. No universalices "todo 407 es 22". CONNECT con 407, sin --fail, fue 56. Mezclarlos es el error que más tiempo pierde.
Con -U correcto, HTTP y HTTPS devolvieron origen 200. El origen no vio Proxy-Authorization. Esa cabecera se queda en el salto del proxy. Si el sitio también pide Basic, añade -u. En /origin-auth, -U bueno y -u malo dejó el proxy autorizado y el origen en 401. Al revés, el proxy ni abre el origen.
- Pon usuario y contraseña del proxy en
-U. - Reserva
-upara el sitio. - Trata 407 HTTP (exit 0 o 22 según
--fail) y CONNECT 56 como desenlaces distintos. - No asumas que "en el navegador entra" prueba el header que mandó curl.
04Cómo curl lee el entorno y NOPROXY
curl no tiene trust_env. Eso es de otros clientes Python. Aquí el aislamiento es -q (nada de curlrc), un proceso hijo sin HTTP_PROXY/HTTPS_PROXY/ALL_PROXY/NO_PROXY/FTP_PROXY en ambos casos, sin REQUEST_METHOD ni resto CGI, sin CURL_CA_BUNDLE, SSL_CERT_FILE, SSL_CERT_DIR, SSLKEYLOGFILE ni REQUESTS_CA_BUNDLE, y sin --netrc. HOME no se reescribe. Un netrc vacío privado existió en el temporal y se borró al terminar. -q como primera opción es lo que evita el curlrc del usuario, no un truco de directorio.
everything curl, variables de entorno (14 de octubre de 2026) es claro: el nombre [scheme]_proxy sigue el esquema del destino. http_proxy se acepta solo en minúsculas. El resto puede ir en mayúsculas, HTTPS_PROXY incluido. CGI crea HTTP_* a partir de cabeceras. Un header Proxy: se volvería HTTP_PROXY y un script que llama curl saldría hacia un proxy ajeno. Por eso curl rechaza HTTP_PROXY.
En macOS las variables distinguen mayúsculas. Con HTTP_PROXY apuntando al proxy de laboratorio y un GET HTTP, verbose no escribió Uses proxy env variable HTTP_PROXY. curl se conectó al origen, GET /, exit 0, proxy inactivo. Con http_proxy en minúsculas, verbose dijo Uses proxy env variable http_proxy == 'http://127.0.0.1:…' y el GET absoluto pasó por alpha.
Para HTTPS, https_proxy y HTTPS_PROXY sí abrieron CONNECT. ALL_PROXY queda por debajo de la variable específica del esquema. No infieras precedencia por un 200. Este laboratorio usó dos proxies instrumentados, alpha y bravo. Con http_proxy en alpha y -x en bravo, bravo recibió el GET y alpha no vio nada. -x ganó al entorno. El status 200 no lo habría dicho.
--noproxy / NO_PROXY es el bypass deliberado. Con http_proxy puesto y --noproxy 127.0.0.1 hacia http://127.0.0.1:origen/, el origen contestó 200 y alpha no vio el GET. Ese es un caso propio, no un efecto secundario que asumas. Para forzar proxy en loopback, --noproxy ''.
CURL_CA_BUNDLE no es "trust_env". En esta corrida, con SecureTransport, CURL_CA_BUNDLE apuntando a la CA de laboratorio y un HTTPS directo a 127.0.0.1 salió 0 y verbose mostró CAfile: de esa CA. --cacert hizo lo mismo. Eso es lo medido en curl 8.7.1 de este equipo. No lo conviertas en una ley para LibreSSL puro u OpenSSL en Linux. Un trust_env=False de Requests también deja de leer CA y netrc. curl reparte esas palancas: -q, no --netrc, --cacert o CURL_CA_BUNDLE, y el entorno de proxy por otro lado.
05Dónde resuelve el nombre SOCKS5 y socks5h
SOCKS5 no es "el proxy más anónimo". Es otro protocolo. everything curl, SOCKS (14 de octubre de 2026) parte las versiones: socks5 resuelve curl, socks5h / --socks5-hostname manda el hostname al proxy.
Un GET SOCKS exitoso no localiza el DNS. Hace falta el ATYP del handshake.
-x socks5://127.0.0.1:PUERTOhaciahttp://127.0.0.1:origen/.- Lee el verbose: este curl escribió
SOCKS5 connect to 127.0.0.1:origen (locally resolved). - El fixture SOCKS registró ATYP ipv4 y host
127.0.0.1. Eso es un destino con IP literal. No ejercita un resolver local de hostname. - Repite con
socks5h://y un hostname de laboratorio. Verbose:SOCKS5 connect to curl-es-lab.origin:origen (remotely resolved). El fixture vio ATYP domain y ese hostname.
El mapeo del hostname a loopback lo hizo el proxy de laboratorio, no un DNS público. El proceso Python instala un getaddrinfo acotado para su propio código. curl 8.7.1 declara AsynchDNS y el hijo de curl no hereda ese parche. No toques /etc/hosts. Si en tu máquina socks5h falla porque el proxy no sabe resolver, eso es configuración del proxy, no "curl no soporta SOCKS".
La feature HTTPS-proxy del binario no se usó como salto TLS hacia el proxy. SOCKS y HTTP CONNECT se midieron en loopback. Un producto residencial que anuncia SOCKS5 sigue siendo otro origen de IP. Mídelo aparte.
06Cómo leer timeouts, 407 y la denegación de conexión
--connect-timeout limita el arranque: DNS, TCP, TLS. --max-time / -m limita toda la operación, incluso a mitad de cuerpo. everything curl, timeouts (14 de octubre de 2026) asigna el código 28 a ambos si se agota el reloj, y avisa de que por defecto curl no pone tope.
Una denegación de conexión no es un timeout. El exit 7 de esta corrida vino de un listener que se cerró en un puerto efímero. curl escribió Connection refused en 0 ms. Un segundo socket se dejó en bind sin listen hasta el finally; es otro puerto, no retiene el puerto rehusado. Un bind sin listen en Darwin a veces no manda RST; esa rama no está entre los 23 casos capturados.
El cuerpo lento es otra historia. /slow-body mandó cabeceras 200, escribió y flusheó el primer byte H, señaló un evento, y se quedó más tiempo que --max-time 0.8 antes del resto. curl había visto headers y ese byte. Salió 28. Eso es un timeout de transferencia después de empezar el cuerpo, no un read timeout estilo Requests y no un retraso antes de headers. Si tu prueba duerme antes de end_headers, no estás midiendo lo mismo.
| Qué viste | Código de este curl 8.7.1 | Qué no es |
|---|---|---|
GET HTTP sin -U, fixture 407 | exit 0, HTTP 407 | Fallo de CONNECT |
Mismo 407 con --fail | exit 22 | Ley universal de todos los 407 |
| CONNECT sin credenciales o mal | exit 56, 407 en el túnel | Origen caído |
| CONNECT bien, CA ausente | exit 60 | Proxy que no abre túnel |
| Puerto sin listener (RST) | exit 7 | Timeout medido |
Cuerpo que se para tras el primer byte, --max-time | exit 28 | Timeout de connect |
No prometas el texto exacto del error entre versiones. Guarda stdout, stderr, estado HTTP, estado CONNECT y exit.
La figura sale de la corrida real: 23 casos capturados y 47 filas de aserto. El original entregó esas 47 filas como 45 comprobaciones condicionales más 2 observaciones incondicionales (True). Esta corrección sustituye esas dos por predicados reales: --fail HTTP exit 22 y CURL_CA_BUNDLE con cuerpo de origen y CAfile. CONNECT 407 es exit 56. CA ausente tras CONNECT es exit 60. Capturar 23 filas no equivale a 23 asertos. La inyección Python no es DNS público.
07Cómo ejecutar el laboratorio incluido
Copia el bloque Python de esta sección a un archivo lab.py. No hay URL pública de descarga y este texto no afirma un adjunto publicado. El padre puede integrar el mismo fuente más adelante; el contrato de entrega es este bloque, no un path de máquina.
- Python 3.14 o compatible,
curl,openssl, Pillow 12.1.1 y cwebp en el PATH. cwebp es obligatorio: el laboratorio llama acwebpal pintar la figura y el exit 0 prometido incluye ese paso. python3 -m pip install Pillow==12.1.1- Extrae el bloque a
lab.pyen un directorio vacío. Ejecutapython3 lab.pydesde ese directorio. - El padre limpia
HTTP_PROXY,HTTPS_PROXY,ALL_PROXY,NO_PROXY,FTP_PROXYy las variantes en minúsculas, más variables CGI y de CA. Relanza un hijo. Cada curl lleva-qcomo primera opción. - Salida esperada: exit 0, 23 capturas, 47 predicados reales, 0 fallidos, 1 inyección de resolver Python (no DNS público). Históricamente el original entregó 47 filas como 45 comprobaciones más 2 observaciones incondicionales.
- Limitación: solo 127.0.0.1. No prueba un proveedor de pago.
El recuadro PLACEHOLDERS más arriba ilustra -x y -U. El laboratorio ejecutable es el bloque siguiente.
#!/usr/bin/env python3
"""Self-contained curl proxy lab. Loopback fixtures only. One entry point."""
from __future__ import annotations
import base64
import hashlib
import json
import os
import re
import select
import shutil
import socket
import ssl
import stat
import subprocess
import sys
import tempfile
import threading
import time
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
if HERE.name == "tests":
ART_ROOT = HERE.parent
RESULTS_DIR = HERE
else:
ART_ROOT = HERE
RESULTS_DIR = HERE
ASSETS = ART_ROOT / "assets"
EVIDENCE = ART_ROOT / "evidence"
RAW = EVIDENCE / "raw"
RESULTS_JSON = RESULTS_DIR / "results.json"
RESULTS_TXT = RESULTS_DIR / "results.txt"
FIGURE_WEBP = ASSETS / "curl-results.webp"
FIGURE_PNG = ASSETS / "curl-results.png"
CA_PUBLIC = EVIDENCE / "ca.pem"
CHILD_FLAG = "CURL_ES_LAB_CHILD"
PARTIAL_HTTPS_FAIL = "CURL_ES_LAB_PARTIAL_HTTPS_FAIL"
BRAVO_START_FAIL = "CURL_ES_LAB_BRAVO_START_FAIL"
CLEANUP_PROBE = "CURL_ES_LAB_CLEANUP_PROBE"
LAB_HOST = "curl-es-lab.origin"
PROXY_USER = "lab-user"
PROXY_PASS = "Aa1@:/#%"
ORIGIN_USER = "origin-user"
ORIGIN_PASS = "origin-lab"
WRONG_PASS = "wrong-lab"
ORIGIN_WRONG_USER = "nobody"
ORIGIN_WRONG_PASS = "nope"
SLOW_BODY = b"H" + b"ELLO-SLOW-BODY!"
SLOW_WAIT_S = 2.4
MAX_TIME_S = "0.8"
CURL_BIN = shutil.which("curl") or "/usr/bin/curl"
CLEAR_KEYS = [
"HTTP_PROXY",
"HTTPS_PROXY",
"ALL_PROXY",
"NO_PROXY",
"FTP_PROXY",
"http_proxy",
"https_proxy",
"all_proxy",
"no_proxy",
"ftp_proxy",
"REQUESTS_CA_BUNDLE",
"CURL_CA_BUNDLE",
"SSL_CERT_FILE",
"SSL_CERT_DIR",
"SSLKEYLOGFILE",
"REQUEST_METHOD",
"GATEWAY_INTERFACE",
"CONTENT_TYPE",
"CONTENT_LENGTH",
"QUERY_STRING",
"SCRIPT_NAME",
"PATH_INFO",
"SERVER_PROTOCOL",
"REMOTE_ADDR",
"SERVER_NAME",
"SERVER_PORT",
"NETRC",
]
def utc_now() -> str:
return datetime.now(timezone.utc).isoformat()
def b64_basic(user: str, password: str) -> str:
return base64.b64encode(f"{user}:{password}".encode()).decode()
def _secret_list() -> list[str]:
pairs = [
(PROXY_USER, PROXY_PASS),
(ORIGIN_USER, ORIGIN_PASS),
(PROXY_USER, WRONG_PASS),
(ORIGIN_WRONG_USER, ORIGIN_WRONG_PASS),
]
secrets = [
PROXY_PASS,
ORIGIN_PASS,
WRONG_PASS,
ORIGIN_WRONG_PASS,
PROXY_USER,
ORIGIN_USER,
ORIGIN_WRONG_USER,
]
for user, password in pairs:
secrets.append(f"{user}:{password}")
secrets.append(quote(password, safe=""))
secrets.append(quote(f"{user}:{password}", safe=""))
secrets.append(b64_basic(user, password))
return [s for s in secrets if s]
def redact_text(value: str) -> str:
out = value
for secret in _secret_list():
out = out.replace(secret, "[redacted]")
out = re.sub(r"(?im)^([>\s]*)(Proxy-Authorization:)\s*.*$", r"\1\2 [redacted]", out)
out = re.sub(r"(?im)^([>\s]*)(Authorization:)\s*.*$", r"\1\2 [redacted]", out)
out = re.sub(r"(?i)(Proxy-Authorization:\s*)\S+", r"\1[redacted]", out)
out = re.sub(r"(?i)(Authorization:\s*)\S+", r"\1[redacted]", out)
out = re.sub(r"://[^/\s]+@", "://[redacted]@", out)
return out
def redact_argv(cmd: list[str]) -> list[str]:
out: list[str] = []
hide_next = False
auth_flags = {"-U", "--proxy-user", "-u", "--user"}
for part in cmd:
if hide_next:
out.append("[redacted-auth]")
hide_next = False
continue
if part in auth_flags:
out.append(part)
hide_next = True
continue
out.append(redact_text(part))
return out
def persisted_text_is_safe(text: str) -> bool:
for secret in _secret_list():
if secret and secret in text:
return False
if re.search(r"(?i)Authorization:\s*Basic\s+[A-Za-z0-9+/=]{8,}", text):
return False
return True
def close_sock(sock: socket.socket | None) -> None:
if sock is None:
return
try:
sock.shutdown(socket.SHUT_RDWR)
except OSError:
pass
try:
sock.close()
except OSError:
pass
def curl_version_blob() -> str:
proc = subprocess.run([CURL_BIN, "-q", "--version"], capture_output=True, text=True)
return (proc.stdout or proc.stderr or "").strip()
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 key, value in extra.items():
self.send_header(key, value)
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 " + b64_basic(ORIGIN_USER, ORIGIN_PASS)
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, http_port: int, https_port: int, user: str | None, password: str | None):
self.name = name
self.http_port = http_port
self.https_port = https_port
self.user = user
self.password = password
self.events: list[dict[str, Any]] = []
self._lock = threading.Lock()
self._stop = threading.Event()
self._workers: list[threading.Thread] = []
self._active: list[socket.socket] = []
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 _track(self, sock: socket.socket) -> None:
with self._lock:
self._active.append(sock)
def workers_alive(self) -> int:
n = 1 if self._thread.is_alive() else 0
n += sum(1 for w in self._workers if w.is_alive())
return n
def start(self) -> None:
self._thread.start()
def _record(self, row: dict[str, Any]) -> None:
with self._lock:
self.events.append(row)
def snapshot(self) -> list[dict[str, Any]]:
with self._lock:
return list(self.events)
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 _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(0.8)
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 _map_dest(self, host: str, port: int) -> tuple[str, int] | None:
key = host.lower()
if key not in ("127.0.0.1", "localhost", LAB_HOST):
return None
if port == self.http_port:
return ("127.0.0.1", self.http_port)
if port == self.https_port:
return ("127.0.0.1", self.https_port)
return None
def _handle(self, conn: socket.socket) -> None:
upstream: socket.socket | None = None
self._track(conn)
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:
key, value = line.split(":", 1)
headers[key.strip().lower()] = value.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)
mapped = self._map_dest(host, int(port_s))
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)
self._track(upstream)
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)
self._track(upstream)
skip = {"proxy-authorization", "proxy-connection"}
out = [f"{method} {path} HTTP/1.1"]
out.append(f"Host: {split.hostname}:{dest_port}" if dest_port != 80 else f"Host: {split.hostname}")
for key, value in headers.items():
if key in skip or key == "host":
continue
out.append(f"{key}: {value}")
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:
close_sock(conn)
close_sock(upstream)
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
finally:
close_sock(a)
close_sock(b)
def stop(self) -> None:
self._stop.set()
with self._lock:
active = list(self._active)
for sock in active:
close_sock(sock)
close_sock(self.sock)
if self._thread.is_alive():
self._thread.join(timeout=2)
for worker in self._workers:
if worker.is_alive():
worker.join(timeout=3)
class Socks5Proxy:
def __init__(self, name: str, allowed_ports: set[int]):
self.name = name
self.allowed_ports = allowed_ports
self.events: list[dict[str, Any]] = []
self._lock = threading.Lock()
self._stop = threading.Event()
self._workers: list[threading.Thread] = []
self._active: list[socket.socket] = []
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 _track(self, sock: socket.socket) -> None:
with self._lock:
self._active.append(sock)
def workers_alive(self) -> int:
n = 1 if self._thread.is_alive() else 0
n += sum(1 for w in self._workers if w.is_alive())
return n
def start(self) -> None:
self._thread.start()
def snapshot(self) -> list[dict[str, Any]]:
with self._lock:
return list(self.events)
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 _handle(self, conn: socket.socket) -> None:
upstream: socket.socket | None = None
self._track(conn)
try:
conn.settimeout(0.8)
hello = conn.recv(2)
if len(hello) < 2 or hello[0] != 5:
return
nmethods = hello[1]
methods = conn.recv(nmethods) if nmethods else b""
if b"\x00" not in methods and nmethods:
conn.sendall(b"\x05\xff")
return
conn.sendall(b"\x05\x00")
req = conn.recv(4)
if len(req) < 4 or req[0] != 5 or req[1] != 1:
conn.sendall(b"\x05\x07\x00\x01\x00\x00\x00\x00\x00\x00")
return
atyp = req[3]
if atyp == 1:
addr_b = conn.recv(4)
host = socket.inet_ntoa(addr_b)
atyp_name = "ipv4"
elif atyp == 3:
ln = conn.recv(1)[0]
host = conn.recv(ln).decode("ascii", "replace")
atyp_name = "domain"
else:
conn.sendall(b"\x05\x08\x00\x01\x00\x00\x00\x00\x00\x00")
return
port_b = conn.recv(2)
port = int.from_bytes(port_b, "big")
with self._lock:
self.events.append({"proxyId": self.name, "atyp": atyp_name, "host": host, "port": port})
if port not in self.allowed_ports:
conn.sendall(b"\x05\x02\x00\x01\x00\x00\x00\x00\x00\x00")
return
if host not in ("127.0.0.1", "localhost", LAB_HOST):
conn.sendall(b"\x05\x02\x00\x01\x00\x00\x00\x00\x00\x00")
return
upstream = socket.create_connection(("127.0.0.1", port), timeout=5)
self._track(upstream)
conn.sendall(b"\x05\x00\x00\x01\x7f\x00\x00\x01" + (0).to_bytes(2, "big"))
sockets = [conn, upstream]
conn.settimeout(None)
upstream.settimeout(None)
while True:
ready, _, _ = select.select(sockets, [], [], 8)
if not ready:
break
for sock in ready:
other = upstream if sock is conn else conn
data = sock.recv(16384)
if not data:
return
other.sendall(data)
except Exception:
return
finally:
close_sock(conn)
close_sock(upstream)
def stop(self) -> None:
self._stop.set()
with self._lock:
active = list(self._active)
for sock in active:
close_sock(sock)
close_sock(self.sock)
if self._thread.is_alive():
self._thread.join(timeout=2)
for worker in self._workers:
if worker.is_alive():
worker.join(timeout=3)
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"
ca_cnf = tmp / "ca.cnf"
origin_cnf = tmp / "origin.cnf"
origin_ext = tmp / "origin.ext"
ca_cnf.write_text(
"[req]\n"
"distinguished_name=dn\n"
"x509_extensions=v3\n"
"prompt=no\n"
"[dn]\n"
"CN=curl-es Lab CA\n"
"[v3]\n"
"basicConstraints=critical,CA:TRUE\n"
"keyUsage=critical,keyCertSign,cRLSign\n"
"subjectKeyIdentifier=hash\n"
)
origin_cnf.write_text("[req]\ndistinguished_name=dn\nprompt=no\n[dn]\nCN=localhost\n")
origin_ext.write_text(
f"subjectAltName=DNS:localhost,DNS:{LAB_HOST},IP:127.0.0.1\n"
"basicConstraints=CA:FALSE\n"
"keyUsage=digitalSignature,keyEncipherment\n"
"extendedKeyUsage=serverAuth\n"
"subjectKeyIdentifier=hash\n"
"authorityKeyIdentifier=keyid,issuer\n"
)
subprocess.run(
[
"openssl",
"req",
"-new",
"-x509",
"-days",
"1",
"-nodes",
"-newkey",
"rsa:2048",
"-keyout",
str(ca_key),
"-out",
str(ca_pem),
"-config",
str(ca_cnf),
],
check=True,
capture_output=True,
text=True,
)
subprocess.run(
[
"openssl",
"req",
"-new",
"-nodes",
"-newkey",
"rsa:2048",
"-keyout",
str(origin_key),
"-out",
str(origin_csr),
"-config",
str(origin_cnf),
],
check=True,
capture_output=True,
text=True,
)
subprocess.run(
[
"openssl",
"x509",
"-req",
"-in",
str(origin_csr),
"-CA",
str(ca_pem),
"-CAkey",
str(ca_key),
"-CAcreateserial",
"-out",
str(origin_crt),
"-days",
"1",
"-extfile",
str(origin_ext),
],
check=True,
capture_output=True,
text=True,
)
for key in (ca_key, origin_key):
key.chmod(stat.S_IRUSR | stat.S_IWUSR)
return {"ca_key": ca_key, "ca_pem": ca_pem, "origin_key": origin_key, "origin_crt": origin_crt}
def render_figure(rows: list[dict[str, Any]], versions: dict[str, Any], ran_at: str) -> dict[str, Any]:
from PIL import Image, ImageDraw, ImageFont
ASSETS.mkdir(parents=True, exist_ok=True)
width = 1400
row_h = 26
header_h = 118
height = header_h + row_h * (len(rows) + 3) + 40
img = Image.new("RGB", (width, height), "#0f172a")
draw = ImageDraw.Draw(img)
try:
font = ImageFont.truetype("/System/Library/Fonts/Supplemental/Arial.ttf", 15)
font_b = ImageFont.truetype("/System/Library/Fonts/Supplemental/Arial Bold.ttf", 22)
font_s = ImageFont.truetype("/System/Library/Fonts/Supplemental/Arial.ttf", 13)
except OSError:
font = ImageFont.load_default()
font_b = font
font_s = font
draw.text((32, 22), "Resultados del laboratorio curl (loopback)", fill="#e2e8f0", font=font_b)
sub = f"{versions.get('curl')} / Python {versions.get('python')} / {ran_at}"
draw.text((32, 56), sub[:110], fill="#c4b5fd", font=font_s)
draw.text(
(32, 78),
"Fixtures locales. Contrasenas y claves omitidas. No es un gateway de pago.",
fill="#94a3b8",
font=font_s,
)
y = header_h
for x, label in ((32, "aserto"), (520, "tipo"), (720, "resultado"), (980, "detalle")):
draw.text((x, y), label, fill="#a78bfa", font=font)
y += row_h
draw.line((32, y, width - 32, y), fill="#4c1d95")
y += 6
for row in rows:
fill = "#86efac" if row.get("passed") else "#fca5a5"
draw.text((32, y), str(row.get("name", ""))[:48], fill="#e2e8f0", font=font)
draw.text((520, y), str(row.get("kind", ""))[:16], fill="#cbd5e1", font=font)
flag = "PASS" if row.get("passed") else "FAIL"
draw.text((720, y), flag, fill=fill, font=font)
draw.text((980, y), str(row.get("detail", ""))[:36], fill="#e2e8f0", font=font)
y += row_h
img.save(FIGURE_PNG, "PNG")
subprocess.run(["cwebp", "-q", "78", str(FIGURE_PNG), "-o", str(FIGURE_WEBP)], check=True, capture_output=True)
data = FIGURE_WEBP.read_bytes()
with Image.open(FIGURE_WEBP) as webp:
dims = webp.size
return {"bytes": len(data), "dimensions": list(dims), "sha256": hashlib.sha256(data).hexdigest()}
def write_results(payload: dict[str, Any]) -> None:
RAW.mkdir(parents=True, exist_ok=True)
text = json.dumps(payload, indent=2, sort_keys=True)
RESULTS_JSON.write_text(text)
lines = [
f"ranAt {payload.get('ranAt')}",
f"finishedAt {payload.get('finishedAt')}",
f"ok {payload.get('ok')}",
f"exitStatus {payload.get('exitStatus')}",
f"captured {payload['counts']['captured']}",
f"asserted {payload['counts']['asserted']}",
f"passed {payload['counts']['passed']}",
f"failed {payload['counts']['failed']}",
f"injected {payload['counts']['injected']}",
f"curl {payload['versions']['curl']}",
f"features {payload['versions']['curlFeatures']}",
"",
]
for row in payload["assertions"]:
flag = "PASS" if row["passed"] else "FAIL"
lines.append(f"{flag} {row['name']} :: {row.get('detail', '')}")
RESULTS_TXT.write_text("\n".join(lines) + "\n")
(RAW / "results.json").write_text(text)
(RAW / "results.txt").write_text(RESULTS_TXT.read_text())
def launch_child() -> int:
env = {
"PATH": os.environ.get("PATH", "/usr/bin:/bin"),
"HOME": os.environ.get("HOME", ""),
"USER": os.environ.get("USER", ""),
"LANG": "C",
"LC_ALL": "C",
"TMPDIR": os.environ.get("TMPDIR", "/tmp"),
CHILD_FLAG: "1",
"PYTHONDONTWRITEBYTECODE": "1",
}
for key in (PARTIAL_HTTPS_FAIL, BRAVO_START_FAIL, CLEANUP_PROBE):
if os.environ.get(key):
env[key] = os.environ[key]
proc = subprocess.run([sys.executable, str(Path(__file__).resolve())], env=env)
return int(proc.returncode)
def run_curl(
extra_args: list[str],
env_inject: dict[str, str] | None = None,
timeout: float = 12.0,
) -> dict[str, Any]:
env = {k: v for k, v in os.environ.items() if k not in CLEAR_KEYS}
if env_inject:
env.update(env_inject)
cmd = [CURL_BIN, "-q", *extra_args]
if cmd[1] != "-q":
raise RuntimeError("curl -q must be the first option")
started = time.monotonic()
proc = subprocess.run(cmd, env=env, capture_output=True, timeout=timeout)
elapsed = time.monotonic() - started
stdout = redact_text(proc.stdout.decode("utf-8", "replace"))
stderr = redact_text(proc.stderr.decode("utf-8", "replace"))
return {
"argv": redact_argv(cmd),
"exit": int(proc.returncode),
"stdout": stdout,
"stderr": stderr,
"elapsedSec": round(elapsed, 3),
"firstOpt": cmd[1],
"httpCode": parse_http_code(stderr, stdout),
"connectStatus": parse_connect_status(stderr),
}
def parse_http_code(stderr: str, stdout: str) -> str | None:
for blob in (stdout, stderr):
m = re.search(r"< HTTP/1\.[01] (\d{3})", blob)
if m:
return m.group(1)
m = re.search(r"HTTP/1\.[01] (\d{3})", blob)
if m:
return m.group(1)
return None
def parse_connect_status(stderr: str) -> str | None:
m = re.search(
r"Establish HTTP proxy tunnel[\s\S]*?< HTTP/1\.[01] (\d{3})",
stderr,
)
if m:
return m.group(1)
if "CONNECT" in stderr:
m = re.search(r"< HTTP/1\.[01] (\d{3})", stderr)
if m:
return m.group(1)
return None
def cleanup(
owned: dict[str, Any],
tmp: Path | None,
real_gai: Any,
shared: dict[str, Any] | None,
prior_netrc: str | None,
) -> dict[str, Any]:
evidence: dict[str, Any] = {
"stopped": [],
"deleted": [],
"resolverRestored": False,
"slowReleased": False,
"workersAliveAfterStop": {},
"netrcRestored": False,
}
if shared is not None:
shared.get("slow_continue", threading.Event()).set()
evidence["slowReleased"] = True
for key in ("alpha", "bravo", "socks"):
obj = owned.get(key)
if obj is not None:
try:
obj.stop()
evidence["stopped"].append(key)
evidence["workersAliveAfterStop"][key] = int(obj.workers_alive())
except Exception as exc:
evidence["stopped"].append(f"{key}:error:{type(exc).__name__}")
for key in ("http_origin", "https_origin"):
obj = owned.get(key)
if obj is not None:
try:
stop_origin(obj)
evidence["stopped"].append(key)
except Exception as exc:
evidence["stopped"].append(f"{key}:error:{type(exc).__name__}")
held = owned.get("held")
if held is not None:
close_sock(held)
evidence["stopped"].append("held-refusal-socket")
if real_gai is not None:
socket.getaddrinfo = real_gai
evidence["resolverRestored"] = True
if prior_netrc is None:
os.environ.pop("NETRC", None)
else:
os.environ["NETRC"] = prior_netrc
evidence["netrcRestored"] = True
if tmp is not None:
for name in ("ca.key", "origin.key", "netrc"):
path = tmp / name
if path.exists():
path.unlink()
evidence["deleted"].append(name)
try:
shutil.rmtree(tmp, ignore_errors=True)
evidence["deleted"].append(str(tmp.name))
except Exception:
pass
evidence["privateKeysPresent"] = any((tmp / n).exists() for n in ("ca.key", "origin.key"))
else:
evidence["privateKeysPresent"] = False
return evidence
def run_lab() -> int:
started = utc_now()
blob = curl_version_blob()
first = blob.splitlines()[0] if blob else ""
features = ""
for line in blob.splitlines():
if line.startswith("Features:"):
features = line.split(":", 1)[1].strip()
versions = {
"python": sys.version.split()[0],
"curl": first,
"curlFeatures": features,
"curlBin": CURL_BIN,
"openssl": subprocess.run(["openssl", "version"], capture_output=True, text=True).stdout.strip(),
"os": sys.platform,
"platform": os.uname().sysname + "-" + os.uname().release + "-" + os.uname().machine,
"interpreter": sys.executable,
"tlsBackend": "SecureTransport" if "SecureTransport" in blob else "unknown",
}
captured: list[dict[str, Any]] = []
assertions: list[dict[str, Any]] = []
injected: list[dict[str, Any]] = []
owned: dict[str, Any] = {}
tmp: Path | None = None
real_gai: Any = None
shared: dict[str, Any] | None = None
prior_netrc = os.environ.get("NETRC")
def capture(name: str, **fields: Any) -> None:
captured.append({"name": name, **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})
cleanup_info: dict[str, Any] = {}
figure_info: dict[str, Any] = {}
ok = False
exit_status = 1
try:
tmp = Path(tempfile.mkdtemp(prefix="curl-es-lab-"))
netrc_path = tmp / "netrc"
netrc_path.write_text("")
netrc_path.chmod(stat.S_IRUSR | stat.S_IWUSR)
os.environ["NETRC"] = str(netrc_path)
real_gai = socket.getaddrinfo
def patched_gai(host: Any, port: Any, *args: Any, **kwargs: Any) -> Any:
if host in (LAB_HOST, b"curl-es-lab.origin"):
host = "127.0.0.1"
return real_gai(host, port, *args, **kwargs)
socket.getaddrinfo = patched_gai # type: ignore[assignment]
injected.append(
{
"branch": "resolver",
"label": "scoped Python getaddrinfo maps curl-es-lab.origin to 127.0.0.1; curl AsynchDNS does not inherit this; not public DNS",
}
)
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", http_port, https_port, PROXY_USER, PROXY_PASS)
owned["alpha"] = alpha
alpha.start()
bravo = HttpForwardProxy("bravo", http_port, https_port, PROXY_USER, PROXY_PASS)
owned["bravo"] = bravo
if os.environ.get(BRAVO_START_FAIL) == "1":
raise RuntimeError("bravo-start-fail")
bravo.start()
socks = Socks5Proxy("socks", {http_port})
owned["socks"] = socks
socks.start()
held = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
held.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
held.bind(("127.0.0.1", 0))
owned["held"] = held
# Darwin drops SYN on bind-without-listen (connect timeout), not RST.
# listen-then-close yields ECONNREFUSED while we still hold `held`.
closed_listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
closed_listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
closed_listener.bind(("127.0.0.1", 0))
refuse_port = int(closed_listener.getsockname()[1])
closed_listener.listen(1)
closed_listener.close()
http_url = f"http://{LAB_HOST}:{http_port}/"
https_url = f"https://{LAB_HOST}:{https_port}/"
http_ip = f"http://127.0.0.1:{http_port}/"
https_ip = f"https://127.0.0.1:{https_port}/"
slow_url = f"http://127.0.0.1:{http_port}/slow-body"
auth_url = f"http://{LAB_HOST}:{http_port}/origin-auth"
proxy_alpha = f"http://127.0.0.1:{alpha.port}"
proxy_bravo = f"http://127.0.0.1:{bravo.port}"
ca = str(certs["ca_pem"])
common = ["-sS", "-v", "--max-time", "8", "--noproxy", ""]
probe = os.environ.get(CLEANUP_PROBE, "")
if probe:
probe_row: dict[str, Any] = {"probe": probe, "alphaPort": alpha.port, "socksPort": socks.port}
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.settimeout(2)
try:
if probe == "partial-header":
client.connect(("127.0.0.1", alpha.port))
client.sendall(b"GET ")
time.sleep(0.15)
probe_row["sent"] = "partial-GET"
elif probe == "active-tunnel":
client.connect(("127.0.0.1", socks.port))
client.sendall(b"\x05\x01\x00")
time.sleep(0.15)
probe_row["sent"] = "socks-hello"
else:
probe_row["sent"] = "none"
finally:
close_sock(client)
RAW.mkdir(parents=True, exist_ok=True)
(RAW / f"cleanup-probe-{probe}.json").write_text(json.dumps(probe_row, indent=2))
raise RuntimeError(f"cleanup-probe:{probe}")
# 1 direct HTTP
before = len(alpha.snapshot())
r = run_curl([*common, http_ip])
capture("direct_http", **r, proxyDelta=alpha.snapshot()[before:])
assert_true("direct_http_first_opt", r["firstOpt"] == "-q", "first option -q")
assert_true("direct_http_200", r["exit"] == 0 and "origin-http" in r["stdout"], f"exit={r['exit']}")
assert_true("direct_http_idle_proxy", len(alpha.snapshot()) == before, "alpha idle")
# 2 explicit HTTP correct auth via -U
before = len(alpha.snapshot())
r = run_curl([*common, "-x", proxy_alpha, "-U", f"{PROXY_USER}:{PROXY_PASS}", http_url])
ev = alpha.snapshot()[before:]
capture("explicit_http_ok", **r, events=ev)
assert_true("explicit_http_ok_200", r["exit"] == 0 and "origin-http" in r["stdout"], f"exit={r['exit']}")
assert_true(
"explicit_http_absolute_form",
any(e["method"] == "GET" and e["authorized"] and LAB_HOST in e["target"] for e in ev),
"GET absolute-form via alpha",
)
assert_true(
"explicit_http_no_leak",
"X-Proxy-Auth-Leaked: 1" not in r["stderr"] and not any(http_origin.leak_flags),
"no origin leak in verbose or fixture",
)
# 3 missing proxy auth HTTP
before = len(alpha.snapshot())
r = run_curl([*common, "-x", proxy_alpha, http_url])
ev = alpha.snapshot()[before:]
code = parse_http_code(r["stderr"], r["stdout"])
capture("http_auth_missing", **r, events=ev)
assert_true("http_auth_missing_407", code == "407" or "407" in r["stderr"] or "407" in r["stdout"], f"code={code}")
assert_true("http_auth_missing_reason", any(e["authReason"] == "missing" for e in ev), "fixture missing")
assert_true(
"http_auth_missing_exit_without_fail",
r["exit"] == 0,
f"exit={r['exit']} without --fail",
)
# 4 wrong proxy auth HTTP
before = len(alpha.snapshot())
r = run_curl([*common, "-x", proxy_alpha, "-U", f"{PROXY_USER}:{WRONG_PASS}", http_url])
ev = alpha.snapshot()[before:]
code = parse_http_code(r["stderr"], r["stdout"])
capture("http_auth_wrong", **r, events=ev)
assert_true("http_auth_wrong_407", code == "407" or "407" in r["stderr"] or "407" in r["stdout"], f"code={code}")
assert_true("http_auth_wrong_reason", any(e["authReason"] == "wrong" for e in ev), "fixture wrong")
# 5 HTTP 407 with --fail
r = run_curl([*common, "--fail", "-x", proxy_alpha, http_url])
capture("http_auth_missing_fail", **r)
assert_true("http_407_fail_nonzero", r["httpCode"] == "407", f"httpCode={r['httpCode']}")
assert_true("http_407_fail_not_universal_22_claim", r["exit"] == 22, f"observed_exit={r['exit']}")
# 6 HTTPS CONNECT correct
before = len(alpha.snapshot())
r = run_curl([*common, "--cacert", ca, "-x", proxy_alpha, "-U", f"{PROXY_USER}:{PROXY_PASS}", https_url])
ev = alpha.snapshot()[before:]
capture("https_connect_ok", **r, events=ev)
assert_true("https_connect_ok_200", r["exit"] == 0 and "origin-https" in r["stdout"], f"exit={r['exit']}")
assert_true(
"https_connect_method",
any(e["method"] == "CONNECT" and e["authorized"] and str(https_port) in e["target"] for e in ev),
"CONNECT then TLS origin",
)
assert_true("https_connect_no_k", "-k" not in r["argv"] and "--insecure" not in r["argv"], "verify on")
# 7 HTTPS CONNECT missing auth
before = len(alpha.snapshot())
r = run_curl([*common, "--cacert", ca, "-x", proxy_alpha, https_url])
ev = alpha.snapshot()[before:]
capture("https_auth_missing", **r, events=ev)
assert_true("https_auth_missing_nonzero", r["exit"] == 56 and r.get("connectStatus") == "407", f"exit={r['exit']} connect={r.get('connectStatus')}")
assert_true(
"https_connect_missing_reason",
any(e["method"] == "CONNECT" and e["authReason"] == "missing" for e in ev),
"CONNECT missing",
)
# 8 HTTPS CONNECT wrong auth
before = len(alpha.snapshot())
r = run_curl([*common, "--cacert", ca, "-x", proxy_alpha, "-U", f"{PROXY_USER}:{WRONG_PASS}", https_url])
ev = alpha.snapshot()[before:]
capture("https_auth_wrong", **r, events=ev)
assert_true("https_auth_wrong_nonzero", r["exit"] == 56 and r.get("connectStatus") == "407", f"exit={r['exit']} connect={r.get('connectStatus')}")
assert_true(
"https_connect_wrong_reason",
any(e["method"] == "CONNECT" and e["authReason"] == "wrong" for e in ev),
"CONNECT wrong",
)
# 9 untrusted CA, still CONNECT
before = len(alpha.snapshot())
r = run_curl([*common, "-x", proxy_alpha, "-U", f"{PROXY_USER}:{PROXY_PASS}", https_url])
ev = alpha.snapshot()[before:]
capture("https_untrusted_ca", **r, events=ev)
assert_true("https_untrusted_nonzero", r["exit"] == 60, f"exit={r['exit']}")
assert_true(
"https_untrusted_connected",
any(e["method"] == "CONNECT" and e["authorized"] for e in ev),
"CONNECT happened; TLS failed",
)
# 10 env http_proxy lowercase
before = len(alpha.snapshot())
r = run_curl(
[*common, "-U", f"{PROXY_USER}:{PROXY_PASS}", http_url],
env_inject={"http_proxy": proxy_alpha},
)
ev = alpha.snapshot()[before:]
capture("env_http_proxy", **r, events=ev)
assert_true("env_http_proxy_used", any(e["proxyId"] == "alpha" and e["method"] == "GET" for e in ev), "alpha GET")
assert_true("env_http_proxy_verbose", "http_proxy" in r["stderr"], r["stderr"][:80])
# 11 uppercase HTTP_PROXY ignored
before_a = len(alpha.snapshot())
r = run_curl(
[*common, http_ip],
env_inject={"HTTP_PROXY": proxy_alpha},
)
capture("env_HTTP_PROXY_ignored", **r, events=alpha.snapshot()[before_a:])
assert_true("env_HTTP_PROXY_idle", len(alpha.snapshot()) == before_a, "alpha idle")
assert_true("env_HTTP_PROXY_direct", "origin-http" in r["stdout"] and r["exit"] == 0, f"exit={r['exit']}")
assert_true("env_HTTP_PROXY_no_uses", "HTTP_PROXY" not in r["stderr"] or "Uses proxy env variable HTTP_PROXY" not in r["stderr"], "not used")
# 12 https_proxy lowercase CONNECT
before = len(alpha.snapshot())
r = run_curl(
[*common, "--cacert", ca, "-U", f"{PROXY_USER}:{PROXY_PASS}", https_url],
env_inject={"https_proxy": proxy_alpha},
)
ev = alpha.snapshot()[before:]
capture("env_https_proxy", **r, events=ev)
assert_true(
"env_https_proxy_connect",
any(e["method"] == "CONNECT" and e["authorized"] for e in ev) and r["exit"] == 0,
f"exit={r['exit']}",
)
# 13 HTTPS_PROXY uppercase CONNECT
before = len(alpha.snapshot())
r = run_curl(
[*common, "--cacert", ca, "-U", f"{PROXY_USER}:{PROXY_PASS}", https_url],
env_inject={"HTTPS_PROXY": proxy_alpha},
)
ev = alpha.snapshot()[before:]
capture("env_HTTPS_PROXY", **r, events=ev)
assert_true(
"env_HTTPS_PROXY_connect",
any(e["method"] == "CONNECT" and e["authorized"] for e in ev) and r["exit"] == 0,
f"exit={r['exit']}",
)
# 14 explicit -x bravo wins over env alpha
before_a = len(alpha.snapshot())
before_b = len(bravo.snapshot())
r = run_curl(
[*common, "-x", proxy_bravo, "-U", f"{PROXY_USER}:{PROXY_PASS}", http_url],
env_inject={"http_proxy": proxy_alpha},
)
ev_a = alpha.snapshot()[before_a:]
ev_b = bravo.snapshot()[before_b:]
capture("explicit_overrides_env", **r, alpha=ev_a, bravo=ev_b)
assert_true("precedence_bravo_used", any(e["method"] == "GET" and e["authorized"] for e in ev_b), "bravo GET")
assert_true("precedence_alpha_idle", not any(e["method"] == "GET" for e in ev_a), "alpha idle")
# 15 bypass --noproxy host
before = len(alpha.snapshot())
r = run_curl(
["-sS", "-v", "--max-time", "8", "--noproxy", "127.0.0.1", "-U", f"{PROXY_USER}:{PROXY_PASS}", http_ip],
env_inject={"http_proxy": proxy_alpha},
)
ev = alpha.snapshot()[before:]
capture("noproxy_bypass", **r, events=ev)
assert_true("noproxy_bypass_direct", r["exit"] == 0 and "origin-http" in r["stdout"], f"exit={r['exit']}")
assert_true("noproxy_bypass_idle", not any(e["method"] == "GET" for e in ev), "alpha idle")
# 16 origin -u vs proxy -U
before = len(alpha.snapshot())
r = run_curl(
[
*common,
"-x",
proxy_alpha,
"-U",
f"{PROXY_USER}:{PROXY_PASS}",
"-u",
f"{ORIGIN_USER}:{ORIGIN_PASS}",
auth_url,
]
)
ev = alpha.snapshot()[before:]
capture("origin_and_proxy_auth", **r, events=ev)
assert_true("dual_auth_200", r["exit"] == 0 and "origin-auth-ok" in r["stdout"], f"exit={r['exit']}")
assert_true("dual_auth_proxy_ok", any(e["authorized"] for e in ev), "proxy authorized")
r = run_curl(
[*common, "-x", proxy_alpha, "-U", f"{PROXY_USER}:{PROXY_PASS}", "-u", f"{ORIGIN_WRONG_USER}:{ORIGIN_WRONG_PASS}", auth_url]
)
capture("origin_wrong_proxy_ok", **r)
assert_true("origin_401_distinct", "401" in r["stderr"] or "ORIGIN-401" in r["stdout"], "origin 401")
# 17 owned refusal, not timeout
r = run_curl(["-sS", "-v", "--max-time", "4", f"http://127.0.0.1:{refuse_port}/"])
capture("connection_refusal", **r, refusePort=refuse_port)
assert_true("refusal_exit7", r["exit"] == 7, f"exit={r['exit']}")
assert_true("refusal_not_timeout28", r["exit"] != 28, "connection error not transfer timeout")
# 18 delayed body / --max-time transfer timeout
shared["first_byte"].clear()
shared["slow_continue"].clear()
r = run_curl(["-sS", "-v", "--max-time", MAX_TIME_S, "--noproxy", "", slow_url], timeout=8)
first = shared["first_byte"].is_set()
capture("slow_body_max_time", **r, firstByte=first)
assert_true("slow_first_byte", first, "headers and first byte flushed")
assert_true("slow_exit28", r["exit"] == 28, f"exit={r['exit']} transfer timeout")
shared["slow_continue"].set()
# 19 SOCKS5 ATYP ipv4
before = len(socks.snapshot())
r = run_curl([*common, "-x", f"socks5://127.0.0.1:{socks.port}", http_ip])
ev = socks.snapshot()[before:]
capture("socks5_ipv4", **r, events=ev)
assert_true("socks5_ok", r["exit"] == 0 and "origin-http" in r["stdout"], f"exit={r['exit']}")
assert_true("socks5_atyp_ipv4", any(e["atyp"] == "ipv4" and e["host"] == "127.0.0.1" for e in ev), str(ev))
# 20 SOCKS5h ATYP domain
before = len(socks.snapshot())
r = run_curl([*common, "-x", f"socks5h://127.0.0.1:{socks.port}", http_url])
ev = socks.snapshot()[before:]
capture("socks5h_domain", **r, events=ev)
assert_true("socks5h_ok", r["exit"] == 0 and "origin-http" in r["stdout"], f"exit={r['exit']}")
assert_true(
"socks5h_atyp_domain",
any(e["atyp"] == "domain" and e["host"] == LAB_HOST for e in ev),
str(ev),
)
# 21 CURL_CA_BUNDLE vs --cacert (measure, do not invent)
r_bundle = run_curl(
["-sS", "-v", "--max-time", "8", "--noproxy", "", https_ip],
env_inject={"CURL_CA_BUNDLE": ca},
)
r_cacert = run_curl(["-sS", "-v", "--max-time", "8", "--noproxy", "", "--cacert", ca, https_ip])
capture("ca_bundle_env", **r_bundle)
capture("cacert_flag", **r_cacert)
assert_true("cacert_direct_https", r_cacert["exit"] == 0 and "origin-https" in r_cacert["stdout"], f"exit={r_cacert['exit']}")
assert_true(
"ca_bundle_measured",
r_bundle["exit"] == 0 and "origin-https" in r_bundle["stdout"] and "CAfile:" in r_bundle["stderr"],
f"CURL_CA_BUNDLE_exit={r_bundle['exit']} cafile_logged={'CAfile:' in r_bundle['stderr']}",
)
# 22 no --netrc in argv
captured_blob = json.dumps(captured)
assert_true(
"no_netrc_flag",
all("--netrc" not in c.get("argv", []) and "--netrc-optional" not in c.get("argv", []) for c in captured)
and persisted_text_is_safe(captured_blob),
"curl invoked without --netrc; persisted captures redacted",
)
failed = [a for a in assertions if not a["passed"]]
ok = len(failed) == 0
exit_status = 0 if ok else 1
figure_rows = [
{
"name": a["name"],
"kind": a.get("kind", "assert"),
"passed": a["passed"],
"detail": a.get("detail", ""),
"result": "PASS" if a["passed"] else "FAIL",
}
for a in assertions
]
figure_info = render_figure(figure_rows, versions, started)
except Exception as exc:
ok = False
exit_status = 2
capture("lab_exception", error=redact_text(f"{type(exc).__name__}: {exc}"))
assert_true("lab_exception", False, redact_text(f"{type(exc).__name__}: {exc}"))
failed = [a for a in assertions if not a["passed"]]
if os.environ.get(PARTIAL_HTTPS_FAIL) == "1":
exit_status = 2
finally:
cleanup_info = cleanup(owned, tmp, real_gai, shared, prior_netrc)
failed = [a for a in assertions if not a["passed"]]
meaningful = [a for a in assertions if a.get("kind") != "observation"]
payload = {
"ranAt": started,
"finishedAt": utc_now(),
"ok": ok and len(failed) == 0,
"exitStatus": exit_status,
"versions": versions,
"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),
"meaningfulChecks": len(meaningful),
"observations": sum(1 for a in assertions if a.get("kind") == "observation"),
"historicalOriginal": {
"asserted": 47,
"meaningfulChecks": 45,
"unconditionalObservations": 2,
},
},
"failedNames": [a["name"] for a in assertions if not a["passed"]],
"injected": injected,
"captured": captured,
"assertions": assertions,
"cleanup": cleanup_info,
"figure": figure_info,
"notes": {
"trust_env": "curl has no trust_env; -q isolates curlrc; env CA measured separately",
"https_proxy_scheme": "HTTP proxy URL used for HTTPS CONNECT; not an HTTPS-proxy TLS hop",
"passwords": "redacted",
"verifyFalse": False,
},
}
skip_main_results = bool(
os.environ.get(PARTIAL_HTTPS_FAIL) == "1"
or os.environ.get(BRAVO_START_FAIL) == "1"
or os.environ.get(CLEANUP_PROBE)
)
if not skip_main_results:
try:
write_results(payload)
except Exception:
pass
RAW.mkdir(parents=True, exist_ok=True)
if os.environ.get(PARTIAL_HTTPS_FAIL) == "1":
(RAW / "partial-https-fail.json").write_text(
json.dumps(
{
"exitStatus": exit_status,
"error": captured[-1] if captured else None,
"cleanup": cleanup_info,
"httpOriginStopped": "http_origin" in cleanup_info.get("stopped", []),
"workersAliveAfterStop": cleanup_info.get("workersAliveAfterStop"),
},
indent=2,
)
)
if os.environ.get(BRAVO_START_FAIL) == "1":
(RAW / "bravo-start-fail.json").write_text(
json.dumps(
{
"exitStatus": exit_status,
"error": captured[-1] if captured else None,
"cleanup": cleanup_info,
"alphaRegistered": "alpha" in owned,
"bravoRegistered": "bravo" in owned,
"workersAliveAfterStop": cleanup_info.get("workersAliveAfterStop"),
},
indent=2,
)
)
if os.environ.get(CLEANUP_PROBE):
name = os.environ.get(CLEANUP_PROBE)
path = RAW / f"cleanup-probe-{name}.json"
prior = {}
if path.exists():
try:
prior = json.loads(path.read_text())
except Exception:
prior = {}
prior.update(
{
"exitStatus": exit_status,
"cleanup": cleanup_info,
"workersAliveAfterStop": cleanup_info.get("workersAliveAfterStop"),
}
)
path.write_text(json.dumps(prior, indent=2))
return exit_status
def main() -> int:
if os.environ.get(CHILD_FLAG) != "1":
return launch_child()
return run_lab()
if __name__ == "__main__":
raise SystemExit(main())
08Cómo elegir un endpoint documentado
El laboratorio no pagó Oxylabs ni Decodo. Lo que sigue es forma de endpoint según su documentación, no latencia ni países.
La guía de Residential Proxies de Oxylabs (14 de octubre de 2026) declara soporte HTTP, HTTPS, HTTP/3 (MASQUE) y SOCKS5, autenticación por usuario/contraseña o por IP en lista, y el nodo pr.oxylabs.io:7777 con customer-USERNAME:PASSWORD. El ejemplo oficial de cURL usa -x y -U. Eso encaja con el modelo de esta guía. No prueba su pool.
El quick start residencial de Decodo (mismo día) muestra gate.decodo.com:7000, cURL con -U y -x, y en Python un mapa http/https hacia la misma URL HTTP del proxy. En el catálogo en español de este sitio esa ficha sigue el slug smartproxy-com. El nombre público actual es Decodo. El slug no se inventa.
Comprueba el host con el checker y mira la IP del navegador en Mi IP si estás depurando el browser. Ni uno ni otro sustituyen curl -q -v.
Oxylabs vende proxies residenciales, de datacenter y móviles. El residencial de autoservicio empieza en $6/GB para 5 GB/mes. Pagos del comprador: tarjetas, transferencia, AliPay, PayPal.
Decodo (antes Smartproxy) vende proxies residenciales y de datacenter. Residencial: $4/GB sin plan mensual, o paquetes mensuales desde $3.75/GB por 3 GB. Pagos: tarjetas, PayPal, Alipay, Google/Apple Pay, cripto (con límites).
Las fichas públicas Oxylabs y Decodo (smartproxy-com) respondieron 200 el 14 de octubre de 2026. Eso es presencia pública, no un test de cuenta pagada.
Si eliges residencial o móvil, filtra por protocolo en el catálogo y vuelve a curl. El laboratorio de esta página solo demuestra loopback.
¿Cómo configuro un proxy en curl?
¿Por qué un destino HTTPS puede usar un proxy HTTP?
¿En qué se diferencian -U y -u?
¿Por qué curl ignora HTTP_PROXY?
¿Dónde resuelve el nombre socks5h?
¿Qué limitan --connect-timeout y --max-time?
Artículos relacionados
Gemini Developer API: configurar un proxy HTTP y diagnosticar la conexión
Enruta google-genai 2.30.1 por HTTPS_PROXY y CONNECT, conserva TLS con SSL_CERT_FILE y separa un fallo de transporte de un 403 o un timeout.
Configurar un proxy en HTTPX: autenticación, entorno y tiempos de espera
Configura proxy= en HTTPX 0.28.1, separa las credenciales del proxy y del sitio, deja HTTPS en CONNECT y distingue entorno, 407 y ReadTimeout.
Cómo configurar un proxy en Playwright
Configura un proxy HTTP o SOCKS5 en Playwright Python, en el browser o en el context, comprueba CONNECT y 407, y deja TLS encendido.
