Function connect_http_proxy

src/websockets/sync/client.py:514–549

(
    proxy: Proxy,
    ws_uri: WebSocketURI,
    deadline: Deadline,
    *,
    user_agent_header: str | None = None,
    ssl: ssl_module.SSLContext | None = None,
    server_hostname: str | None = None,
    **kwargs: Any,
)

def connect_http_proxy(
    proxy: Proxy,
    ws_uri: WebSocketURI,
    deadline: Deadline,
    *,
    user_agent_header: str | None = None,
    ssl: ssl_module.SSLContext | None = None,
    server_hostname: str | None = None,
    **kwargs: Any,
) -> socket.socket:
    # Connect socket

    kwargs.setdefault("timeout", deadline.timeout())
    sock = socket.create_connection((proxy.host, proxy.port), **kwargs)

    # Initialize TLS wrapper and perform TLS handshake

    if proxy.scheme == "https":
        if ssl is None:
            ssl = ssl_module.create_default_context()
        if server_hostname is None:
            server_hostname = proxy.host
        sock.settimeout(deadline.timeout())
        sock = ssl.wrap_socket(sock, server_hostname=server_hostname)
        sock.settimeout(None)

    # Send CONNECT request to the proxy and read response.

    sock.sendall(prepare_connect_request(proxy, ws_uri, user_agent_header))
    try:
        read_connect_response(sock, deadline)
    except Exception:
        sock.close()
        raise

    return sock


T = TypeVar("T")