Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/websocket/_http.py: 34%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""
2_http.py
3websocket - WebSocket client library for Python
5Copyright 2026 engn33r
7Licensed under the Apache License, Version 2.0 (the "License");
8you may not use this file except in compliance with the License.
9You may obtain a copy of the License at
11 http://www.apache.org/licenses/LICENSE-2.0
13Unless required by applicable law or agreed to in writing, software
14distributed under the License is distributed on an "AS IS" BASIS,
15WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16See the License for the specific language governing permissions and
17limitations under the License.
18"""
20import errno
21import os
22import socket
23from base64 import encodebytes as base64encode
24from typing import Any, Optional, Tuple
26from ._exceptions import (
27 WebSocketAddressException,
28 WebSocketException,
29 WebSocketProxyException,
30)
31from ._logging import debug, dump, trace
32from ._socket import DEFAULT_SOCKET_OPTION, recv_line, send
33from ._ssl_compat import HAVE_SSL, ssl
34from ._url import get_proxy_info, parse_url
36__all__ = ["proxy_info", "connect", "read_headers"]
38# Import python_socks if available, otherwise define fallback classes
39try:
40 from python_socks._errors import ProxyConnectionError, ProxyError, ProxyTimeoutError
41 from python_socks._types import ProxyType
42 from python_socks.sync import Proxy
44 HAVE_PYTHON_SOCKS = True
45except ImportError:
46 HAVE_PYTHON_SOCKS = False
48 class ProxyError(Exception): # type: ignore[no-redef]
49 pass
51 class ProxyTimeoutError(Exception): # type: ignore[no-redef]
52 pass
54 class ProxyConnectionError(Exception): # type: ignore[no-redef]
55 pass
57 class ProxyType: # type: ignore[no-redef]
58 pass
61class proxy_info:
62 def __init__(self, **options):
63 self.proxy_host = options.get("http_proxy_host", None)
64 if self.proxy_host:
65 self.proxy_port = options.get("http_proxy_port", 0)
66 self.auth = options.get("http_proxy_auth", None)
67 self.no_proxy = options.get("http_no_proxy", None)
68 self.proxy_protocol = options.get("proxy_type", "http")
69 # Note: If timeout not specified, default python-socks timeout is 60 seconds
70 self.proxy_timeout = options.get("http_proxy_timeout", None)
71 if self.proxy_protocol not in [
72 "http",
73 "socks4",
74 "socks4a",
75 "socks5",
76 "socks5h",
77 ]:
78 raise ProxyError(
79 "Only http, socks4, socks5 proxy protocols are supported"
80 )
81 else:
82 self.proxy_port = 0
83 self.auth = None
84 self.no_proxy = None
85 self.proxy_protocol = "http"
86 self.proxy_timeout = None
89def _start_proxied_socket(
90 url: str, options: Any, proxy: Any
91) -> Tuple[socket.socket, Tuple[str, int, str]]:
92 if proxy.proxy_host and not proxy.proxy_port:
93 raise WebSocketProxyException("Cannot use port 0 when proxy_host specified")
94 if not HAVE_PYTHON_SOCKS:
95 raise WebSocketException(
96 "Python Socks is needed for SOCKS proxying but is not available"
97 )
99 hostname, port, resource, is_secure = parse_url(url)
101 if proxy.proxy_protocol == "socks4":
102 rdns = False
103 proxy_type = ProxyType.SOCKS4
104 # socks4a sends DNS through proxy
105 elif proxy.proxy_protocol == "socks4a":
106 rdns = True
107 proxy_type = ProxyType.SOCKS4
108 elif proxy.proxy_protocol == "socks5":
109 rdns = False
110 proxy_type = ProxyType.SOCKS5
111 # socks5h sends DNS through proxy
112 elif proxy.proxy_protocol == "socks5h":
113 rdns = True
114 proxy_type = ProxyType.SOCKS5
116 ws_proxy = Proxy.create(
117 proxy_type=proxy_type,
118 host=proxy.proxy_host,
119 port=int(proxy.proxy_port),
120 username=proxy.auth[0] if proxy.auth else None,
121 password=proxy.auth[1] if proxy.auth else None,
122 rdns=rdns,
123 )
125 sock = ws_proxy.connect(hostname, port, timeout=proxy.proxy_timeout)
127 try:
128 if is_secure:
129 if HAVE_SSL:
130 sock = _ssl_socket(sock, options.sslopt, hostname)
131 else:
132 raise WebSocketException("SSL not available.")
133 except BaseException:
134 sock.close()
135 raise
137 return sock, (hostname, port, resource)
140def connect(
141 url: str, options: Any, proxy: Any, socket: Any
142) -> Tuple[socket.socket, Tuple[str, int, str]]:
143 # Use _start_proxied_socket() only for socks4 or socks5 proxy
144 # Use _tunnel() for http proxy
145 # TODO: Use python-socks for http protocol also, to standardize flow
146 if proxy.proxy_host and not socket and proxy.proxy_protocol != "http":
147 return _start_proxied_socket(url, options, proxy)
149 hostname, port_from_url, resource, is_secure = parse_url(url)
151 if socket:
152 # A wss:// URL always runs over TLS: wrap a caller-supplied plain socket
153 # instead of silently sending the handshake and data in cleartext.
154 if is_secure and not (HAVE_SSL and isinstance(socket, ssl.SSLSocket)):
155 if not HAVE_SSL:
156 raise WebSocketException("SSL not available.")
157 socket = _ssl_socket(socket, options.sslopt, hostname)
158 return socket, (hostname, port_from_url, resource)
160 addrinfo_list, need_tunnel, auth = _get_addrinfo_list(
161 hostname, port_from_url, is_secure, proxy
162 )
163 if not addrinfo_list:
164 raise WebSocketException(f"Host not found.: {hostname}:{port_from_url}")
166 sock = None
167 try:
168 sock = _open_socket(addrinfo_list, options.sockopt, options.timeout)
169 if need_tunnel:
170 sock = _tunnel(sock, hostname, port_from_url, auth)
172 if is_secure:
173 if HAVE_SSL:
174 sock = _ssl_socket(sock, options.sslopt, hostname)
175 else:
176 raise WebSocketException("SSL not available.")
178 return sock, (hostname, port_from_url, resource)
179 except:
180 if sock:
181 sock.close()
182 raise
185def _get_addrinfo_list(
186 hostname: str, port: int, is_secure: bool, proxy: Any
187) -> Tuple[list, bool, Any]:
188 try:
189 phost, pport, pauth = get_proxy_info(
190 hostname,
191 is_secure,
192 proxy.proxy_host,
193 proxy.proxy_port,
194 proxy.auth,
195 proxy.no_proxy,
196 )
197 except TypeError as e:
198 raise WebSocketAddressException(e)
199 try:
200 # when running on windows 10, getaddrinfo without socktype returns a socktype 0.
201 # This generates an error exception: `_on_error: exception Socket type must be stream or datagram, not 0`
202 # or `OSError: [Errno 22] Invalid argument` when creating socket. Force the socket type to SOCK_STREAM.
203 if not phost:
204 addrinfo_list = socket.getaddrinfo(
205 hostname, port, 0, socket.SOCK_STREAM, socket.SOL_TCP
206 )
207 return addrinfo_list, False, None
208 else:
209 pport = pport and pport or 80
210 # when running on windows 10, the getaddrinfo used above
211 # returns a socktype 0. This generates an error exception:
212 # _on_error: exception Socket type must be stream or datagram, not 0
213 # Force the socket type to SOCK_STREAM
214 addrinfo_list = socket.getaddrinfo(
215 phost, pport, 0, socket.SOCK_STREAM, socket.SOL_TCP
216 )
217 return addrinfo_list, True, pauth
218 except (socket.gaierror, TypeError) as e:
219 raise WebSocketAddressException(e)
222def _open_socket(addrinfo_list, sockopt, timeout):
223 err = None
224 for addrinfo in addrinfo_list:
225 family, socktype, proto = addrinfo[:3]
226 sock = socket.socket(family, socktype, proto)
227 sock.settimeout(timeout)
228 for opts in DEFAULT_SOCKET_OPTION:
229 sock.setsockopt(*opts)
230 for opts in sockopt:
231 sock.setsockopt(*opts)
233 address = addrinfo[4]
234 err = None
235 while not err:
236 try:
237 sock.connect(address)
238 except socket.error as error:
239 sock.close()
240 error.remote_ip = str(address[0]) # type: ignore[attr-defined]
241 eConnRefused = (
242 errno.ECONNREFUSED,
243 getattr(errno, "WSAECONNREFUSED", errno.ECONNREFUSED),
244 errno.ENETUNREACH,
245 )
246 if error.errno not in eConnRefused:
247 raise error
248 err = error
249 continue
250 else:
251 break
252 else:
253 continue
254 break
255 else:
256 if err:
257 raise err
259 return sock
262def _sslopt_check_hostname(sslopt: dict) -> Optional[bool]:
263 """sslopt["check_hostname"] as True, False or None (use the default).
264 Any other value is rejected, so a falsy placeholder such as 0 or ""
265 cannot silently turn hostname verification off."""
266 value = sslopt.get("check_hostname")
267 if value is None or isinstance(value, bool):
268 return value
269 raise WebSocketException(
270 f"sslopt check_hostname must be True, False or None, not {value!r}"
271 )
274def _wrap_sni_socket(
275 sock: socket.socket, sslopt: dict, hostname: str, check_hostname: bool
276) -> Any:
277 context = sslopt.get("context", None)
278 if not context:
279 context = ssl.SSLContext(sslopt.get("ssl_version", ssl.PROTOCOL_TLS_CLIENT))
280 # Non default context need to manually enable SSLKEYLOGFILE support by setting the keylog_filename attribute.
281 # For more details see also:
282 # * https://docs.python.org/3.8/library/ssl.html?highlight=sslkeylogfile#context-creation
283 # * https://docs.python.org/3.8/library/ssl.html?highlight=sslkeylogfile#ssl.SSLContext.keylog_filename
284 keylog_file = os.environ.get("SSLKEYLOGFILE")
285 if keylog_file is not None:
286 context.keylog_filename = keylog_file
288 if sslopt.get("cert_reqs", ssl.CERT_NONE) != ssl.CERT_NONE:
289 # An empty value ("" or False, e.g. an unset config entry) counts
290 # as unset, the same as in _ssl_socket.
291 cafile = sslopt.get("ca_certs") or None
292 capath = sslopt.get("ca_cert_path") or None
293 if cafile or capath:
294 try:
295 context.load_verify_locations(cafile=cafile, capath=capath)
296 except (FileNotFoundError, ssl.SSLError, ValueError) as e:
297 raise WebSocketException(f"SSL CA certificate loading failed: {e}")
298 elif hasattr(context, "load_default_certs"):
299 try:
300 context.load_default_certs(ssl.Purpose.SERVER_AUTH)
301 except ssl.SSLError as e:
302 raise WebSocketException(
303 f"SSL default certificate loading failed: {e}"
304 )
305 if sslopt.get("certfile", None):
306 try:
307 context.load_cert_chain(
308 sslopt["certfile"],
309 sslopt.get("keyfile", None),
310 sslopt.get("password", None),
311 )
312 except (FileNotFoundError, ValueError) as e:
313 raise WebSocketException(f"SSL client certificate loading failed: {e}")
314 except ssl.SSLError as e:
315 raise WebSocketException(f"SSL client certificate loading failed: {e}")
317 # Python 3.10 switch to PROTOCOL_TLS_CLIENT defaults to "cert_reqs = ssl.CERT_REQUIRED" and "check_hostname = True"
318 # If both disabled, set check_hostname before verify_mode
319 # see https://github.com/liris/websocket-client/commit/b96a2e8fa765753e82eea531adb19716b52ca3ca#commitcomment-10803153
320 verify_hostname = _sslopt_check_hostname(sslopt)
321 if (
322 sslopt.get("cert_reqs", ssl.CERT_NONE) == ssl.CERT_NONE
323 and not verify_hostname
324 ):
325 context.check_hostname = False
326 context.verify_mode = ssl.CERT_NONE
327 else:
328 cert_reqs = sslopt.get("cert_reqs", ssl.CERT_REQUIRED)
329 if verify_hostname is None:
330 # Only an explicit False turns hostname verification off.
331 verify_hostname = True
332 if verify_hostname and cert_reqs == ssl.CERT_NONE:
333 raise WebSocketException(
334 "SSL certificate verification configuration failed: "
335 "check_hostname cannot be enabled when cert_reqs is CERT_NONE"
336 )
337 context.check_hostname = verify_hostname
338 context.verify_mode = cert_reqs
340 if "ciphers" in sslopt:
341 try:
342 context.set_ciphers(sslopt["ciphers"])
343 except ssl.SSLError as e:
344 raise WebSocketException(f"SSL cipher configuration failed: {e}")
345 if "cert_chain" in sslopt:
346 try:
347 cert_chain = sslopt["cert_chain"]
348 if not isinstance(cert_chain, (tuple, list)) or len(cert_chain) != 3:
349 raise ValueError(
350 "cert_chain must be a tuple/list of (certfile, keyfile, password)"
351 )
352 certfile, keyfile, password = cert_chain
353 context.load_cert_chain(certfile, keyfile, password)
354 except ValueError:
355 raise
356 except (FileNotFoundError, ssl.SSLError) as e:
357 raise WebSocketException(
358 f"SSL client certificate configuration failed: {e}"
359 )
360 if "ecdh_curve" in sslopt:
361 try:
362 context.set_ecdh_curve(sslopt["ecdh_curve"])
363 except ValueError as e:
364 raise WebSocketException(f"SSL ECDH curve configuration failed: {e}")
366 return context.wrap_socket(
367 sock,
368 do_handshake_on_connect=sslopt.get("do_handshake_on_connect", True),
369 suppress_ragged_eofs=sslopt.get("suppress_ragged_eofs", True),
370 server_hostname=hostname,
371 )
374def _ssl_socket(sock: socket.socket, user_sslopt: dict, hostname: str) -> Any:
375 sslopt: dict = {"cert_reqs": ssl.CERT_REQUIRED}
376 sslopt.update(user_sslopt)
378 # WEBSOCKET_CLIENT_CA_BUNDLE is only a default. An explicit ca_certs or
379 # ca_cert_path pins the trusted CAs, and the variable must not add to it.
380 # It is read only when it would be used: certificates are verified and no
381 # custom context is given.
382 cert_path = os.environ.get("WEBSOCKET_CLIENT_CA_BUNDLE")
383 if (
384 cert_path
385 # An empty ca_certs/ca_cert_path ("" or False) is not a pin: treating it
386 # as one would skip the bundle and fall back to the system CAs.
387 and not user_sslopt.get("ca_certs")
388 and not user_sslopt.get("ca_cert_path")
389 and sslopt["cert_reqs"] != ssl.CERT_NONE
390 and not sslopt.get("context")
391 ):
392 if os.path.isfile(cert_path):
393 sslopt["ca_certs"] = cert_path
394 elif os.path.isdir(cert_path):
395 sslopt["ca_cert_path"] = cert_path
396 else:
397 # Falling back to the system CAs would silently widen the trust
398 # the variable was meant to restrict.
399 raise WebSocketException(
400 "WEBSOCKET_CLIENT_CA_BUNDLE is neither a file nor a directory: "
401 f"{cert_path!r}"
402 )
404 if sslopt.get("server_hostname", None):
405 hostname = sslopt["server_hostname"]
407 check_hostname = sslopt.get("check_hostname", True)
408 sock = _wrap_sni_socket(sock, sslopt, hostname, check_hostname)
410 return sock
413def _tunnel(sock: socket.socket, host: str, port: int, auth: Any) -> socket.socket:
414 debug("Connecting proxy...")
415 if ":" in host:
416 host = f"[{host}]"
417 connect_header = f"CONNECT {host}:{port} HTTP/1.1\r\n"
418 connect_header += f"Host: {host}:{port}\r\n"
420 # TODO: support digest auth.
421 if auth and auth[0]:
422 auth_str = auth[0]
423 if auth[1]:
424 auth_str += f":{auth[1]}"
425 encoded_str = base64encode(auth_str.encode()).strip().decode().replace("\n", "")
426 connect_header += f"Proxy-Authorization: Basic {encoded_str}\r\n"
427 connect_header += "\r\n"
428 dump("request header", connect_header)
430 send(sock, connect_header)
432 try:
433 status, _, _ = read_headers(sock)
434 except (socket.error, WebSocketException) as e:
435 raise WebSocketProxyException(str(e))
437 if status != 200:
438 raise WebSocketProxyException(f"failed CONNECT via proxy status: {status}")
440 return sock
443MAX_HEADER_BYTES = 65536
446def read_headers(sock: socket.socket) -> tuple:
447 status = None
448 status_message = None
449 headers: dict = {}
450 trace("--- response header ---")
452 total = 0
453 while True:
454 line = recv_line(sock)
455 total += len(line)
456 if total > MAX_HEADER_BYTES:
457 raise WebSocketException("Response headers too large")
458 try:
459 line = line.decode("utf-8").strip()
460 except UnicodeDecodeError as e:
461 raise WebSocketException(
462 f"Invalid header line, not valid UTF-8 at byte {e.start}: {line!r}"
463 )
464 if not line:
465 if not status:
466 raise WebSocketException(f"Invalid status line: {line!r}")
467 break
468 trace(line)
469 if not status:
470 status_info = line.split(" ", 2)
471 try:
472 status = int(status_info[1])
473 except (IndexError, ValueError):
474 raise WebSocketException(f"Invalid status line: {line!r}")
475 if len(status_info) > 2:
476 status_message = status_info[2]
477 else:
478 kv = line.split(":", 1)
479 if len(kv) != 2:
480 raise WebSocketException("Invalid header")
481 key, value = kv
482 if key.lower() == "set-cookie":
483 if "set-cookie" in headers:
484 existing = headers["set-cookie"]
485 if isinstance(existing, list):
486 existing.append(value.strip())
487 else:
488 headers["set-cookie"] = [existing, value.strip()]
489 else:
490 headers["set-cookie"] = value.strip()
491 else:
492 headers[key.lower()] = value.strip()
494 trace("-----------------------")
496 return status, headers, status_message