client.py 30 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804
  1. from __future__ import annotations
  2. import asyncio
  3. import logging
  4. import os
  5. import socket
  6. import ssl as ssl_module
  7. import traceback
  8. import urllib.parse
  9. from collections.abc import AsyncIterator, Generator, Sequence
  10. from types import TracebackType
  11. from typing import Any, Callable, Literal, cast
  12. from ..client import ClientProtocol, backoff
  13. from ..datastructures import HeadersLike
  14. from ..exceptions import (
  15. InvalidMessage,
  16. InvalidProxyMessage,
  17. InvalidProxyStatus,
  18. InvalidStatus,
  19. ProxyError,
  20. SecurityError,
  21. )
  22. from ..extensions.base import ClientExtensionFactory
  23. from ..extensions.permessage_deflate import enable_client_permessage_deflate
  24. from ..headers import validate_subprotocols
  25. from ..http11 import USER_AGENT, Response
  26. from ..protocol import CONNECTING, Event
  27. from ..proxy import Proxy, get_proxy, parse_proxy, prepare_connect_request
  28. from ..streams import StreamReader
  29. from ..typing import LoggerLike, Origin, Subprotocol
  30. from ..uri import WebSocketURI, parse_uri
  31. from .compatibility import TimeoutError, asyncio_timeout
  32. from .connection import Connection
  33. __all__ = ["connect", "unix_connect", "ClientConnection"]
  34. MAX_REDIRECTS = int(os.environ.get("WEBSOCKETS_MAX_REDIRECTS", "10"))
  35. class ClientConnection(Connection):
  36. """
  37. :mod:`asyncio` implementation of a WebSocket client connection.
  38. :class:`ClientConnection` provides :meth:`recv` and :meth:`send` coroutines
  39. for receiving and sending messages.
  40. It supports asynchronous iteration to receive messages::
  41. async for message in websocket:
  42. await process(message)
  43. The iterator exits normally when the connection is closed with code
  44. 1000 (OK) or 1001 (going away) or without a close code. It raises a
  45. :exc:`~websockets.exceptions.ConnectionClosedError` when the connection is
  46. closed with any other code.
  47. The ``ping_interval``, ``ping_timeout``, ``close_timeout``, ``max_queue``,
  48. and ``write_limit`` arguments have the same meaning as in :func:`connect`.
  49. Args:
  50. protocol: Sans-I/O connection.
  51. """
  52. def __init__(
  53. self,
  54. protocol: ClientProtocol,
  55. *,
  56. ping_interval: float | None = 20,
  57. ping_timeout: float | None = 20,
  58. close_timeout: float | None = 10,
  59. max_queue: int | None | tuple[int | None, int | None] = 16,
  60. write_limit: int | tuple[int, int | None] = 2**15,
  61. ) -> None:
  62. self.protocol: ClientProtocol
  63. super().__init__(
  64. protocol,
  65. ping_interval=ping_interval,
  66. ping_timeout=ping_timeout,
  67. close_timeout=close_timeout,
  68. max_queue=max_queue,
  69. write_limit=write_limit,
  70. )
  71. self.response_rcvd: asyncio.Future[None] = self.loop.create_future()
  72. async def handshake(
  73. self,
  74. additional_headers: HeadersLike | None = None,
  75. user_agent_header: str | None = USER_AGENT,
  76. ) -> None:
  77. """
  78. Perform the opening handshake.
  79. """
  80. async with self.send_context(expected_state=CONNECTING):
  81. self.request = self.protocol.connect()
  82. if additional_headers is not None:
  83. self.request.headers.update(additional_headers)
  84. if user_agent_header is not None:
  85. self.request.headers.setdefault("User-Agent", user_agent_header)
  86. self.protocol.send_request(self.request)
  87. await asyncio.wait(
  88. [self.response_rcvd, self.connection_lost_waiter],
  89. return_when=asyncio.FIRST_COMPLETED,
  90. )
  91. # self.protocol.handshake_exc is set when the connection is lost before
  92. # receiving a response, when the response cannot be parsed, or when the
  93. # response fails the handshake.
  94. if self.protocol.handshake_exc is not None:
  95. raise self.protocol.handshake_exc
  96. def process_event(self, event: Event) -> None:
  97. """
  98. Process one incoming event.
  99. """
  100. # First event - handshake response.
  101. if self.response is None:
  102. assert isinstance(event, Response)
  103. self.response = event
  104. self.response_rcvd.set_result(None)
  105. # Later events - frames.
  106. else:
  107. super().process_event(event)
  108. def process_exception(exc: Exception) -> Exception | None:
  109. """
  110. Determine whether a connection error is retryable or fatal.
  111. When reconnecting automatically with ``async for ... in connect(...)``, if a
  112. connection attempt fails, :func:`process_exception` is called to determine
  113. whether to retry connecting or to raise the exception.
  114. This function defines the default behavior, which is to retry on:
  115. * :exc:`EOFError`, :exc:`OSError`, :exc:`asyncio.TimeoutError`: network
  116. errors;
  117. * :exc:`~websockets.exceptions.InvalidStatus` when the status code is 500,
  118. 502, 503, or 504: server or proxy errors.
  119. All other exceptions are considered fatal.
  120. You can change this behavior with the ``process_exception`` argument of
  121. :func:`connect`.
  122. Return :obj:`None` if the exception is retryable i.e. when the error could
  123. be transient and trying to reconnect with the same parameters could succeed.
  124. The exception will be logged at the ``INFO`` level.
  125. Return an exception, either ``exc`` or a new exception, if the exception is
  126. fatal i.e. when trying to reconnect will most likely produce the same error.
  127. That exception will be raised, breaking out of the retry loop.
  128. """
  129. # This catches python-socks' ProxyConnectionError and ProxyTimeoutError.
  130. # Remove asyncio.TimeoutError when dropping Python < 3.11.
  131. if isinstance(exc, (OSError, TimeoutError, asyncio.TimeoutError)):
  132. return None
  133. if isinstance(exc, InvalidMessage) and isinstance(exc.__cause__, EOFError):
  134. return None
  135. if isinstance(exc, InvalidStatus) and exc.response.status_code in [
  136. 500, # Internal Server Error
  137. 502, # Bad Gateway
  138. 503, # Service Unavailable
  139. 504, # Gateway Timeout
  140. ]:
  141. return None
  142. return exc
  143. # This is spelled in lower case because it's exposed as a callable in the API.
  144. class connect:
  145. """
  146. Connect to the WebSocket server at ``uri``.
  147. This coroutine returns a :class:`ClientConnection` instance, which you can
  148. use to send and receive messages.
  149. :func:`connect` may be used as an asynchronous context manager::
  150. from websockets.asyncio.client import connect
  151. async with connect(...) as websocket:
  152. ...
  153. The connection is closed automatically when exiting the context.
  154. :func:`connect` can be used as an infinite asynchronous iterator to
  155. reconnect automatically on errors::
  156. async for websocket in connect(...):
  157. try:
  158. ...
  159. except websockets.exceptions.ConnectionClosed:
  160. continue
  161. If the connection fails with a transient error, it is retried with
  162. exponential backoff. If it fails with a fatal error, the exception is
  163. raised, breaking out of the loop.
  164. The connection is closed automatically after each iteration of the loop.
  165. Args:
  166. uri: URI of the WebSocket server.
  167. origin: Value of the ``Origin`` header, for servers that require it.
  168. extensions: List of supported extensions, in order in which they
  169. should be negotiated and run.
  170. subprotocols: List of supported subprotocols, in order of decreasing
  171. preference.
  172. compression: The "permessage-deflate" extension is enabled by default.
  173. Set ``compression`` to :obj:`None` to disable it. See the
  174. :doc:`compression guide <../../topics/compression>` for details.
  175. additional_headers (HeadersLike | None): Arbitrary HTTP headers to add
  176. to the handshake request.
  177. user_agent_header: Value of the ``User-Agent`` request header.
  178. It defaults to ``"Python/x.y.z websockets/X.Y"``.
  179. Setting it to :obj:`None` removes the header.
  180. proxy: If a proxy is configured, it is used by default. Set ``proxy``
  181. to :obj:`None` to disable the proxy or to the address of a proxy
  182. to override the system configuration. See the :doc:`proxy docs
  183. <../../topics/proxies>` for details.
  184. process_exception: When reconnecting automatically, tell whether an
  185. error is transient or fatal. The default behavior is defined by
  186. :func:`process_exception`. Refer to its documentation for details.
  187. open_timeout: Timeout for opening the connection in seconds.
  188. :obj:`None` disables the timeout.
  189. ping_interval: Interval between keepalive pings in seconds.
  190. :obj:`None` disables keepalive.
  191. ping_timeout: Timeout for keepalive pings in seconds.
  192. :obj:`None` disables timeouts.
  193. close_timeout: Timeout for closing the connection in seconds.
  194. :obj:`None` disables the timeout.
  195. max_size: Maximum size of incoming messages in bytes.
  196. :obj:`None` disables the limit. You may pass a ``(max_message_size,
  197. max_fragment_size)`` tuple to set different limits for messages and
  198. fragments when you expect long messages sent in short fragments.
  199. max_queue: High-water mark of the buffer where frames are received.
  200. It defaults to 16 frames. The low-water mark defaults to ``max_queue
  201. // 4``. You may pass a ``(high, low)`` tuple to set the high-water
  202. and low-water marks. If you want to disable flow control entirely,
  203. you may set it to ``None``, although that's a bad idea.
  204. write_limit: High-water mark of write buffer in bytes. It is passed to
  205. :meth:`~asyncio.WriteTransport.set_write_buffer_limits`. It defaults
  206. to 32 KiB. You may pass a ``(high, low)`` tuple to set the
  207. high-water and low-water marks.
  208. logger: Logger for this client.
  209. It defaults to ``logging.getLogger("websockets.client")``.
  210. See the :doc:`logging guide <../../topics/logging>` for details.
  211. create_connection: Factory for the :class:`ClientConnection` managing
  212. the connection. Set it to a wrapper or a subclass to customize
  213. connection handling.
  214. Any other keyword arguments are passed to the event loop's
  215. :meth:`~asyncio.loop.create_connection` method.
  216. For example:
  217. * You can set ``ssl`` to a :class:`~ssl.SSLContext` to enforce TLS settings.
  218. When connecting to a ``wss://`` URI, if ``ssl`` isn't provided, a TLS
  219. context is created with :func:`~ssl.create_default_context`.
  220. * You can set ``server_hostname`` to override the host name from ``uri`` in
  221. the TLS handshake.
  222. * You can set ``host`` and ``port`` to connect to a different host and port
  223. from those found in ``uri``. This only changes the destination of the TCP
  224. connection. The host name from ``uri`` is still used in the TLS handshake
  225. for secure connections and in the ``Host`` header.
  226. * You can set ``sock`` to provide a preexisting TCP socket. You may call
  227. :func:`socket.create_connection` (not to be confused with the event loop's
  228. :meth:`~asyncio.loop.create_connection` method) to create a suitable
  229. client socket and customize it.
  230. When using a proxy:
  231. * Prefix keyword arguments with ``proxy_`` for configuring TLS between the
  232. client and an HTTPS proxy: ``proxy_ssl``, ``proxy_server_hostname``,
  233. ``proxy_ssl_handshake_timeout``, and ``proxy_ssl_shutdown_timeout``.
  234. * Use the standard keyword arguments for configuring TLS between the proxy
  235. and the WebSocket server: ``ssl``, ``server_hostname``,
  236. ``ssl_handshake_timeout``, and ``ssl_shutdown_timeout``.
  237. * Other keyword arguments are used only for connecting to the proxy.
  238. Raises:
  239. InvalidURI: If ``uri`` isn't a valid WebSocket URI.
  240. InvalidProxy: If ``proxy`` isn't a valid proxy.
  241. OSError: If the TCP connection fails.
  242. InvalidHandshake: If the opening handshake fails.
  243. TimeoutError: If the opening handshake times out.
  244. """
  245. def __init__(
  246. self,
  247. uri: str,
  248. *,
  249. # WebSocket
  250. origin: Origin | None = None,
  251. extensions: Sequence[ClientExtensionFactory] | None = None,
  252. subprotocols: Sequence[Subprotocol] | None = None,
  253. compression: str | None = "deflate",
  254. # HTTP
  255. additional_headers: HeadersLike | None = None,
  256. user_agent_header: str | None = USER_AGENT,
  257. proxy: str | Literal[True] | None = True,
  258. process_exception: Callable[[Exception], Exception | None] = process_exception,
  259. # Timeouts
  260. open_timeout: float | None = 10,
  261. ping_interval: float | None = 20,
  262. ping_timeout: float | None = 20,
  263. close_timeout: float | None = 10,
  264. # Limits
  265. max_size: int | None | tuple[int | None, int | None] = 2**20,
  266. max_queue: int | None | tuple[int | None, int | None] = 16,
  267. write_limit: int | tuple[int, int | None] = 2**15,
  268. # Logging
  269. logger: LoggerLike | None = None,
  270. # Escape hatch for advanced customization
  271. create_connection: type[ClientConnection] | None = None,
  272. # Other keyword arguments are passed to loop.create_connection
  273. **kwargs: Any,
  274. ) -> None:
  275. self.uri = uri
  276. if subprotocols is not None:
  277. validate_subprotocols(subprotocols)
  278. if compression == "deflate":
  279. extensions = enable_client_permessage_deflate(extensions)
  280. elif compression is not None:
  281. raise ValueError(f"unsupported compression: {compression}")
  282. if logger is None:
  283. logger = logging.getLogger("websockets.client")
  284. if create_connection is None:
  285. create_connection = ClientConnection
  286. def protocol_factory(uri: WebSocketURI) -> ClientConnection:
  287. # This is a protocol in the Sans-I/O implementation of websockets.
  288. protocol = ClientProtocol(
  289. uri,
  290. origin=origin,
  291. extensions=extensions,
  292. subprotocols=subprotocols,
  293. max_size=max_size,
  294. logger=logger,
  295. )
  296. # This is a connection in websockets and a protocol in asyncio.
  297. connection = create_connection(
  298. protocol,
  299. ping_interval=ping_interval,
  300. ping_timeout=ping_timeout,
  301. close_timeout=close_timeout,
  302. max_queue=max_queue,
  303. write_limit=write_limit,
  304. )
  305. return connection
  306. self.proxy = proxy
  307. self.protocol_factory = protocol_factory
  308. self.additional_headers = additional_headers
  309. self.user_agent_header = user_agent_header
  310. self.process_exception = process_exception
  311. self.open_timeout = open_timeout
  312. self.logger = logger
  313. self.connection_kwargs = kwargs
  314. async def create_connection(self) -> ClientConnection:
  315. """Create TCP or Unix connection."""
  316. loop = asyncio.get_running_loop()
  317. kwargs = self.connection_kwargs.copy()
  318. ws_uri = parse_uri(self.uri)
  319. proxy = self.proxy
  320. if kwargs.get("unix", False):
  321. proxy = None
  322. if kwargs.get("sock") is not None:
  323. proxy = None
  324. if proxy is True:
  325. proxy = get_proxy(ws_uri)
  326. def factory() -> ClientConnection:
  327. return self.protocol_factory(ws_uri)
  328. if ws_uri.secure:
  329. kwargs.setdefault("ssl", True)
  330. kwargs.setdefault("server_hostname", ws_uri.host)
  331. if kwargs.get("ssl") is None:
  332. raise ValueError("ssl=None is incompatible with a wss:// URI")
  333. else:
  334. if kwargs.get("ssl") is not None:
  335. raise ValueError("ssl argument is incompatible with a ws:// URI")
  336. if kwargs.pop("unix", False):
  337. _, connection = await loop.create_unix_connection(factory, **kwargs)
  338. elif proxy is not None:
  339. proxy_parsed = parse_proxy(proxy)
  340. if proxy_parsed.scheme[:5] == "socks":
  341. # Connect to the server through the proxy.
  342. sock = await connect_socks_proxy(
  343. proxy_parsed,
  344. ws_uri,
  345. local_addr=kwargs.pop("local_addr", None),
  346. )
  347. # Initialize WebSocket connection via the proxy.
  348. _, connection = await loop.create_connection(
  349. factory,
  350. sock=sock,
  351. **kwargs,
  352. )
  353. elif proxy_parsed.scheme[:4] == "http":
  354. # Split keyword arguments between the proxy and the server.
  355. all_kwargs, proxy_kwargs, kwargs = kwargs, {}, {}
  356. for key, value in all_kwargs.items():
  357. if key.startswith("ssl") or key == "server_hostname":
  358. kwargs[key] = value
  359. elif key.startswith("proxy_"):
  360. proxy_kwargs[key[6:]] = value
  361. else:
  362. proxy_kwargs[key] = value
  363. # Validate the proxy_ssl argument.
  364. if proxy_parsed.scheme == "https":
  365. proxy_kwargs.setdefault("ssl", True)
  366. if proxy_kwargs.get("ssl") is None:
  367. raise ValueError(
  368. "proxy_ssl=None is incompatible with an https:// proxy"
  369. )
  370. else:
  371. if proxy_kwargs.get("ssl") is not None:
  372. raise ValueError(
  373. "proxy_ssl argument is incompatible with an http:// proxy"
  374. )
  375. # Connect to the server through the proxy.
  376. transport = await connect_http_proxy(
  377. proxy_parsed,
  378. ws_uri,
  379. user_agent_header=self.user_agent_header,
  380. **proxy_kwargs,
  381. )
  382. # Initialize WebSocket connection via the proxy.
  383. connection = factory()
  384. transport.set_protocol(connection)
  385. ssl = kwargs.pop("ssl", None)
  386. if ssl is True:
  387. ssl = ssl_module.create_default_context()
  388. if ssl is not None:
  389. new_transport = await loop.start_tls(
  390. transport, connection, ssl, **kwargs
  391. )
  392. assert new_transport is not None # help mypy
  393. transport = new_transport
  394. connection.connection_made(transport)
  395. else:
  396. raise AssertionError("unsupported proxy")
  397. else:
  398. # Connect to the server directly.
  399. if kwargs.get("sock") is None:
  400. kwargs.setdefault("host", ws_uri.host)
  401. kwargs.setdefault("port", ws_uri.port)
  402. # Initialize WebSocket connection.
  403. _, connection = await loop.create_connection(factory, **kwargs)
  404. return connection
  405. def process_redirect(self, exc: Exception) -> Exception | str:
  406. """
  407. Determine whether a connection error is a redirect that can be followed.
  408. Return the new URI if it's a valid redirect. Else, return an exception.
  409. """
  410. if not (
  411. isinstance(exc, InvalidStatus)
  412. and exc.response.status_code
  413. in [
  414. 300, # Multiple Choices
  415. 301, # Moved Permanently
  416. 302, # Found
  417. 303, # See Other
  418. 307, # Temporary Redirect
  419. 308, # Permanent Redirect
  420. ]
  421. and "Location" in exc.response.headers
  422. ):
  423. return exc
  424. old_ws_uri = parse_uri(self.uri)
  425. new_uri = urllib.parse.urljoin(self.uri, exc.response.headers["Location"])
  426. new_ws_uri = parse_uri(new_uri)
  427. # If connect() received a socket, it is closed and cannot be reused.
  428. if self.connection_kwargs.get("sock") is not None:
  429. return ValueError(
  430. f"cannot follow redirect to {new_uri} with a preexisting socket"
  431. )
  432. # TLS downgrade is forbidden.
  433. if old_ws_uri.secure and not new_ws_uri.secure:
  434. return SecurityError(f"cannot follow redirect to non-secure URI {new_uri}")
  435. # Apply restrictions to cross-origin redirects.
  436. if (
  437. old_ws_uri.secure != new_ws_uri.secure
  438. or old_ws_uri.host != new_ws_uri.host
  439. or old_ws_uri.port != new_ws_uri.port
  440. ):
  441. # Cross-origin redirects on Unix sockets don't quite make sense.
  442. if self.connection_kwargs.get("unix", False):
  443. return ValueError(
  444. f"cannot follow cross-origin redirect to {new_uri} "
  445. f"with a Unix socket"
  446. )
  447. # Cross-origin redirects when host and port are overridden are ill-defined.
  448. if (
  449. self.connection_kwargs.get("host") is not None
  450. or self.connection_kwargs.get("port") is not None
  451. ):
  452. return ValueError(
  453. f"cannot follow cross-origin redirect to {new_uri} "
  454. f"with an explicit host or port"
  455. )
  456. return new_uri
  457. # ... = await connect(...)
  458. def __await__(self) -> Generator[Any, None, ClientConnection]:
  459. # Create a suitable iterator by calling __await__ on a coroutine.
  460. return self.__await_impl__().__await__()
  461. async def __await_impl__(self) -> ClientConnection:
  462. try:
  463. async with asyncio_timeout(self.open_timeout):
  464. for _ in range(MAX_REDIRECTS):
  465. self.connection = await self.create_connection()
  466. try:
  467. await self.connection.handshake(
  468. self.additional_headers,
  469. self.user_agent_header,
  470. )
  471. except asyncio.CancelledError:
  472. self.connection.transport.abort()
  473. raise
  474. except Exception as exc:
  475. # Always close the connection even though keep-alive is
  476. # the default in HTTP/1.1 because create_connection ties
  477. # opening the network connection with initializing the
  478. # protocol. In the current design of connect(), there is
  479. # no easy way to reuse the network connection that works
  480. # in every case nor to reinitialize the protocol.
  481. self.connection.transport.abort()
  482. uri_or_exc = self.process_redirect(exc)
  483. # Response is a valid redirect; follow it.
  484. if isinstance(uri_or_exc, str):
  485. self.uri = uri_or_exc
  486. continue
  487. # Response isn't a valid redirect; raise the exception.
  488. if uri_or_exc is exc:
  489. raise
  490. else:
  491. raise uri_or_exc from exc
  492. else:
  493. self.connection.start_keepalive()
  494. return self.connection
  495. else:
  496. raise SecurityError(f"more than {MAX_REDIRECTS} redirects")
  497. except TimeoutError as exc:
  498. # Re-raise exception with an informative error message.
  499. raise TimeoutError("timed out during opening handshake") from exc
  500. # ... = yield from connect(...) - remove when dropping Python < 3.11
  501. __iter__ = __await__
  502. # async with connect(...) as ...: ...
  503. async def __aenter__(self) -> ClientConnection:
  504. return await self
  505. async def __aexit__(
  506. self,
  507. exc_type: type[BaseException] | None,
  508. exc_value: BaseException | None,
  509. traceback: TracebackType | None,
  510. ) -> None:
  511. await self.connection.close()
  512. # async for ... in connect(...):
  513. async def __aiter__(self) -> AsyncIterator[ClientConnection]:
  514. delays: Generator[float] | None = None
  515. while True:
  516. try:
  517. async with self as protocol:
  518. yield protocol
  519. except Exception as exc:
  520. # Determine whether the exception is retryable or fatal.
  521. # The API of process_exception is "return an exception or None";
  522. # "raise an exception" is also supported because it's a frequent
  523. # mistake. It isn't documented in order to keep the API simple.
  524. try:
  525. new_exc = self.process_exception(exc)
  526. except Exception as raised_exc:
  527. new_exc = raised_exc
  528. # The connection failed with a fatal error.
  529. # Raise the exception and exit the loop.
  530. if new_exc is exc:
  531. raise
  532. if new_exc is not None:
  533. raise new_exc from exc
  534. # The connection failed with a retryable error.
  535. # Start or continue backoff and reconnect.
  536. if delays is None:
  537. delays = backoff()
  538. delay = next(delays)
  539. self.logger.info(
  540. "connect failed; reconnecting in %.1f seconds: %s",
  541. delay,
  542. traceback.format_exception_only(exc)[0].strip(),
  543. )
  544. await asyncio.sleep(delay)
  545. continue
  546. else:
  547. # The connection succeeded. Reset backoff.
  548. delays = None
  549. def unix_connect(
  550. path: str | None = None,
  551. uri: str | None = None,
  552. **kwargs: Any,
  553. ) -> connect:
  554. """
  555. Connect to a WebSocket server listening on a Unix socket.
  556. This function accepts the same keyword arguments as :func:`connect`.
  557. It's only available on Unix.
  558. It's mainly useful for debugging servers listening on Unix sockets.
  559. Args:
  560. path: File system path to the Unix socket.
  561. uri: URI of the WebSocket server. ``uri`` defaults to
  562. ``ws://localhost/`` or, when a ``ssl`` argument is provided, to
  563. ``wss://localhost/``.
  564. """
  565. if uri is None:
  566. if kwargs.get("ssl") is None:
  567. uri = "ws://localhost/"
  568. else:
  569. uri = "wss://localhost/"
  570. return connect(uri=uri, unix=True, path=path, **kwargs)
  571. try:
  572. from python_socks import ProxyType
  573. from python_socks.async_.asyncio import Proxy as SocksProxy
  574. except ImportError:
  575. async def connect_socks_proxy(
  576. proxy: Proxy,
  577. ws_uri: WebSocketURI,
  578. **kwargs: Any,
  579. ) -> socket.socket:
  580. raise ImportError("connecting through a SOCKS proxy requires python-socks")
  581. else:
  582. SOCKS_PROXY_TYPES = {
  583. "socks5h": ProxyType.SOCKS5,
  584. "socks5": ProxyType.SOCKS5,
  585. "socks4a": ProxyType.SOCKS4,
  586. "socks4": ProxyType.SOCKS4,
  587. }
  588. SOCKS_PROXY_RDNS = {
  589. "socks5h": True,
  590. "socks5": False,
  591. "socks4a": True,
  592. "socks4": False,
  593. }
  594. async def connect_socks_proxy(
  595. proxy: Proxy,
  596. ws_uri: WebSocketURI,
  597. **kwargs: Any,
  598. ) -> socket.socket:
  599. """Connect via a SOCKS proxy and return the socket."""
  600. socks_proxy = SocksProxy(
  601. SOCKS_PROXY_TYPES[proxy.scheme],
  602. proxy.host,
  603. proxy.port,
  604. proxy.username,
  605. proxy.password,
  606. SOCKS_PROXY_RDNS[proxy.scheme],
  607. )
  608. # connect() is documented to raise OSError.
  609. # socks_proxy.connect() doesn't raise TimeoutError; it gets canceled.
  610. # Wrap other exceptions in ProxyError, a subclass of InvalidHandshake.
  611. try:
  612. return await socks_proxy.connect(ws_uri.host, ws_uri.port, **kwargs)
  613. except OSError:
  614. raise
  615. except Exception as exc:
  616. raise ProxyError("failed to connect to SOCKS proxy") from exc
  617. class HTTPProxyConnection(asyncio.Protocol):
  618. def __init__(
  619. self,
  620. ws_uri: WebSocketURI,
  621. proxy: Proxy,
  622. user_agent_header: str | None = None,
  623. ):
  624. self.ws_uri = ws_uri
  625. self.proxy = proxy
  626. self.user_agent_header = user_agent_header
  627. self.reader = StreamReader()
  628. self.parser = Response.parse(
  629. self.reader.read_line,
  630. self.reader.read_exact,
  631. self.reader.read_to_eof,
  632. proxy=True,
  633. )
  634. loop = asyncio.get_running_loop()
  635. self.response: asyncio.Future[Response] = loop.create_future()
  636. def run_parser(self) -> None:
  637. try:
  638. next(self.parser)
  639. except StopIteration as exc:
  640. response = exc.value
  641. if 200 <= response.status_code < 300:
  642. self.response.set_result(response)
  643. else:
  644. self.response.set_exception(InvalidProxyStatus(response))
  645. except Exception as exc:
  646. proxy_exc = InvalidProxyMessage(
  647. "did not receive a valid HTTP response from proxy"
  648. )
  649. proxy_exc.__cause__ = exc
  650. self.response.set_exception(proxy_exc)
  651. def connection_made(self, transport: asyncio.BaseTransport) -> None:
  652. transport = cast(asyncio.Transport, transport)
  653. self.transport = transport
  654. self.transport.write(
  655. prepare_connect_request(self.proxy, self.ws_uri, self.user_agent_header)
  656. )
  657. def data_received(self, data: bytes) -> None:
  658. self.reader.feed_data(data)
  659. self.run_parser()
  660. def eof_received(self) -> None:
  661. self.reader.feed_eof()
  662. self.run_parser()
  663. def connection_lost(self, exc: Exception | None) -> None:
  664. self.reader.feed_eof()
  665. if exc is not None:
  666. self.response.set_exception(exc)
  667. async def connect_http_proxy(
  668. proxy: Proxy,
  669. ws_uri: WebSocketURI,
  670. user_agent_header: str | None = None,
  671. **kwargs: Any,
  672. ) -> asyncio.Transport:
  673. transport, protocol = await asyncio.get_running_loop().create_connection(
  674. lambda: HTTPProxyConnection(ws_uri, proxy, user_agent_header),
  675. proxy.host,
  676. proxy.port,
  677. **kwargs,
  678. )
  679. try:
  680. # This raises exceptions if the connection to the proxy fails.
  681. await protocol.response
  682. except Exception:
  683. transport.close()
  684. raise
  685. return transport