client.py 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633
  1. from __future__ import annotations
  2. import socket
  3. import ssl as ssl_module
  4. import threading
  5. import warnings
  6. from collections.abc import Sequence
  7. from typing import Any, Callable, Literal, TypeVar, cast
  8. from ..client import ClientProtocol
  9. from ..datastructures import HeadersLike
  10. from ..exceptions import InvalidProxyMessage, InvalidProxyStatus, ProxyError
  11. from ..extensions.base import ClientExtensionFactory
  12. from ..extensions.permessage_deflate import enable_client_permessage_deflate
  13. from ..headers import validate_subprotocols
  14. from ..http11 import USER_AGENT, Response
  15. from ..protocol import CONNECTING, Event
  16. from ..proxy import Proxy, get_proxy, parse_proxy, prepare_connect_request
  17. from ..streams import StreamReader
  18. from ..typing import BytesLike, LoggerLike, Origin, Subprotocol
  19. from ..uri import WebSocketURI, parse_uri
  20. from .connection import Connection
  21. from .utils import Deadline
  22. __all__ = ["connect", "unix_connect", "ClientConnection"]
  23. class ClientConnection(Connection):
  24. """
  25. :mod:`threading` implementation of a WebSocket client connection.
  26. :class:`ClientConnection` provides :meth:`recv` and :meth:`send` methods for
  27. receiving and sending messages.
  28. It supports iteration to receive messages::
  29. for message in websocket:
  30. process(message)
  31. The iterator exits normally when the connection is closed with code
  32. 1000 (OK) or 1001 (going away) or without a close code. It raises a
  33. :exc:`~websockets.exceptions.ConnectionClosedError` when the connection is
  34. closed with any other code.
  35. The ``ping_interval``, ``ping_timeout``, ``close_timeout``, and
  36. ``max_queue`` arguments have the same meaning as in :func:`connect`.
  37. Args:
  38. socket: Socket connected to a WebSocket server.
  39. protocol: Sans-I/O connection.
  40. """
  41. def __init__(
  42. self,
  43. socket: socket.socket,
  44. protocol: ClientProtocol,
  45. *,
  46. ping_interval: float | None = 20,
  47. ping_timeout: float | None = 20,
  48. close_timeout: float | None = 10,
  49. max_queue: int | None | tuple[int | None, int | None] = 16,
  50. ) -> None:
  51. self.protocol: ClientProtocol
  52. self.response_rcvd = threading.Event()
  53. super().__init__(
  54. socket,
  55. protocol,
  56. ping_interval=ping_interval,
  57. ping_timeout=ping_timeout,
  58. close_timeout=close_timeout,
  59. max_queue=max_queue,
  60. )
  61. def handshake(
  62. self,
  63. additional_headers: HeadersLike | None = None,
  64. user_agent_header: str | None = USER_AGENT,
  65. timeout: float | None = None,
  66. ) -> None:
  67. """
  68. Perform the opening handshake.
  69. """
  70. with self.send_context(expected_state=CONNECTING):
  71. self.request = self.protocol.connect()
  72. if additional_headers is not None:
  73. self.request.headers.update(additional_headers)
  74. if user_agent_header is not None:
  75. self.request.headers.setdefault("User-Agent", user_agent_header)
  76. self.protocol.send_request(self.request)
  77. if not self.response_rcvd.wait(timeout):
  78. raise TimeoutError("timed out while waiting for handshake response")
  79. # self.protocol.handshake_exc is set when the connection is lost before
  80. # receiving a response, when the response cannot be parsed, or when the
  81. # response fails the handshake.
  82. if self.protocol.handshake_exc is not None:
  83. raise self.protocol.handshake_exc
  84. def process_event(self, event: Event) -> None:
  85. """
  86. Process one incoming event.
  87. """
  88. # First event - handshake response.
  89. if self.response is None:
  90. assert isinstance(event, Response)
  91. self.response = event
  92. self.response_rcvd.set()
  93. # Later events - frames.
  94. else:
  95. super().process_event(event)
  96. def recv_events(self) -> None:
  97. """
  98. Read incoming data from the socket and process events.
  99. """
  100. try:
  101. super().recv_events()
  102. finally:
  103. # If the connection is closed during the handshake, unblock it.
  104. self.response_rcvd.set()
  105. def connect(
  106. uri: str,
  107. *,
  108. # TCP/TLS
  109. sock: socket.socket | None = None,
  110. ssl: ssl_module.SSLContext | None = None,
  111. server_hostname: str | None = None,
  112. # WebSocket
  113. origin: Origin | None = None,
  114. extensions: Sequence[ClientExtensionFactory] | None = None,
  115. subprotocols: Sequence[Subprotocol] | None = None,
  116. compression: str | None = "deflate",
  117. # HTTP
  118. additional_headers: HeadersLike | None = None,
  119. user_agent_header: str | None = USER_AGENT,
  120. proxy: str | Literal[True] | None = True,
  121. proxy_ssl: ssl_module.SSLContext | None = None,
  122. proxy_server_hostname: str | None = None,
  123. # Timeouts
  124. open_timeout: float | None = 10,
  125. ping_interval: float | None = 20,
  126. ping_timeout: float | None = 20,
  127. close_timeout: float | None = 10,
  128. # Limits
  129. max_size: int | None | tuple[int | None, int | None] = 2**20,
  130. max_queue: int | None | tuple[int | None, int | None] = 16,
  131. # Logging
  132. logger: LoggerLike | None = None,
  133. # Escape hatch for advanced customization
  134. create_connection: type[ClientConnection] | None = None,
  135. **kwargs: Any,
  136. ) -> ClientConnection:
  137. """
  138. Connect to the WebSocket server at ``uri``.
  139. This function returns a :class:`ClientConnection` instance, which you can
  140. use to send and receive messages.
  141. :func:`connect` may be used as a context manager::
  142. from websockets.sync.client import connect
  143. with connect(...) as websocket:
  144. ...
  145. The connection is closed automatically when exiting the context.
  146. Args:
  147. uri: URI of the WebSocket server.
  148. sock: Preexisting TCP socket. ``sock`` overrides the host and port
  149. from ``uri``. You may call :func:`socket.create_connection` to
  150. create a suitable TCP socket.
  151. ssl: Configuration for enabling TLS on the connection.
  152. server_hostname: Host name for the TLS handshake. ``server_hostname``
  153. overrides the host name from ``uri``.
  154. origin: Value of the ``Origin`` header, for servers that require it.
  155. extensions: List of supported extensions, in order in which they
  156. should be negotiated and run.
  157. subprotocols: List of supported subprotocols, in order of decreasing
  158. preference.
  159. compression: The "permessage-deflate" extension is enabled by default.
  160. Set ``compression`` to :obj:`None` to disable it. See the
  161. :doc:`compression guide <../../topics/compression>` for details.
  162. additional_headers (HeadersLike | None): Arbitrary HTTP headers to add
  163. to the handshake request.
  164. user_agent_header: Value of the ``User-Agent`` request header.
  165. It defaults to ``"Python/x.y.z websockets/X.Y"``.
  166. Setting it to :obj:`None` removes the header.
  167. proxy: If a proxy is configured, it is used by default. Set ``proxy``
  168. to :obj:`None` to disable the proxy or to the address of a proxy
  169. to override the system configuration. See the :doc:`proxy docs
  170. <../../topics/proxies>` for details.
  171. proxy_ssl: Configuration for enabling TLS on the proxy connection.
  172. proxy_server_hostname: Host name for the TLS handshake with the proxy.
  173. ``proxy_server_hostname`` overrides the host name from ``proxy``.
  174. open_timeout: Timeout for opening the connection in seconds.
  175. :obj:`None` disables the timeout.
  176. ping_interval: Interval between keepalive pings in seconds.
  177. :obj:`None` disables keepalive.
  178. ping_timeout: Timeout for keepalive pings in seconds.
  179. :obj:`None` disables timeouts.
  180. close_timeout: Timeout for closing the connection in seconds.
  181. :obj:`None` disables the timeout.
  182. max_size: Maximum size of incoming messages in bytes.
  183. :obj:`None` disables the limit. You may pass a ``(max_message_size,
  184. max_fragment_size)`` tuple to set different limits for messages and
  185. fragments when you expect long messages sent in short fragments.
  186. max_queue: High-water mark of the buffer where frames are received.
  187. It defaults to 16 frames. The low-water mark defaults to ``max_queue
  188. // 4``. You may pass a ``(high, low)`` tuple to set the high-water
  189. and low-water marks. If you want to disable flow control entirely,
  190. you may set it to ``None``, although that's a bad idea.
  191. logger: Logger for this client.
  192. It defaults to ``logging.getLogger("websockets.client")``.
  193. See the :doc:`logging guide <../../topics/logging>` for details.
  194. create_connection: Factory for the :class:`ClientConnection` managing
  195. the connection. Set it to a wrapper or a subclass to customize
  196. connection handling.
  197. Any other keyword arguments are passed to :func:`~socket.create_connection`.
  198. Raises:
  199. InvalidURI: If ``uri`` isn't a valid WebSocket URI.
  200. OSError: If the TCP connection fails.
  201. InvalidHandshake: If the opening handshake fails.
  202. TimeoutError: If the opening handshake times out.
  203. """
  204. # Process parameters
  205. # Backwards compatibility: ssl used to be called ssl_context.
  206. if ssl is None and "ssl_context" in kwargs:
  207. ssl = kwargs.pop("ssl_context")
  208. warnings.warn( # deprecated in 13.0 - 2024-08-20
  209. "ssl_context was renamed to ssl",
  210. DeprecationWarning,
  211. )
  212. ws_uri = parse_uri(uri)
  213. if not ws_uri.secure and ssl is not None:
  214. raise ValueError("ssl argument is incompatible with a ws:// URI")
  215. # Private APIs for unix_connect()
  216. unix: bool = kwargs.pop("unix", False)
  217. path: str | None = kwargs.pop("path", None)
  218. if unix:
  219. if path is None and sock is None:
  220. raise ValueError("missing path argument")
  221. elif path is not None and sock is not None:
  222. raise ValueError("path and sock arguments are incompatible")
  223. if subprotocols is not None:
  224. validate_subprotocols(subprotocols)
  225. if compression == "deflate":
  226. extensions = enable_client_permessage_deflate(extensions)
  227. elif compression is not None:
  228. raise ValueError(f"unsupported compression: {compression}")
  229. if unix:
  230. proxy = None
  231. if sock is not None:
  232. proxy = None
  233. if proxy is True:
  234. proxy = get_proxy(ws_uri)
  235. # Calculate timeouts on the TCP, TLS, and WebSocket handshakes.
  236. # The TCP and TLS timeouts must be set on the socket, then removed
  237. # to avoid conflicting with the WebSocket timeout in handshake().
  238. deadline = Deadline(open_timeout)
  239. if create_connection is None:
  240. create_connection = ClientConnection
  241. try:
  242. # Connect socket
  243. if sock is None:
  244. if unix:
  245. sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
  246. sock.settimeout(deadline.timeout())
  247. assert path is not None # mypy cannot figure this out
  248. sock.connect(path)
  249. elif proxy is not None:
  250. proxy_parsed = parse_proxy(proxy)
  251. if proxy_parsed.scheme[:5] == "socks":
  252. # Connect to the server through the proxy.
  253. sock = connect_socks_proxy(
  254. proxy_parsed,
  255. ws_uri,
  256. deadline,
  257. # websockets is consistent with the socket module while
  258. # python_socks is consistent across implementations.
  259. local_addr=kwargs.pop("source_address", None),
  260. )
  261. elif proxy_parsed.scheme[:4] == "http":
  262. # Validate the proxy_ssl argument.
  263. if proxy_parsed.scheme != "https" and proxy_ssl is not None:
  264. raise ValueError(
  265. "proxy_ssl argument is incompatible with an http:// proxy"
  266. )
  267. # Connect to the server through the proxy.
  268. sock = connect_http_proxy(
  269. proxy_parsed,
  270. ws_uri,
  271. deadline,
  272. user_agent_header=user_agent_header,
  273. ssl=proxy_ssl,
  274. server_hostname=proxy_server_hostname,
  275. **kwargs,
  276. )
  277. else:
  278. raise AssertionError("unsupported proxy")
  279. else:
  280. kwargs.setdefault("timeout", deadline.timeout())
  281. sock = socket.create_connection(
  282. (ws_uri.host, ws_uri.port),
  283. **kwargs,
  284. )
  285. sock.settimeout(None)
  286. # Disable Nagle algorithm
  287. if not unix:
  288. sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, True)
  289. # Initialize TLS wrapper and perform TLS handshake
  290. if ws_uri.secure:
  291. if ssl is None:
  292. ssl = ssl_module.create_default_context()
  293. if server_hostname is None:
  294. server_hostname = ws_uri.host
  295. sock.settimeout(deadline.timeout())
  296. if proxy_ssl is None:
  297. sock = ssl.wrap_socket(sock, server_hostname=server_hostname)
  298. else:
  299. sock_2 = SSLSSLSocket(sock, ssl, server_hostname=server_hostname)
  300. # Let's pretend that sock is a socket, even though it isn't.
  301. sock = cast(socket.socket, sock_2)
  302. sock.settimeout(None)
  303. # Initialize WebSocket protocol
  304. protocol = ClientProtocol(
  305. ws_uri,
  306. origin=origin,
  307. extensions=extensions,
  308. subprotocols=subprotocols,
  309. max_size=max_size,
  310. logger=logger,
  311. )
  312. # Initialize WebSocket connection
  313. connection = create_connection(
  314. sock,
  315. protocol,
  316. ping_interval=ping_interval,
  317. ping_timeout=ping_timeout,
  318. close_timeout=close_timeout,
  319. max_queue=max_queue,
  320. )
  321. except Exception:
  322. if sock is not None:
  323. sock.close()
  324. raise
  325. try:
  326. connection.handshake(
  327. additional_headers,
  328. user_agent_header,
  329. deadline.timeout(),
  330. )
  331. except Exception:
  332. connection.close_socket()
  333. connection.recv_events_thread.join()
  334. raise
  335. connection.start_keepalive()
  336. return connection
  337. def unix_connect(
  338. path: str | None = None,
  339. uri: str | None = None,
  340. **kwargs: Any,
  341. ) -> ClientConnection:
  342. """
  343. Connect to a WebSocket server listening on a Unix socket.
  344. This function accepts the same keyword arguments as :func:`connect`.
  345. It's only available on Unix.
  346. It's mainly useful for debugging servers listening on Unix sockets.
  347. Args:
  348. path: File system path to the Unix socket.
  349. uri: URI of the WebSocket server. ``uri`` defaults to
  350. ``ws://localhost/`` or, when a ``ssl`` is provided, to
  351. ``wss://localhost/``.
  352. """
  353. if uri is None:
  354. # Backwards compatibility: ssl used to be called ssl_context.
  355. if kwargs.get("ssl") is None and kwargs.get("ssl_context") is None:
  356. uri = "ws://localhost/"
  357. else:
  358. uri = "wss://localhost/"
  359. return connect(uri=uri, unix=True, path=path, **kwargs)
  360. try:
  361. from python_socks import ProxyType
  362. from python_socks.sync import Proxy as SocksProxy
  363. except ImportError:
  364. def connect_socks_proxy(
  365. proxy: Proxy,
  366. ws_uri: WebSocketURI,
  367. deadline: Deadline,
  368. **kwargs: Any,
  369. ) -> socket.socket:
  370. raise ImportError("connecting through a SOCKS proxy requires python-socks")
  371. else:
  372. SOCKS_PROXY_TYPES = {
  373. "socks5h": ProxyType.SOCKS5,
  374. "socks5": ProxyType.SOCKS5,
  375. "socks4a": ProxyType.SOCKS4,
  376. "socks4": ProxyType.SOCKS4,
  377. }
  378. SOCKS_PROXY_RDNS = {
  379. "socks5h": True,
  380. "socks5": False,
  381. "socks4a": True,
  382. "socks4": False,
  383. }
  384. def connect_socks_proxy(
  385. proxy: Proxy,
  386. ws_uri: WebSocketURI,
  387. deadline: Deadline,
  388. **kwargs: Any,
  389. ) -> socket.socket:
  390. """Connect via a SOCKS proxy and return the socket."""
  391. socks_proxy = SocksProxy(
  392. SOCKS_PROXY_TYPES[proxy.scheme],
  393. proxy.host,
  394. proxy.port,
  395. proxy.username,
  396. proxy.password,
  397. SOCKS_PROXY_RDNS[proxy.scheme],
  398. )
  399. kwargs.setdefault("timeout", deadline.timeout())
  400. # connect() is documented to raise OSError and TimeoutError.
  401. # Wrap other exceptions in ProxyError, a subclass of InvalidHandshake.
  402. try:
  403. return socks_proxy.connect(ws_uri.host, ws_uri.port, **kwargs)
  404. except (OSError, TimeoutError, socket.timeout):
  405. raise
  406. except Exception as exc:
  407. raise ProxyError("failed to connect to SOCKS proxy") from exc
  408. def read_connect_response(sock: socket.socket, deadline: Deadline) -> Response:
  409. reader = StreamReader()
  410. parser = Response.parse(
  411. reader.read_line,
  412. reader.read_exact,
  413. reader.read_to_eof,
  414. proxy=True,
  415. )
  416. try:
  417. while True:
  418. sock.settimeout(deadline.timeout())
  419. data = sock.recv(4096)
  420. if data:
  421. reader.feed_data(data)
  422. else:
  423. reader.feed_eof()
  424. next(parser)
  425. except StopIteration as exc:
  426. assert isinstance(exc.value, Response) # help mypy
  427. response = exc.value
  428. if 200 <= response.status_code < 300:
  429. return response
  430. else:
  431. raise InvalidProxyStatus(response)
  432. except socket.timeout:
  433. raise TimeoutError("timed out while connecting to HTTP proxy")
  434. except Exception as exc:
  435. raise InvalidProxyMessage(
  436. "did not receive a valid HTTP response from proxy"
  437. ) from exc
  438. finally:
  439. sock.settimeout(None)
  440. def connect_http_proxy(
  441. proxy: Proxy,
  442. ws_uri: WebSocketURI,
  443. deadline: Deadline,
  444. *,
  445. user_agent_header: str | None = None,
  446. ssl: ssl_module.SSLContext | None = None,
  447. server_hostname: str | None = None,
  448. **kwargs: Any,
  449. ) -> socket.socket:
  450. # Connect socket
  451. kwargs.setdefault("timeout", deadline.timeout())
  452. sock = socket.create_connection((proxy.host, proxy.port), **kwargs)
  453. # Initialize TLS wrapper and perform TLS handshake
  454. if proxy.scheme == "https":
  455. if ssl is None:
  456. ssl = ssl_module.create_default_context()
  457. if server_hostname is None:
  458. server_hostname = proxy.host
  459. sock.settimeout(deadline.timeout())
  460. sock = ssl.wrap_socket(sock, server_hostname=server_hostname)
  461. sock.settimeout(None)
  462. # Send CONNECT request to the proxy and read response.
  463. sock.sendall(prepare_connect_request(proxy, ws_uri, user_agent_header))
  464. try:
  465. read_connect_response(sock, deadline)
  466. except Exception:
  467. sock.close()
  468. raise
  469. return sock
  470. T = TypeVar("T")
  471. F = TypeVar("F", bound=Callable[..., T])
  472. class SSLSSLSocket:
  473. """
  474. Socket-like object providing TLS-in-TLS.
  475. Only methods that are used by websockets are implemented.
  476. """
  477. recv_bufsize = 65536
  478. def __init__(
  479. self,
  480. sock: socket.socket,
  481. ssl_context: ssl_module.SSLContext,
  482. server_hostname: str | None = None,
  483. ) -> None:
  484. self.incoming = ssl_module.MemoryBIO()
  485. self.outgoing = ssl_module.MemoryBIO()
  486. self.ssl_socket = sock
  487. self.ssl_object = ssl_context.wrap_bio(
  488. self.incoming,
  489. self.outgoing,
  490. server_hostname=server_hostname,
  491. )
  492. self.run_io(self.ssl_object.do_handshake)
  493. def run_io(self, func: Callable[..., T], *args: Any) -> T:
  494. while True:
  495. want_read = False
  496. want_write = False
  497. try:
  498. result = func(*args)
  499. except ssl_module.SSLWantReadError:
  500. want_read = True
  501. except ssl_module.SSLWantWriteError: # pragma: no cover
  502. want_write = True
  503. # Write outgoing data in all cases.
  504. data = self.outgoing.read()
  505. if data:
  506. self.ssl_socket.sendall(data)
  507. # Read incoming data and retry on SSLWantReadError.
  508. if want_read:
  509. data = self.ssl_socket.recv(self.recv_bufsize)
  510. if data:
  511. self.incoming.write(data)
  512. else:
  513. self.incoming.write_eof()
  514. continue
  515. # Retry after writing outgoing data on SSLWantWriteError.
  516. if want_write: # pragma: no cover
  517. continue
  518. # Return result if no error happened.
  519. return result
  520. def recv(self, buflen: int) -> bytes:
  521. try:
  522. return self.run_io(self.ssl_object.read, buflen)
  523. except ssl_module.SSLEOFError:
  524. return b"" # always ignore ragged EOFs
  525. def send(self, data: BytesLike) -> int:
  526. return self.run_io(self.ssl_object.write, data)
  527. def sendall(self, data: BytesLike) -> None:
  528. # adapted from ssl_module.SSLSocket.sendall()
  529. count = 0
  530. with memoryview(data) as view, view.cast("B") as byte_view:
  531. amount = len(byte_view)
  532. while count < amount:
  533. count += self.send(byte_view[count:])
  534. # recv_into(), recvfrom(), recvfrom_into(), sendto(), unwrap(), and the
  535. # flags argument aren't implemented because websockets doesn't need them.
  536. def __getattr__(self, name: str) -> Any:
  537. return getattr(self.ssl_socket, name)