Proxy in Playwright konfigurieren und prüfen

Proxy in Playwright Python am Browser oder Context setzen, CONNECT und 407 prüfen, SOCKS5 nur mit Engine-Support und TLS anlassen.

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

01Proxy am Browser oder am Context setzen#

Die Python-Doku zu HTTP-Proxy vom 14. Oktober 2026 sagt klar: der Proxy gilt entweder für den ganzen Browser oder für einen einzelnen Context. Dieselbe Form steht an browser_type.launch und an browser.new_context. Felder sind server, optional bypass, optional username, optional password. Das Labor hat Playwright for Python 1.64.0 gegen Chromium 156.0.8078.4 auf Python 3.14.6 ausgeführt. Node war kein zweites Labor.

Setzen Sie den Proxy auf chromium.launch, wenn jeder Context desselben Prozesses denselben Ausgang nutzen soll. Setzen Sie ihn auf browser.new_context, wenn ein Prozess zwei Ausgänge oder zwei Cookie-Gläser braucht.

1. Entscheiden Sie, ob der Lauf einen Exit oder zwei Identitäten braucht.
2. Für einen Exit: proxy an launch übergeben.
3. Für zwei Exits: Browser ohne zweite Identität starten, dann je Context ein anderes proxy.
4. Jeden selbst erzeugten Context schließen, bevor browser.close() kommt. Das steht in der Browser-API als Aufräumhinweis.

Ohne proxy traf Chromium den lokalen HTTP-Origin mit Status 200. Der offene Loopback-Proxy zählte null Ereignisse. Vor dem Start hat das Labor HTTP_PROXY, HTTPS_PROXY, ALL_PROXY und NO_PROXY gelöscht. Das ist eine saubere Basis ohne Proxy-Option, kein Beweis, wie Playwright eine geerbte HTTP_PROXY-Variable behandeln würde.

Ein zweiter Browser startete mit dem offenen Proxy als Launch-Default. Ein Context bekam proxy.server auf einen anderen lokalen Forwarder. Der Request kam mit 200 zurück. Die Treffer lagen auf dem zweiten Forwarder. Der erste Forwarder sah die Origin-URL dieses Contexts nicht. Context-proxy ersetzt den Browser-Default für genau diesen Context. Es wird nicht gemischt.

Es gibt in 1.64.0 kein rotateSession an launch oder new_context. Wer IPs drehen will, macht das im Anbieterprodukt, nicht in dieser API.

02Host, Port und Anmeldung übergeben, ohne Geheimnisse zu loggen#

server ist eine URL. Die Launch-Doku nennt http://myproxy.com:3128 und socks5://myproxy.com:3128. Die Kurzform myproxy.com:3128 gilt als HTTP. Wenn der Anbieter SOCKS5 meint, muss das Schema socks5:// stehen. username und password sind für einen HTTP-Proxy mit Anmeldung dokumentiert, nicht als SOCKS5-Userpass.

Das Passwort gehört in die Umgebung oder den Secret-Store, nicht in ein committed Config-File. Playwright schickt trotzdem Proxy-Authorization auf der Leitung. Das Labor nutzte den Dummy-Benutzer lab-user und ein Dummy-Passwort mit @, :, /, # und %. In den Ergebnislogs steht ***.

Gegen den lokalen Basic-Proxy sah Chromium so aus:

1. Nur server, keine Userdaten. page.goto auf den HTTP-Origin lieferte Status 407. Der Proxy schrieb authReason: missing. Playwright warf nicht. Der Status am Response-Objekt war der Beweis.
2. Benutzer lab-user und ein falsches Passwort. Wieder Status 407. Der Log zeigte authReason: wrong.
3. Das Dummy-Passwort korrekt. Status 200. Der Log zeigte authorized: true.

Falsche Zugangsdaten sind in diesem Chromium-Build kein hängender Socket. Sie sind HTTP 407 vom Proxy, sichtbar als Navigationsstatus. Wenn ein Wrapper später wirft, behalten Sie 407 und die Exception-Klasse. Beides nicht zu "Proxy geht nicht" zusammenziehen.

HTTP Authentication auf derselben Network-Seite ist http_credentials für den Origin, nicht proxy.username. Origin-401 und Proxy-407 sind verschiedene Hops. Wer die beiden mischt, debuggt die falsche Schicht.

03HTTPS-Ziele über HTTP-Proxy (CONNECT) prüfen#

