Como configurar proxy no Playwright

Configure um proxy HTTP ou SOCKS5 no Playwright Python, no browser ou no context, confira CONNECT e 407, e deixe o TLS ligado.

Este guia é para quem sobe Playwright em Python e quer que o Chromium saia por um proxy HTTP ou SOCKS5, e que saiba ler um 407 diferente de um certificado recusado. Os exemplos usam credenciais dummy. A Top Proxy não vende IP e não distribui lista grátis. Um 200 de laboratório não prova geografia, conta nem qualidade de fornecedor.

01Definir o proxy no browser ou no context#

A documentação de rede do Playwright para Python de 14 de outubro de 2026 diz que o proxy pode ir no browser inteiro ou em um context. A mesma forma aparece em browser_type.launch e em browser.new_context. Os campos são server, bypass opcional, username opcional e password opcional. O laboratório executou Playwright para Python 1.64.0 contra Chromium 156.0.8078.4 em Python 3.14.6. Node não foi um segundo laboratório.

Coloque o proxy em chromium.launch se todos os contexts do mesmo processo devem sair pelo mesmo host. Coloque em browser.new_context se um processo precisa de duas saídas ou de dois potes de cookie.

1. Decida se o percurso pede uma só saída ou duas identidades.
2. Para uma saída, passe proxy para launch.
3. Para duas saídas, suba o browser sem a segunda identidade e depois um proxy distinto por context.
4. Feche cada context que você criou antes de browser.close(). A API do browser marca isso como limpeza, não como enfeite.

Sem proxy, o Chromium bateu na origem HTTP local com estado 200. O proxy de loop aberto contou zero eventos. Antes de subir, o laboratório apagou HTTP_PROXY, HTTPS_PROXY, ALL_PROXY e NO_PROXY. Isso é uma base limpa sem opção de proxy. Não prova como o Playwright trataria uma variável HTTP_PROXY herdada.

Um segundo browser subiu com o proxy aberto como valor de launch. Um context recebeu proxy.server apontando para outro forwarder local. O pedido voltou com 200. Os golpes caíram no segundo forwarder. O primeiro não viu a URL de origem daquele context. O proxy do context substitui o valor do browser para aquele context. Não se misturam.

Em 1.64.0 não existe rotateSession em launch nem em new_context. Quem quiser girar IP faz isso no produto do fornecedor, não nesta API.

02Passar host, porta e credenciais sem vazá-las#

server é uma URL. A documentação de launch cita http://myproxy.com:3128 e socks5://myproxy.com:3128. A forma curta myproxy.com:3128 é tomada como HTTP. Se o fornecedor fala SOCKS5, o esquema tem de ser socks5://. username e password estão documentados para um proxy HTTP com login, não como userpass de SOCKS5.

A senha vive no ambiente ou no cofre de segredos, não em arquivo commitado. O Playwright mesmo assim manda Proxy-Authorization no fio. O laboratório usou o usuário dummy lab-user e uma senha dummy com @, :, /, # e %. Nos logs de resultado aparece ***.

Contra o proxy Basic local, o Chromium se comportou assim:

1. Só server, sem usuário. page.goto na origem HTTP devolveu estado 407. O proxy escreveu authReason: missing. O Playwright não lançou exceção. O estado do objeto de resposta foi a prova.
2. Usuário lab-user e uma senha falsa. Outra vez estado 407. O log mostrou authReason: wrong.
3. A senha dummy correta. Estado 200. O log mostrou authorized: true.

Credencial falsa, neste Chromium, não é socket pendurado. É HTTP 407 do proxy, visível como estado de navegação. Se um wrapper mais acima lança, conserve o 407 e a classe da exceção. Não funda os dois em "o proxy não vai".

A autenticação HTTP da mesma página de rede é http_credentials para a origem, não proxy.username. Um 401 da origem e um 407 do proxy são saltos distintos. Quem mistura os dois debuga a camada errada.

03Confirmar HTTPS via proxy HTTP (CONNECT)#

O esquema da URL de destino e o esquema da URL do proxy são dois controles. Uma origem https://example.com/ ainda sai por um proxy HTTP em http://127.0.0.1:<porta>. O método do proxy é CONNECT, destino example.com:443. O laboratório viu exatamente esse CONNECT. Depois o Chromium conferiu TLS contra o certificado público e page.goto devolveu 200. Isso é HTTPS através de um proxy HTTP com verificação ligada.

Uma origem HTTPS local com CA de laboratório é a outra metade. Chromium 156.0.8078.4, ignore_https_errors no valor documentado False, navegou para https://127.0.0.1:<porta-https>/ pelo mesmo proxy HTTP. O proxy escreveu CONNECT 127.0.0.1:<porta-https>. Depois o Playwright lançou Error com net::ERR_CERT_AUTHORITY_INVALID. O CONNECT ocorreu. O TLS mesmo assim falhou. O destino carregar https:// não transforma o proxy em um "proxy https". O proxy aceitar CONNECT não faz o Chromium confiar numa CA privada.

Python ssl contra a mesma origem sem CA terminou em SSLCertVerificationError: unable to get local issuer certificate. Com a CA de laboratório e verify_mode=CERT_REQUIRED chegou HTTP 200. Esse é o caminho de confiança fora do browser. client_certificates em new_context, acrescentado na v1.46, é certificado de cliente para uma origem nomeada, não um arquivo de CA. O interruptor documentado do Playwright que pulharia o erro é ignore_https_errors=True. Este texto não o liga. verify=False não foi um caso.

