(
proxy: Proxy,
ws_uri: WebSocketURI,
user_agent_header: str | None = USER_AGENT,
)
| 132 | |
| 133 | |
| 134 | def prepare_connect_request( |
| 135 | proxy: Proxy, |
| 136 | ws_uri: WebSocketURI, |
| 137 | user_agent_header: str | None = USER_AGENT, |
| 138 | ) -> bytes: |
| 139 | host = build_host(ws_uri.host, ws_uri.port, ws_uri.secure, always_include_port=True) |
| 140 | headers = Headers() |
| 141 | headers["Host"] = build_host(ws_uri.host, ws_uri.port, ws_uri.secure) |
| 142 | if user_agent_header is not None: |
| 143 | headers["User-Agent"] = user_agent_header |
| 144 | if proxy.username is not None: |
| 145 | assert proxy.password is not None # enforced by parse_proxy() |
| 146 | headers["Proxy-Authorization"] = build_authorization_basic( |
| 147 | proxy.username, proxy.password |
| 148 | ) |
| 149 | # We cannot use the Request class because it supports only GET requests. |
| 150 | return f"CONNECT {host} HTTP/1.1\r\n".encode() + headers.serialize() |
searching dependent graphs…