Das Schema der Ziel-URL und das Schema der Proxy-URL sind zwei Knöpfe. Ein Origin https://example.com/ geht trotzdem über einen HTTP-Proxy auf http://127.0.0.1:<port>. Die Proxy-Methode ist CONNECT, Ziel example.com:443. Das Labor hat genau dieses CONNECT gesehen. Danach hat Chromium TLS gegen das öffentliche Zertifikat geprüft und page.goto lieferte 200. Das ist HTTPS durch einen HTTP-Proxy mit Prüfung an.

Ein lokaler HTTPS-Origin mit Labor-CA ist die andere Hälfte. Chromium 156.0.8078.4, ignore_https_errors auf dem dokumentierten Default False, navigierte zu https://127.0.0.1:<https-port>/ durch denselben HTTP-Proxy. Der Proxy schrieb CONNECT 127.0.0.1:<https-port>. Danach warf Playwright Error mit net::ERR_CERT_AUTHORITY_INVALID. CONNECT ist passiert. TLS ist trotzdem gescheitert. Dass das Ziel https:// trägt, macht den Proxy nicht zum "https-Proxy". Dass der Proxy CONNECT annimmt, bringt Chromium nicht dazu, einer privaten CA zu trauen.

Python ssl gegen denselben Origin ohne CA endete in SSLCertVerificationError: unable to get local issuer certificate. Mit der Labor-CA und verify_mode=CERT_REQUIRED kam HTTP 200. Das ist der Trust-Pfad außerhalb des Browsers. client_certificates an new_context, ergänzt in v1.46, ist ein Client-Zertifikat für einen genannten Origin, keine CA-Datei. Der dokumentierte Playwright-Schalter, der den Fehler überspringen würde, ist ignore_https_errors=True. Dieser Text setzt ihn nicht. verify=False war kein Fall.

Wenn die Aufgabe "CONNECT beweisen" heißt, lesen Sie den Proxy-Log auf CONNECT host:port. Wenn die Aufgabe "Seite geladen" heißt, brauchen Sie ein Zertifikat, dem Chromium schon traut, oder eine CA außerhalb dieses Guides. Eine Checker-Seite im privaten Browser ist der Browser-Ausgang, nicht der Playwright-Weg. Meine IP hilft nur für den Browser-Exit. Der Proxy-Checker sagt, ob ein Host HTTP oder SOCKS5 annimmt, bevor Sie Playwright darauf zeigen. Beides ersetzt kein Playwright-Trace.

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

Das vollständige öffentliche Skript unten startet Origin, CONNECT-Proxy und Chromium lokal, schreibt JSON auf stdout und beendet mit Code 0 oder 1. Es braucht keine privaten Sidecars. Temporäre Schlüssel löscht es im finally.

04Python-Binding: Launch-Optionen und Context-Optionen#

Die Python-Beispiele in der Network-Doku nutzen chromium.launch(proxy={...}) und browser.new_context(proxy={...}). Sync und Async tragen dieselben Schlüssel. Dieses Labor führt das Sync-Binding aus.

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()

new_page() ohne eigenen Context erbt den Browser-Proxy. Ein eigener Context mit anderem server überschreibt, wie der zweite Forwarder im Labor gezeigt hat. Timeout setzen. Ohne Timeout kann goto an einem stummen Proxy hängen. Im Labor hat ein Listener, der TCP annahm und nie antwortete, nach 8000 ms TimeoutError geliefert, gemessen 8061 ms. Das ist begrenzt, nicht endlos.

bypass ist eine Kommaliste. Die API-Beispiele nennen Hosts wie .com oder chromium.org. Es ist kein Universalschalter.

Das folgende Skript ist dasselbe Byte für Byte wie tests/public_lab.py. Es läuft allein.