Se a tarefa é "provar CONNECT", leia o log do proxy em CONNECT host:porta. Se a tarefa é "página carregada", você precisa de um certificado que o Chromium já aceite, ou de uma CA fora deste guia. Uma página de checagem no browser privado é a saída do browser, não o caminho do Playwright. Qual é o meu IP serve para a saída do browser. O checador de proxies diz se um host aceita HTTP ou SOCKS5 antes de apontar o Playwright. Nenhum substitui um trace do Playwright.

A figura é a tabela deste percurso, não uma capa 16:9 nem uma captura do checador.

O script público abaixo sobe origem, proxy CONNECT e Chromium no local, escreve JSON no stdout e termina com código 0 ou 1. Não pede sidecars privados. As chaves temporárias apaga no finally.

04Percurso executável em Playwright para Python#

Os exemplos de rede em Python usam chromium.launch(proxy={...}) e browser.new_context(proxy={...}). Sync e async carregam as mesmas chaves. Este laboratório executa o binding sync.

from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch(
        proxy={
            "server": "http://127.0.0.1:3128",
            "username": "lab-user",
            "password": "***",
        }
    )
    page = browser.new_page()
    page.set_default_timeout(8000)
    page.goto("http://127.0.0.1:8080/")
    browser.close()

O recorte acima é o desenho mínimo. O sidecar abaixo é o percurso que o leitor consegue rodar sozinho. Origem HTTP, origem HTTPS com CA de um dia, proxy CONNECT com Basic dummy, Chromium com TLS ligado. O JSON no stdout é a prova, não um print de marketing. As portas são efêmeras. As chaves .key saem no finally.

#!/usr/bin/env python3
"""Laboratório independente de loopback para Playwright Python.

Credenciais dummy. Origem local e proxy CONNECT. Sem provedor pago.
TLS permanece verificado. Este script não liga ignore_https_errors.

    python3 -m pip install playwright==1.64.0
    python3 -m playwright install chromium
    python3 public_lab.py
"""
from __future__ import annotations

import base64
import json
import os
import shutil
import socket
import ssl
import subprocess
import sys
import tempfile
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlsplit

USER = "lab-user"
PASSWORD = "p@ss:" + "/#%"
REDACT = "***"
TIMEOUT_MS = 8000
JOIN_S = 2.0
STRIP = [
    "HTTP_PROXY",
    "HTTPS_PROXY",
    "ALL_PROXY",
    "NO_PROXY",
    "http_proxy",
    "https_proxy",
    "all_proxy",
    "no_proxy",
]


def redact(s: object) -> str:
    text = str(s)
    blob = base64.b64encode(f"{USER}:{PASSWORD}".encode("utf-8")).decode("ascii")
    return text.replace(PASSWORD, REDACT).replace(blob, REDACT)


def isolate_env() -> None:
    for key in STRIP:
        os.environ.pop(key, None)


def basic() -> str:
    return "Basic " + base64.b64encode(f"{USER}:{PASSWORD}".encode("utf-8")).decode("ascii")


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 leftover_listen(ports: list[int]) -> list[int]:
    still = []
    for port in ports:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(0.2)
        try:
            s.connect(("127.0.0.1", int(port)))
            still.append(int(port))
        except OSError:
            pass
        finally:
            s.close()
    return still


class Origin(BaseHTTPRequestHandler):
    def log_message(self, format: str, *args: object) -> None:  # noqa: A003
        return

    def do_GET(self) -> None:  # noqa: N802
        body = json.dumps({"ok": True, "path": self.path, "via": "http-origin"}).encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)


class TrackingHTTPServer(ThreadingHTTPServer):
    def __init__(self, addr: tuple[str, int], handler: type[BaseHTTPRequestHandler], track_name: str = "http"):
        super().__init__(addr, handler)
        self.track_name = track_name
        self.accepted: list[socket.socket] = []
        self.handlers: list[threading.Thread] = []
        self.handler_recs: list[dict[str, object]] = []
        self.daemon_threads = True
        self.ssl_ctx: ssl.SSLContext | None = None
        self.handler_start_fault = False
        self.last_start_error: str | None = None
        self.handshake_s = 1.0

    def get_request(self) -> tuple[socket.socket, object]:
        raw, addr = self.socket.accept()
        self.accepted.append(raw)
        if self.ssl_ctx is None:
            return raw, addr
        raw.settimeout(self.handshake_s)
        try:
            tls = self.ssl_ctx.wrap_socket(raw, server_side=True)
        except Exception:
            raise
        tls.settimeout(None)
        self.accepted.append(tls)
        return tls, addr

    def process_request(self, request: socket.socket, client_address: object) -> None:
        t = threading.Thread(
            target=self.process_request_thread,
            args=(request, client_address),
            daemon=True,
            name=f"{self.track_name}-handler",
        )
        rec: dict[str, object] = {"thread": t, "started": False, "name": t.name}
        self.handlers.append(t)
        self.handler_recs.append(rec)
        try:
            if self.handler_start_fault:
                self.handler_start_fault = False
                real = threading.Thread.start

                def boom(th_self: threading.Thread, *a: object, **k: object) -> None:
                    threading.Thread.start = real
                    raise RuntimeError("inject-handler-start")

                threading.Thread.start = boom  # type: ignore[method-assign]
                try:
                    t.start()
                finally:
                    threading.Thread.start = real
            else:
                t.start()
            rec["started"] = True
        except Exception as e:
            rec["started"] = False
            self.last_start_error = str(e)
            close_sock(request)
            raise


def _serve_loop(server: ThreadingHTTPServer, stop: threading.Event) -> None:
    try:
        server.socket.settimeout(0.2)
        while not stop.is_set():
            try:
                server.handle_request()
            except OSError:
                if stop.is_set():
                    break
    except Exception:
        return


class OwnedHttp:
    def __init__(self, server: TrackingHTTPServer, thread: threading.Thread | None, started: bool):
        self.server = server
        self.thread = thread
        self.started = started
        self.port = int(server.server_address[1])
        self.stop = threading.Event()
        self.helpers: list[threading.Thread] = []

    def close(self) -> list[str]:
        errors: list[str] = []
        try:
            self.stop.set()
        except Exception as e:  # noqa: BLE001
            errors.append(str(e))
        for s in list(self.server.accepted):
            try:
                close_sock(s)
            except Exception as e:  # noqa: BLE001
                errors.append(str(e))
        try:
            self.server.server_close()
        except Exception as e:  # noqa: BLE001
            errors.append(str(e))
        try:
            close_sock(getattr(self.server, "socket", None))
        except Exception as e:  # noqa: BLE001
            errors.append(str(e))
        recs = list(getattr(self.server, "handler_recs", []) or [])
        if recs:
            for rec in recs:
                th = rec.get("thread")
                if rec.get("started") and isinstance(th, threading.Thread):
                    try:
                        th.join(JOIN_S)
                        if th.is_alive():
                            errors.append(f"handler-alive:{th.name}")
                    except RuntimeError as e:
                        errors.append(str(e))
        else:
            for th in list(self.server.handlers):
                try:
                    th.join(JOIN_S)
                    if th.is_alive():
                        errors.append(f"handler-alive:{th.name}")
                except RuntimeError as e:
                    errors.append(str(e))
        if self.started and self.thread is not None:
            try:
                self.thread.join(JOIN_S)
                if self.thread.is_alive():
                    errors.append(f"thread-alive:{self.thread.name}")
            except RuntimeError as e:
                errors.append(str(e))
        for th in list(self.helpers):
            try:
                th.join(JOIN_S)
                if th.is_alive():
                    errors.append(f"helper-alive:{th.name}")
            except RuntimeError as e:
                errors.append(str(e))
        return errors

    def live_threads(self) -> list[threading.Thread]:
        out: list[threading.Thread] = []
        if self.thread is not None:
            out.append(self.thread)
        out.extend(self.server.handlers)
        out.extend(self.helpers)
        return out


def _call_thread_start(thread: threading.Thread, fault: bool) -> None:
    if not fault:
        thread.start()
        return
    real = threading.Thread.start

    def boom(th_self: threading.Thread, *a: object, **k: object) -> None:
        threading.Thread.start = real
        raise RuntimeError("inject-thread-start")

    threading.Thread.start = boom  # type: ignore[method-assign]
    try:
        thread.start()
    finally:
        threading.Thread.start = real


def start_http_origin(fault_start: bool = False) -> OwnedHttp:
    server = TrackingHTTPServer(("127.0.0.1", 0), Origin, track_name="public-http")
    owned = OwnedHttp(server, None, False)
    thread = threading.Thread(target=lambda: _serve_loop(server, owned.stop), daemon=True, name="public-http")
    owned.thread = thread
    try:
        _call_thread_start(thread, fault_start)
        owned.started = True
        return owned
    except Exception:
        owned.close()
        raise


def make_certs(tmp: str) -> dict[str, str]:
    ca_key = os.path.join(tmp, "ca.key")
    ca_pem = os.path.join(tmp, "ca.pem")
    origin_key = os.path.join(tmp, "origin.key")
    origin_csr = os.path.join(tmp, "origin.csr")
    origin_pem = os.path.join(tmp, "origin.pem")
    ca_cnf = os.path.join(tmp, "ca.cnf")
    ext = os.path.join(tmp, "ext.cnf")
    with open(ca_cnf, "w", encoding="utf-8") as fh:
        fh.write(
            "[req]\n"
            "distinguished_name=dn\n"
            "x509_extensions=v3_ca\n"
            "prompt=no\n"
            "[dn]\n"
            "CN=pw-pt-public-lab-ca\n"
            "[v3_ca]\n"
            "basicConstraints=critical,CA:TRUE\n"
            "keyUsage=critical,keyCertSign,cRLSign\n"
            "subjectKeyIdentifier=hash\n"
        )
    with open(ext, "w", encoding="utf-8") as fh:
        fh.write(
            "subjectAltName=IP:127.0.0.1,DNS:localhost\n"
            "basicConstraints=CA:FALSE\n"
            "keyUsage=digitalSignature,keyEncipherment\n"
            "extendedKeyUsage=serverAuth\n"
            "subjectKeyIdentifier=hash\n"
            "authorityKeyIdentifier=keyid,issuer\n"
        )

    def run(args: list[str]) -> None:
        r = subprocess.run(["openssl", *args], capture_output=True, text=True)
        if r.returncode != 0:
            raise RuntimeError(r.stderr or r.stdout)

    run(["req", "-x509", "-newkey", "rsa:2048", "-keyout", ca_key, "-out", ca_pem, "-days", "1", "-nodes", "-config", ca_cnf])
    run(["req", "-newkey", "rsa:2048", "-keyout", origin_key, "-out", origin_csr, "-nodes", "-subj", "/CN=localhost"])
    run(["x509", "-req", "-in", origin_csr, "-CA", ca_pem, "-CAkey", ca_key, "-CAcreateserial", "-out", origin_pem, "-days", "1", "-extfile", ext])
    return {
        "caPem": open(ca_pem, encoding="utf-8").read(),
        "originKey": origin_key,
        "originPem": origin_pem,
    }