#!/usr/bin/env python3
"""Eigenstaendiges Loopback-Labor fuer Playwright Python.

Dummy-Zugangsdaten. Lokaler Origin und CONNECT-Proxy. Kein bezahlter Anbieter.
TLS bleibt geprueft. Dieses Skript setzt ignore_https_errors nie auf True.

    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[Origin], track_name: str = "public-http"):
        super().__init__(addr, handler)
        self.track_name = track_name
        self.tls_ctx: ssl.SSLContext | None = None
        self.handshake_timeout = 1.0
        self.accepted: list[socket.socket] = []
        self.handlers: list[threading.Thread] = []
        self.daemon_threads = True

    def get_request(self) -> tuple[socket.socket, object]:
        req, addr = super().get_request()
        self.accepted.append(req)
        if self.tls_ctx is None:
            return req, addr
        try:
            req.settimeout(self.handshake_timeout)
            tls_sock = self.tls_ctx.wrap_socket(req, server_side=True)
            self.accepted.append(tls_sock)
            return tls_sock, addr
        except Exception:
            close_sock(req)
            raise

    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",
        )
        self.handlers.append(t)
        t.start()


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] = []
        self.baseServerShutdownInvoked = False

    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):
            close_sock(s)
        try:
            self.server.server_close()
        except Exception as e:  # noqa: BLE001
            errors.append(str(e))
        close_sock(getattr(self.server, "socket", None))
        for t in list(self.server.handlers):
            t.join(JOIN_S)
            if t.is_alive():
                errors.append(f"handler-alive:{t.name}")
        if self.started and self.thread is not None:
            self.thread.join(JOIN_S)
            if self.thread.is_alive():
                errors.append(f"thread-alive:{self.thread.name}")
        for t in list(self.helpers):
            t.join(JOIN_S)
            if t.is_alive():
                errors.append(f"helper-alive:{t.name}")
        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-de-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-de-public.pem", "/nonexistent-pw-de-public.key")
        else:
            ctx.load_cert_chain(certs["originPem"], certs["originKey"])
        server.tls_ctx = ctx
        server.handshake_timeout = 1.0
        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) -> None:
        self.events: list[dict[str, object]] = []
        self.clients: list[socket.socket] = []
        self.threads: list[threading.Thread] = []
        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")
            self.threads.append(t)
            t.start()

    def close(self) -> list[str]:
        errors: list[str] = []
        self.stop.set()
        close_sock(self.sock)
        for client in list(self.clients):
            close_sock(client)
        if self.accept_started:
            self.accept_thread.join(JOIN_S)
            if self.accept_thread.is_alive():
                errors.append("accept-thread-alive")
        for t in self.threads:
            t.join(JOIN_S)
            if t.is_alive():
                errors.append(f"handler-alive:{t.name}")
        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)
    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()
    if "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 _live_names(threads: list[threading.Thread]) -> list[str]:
    return [t.name for t in threads if t is not None and t.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-de-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-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-tls-prehandshake":
            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-set":
            http_owned = start_http_origin()
            certs = make_certs(tmp)
            https_owned = start_https_origin(certs)
            ports = [http_owned.port, https_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 == "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)
        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] = []
    for owned in (http_owned, https_owned):
        if owned is None:
            continue
        try:
            cleanup_errors.extend(owned.close())
        except Exception as e:  # noqa: BLE001
            cleanup_errors.append(str(e))
    if proxy is not None:
        try:
            cleanup_errors.extend(proxy.close())
        except Exception as e:  # noqa: BLE001
            cleanup_errors.append(str(e))
    close_sock(idle_client)
    try:
        _delete_tmp(tmp)
    except Exception as e:  # noqa: BLE001
        cleanup_errors.append(str(e))
    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 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_DE_PUBLIC_FAULT", "")
    if fault:
        return run_public_fault(fault)
    isolate_env()
    tmp = tempfile.mkdtemp(prefix="pw-de-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] = []
        for owned in (http_owned, https_owned):
            if owned is None:
                continue
            try:
                cleanup_errors.extend(owned.close())
            except Exception as e:  # noqa: BLE001
                cleanup_errors.append(str(e))
        if proxy is not None:
            try:
                cleanup_errors.extend(proxy.close())
            except Exception as e:  # noqa: BLE001
                cleanup_errors.append(str(e))
        try:
            _delete_tmp(tmp)
        except Exception as e:  # noqa: BLE001
            cleanup_errors.append(str(e))
        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())

05SOCKS5 nur mit bewiesener Engine-Unterstützung#

Die Network-Doku sagt, Seiten können über HTTP-Proxy oder SOCKSv5 laden. Chromium 156.0.8078.4 mit server: "socks5://127.0.0.1:<port>" ohne Benutzer holte den HTTP-Origin mit Status 200. Die SOCKS-Fixture schrieb CONNECT 127.0.0.1:<origin-port>. Unauthentifiziertes SOCKS5 ist in diesem Pin für Chromium im Walkthrough.

Authentifiziertes SOCKS5 ist es nicht. chromium.launch, firefox.launch und webkit.launch in Playwright 1.64.0 warfen alle Error: Browser does not support socks5 proxy authentication. Der Wurf kam am Launch, vor page.goto. Playwright 1.64.0 hatte Firefox 157.0 und WebKit 27.2 für diese Starts geladen. Die Browser-version() blieb leer, weil der Launch scheiterte. Die Meldung trifft die offene Produktanfrage microsoft/playwright#10567, am 14. Oktober 2026 weiterhin Open mit Label P3-collecting-feedback. Ein Maintainer-Kommentar vom 28. November 2021 hält fest, dass Chromium und Firefox SOCKS5-Userpass in den Engines selbst nicht können.

Wenn ein Anbieter nur SOCKS5 mit Benutzer anbietet, schickt dieser Playwright-Pin den Benutzer nicht. Nutzen Sie einen HTTP-Proxy mit username und password, oder ein unauthentifiziertes SOCKS5, das die Engine in einem Lauf wie oben wirklich gewählt hat. Der Produktvergleich steht in SOCKS5 vs HTTP-Proxys.

06Zwei Contexts isolieren#

browser.new_context() erzeugt ein Glas, das Cookies nicht mit anderen Contexts teilt. Kombinieren Sie das mit per-Context-proxy, wenn zwei Läufe eine Session nicht auf den falschen Exit legen dürfen.

1. Chromium ohne Browser-Proxy starten.
2. Context A mit Proxy A, Context B mit Proxy B.
3. In beiden page.goto auf denselben Origin.
4. Cookies vergleichen.

Im Labor kamen beide Navigationen mit 200. Die Cookie-Werte lab= unterschieden sich. context.cookies() der beiden Gläser war nicht identisch. Der Origin hatte jedem Context ein eigenes Token gegeben. Ein gemeinsames Glas hätte dasselbe Token zurückgegeben.

Isolation ist keine Magie gegen Leak in der App. Sie trennt Playwright-State. Wenn die getestete Seite die Session in LocalStorage an eine feste ID bindet, die Sie selbst setzen, hilft der zweite Context nicht. Messen Sie die Gläser, nicht die Absicht.

Schließen Sie beide Contexts explizit. browser.close() reißt sonst mit und spült HAR oder Video nicht sauber.

07Verbindungsfehler, 407 und Zertifikate ohne TLS-Abschaltung#

Die Fehler sehen ähnlich aus, bis Sie Klasse, Status und Proxy-Log trennen. Ein abgelehnter Port http://127.0.0.1:1 lieferte net::ERR_PROXY_CONNECTION_FAILED. Der stumme Listener lieferte TimeoutError nach 8000 ms. Fehlende und falsche Basic-Auth lieferten 407 als Status, ohne Throw. Die untrusted Labor-CA lieferte ERR_CERT_AUTHORITY_INVALID nach CONNECT. Python ohne CA lieferte SSLCertVerificationError. Das sind fünf verschiedene Schichten.

bypass: "127.0.0.1" übersprang den Proxy, Status 200, Proxy-Log leer. bypass: "example.com" bei Ziel 127.0.0.1 ging weiter durch den Proxy. Bypass ist die dokumentierte Hostliste, kein allgemeines Entkommen.

Wenn der Dienst hängt, grenzt Proxy funktioniert nicht die Schicht ein, bevor Sie den Client umschreiben. Für die längere Checkliste siehe So prüfen Sie, ob ein Proxy funktioniert. Residential- und Mobile-Exits unterscheiden sich in der Herkunft, nicht im proxy-Dict. Die Katalogseiten Residential-Proxys, Mobile Proxys und Datacenter filtern den Typ. Playwright bleibt dieselbe Bibliothek.

SOAX: Residential- und Mobile-Proxys3.9219

SOAX verkauft rotierende Residential- und Mobile-Proxys über einen gemeinsamen Credit-Pool, mit Datacenter und ISP in denselben Plänen. Der Builder startet bei $200/mo + VAT (200 Credits); Tier-1-Traffic kostet $3/GB im Builder.

85 Länder

3,00 $

pro 1 GB

Bright Data: Residential- und Datacenter-Proxys4.41438

Bright Data ist ein großes Proxy-Netz für Datenerhebung: Residential-, Mobile-, ISP- und Datacenter-IPs in rund 195 Ländern. Residential-Traffic lässt sich ohne Abo nach Verbrauch bezahlen, Datacenter- und ISP-Adressen gibt es in Paketen ab 10 IPs. Für Residential und Mobile ist eine Identitätsprüfung nötig.

10 Länder

5,00 $

pro 1 GB

Oxylabs: Residential- und Datacenter-Proxys4.0770

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

10 Länder

0,59 $

pro 1 GB

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

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

83 Länder

0,38 $

pro 1 GB

Die Karten unten sind DE-Katalogeinträge. Für Playwright 1.64.0 brauchen Sie einen HTTP-Forward-Proxy mit CONNECT für HTTPS-Ziele und Login in username/password, oder SOCKS5 ohne Benutzer. SOCKS5 mit Benutzer passt in diesem Pin nicht, auch wenn ein Anbieter SOCKS5 nennt. Ob der gewählte Endpunkt das so spricht, steht in der aktuellen Anbieterdoku, nicht in diesem Loopback-Lauf.