def start_https_origin(certs: dict[str, str], fault_tls: bool = False, fault_start: bool = False) -> OwnedHttp:
    server = TrackingHTTPServer(("127.0.0.1", 0), Origin, track_name="public-https")
    owned = OwnedHttp(server, None, False)
    try:
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        ctx.minimum_version = ssl.TLSVersion.TLSv1_2
        if fault_tls:
            ctx.load_cert_chain("/nonexistent-pw-pt-public.pem", "/nonexistent-pw-pt-public.key")
        else:
            ctx.load_cert_chain(certs["originPem"], certs["originKey"])
        server.ssl_ctx = ctx
        thread = threading.Thread(target=lambda: _serve_loop(server, owned.stop), daemon=True, name="public-https")
        owned.thread = thread
        _call_thread_start(thread, fault_start)
        owned.started = True
        return owned
    except Exception:
        owned.close()
        raise


class ConnectProxy:
    def __init__(self, fault_handler: bool = False) -> None:
        self.events: list[dict[str, object]] = []
        self.clients: list[socket.socket] = []
        self.threads: list[threading.Thread] = []
        self.handler_recs: list[dict[str, object]] = []
        self.handler_start_fault = fault_handler
        self.last_start_error: str | None = None
        self.stop = threading.Event()
        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(16)
        self.port = int(self.sock.getsockname()[1])
        self.accept_thread = threading.Thread(target=self._accept, daemon=True, name="public-proxy-accept")
        self.accept_started = False
        try:
            self.accept_thread.start()
            self.accept_started = True
        except Exception:
            close_sock(self.sock)
            for client in list(self.clients):
                close_sock(client)
            raise

    def _handle(self, client: socket.socket) -> None:
        dest = None
        try:
            buf = b""
            while b"\r\n\r\n" not in buf:
                chunk = client.recv(4096)
                if not chunk:
                    return
                buf += chunk
            head, rest = buf.split(b"\r\n\r\n", 1)
            lines = head.decode("latin1").split("\r\n")
            method, target, *_rest = lines[0].split(" ")
            headers = {}
            for line in lines[1:]:
                if ":" in line:
                    k, v = line.split(":", 1)
                    headers[k.strip().lower()] = v.strip()
            authorized = headers.get("proxy-authorization") == basic()
            self.events.append({"method": method, "target": target, "auth": authorized})
            if not authorized:
                client.sendall(
                    b"HTTP/1.1 407 Proxy Authentication Required\r\nProxy-Authenticate: Basic realm=\"lab\"\r\nContent-Length: 0\r\n\r\n"
                )
                return
            if method == "CONNECT":
                host, port_s = target.split(":")
                dest = socket.create_connection((host, int(port_s)), timeout=5)
                self.clients.append(dest)
                client.sendall(b"HTTP/1.1 200 Connection Established\r\n\r\n")
                _pipe(client, dest)
                return
            split = urlsplit(target)
            dest = socket.create_connection((split.hostname, split.port or 80), timeout=5)
            self.clients.append(dest)
            path = split.path or "/"
            if split.query:
                path += "?" + split.query
            dest.sendall(f"{method} {path} HTTP/1.1\r\nHost: {split.netloc}\r\nConnection: close\r\n\r\n".encode("latin1") + rest)
            while True:
                data = dest.recv(8192)
                if not data:
                    break
                client.sendall(data)
        except OSError:
            return
        finally:
            close_sock(client)
            close_sock(dest)

    def _accept(self) -> None:
        self.sock.settimeout(0.3)
        while not self.stop.is_set():
            try:
                client, _addr = self.sock.accept()
            except socket.timeout:
                continue
            except OSError:
                break
            self.clients.append(client)
            t = threading.Thread(target=self._handle, args=(client,), daemon=True, name="public-proxy-handler")
            rec: dict[str, object] = {"thread": t, "started": False, "name": t.name}
            self.threads.append(t)
            self.handler_recs.append(rec)
            try:
                if self.handler_start_fault:
                    self.handler_start_fault = False
                    real = threading.Thread.start

                    def boom(th_self: threading.Thread, *a: object, **k: object) -> None:
                        threading.Thread.start = real
                        raise RuntimeError("inject-handler-start")

                    threading.Thread.start = boom  # type: ignore[method-assign]
                    try:
                        t.start()
                    finally:
                        threading.Thread.start = real
                else:
                    t.start()
                rec["started"] = True
            except Exception as e:
                rec["started"] = False
                self.last_start_error = str(e)
                close_sock(client)

    def close(self) -> list[str]:
        errors: list[str] = []
        try:
            self.stop.set()
        except Exception as e:  # noqa: BLE001
            errors.append(str(e))
        try:
            close_sock(self.sock)
        except Exception as e:  # noqa: BLE001
            errors.append(str(e))
        for client in list(self.clients):
            try:
                close_sock(client)
            except Exception as e:  # noqa: BLE001
                errors.append(str(e))
        if self.accept_started:
            try:
                self.accept_thread.join(JOIN_S)
                if self.accept_thread.is_alive():
                    errors.append("accept-thread-alive")
            except RuntimeError as e:
                errors.append(str(e))
        recs = list(self.handler_recs)
        if recs:
            for rec in recs:
                th = rec.get("thread")
                if rec.get("started") and isinstance(th, threading.Thread):
                    try:
                        th.join(JOIN_S)
                        if th.is_alive():
                            errors.append(f"handler-alive:{th.name}")
                    except RuntimeError as e:
                        errors.append(str(e))
        else:
            for th in self.threads:
                try:
                    th.join(JOIN_S)
                    if th.is_alive():
                        errors.append(f"handler-alive:{th.name}")
                except RuntimeError as e:
                    errors.append(str(e))
        return errors


def _pipe(a: socket.socket, b: socket.socket) -> None:
    def one(src: socket.socket, dst: socket.socket) -> None:
        try:
            while True:
                data = src.recv(8192)
                if not data:
                    break
                dst.sendall(data)
        except OSError:
            return

    t1 = threading.Thread(target=one, args=(a, b), daemon=True)
    t2 = threading.Thread(target=one, args=(b, a), daemon=True)
    t1.start()
    t2.start()
    t1.join(JOIN_S)
    t2.join(JOIN_S)


def python_https_with_ca(port: int, ca_pem: str, timeout_ms: int = TIMEOUT_MS) -> dict[str, object]:
    deadline = time.monotonic() + timeout_ms / 1000.0

    def remaining() -> float:
        left = deadline - time.monotonic()
        if left <= 0:
            raise TimeoutError("total response deadline exceeded")
        return left

    ctx = ssl.create_default_context(cadata=ca_pem)
    ctx.check_hostname = True
    ctx.verify_mode = ssl.CERT_REQUIRED
    raw = socket.create_connection(("localhost", port), timeout=remaining())
    try:
        raw.settimeout(remaining())
        tls = ctx.wrap_socket(raw, server_hostname="localhost")
        try:
            tls.settimeout(remaining())
            tls.sendall(b"GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
            data = b""
            while True:
                tls.settimeout(remaining())
                chunk = tls.recv(8192)
                if not chunk:
                    break
                data += chunk
                if len(data) > 1_048_576:
                    raise TimeoutError("response exceeded bound")
        finally:
            close_sock(tls)
    finally:
        close_sock(raw)
    return finish_http_response(data)


def decode_chunked_body(body: bytes) -> bytes:
    out = b""
    i = 0
    n = len(body)
    while True:
        nl = body.find(b"\r\n", i)
        if nl < 0:
            raise TimeoutError("incomplete chunk size")
        size_line = body[i:nl].split(b";", 1)[0].strip()
        try:
            size = int(size_line, 16)
        except ValueError as e:
            raise TimeoutError("invalid chunk size") from e
        i = nl + 2
        if size == 0:
            if i + 2 <= n and body[i:i + 2] == b"\r\n":
                return out
            end = body.find(b"\r\n\r\n", i)
            if end < 0:
                raise TimeoutError("incomplete chunked trailer")
            return out
        if n < i + size + 2:
            raise TimeoutError("incomplete chunk data")
        out += body[i:i + size]
        if body[i + size:i + size + 2] != b"\r\n":
            raise TimeoutError("incomplete chunk CRLF")
        i = i + size + 2


def finish_http_response(data: bytes) -> dict[str, object]:
    if b"\r\n\r\n" not in data:
        raise TimeoutError("incomplete HTTP framing")
    head, body = data.split(b"\r\n\r\n", 1)
    status = int(head.decode("latin1").split(" ")[1])
    headers: dict[str, str] = {}
    for line in head.decode("latin1").split("\r\n")[1:]:
        if ":" in line:
            k, v = line.split(":", 1)
            headers[k.strip().lower()] = v.strip()
    te = headers.get("transfer-encoding", "").strip().lower()
    if te:
        if te != "chunked":
            raise TimeoutError(f"unsupported transfer-encoding {te}")
        body = decode_chunked_body(body)
    elif "content-length" in headers:
        need = int(headers["content-length"])
        if len(body) < need:
            raise TimeoutError(f"incomplete body {len(body)}/{need}")
        body = body[:need]
    return {"status": status, "body": body.decode("utf-8", "replace"), "complete": True}


def _delete_tmp(tmp: str) -> None:
    if not os.path.isdir(tmp):
        return
    for name in os.listdir(tmp):
        if name.endswith(".key"):
            try:
                os.remove(os.path.join(tmp, name))
            except OSError:
                pass
    shutil.rmtree(tmp, ignore_errors=True)


def _readback(ports: list[int]) -> dict[str, object]:
    immediate = leftover_listen(ports)
    time.sleep(0.7)
    after = leftover_listen(ports)
    return {"immediate": immediate, "after700ms": after, "leftoverListen": bool(immediate or after)}


def _live_names(threads: list[threading.Thread]) -> list[str]:
    return [th.name for th in threads if th is not None and th.is_alive()]


def _snapshot(ports: list[int], threads: list[threading.Thread]) -> dict[str, object]:
    return {"listen": leftover_listen(ports), "threads": _live_names(threads)}


def run_public_fault(kind: str) -> int:
    isolate_env()
    tmp = tempfile.mkdtemp(prefix="pw-pt-public-fault-")
    http_owned = None
    https_owned = None
    proxy = None
    idle_client = None
    operation_error = None
    ports: list[int] = []
    try:
        if kind == "tls":
            certs = make_certs(tmp)
            try:
                https_owned = start_https_origin(certs, fault_tls=True)
            except Exception as e:  # noqa: BLE001
                operation_error = redact(e)
        elif kind == "http-start":
            try:
                http_owned = start_http_origin(fault_start=True)
            except Exception as e:  # noqa: BLE001
                operation_error = redact(e)
        elif kind == "proxy-start":
            real = threading.Thread.start

            def boom(th_self: threading.Thread, *a: object, **k: object) -> None:
                threading.Thread.start = real
                raise RuntimeError("inject-thread-start")

            threading.Thread.start = boom  # type: ignore[method-assign]
            try:
                proxy = ConnectProxy()
            except Exception as e:  # noqa: BLE001
                operation_error = redact(e)
            finally:
                threading.Thread.start = real
        elif kind == "shutdown":
            http_owned = start_http_origin()
            certs = make_certs(tmp)
            https_owned = start_https_origin(certs)
            ports = [http_owned.port, https_owned.port]
        elif kind == "serverclose":
            http_owned = start_http_origin()
            ports = [http_owned.port]

            def boom_close(*_a: object, **_k: object) -> None:
                raise RuntimeError("inject-public-server-close")

            http_owned.server.server_close = boom_close  # type: ignore[method-assign]
        elif kind == "idle-origin":
            http_owned = start_http_origin()
            ports = [http_owned.port]
            idle_client = socket.create_connection(("127.0.0.1", http_owned.port), timeout=2)
            time.sleep(0.2)
        elif kind == "idle":
            proxy = ConnectProxy()
            idle_client = socket.create_connection(("127.0.0.1", proxy.port), timeout=2)
            proxy.clients.append(idle_client)
            ports = [proxy.port]
            time.sleep(0.2)
        elif kind == "tls-idle":
            certs = make_certs(tmp)
            https_owned = start_https_origin(certs)
            ports = [https_owned.port]
            idle_client = socket.create_connection(("127.0.0.1", https_owned.port), timeout=2)
            time.sleep(0.3)
        elif kind == "stop":
            http_owned = start_http_origin()
            ports = [http_owned.port]

            def boom_stop(*_a: object, **_k: object) -> None:
                raise RuntimeError("inject-stop-set")

            http_owned.stop.set = boom_stop  # type: ignore[method-assign]
        elif kind == "handler-http":
            http_owned = start_http_origin()
            http_owned.server.handler_start_fault = True
            ports = [http_owned.port]
            idle_client = socket.create_connection(("127.0.0.1", http_owned.port), timeout=2)
            try:
                idle_client.sendall(b"GET / HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n")
            except OSError:
                pass
            time.sleep(0.3)
        elif kind == "handler-proxy":
            proxy = ConnectProxy(fault_handler=True)
            ports = [proxy.port]
            idle_client = socket.create_connection(("127.0.0.1", proxy.port), timeout=2)
            try:
                idle_client.sendall(b"GET http://127.0.0.1/ HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n")
            except OSError:
                pass
            time.sleep(0.3)
        else:
            raise RuntimeError("unknown public fault")
    except Exception as e:  # noqa: BLE001
        operation_error = operation_error or redact(e)
    if http_owned is not None:
        ports.append(http_owned.port)
    if https_owned is not None:
        ports.append(https_owned.port)
    if proxy is not None:
        ports.append(proxy.port)
    ports = list(dict.fromkeys(ports))
    cleanup_errors: list[str] = []
    if http_owned is not None:
        cleanup_errors.extend(http_owned.close())
    if https_owned is not None:
        cleanup_errors.extend(https_owned.close())
    if proxy is not None:
        cleanup_errors.extend(proxy.close())
    close_sock(idle_client)
    _delete_tmp(tmp)
    threads: list[threading.Thread] = []
    if http_owned is not None:
        threads.extend(http_owned.live_threads())
    if https_owned is not None:
        threads.extend(https_owned.live_threads())
    if proxy is not None:
        threads.append(proxy.accept_thread)
        threads.extend(proxy.threads)
    immediate = _snapshot(ports, threads)
    time.sleep(0.7)
    after = _snapshot(ports, threads)
    leftover = bool(immediate["listen"] or after["listen"] or immediate["threads"] or after["threads"])
    alive_errors = [e for e in cleanup_errors if "alive" in e]
    if operation_error is None:
        for owned in (http_owned, https_owned):
            if owned is None:
                continue
            err = getattr(owned.server, "last_start_error", None)
            if err:
                operation_error = redact(err)
                break
        if operation_error is None and proxy is not None and getattr(proxy, "last_start_error", None):
            operation_error = redact(proxy.last_start_error)
    if operation_error is None:
        for e in cleanup_errors:
            if str(e).startswith("inject-"):
                operation_error = str(e)
                break
    out = {
        "fault": kind,
        "operationError": operation_error,
        "cleanupErrors": cleanup_errors,
        "immediate": immediate,
        "after700ms": after,
        "leftoverListen": bool(immediate["listen"] or after["listen"]),
        "cleaned": not leftover and not alive_errors,
    }
    print(json.dumps(out, indent=2, ensure_ascii=False))
    return 0 if out["cleaned"] else 1


def main() -> int:
    fault = os.environ.get("PLAYWRIGHT_PT_PUBLIC_FAULT", "")
    if fault:
        return run_public_fault(fault)
    isolate_env()
    tmp = tempfile.mkdtemp(prefix="pw-pt-public-lab-")
    http_owned = None
    https_owned = None
    proxy = None
    browser = None
    result: dict[str, object] = {"ok": False, "ignore_https_errors": False, "cases": []}
    try:
        from playwright.sync_api import sync_playwright

        certs = make_certs(tmp)
        http_owned = start_http_origin()
        https_owned = start_https_origin(certs)
        proxy = ConnectProxy()
        http_port = http_owned.port
        https_port = https_owned.port
        with sync_playwright() as p:
            browser = p.chromium.launch(
                proxy={
                    "server": f"http://127.0.0.1:{proxy.port}",
                    "username": USER,
                    "password": PASSWORD,
                },
                timeout=30000,
            )
            context = browser.new_context(ignore_https_errors=False)
            page = context.new_page()
            page.set_default_timeout(TIMEOUT_MS)
            http_res = page.goto(f"http://127.0.0.1:{http_port}/")
            http_body = page.text_content("body")
            result["cases"].append(  # type: ignore[union-attr]
                {
                    "name": "browser-http-via-proxy",
                    "status": http_res.status if http_res else None,
                    "body": http_body,
                    "proxyEvents": [
                        {"method": e["method"], "target": e["target"]}
                        for e in proxy.events
                        if e["method"] == "GET"
                    ],
                }
            )
            tls_err = None
            try:
                page.goto(f"https://127.0.0.1:{https_port}/")
            except Exception as e:  # noqa: BLE001
                tls_err = {"name": type(e).__name__, "message": redact(e)}
            result["cases"].append(  # type: ignore[union-attr]
                {
                    "name": "browser-https-untrusted-ca",
                    "error": tls_err,
                    "connectEvents": [e for e in proxy.events if e["method"] == "CONNECT"],
                }
            )
            trusted = python_https_with_ca(https_port, certs["caPem"])
            result["cases"].append({"name": "python-https-with-lab-ca", **trusted})  # type: ignore[arg-type]
            context.close()
            try:
                browser.close()
            except Exception:
                pass
            browser = None
        cases = result["cases"]  # type: ignore[assignment]
        result["ok"] = (
            cases[0]["status"] == 200
            and tls_err is not None
            and trusted["status"] == 200
            and len(cases[1]["connectEvents"]) >= 1
        )
        result["httpOrigin"] = http_port
        result["httpsOrigin"] = https_port
        result["proxyPort"] = proxy.port
    except Exception as e:  # noqa: BLE001
        print(redact(e), file=sys.stderr)
        result["error"] = redact(e)
        result["ok"] = False
    finally:
        if browser is not None:
            try:
                browser.close()
            except Exception:
                pass
        cleanup_errors: list[str] = []
        if http_owned is not None:
            cleanup_errors.extend(http_owned.close())
        if https_owned is not None:
            cleanup_errors.extend(https_owned.close())
        if proxy is not None:
            cleanup_errors.extend(proxy.close())
        _delete_tmp(tmp)
        if cleanup_errors:
            result["cleanupErrors"] = cleanup_errors
    print(json.dumps(result, indent=2, ensure_ascii=False))
    if result.get("ok") and not result.get("cleanupErrors"):
        return 0
    return 1


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

Rode com o Playwright 1.64.0 já instalado e com o Chromium empacotado. page.set_default_timeout(8000) limita o goto. Um proxy que aceita o TCP e depois cala não pode ficar aberto até o default de 30 s do launch.

O sidecar também aceita falhas injetadas. Carga de certificado inexistente, Thread.start recusado no accept e no handler, server_close que lança, cliente ocioso no worker, TCP ocioso antes do handshake TLS. O loop customizado não chama HTTPServer.shutdown. Em cada falha o processo ainda tenta fechar sockets, threads, listeners e handles de filho. A leitura imediata e a leitura depois de 700 ms mostram o que restou. Operação que falhou e limpeza que passou ou falhou de forma explícita são linhas distintas.

Python ssl com a CA do laboratório contra um destino silencioso ou que pinga um byte por segundo estoura TimeoutError no prazo total de 8000 ms. Não devolve um 200 truncado. Isso importa quando o proxy CONNECT abre e o TLS da origem não termina.

A consulta playwright proxy python aqui é o binding do Playwright, não um cliente HTTP de biblioteca. Só o Playwright fala launch e new_context. O que é um proxy de rede continua valendo, mas a API muda.

05SOCKS5 só com suporte comprovado do motor#

A página de rede cita SOCKSv5. A opção server aceita socks5://host:porta. O laboratório subiu um SOCKS5 local sem autenticação. Chromium 156.0.8078.4 lançou, fez CONNECT no sentido SOCKS e a origem HTTP local respondeu 200. Isso é SOCKS5 sem senha neste motor.

SOCKS5 com usuário e senha é outra história. A issue aberta microsoft/playwright#10567 segue em P3-collecting-feedback. Em 28 de novembro de 2021, Max Schmitt escreveu que Chromium e Firefox não suportam userpass SOCKS5, e que o Playwright depende desses motores. Pavel Feldman, no dia seguinte, confirmou que SOCKS5 com auth não é suportado pelos browsers principais. Em 14 de outubro de 2026 a issue continua aberta. O laboratório tentou socks5:// com username e password em Chromium, Firefox e WebKit. Os três recusaram o launch. browser.version nesses motores ficou vazio porque o processo não chegou a subir. As versões de instalador baixadas neste host foram Firefox 157.0 e WebKit 27.2.

Não invente um campo socksUsername. Não passe userpass na URL socks5://user:pass@host e chame isso de API do Playwright. Se o fornecedor só entrega SOCKS5 com senha, o caminho documentado hoje é um relé local sem senha na frente, ou um proxy HTTP com Basic, que este Chromium já prova. Um wrapper não é o Playwright.

O SOCKS5 versus HTTP explica o transporte. Protocolo não é origem. Um IP residencial ou móvel pode chegar como HTTP. Um datacenter também. Trocar o esquema na string do Playwright não troca o ASN.

06Isolar dois contexts#

Dois contexts no mesmo browser não compartilham cookie. O laboratório abriu um context pelo proxy aberto e outro pelo segundo forwarder. Os dois pediram a mesma origem HTTP. Cada um recebeu um Set-Cookie com token diferente. context.cookies() dos dois lados não coincidiu. Fechar um context não derruba o outro.

Isso é o isolamento que a documentação de browser contexts descreve. Serve quando um teste precisa de duas sessões. Não serve como prova de que o fornecedor gira IP a cada context. Girar IP é configuração do produto, fora desta API.

bypass é uma lista separada por vírgula. Com bypass: "127.0.0.1" o Chromium foi direto à origem local e o proxy aberto ficou em zero eventos. Com bypass: "example.com" a origem 127.0.0.1 ainda passou pelo proxy. Bypass não é um interruptor universal. É uma lista. Host que não está nela continua proxied.

Feche o context que você criou. Depois feche o browser. O laboratório registra browsers e contexts e chama close no shutdown. Sobra inesperada de listener, thread, filho ou chave .key imediatamente e depois de 700 ms falha a limpeza. Recusa persistente e intencional de Process.join ou Process.close é um controle negativo: a limpeza fica FAILED, o sentinel continua sob posse do laboratório e o erro fica no recibo. Isso não é fechamento de todos os handles. O harness deste percurso tem 107 registros de afirmação, 103 nomes distintos e 51 capturas, não 107 limpezas distintas. Isso é higiene do processo de teste, não um recurso do Playwright.

07Diagnosticar 407, recusa e certificados sem desligar TLS#

Leia a falha pelo que ela é. A tabela abaixo saiu deste loopback, não de um fornecedor pago.

Porta recusada é imediata. Proxy que engole o TCP e não responde é o caso do timeout limitado. Os dois são feios no log. Só o segundo gasta os 8 s. Não misture com 407. 407 é o proxy falando. Recusa é o kernel. Timeout é o relógio do Playwright.

Antes de culpar o binding, rode o host no checador e olhe a saída no meu IP. Confira também como checar se o proxy funciona. O Playwright só entra depois que o host aceita o protocolo que você passou em server.

No catálogo em português, as páginas de SOAX, Bright Data, Oxylabs e Smartproxy responderam 200 neste dia, em pt-BR, e as fichas falam HTTP, HTTPS e SOCKS5. A ficha da Bright Data menciona Playwright. SOCKS5 com usuário não é encaixe do Playwright 1.64.0. Isso não é teste desses fornecedores. A ficha da NetNut em português não abriu.

SOAX: proxies residenciais e móveis3.9219

A SOAX vende proxies residenciais e móveis rotativos num pool compartilhado de créditos, com datacenter e ISP nos mesmos planos. O Builder começa em $200/mês + IVA (200 créditos); o tráfego Tier 1 é $3/GB no Builder.

85 países

US$ 3,00

por 1 GB

Bright Data: proxies residenciais e de datacenter4.41438

A Bright Data é uma grande rede de proxies para coleta de dados: IPs residenciais, móveis, ISP e de datacenter em cerca de 195 países. O tráfego residencial pode ser pago por uso, sem assinatura, e os endereços de datacenter e ISP são vendidos em pacotes a partir de 10 IPs. O acesso residencial e móvel exige verificação de identidade.

10 países

US$ 5,00

por 1 GB

Oxylabs: proxies residenciais e de datacenter4.0770

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

10 países

US$ 0,59

por 1 GB

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

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

83 países

US$ 0,38

por 1 GB