diff options
| author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-06-12 05:43:14 +0000 |
|---|---|---|
| committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-06-12 05:43:14 +0000 |
| commit | 8dd16259287f58f9273002717ec4d27e97127719 (patch) | |
| tree | 3863e62a53829a84037444beab3abd4ed9dfc7d0 /testing/web-platform/tests/tools/third_party/websockets/src | |
| parent | Releasing progress-linux version 126.0.1-1~progress7.99u1. (diff) | |
| download | firefox-8dd16259287f58f9273002717ec4d27e97127719.tar.xz firefox-8dd16259287f58f9273002717ec4d27e97127719.zip | |
Merging upstream version 127.0.
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'testing/web-platform/tests/tools/third_party/websockets/src')
38 files changed, 3903 insertions, 1694 deletions
diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets.egg-info/PKG-INFO b/testing/web-platform/tests/tools/third_party/websockets/src/websockets.egg-info/PKG-INFO deleted file mode 100644 index 3b042a3f9f..0000000000 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets.egg-info/PKG-INFO +++ /dev/null @@ -1,174 +0,0 @@ -Metadata-Version: 2.1 -Name: websockets -Version: 10.3 -Summary: An implementation of the WebSocket Protocol (RFC 6455 & 7692) -Home-page: https://github.com/aaugustin/websockets -Author: Aymeric Augustin -Author-email: aymeric.augustin@m4x.org -License: BSD -Project-URL: Changelog, https://websockets.readthedocs.io/en/stable/project/changelog.html -Project-URL: Documentation, https://websockets.readthedocs.io/ -Project-URL: Funding, https://tidelift.com/subscription/pkg/pypi-websockets?utm_source=pypi-websockets&utm_medium=referral&utm_campaign=readme -Project-URL: Tracker, https://github.com/aaugustin/websockets/issues -Platform: UNKNOWN -Classifier: Development Status :: 5 - Production/Stable -Classifier: Environment :: Web Environment -Classifier: Intended Audience :: Developers -Classifier: License :: OSI Approved :: BSD License -Classifier: Operating System :: OS Independent -Classifier: Programming Language :: Python -Classifier: Programming Language :: Python :: 3 -Classifier: Programming Language :: Python :: 3.7 -Classifier: Programming Language :: Python :: 3.8 -Classifier: Programming Language :: Python :: 3.9 -Classifier: Programming Language :: Python :: 3.10 -Requires-Python: >=3.7 -License-File: LICENSE - -.. image:: logo/horizontal.svg - :width: 480px - :alt: websockets - -|licence| |version| |pyversions| |wheel| |tests| |docs| - -.. |licence| image:: https://img.shields.io/pypi/l/websockets.svg - :target: https://pypi.python.org/pypi/websockets - -.. |version| image:: https://img.shields.io/pypi/v/websockets.svg - :target: https://pypi.python.org/pypi/websockets - -.. |pyversions| image:: https://img.shields.io/pypi/pyversions/websockets.svg - :target: https://pypi.python.org/pypi/websockets - -.. |wheel| image:: https://img.shields.io/pypi/wheel/websockets.svg - :target: https://pypi.python.org/pypi/websockets - -.. |tests| image:: https://img.shields.io/github/checks-status/aaugustin/websockets/main - :target: https://github.com/aaugustin/websockets/actions/workflows/tests.yml - -.. |docs| image:: https://img.shields.io/readthedocs/websockets.svg - :target: https://websockets.readthedocs.io/ - -What is ``websockets``? ------------------------ - -websockets is a library for building WebSocket_ servers and clients in Python -with a focus on correctness, simplicity, robustness, and performance. - -.. _WebSocket: https://developer.mozilla.org/en-US/docs/Web/API/WebSockets_API - -Built on top of ``asyncio``, Python's standard asynchronous I/O framework, it -provides an elegant coroutine-based API. - -`Documentation is available on Read the Docs. <https://websockets.readthedocs.io/>`_ - -Here's how a client sends and receives messages: - -.. copy-pasted because GitHub doesn't support the include directive - -.. code:: python - - #!/usr/bin/env python - - import asyncio - from websockets import connect - - async def hello(uri): - async with connect(uri) as websocket: - await websocket.send("Hello world!") - await websocket.recv() - - asyncio.run(hello("ws://localhost:8765")) - -And here's an echo server: - -.. code:: python - - #!/usr/bin/env python - - import asyncio - from websockets import serve - - async def echo(websocket): - async for message in websocket: - await websocket.send(message) - - async def main(): - async with serve(echo, "localhost", 8765): - await asyncio.Future() # run forever - - asyncio.run(main()) - -Does that look good? - -`Get started with the tutorial! <https://websockets.readthedocs.io/en/stable/intro/index.html>`_ - -Why should I use ``websockets``? --------------------------------- - -The development of ``websockets`` is shaped by four principles: - -1. **Correctness**: ``websockets`` is heavily tested for compliance - with :rfc:`6455`. Continuous integration fails under 100% branch - coverage. - -2. **Simplicity**: all you need to understand is ``msg = await ws.recv()`` and - ``await ws.send(msg)``. ``websockets`` takes care of managing connections - so you can focus on your application. - -3. **Robustness**: ``websockets`` is built for production. For example, it was - the only library to `handle backpressure correctly`_ before the issue - became widely known in the Python community. - -4. **Performance**: memory usage is optimized and configurable. A C extension - accelerates expensive operations. It's pre-compiled for Linux, macOS and - Windows and packaged in the wheel format for each system and Python version. - -Documentation is a first class concern in the project. Head over to `Read the -Docs`_ and see for yourself. - -.. _Read the Docs: https://websockets.readthedocs.io/ -.. _handle backpressure correctly: https://vorpus.org/blog/some-thoughts-on-asynchronous-api-design-in-a-post-asyncawait-world/#websocket-servers - -Why shouldn't I use ``websockets``? ------------------------------------ - -* If you prefer callbacks over coroutines: ``websockets`` was created to - provide the best coroutine-based API to manage WebSocket connections in - Python. Pick another library for a callback-based API. - -* If you're looking for a mixed HTTP / WebSocket library: ``websockets`` aims - at being an excellent implementation of :rfc:`6455`: The WebSocket Protocol - and :rfc:`7692`: Compression Extensions for WebSocket. Its support for HTTP - is minimal — just enough for a HTTP health check. - - If you want to do both in the same server, look at HTTP frameworks that - build on top of ``websockets`` to support WebSocket connections, like - Sanic_. - -.. _Sanic: https://sanicframework.org/en/ - -What else? ----------- - -Bug reports, patches and suggestions are welcome! - -To report a security vulnerability, please use the `Tidelift security -contact`_. Tidelift will coordinate the fix and disclosure. - -.. _Tidelift security contact: https://tidelift.com/security - -For anything else, please open an issue_ or send a `pull request`_. - -.. _issue: https://github.com/aaugustin/websockets/issues/new -.. _pull request: https://github.com/aaugustin/websockets/compare/ - -Participants must uphold the `Contributor Covenant code of conduct`_. - -.. _Contributor Covenant code of conduct: https://github.com/aaugustin/websockets/blob/main/CODE_OF_CONDUCT.md - -``websockets`` is released under the `BSD license`_. - -.. _BSD license: https://github.com/aaugustin/websockets/blob/main/LICENSE - - diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets.egg-info/SOURCES.txt b/testing/web-platform/tests/tools/third_party/websockets/src/websockets.egg-info/SOURCES.txt deleted file mode 100644 index 2a51106fee..0000000000 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets.egg-info/SOURCES.txt +++ /dev/null @@ -1,42 +0,0 @@ -LICENSE -MANIFEST.in -README.rst -setup.cfg -setup.py -src/websockets/__init__.py -src/websockets/__main__.py -src/websockets/auth.py -src/websockets/client.py -src/websockets/connection.py -src/websockets/datastructures.py -src/websockets/exceptions.py -src/websockets/frames.py -src/websockets/headers.py -src/websockets/http.py -src/websockets/http11.py -src/websockets/imports.py -src/websockets/py.typed -src/websockets/server.py -src/websockets/speedups.c -src/websockets/streams.py -src/websockets/typing.py -src/websockets/uri.py -src/websockets/utils.py -src/websockets/version.py -src/websockets.egg-info/PKG-INFO -src/websockets.egg-info/SOURCES.txt -src/websockets.egg-info/dependency_links.txt -src/websockets.egg-info/not-zip-safe -src/websockets.egg-info/top_level.txt -src/websockets/extensions/__init__.py -src/websockets/extensions/base.py -src/websockets/extensions/permessage_deflate.py -src/websockets/legacy/__init__.py -src/websockets/legacy/auth.py -src/websockets/legacy/client.py -src/websockets/legacy/compatibility.py -src/websockets/legacy/framing.py -src/websockets/legacy/handshake.py -src/websockets/legacy/http.py -src/websockets/legacy/protocol.py -src/websockets/legacy/server.py
\ No newline at end of file diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets.egg-info/dependency_links.txt b/testing/web-platform/tests/tools/third_party/websockets/src/websockets.egg-info/dependency_links.txt deleted file mode 100644 index 8b13789179..0000000000 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets.egg-info/dependency_links.txt +++ /dev/null @@ -1 +0,0 @@ - diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets.egg-info/not-zip-safe b/testing/web-platform/tests/tools/third_party/websockets/src/websockets.egg-info/not-zip-safe deleted file mode 100644 index 8b13789179..0000000000 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets.egg-info/not-zip-safe +++ /dev/null @@ -1 +0,0 @@ - diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets.egg-info/top_level.txt b/testing/web-platform/tests/tools/third_party/websockets/src/websockets.egg-info/top_level.txt deleted file mode 100644 index 5474af7431..0000000000 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets.egg-info/top_level.txt +++ /dev/null @@ -1,3 +0,0 @@ -websockets -websockets/extensions -websockets/legacy diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/__init__.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/__init__.py index ec34841247..fdb028f4c4 100644 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/__init__.py +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/__init__.py @@ -1,23 +1,24 @@ from __future__ import annotations +import typing + from .imports import lazy_import -from .version import version as __version__ # noqa +from .version import version as __version__ # noqa: F401 -__all__ = [ # noqa +__all__ = [ + # .client + "ClientProtocol", + # .datastructures + "Headers", + "HeadersLike", + "MultipleValuesError", + # .exceptions "AbortHandshake", - "basic_auth_protocol_factory", - "BasicAuthWebSocketServerProtocol", - "broadcast", - "ClientConnection", - "connect", "ConnectionClosed", "ConnectionClosedError", "ConnectionClosedOK", - "Data", "DuplicateParameter", - "ExtensionName", - "ExtensionParameter", "InvalidHandshake", "InvalidHeader", "InvalidHeaderFormat", @@ -31,84 +32,159 @@ __all__ = [ # noqa "InvalidStatusCode", "InvalidUpgrade", "InvalidURI", - "LoggerLike", "NegotiationError", - "Origin", - "parse_uri", "PayloadTooBig", "ProtocolError", "RedirectHandshake", "SecurityError", - "serve", - "ServerConnection", - "Subprotocol", - "unix_connect", - "unix_serve", - "WebSocketClientProtocol", - "WebSocketCommonProtocol", "WebSocketException", "WebSocketProtocolError", + # .legacy.auth + "BasicAuthWebSocketServerProtocol", + "basic_auth_protocol_factory", + # .legacy.client + "WebSocketClientProtocol", + "connect", + "unix_connect", + # .legacy.protocol + "WebSocketCommonProtocol", + "broadcast", + # .legacy.server "WebSocketServer", "WebSocketServerProtocol", - "WebSocketURI", + "serve", + "unix_serve", + # .server + "ServerProtocol", + # .typing + "Data", + "ExtensionName", + "ExtensionParameter", + "LoggerLike", + "StatusLike", + "Origin", + "Subprotocol", ] -lazy_import( - globals(), - aliases={ - "auth": ".legacy", - "basic_auth_protocol_factory": ".legacy.auth", - "BasicAuthWebSocketServerProtocol": ".legacy.auth", - "broadcast": ".legacy.protocol", - "ClientConnection": ".client", - "connect": ".legacy.client", - "unix_connect": ".legacy.client", - "WebSocketClientProtocol": ".legacy.client", - "Headers": ".datastructures", - "MultipleValuesError": ".datastructures", - "WebSocketException": ".exceptions", - "ConnectionClosed": ".exceptions", - "ConnectionClosedError": ".exceptions", - "ConnectionClosedOK": ".exceptions", - "InvalidHandshake": ".exceptions", - "SecurityError": ".exceptions", - "InvalidMessage": ".exceptions", - "InvalidHeader": ".exceptions", - "InvalidHeaderFormat": ".exceptions", - "InvalidHeaderValue": ".exceptions", - "InvalidOrigin": ".exceptions", - "InvalidUpgrade": ".exceptions", - "InvalidStatus": ".exceptions", - "InvalidStatusCode": ".exceptions", - "NegotiationError": ".exceptions", - "DuplicateParameter": ".exceptions", - "InvalidParameterName": ".exceptions", - "InvalidParameterValue": ".exceptions", - "AbortHandshake": ".exceptions", - "RedirectHandshake": ".exceptions", - "InvalidState": ".exceptions", - "InvalidURI": ".exceptions", - "PayloadTooBig": ".exceptions", - "ProtocolError": ".exceptions", - "WebSocketProtocolError": ".exceptions", - "protocol": ".legacy", - "WebSocketCommonProtocol": ".legacy.protocol", - "ServerConnection": ".server", - "serve": ".legacy.server", - "unix_serve": ".legacy.server", - "WebSocketServerProtocol": ".legacy.server", - "WebSocketServer": ".legacy.server", - "Data": ".typing", - "LoggerLike": ".typing", - "Origin": ".typing", - "ExtensionHeader": ".typing", - "ExtensionParameter": ".typing", - "Subprotocol": ".typing", - }, - deprecated_aliases={ - "framing": ".legacy", - "handshake": ".legacy", - "parse_uri": ".uri", - "WebSocketURI": ".uri", - }, -) +# When type checking, import non-deprecated aliases eagerly. Else, import on demand. +if typing.TYPE_CHECKING: + from .client import ClientProtocol + from .datastructures import Headers, HeadersLike, MultipleValuesError + from .exceptions import ( + AbortHandshake, + ConnectionClosed, + ConnectionClosedError, + ConnectionClosedOK, + DuplicateParameter, + InvalidHandshake, + InvalidHeader, + InvalidHeaderFormat, + InvalidHeaderValue, + InvalidMessage, + InvalidOrigin, + InvalidParameterName, + InvalidParameterValue, + InvalidState, + InvalidStatus, + InvalidStatusCode, + InvalidUpgrade, + InvalidURI, + NegotiationError, + PayloadTooBig, + ProtocolError, + RedirectHandshake, + SecurityError, + WebSocketException, + WebSocketProtocolError, + ) + from .legacy.auth import ( + BasicAuthWebSocketServerProtocol, + basic_auth_protocol_factory, + ) + from .legacy.client import WebSocketClientProtocol, connect, unix_connect + from .legacy.protocol import WebSocketCommonProtocol, broadcast + from .legacy.server import ( + WebSocketServer, + WebSocketServerProtocol, + serve, + unix_serve, + ) + from .server import ServerProtocol + from .typing import ( + Data, + ExtensionName, + ExtensionParameter, + LoggerLike, + Origin, + StatusLike, + Subprotocol, + ) +else: + lazy_import( + globals(), + aliases={ + # .client + "ClientProtocol": ".client", + # .datastructures + "Headers": ".datastructures", + "HeadersLike": ".datastructures", + "MultipleValuesError": ".datastructures", + # .exceptions + "AbortHandshake": ".exceptions", + "ConnectionClosed": ".exceptions", + "ConnectionClosedError": ".exceptions", + "ConnectionClosedOK": ".exceptions", + "DuplicateParameter": ".exceptions", + "InvalidHandshake": ".exceptions", + "InvalidHeader": ".exceptions", + "InvalidHeaderFormat": ".exceptions", + "InvalidHeaderValue": ".exceptions", + "InvalidMessage": ".exceptions", + "InvalidOrigin": ".exceptions", + "InvalidParameterName": ".exceptions", + "InvalidParameterValue": ".exceptions", + "InvalidState": ".exceptions", + "InvalidStatus": ".exceptions", + "InvalidStatusCode": ".exceptions", + "InvalidUpgrade": ".exceptions", + "InvalidURI": ".exceptions", + "NegotiationError": ".exceptions", + "PayloadTooBig": ".exceptions", + "ProtocolError": ".exceptions", + "RedirectHandshake": ".exceptions", + "SecurityError": ".exceptions", + "WebSocketException": ".exceptions", + "WebSocketProtocolError": ".exceptions", + # .legacy.auth + "BasicAuthWebSocketServerProtocol": ".legacy.auth", + "basic_auth_protocol_factory": ".legacy.auth", + # .legacy.client + "WebSocketClientProtocol": ".legacy.client", + "connect": ".legacy.client", + "unix_connect": ".legacy.client", + # .legacy.protocol + "WebSocketCommonProtocol": ".legacy.protocol", + "broadcast": ".legacy.protocol", + # .legacy.server + "WebSocketServer": ".legacy.server", + "WebSocketServerProtocol": ".legacy.server", + "serve": ".legacy.server", + "unix_serve": ".legacy.server", + # .server + "ServerProtocol": ".server", + # .typing + "Data": ".typing", + "ExtensionName": ".typing", + "ExtensionParameter": ".typing", + "LoggerLike": ".typing", + "Origin": ".typing", + "StatusLike": "typing", + "Subprotocol": ".typing", + }, + deprecated_aliases={ + "framing": ".legacy", + "handshake": ".legacy", + "parse_uri": ".uri", + "WebSocketURI": ".uri", + }, + ) diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/__main__.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/__main__.py index c562d21b54..f2ea5cf4e8 100644 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/__main__.py +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/__main__.py @@ -1,16 +1,18 @@ from __future__ import annotations import argparse -import asyncio import os import signal import sys import threading -from typing import Any, Set -from .exceptions import ConnectionClosed -from .frames import Close -from .legacy.client import connect + +try: + import readline # noqa: F401 +except ImportError: # Windows has no `readline` normally + pass + +from .sync.client import ClientConnection, connect from .version import version as websockets_version @@ -46,21 +48,6 @@ if sys.platform == "win32": raise RuntimeError("unable to set console mode") -def exit_from_event_loop_thread( - loop: asyncio.AbstractEventLoop, - stop: asyncio.Future[None], -) -> None: - loop.stop() - if not stop.done(): - # When exiting the thread that runs the event loop, raise - # KeyboardInterrupt in the main thread to exit the program. - if sys.platform == "win32": - ctrl_c = signal.CTRL_C_EVENT - else: - ctrl_c = signal.SIGINT - os.kill(os.getpid(), ctrl_c) - - def print_during_input(string: str) -> None: sys.stdout.write( # Save cursor position @@ -93,63 +80,20 @@ def print_over_input(string: str) -> None: sys.stdout.flush() -async def run_client( - uri: str, - loop: asyncio.AbstractEventLoop, - inputs: asyncio.Queue[str], - stop: asyncio.Future[None], -) -> None: - try: - websocket = await connect(uri) - except Exception as exc: - print_over_input(f"Failed to connect to {uri}: {exc}.") - exit_from_event_loop_thread(loop, stop) - return - else: - print_during_input(f"Connected to {uri}.") - - try: - while True: - incoming: asyncio.Future[Any] = asyncio.create_task(websocket.recv()) - outgoing: asyncio.Future[Any] = asyncio.create_task(inputs.get()) - done: Set[asyncio.Future[Any]] - pending: Set[asyncio.Future[Any]] - done, pending = await asyncio.wait( - [incoming, outgoing, stop], return_when=asyncio.FIRST_COMPLETED - ) - - # Cancel pending tasks to avoid leaking them. - if incoming in pending: - incoming.cancel() - if outgoing in pending: - outgoing.cancel() - - if incoming in done: - try: - message = incoming.result() - except ConnectionClosed: - break - else: - if isinstance(message, str): - print_during_input("< " + message) - else: - print_during_input("< (binary) " + message.hex()) - - if outgoing in done: - message = outgoing.result() - await websocket.send(message) - - if stop in done: - break - - finally: - await websocket.close() - assert websocket.close_code is not None and websocket.close_reason is not None - close_status = Close(websocket.close_code, websocket.close_reason) - - print_over_input(f"Connection closed: {close_status}.") - - exit_from_event_loop_thread(loop, stop) +def print_incoming_messages(websocket: ClientConnection, stop: threading.Event) -> None: + for message in websocket: + if isinstance(message, str): + print_during_input("< " + message) + else: + print_during_input("< (binary) " + message.hex()) + if not stop.is_set(): + # When the server closes the connection, raise KeyboardInterrupt + # in the main thread to exit the program. + if sys.platform == "win32": + ctrl_c = signal.CTRL_C_EVENT + else: + ctrl_c = signal.SIGINT + os.kill(os.getpid(), ctrl_c) def main() -> None: @@ -184,29 +128,17 @@ def main() -> None: sys.stderr.flush() try: - import readline # noqa - except ImportError: # Windows has no `readline` normally - pass - - # Create an event loop that will run in a background thread. - loop = asyncio.new_event_loop() - - # Due to zealous removal of the loop parameter in the Queue constructor, - # we need a factory coroutine to run in the freshly created event loop. - async def queue_factory() -> asyncio.Queue[str]: - return asyncio.Queue() - - # Create a queue of user inputs. There's no need to limit its size. - inputs: asyncio.Queue[str] = loop.run_until_complete(queue_factory()) - - # Create a stop condition when receiving SIGINT or SIGTERM. - stop: asyncio.Future[None] = loop.create_future() + websocket = connect(args.uri) + except Exception as exc: + print(f"Failed to connect to {args.uri}: {exc}.") + sys.exit(1) + else: + print(f"Connected to {args.uri}.") - # Schedule the task that will manage the connection. - loop.create_task(run_client(args.uri, loop, inputs, stop)) + stop = threading.Event() - # Start the event loop in a background thread. - thread = threading.Thread(target=loop.run_forever) + # Start the thread that reads messages from the connection. + thread = threading.Thread(target=print_incoming_messages, args=(websocket, stop)) thread.start() # Read from stdin in the main thread in order to receive signals. @@ -214,17 +146,14 @@ def main() -> None: while True: # Since there's no size limit, put_nowait is identical to put. message = input("> ") - loop.call_soon_threadsafe(inputs.put_nowait, message) + websocket.send(message) except (KeyboardInterrupt, EOFError): # ^C, ^D - loop.call_soon_threadsafe(stop.set_result, None) + stop.set() + websocket.close() + print_over_input("Connection closed.") - # Wait for the event loop to terminate. thread.join() - # For reasons unclear, even though the loop is closed in the thread, - # it still thinks it's running here. - loop.close() - if __name__ == "__main__": main() diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/auth.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/auth.py index afcb38cffe..b792e02f5c 100644 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/auth.py +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/auth.py @@ -1,4 +1,6 @@ from __future__ import annotations # See #940 for why lazy_import isn't used here for backwards compatibility. -from .legacy.auth import * # noqa +# See #1400 for why listing compatibility imports in __all__ helps PyCharm. +from .legacy.auth import * +from .legacy.auth import __all__ # noqa: F401 diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/client.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/client.py index df8e53429a..b2f622042d 100644 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/client.py +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/client.py @@ -1,8 +1,8 @@ from __future__ import annotations -from typing import Generator, List, Optional, Sequence +import warnings +from typing import Any, Generator, List, Optional, Sequence -from .connection import CLIENT, CONNECTING, OPEN, Connection, State from .datastructures import Headers, MultipleValuesError from .exceptions import ( InvalidHandshake, @@ -23,8 +23,8 @@ from .headers import ( parse_subprotocol, parse_upgrade, ) -from .http import USER_AGENT from .http11 import Request, Response +from .protocol import CLIENT, CONNECTING, OPEN, Protocol, State from .typing import ( ConnectionOption, ExtensionHeader, @@ -38,13 +38,15 @@ from .utils import accept_key, generate_key # See #940 for why lazy_import isn't used here for backwards compatibility. -from .legacy.client import * # isort:skip # noqa +# See #1400 for why listing compatibility imports in __all__ helps PyCharm. +from .legacy.client import * # isort:skip # noqa: I001 +from .legacy.client import __all__ as legacy__all__ -__all__ = ["ClientConnection"] +__all__ = ["ClientProtocol"] + legacy__all__ -class ClientConnection(Connection): +class ClientProtocol(Protocol): """ Sans-I/O implementation of a WebSocket client connection. @@ -60,16 +62,17 @@ class ClientConnection(Connection): preference. state: initial state of the WebSocket connection. max_size: maximum size of incoming messages in bytes; - :obj:`None` to disable the limit. + :obj:`None` disables the limit. logger: logger for this connection; defaults to ``logging.getLogger("websockets.client")``; - see the :doc:`logging guide <../topics/logging>` for details. + see the :doc:`logging guide <../../topics/logging>` for details. """ def __init__( self, wsuri: WebSocketURI, + *, origin: Optional[Origin] = None, extensions: Optional[Sequence[ClientExtensionFactory]] = None, subprotocols: Optional[Sequence[Subprotocol]] = None, @@ -89,7 +92,7 @@ class ClientConnection(Connection): self.available_subprotocols = subprotocols self.key = generate_key() - def connect(self) -> Request: # noqa: F811 + def connect(self) -> Request: """ Create a handshake request to open a connection. @@ -131,8 +134,6 @@ class ClientConnection(Connection): protocol_header = build_subprotocol(self.available_subprotocols) headers["Sec-WebSocket-Protocol"] = protocol_header - headers["User-Agent"] = USER_AGENT - return Request(self.wsuri.resource_name, headers) def process_response(self, response: Response) -> None: @@ -223,7 +224,6 @@ class ClientConnection(Connection): extensions = headers.get_all("Sec-WebSocket-Extensions") if extensions: - if self.available_extensions is None: raise InvalidHandshake("no extensions supported") @@ -232,9 +232,7 @@ class ClientConnection(Connection): ) for name, response_params in parsed_extensions: - for extension_factory in self.available_extensions: - # Skip non-matching extensions based on their name. if extension_factory.name != name: continue @@ -281,7 +279,6 @@ class ClientConnection(Connection): subprotocols = headers.get_all("Sec-WebSocket-Protocol") if subprotocols: - if self.available_subprotocols is None: raise InvalidHandshake("no subprotocols supported") @@ -317,11 +314,17 @@ class ClientConnection(Connection): def parse(self) -> Generator[None, None, None]: if self.state is CONNECTING: - response = yield from Response.parse( - self.reader.read_line, - self.reader.read_exact, - self.reader.read_to_eof, - ) + try: + response = yield from Response.parse( + self.reader.read_line, + self.reader.read_exact, + self.reader.read_to_eof, + ) + except Exception as exc: + self.handshake_exc = exc + self.parser = self.discard() + next(self.parser) # start coroutine + yield if self.debug: code, phrase = response.status_code, response.reason_phrase @@ -335,13 +338,23 @@ class ClientConnection(Connection): self.process_response(response) except InvalidHandshake as exc: response._exception = exc + self.events.append(response) self.handshake_exc = exc self.parser = self.discard() next(self.parser) # start coroutine - else: - assert self.state is CONNECTING - self.state = OPEN - finally: - self.events.append(response) + yield + + assert self.state is CONNECTING + self.state = OPEN + self.events.append(response) yield from super().parse() + + +class ClientConnection(ClientProtocol): + def __init__(self, *args: Any, **kwargs: Any) -> None: + warnings.warn( + "ClientConnection was renamed to ClientProtocol", + DeprecationWarning, + ) + super().__init__(*args, **kwargs) diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/connection.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/connection.py index db8b536993..88bcda1aaf 100644 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/connection.py +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/connection.py @@ -1,702 +1,13 @@ from __future__ import annotations -import enum -import logging -import uuid -from typing import Generator, List, Optional, Type, Union +import warnings -from .exceptions import ( - ConnectionClosed, - ConnectionClosedError, - ConnectionClosedOK, - InvalidState, - PayloadTooBig, - ProtocolError, -) -from .extensions import Extension -from .frames import ( - OK_CLOSE_CODES, - OP_BINARY, - OP_CLOSE, - OP_CONT, - OP_PING, - OP_PONG, - OP_TEXT, - Close, - Frame, -) -from .http11 import Request, Response -from .streams import StreamReader -from .typing import LoggerLike, Origin, Subprotocol - - -__all__ = [ - "Connection", - "Side", - "State", - "SEND_EOF", -] - -Event = Union[Request, Response, Frame] -"""Events that :meth:`~Connection.events_received` may return.""" - - -class Side(enum.IntEnum): - """A WebSocket connection is either a server or a client.""" - - SERVER, CLIENT = range(2) - - -SERVER = Side.SERVER -CLIENT = Side.CLIENT - - -class State(enum.IntEnum): - """A WebSocket connection is in one of these four states.""" - - CONNECTING, OPEN, CLOSING, CLOSED = range(4) - - -CONNECTING = State.CONNECTING -OPEN = State.OPEN -CLOSING = State.CLOSING -CLOSED = State.CLOSED - - -SEND_EOF = b"" -"""Sentinel signaling that the TCP connection must be half-closed.""" - - -class Connection: - """ - Sans-I/O implementation of a WebSocket connection. - - Args: - side: :attr:`~Side.CLIENT` or :attr:`~Side.SERVER`. - state: initial state of the WebSocket connection. - max_size: maximum size of incoming messages in bytes; - :obj:`None` to disable the limit. - logger: logger for this connection; depending on ``side``, - defaults to ``logging.getLogger("websockets.client")`` - or ``logging.getLogger("websockets.server")``; - see the :doc:`logging guide <../topics/logging>` for details. - - """ - - def __init__( - self, - side: Side, - state: State = OPEN, - max_size: Optional[int] = 2**20, - logger: Optional[LoggerLike] = None, - ) -> None: - # Unique identifier. For logs. - self.id: uuid.UUID = uuid.uuid4() - """Unique identifier of the connection. Useful in logs.""" - - # Logger or LoggerAdapter for this connection. - if logger is None: - logger = logging.getLogger(f"websockets.{side.name.lower()}") - self.logger: LoggerLike = logger - """Logger for this connection.""" - - # Track if DEBUG is enabled. Shortcut logging calls if it isn't. - self.debug = logger.isEnabledFor(logging.DEBUG) - - # Connection side. CLIENT or SERVER. - self.side = side - - # Connection state. Initially OPEN because subclasses handle CONNECTING. - self.state = state - - # Maximum size of incoming messages in bytes. - self.max_size = max_size - - # Current size of incoming message in bytes. Only set while reading a - # fragmented message i.e. a data frames with the FIN bit not set. - self.cur_size: Optional[int] = None - - # True while sending a fragmented message i.e. a data frames with the - # FIN bit not set. - self.expect_continuation_frame = False - - # WebSocket protocol parameters. - self.origin: Optional[Origin] = None - self.extensions: List[Extension] = [] - self.subprotocol: Optional[Subprotocol] = None - - # Close code and reason, set when a close frame is sent or received. - self.close_rcvd: Optional[Close] = None - self.close_sent: Optional[Close] = None - self.close_rcvd_then_sent: Optional[bool] = None - - # Track if an exception happened during the handshake. - self.handshake_exc: Optional[Exception] = None - """ - Exception to raise if the opening handshake failed. - - :obj:`None` if the opening handshake succeeded. - - """ - - # Track if send_eof() was called. - self.eof_sent = False - - # Parser state. - self.reader = StreamReader() - self.events: List[Event] = [] - self.writes: List[bytes] = [] - self.parser = self.parse() - next(self.parser) # start coroutine - self.parser_exc: Optional[Exception] = None - - @property - def state(self) -> State: - """ - WebSocket connection state. - - Defined in 4.1, 4.2, 7.1.3, and 7.1.4 of :rfc:`6455`. - - """ - return self._state - - @state.setter - def state(self, state: State) -> None: - if self.debug: - self.logger.debug("= connection is %s", state.name) - self._state = state - - @property - def close_code(self) -> Optional[int]: - """ - `WebSocket close code`_. - - .. _WebSocket close code: - https://www.rfc-editor.org/rfc/rfc6455.html#section-7.1.5 - - :obj:`None` if the connection isn't closed yet. - - """ - if self.state is not CLOSED: - return None - elif self.close_rcvd is None: - return 1006 - else: - return self.close_rcvd.code - - @property - def close_reason(self) -> Optional[str]: - """ - `WebSocket close reason`_. - - .. _WebSocket close reason: - https://www.rfc-editor.org/rfc/rfc6455.html#section-7.1.6 - - :obj:`None` if the connection isn't closed yet. - - """ - if self.state is not CLOSED: - return None - elif self.close_rcvd is None: - return "" - else: - return self.close_rcvd.reason - - @property - def close_exc(self) -> ConnectionClosed: - """ - Exception to raise when trying to interact with a closed connection. - - Don't raise this exception while the connection :attr:`state` - is :attr:`~websockets.connection.State.CLOSING`; wait until - it's :attr:`~websockets.connection.State.CLOSED`. - - Indeed, the exception includes the close code and reason, which are - known only once the connection is closed. - - Raises: - AssertionError: if the connection isn't closed yet. - - """ - assert self.state is CLOSED, "connection isn't closed yet" - exc_type: Type[ConnectionClosed] - if ( - self.close_rcvd is not None - and self.close_sent is not None - and self.close_rcvd.code in OK_CLOSE_CODES - and self.close_sent.code in OK_CLOSE_CODES - ): - exc_type = ConnectionClosedOK - else: - exc_type = ConnectionClosedError - exc: ConnectionClosed = exc_type( - self.close_rcvd, - self.close_sent, - self.close_rcvd_then_sent, - ) - # Chain to the exception raised in the parser, if any. - exc.__cause__ = self.parser_exc - return exc - - # Public methods for receiving data. - - def receive_data(self, data: bytes) -> None: - """ - Receive data from the network. - - After calling this method: - - - You must call :meth:`data_to_send` and send this data to the network. - - You should call :meth:`events_received` and process resulting events. - - Raises: - EOFError: if :meth:`receive_eof` was called earlier. - - """ - self.reader.feed_data(data) - next(self.parser) - - def receive_eof(self) -> None: - """ - Receive the end of the data stream from the network. - - After calling this method: - - - You must call :meth:`data_to_send` and send this data to the network. - - You aren't expected to call :meth:`events_received`; it won't return - any new events. - - Raises: - EOFError: if :meth:`receive_eof` was called earlier. - - """ - self.reader.feed_eof() - next(self.parser) - - # Public methods for sending events. - - def send_continuation(self, data: bytes, fin: bool) -> None: - """ - Send a `Continuation frame`_. - - .. _Continuation frame: - https://datatracker.ietf.org/doc/html/rfc6455#section-5.6 - - Parameters: - data: payload containing the same kind of data - as the initial frame. - fin: FIN bit; set it to :obj:`True` if this is the last frame - of a fragmented message and to :obj:`False` otherwise. - - Raises: - ProtocolError: if a fragmented message isn't in progress. - - """ - if not self.expect_continuation_frame: - raise ProtocolError("unexpected continuation frame") - self.expect_continuation_frame = not fin - self.send_frame(Frame(OP_CONT, data, fin)) - - def send_text(self, data: bytes, fin: bool = True) -> None: - """ - Send a `Text frame`_. - - .. _Text frame: - https://datatracker.ietf.org/doc/html/rfc6455#section-5.6 - - Parameters: - data: payload containing text encoded with UTF-8. - fin: FIN bit; set it to :obj:`False` if this is the first frame of - a fragmented message. - - Raises: - ProtocolError: if a fragmented message is in progress. - - """ - if self.expect_continuation_frame: - raise ProtocolError("expected a continuation frame") - self.expect_continuation_frame = not fin - self.send_frame(Frame(OP_TEXT, data, fin)) - - def send_binary(self, data: bytes, fin: bool = True) -> None: - """ - Send a `Binary frame`_. +# lazy_import doesn't support this use case. +from .protocol import SEND_EOF, Protocol as Connection, Side, State # noqa: F401 - .. _Binary frame: - https://datatracker.ietf.org/doc/html/rfc6455#section-5.6 - Parameters: - data: payload containing arbitrary binary data. - fin: FIN bit; set it to :obj:`False` if this is the first frame of - a fragmented message. - - Raises: - ProtocolError: if a fragmented message is in progress. - - """ - if self.expect_continuation_frame: - raise ProtocolError("expected a continuation frame") - self.expect_continuation_frame = not fin - self.send_frame(Frame(OP_BINARY, data, fin)) - - def send_close(self, code: Optional[int] = None, reason: str = "") -> None: - """ - Send a `Close frame`_. - - .. _Close frame: - https://datatracker.ietf.org/doc/html/rfc6455#section-5.5.1 - - Parameters: - code: close code. - reason: close reason. - - Raises: - ProtocolError: if a fragmented message is being sent, if the code - isn't valid, or if a reason is provided without a code - - """ - if self.expect_continuation_frame: - raise ProtocolError("expected a continuation frame") - if code is None: - if reason != "": - raise ProtocolError("cannot send a reason without a code") - close = Close(1005, "") - data = b"" - else: - close = Close(code, reason) - data = close.serialize() - # send_frame() guarantees that self.state is OPEN at this point. - # 7.1.3. The WebSocket Closing Handshake is Started - self.send_frame(Frame(OP_CLOSE, data)) - self.close_sent = close - self.state = CLOSING - - def send_ping(self, data: bytes) -> None: - """ - Send a `Ping frame`_. - - .. _Ping frame: - https://datatracker.ietf.org/doc/html/rfc6455#section-5.5.2 - - Parameters: - data: payload containing arbitrary binary data. - - """ - self.send_frame(Frame(OP_PING, data)) - - def send_pong(self, data: bytes) -> None: - """ - Send a `Pong frame`_. - - .. _Pong frame: - https://datatracker.ietf.org/doc/html/rfc6455#section-5.5.3 - - Parameters: - data: payload containing arbitrary binary data. - - """ - self.send_frame(Frame(OP_PONG, data)) - - def fail(self, code: int, reason: str = "") -> None: - """ - `Fail the WebSocket connection`_. - - .. _Fail the WebSocket connection: - https://datatracker.ietf.org/doc/html/rfc6455#section-7.1.7 - - Parameters: - code: close code - reason: close reason - - Raises: - ProtocolError: if the code isn't valid. - """ - # 7.1.7. Fail the WebSocket Connection - - # Send a close frame when the state is OPEN (a close frame was already - # sent if it's CLOSING), except when failing the connection because - # of an error reading from or writing to the network. - if self.state is OPEN: - if code != 1006: - close = Close(code, reason) - data = close.serialize() - self.send_frame(Frame(OP_CLOSE, data)) - self.close_sent = close - self.state = CLOSING - - # When failing the connection, a server closes the TCP connection - # without waiting for the client to complete the handshake, while a - # client waits for the server to close the TCP connection, possibly - # after sending a close frame that the client will ignore. - if self.side is SERVER and not self.eof_sent: - self.send_eof() - - # 7.1.7. Fail the WebSocket Connection "An endpoint MUST NOT continue - # to attempt to process data(including a responding Close frame) from - # the remote endpoint after being instructed to _Fail the WebSocket - # Connection_." - self.parser = self.discard() - next(self.parser) # start coroutine - - # Public method for getting incoming events after receiving data. - - def events_received(self) -> List[Event]: - """ - Fetch events generated from data received from the network. - - Call this method immediately after any of the ``receive_*()`` methods. - - Process resulting events, likely by passing them to the application. - - Returns: - List[Event]: Events read from the connection. - """ - events, self.events = self.events, [] - return events - - # Public method for getting outgoing data after receiving data or sending events. - - def data_to_send(self) -> List[bytes]: - """ - Obtain data to send to the network. - - Call this method immediately after any of the ``receive_*()``, - ``send_*()``, or :meth:`fail` methods. - - Write resulting data to the connection. - - The empty bytestring :data:`~websockets.connection.SEND_EOF` signals - the end of the data stream. When you receive it, half-close the TCP - connection. - - Returns: - List[bytes]: Data to write to the connection. - - """ - writes, self.writes = self.writes, [] - return writes - - def close_expected(self) -> bool: - """ - Tell if the TCP connection is expected to close soon. - - Call this method immediately after any of the ``receive_*()`` or - :meth:`fail` methods. - - If it returns :obj:`True`, schedule closing the TCP connection after a - short timeout if the other side hasn't already closed it. - - Returns: - bool: Whether the TCP connection is expected to close soon. - - """ - # We expect a TCP close if and only if we sent a close frame: - # * Normal closure: once we send a close frame, we expect a TCP close: - # server waits for client to complete the TCP closing handshake; - # client waits for server to initiate the TCP closing handshake. - # * Abnormal closure: we always send a close frame and the same logic - # applies, except on EOFError where we don't send a close frame - # because we already received the TCP close, so we don't expect it. - # We already got a TCP Close if and only if the state is CLOSED. - return self.state is CLOSING or self.handshake_exc is not None - - # Private methods for receiving data. - - def parse(self) -> Generator[None, None, None]: - """ - Parse incoming data into frames. - - :meth:`receive_data` and :meth:`receive_eof` run this generator - coroutine until it needs more data or reaches EOF. - - """ - try: - while True: - if (yield from self.reader.at_eof()): - if self.debug: - self.logger.debug("< EOF") - # If the WebSocket connection is closed cleanly, with a - # closing handhshake, recv_frame() substitutes parse() - # with discard(). This branch is reached only when the - # connection isn't closed cleanly. - raise EOFError("unexpected end of stream") - - if self.max_size is None: - max_size = None - elif self.cur_size is None: - max_size = self.max_size - else: - max_size = self.max_size - self.cur_size - - # During a normal closure, execution ends here on the next - # iteration of the loop after receiving a close frame. At - # this point, recv_frame() replaced parse() by discard(). - frame = yield from Frame.parse( - self.reader.read_exact, - mask=self.side is SERVER, - max_size=max_size, - extensions=self.extensions, - ) - - if self.debug: - self.logger.debug("< %s", frame) - - self.recv_frame(frame) - - except ProtocolError as exc: - self.fail(1002, str(exc)) - self.parser_exc = exc - - except EOFError as exc: - self.fail(1006, str(exc)) - self.parser_exc = exc - - except UnicodeDecodeError as exc: - self.fail(1007, f"{exc.reason} at position {exc.start}") - self.parser_exc = exc - - except PayloadTooBig as exc: - self.fail(1009, str(exc)) - self.parser_exc = exc - - except Exception as exc: - self.logger.error("parser failed", exc_info=True) - # Don't include exception details, which may be security-sensitive. - self.fail(1011) - self.parser_exc = exc - - # During an abnormal closure, execution ends here after catching an - # exception. At this point, fail() replaced parse() by discard(). - yield - raise AssertionError("parse() shouldn't step after error") # pragma: no cover - - def discard(self) -> Generator[None, None, None]: - """ - Discard incoming data. - - This coroutine replaces :meth:`parse`: - - - after receiving a close frame, during a normal closure (1.4); - - after sending a close frame, during an abnormal closure (7.1.7). - - """ - # The server close the TCP connection in the same circumstances where - # discard() replaces parse(). The client closes the connection later, - # after the server closes the connection or a timeout elapses. - # (The latter case cannot be handled in this Sans-I/O layer.) - assert (self.side is SERVER) == (self.eof_sent) - while not (yield from self.reader.at_eof()): - self.reader.discard() - if self.debug: - self.logger.debug("< EOF") - # A server closes the TCP connection immediately, while a client - # waits for the server to close the TCP connection. - if self.side is CLIENT: - self.send_eof() - self.state = CLOSED - # If discard() completes normally, execution ends here. - yield - # Once the reader reaches EOF, its feed_data/eof() methods raise an - # error, so our receive_data/eof() methods don't step the generator. - raise AssertionError("discard() shouldn't step after EOF") # pragma: no cover - - def recv_frame(self, frame: Frame) -> None: - """ - Process an incoming frame. - - """ - if frame.opcode is OP_TEXT or frame.opcode is OP_BINARY: - if self.cur_size is not None: - raise ProtocolError("expected a continuation frame") - if frame.fin: - self.cur_size = None - else: - self.cur_size = len(frame.data) - - elif frame.opcode is OP_CONT: - if self.cur_size is None: - raise ProtocolError("unexpected continuation frame") - if frame.fin: - self.cur_size = None - else: - self.cur_size += len(frame.data) - - elif frame.opcode is OP_PING: - # 5.5.2. Ping: "Upon receipt of a Ping frame, an endpoint MUST - # send a Pong frame in response" - pong_frame = Frame(OP_PONG, frame.data) - self.send_frame(pong_frame) - - elif frame.opcode is OP_PONG: - # 5.5.3 Pong: "A response to an unsolicited Pong frame is not - # expected." - pass - - elif frame.opcode is OP_CLOSE: - # 7.1.5. The WebSocket Connection Close Code - # 7.1.6. The WebSocket Connection Close Reason - self.close_rcvd = Close.parse(frame.data) - if self.state is CLOSING: - assert self.close_sent is not None - self.close_rcvd_then_sent = False - - if self.cur_size is not None: - raise ProtocolError("incomplete fragmented message") - - # 5.5.1 Close: "If an endpoint receives a Close frame and did - # not previously send a Close frame, the endpoint MUST send a - # Close frame in response. (When sending a Close frame in - # response, the endpoint typically echos the status code it - # received.)" - - if self.state is OPEN: - # Echo the original data instead of re-serializing it with - # Close.serialize() because that fails when the close frame - # is empty and Close.parse() synthetizes a 1005 close code. - # The rest is identical to send_close(). - self.send_frame(Frame(OP_CLOSE, frame.data)) - self.close_sent = self.close_rcvd - self.close_rcvd_then_sent = True - self.state = CLOSING - - # 7.1.2. Start the WebSocket Closing Handshake: "Once an - # endpoint has both sent and received a Close control frame, - # that endpoint SHOULD _Close the WebSocket Connection_" - - # A server closes the TCP connection immediately, while a client - # waits for the server to close the TCP connection. - if self.side is SERVER: - self.send_eof() - - # 1.4. Closing Handshake: "after receiving a control frame - # indicating the connection should be closed, a peer discards - # any further data received." - self.parser = self.discard() - next(self.parser) # start coroutine - - else: # pragma: no cover - # This can't happen because Frame.parse() validates opcodes. - raise AssertionError(f"unexpected opcode: {frame.opcode:02x}") - - self.events.append(frame) - - # Private methods for sending events. - - def send_frame(self, frame: Frame) -> None: - if self.state is not OPEN: - raise InvalidState( - f"cannot write to a WebSocket in the {self.state.name} state" - ) - - if self.debug: - self.logger.debug("> %s", frame) - self.writes.append( - frame.serialize(mask=self.side is CLIENT, extensions=self.extensions) - ) - - def send_eof(self) -> None: - assert not self.eof_sent - self.eof_sent = True - if self.debug: - self.logger.debug("> EOF") - self.writes.append(SEND_EOF) +warnings.warn( + "websockets.connection was renamed to websockets.protocol " + "and Connection was renamed to Protocol", + DeprecationWarning, +) diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/datastructures.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/datastructures.py index 36a2cbaf99..a0a648463a 100644 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/datastructures.py +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/datastructures.py @@ -1,6 +1,5 @@ from __future__ import annotations -import sys from typing import ( Any, Dict, @@ -9,17 +8,12 @@ from typing import ( List, Mapping, MutableMapping, + Protocol, Tuple, Union, ) -if sys.version_info[:2] >= (3, 8): - from typing import Protocol -else: # pragma: no cover - Protocol = object # mypy will report errors on Python 3.7. - - __all__ = ["Headers", "HeadersLike", "MultipleValuesError"] diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/exceptions.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/exceptions.py index 0c4fc51851..f7169e3b17 100644 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/exceptions.py +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/exceptions.py @@ -34,6 +34,7 @@ import http from typing import Optional from . import datastructures, frames, http11 +from .typing import StatusLike __all__ = [ @@ -120,19 +121,23 @@ class ConnectionClosed(WebSocketException): @property def code(self) -> int: - return 1006 if self.rcvd is None else self.rcvd.code + if self.rcvd is None: + return frames.CloseCode.ABNORMAL_CLOSURE + return self.rcvd.code @property def reason(self) -> str: - return "" if self.rcvd is None else self.rcvd.reason + if self.rcvd is None: + return "" + return self.rcvd.reason class ConnectionClosedError(ConnectionClosed): """ Like :exc:`ConnectionClosed`, when the connection terminated with an error. - A close code other than 1000 (OK) or 1001 (going away) was received or - sent, or the closing handshake didn't complete properly. + A close frame with a code other than 1000 (OK) or 1001 (going away) was + received or sent, or the closing handshake didn't complete properly. """ @@ -141,7 +146,8 @@ class ConnectionClosedOK(ConnectionClosed): """ Like :exc:`ConnectionClosed`, when the connection terminated properly. - A close code 1000 (OK) or 1001 (going away) was received and sent. + A close code with code 1000 (OK) or 1001 (going away) or without a code was + received and sent. """ @@ -171,7 +177,7 @@ class InvalidMessage(InvalidHandshake): class InvalidHeader(InvalidHandshake): """ - Raised when a HTTP header doesn't have a valid format or value. + Raised when an HTTP header doesn't have a valid format or value. """ @@ -190,7 +196,7 @@ class InvalidHeader(InvalidHandshake): class InvalidHeaderFormat(InvalidHeader): """ - Raised when a HTTP header cannot be parsed. + Raised when an HTTP header cannot be parsed. The format of the header doesn't match the grammar for that header. @@ -202,7 +208,7 @@ class InvalidHeaderFormat(InvalidHeader): class InvalidHeaderValue(InvalidHeader): """ - Raised when a HTTP header has a wrong value. + Raised when an HTTP header has a wrong value. The format of the header is correct but a value isn't acceptable. @@ -310,7 +316,7 @@ class InvalidParameterValue(NegotiationError): class AbortHandshake(InvalidHandshake): """ - Raised to abort the handshake on purpose and return a HTTP response. + Raised to abort the handshake on purpose and return an HTTP response. This exception is an implementation detail. @@ -325,11 +331,12 @@ class AbortHandshake(InvalidHandshake): def __init__( self, - status: http.HTTPStatus, + status: StatusLike, headers: datastructures.HeadersLike, body: bytes = b"", ) -> None: - self.status = status + # If a user passes an int instead of a HTTPStatus, fix it automatically. + self.status = http.HTTPStatus(status) self.headers = datastructures.Headers(headers) self.body = body @@ -369,7 +376,7 @@ class InvalidState(WebSocketException, AssertionError): class InvalidURI(WebSocketException): """ - Raised when connecting to an URI that isn't a valid WebSocket URI. + Raised when connecting to a URI that isn't a valid WebSocket URI. """ diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/extensions/base.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/extensions/base.py index 0609676185..6c481a46cc 100644 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/extensions/base.py +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/extensions/base.py @@ -38,6 +38,7 @@ class Extension: PayloadTooBig: if decoding the payload exceeds ``max_size``. """ + raise NotImplementedError def encode(self, frame: frames.Frame) -> frames.Frame: """ @@ -50,6 +51,7 @@ class Extension: Frame: Encoded frame. """ + raise NotImplementedError class ClientExtensionFactory: @@ -69,6 +71,7 @@ class ClientExtensionFactory: List[ExtensionParameter]: Parameters to send to the server. """ + raise NotImplementedError def process_response_params( self, @@ -91,6 +94,7 @@ class ClientExtensionFactory: NegotiationError: if parameters aren't acceptable. """ + raise NotImplementedError class ServerExtensionFactory: @@ -126,3 +130,4 @@ class ServerExtensionFactory: the client aren't acceptable. """ + raise NotImplementedError diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/extensions/permessage_deflate.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/extensions/permessage_deflate.py index e0de5e8f85..b391837c66 100644 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/extensions/permessage_deflate.py +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/extensions/permessage_deflate.py @@ -211,7 +211,6 @@ def _extract_parameters( client_max_window_bits: Optional[Union[int, bool]] = None for name, value in params: - if name == "server_no_context_takeover": if server_no_context_takeover: raise exceptions.DuplicateParameter(name) diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/frames.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/frames.py index 043b688b52..6b1befb2e0 100644 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/frames.py +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/frames.py @@ -13,7 +13,7 @@ from .typing import Data try: from .speedups import apply_mask -except ImportError: # pragma: no cover +except ImportError: from .utils import apply_mask @@ -52,45 +52,70 @@ DATA_OPCODES = OP_CONT, OP_TEXT, OP_BINARY CTRL_OPCODES = OP_CLOSE, OP_PING, OP_PONG -# See https://www.iana.org/assignments/websocket/websocket.xhtml -CLOSE_CODES = { - 1000: "OK", - 1001: "going away", - 1002: "protocol error", - 1003: "unsupported type", +class CloseCode(enum.IntEnum): + """Close code values for WebSocket close frames.""" + + NORMAL_CLOSURE = 1000 + GOING_AWAY = 1001 + PROTOCOL_ERROR = 1002 + UNSUPPORTED_DATA = 1003 # 1004 is reserved - 1005: "no status code [internal]", - 1006: "connection closed abnormally [internal]", - 1007: "invalid data", - 1008: "policy violation", - 1009: "message too big", - 1010: "extension required", - 1011: "unexpected error", - 1012: "service restart", - 1013: "try again later", - 1014: "bad gateway", - 1015: "TLS failure [internal]", + NO_STATUS_RCVD = 1005 + ABNORMAL_CLOSURE = 1006 + INVALID_DATA = 1007 + POLICY_VIOLATION = 1008 + MESSAGE_TOO_BIG = 1009 + MANDATORY_EXTENSION = 1010 + INTERNAL_ERROR = 1011 + SERVICE_RESTART = 1012 + TRY_AGAIN_LATER = 1013 + BAD_GATEWAY = 1014 + TLS_HANDSHAKE = 1015 + + +# See https://www.iana.org/assignments/websocket/websocket.xhtml +CLOSE_CODE_EXPLANATIONS: dict[int, str] = { + CloseCode.NORMAL_CLOSURE: "OK", + CloseCode.GOING_AWAY: "going away", + CloseCode.PROTOCOL_ERROR: "protocol error", + CloseCode.UNSUPPORTED_DATA: "unsupported data", + CloseCode.NO_STATUS_RCVD: "no status received [internal]", + CloseCode.ABNORMAL_CLOSURE: "abnormal closure [internal]", + CloseCode.INVALID_DATA: "invalid frame payload data", + CloseCode.POLICY_VIOLATION: "policy violation", + CloseCode.MESSAGE_TOO_BIG: "message too big", + CloseCode.MANDATORY_EXTENSION: "mandatory extension", + CloseCode.INTERNAL_ERROR: "internal error", + CloseCode.SERVICE_RESTART: "service restart", + CloseCode.TRY_AGAIN_LATER: "try again later", + CloseCode.BAD_GATEWAY: "bad gateway", + CloseCode.TLS_HANDSHAKE: "TLS handshake failure [internal]", } # Close code that are allowed in a close frame. # Using a set optimizes `code in EXTERNAL_CLOSE_CODES`. EXTERNAL_CLOSE_CODES = { - 1000, - 1001, - 1002, - 1003, - 1007, - 1008, - 1009, - 1010, - 1011, - 1012, - 1013, - 1014, + CloseCode.NORMAL_CLOSURE, + CloseCode.GOING_AWAY, + CloseCode.PROTOCOL_ERROR, + CloseCode.UNSUPPORTED_DATA, + CloseCode.INVALID_DATA, + CloseCode.POLICY_VIOLATION, + CloseCode.MESSAGE_TOO_BIG, + CloseCode.MANDATORY_EXTENSION, + CloseCode.INTERNAL_ERROR, + CloseCode.SERVICE_RESTART, + CloseCode.TRY_AGAIN_LATER, + CloseCode.BAD_GATEWAY, } -OK_CLOSE_CODES = {1000, 1001} + +OK_CLOSE_CODES = { + CloseCode.NORMAL_CLOSURE, + CloseCode.GOING_AWAY, + CloseCode.NO_STATUS_RCVD, +} BytesLike = bytes, bytearray, memoryview @@ -123,7 +148,7 @@ class Frame: def __str__(self) -> str: """ - Return a human-readable represention of a frame. + Return a human-readable representation of a frame. """ coding = None @@ -191,6 +216,8 @@ class Frame: extensions: list of extensions, applied in reverse order. Raises: + EOFError: if the connection is closed without a full WebSocket frame. + UnicodeDecodeError: if the frame contains invalid UTF-8. PayloadTooBig: if the frame's payload size exceeds ``max_size``. ProtocolError: if the frame contains incorrect values. @@ -383,7 +410,7 @@ class Close: def __str__(self) -> str: """ - Return a human-readable represention of a close code and reason. + Return a human-readable representation of a close code and reason. """ if 3000 <= self.code < 4000: @@ -391,7 +418,7 @@ class Close: elif 4000 <= self.code < 5000: explanation = "private use" else: - explanation = CLOSE_CODES.get(self.code, "unknown") + explanation = CLOSE_CODE_EXPLANATIONS.get(self.code, "unknown") result = f"{self.code} ({explanation})" if self.reason: @@ -419,7 +446,7 @@ class Close: close.check() return close elif len(data) == 0: - return cls(1005, "") + return cls(CloseCode.NO_STATUS_RCVD, "") else: raise exceptions.ProtocolError("close frame too short") diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/http.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/http.py index b14fa94bdc..9f86f6a1ff 100644 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/http.py +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/http.py @@ -1,6 +1,7 @@ from __future__ import annotations import sys +import typing from .imports import lazy_import from .version import version as websockets_version @@ -9,18 +10,22 @@ from .version import version as websockets_version # For backwards compatibility: -lazy_import( - globals(), - # Headers and MultipleValuesError used to be defined in this module. - aliases={ - "Headers": ".datastructures", - "MultipleValuesError": ".datastructures", - }, - deprecated_aliases={ - "read_request": ".legacy.http", - "read_response": ".legacy.http", - }, -) +# When type checking, import non-deprecated aliases eagerly. Else, import on demand. +if typing.TYPE_CHECKING: + from .datastructures import Headers, MultipleValuesError # noqa: F401 +else: + lazy_import( + globals(), + # Headers and MultipleValuesError used to be defined in this module. + aliases={ + "Headers": ".datastructures", + "MultipleValuesError": ".datastructures", + }, + deprecated_aliases={ + "read_request": ".legacy.http", + "read_response": ".legacy.http", + }, + ) __all__ = ["USER_AGENT"] diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/http11.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/http11.py index 84048fa47b..ec4e3b8b7d 100644 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/http11.py +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/http11.py @@ -8,14 +8,12 @@ from typing import Callable, Generator, Optional from . import datastructures, exceptions -# Maximum total size of headers is around 256 * 4 KiB = 1 MiB -MAX_HEADERS = 256 +# Maximum total size of headers is around 128 * 8 KiB = 1 MiB. +MAX_HEADERS = 128 -# We can use the same limit for the request line and header lines: -# "GET <4096 bytes> HTTP/1.1\r\n" = 4111 bytes -# "Set-Cookie: <4097 bytes>\r\n" = 4111 bytes -# (RFC requires 4096 bytes; for some reason Firefox supports 4097 bytes.) -MAX_LINE = 4111 +# Limit request line and header lines. 8KiB is the most common default +# configuration of popular HTTP servers. +MAX_LINE = 8192 # Support for HTTP response bodies is intended to read an error message # returned by a server. It isn't designed to perform large file transfers. @@ -70,7 +68,7 @@ class Request: def exception(self) -> Optional[Exception]: # pragma: no cover warnings.warn( "Request.exception is deprecated; " - "use ServerConnection.handshake_exc instead", + "use ServerProtocol.handshake_exc instead", DeprecationWarning, ) return self._exception @@ -174,7 +172,7 @@ class Response: def exception(self) -> Optional[Exception]: # pragma: no cover warnings.warn( "Response.exception is deprecated; " - "use ClientConnection.handshake_exc instead", + "use ClientProtocol.handshake_exc instead", DeprecationWarning, ) return self._exception diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/async_timeout.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/async_timeout.py new file mode 100644 index 0000000000..8264094f5b --- /dev/null +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/async_timeout.py @@ -0,0 +1,265 @@ +# From https://github.com/aio-libs/async-timeout/blob/master/async_timeout/__init__.py +# Licensed under the Apache License (Apache-2.0) + +import asyncio +import enum +import sys +import warnings +from types import TracebackType +from typing import Optional, Type + + +# From https://github.com/python/typing_extensions/blob/main/src/typing_extensions.py +# Licensed under the Python Software Foundation License (PSF-2.0) + +if sys.version_info >= (3, 11): + from typing import final +else: + # @final exists in 3.8+, but we backport it for all versions + # before 3.11 to keep support for the __final__ attribute. + # See https://bugs.python.org/issue46342 + def final(f): + """This decorator can be used to indicate to type checkers that + the decorated method cannot be overridden, and decorated class + cannot be subclassed. For example: + + class Base: + @final + def done(self) -> None: + ... + class Sub(Base): + def done(self) -> None: # Error reported by type checker + ... + @final + class Leaf: + ... + class Other(Leaf): # Error reported by type checker + ... + + There is no runtime checking of these properties. The decorator + sets the ``__final__`` attribute to ``True`` on the decorated object + to allow runtime introspection. + """ + try: + f.__final__ = True + except (AttributeError, TypeError): + # Skip the attribute silently if it is not writable. + # AttributeError happens if the object has __slots__ or a + # read-only property, TypeError if it's a builtin class. + pass + return f + + +# End https://github.com/aio-libs/async-timeout/blob/master/async_timeout/__init__.py + +__version__ = "4.0.2" + + +__all__ = ("timeout", "timeout_at", "Timeout") + + +def timeout(delay: Optional[float]) -> "Timeout": + """timeout context manager. + + Useful in cases when you want to apply timeout logic around block + of code or in cases when asyncio.wait_for is not suitable. For example: + + >>> async with timeout(0.001): + ... async with aiohttp.get('https://github.com') as r: + ... await r.text() + + + delay - value in seconds or None to disable timeout logic + """ + loop = asyncio.get_running_loop() + if delay is not None: + deadline = loop.time() + delay # type: Optional[float] + else: + deadline = None + return Timeout(deadline, loop) + + +def timeout_at(deadline: Optional[float]) -> "Timeout": + """Schedule the timeout at absolute time. + + deadline argument points on the time in the same clock system + as loop.time(). + + Please note: it is not POSIX time but a time with + undefined starting base, e.g. the time of the system power on. + + >>> async with timeout_at(loop.time() + 10): + ... async with aiohttp.get('https://github.com') as r: + ... await r.text() + + + """ + loop = asyncio.get_running_loop() + return Timeout(deadline, loop) + + +class _State(enum.Enum): + INIT = "INIT" + ENTER = "ENTER" + TIMEOUT = "TIMEOUT" + EXIT = "EXIT" + + +@final +class Timeout: + # Internal class, please don't instantiate it directly + # Use timeout() and timeout_at() public factories instead. + # + # Implementation note: `async with timeout()` is preferred + # over `with timeout()`. + # While technically the Timeout class implementation + # doesn't need to be async at all, + # the `async with` statement explicitly points that + # the context manager should be used from async function context. + # + # This design allows to avoid many silly misusages. + # + # TimeoutError is raised immediately when scheduled + # if the deadline is passed. + # The purpose is to time out as soon as possible + # without waiting for the next await expression. + + __slots__ = ("_deadline", "_loop", "_state", "_timeout_handler") + + def __init__( + self, deadline: Optional[float], loop: asyncio.AbstractEventLoop + ) -> None: + self._loop = loop + self._state = _State.INIT + + self._timeout_handler = None # type: Optional[asyncio.Handle] + if deadline is None: + self._deadline = None # type: Optional[float] + else: + self.update(deadline) + + def __enter__(self) -> "Timeout": + warnings.warn( + "with timeout() is deprecated, use async with timeout() instead", + DeprecationWarning, + stacklevel=2, + ) + self._do_enter() + return self + + def __exit__( + self, + exc_type: Optional[Type[BaseException]], + exc_val: Optional[BaseException], + exc_tb: Optional[TracebackType], + ) -> Optional[bool]: + self._do_exit(exc_type) + return None + + async def __aenter__(self) -> "Timeout": + self._do_enter() + return self + + async def __aexit__( + self, + exc_type: Optional[Type[BaseException]], + exc_val: Optional[BaseException], + exc_tb: Optional[TracebackType], + ) -> Optional[bool]: + self._do_exit(exc_type) + return None + + @property + def expired(self) -> bool: + """Is timeout expired during execution?""" + return self._state == _State.TIMEOUT + + @property + def deadline(self) -> Optional[float]: + return self._deadline + + def reject(self) -> None: + """Reject scheduled timeout if any.""" + # cancel is maybe better name but + # task.cancel() raises CancelledError in asyncio world. + if self._state not in (_State.INIT, _State.ENTER): + raise RuntimeError(f"invalid state {self._state.value}") + self._reject() + + def _reject(self) -> None: + if self._timeout_handler is not None: + self._timeout_handler.cancel() + self._timeout_handler = None + + def shift(self, delay: float) -> None: + """Advance timeout on delay seconds. + + The delay can be negative. + + Raise RuntimeError if shift is called when deadline is not scheduled + """ + deadline = self._deadline + if deadline is None: + raise RuntimeError("cannot shift timeout if deadline is not scheduled") + self.update(deadline + delay) + + def update(self, deadline: float) -> None: + """Set deadline to absolute value. + + deadline argument points on the time in the same clock system + as loop.time(). + + If new deadline is in the past the timeout is raised immediately. + + Please note: it is not POSIX time but a time with + undefined starting base, e.g. the time of the system power on. + """ + if self._state == _State.EXIT: + raise RuntimeError("cannot reschedule after exit from context manager") + if self._state == _State.TIMEOUT: + raise RuntimeError("cannot reschedule expired timeout") + if self._timeout_handler is not None: + self._timeout_handler.cancel() + self._deadline = deadline + if self._state != _State.INIT: + self._reschedule() + + def _reschedule(self) -> None: + assert self._state == _State.ENTER + deadline = self._deadline + if deadline is None: + return + + now = self._loop.time() + if self._timeout_handler is not None: + self._timeout_handler.cancel() + + task = asyncio.current_task() + if deadline <= now: + self._timeout_handler = self._loop.call_soon(self._on_timeout, task) + else: + self._timeout_handler = self._loop.call_at(deadline, self._on_timeout, task) + + def _do_enter(self) -> None: + if self._state != _State.INIT: + raise RuntimeError(f"invalid state {self._state.value}") + self._state = _State.ENTER + self._reschedule() + + def _do_exit(self, exc_type: Optional[Type[BaseException]]) -> None: + if exc_type is asyncio.CancelledError and self._state == _State.TIMEOUT: + self._timeout_handler = None + raise asyncio.TimeoutError + # timeout has not expired + self._state = _State.EXIT + self._reject() + return None + + def _on_timeout(self, task: "asyncio.Task[None]") -> None: + task.cancel() + self._state = _State.TIMEOUT + # drop the reference early + self._timeout_handler = None + + +# End https://github.com/aio-libs/async-timeout/blob/master/async_timeout/__init__.py diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/auth.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/auth.py index 8825c14ecf..d3425836e1 100644 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/auth.py +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/auth.py @@ -67,7 +67,7 @@ class BasicAuthWebSocketServerProtocol(WebSocketServerProtocol): Returns: bool: :obj:`True` if the handshake should continue; - :obj:`False` if it should fail with a HTTP 401 error. + :obj:`False` if it should fail with an HTTP 401 error. """ if self._check_credentials is not None: @@ -81,7 +81,7 @@ class BasicAuthWebSocketServerProtocol(WebSocketServerProtocol): request_headers: Headers, ) -> Optional[HTTPResponse]: """ - Check HTTP Basic Auth and return a HTTP 401 response if needed. + Check HTTP Basic Auth and return an HTTP 401 response if needed. """ try: @@ -118,8 +118,8 @@ def basic_auth_protocol_factory( realm: Optional[str] = None, credentials: Optional[Union[Credentials, Iterable[Credentials]]] = None, check_credentials: Optional[Callable[[str, str], Awaitable[bool]]] = None, - create_protocol: Optional[Callable[[Any], BasicAuthWebSocketServerProtocol]] = None, -) -> Callable[[Any], BasicAuthWebSocketServerProtocol]: + create_protocol: Optional[Callable[..., BasicAuthWebSocketServerProtocol]] = None, +) -> Callable[..., BasicAuthWebSocketServerProtocol]: """ Protocol factory that enforces HTTP Basic Auth. @@ -135,20 +135,20 @@ def basic_auth_protocol_factory( ) Args: - realm: indicates the scope of protection. It should contain only ASCII - characters because the encoding of non-ASCII characters is - undefined. Refer to section 2.2 of :rfc:`7235` for details. - credentials: defines hard coded authorized credentials. It can be a + realm: Scope of protection. It should contain only ASCII characters + because the encoding of non-ASCII characters is undefined. + Refer to section 2.2 of :rfc:`7235` for details. + credentials: Hard coded authorized credentials. It can be a ``(username, password)`` pair or a list of such pairs. - check_credentials: defines a coroutine that verifies credentials. - This coroutine receives ``username`` and ``password`` arguments + check_credentials: Coroutine that verifies credentials. + It receives ``username`` and ``password`` arguments and returns a :class:`bool`. One of ``credentials`` or ``check_credentials`` must be provided but not both. - create_protocol: factory that creates the protocol. By default, this + create_protocol: Factory that creates the protocol. By default, this is :class:`BasicAuthWebSocketServerProtocol`. It can be replaced by a subclass. Raises: - TypeError: if the ``credentials`` or ``check_credentials`` argument is + TypeError: If the ``credentials`` or ``check_credentials`` argument is wrong. """ @@ -175,11 +175,7 @@ def basic_auth_protocol_factory( return hmac.compare_digest(expected_password, password) if create_protocol is None: - # Not sure why mypy cannot figure this out. - create_protocol = cast( - Callable[[Any], BasicAuthWebSocketServerProtocol], - BasicAuthWebSocketServerProtocol, - ) + create_protocol = BasicAuthWebSocketServerProtocol return functools.partial( create_protocol, diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/client.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/client.py index fadc3efe87..48622523ee 100644 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/client.py +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/client.py @@ -44,6 +44,7 @@ from ..headers import ( from ..http import USER_AGENT from ..typing import ExtensionHeader, LoggerLike, Origin, Subprotocol from ..uri import WebSocketURI, parse_uri +from .compatibility import asyncio_timeout from .handshake import build_request, check_response from .http import read_response from .protocol import WebSocketCommonProtocol @@ -65,12 +66,13 @@ class WebSocketClientProtocol(WebSocketCommonProtocol): await process(message) The iterator exits normally when the connection is closed with close code - 1000 (OK) or 1001 (going away). It raises + 1000 (OK) or 1001 (going away) or without a close code. It raises a :exc:`~websockets.exceptions.ConnectionClosedError` when the connection is closed with any other code. See :func:`connect` for the documentation of ``logger``, ``origin``, - ``extensions``, ``subprotocols``, and ``extra_headers``. + ``extensions``, ``subprotocols``, ``extra_headers``, and + ``user_agent_header``. See :class:`~websockets.legacy.protocol.WebSocketCommonProtocol` for the documentation of ``ping_interval``, ``ping_timeout``, ``close_timeout``, @@ -89,6 +91,7 @@ class WebSocketClientProtocol(WebSocketCommonProtocol): extensions: Optional[Sequence[ClientExtensionFactory]] = None, subprotocols: Optional[Sequence[Subprotocol]] = None, extra_headers: Optional[HeadersLike] = None, + user_agent_header: Optional[str] = USER_AGENT, **kwargs: Any, ) -> None: if logger is None: @@ -98,6 +101,7 @@ class WebSocketClientProtocol(WebSocketCommonProtocol): self.available_extensions = extensions self.available_subprotocols = subprotocols self.extra_headers = extra_headers + self.user_agent_header = user_agent_header def write_http_request(self, path: str, headers: Headers) -> None: """ @@ -127,16 +131,12 @@ class WebSocketClientProtocol(WebSocketCommonProtocol): after this coroutine returns. Raises: - InvalidMessage: if the HTTP message is malformed or isn't an + InvalidMessage: If the HTTP message is malformed or isn't an HTTP/1.1 GET response. """ try: status_code, reason, headers = await read_response(self.reader) - # Remove this branch when dropping support for Python < 3.8 - # because CancelledError no longer inherits Exception. - except asyncio.CancelledError: # pragma: no cover - raise except Exception as exc: raise InvalidMessage("did not receive a valid HTTP response") from exc @@ -185,7 +185,6 @@ class WebSocketClientProtocol(WebSocketCommonProtocol): header_values = headers.get_all("Sec-WebSocket-Extensions") if header_values: - if available_extensions is None: raise InvalidHandshake("no extensions supported") @@ -194,9 +193,7 @@ class WebSocketClientProtocol(WebSocketCommonProtocol): ) for name, response_params in parsed_header_values: - for extension_factory in available_extensions: - # Skip non-matching extensions based on their name. if extension_factory.name != name: continue @@ -242,7 +239,6 @@ class WebSocketClientProtocol(WebSocketCommonProtocol): header_values = headers.get_all("Sec-WebSocket-Protocol") if header_values: - if available_subprotocols is None: raise InvalidHandshake("no subprotocols supported") @@ -274,15 +270,15 @@ class WebSocketClientProtocol(WebSocketCommonProtocol): Args: wsuri: URI of the WebSocket server. - origin: value of the ``Origin`` header. - available_extensions: list of supported extensions, in order in - which they should be tried. - available_subprotocols: list of supported subprotocols, in order - of decreasing preference. - extra_headers: arbitrary HTTP headers to add to the request. + origin: Value of the ``Origin`` header. + extensions: List of supported extensions, in order in which they + should be negotiated and run. + subprotocols: List of supported subprotocols, in order of decreasing + preference. + extra_headers: Arbitrary HTTP headers to add to the handshake request. Raises: - InvalidHandshake: if the handshake fails. + InvalidHandshake: If the handshake fails. """ request_headers = Headers() @@ -315,7 +311,8 @@ class WebSocketClientProtocol(WebSocketCommonProtocol): if self.extra_headers is not None: request_headers.update(self.extra_headers) - request_headers.setdefault("User-Agent", USER_AGENT) + if self.user_agent_header is not None: + request_headers.setdefault("User-Agent", self.user_agent_header) self.write_http_request(wsuri.resource_name, request_headers) @@ -376,25 +373,26 @@ class Connect: Args: uri: URI of the WebSocket server. - create_protocol: factory for the :class:`asyncio.Protocol` managing - the connection; defaults to :class:`WebSocketClientProtocol`; may - be set to a wrapper or a subclass to customize connection handling. - logger: logger for this connection; - defaults to ``logging.getLogger("websockets.client")``; - see the :doc:`logging guide <../topics/logging>` for details. - compression: shortcut that enables the "permessage-deflate" extension - by default; may be set to :obj:`None` to disable compression; - see the :doc:`compression guide <../topics/compression>` for details. - origin: value of the ``Origin`` header. This is useful when connecting - to a server that validates the ``Origin`` header to defend against - Cross-Site WebSocket Hijacking attacks. - extensions: list of supported extensions, in order in which they - should be tried. - subprotocols: list of supported subprotocols, in order of decreasing + create_protocol: Factory for the :class:`asyncio.Protocol` managing + the connection. It defaults to :class:`WebSocketClientProtocol`. + Set it to a wrapper or a subclass to customize connection handling. + logger: Logger for this client. + It defaults to ``logging.getLogger("websockets.client")``. + See the :doc:`logging guide <../../topics/logging>` for details. + compression: The "permessage-deflate" extension is enabled by default. + Set ``compression`` to :obj:`None` to disable it. See the + :doc:`compression guide <../../topics/compression>` for details. + origin: Value of the ``Origin`` header, for servers that require it. + extensions: List of supported extensions, in order in which they + should be negotiated and run. + subprotocols: List of supported subprotocols, in order of decreasing preference. - extra_headers: arbitrary HTTP headers to add to the request. - open_timeout: timeout for opening the connection in seconds; - :obj:`None` to disable the timeout + extra_headers: Arbitrary HTTP headers to add to the handshake request. + user_agent_header: Value of the ``User-Agent`` request header. + It defaults to ``"Python/x.y.z websockets/X.Y"``. + Setting it to :obj:`None` removes the header. + open_timeout: Timeout for opening the connection in seconds. + :obj:`None` disables the timeout. See :class:`~websockets.legacy.protocol.WebSocketCommonProtocol` for the documentation of ``ping_interval``, ``ping_timeout``, ``close_timeout``, @@ -415,13 +413,11 @@ class Connect: the TCP connection. The host name from ``uri`` is still used in the TLS handshake for secure connections and in the ``Host`` header. - Returns: - WebSocketClientProtocol: WebSocket connection. - Raises: - InvalidURI: if ``uri`` isn't a valid WebSocket URI. - InvalidHandshake: if the opening handshake fails. - ~asyncio.TimeoutError: if the opening handshake times out. + InvalidURI: If ``uri`` isn't a valid WebSocket URI. + OSError: If the TCP connection fails. + InvalidHandshake: If the opening handshake fails. + ~asyncio.TimeoutError: If the opening handshake times out. """ @@ -431,13 +427,14 @@ class Connect: self, uri: str, *, - create_protocol: Optional[Callable[[Any], WebSocketClientProtocol]] = None, + create_protocol: Optional[Callable[..., WebSocketClientProtocol]] = None, logger: Optional[LoggerLike] = None, compression: Optional[str] = "deflate", origin: Optional[Origin] = None, extensions: Optional[Sequence[ClientExtensionFactory]] = None, subprotocols: Optional[Sequence[Subprotocol]] = None, extra_headers: Optional[HeadersLike] = None, + user_agent_header: Optional[str] = USER_AGENT, open_timeout: Optional[float] = 10, ping_interval: Optional[float] = 20, ping_timeout: Optional[float] = 20, @@ -503,6 +500,7 @@ class Connect: extensions=extensions, subprotocols=subprotocols, extra_headers=extra_headers, + user_agent_header=user_agent_header, ping_interval=ping_interval, ping_timeout=ping_timeout, close_timeout=close_timeout, @@ -530,6 +528,8 @@ class Connect: else: # If sock is given, host and port shouldn't be specified. host, port = None, None + if kwargs.get("ssl"): + kwargs.setdefault("server_hostname", wsuri.host) # If host and port are given, override values from the URI. host = kwargs.pop("host", host) port = kwargs.pop("port", port) @@ -597,10 +597,6 @@ class Connect: try: async with self as protocol: yield protocol - # Remove this branch when dropping support for Python < 3.8 - # because CancelledError no longer inherits Exception. - except asyncio.CancelledError: # pragma: no cover - raise except Exception: # Add a random initial delay between 0 and 5 seconds. # See 7.2.3. Recovering from Abnormal Closure in RFC 6544. @@ -647,13 +643,13 @@ class Connect: return self.__await_impl_timeout__().__await__() async def __await_impl_timeout__(self) -> WebSocketClientProtocol: - return await asyncio.wait_for(self.__await_impl__(), self.open_timeout) + async with asyncio_timeout(self.open_timeout): + return await self.__await_impl__() async def __await_impl__(self) -> WebSocketClientProtocol: for redirects in range(self.MAX_REDIRECTS_ALLOWED): - transport, protocol = await self._create_connection() - protocol = cast(WebSocketClientProtocol, protocol) - + _transport, _protocol = await self._create_connection() + protocol = cast(WebSocketClientProtocol, _protocol) try: await protocol.handshake( self._wsuri, @@ -701,7 +697,7 @@ def unix_connect( It's mainly useful for debugging servers listening on Unix sockets. Args: - path: file system path to the Unix socket. + path: File system path to the Unix socket. uri: URI of the WebSocket server; the host is used in the TLS handshake for secure connections and in the ``Host`` header. diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/compatibility.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/compatibility.py index df81de9dbc..6bd01e70de 100644 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/compatibility.py +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/compatibility.py @@ -1,13 +1,12 @@ from __future__ import annotations -import asyncio import sys -from typing import Any, Dict -def loop_if_py_lt_38(loop: asyncio.AbstractEventLoop) -> Dict[str, Any]: - """ - Helper for the removal of the loop argument in Python 3.10. +__all__ = ["asyncio_timeout"] - """ - return {"loop": loop} if sys.version_info[:2] < (3, 8) else {} + +if sys.version_info[:2] >= (3, 11): + from asyncio import timeout as asyncio_timeout # noqa: F401 +else: + from .async_timeout import timeout as asyncio_timeout # noqa: F401 diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/framing.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/framing.py index c4de7eb28b..b77b869e3f 100644 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/framing.py +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/framing.py @@ -1,6 +1,5 @@ from __future__ import annotations -import dataclasses import struct from typing import Any, Awaitable, Callable, NamedTuple, Optional, Sequence, Tuple @@ -10,12 +9,11 @@ from ..exceptions import PayloadTooBig, ProtocolError try: from ..speedups import apply_mask -except ImportError: # pragma: no cover +except ImportError: from ..utils import apply_mask class Frame(NamedTuple): - fin: bool opcode: frames.Opcode data: bytes @@ -53,16 +51,16 @@ class Frame(NamedTuple): Read a WebSocket frame. Args: - reader: coroutine that reads exactly the requested number of + reader: Coroutine that reads exactly the requested number of bytes, unless the end of file is reached. - mask: whether the frame should be masked i.e. whether the read + mask: Whether the frame should be masked i.e. whether the read happens on the server side. - max_size: maximum payload size in bytes. - extensions: list of extensions, applied in reverse order. + max_size: Maximum payload size in bytes. + extensions: List of extensions, applied in reverse order. Raises: - PayloadTooBig: if the frame exceeds ``max_size``. - ProtocolError: if the frame contains incorrect values. + PayloadTooBig: If the frame exceeds ``max_size``. + ProtocolError: If the frame contains incorrect values. """ @@ -130,14 +128,14 @@ class Frame(NamedTuple): Write a WebSocket frame. Args: - frame: frame to write. - write: function that writes bytes. - mask: whether the frame should be masked i.e. whether the write + frame: Frame to write. + write: Function that writes bytes. + mask: Whether the frame should be masked i.e. whether the write happens on the client side. - extensions: list of extensions, applied in order. + extensions: List of extensions, applied in order. Raises: - ProtocolError: if the frame contains incorrect values. + ProtocolError: If the frame contains incorrect values. """ # The frame is written in a single call to write in order to prevent @@ -147,8 +145,11 @@ class Frame(NamedTuple): # Backwards compatibility with previously documented public APIs - -from ..frames import Close, prepare_ctrl as encode_data, prepare_data # noqa +from ..frames import ( # noqa: E402, F401, I001 + Close, + prepare_ctrl as encode_data, + prepare_data, +) def parse_close(data: bytes) -> Tuple[int, str]: @@ -156,14 +157,15 @@ def parse_close(data: bytes) -> Tuple[int, str]: Parse the payload from a close frame. Returns: - Tuple[int, str]: close code and reason. + Close code and reason. Raises: - ProtocolError: if data is ill-formed. - UnicodeDecodeError: if the reason isn't valid UTF-8. + ProtocolError: If data is ill-formed. + UnicodeDecodeError: If the reason isn't valid UTF-8. """ - return dataclasses.astuple(Close.parse(data)) # type: ignore + close = Close.parse(data) + return close.code, close.reason def serialize_close(code: int, reason: str) -> bytes: diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/handshake.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/handshake.py index 569937bb9a..ad8faf0404 100644 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/handshake.py +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/handshake.py @@ -21,7 +21,7 @@ def build_request(headers: Headers) -> str: Update request headers passed in argument. Args: - headers: handshake request headers. + headers: Handshake request headers. Returns: str: ``key`` that must be passed to :func:`check_response`. @@ -45,14 +45,14 @@ def check_request(headers: Headers) -> str: the responsibility of the caller. Args: - headers: handshake request headers. + headers: Handshake request headers. Returns: str: ``key`` that must be passed to :func:`build_response`. Raises: - InvalidHandshake: if the handshake request is invalid; - then the server must return 400 Bad Request error. + InvalidHandshake: If the handshake request is invalid. + Then, the server must return a 400 Bad Request error. """ connection: List[ConnectionOption] = sum( @@ -110,8 +110,8 @@ def build_response(headers: Headers, key: str) -> None: Update response headers passed in argument. Args: - headers: handshake response headers. - key: returned by :func:`check_request`. + headers: Handshake response headers. + key: Returned by :func:`check_request`. """ headers["Upgrade"] = "websocket" @@ -128,11 +128,11 @@ def check_response(headers: Headers, key: str) -> None: the caller. Args: - headers: handshake response headers. - key: returned by :func:`build_request`. + headers: Handshake response headers. + key: Returned by :func:`build_request`. Raises: - InvalidHandshake: if the handshake response is invalid. + InvalidHandshake: If the handshake response is invalid. """ connection: List[ConnectionOption] = sum( diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/http.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/http.py index cc2ef1f067..2ac7f7092d 100644 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/http.py +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/http.py @@ -10,8 +10,8 @@ from ..exceptions import SecurityError __all__ = ["read_request", "read_response"] -MAX_HEADERS = 256 -MAX_LINE = 4110 +MAX_HEADERS = 128 +MAX_LINE = 8192 def d(value: bytes) -> str: @@ -56,12 +56,12 @@ async def read_request(stream: asyncio.StreamReader) -> Tuple[str, Headers]: body, it may be read from ``stream`` after this coroutine returns. Args: - stream: input to read the request from + stream: Input to read the request from. Raises: - EOFError: if the connection is closed without a full HTTP request - SecurityError: if the request exceeds a security limit - ValueError: if the request isn't well formatted + EOFError: If the connection is closed without a full HTTP request. + SecurityError: If the request exceeds a security limit. + ValueError: If the request isn't well formatted. """ # https://www.rfc-editor.org/rfc/rfc7230.html#section-3.1.1 @@ -103,12 +103,12 @@ async def read_response(stream: asyncio.StreamReader) -> Tuple[int, str, Headers body, it may be read from ``stream`` after this coroutine returns. Args: - stream: input to read the response from + stream: Input to read the response from. Raises: - EOFError: if the connection is closed without a full HTTP response - SecurityError: if the response exceeds a security limit - ValueError: if the response isn't well formatted + EOFError: If the connection is closed without a full HTTP response. + SecurityError: If the response exceeds a security limit. + ValueError: If the response isn't well formatted. """ # https://www.rfc-editor.org/rfc/rfc7230.html#section-3.1.2 @@ -192,7 +192,7 @@ async def read_line(stream: asyncio.StreamReader) -> bytes: """ # Security: this is bounded by the StreamReader's limit (default = 32 KiB). line = await stream.readline() - # Security: this guarantees header values are small (hard-coded = 4 KiB) + # Security: this guarantees header values are small (hard-coded = 8 KiB) if len(line) > MAX_LINE: raise SecurityError("line too long") # Not mandatory but safe - https://www.rfc-editor.org/rfc/rfc7230.html#section-3.5 diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/protocol.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/protocol.py index 3f734fe760..19cee0e652 100644 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/protocol.py +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/protocol.py @@ -7,6 +7,8 @@ import logging import random import ssl import struct +import sys +import time import uuid import warnings from typing import ( @@ -14,17 +16,18 @@ from typing import ( AsyncIterable, AsyncIterator, Awaitable, + Callable, Deque, Dict, Iterable, List, Mapping, Optional, + Tuple, Union, cast, ) -from ..connection import State from ..datastructures import Headers from ..exceptions import ( ConnectionClosed, @@ -44,12 +47,14 @@ from ..frames import ( OP_PONG, OP_TEXT, Close, + CloseCode, Opcode, prepare_ctrl, prepare_data, ) +from ..protocol import State from ..typing import Data, LoggerLike, Subprotocol -from .compatibility import loop_if_py_lt_38 +from .compatibility import asyncio_timeout from .framing import Frame @@ -76,38 +81,38 @@ class WebSocketCommonProtocol(asyncio.Protocol): simplicity. Once the connection is open, a Ping_ frame is sent every ``ping_interval`` - seconds. This serves as a keepalive. It helps keeping the connection - open, especially in the presence of proxies with short timeouts on - inactive connections. Set ``ping_interval`` to :obj:`None` to disable - this behavior. + seconds. This serves as a keepalive. It helps keeping the connection open, + especially in the presence of proxies with short timeouts on inactive + connections. Set ``ping_interval`` to :obj:`None` to disable this behavior. .. _Ping: https://www.rfc-editor.org/rfc/rfc6455.html#section-5.5.2 If the corresponding Pong_ frame isn't received within ``ping_timeout`` - seconds, the connection is considered unusable and is closed with code - 1011. This ensures that the remote endpoint remains responsive. Set + seconds, the connection is considered unusable and is closed with code 1011. + This ensures that the remote endpoint remains responsive. Set ``ping_timeout`` to :obj:`None` to disable this behavior. .. _Pong: https://www.rfc-editor.org/rfc/rfc6455.html#section-5.5.3 + See the discussion of :doc:`timeouts <../../topics/timeouts>` for details. + The ``close_timeout`` parameter defines a maximum wait time for completing the closing handshake and terminating the TCP connection. For legacy reasons, :meth:`close` completes in at most ``5 * close_timeout`` seconds for clients and ``4 * close_timeout`` for servers. - See the discussion of :doc:`timeouts <../topics/timeouts>` for details. - - ``close_timeout`` needs to be a parameter of the protocol because - websockets usually calls :meth:`close` implicitly upon exit: + ``close_timeout`` is a parameter of the protocol because websockets usually + calls :meth:`close` implicitly upon exit: - * on the client side, when :func:`~websockets.client.connect` is used as a + * on the client side, when using :func:`~websockets.client.connect` as a context manager; - * on the server side, when the connection handler terminates; + * on the server side, when the connection handler terminates. - To apply a timeout to any other API, wrap it in :func:`~asyncio.wait_for`. + To apply a timeout to any other API, wrap it in :func:`~asyncio.timeout` or + :func:`~asyncio.wait_for`. The ``max_size`` parameter enforces the maximum size for incoming messages - in bytes. The default value is 1 MiB. If a larger message is received, + in bytes. The default value is 1 MiB. If a larger message is received, :meth:`recv` will raise :exc:`~websockets.exceptions.ConnectionClosedError` and the connection will be closed with code 1009. @@ -124,38 +129,38 @@ class WebSocketCommonProtocol(asyncio.Protocol): Since Python can use up to 4 bytes of memory to represent a single character, each connection may use up to ``4 * max_size * max_queue`` - bytes of memory to store incoming messages. By default, this is 128 MiB. + bytes of memory to store incoming messages. By default, this is 128 MiB. You may want to lower the limits, depending on your application's requirements. The ``read_limit`` argument sets the high-water limit of the buffer for incoming bytes. The low-water limit is half the high-water limit. The - default value is 64 KiB, half of asyncio's default (based on the current + default value is 64 KiB, half of asyncio's default (based on the current implementation of :class:`~asyncio.StreamReader`). The ``write_limit`` argument sets the high-water limit of the buffer for outgoing bytes. The low-water limit is a quarter of the high-water limit. - The default value is 64 KiB, equal to asyncio's default (based on the + The default value is 64 KiB, equal to asyncio's default (based on the current implementation of ``FlowControlMixin``). - See the discussion of :doc:`memory usage <../topics/memory>` for details. + See the discussion of :doc:`memory usage <../../topics/memory>` for details. Args: - logger: logger for this connection; - defaults to ``logging.getLogger("websockets.protocol")``; - see the :doc:`logging guide <../topics/logging>` for details. - ping_interval: delay between keepalive pings in seconds; - :obj:`None` to disable keepalive pings. - ping_timeout: timeout for keepalive pings in seconds; - :obj:`None` to disable timeouts. - close_timeout: timeout for closing the connection in seconds; - for legacy reasons, the actual timeout is 4 or 5 times larger. - max_size: maximum size of incoming messages in bytes; - :obj:`None` to disable the limit. - max_queue: maximum number of incoming messages in receive buffer; - :obj:`None` to disable the limit. - read_limit: high-water mark of read buffer in bytes. - write_limit: high-water mark of write buffer in bytes. + logger: Logger for this server. + It defaults to ``logging.getLogger("websockets.protocol")``. + See the :doc:`logging guide <../../topics/logging>` for details. + ping_interval: Delay between keepalive pings in seconds. + :obj:`None` disables keepalive pings. + ping_timeout: Timeout for keepalive pings in seconds. + :obj:`None` disables timeouts. + close_timeout: Timeout for closing the connection in seconds. + For legacy reasons, the actual timeout is 4 or 5 times larger. + max_size: Maximum size of incoming messages in bytes. + :obj:`None` disables the limit. + max_queue: Maximum number of incoming messages in receive buffer. + :obj:`None` disables the limit. + read_limit: High-water mark of read buffer in bytes. + write_limit: High-water mark of write buffer in bytes. """ @@ -217,8 +222,6 @@ class WebSocketCommonProtocol(asyncio.Protocol): # Logger or LoggerAdapter for this connection. if logger is None: logger = logging.getLogger("websockets.protocol") - # https://github.com/python/typeshed/issues/5561 - logger = cast(logging.Logger, logger) self.logger: LoggerLike = logging.LoggerAdapter(logger, {"websocket": self}) """Logger for this connection.""" @@ -242,7 +245,7 @@ class WebSocketCommonProtocol(asyncio.Protocol): self._paused = False self._drain_waiter: Optional[asyncio.Future[None]] = None - self._drain_lock = asyncio.Lock(**loop_if_py_lt_38(loop)) + self._drain_lock = asyncio.Lock() # This class implements the data transfer and closing handshake, which # are shared between the client-side and the server-side. @@ -285,7 +288,19 @@ class WebSocketCommonProtocol(asyncio.Protocol): self._fragmented_message_waiter: Optional[asyncio.Future[None]] = None # Mapping of ping IDs to pong waiters, in chronological order. - self.pings: Dict[bytes, asyncio.Future[None]] = {} + self.pings: Dict[bytes, Tuple[asyncio.Future[float], float]] = {} + + self.latency: float = 0 + """ + Latency of the connection, in seconds. + + This value is updated after sending a ping frame and receiving a + matching pong frame. Before the first ping, :attr:`latency` is ``0``. + + By default, websockets enables a :ref:`keepalive <keepalive>` mechanism + that sends ping frames automatically at regular intervals. You can also + send ping frames and measure latency with :meth:`ping`. + """ # Task running the data transfer. self.transfer_data_task: asyncio.Task[None] @@ -325,7 +340,7 @@ class WebSocketCommonProtocol(asyncio.Protocol): # write(...); yield from drain() # in a loop would never call connection_lost(), so it # would not see an error when the socket is closed. - await asyncio.sleep(0, **loop_if_py_lt_38(self.loop)) + await asyncio.sleep(0) await self._drain_helper() def connection_open(self) -> None: @@ -445,7 +460,7 @@ class WebSocketCommonProtocol(asyncio.Protocol): if self.state is not State.CLOSED: return None elif self.close_rcvd is None: - return 1006 + return CloseCode.ABNORMAL_CLOSURE else: return self.close_rcvd.code @@ -471,10 +486,11 @@ class WebSocketCommonProtocol(asyncio.Protocol): """ Iterate on incoming messages. - The iterator exits normally when the connection is closed with the - close code 1000 (OK) or 1001(going away). It raises - a :exc:`~websockets.exceptions.ConnectionClosedError` exception when - the connection is closed with any other code. + The iterator exits normally when the connection is closed with the close + code 1000 (OK) or 1001 (going away) or without a close code. + + It raises a :exc:`~websockets.exceptions.ConnectionClosedError` + exception when the connection is closed with any other code. """ try: @@ -488,8 +504,8 @@ class WebSocketCommonProtocol(asyncio.Protocol): Receive the next message. When the connection is closed, :meth:`recv` raises - :exc:`~websockets.exceptions.ConnectionClosed`. Specifically, it - raises :exc:`~websockets.exceptions.ConnectionClosedOK` after a normal + :exc:`~websockets.exceptions.ConnectionClosed`. Specifically, it raises + :exc:`~websockets.exceptions.ConnectionClosedOK` after a normal connection closure and :exc:`~websockets.exceptions.ConnectionClosedError` after a protocol error or a network failure. This is how you detect the end of the @@ -498,8 +514,8 @@ class WebSocketCommonProtocol(asyncio.Protocol): Canceling :meth:`recv` is safe. There's no risk of losing the next message. The next invocation of :meth:`recv` will return it. - This makes it possible to enforce a timeout by wrapping :meth:`recv` - in :func:`~asyncio.wait_for`. + This makes it possible to enforce a timeout by wrapping :meth:`recv` in + :func:`~asyncio.timeout` or :func:`~asyncio.wait_for`. Returns: Data: A string (:class:`str`) for a Text_ frame. A bytestring @@ -509,8 +525,8 @@ class WebSocketCommonProtocol(asyncio.Protocol): .. _Binary: https://www.rfc-editor.org/rfc/rfc6455.html#section-5.6 Raises: - ConnectionClosed: when the connection is closed. - RuntimeError: if two coroutines call :meth:`recv` concurrently. + ConnectionClosed: When the connection is closed. + RuntimeError: If two coroutines call :meth:`recv` concurrently. """ if self._pop_message_waiter is not None: @@ -536,7 +552,6 @@ class WebSocketCommonProtocol(asyncio.Protocol): await asyncio.wait( [pop_message_waiter, self.transfer_data_task], return_when=asyncio.FIRST_COMPLETED, - **loop_if_py_lt_38(self.loop), ) finally: self._pop_message_waiter = None @@ -613,8 +628,8 @@ class WebSocketCommonProtocol(asyncio.Protocol): to send. Raises: - ConnectionClosed: when the connection is closed. - TypeError: if ``message`` doesn't have a supported type. + ConnectionClosed: When the connection is closed. + TypeError: If ``message`` doesn't have a supported type. """ await self.ensure_open() @@ -639,16 +654,15 @@ class WebSocketCommonProtocol(asyncio.Protocol): # Fragmented message -- regular iterator. elif isinstance(message, Iterable): - # Work around https://github.com/python/mypy/issues/6227 message = cast(Iterable[Data], message) iter_message = iter(message) try: - message_chunk = next(iter_message) + fragment = next(iter_message) except StopIteration: return - opcode, data = prepare_data(message_chunk) + opcode, data = prepare_data(fragment) self._fragmented_message_waiter = asyncio.Future() try: @@ -656,8 +670,8 @@ class WebSocketCommonProtocol(asyncio.Protocol): await self.write_frame(False, opcode, data) # Other fragments. - for message_chunk in iter_message: - confirm_opcode, data = prepare_data(message_chunk) + for fragment in iter_message: + confirm_opcode, data = prepare_data(fragment) if confirm_opcode != opcode: raise TypeError("data contains inconsistent types") await self.write_frame(False, OP_CONT, data) @@ -668,7 +682,7 @@ class WebSocketCommonProtocol(asyncio.Protocol): except (Exception, asyncio.CancelledError): # We're half-way through a fragmented message and we can't # complete it. This makes the connection unusable. - self.fail_connection(1011) + self.fail_connection(CloseCode.INTERNAL_ERROR) raise finally: @@ -678,18 +692,22 @@ class WebSocketCommonProtocol(asyncio.Protocol): # Fragmented message -- asynchronous iterator elif isinstance(message, AsyncIterable): - # aiter_message = aiter(message) without aiter - # https://github.com/python/mypy/issues/5738 - aiter_message = type(message).__aiter__(message) # type: ignore + # Implement aiter_message = aiter(message) without aiter + # Work around https://github.com/python/mypy/issues/5738 + aiter_message = cast( + Callable[[AsyncIterable[Data]], AsyncIterator[Data]], + type(message).__aiter__, + )(message) try: - # message_chunk = anext(aiter_message) without anext - # https://github.com/python/mypy/issues/5738 - message_chunk = await type(aiter_message).__anext__( # type: ignore - aiter_message - ) + # Implement fragment = anext(aiter_message) without anext + # Work around https://github.com/python/mypy/issues/5738 + fragment = await cast( + Callable[[AsyncIterator[Data]], Awaitable[Data]], + type(aiter_message).__anext__, + )(aiter_message) except StopAsyncIteration: return - opcode, data = prepare_data(message_chunk) + opcode, data = prepare_data(fragment) self._fragmented_message_waiter = asyncio.Future() try: @@ -697,11 +715,8 @@ class WebSocketCommonProtocol(asyncio.Protocol): await self.write_frame(False, opcode, data) # Other fragments. - # https://github.com/python/mypy/issues/5738 - # coverage reports this code as not covered, but it is - # exercised by tests - changing it breaks the tests! - async for message_chunk in aiter_message: # type: ignore # pragma: no cover # noqa - confirm_opcode, data = prepare_data(message_chunk) + async for fragment in aiter_message: + confirm_opcode, data = prepare_data(fragment) if confirm_opcode != opcode: raise TypeError("data contains inconsistent types") await self.write_frame(False, OP_CONT, data) @@ -712,7 +727,7 @@ class WebSocketCommonProtocol(asyncio.Protocol): except (Exception, asyncio.CancelledError): # We're half-way through a fragmented message and we can't # complete it. This makes the connection unusable. - self.fail_connection(1011) + self.fail_connection(CloseCode.INTERNAL_ERROR) raise finally: @@ -722,7 +737,11 @@ class WebSocketCommonProtocol(asyncio.Protocol): else: raise TypeError("data must be str, bytes-like, or iterable") - async def close(self, code: int = 1000, reason: str = "") -> None: + async def close( + self, + code: int = CloseCode.NORMAL_CLOSURE, + reason: str = "", + ) -> None: """ Perform the closing handshake. @@ -747,19 +766,16 @@ class WebSocketCommonProtocol(asyncio.Protocol): """ try: - await asyncio.wait_for( - self.write_close_frame(Close(code, reason)), - self.close_timeout, - **loop_if_py_lt_38(self.loop), - ) + async with asyncio_timeout(self.close_timeout): + await self.write_close_frame(Close(code, reason)) except asyncio.TimeoutError: # If the close frame cannot be sent because the send buffers # are full, the closing handshake won't complete anyway. # Fail the connection to shut down faster. self.fail_connection() - # If no close frame is received within the timeout, wait_for() cancels - # the data transfer task and raises TimeoutError. + # If no close frame is received within the timeout, asyncio_timeout() + # cancels the data transfer task and raises TimeoutError. # If close() is called multiple times concurrently and one of these # calls hits the timeout, the data transfer task will be canceled. @@ -768,11 +784,8 @@ class WebSocketCommonProtocol(asyncio.Protocol): try: # If close() is canceled during the wait, self.transfer_data_task # is canceled before the timeout elapses. - await asyncio.wait_for( - self.transfer_data_task, - self.close_timeout, - **loop_if_py_lt_38(self.loop), - ) + async with asyncio_timeout(self.close_timeout): + await self.transfer_data_task except (asyncio.TimeoutError, asyncio.CancelledError): pass @@ -798,8 +811,8 @@ class WebSocketCommonProtocol(asyncio.Protocol): .. _Ping: https://www.rfc-editor.org/rfc/rfc6455.html#section-5.5.2 - A ping may serve as a keepalive or as a check that the remote endpoint - received all messages up to this point + A ping may serve as a keepalive, as a check that the remote endpoint + received all messages up to this point, or to measure :attr:`latency`. Canceling :meth:`ping` is discouraged. If :meth:`ping` doesn't return immediately, it means the write buffer is full. If you don't want to @@ -814,18 +827,20 @@ class WebSocketCommonProtocol(asyncio.Protocol): containing four random bytes. Returns: - ~asyncio.Future: A future that will be completed when the - corresponding pong is received. You can ignore it if you - don't intend to wait. + ~asyncio.Future[float]: A future that will be completed when the + corresponding pong is received. You can ignore it if you don't + intend to wait. The result of the future is the latency of the + connection in seconds. :: pong_waiter = await ws.ping() - await pong_waiter # only if you want to wait for the pong + # only if you want to wait for the corresponding pong + latency = await pong_waiter Raises: - ConnectionClosed: when the connection is closed. - RuntimeError: if another ping was sent with the same data and + ConnectionClosed: When the connection is closed. + RuntimeError: If another ping was sent with the same data and the corresponding pong wasn't received yet. """ @@ -842,11 +857,14 @@ class WebSocketCommonProtocol(asyncio.Protocol): while data is None or data in self.pings: data = struct.pack("!I", random.getrandbits(32)) - self.pings[data] = self.loop.create_future() + pong_waiter = self.loop.create_future() + # Resolution of time.monotonic() may be too low on Windows. + ping_timestamp = time.perf_counter() + self.pings[data] = (pong_waiter, ping_timestamp) await self.write_frame(True, OP_PING, data) - return asyncio.shield(self.pings[data]) + return asyncio.shield(pong_waiter) async def pong(self, data: Data = b"") -> None: """ @@ -861,11 +879,11 @@ class WebSocketCommonProtocol(asyncio.Protocol): wait, you should close the connection. Args: - data (Data): payload of the pong; a string will be encoded to + data (Data): Payload of the pong. A string will be encoded to UTF-8. Raises: - ConnectionClosed: when the connection is closed. + ConnectionClosed: When the connection is closed. """ await self.ensure_open() @@ -973,7 +991,7 @@ class WebSocketCommonProtocol(asyncio.Protocol): except ProtocolError as exc: self.transfer_data_exc = exc - self.fail_connection(1002) + self.fail_connection(CloseCode.PROTOCOL_ERROR) except (ConnectionError, TimeoutError, EOFError, ssl.SSLError) as exc: # Reading data with self.reader.readexactly may raise: @@ -984,15 +1002,15 @@ class WebSocketCommonProtocol(asyncio.Protocol): # bytes are available than requested; # - ssl.SSLError if the other side infringes the TLS protocol. self.transfer_data_exc = exc - self.fail_connection(1006) + self.fail_connection(CloseCode.ABNORMAL_CLOSURE) except UnicodeDecodeError as exc: self.transfer_data_exc = exc - self.fail_connection(1007) + self.fail_connection(CloseCode.INVALID_DATA) except PayloadTooBig as exc: self.transfer_data_exc = exc - self.fail_connection(1009) + self.fail_connection(CloseCode.MESSAGE_TOO_BIG) except Exception as exc: # This shouldn't happen often because exceptions expected under @@ -1001,7 +1019,7 @@ class WebSocketCommonProtocol(asyncio.Protocol): self.logger.error("data transfer failed", exc_info=True) self.transfer_data_exc = exc - self.fail_connection(1011) + self.fail_connection(CloseCode.INTERNAL_ERROR) async def read_message(self) -> Optional[Data]: """ @@ -1030,7 +1048,7 @@ class WebSocketCommonProtocol(asyncio.Protocol): return frame.data.decode("utf-8") if text else frame.data # 5.4. Fragmentation - chunks: List[Data] = [] + fragments: List[Data] = [] max_size = self.max_size if text: decoder_factory = codecs.getincrementaldecoder("utf-8") @@ -1038,14 +1056,14 @@ class WebSocketCommonProtocol(asyncio.Protocol): if max_size is None: def append(frame: Frame) -> None: - nonlocal chunks - chunks.append(decoder.decode(frame.data, frame.fin)) + nonlocal fragments + fragments.append(decoder.decode(frame.data, frame.fin)) else: def append(frame: Frame) -> None: - nonlocal chunks, max_size - chunks.append(decoder.decode(frame.data, frame.fin)) + nonlocal fragments, max_size + fragments.append(decoder.decode(frame.data, frame.fin)) assert isinstance(max_size, int) max_size -= len(frame.data) @@ -1053,14 +1071,14 @@ class WebSocketCommonProtocol(asyncio.Protocol): if max_size is None: def append(frame: Frame) -> None: - nonlocal chunks - chunks.append(frame.data) + nonlocal fragments + fragments.append(frame.data) else: def append(frame: Frame) -> None: - nonlocal chunks, max_size - chunks.append(frame.data) + nonlocal fragments, max_size + fragments.append(frame.data) assert isinstance(max_size, int) max_size -= len(frame.data) @@ -1074,7 +1092,7 @@ class WebSocketCommonProtocol(asyncio.Protocol): raise ProtocolError("unexpected opcode") append(frame) - return ("" if text else b"").join(chunks) + return ("" if text else b"").join(fragments) async def read_data_frame(self, max_size: Optional[int]) -> Optional[Frame]: """ @@ -1099,7 +1117,7 @@ class WebSocketCommonProtocol(asyncio.Protocol): try: # Echo the original data instead of re-serializing it with # Close.serialize() because that fails when the close frame - # is empty and Close.parse() synthetizes a 1005 close code. + # is empty and Close.parse() synthesizes a 1005 close code. await self.write_close_frame(self.close_rcvd, frame.data) except ConnectionClosed: # Connection closed before we could echo the close frame. @@ -1117,18 +1135,20 @@ class WebSocketCommonProtocol(asyncio.Protocol): elif frame.opcode == OP_PONG: if frame.data in self.pings: + pong_timestamp = time.perf_counter() # Sending a pong for only the most recent ping is legal. # Acknowledge all previous pings too in that case. ping_id = None ping_ids = [] - for ping_id, ping in self.pings.items(): + for ping_id, (pong_waiter, ping_timestamp) in self.pings.items(): ping_ids.append(ping_id) - if not ping.done(): - ping.set_result(None) + if not pong_waiter.done(): + pong_waiter.set_result(pong_timestamp - ping_timestamp) if ping_id == frame.data: + self.latency = pong_timestamp - ping_timestamp break - else: # pragma: no cover - assert False, "ping_id is in self.pings" + else: + raise AssertionError("solicited pong not found in pings") # Remove acknowledged pings from self.pings. for ping_id in ping_ids: del self.pings[ping_id] @@ -1231,10 +1251,7 @@ class WebSocketCommonProtocol(asyncio.Protocol): try: while True: - await asyncio.sleep( - self.ping_interval, - **loop_if_py_lt_38(self.loop), - ) + await asyncio.sleep(self.ping_interval) # ping() raises CancelledError if the connection is closed, # when close_connection() cancels self.keepalive_ping_task. @@ -1247,23 +1264,18 @@ class WebSocketCommonProtocol(asyncio.Protocol): if self.ping_timeout is not None: try: - await asyncio.wait_for( - pong_waiter, - self.ping_timeout, - **loop_if_py_lt_38(self.loop), - ) + async with asyncio_timeout(self.ping_timeout): + await pong_waiter self.logger.debug("% received keepalive pong") except asyncio.TimeoutError: if self.debug: self.logger.debug("! timed out waiting for keepalive pong") - self.fail_connection(1011, "keepalive ping timeout") + self.fail_connection( + CloseCode.INTERNAL_ERROR, + "keepalive ping timeout", + ) break - # Remove this branch when dropping support for Python < 3.8 - # because CancelledError no longer inherits Exception. - except asyncio.CancelledError: - raise - except ConnectionClosed: pass @@ -1297,9 +1309,7 @@ class WebSocketCommonProtocol(asyncio.Protocol): # A client should wait for a TCP close from the server. if self.is_client and hasattr(self, "transfer_data_task"): if await self.wait_for_connection_lost(): - # Coverage marks this line as a partially executed branch. - # I supect a bug in coverage. Ignore it for now. - return # pragma: no cover + return if self.debug: self.logger.debug("! timed out waiting for TCP close") @@ -1317,9 +1327,7 @@ class WebSocketCommonProtocol(asyncio.Protocol): pass if await self.wait_for_connection_lost(): - # Coverage marks this line as a partially executed branch. - # I supect a bug in coverage. Ignore it for now. - return # pragma: no cover + return if self.debug: self.logger.debug("! timed out waiting for TCP close") @@ -1352,12 +1360,11 @@ class WebSocketCommonProtocol(asyncio.Protocol): # Abort the TCP connection. Buffers are discarded. if self.debug: self.logger.debug("x aborting TCP connection") - self.transport.abort() + # Due to a bug in coverage, this is erroneously reported as not covered. + self.transport.abort() # pragma: no cover # connection_lost() is called quickly after aborting. - # Coverage marks this line as a partially executed branch. - # I supect a bug in coverage. Ignore it for now. - await self.wait_for_connection_lost() # pragma: no cover + await self.wait_for_connection_lost() async def wait_for_connection_lost(self) -> bool: """ @@ -1369,11 +1376,8 @@ class WebSocketCommonProtocol(asyncio.Protocol): """ if not self.connection_lost_waiter.done(): try: - await asyncio.wait_for( - asyncio.shield(self.connection_lost_waiter), - self.close_timeout, - **loop_if_py_lt_38(self.loop), - ) + async with asyncio_timeout(self.close_timeout): + await asyncio.shield(self.connection_lost_waiter) except asyncio.TimeoutError: pass # Re-check self.connection_lost_waiter.done() synchronously because @@ -1381,7 +1385,11 @@ class WebSocketCommonProtocol(asyncio.Protocol): # and the moment this coroutine resumes running. return self.connection_lost_waiter.done() - def fail_connection(self, code: int = 1006, reason: str = "") -> None: + def fail_connection( + self, + code: int = CloseCode.ABNORMAL_CLOSURE, + reason: str = "", + ) -> None: """ 7.1.7. Fail the WebSocket Connection @@ -1412,7 +1420,7 @@ class WebSocketCommonProtocol(asyncio.Protocol): # sent if it's CLOSING), except when failing the connection because of # an error reading from or writing to the network. # Don't send a close frame if the connection is broken. - if code != 1006 and self.state is State.OPEN: + if code != CloseCode.ABNORMAL_CLOSURE and self.state is State.OPEN: close = Close(code, reason) # Write the close frame without draining the write buffer. @@ -1449,13 +1457,13 @@ class WebSocketCommonProtocol(asyncio.Protocol): assert self.state is State.CLOSED exc = self.connection_closed_exc() - for ping in self.pings.values(): - ping.set_exception(exc) + for pong_waiter, _ping_timestamp in self.pings.values(): + pong_waiter.set_exception(exc) # If the exception is never retrieved, it will be logged when ping # is garbage-collected. This is confusing for users. # Given that ping is done (with an exception), canceling it does # nothing, but it prevents logging the exception. - ping.cancel() + pong_waiter.cancel() # asyncio.Protocol methods @@ -1496,7 +1504,6 @@ class WebSocketCommonProtocol(asyncio.Protocol): self.connection_lost_waiter.set_result(None) if True: # pragma: no cover - # Copied from asyncio.StreamReaderProtocol if self.reader is not None: if exc is None: @@ -1552,13 +1559,17 @@ class WebSocketCommonProtocol(asyncio.Protocol): self.reader.feed_eof() -def broadcast(websockets: Iterable[WebSocketCommonProtocol], message: Data) -> None: +def broadcast( + websockets: Iterable[WebSocketCommonProtocol], + message: Data, + raise_exceptions: bool = False, +) -> None: """ Broadcast a message to several WebSocket connections. - A string (:class:`str`) is sent as a Text_ frame. A bytestring or - bytes-like object (:class:`bytes`, :class:`bytearray`, or - :class:`memoryview`) is sent as a Binary_ frame. + A string (:class:`str`) is sent as a Text_ frame. A bytestring or bytes-like + object (:class:`bytes`, :class:`bytearray`, or :class:`memoryview`) is sent + as a Binary_ frame. .. _Text: https://www.rfc-editor.org/rfc/rfc6455.html#section-5.6 .. _Binary: https://www.rfc-editor.org/rfc/rfc6455.html#section-5.6 @@ -1566,33 +1577,42 @@ def broadcast(websockets: Iterable[WebSocketCommonProtocol], message: Data) -> N :func:`broadcast` pushes the message synchronously to all connections even if their write buffers are overflowing. There's no backpressure. - :func:`broadcast` skips silently connections that aren't open in order to - avoid errors on connections where the closing handshake is in progress. - - If you broadcast messages faster than a connection can handle them, - messages will pile up in its write buffer until the connection times out. - Keep low values for ``ping_interval`` and ``ping_timeout`` to prevent - excessive memory usage by slow connections when you use :func:`broadcast`. + If you broadcast messages faster than a connection can handle them, messages + will pile up in its write buffer until the connection times out. Keep + ``ping_interval`` and ``ping_timeout`` low to prevent excessive memory usage + from slow connections. Unlike :meth:`~websockets.server.WebSocketServerProtocol.send`, :func:`broadcast` doesn't support sending fragmented messages. Indeed, - fragmentation is useful for sending large messages without buffering - them in memory, while :func:`broadcast` buffers one copy per connection - as fast as possible. + fragmentation is useful for sending large messages without buffering them in + memory, while :func:`broadcast` buffers one copy per connection as fast as + possible. + + :func:`broadcast` skips connections that aren't open in order to avoid + errors on connections where the closing handshake is in progress. + + :func:`broadcast` ignores failures to write the message on some connections. + It continues writing to other connections. On Python 3.11 and above, you + may set ``raise_exceptions`` to :obj:`True` to record failures and raise all + exceptions in a :pep:`654` :exc:`ExceptionGroup`. Args: - websockets (Iterable[WebSocketCommonProtocol]): WebSocket connections - to which the message will be sent. - message (Data): message to send. + websockets: WebSocket connections to which the message will be sent. + message: Message to send. + raise_exceptions: Whether to raise an exception in case of failures. Raises: - RuntimeError: if a connection is busy sending a fragmented message. - TypeError: if ``message`` doesn't have a supported type. + TypeError: If ``message`` doesn't have a supported type. """ if not isinstance(message, (str, bytes, bytearray, memoryview)): raise TypeError("data must be str or bytes-like") + if raise_exceptions: + if sys.version_info[:2] < (3, 11): # pragma: no cover + raise ValueError("raise_exceptions requires at least Python 3.11") + exceptions = [] + opcode, data = prepare_data(message) for websocket in websockets: @@ -1600,6 +1620,26 @@ def broadcast(websockets: Iterable[WebSocketCommonProtocol], message: Data) -> N continue if websocket._fragmented_message_waiter is not None: - raise RuntimeError("busy sending a fragmented message") + if raise_exceptions: + exception = RuntimeError("sending a fragmented message") + exceptions.append(exception) + else: + websocket.logger.warning( + "skipped broadcast: sending a fragmented message", + ) + + try: + websocket.write_frame_sync(True, opcode, data) + except Exception as write_exception: + if raise_exceptions: + exception = RuntimeError("failed to write message") + exception.__cause__ = write_exception + exceptions.append(exception) + else: + websocket.logger.warning( + "skipped broadcast: failed to write message", + exc_info=True, + ) - websocket.write_frame_sync(True, opcode, data) + if raise_exceptions: + raise ExceptionGroup("skipped broadcast", exceptions) diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/server.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/server.py index 3e51db1b71..7c24dd74af 100644 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/server.py +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/legacy/server.py @@ -25,7 +25,6 @@ from typing import ( cast, ) -from ..connection import State from ..datastructures import Headers, HeadersLike, MultipleValuesError from ..exceptions import ( AbortHandshake, @@ -45,8 +44,9 @@ from ..headers import ( validate_subprotocols, ) from ..http import USER_AGENT -from ..typing import ExtensionHeader, LoggerLike, Origin, Subprotocol -from .compatibility import loop_if_py_lt_38 +from ..protocol import State +from ..typing import ExtensionHeader, LoggerLike, Origin, StatusLike, Subprotocol +from .compatibility import asyncio_timeout from .handshake import build_response, check_request from .http import read_request from .protocol import WebSocketCommonProtocol @@ -57,7 +57,7 @@ __all__ = ["serve", "unix_serve", "WebSocketServerProtocol", "WebSocketServer"] HeadersLikeOrCallable = Union[HeadersLike, Callable[[str, Headers], HeadersLike]] -HTTPResponse = Tuple[http.HTTPStatus, HeadersLike, bytes] +HTTPResponse = Tuple[StatusLike, HeadersLike, bytes] class WebSocketServerProtocol(WebSocketCommonProtocol): @@ -73,7 +73,7 @@ class WebSocketServerProtocol(WebSocketCommonProtocol): await process(message) The iterator exits normally when the connection is closed with close code - 1000 (OK) or 1001 (going away). It raises + 1000 (OK) or 1001 (going away) or without a close code. It raises a :exc:`~websockets.exceptions.ConnectionClosedError` when the connection is closed with any other code. @@ -84,7 +84,7 @@ class WebSocketServerProtocol(WebSocketCommonProtocol): ws_server: WebSocket server that created this connection. See :func:`serve` for the documentation of ``ws_handler``, ``logger``, ``origins``, - ``extensions``, ``subprotocols``, and ``extra_headers``. + ``extensions``, ``subprotocols``, ``extra_headers``, and ``server_header``. See :class:`~websockets.legacy.protocol.WebSocketCommonProtocol` for the documentation of ``ping_interval``, ``ping_timeout``, ``close_timeout``, @@ -108,12 +108,14 @@ class WebSocketServerProtocol(WebSocketCommonProtocol): extensions: Optional[Sequence[ServerExtensionFactory]] = None, subprotocols: Optional[Sequence[Subprotocol]] = None, extra_headers: Optional[HeadersLikeOrCallable] = None, + server_header: Optional[str] = USER_AGENT, process_request: Optional[ Callable[[str, Headers], Awaitable[Optional[HTTPResponse]]] ] = None, select_subprotocol: Optional[ Callable[[Sequence[Subprotocol], Sequence[Subprotocol]], Subprotocol] ] = None, + open_timeout: Optional[float] = 10, **kwargs: Any, ) -> None: if logger is None: @@ -132,8 +134,10 @@ class WebSocketServerProtocol(WebSocketCommonProtocol): self.available_extensions = extensions self.available_subprotocols = subprotocols self.extra_headers = extra_headers + self.server_header = server_header self._process_request = process_request self._select_subprotocol = select_subprotocol + self.open_timeout = open_timeout def connection_made(self, transport: asyncio.BaseTransport) -> None: """ @@ -153,22 +157,20 @@ class WebSocketServerProtocol(WebSocketCommonProtocol): Handle the lifecycle of a WebSocket connection. Since this method doesn't have a caller able to handle exceptions, it - attemps to log relevant ones and guarantees that the TCP connection is + attempts to log relevant ones and guarantees that the TCP connection is closed before exiting. """ try: - try: - await self.handshake( - origins=self.origins, - available_extensions=self.available_extensions, - available_subprotocols=self.available_subprotocols, - extra_headers=self.extra_headers, - ) - # Remove this branch when dropping support for Python < 3.8 - # because CancelledError no longer inherits Exception. - except asyncio.CancelledError: # pragma: no cover + async with asyncio_timeout(self.open_timeout): + await self.handshake( + origins=self.origins, + available_extensions=self.available_extensions, + available_subprotocols=self.available_subprotocols, + extra_headers=self.extra_headers, + ) + except asyncio.TimeoutError: # pragma: no cover raise except ConnectionError: raise @@ -216,14 +218,16 @@ class WebSocketServerProtocol(WebSocketCommonProtocol): ) headers.setdefault("Date", email.utils.formatdate(usegmt=True)) - headers.setdefault("Server", USER_AGENT) + if self.server_header is not None: + headers.setdefault("Server", self.server_header) + headers.setdefault("Content-Length", str(len(body))) headers.setdefault("Content-Type", "text/plain") headers.setdefault("Connection", "close") self.write_http_response(status, headers, body) self.logger.info( - "connection failed (%d %s)", status.value, status.phrase + "connection rejected (%d %s)", status.value, status.phrase ) await self.close_transport() return @@ -325,9 +329,9 @@ class WebSocketServerProtocol(WebSocketCommonProtocol): You may override this method in a :class:`WebSocketServerProtocol` subclass, for example: - * to return a HTTP 200 OK response on a given path; then a load + * to return an HTTP 200 OK response on a given path; then a load balancer can use this path for a health check; - * to authenticate the request and return a HTTP 401 Unauthorized or a + * to authenticate the request and return an HTTP 401 Unauthorized or an HTTP 403 Forbidden when authentication fails. You may also override this method with the ``process_request`` @@ -345,7 +349,7 @@ class WebSocketServerProtocol(WebSocketCommonProtocol): request_headers: request headers. Returns: - Optional[Tuple[http.HTTPStatus, HeadersLike, bytes]]: :obj:`None` + Optional[Tuple[StatusLike, HeadersLike, bytes]]: :obj:`None` to continue the WebSocket handshake normally. An HTTP response, represented by a 3-uple of the response status, @@ -439,15 +443,12 @@ class WebSocketServerProtocol(WebSocketCommonProtocol): header_values = headers.get_all("Sec-WebSocket-Extensions") if header_values and available_extensions: - parsed_header_values: List[ExtensionHeader] = sum( [parse_extension(header_value) for header_value in header_values], [] ) for name, request_params in parsed_header_values: - for ext_factory in available_extensions: - # Skip non-matching extensions based on their name. if ext_factory.name != name: continue @@ -499,7 +500,6 @@ class WebSocketServerProtocol(WebSocketCommonProtocol): header_values = headers.get_all("Sec-WebSocket-Protocol") if header_values and available_subprotocols: - parsed_header_values: List[Subprotocol] = sum( [parse_subprotocol(header_value) for header_value in header_values], [] ) @@ -516,31 +516,29 @@ class WebSocketServerProtocol(WebSocketCommonProtocol): server_subprotocols: Sequence[Subprotocol], ) -> Optional[Subprotocol]: """ - Pick a subprotocol among those offered by the client. + Pick a subprotocol among those supported by the client and the server. - If several subprotocols are supported by the client and the server, - the default implementation selects the preferred subprotocol by - giving equal value to the priorities of the client and the server. - If no subprotocol is supported by the client and the server, it - proceeds without a subprotocol. + If several subprotocols are available, select the preferred subprotocol + by giving equal weight to the preferences of the client and the server. - This is unlikely to be the most useful implementation in practice. - Many servers providing a subprotocol will require that the client - uses that subprotocol. Such rules can be implemented in a subclass. + If no subprotocol is available, proceed without a subprotocol. - You may also override this method with the ``select_subprotocol`` - argument of :func:`serve` and :class:`WebSocketServerProtocol`. + You may provide a ``select_subprotocol`` argument to :func:`serve` or + :class:`WebSocketServerProtocol` to override this logic. For example, + you could reject the handshake if the client doesn't support a + particular subprotocol, rather than accept the handshake without that + subprotocol. Args: client_subprotocols: list of subprotocols offered by the client. server_subprotocols: list of subprotocols available on the server. Returns: - Optional[Subprotocol]: Selected subprotocol. + Optional[Subprotocol]: Selected subprotocol, if a common subprotocol + was found. :obj:`None` to continue without a subprotocol. - """ if self._select_subprotocol is not None: return self._select_subprotocol(client_subprotocols, server_subprotocols) @@ -548,10 +546,10 @@ class WebSocketServerProtocol(WebSocketCommonProtocol): subprotocols = set(client_subprotocols) & set(server_subprotocols) if not subprotocols: return None - priority = lambda p: ( - client_subprotocols.index(p) + server_subprotocols.index(p) - ) - return sorted(subprotocols, key=priority)[0] + return sorted( + subprotocols, + key=lambda p: client_subprotocols.index(p) + server_subprotocols.index(p), + )[0] async def handshake( self, @@ -594,7 +592,8 @@ class WebSocketServerProtocol(WebSocketCommonProtocol): # The connection may drop while process_request is running. if self.state is State.CLOSED: - raise self.connection_closed_exc() # pragma: no cover + # This subclass of ConnectionError is silently ignored in handler(). + raise BrokenPipeError("connection closed during opening handshake") # Change the response to a 503 error if the server is shutting down. if not self.ws_server.is_serving(): @@ -635,7 +634,8 @@ class WebSocketServerProtocol(WebSocketCommonProtocol): response_headers.update(extra_headers) response_headers.setdefault("Date", email.utils.formatdate(usegmt=True)) - response_headers.setdefault("Server", USER_AGENT) + if self.server_header is not None: + response_headers.setdefault("Server", self.server_header) self.write_http_response(http.HTTPStatus.SWITCHING_PROTOCOLS, response_headers) @@ -658,9 +658,9 @@ class WebSocketServer: when shutting down. Args: - logger: logger for this server; - defaults to ``logging.getLogger("websockets.server")``; - see the :doc:`logging guide <../topics/logging>` for details. + logger: Logger for this server. + It defaults to ``logging.getLogger("websockets.server")``. + See the :doc:`logging guide <../../topics/logging>` for details. """ @@ -707,7 +707,7 @@ class WebSocketServer: self.logger.info("server listening on %s", name) # Initialized here because we need a reference to the event loop. - # This should be moved back to __init__ in Python 3.10. + # This should be moved back to __init__ when dropping Python < 3.10. self.closed_waiter = server.get_loop().create_future() def register(self, protocol: WebSocketServerProtocol) -> None: @@ -724,26 +724,30 @@ class WebSocketServer: """ self.websockets.remove(protocol) - def close(self) -> None: + def close(self, close_connections: bool = True) -> None: """ Close the server. - This method: + * Close the underlying :class:`~asyncio.Server`. + * When ``close_connections`` is :obj:`True`, which is the default, + close existing connections. Specifically: - * closes the underlying :class:`~asyncio.Server`; - * rejects new WebSocket connections with an HTTP 503 (service - unavailable) error; this happens when the server accepted the TCP - connection but didn't complete the WebSocket opening handshake prior - to closing; - * closes open WebSocket connections with close code 1001 (going away). + * Reject opening WebSocket connections with an HTTP 503 (service + unavailable) error. This happens when the server accepted the TCP + connection but didn't complete the opening handshake before closing. + * Close open WebSocket connections with close code 1001 (going away). + + * Wait until all connection handlers terminate. :meth:`close` is idempotent. """ if self.close_task is None: - self.close_task = self.get_loop().create_task(self._close()) + self.close_task = self.get_loop().create_task( + self._close(close_connections) + ) - async def _close(self) -> None: + async def _close(self, close_connections: bool) -> None: """ Implementation of :meth:`close`. @@ -757,36 +761,30 @@ class WebSocketServer: # Stop accepting new connections. self.server.close() - # Wait until self.server.close() completes. - await self.server.wait_closed() - # Wait until all accepted connections reach connection_made() and call # register(). See https://bugs.python.org/issue34852 for details. - await asyncio.sleep(0, **loop_if_py_lt_38(self.get_loop())) - - # Close OPEN connections with status code 1001. Since the server was - # closed, handshake() closes OPENING connections with a HTTP 503 - # error. Wait until all connections are closed. - - close_tasks = [ - asyncio.create_task(websocket.close(1001)) - for websocket in self.websockets - if websocket.state is not State.CONNECTING - ] - # asyncio.wait doesn't accept an empty first argument. - if close_tasks: - await asyncio.wait( - close_tasks, - **loop_if_py_lt_38(self.get_loop()), - ) - - # Wait until all connection handlers are complete. + await asyncio.sleep(0) + + if close_connections: + # Close OPEN connections with close code 1001. After server.close(), + # handshake() closes OPENING connections with an HTTP 503 error. + close_tasks = [ + asyncio.create_task(websocket.close(1001)) + for websocket in self.websockets + if websocket.state is not State.CONNECTING + ] + # asyncio.wait doesn't accept an empty first argument. + if close_tasks: + await asyncio.wait(close_tasks) + + # Wait until all TCP connections are closed. + await self.server.wait_closed() + # Wait until all connection handlers terminate. # asyncio.wait doesn't accept an empty first argument. if self.websockets: await asyncio.wait( - [websocket.handler_task for websocket in self.websockets], - **loop_if_py_lt_38(self.get_loop()), + [websocket.handler_task for websocket in self.websockets] ) # Tell wait_closed() to return. @@ -829,19 +827,37 @@ class WebSocketServer: """ return self.server.is_serving() - async def start_serving(self) -> None: + async def start_serving(self) -> None: # pragma: no cover """ See :meth:`asyncio.Server.start_serving`. + Typical use:: + + server = await serve(..., start_serving=False) + # perform additional setup here... + # ... then start the server + await server.start_serving() + """ - await self.server.start_serving() # pragma: no cover + await self.server.start_serving() - async def serve_forever(self) -> None: + async def serve_forever(self) -> None: # pragma: no cover """ See :meth:`asyncio.Server.serve_forever`. + Typical use:: + + server = await serve(...) + # this coroutine doesn't return + # canceling it stops the server + await server.serve_forever() + + This is an alternative to using :func:`serve` as an asynchronous context + manager. Shutdown is triggered by canceling :meth:`serve_forever` + instead of exiting a :func:`serve` context. + """ - await self.server.serve_forever() # pragma: no cover + await self.server.serve_forever() @property def sockets(self) -> Iterable[socket.socket]: @@ -851,17 +867,17 @@ class WebSocketServer: """ return self.server.sockets - async def __aenter__(self) -> WebSocketServer: - return self # pragma: no cover + async def __aenter__(self) -> WebSocketServer: # pragma: no cover + return self async def __aexit__( self, exc_type: Optional[Type[BaseException]], exc_value: Optional[BaseException], traceback: Optional[TracebackType], - ) -> None: - self.close() # pragma: no cover - await self.wait_closed() # pragma: no cover + ) -> None: # pragma: no cover + self.close() + await self.wait_closed() class Serve: @@ -879,53 +895,61 @@ class Serve: server performs the closing handshake and closes the connection. Awaiting :func:`serve` yields a :class:`WebSocketServer`. This object - provides :meth:`~WebSocketServer.close` and - :meth:`~WebSocketServer.wait_closed` methods for shutting down the server. + provides a :meth:`~WebSocketServer.close` method to shut down the server:: - :func:`serve` can be used as an asynchronous context manager:: + stop = asyncio.Future() # set this future to exit the server + + server = await serve(...) + await stop + await server.close() + + :func:`serve` can be used as an asynchronous context manager. Then, the + server is shut down automatically when exiting the context:: stop = asyncio.Future() # set this future to exit the server async with serve(...): await stop - The server is shut down automatically when exiting the context. - Args: - ws_handler: connection handler. It receives the WebSocket connection, + ws_handler: Connection handler. It receives the WebSocket connection, which is a :class:`WebSocketServerProtocol`, in argument. - host: network interfaces the server is bound to; - see :meth:`~asyncio.loop.create_server` for details. - port: TCP port the server listens on; - see :meth:`~asyncio.loop.create_server` for details. - create_protocol: factory for the :class:`asyncio.Protocol` managing - the connection; defaults to :class:`WebSocketServerProtocol`; may - be set to a wrapper or a subclass to customize connection handling. - logger: logger for this server; - defaults to ``logging.getLogger("websockets.server")``; - see the :doc:`logging guide <../topics/logging>` for details. - compression: shortcut that enables the "permessage-deflate" extension - by default; may be set to :obj:`None` to disable compression; - see the :doc:`compression guide <../topics/compression>` for details. - origins: acceptable values of the ``Origin`` header; include - :obj:`None` in the list if the lack of an origin is acceptable. - This is useful for defending against Cross-Site WebSocket - Hijacking attacks. - extensions: list of supported extensions, in order in which they - should be tried. - subprotocols: list of supported subprotocols, in order of decreasing + host: Network interfaces the server binds to. + See :meth:`~asyncio.loop.create_server` for details. + port: TCP port the server listens on. + See :meth:`~asyncio.loop.create_server` for details. + create_protocol: Factory for the :class:`asyncio.Protocol` managing + the connection. It defaults to :class:`WebSocketServerProtocol`. + Set it to a wrapper or a subclass to customize connection handling. + logger: Logger for this server. + It defaults to ``logging.getLogger("websockets.server")``. + See the :doc:`logging guide <../../topics/logging>` for details. + compression: The "permessage-deflate" extension is enabled by default. + Set ``compression`` to :obj:`None` to disable it. See the + :doc:`compression guide <../../topics/compression>` for details. + origins: Acceptable values of the ``Origin`` header, for defending + against Cross-Site WebSocket Hijacking attacks. Include :obj:`None` + in the list if the lack of an origin is acceptable. + extensions: List of supported extensions, in order in which they + should be negotiated and run. + subprotocols: List of supported subprotocols, in order of decreasing preference. extra_headers (Union[HeadersLike, Callable[[str, Headers], HeadersLike]]): - arbitrary HTTP headers to add to the request; this can be + Arbitrary HTTP headers to add to the response. This can be a :data:`~websockets.datastructures.HeadersLike` or a callable taking the request path and headers in arguments and returning a :data:`~websockets.datastructures.HeadersLike`. + server_header: Value of the ``Server`` response header. + It defaults to ``"Python/x.y.z websockets/X.Y"``. + Setting it to :obj:`None` removes the header. process_request (Optional[Callable[[str, Headers], \ - Awaitable[Optional[Tuple[http.HTTPStatus, HeadersLike, bytes]]]]]): - intercept HTTP request before the opening handshake; - see :meth:`~WebSocketServerProtocol.process_request` for details. - select_subprotocol: select a subprotocol supported by the client; - see :meth:`~WebSocketServerProtocol.select_subprotocol` for details. + Awaitable[Optional[Tuple[StatusLike, HeadersLike, bytes]]]]]): + Intercept HTTP request before the opening handshake. + See :meth:`~WebSocketServerProtocol.process_request` for details. + select_subprotocol: Select a subprotocol supported by the client. + See :meth:`~WebSocketServerProtocol.select_subprotocol` for details. + open_timeout: Timeout for opening connections in seconds. + :obj:`None` disables the timeout. See :class:`~websockets.legacy.protocol.WebSocketCommonProtocol` for the documentation of ``ping_interval``, ``ping_timeout``, ``close_timeout``, @@ -955,19 +979,21 @@ class Serve: host: Optional[Union[str, Sequence[str]]] = None, port: Optional[int] = None, *, - create_protocol: Optional[Callable[[Any], WebSocketServerProtocol]] = None, + create_protocol: Optional[Callable[..., WebSocketServerProtocol]] = None, logger: Optional[LoggerLike] = None, compression: Optional[str] = "deflate", origins: Optional[Sequence[Optional[Origin]]] = None, extensions: Optional[Sequence[ServerExtensionFactory]] = None, subprotocols: Optional[Sequence[Subprotocol]] = None, extra_headers: Optional[HeadersLikeOrCallable] = None, + server_header: Optional[str] = USER_AGENT, process_request: Optional[ Callable[[str, Headers], Awaitable[Optional[HTTPResponse]]] ] = None, select_subprotocol: Optional[ Callable[[Sequence[Subprotocol], Sequence[Subprotocol]], Subprotocol] ] = None, + open_timeout: Optional[float] = 10, ping_interval: Optional[float] = 20, ping_timeout: Optional[float] = 20, close_timeout: Optional[float] = None, @@ -1030,6 +1056,7 @@ class Serve: host=host, port=port, secure=secure, + open_timeout=open_timeout, ping_interval=ping_interval, ping_timeout=ping_timeout, close_timeout=close_timeout, @@ -1043,6 +1070,7 @@ class Serve: extensions=extensions, subprotocols=subprotocols, extra_headers=extra_headers, + server_header=server_header, process_request=process_request, select_subprotocol=select_subprotocol, logger=logger, @@ -1106,17 +1134,18 @@ def unix_serve( **kwargs: Any, ) -> Serve: """ - Similar to :func:`serve`, but for listening on Unix sockets. + Start a WebSocket server listening on a Unix socket. - This function builds upon the event - loop's :meth:`~asyncio.loop.create_unix_server` method. + This function is identical to :func:`serve`, except the ``host`` and + ``port`` arguments are replaced by ``path``. It is only available on Unix. - It is only available on Unix. + Unrecognized keyword arguments are passed the event loop's + :meth:`~asyncio.loop.create_unix_server` method. It's useful for deploying a server behind a reverse proxy such as nginx. Args: - path: file system path to the Unix socket. + path: File system path to the Unix socket. """ return serve(ws_handler, path=path, unix=True, **kwargs) diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/protocol.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/protocol.py new file mode 100644 index 0000000000..765e6b9bb4 --- /dev/null +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/protocol.py @@ -0,0 +1,708 @@ +from __future__ import annotations + +import enum +import logging +import uuid +from typing import Generator, List, Optional, Type, Union + +from .exceptions import ( + ConnectionClosed, + ConnectionClosedError, + ConnectionClosedOK, + InvalidState, + PayloadTooBig, + ProtocolError, +) +from .extensions import Extension +from .frames import ( + OK_CLOSE_CODES, + OP_BINARY, + OP_CLOSE, + OP_CONT, + OP_PING, + OP_PONG, + OP_TEXT, + Close, + CloseCode, + Frame, +) +from .http11 import Request, Response +from .streams import StreamReader +from .typing import LoggerLike, Origin, Subprotocol + + +__all__ = [ + "Protocol", + "Side", + "State", + "SEND_EOF", +] + +Event = Union[Request, Response, Frame] +"""Events that :meth:`~Protocol.events_received` may return.""" + + +class Side(enum.IntEnum): + """A WebSocket connection is either a server or a client.""" + + SERVER, CLIENT = range(2) + + +SERVER = Side.SERVER +CLIENT = Side.CLIENT + + +class State(enum.IntEnum): + """A WebSocket connection is in one of these four states.""" + + CONNECTING, OPEN, CLOSING, CLOSED = range(4) + + +CONNECTING = State.CONNECTING +OPEN = State.OPEN +CLOSING = State.CLOSING +CLOSED = State.CLOSED + + +SEND_EOF = b"" +"""Sentinel signaling that the TCP connection must be half-closed.""" + + +class Protocol: + """ + Sans-I/O implementation of a WebSocket connection. + + Args: + side: :attr:`~Side.CLIENT` or :attr:`~Side.SERVER`. + state: initial state of the WebSocket connection. + max_size: maximum size of incoming messages in bytes; + :obj:`None` disables the limit. + logger: logger for this connection; depending on ``side``, + defaults to ``logging.getLogger("websockets.client")`` + or ``logging.getLogger("websockets.server")``; + see the :doc:`logging guide <../../topics/logging>` for details. + + """ + + def __init__( + self, + side: Side, + *, + state: State = OPEN, + max_size: Optional[int] = 2**20, + logger: Optional[LoggerLike] = None, + ) -> None: + # Unique identifier. For logs. + self.id: uuid.UUID = uuid.uuid4() + """Unique identifier of the connection. Useful in logs.""" + + # Logger or LoggerAdapter for this connection. + if logger is None: + logger = logging.getLogger(f"websockets.{side.name.lower()}") + self.logger: LoggerLike = logger + """Logger for this connection.""" + + # Track if DEBUG is enabled. Shortcut logging calls if it isn't. + self.debug = logger.isEnabledFor(logging.DEBUG) + + # Connection side. CLIENT or SERVER. + self.side = side + + # Connection state. Initially OPEN because subclasses handle CONNECTING. + self.state = state + + # Maximum size of incoming messages in bytes. + self.max_size = max_size + + # Current size of incoming message in bytes. Only set while reading a + # fragmented message i.e. a data frames with the FIN bit not set. + self.cur_size: Optional[int] = None + + # True while sending a fragmented message i.e. a data frames with the + # FIN bit not set. + self.expect_continuation_frame = False + + # WebSocket protocol parameters. + self.origin: Optional[Origin] = None + self.extensions: List[Extension] = [] + self.subprotocol: Optional[Subprotocol] = None + + # Close code and reason, set when a close frame is sent or received. + self.close_rcvd: Optional[Close] = None + self.close_sent: Optional[Close] = None + self.close_rcvd_then_sent: Optional[bool] = None + + # Track if an exception happened during the handshake. + self.handshake_exc: Optional[Exception] = None + """ + Exception to raise if the opening handshake failed. + + :obj:`None` if the opening handshake succeeded. + + """ + + # Track if send_eof() was called. + self.eof_sent = False + + # Parser state. + self.reader = StreamReader() + self.events: List[Event] = [] + self.writes: List[bytes] = [] + self.parser = self.parse() + next(self.parser) # start coroutine + self.parser_exc: Optional[Exception] = None + + @property + def state(self) -> State: + """ + WebSocket connection state. + + Defined in 4.1, 4.2, 7.1.3, and 7.1.4 of :rfc:`6455`. + + """ + return self._state + + @state.setter + def state(self, state: State) -> None: + if self.debug: + self.logger.debug("= connection is %s", state.name) + self._state = state + + @property + def close_code(self) -> Optional[int]: + """ + `WebSocket close code`_. + + .. _WebSocket close code: + https://www.rfc-editor.org/rfc/rfc6455.html#section-7.1.5 + + :obj:`None` if the connection isn't closed yet. + + """ + if self.state is not CLOSED: + return None + elif self.close_rcvd is None: + return CloseCode.ABNORMAL_CLOSURE + else: + return self.close_rcvd.code + + @property + def close_reason(self) -> Optional[str]: + """ + `WebSocket close reason`_. + + .. _WebSocket close reason: + https://www.rfc-editor.org/rfc/rfc6455.html#section-7.1.6 + + :obj:`None` if the connection isn't closed yet. + + """ + if self.state is not CLOSED: + return None + elif self.close_rcvd is None: + return "" + else: + return self.close_rcvd.reason + + @property + def close_exc(self) -> ConnectionClosed: + """ + Exception to raise when trying to interact with a closed connection. + + Don't raise this exception while the connection :attr:`state` + is :attr:`~websockets.protocol.State.CLOSING`; wait until + it's :attr:`~websockets.protocol.State.CLOSED`. + + Indeed, the exception includes the close code and reason, which are + known only once the connection is closed. + + Raises: + AssertionError: if the connection isn't closed yet. + + """ + assert self.state is CLOSED, "connection isn't closed yet" + exc_type: Type[ConnectionClosed] + if ( + self.close_rcvd is not None + and self.close_sent is not None + and self.close_rcvd.code in OK_CLOSE_CODES + and self.close_sent.code in OK_CLOSE_CODES + ): + exc_type = ConnectionClosedOK + else: + exc_type = ConnectionClosedError + exc: ConnectionClosed = exc_type( + self.close_rcvd, + self.close_sent, + self.close_rcvd_then_sent, + ) + # Chain to the exception raised in the parser, if any. + exc.__cause__ = self.parser_exc + return exc + + # Public methods for receiving data. + + def receive_data(self, data: bytes) -> None: + """ + Receive data from the network. + + After calling this method: + + - You must call :meth:`data_to_send` and send this data to the network. + - You should call :meth:`events_received` and process resulting events. + + Raises: + EOFError: if :meth:`receive_eof` was called earlier. + + """ + self.reader.feed_data(data) + next(self.parser) + + def receive_eof(self) -> None: + """ + Receive the end of the data stream from the network. + + After calling this method: + + - You must call :meth:`data_to_send` and send this data to the network; + it will return ``[b""]``, signaling the end of the stream, or ``[]``. + - You aren't expected to call :meth:`events_received`; it won't return + any new events. + + Raises: + EOFError: if :meth:`receive_eof` was called earlier. + + """ + self.reader.feed_eof() + next(self.parser) + + # Public methods for sending events. + + def send_continuation(self, data: bytes, fin: bool) -> None: + """ + Send a `Continuation frame`_. + + .. _Continuation frame: + https://datatracker.ietf.org/doc/html/rfc6455#section-5.6 + + Parameters: + data: payload containing the same kind of data + as the initial frame. + fin: FIN bit; set it to :obj:`True` if this is the last frame + of a fragmented message and to :obj:`False` otherwise. + + Raises: + ProtocolError: if a fragmented message isn't in progress. + + """ + if not self.expect_continuation_frame: + raise ProtocolError("unexpected continuation frame") + self.expect_continuation_frame = not fin + self.send_frame(Frame(OP_CONT, data, fin)) + + def send_text(self, data: bytes, fin: bool = True) -> None: + """ + Send a `Text frame`_. + + .. _Text frame: + https://datatracker.ietf.org/doc/html/rfc6455#section-5.6 + + Parameters: + data: payload containing text encoded with UTF-8. + fin: FIN bit; set it to :obj:`False` if this is the first frame of + a fragmented message. + + Raises: + ProtocolError: if a fragmented message is in progress. + + """ + if self.expect_continuation_frame: + raise ProtocolError("expected a continuation frame") + self.expect_continuation_frame = not fin + self.send_frame(Frame(OP_TEXT, data, fin)) + + def send_binary(self, data: bytes, fin: bool = True) -> None: + """ + Send a `Binary frame`_. + + .. _Binary frame: + https://datatracker.ietf.org/doc/html/rfc6455#section-5.6 + + Parameters: + data: payload containing arbitrary binary data. + fin: FIN bit; set it to :obj:`False` if this is the first frame of + a fragmented message. + + Raises: + ProtocolError: if a fragmented message is in progress. + + """ + if self.expect_continuation_frame: + raise ProtocolError("expected a continuation frame") + self.expect_continuation_frame = not fin + self.send_frame(Frame(OP_BINARY, data, fin)) + + def send_close(self, code: Optional[int] = None, reason: str = "") -> None: + """ + Send a `Close frame`_. + + .. _Close frame: + https://datatracker.ietf.org/doc/html/rfc6455#section-5.5.1 + + Parameters: + code: close code. + reason: close reason. + + Raises: + ProtocolError: if a fragmented message is being sent, if the code + isn't valid, or if a reason is provided without a code + + """ + if self.expect_continuation_frame: + raise ProtocolError("expected a continuation frame") + if code is None: + if reason != "": + raise ProtocolError("cannot send a reason without a code") + close = Close(CloseCode.NO_STATUS_RCVD, "") + data = b"" + else: + close = Close(code, reason) + data = close.serialize() + # send_frame() guarantees that self.state is OPEN at this point. + # 7.1.3. The WebSocket Closing Handshake is Started + self.send_frame(Frame(OP_CLOSE, data)) + self.close_sent = close + self.state = CLOSING + + def send_ping(self, data: bytes) -> None: + """ + Send a `Ping frame`_. + + .. _Ping frame: + https://datatracker.ietf.org/doc/html/rfc6455#section-5.5.2 + + Parameters: + data: payload containing arbitrary binary data. + + """ + self.send_frame(Frame(OP_PING, data)) + + def send_pong(self, data: bytes) -> None: + """ + Send a `Pong frame`_. + + .. _Pong frame: + https://datatracker.ietf.org/doc/html/rfc6455#section-5.5.3 + + Parameters: + data: payload containing arbitrary binary data. + + """ + self.send_frame(Frame(OP_PONG, data)) + + def fail(self, code: int, reason: str = "") -> None: + """ + `Fail the WebSocket connection`_. + + .. _Fail the WebSocket connection: + https://datatracker.ietf.org/doc/html/rfc6455#section-7.1.7 + + Parameters: + code: close code + reason: close reason + + Raises: + ProtocolError: if the code isn't valid. + """ + # 7.1.7. Fail the WebSocket Connection + + # Send a close frame when the state is OPEN (a close frame was already + # sent if it's CLOSING), except when failing the connection because + # of an error reading from or writing to the network. + if self.state is OPEN: + if code != CloseCode.ABNORMAL_CLOSURE: + close = Close(code, reason) + data = close.serialize() + self.send_frame(Frame(OP_CLOSE, data)) + self.close_sent = close + self.state = CLOSING + + # When failing the connection, a server closes the TCP connection + # without waiting for the client to complete the handshake, while a + # client waits for the server to close the TCP connection, possibly + # after sending a close frame that the client will ignore. + if self.side is SERVER and not self.eof_sent: + self.send_eof() + + # 7.1.7. Fail the WebSocket Connection "An endpoint MUST NOT continue + # to attempt to process data(including a responding Close frame) from + # the remote endpoint after being instructed to _Fail the WebSocket + # Connection_." + self.parser = self.discard() + next(self.parser) # start coroutine + + # Public method for getting incoming events after receiving data. + + def events_received(self) -> List[Event]: + """ + Fetch events generated from data received from the network. + + Call this method immediately after any of the ``receive_*()`` methods. + + Process resulting events, likely by passing them to the application. + + Returns: + List[Event]: Events read from the connection. + """ + events, self.events = self.events, [] + return events + + # Public method for getting outgoing data after receiving data or sending events. + + def data_to_send(self) -> List[bytes]: + """ + Obtain data to send to the network. + + Call this method immediately after any of the ``receive_*()``, + ``send_*()``, or :meth:`fail` methods. + + Write resulting data to the connection. + + The empty bytestring :data:`~websockets.protocol.SEND_EOF` signals + the end of the data stream. When you receive it, half-close the TCP + connection. + + Returns: + List[bytes]: Data to write to the connection. + + """ + writes, self.writes = self.writes, [] + return writes + + def close_expected(self) -> bool: + """ + Tell if the TCP connection is expected to close soon. + + Call this method immediately after any of the ``receive_*()``, + ``send_close()``, or :meth:`fail` methods. + + If it returns :obj:`True`, schedule closing the TCP connection after a + short timeout if the other side hasn't already closed it. + + Returns: + bool: Whether the TCP connection is expected to close soon. + + """ + # We expect a TCP close if and only if we sent a close frame: + # * Normal closure: once we send a close frame, we expect a TCP close: + # server waits for client to complete the TCP closing handshake; + # client waits for server to initiate the TCP closing handshake. + # * Abnormal closure: we always send a close frame and the same logic + # applies, except on EOFError where we don't send a close frame + # because we already received the TCP close, so we don't expect it. + # We already got a TCP Close if and only if the state is CLOSED. + return self.state is CLOSING or self.handshake_exc is not None + + # Private methods for receiving data. + + def parse(self) -> Generator[None, None, None]: + """ + Parse incoming data into frames. + + :meth:`receive_data` and :meth:`receive_eof` run this generator + coroutine until it needs more data or reaches EOF. + + :meth:`parse` never raises an exception. Instead, it sets the + :attr:`parser_exc` and yields control. + + """ + try: + while True: + if (yield from self.reader.at_eof()): + if self.debug: + self.logger.debug("< EOF") + # If the WebSocket connection is closed cleanly, with a + # closing handhshake, recv_frame() substitutes parse() + # with discard(). This branch is reached only when the + # connection isn't closed cleanly. + raise EOFError("unexpected end of stream") + + if self.max_size is None: + max_size = None + elif self.cur_size is None: + max_size = self.max_size + else: + max_size = self.max_size - self.cur_size + + # During a normal closure, execution ends here on the next + # iteration of the loop after receiving a close frame. At + # this point, recv_frame() replaced parse() by discard(). + frame = yield from Frame.parse( + self.reader.read_exact, + mask=self.side is SERVER, + max_size=max_size, + extensions=self.extensions, + ) + + if self.debug: + self.logger.debug("< %s", frame) + + self.recv_frame(frame) + + except ProtocolError as exc: + self.fail(CloseCode.PROTOCOL_ERROR, str(exc)) + self.parser_exc = exc + + except EOFError as exc: + self.fail(CloseCode.ABNORMAL_CLOSURE, str(exc)) + self.parser_exc = exc + + except UnicodeDecodeError as exc: + self.fail(CloseCode.INVALID_DATA, f"{exc.reason} at position {exc.start}") + self.parser_exc = exc + + except PayloadTooBig as exc: + self.fail(CloseCode.MESSAGE_TOO_BIG, str(exc)) + self.parser_exc = exc + + except Exception as exc: + self.logger.error("parser failed", exc_info=True) + # Don't include exception details, which may be security-sensitive. + self.fail(CloseCode.INTERNAL_ERROR) + self.parser_exc = exc + + # During an abnormal closure, execution ends here after catching an + # exception. At this point, fail() replaced parse() by discard(). + yield + raise AssertionError("parse() shouldn't step after error") + + def discard(self) -> Generator[None, None, None]: + """ + Discard incoming data. + + This coroutine replaces :meth:`parse`: + + - after receiving a close frame, during a normal closure (1.4); + - after sending a close frame, during an abnormal closure (7.1.7). + + """ + # The server close the TCP connection in the same circumstances where + # discard() replaces parse(). The client closes the connection later, + # after the server closes the connection or a timeout elapses. + # (The latter case cannot be handled in this Sans-I/O layer.) + assert (self.side is SERVER) == (self.eof_sent) + while not (yield from self.reader.at_eof()): + self.reader.discard() + if self.debug: + self.logger.debug("< EOF") + # A server closes the TCP connection immediately, while a client + # waits for the server to close the TCP connection. + if self.side is CLIENT: + self.send_eof() + self.state = CLOSED + # If discard() completes normally, execution ends here. + yield + # Once the reader reaches EOF, its feed_data/eof() methods raise an + # error, so our receive_data/eof() methods don't step the generator. + raise AssertionError("discard() shouldn't step after EOF") + + def recv_frame(self, frame: Frame) -> None: + """ + Process an incoming frame. + + """ + if frame.opcode is OP_TEXT or frame.opcode is OP_BINARY: + if self.cur_size is not None: + raise ProtocolError("expected a continuation frame") + if frame.fin: + self.cur_size = None + else: + self.cur_size = len(frame.data) + + elif frame.opcode is OP_CONT: + if self.cur_size is None: + raise ProtocolError("unexpected continuation frame") + if frame.fin: + self.cur_size = None + else: + self.cur_size += len(frame.data) + + elif frame.opcode is OP_PING: + # 5.5.2. Ping: "Upon receipt of a Ping frame, an endpoint MUST + # send a Pong frame in response" + pong_frame = Frame(OP_PONG, frame.data) + self.send_frame(pong_frame) + + elif frame.opcode is OP_PONG: + # 5.5.3 Pong: "A response to an unsolicited Pong frame is not + # expected." + pass + + elif frame.opcode is OP_CLOSE: + # 7.1.5. The WebSocket Connection Close Code + # 7.1.6. The WebSocket Connection Close Reason + self.close_rcvd = Close.parse(frame.data) + if self.state is CLOSING: + assert self.close_sent is not None + self.close_rcvd_then_sent = False + + if self.cur_size is not None: + raise ProtocolError("incomplete fragmented message") + + # 5.5.1 Close: "If an endpoint receives a Close frame and did + # not previously send a Close frame, the endpoint MUST send a + # Close frame in response. (When sending a Close frame in + # response, the endpoint typically echos the status code it + # received.)" + + if self.state is OPEN: + # Echo the original data instead of re-serializing it with + # Close.serialize() because that fails when the close frame + # is empty and Close.parse() synthesizes a 1005 close code. + # The rest is identical to send_close(). + self.send_frame(Frame(OP_CLOSE, frame.data)) + self.close_sent = self.close_rcvd + self.close_rcvd_then_sent = True + self.state = CLOSING + + # 7.1.2. Start the WebSocket Closing Handshake: "Once an + # endpoint has both sent and received a Close control frame, + # that endpoint SHOULD _Close the WebSocket Connection_" + + # A server closes the TCP connection immediately, while a client + # waits for the server to close the TCP connection. + if self.side is SERVER: + self.send_eof() + + # 1.4. Closing Handshake: "after receiving a control frame + # indicating the connection should be closed, a peer discards + # any further data received." + self.parser = self.discard() + next(self.parser) # start coroutine + + else: + # This can't happen because Frame.parse() validates opcodes. + raise AssertionError(f"unexpected opcode: {frame.opcode:02x}") + + self.events.append(frame) + + # Private methods for sending events. + + def send_frame(self, frame: Frame) -> None: + if self.state is not OPEN: + raise InvalidState( + f"cannot write to a WebSocket in the {self.state.name} state" + ) + + if self.debug: + self.logger.debug("> %s", frame) + self.writes.append( + frame.serialize(mask=self.side is CLIENT, extensions=self.extensions) + ) + + def send_eof(self) -> None: + assert not self.eof_sent + self.eof_sent = True + if self.debug: + self.logger.debug("> EOF") + self.writes.append(SEND_EOF) diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/server.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/server.py index 5dad50b6a1..191660553f 100644 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/server.py +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/server.py @@ -4,9 +4,9 @@ import base64 import binascii import email.utils import http -from typing import Generator, List, Optional, Sequence, Tuple, cast +import warnings +from typing import Any, Callable, Generator, List, Optional, Sequence, Tuple, cast -from .connection import CONNECTING, OPEN, SERVER, Connection, State from .datastructures import Headers, MultipleValuesError from .exceptions import ( InvalidHandshake, @@ -25,13 +25,14 @@ from .headers import ( parse_subprotocol, parse_upgrade, ) -from .http import USER_AGENT from .http11 import Request, Response +from .protocol import CONNECTING, OPEN, SERVER, Protocol, State from .typing import ( ConnectionOption, ExtensionHeader, LoggerLike, Origin, + StatusLike, Subprotocol, UpgradeProtocol, ) @@ -39,13 +40,15 @@ from .utils import accept_key # See #940 for why lazy_import isn't used here for backwards compatibility. -from .legacy.server import * # isort:skip # noqa +# See #1400 for why listing compatibility imports in __all__ helps PyCharm. +from .legacy.server import * # isort:skip # noqa: I001 +from .legacy.server import __all__ as legacy__all__ -__all__ = ["ServerConnection"] +__all__ = ["ServerProtocol"] + legacy__all__ -class ServerConnection(Connection): +class ServerProtocol(Protocol): """ Sans-I/O implementation of a WebSocket server connection. @@ -58,20 +61,31 @@ class ServerConnection(Connection): should be tried. subprotocols: list of supported subprotocols, in order of decreasing preference. + select_subprotocol: Callback for selecting a subprotocol among + those supported by the client and the server. It has the same + signature as the :meth:`select_subprotocol` method, including a + :class:`ServerProtocol` instance as first argument. state: initial state of the WebSocket connection. max_size: maximum size of incoming messages in bytes; - :obj:`None` to disable the limit. + :obj:`None` disables the limit. logger: logger for this connection; defaults to ``logging.getLogger("websockets.client")``; - see the :doc:`logging guide <../topics/logging>` for details. + see the :doc:`logging guide <../../topics/logging>` for details. """ def __init__( self, + *, origins: Optional[Sequence[Optional[Origin]]] = None, extensions: Optional[Sequence[ServerExtensionFactory]] = None, subprotocols: Optional[Sequence[Subprotocol]] = None, + select_subprotocol: Optional[ + Callable[ + [ServerProtocol, Sequence[Subprotocol]], + Optional[Subprotocol], + ] + ] = None, state: State = CONNECTING, max_size: Optional[int] = 2**20, logger: Optional[LoggerLike] = None, @@ -85,6 +99,14 @@ class ServerConnection(Connection): self.origins = origins self.available_extensions = extensions self.available_subprotocols = subprotocols + if select_subprotocol is not None: + # Bind select_subprotocol then shadow self.select_subprotocol. + # Use setattr to work around https://github.com/python/mypy/issues/2427. + setattr( + self, + "select_subprotocol", + select_subprotocol.__get__(self, self.__class__), + ) def accept(self, request: Request) -> Response: """ @@ -95,13 +117,13 @@ class ServerConnection(Connection): You must send the handshake response with :meth:`send_response`. - You can modify it before sending it, for example to add HTTP headers. + You may modify it before sending it, for example to add HTTP headers. Args: request: WebSocket handshake request event received from the client. Returns: - Response: WebSocket handshake response event to send to the client. + WebSocket handshake response event to send to the client. """ try: @@ -145,6 +167,8 @@ class ServerConnection(Connection): f"Failed to open a WebSocket connection: {exc}.\n", ) except Exception as exc: + # Handle exceptions raised by user-provided select_subprotocol and + # unexpected errors. request._exception = exc self.handshake_exc = exc self.logger.error("opening handshake failed", exc_info=True) @@ -170,13 +194,12 @@ class ServerConnection(Connection): if protocol_header is not None: headers["Sec-WebSocket-Protocol"] = protocol_header - headers["Server"] = USER_AGENT - self.logger.info("connection open") return Response(101, "Switching Protocols", headers) def process_request( - self, request: Request + self, + request: Request, ) -> Tuple[str, Optional[str], Optional[str]]: """ Check a handshake request and negotiate extensions and subprotocol. @@ -274,6 +297,7 @@ class ServerConnection(Connection): Optional[Origin]: origin, if it is acceptable. Raises: + InvalidHandshake: if the Origin header is invalid. InvalidOrigin: if the origin isn't acceptable. """ @@ -298,8 +322,8 @@ class ServerConnection(Connection): Accept or reject each extension proposed in the client request. Negotiate parameters for accepted extensions. - :rfc:`6455` leaves the rules up to the specification of each - :extension. + Per :rfc:`6455`, negotiation rules are defined by the specification of + each extension. To provide this level of flexibility, for each extension proposed by the client, we check for a match with each extension available in the @@ -324,7 +348,7 @@ class ServerConnection(Connection): HTTP response header and list of accepted extensions. Raises: - InvalidHandshake: to abort the handshake with an HTTP 400 error. + InvalidHandshake: if the Sec-WebSocket-Extensions header is invalid. """ response_header_value: Optional[str] = None @@ -335,15 +359,12 @@ class ServerConnection(Connection): header_values = headers.get_all("Sec-WebSocket-Extensions") if header_values and self.available_extensions: - parsed_header_values: List[ExtensionHeader] = sum( [parse_extension(header_value) for header_value in header_values], [] ) for name, request_params in parsed_header_values: - for ext_factory in self.available_extensions: - # Skip non-matching extensions based on their name. if ext_factory.name != name: continue @@ -384,64 +405,83 @@ class ServerConnection(Connection): also the value of the ``Sec-WebSocket-Protocol`` response header. Raises: - InvalidHandshake: to abort the handshake with an HTTP 400 error. + InvalidHandshake: if the Sec-WebSocket-Subprotocol header is invalid. """ - subprotocol: Optional[Subprotocol] = None - - header_values = headers.get_all("Sec-WebSocket-Protocol") - - if header_values and self.available_subprotocols: - - parsed_header_values: List[Subprotocol] = sum( - [parse_subprotocol(header_value) for header_value in header_values], [] - ) - - subprotocol = self.select_subprotocol( - parsed_header_values, self.available_subprotocols - ) + subprotocols: Sequence[Subprotocol] = sum( + [ + parse_subprotocol(header_value) + for header_value in headers.get_all("Sec-WebSocket-Protocol") + ], + [], + ) - return subprotocol + return self.select_subprotocol(subprotocols) def select_subprotocol( self, - client_subprotocols: Sequence[Subprotocol], - server_subprotocols: Sequence[Subprotocol], + subprotocols: Sequence[Subprotocol], ) -> Optional[Subprotocol]: """ Pick a subprotocol among those offered by the client. - If several subprotocols are supported by the client and the server, - the default implementation selects the preferred subprotocols by - giving equal value to the priorities of the client and the server. + If several subprotocols are supported by both the client and the server, + pick the first one in the list declared the server. + + If the server doesn't support any subprotocols, continue without a + subprotocol, regardless of what the client offers. + + If the server supports at least one subprotocol and the client doesn't + offer any, abort the handshake with an HTTP 400 error. - If no common subprotocol is supported by the client and the server, it - proceeds without a subprotocol. + You provide a ``select_subprotocol`` argument to :class:`ServerProtocol` + to override this logic. For example, you could accept the connection + even if client doesn't offer a subprotocol, rather than reject it. - This is unlikely to be the most useful implementation in practice, as - many servers providing a subprotocol will require that the client uses - that subprotocol. + Here's how to negotiate the ``chat`` subprotocol if the client supports + it and continue without a subprotocol otherwise:: + + def select_subprotocol(protocol, subprotocols): + if "chat" in subprotocols: + return "chat" Args: - client_subprotocols: list of subprotocols offered by the client. - server_subprotocols: list of subprotocols available on the server. + subprotocols: list of subprotocols offered by the client. Returns: - Optional[Subprotocol]: Subprotocol, if a common subprotocol was - found. + Optional[Subprotocol]: Selected subprotocol, if a common subprotocol + was found. + + :obj:`None` to continue without a subprotocol. + + Raises: + NegotiationError: custom implementations may raise this exception + to abort the handshake with an HTTP 400 error. """ - subprotocols = set(client_subprotocols) & set(server_subprotocols) - if not subprotocols: + # Server doesn't offer any subprotocols. + if not self.available_subprotocols: # None or empty list return None - priority = lambda p: ( - client_subprotocols.index(p) + server_subprotocols.index(p) + + # Server offers at least one subprotocol but client doesn't offer any. + if not subprotocols: + raise NegotiationError("missing subprotocol") + + # Server and client both offer subprotocols. Look for a shared one. + proposed_subprotocols = set(subprotocols) + for subprotocol in self.available_subprotocols: + if subprotocol in proposed_subprotocols: + return subprotocol + + # No common subprotocol was found. + raise NegotiationError( + "invalid subprotocol; expected one of " + + ", ".join(self.available_subprotocols) ) - return sorted(subprotocols, key=priority)[0] def reject( self, - status: http.HTTPStatus, + status: StatusLike, text: str, ) -> Response: """ @@ -462,6 +502,8 @@ class ServerConnection(Connection): Response: WebSocket handshake response event to send to the client. """ + # If a user passes an int instead of a HTTPStatus, fix it automatically. + status = http.HTTPStatus(status) body = text.encode() headers = Headers( [ @@ -469,16 +511,15 @@ class ServerConnection(Connection): ("Connection", "close"), ("Content-Length", str(len(body))), ("Content-Type", "text/plain; charset=utf-8"), - ("Server", USER_AGENT), ] ) response = Response(status.value, status.phrase, headers, body) # When reject() is called from accept(), handshake_exc is already set. # If a user calls reject(), set handshake_exc to guarantee invariant: - # "handshake_exc is None if and only if opening handshake succeded." + # "handshake_exc is None if and only if opening handshake succeeded." if self.handshake_exc is None: self.handshake_exc = InvalidStatus(response) - self.logger.info("connection failed (%d %s)", status.value, status.phrase) + self.logger.info("connection rejected (%d %s)", status.value, status.phrase) return response def send_response(self, response: Response) -> None: @@ -509,7 +550,16 @@ class ServerConnection(Connection): def parse(self) -> Generator[None, None, None]: if self.state is CONNECTING: - request = yield from Request.parse(self.reader.read_line) + try: + request = yield from Request.parse( + self.reader.read_line, + ) + except Exception as exc: + self.handshake_exc = exc + self.send_eof() + self.parser = self.discard() + next(self.parser) # start coroutine + yield if self.debug: self.logger.debug("< GET %s HTTP/1.1", request.path) @@ -519,3 +569,12 @@ class ServerConnection(Connection): self.events.append(request) yield from super().parse() + + +class ServerConnection(ServerProtocol): + def __init__(self, *args: Any, **kwargs: Any) -> None: + warnings.warn( + "ServerConnection was renamed to ServerProtocol", + DeprecationWarning, + ) + super().__init__(*args, **kwargs) diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/speedups.pyi b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/speedups.pyi new file mode 100644 index 0000000000..821438a064 --- /dev/null +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/speedups.pyi @@ -0,0 +1 @@ +def apply_mask(data: bytes, mask: bytes) -> bytes: ... diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/sync/__init__.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/sync/__init__.py new file mode 100644 index 0000000000..e69de29bb2 --- /dev/null +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/sync/__init__.py diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/sync/client.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/sync/client.py new file mode 100644 index 0000000000..087ff5f569 --- /dev/null +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/sync/client.py @@ -0,0 +1,328 @@ +from __future__ import annotations + +import socket +import ssl +import threading +from typing import Any, Optional, Sequence, Type + +from ..client import ClientProtocol +from ..datastructures import HeadersLike +from ..extensions.base import ClientExtensionFactory +from ..extensions.permessage_deflate import enable_client_permessage_deflate +from ..headers import validate_subprotocols +from ..http import USER_AGENT +from ..http11 import Response +from ..protocol import CONNECTING, OPEN, Event +from ..typing import LoggerLike, Origin, Subprotocol +from ..uri import parse_uri +from .connection import Connection +from .utils import Deadline + + +__all__ = ["connect", "unix_connect", "ClientConnection"] + + +class ClientConnection(Connection): + """ + Threaded implementation of a WebSocket client connection. + + :class:`ClientConnection` provides :meth:`recv` and :meth:`send` methods for + receiving and sending messages. + + It supports iteration to receive messages:: + + for message in websocket: + process(message) + + The iterator exits normally when the connection is closed with close code + 1000 (OK) or 1001 (going away) or without a close code. It raises a + :exc:`~websockets.exceptions.ConnectionClosedError` when the connection is + closed with any other code. + + Args: + socket: Socket connected to a WebSocket server. + protocol: Sans-I/O connection. + close_timeout: Timeout for closing the connection in seconds. + + """ + + def __init__( + self, + socket: socket.socket, + protocol: ClientProtocol, + *, + close_timeout: Optional[float] = 10, + ) -> None: + self.protocol: ClientProtocol + self.response_rcvd = threading.Event() + super().__init__( + socket, + protocol, + close_timeout=close_timeout, + ) + + def handshake( + self, + additional_headers: Optional[HeadersLike] = None, + user_agent_header: Optional[str] = USER_AGENT, + timeout: Optional[float] = None, + ) -> None: + """ + Perform the opening handshake. + + """ + with self.send_context(expected_state=CONNECTING): + self.request = self.protocol.connect() + if additional_headers is not None: + self.request.headers.update(additional_headers) + if user_agent_header is not None: + self.request.headers["User-Agent"] = user_agent_header + self.protocol.send_request(self.request) + + if not self.response_rcvd.wait(timeout): + self.close_socket() + self.recv_events_thread.join() + raise TimeoutError("timed out during handshake") + + if self.response is None: + self.close_socket() + self.recv_events_thread.join() + raise ConnectionError("connection closed during handshake") + + if self.protocol.state is not OPEN: + self.recv_events_thread.join(self.close_timeout) + self.close_socket() + self.recv_events_thread.join() + + if self.protocol.handshake_exc is not None: + raise self.protocol.handshake_exc + + def process_event(self, event: Event) -> None: + """ + Process one incoming event. + + """ + # First event - handshake response. + if self.response is None: + assert isinstance(event, Response) + self.response = event + self.response_rcvd.set() + # Later events - frames. + else: + super().process_event(event) + + def recv_events(self) -> None: + """ + Read incoming data from the socket and process events. + + """ + try: + super().recv_events() + finally: + # If the connection is closed during the handshake, unblock it. + self.response_rcvd.set() + + +def connect( + uri: str, + *, + # TCP/TLS — unix and path are only for unix_connect() + sock: Optional[socket.socket] = None, + ssl_context: Optional[ssl.SSLContext] = None, + server_hostname: Optional[str] = None, + unix: bool = False, + path: Optional[str] = None, + # WebSocket + origin: Optional[Origin] = None, + extensions: Optional[Sequence[ClientExtensionFactory]] = None, + subprotocols: Optional[Sequence[Subprotocol]] = None, + additional_headers: Optional[HeadersLike] = None, + user_agent_header: Optional[str] = USER_AGENT, + compression: Optional[str] = "deflate", + # Timeouts + open_timeout: Optional[float] = 10, + close_timeout: Optional[float] = 10, + # Limits + max_size: Optional[int] = 2**20, + # Logging + logger: Optional[LoggerLike] = None, + # Escape hatch for advanced customization + create_connection: Optional[Type[ClientConnection]] = None, +) -> ClientConnection: + """ + Connect to the WebSocket server at ``uri``. + + This function returns a :class:`ClientConnection` instance, which you can + use to send and receive messages. + + :func:`connect` may be used as a context manager:: + + async with websockets.sync.client.connect(...) as websocket: + ... + + The connection is closed automatically when exiting the context. + + Args: + uri: URI of the WebSocket server. + sock: Preexisting TCP socket. ``sock`` overrides the host and port + from ``uri``. You may call :func:`socket.create_connection` to + create a suitable TCP socket. + ssl_context: Configuration for enabling TLS on the connection. + server_hostname: Host name for the TLS handshake. ``server_hostname`` + overrides the host name from ``uri``. + origin: Value of the ``Origin`` header, for servers that require it. + extensions: List of supported extensions, in order in which they + should be negotiated and run. + subprotocols: List of supported subprotocols, in order of decreasing + preference. + additional_headers (HeadersLike | None): Arbitrary HTTP headers to add + to the handshake request. + user_agent_header: Value of the ``User-Agent`` request header. + It defaults to ``"Python/x.y.z websockets/X.Y"``. + Setting it to :obj:`None` removes the header. + compression: The "permessage-deflate" extension is enabled by default. + Set ``compression`` to :obj:`None` to disable it. See the + :doc:`compression guide <../../topics/compression>` for details. + open_timeout: Timeout for opening the connection in seconds. + :obj:`None` disables the timeout. + close_timeout: Timeout for closing the connection in seconds. + :obj:`None` disables the timeout. + max_size: Maximum size of incoming messages in bytes. + :obj:`None` disables the limit. + logger: Logger for this client. + It defaults to ``logging.getLogger("websockets.client")``. + See the :doc:`logging guide <../../topics/logging>` for details. + create_connection: Factory for the :class:`ClientConnection` managing + the connection. Set it to a wrapper or a subclass to customize + connection handling. + + Raises: + InvalidURI: If ``uri`` isn't a valid WebSocket URI. + OSError: If the TCP connection fails. + InvalidHandshake: If the opening handshake fails. + TimeoutError: If the opening handshake times out. + + """ + + # Process parameters + + wsuri = parse_uri(uri) + if not wsuri.secure and ssl_context is not None: + raise TypeError("ssl_context argument is incompatible with a ws:// URI") + + if unix: + if path is None and sock is None: + raise TypeError("missing path argument") + elif path is not None and sock is not None: + raise TypeError("path and sock arguments are incompatible") + else: + assert path is None # private argument, only set by unix_connect() + + if subprotocols is not None: + validate_subprotocols(subprotocols) + + if compression == "deflate": + extensions = enable_client_permessage_deflate(extensions) + elif compression is not None: + raise ValueError(f"unsupported compression: {compression}") + + # Calculate timeouts on the TCP, TLS, and WebSocket handshakes. + # The TCP and TLS timeouts must be set on the socket, then removed + # to avoid conflicting with the WebSocket timeout in handshake(). + deadline = Deadline(open_timeout) + + if create_connection is None: + create_connection = ClientConnection + + try: + # Connect socket + + if sock is None: + if unix: + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + sock.settimeout(deadline.timeout()) + assert path is not None # validated above -- this is for mpypy + sock.connect(path) + else: + sock = socket.create_connection( + (wsuri.host, wsuri.port), + deadline.timeout(), + ) + sock.settimeout(None) + + # Disable Nagle algorithm + + if not unix: + sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, True) + + # Initialize TLS wrapper and perform TLS handshake + + if wsuri.secure: + if ssl_context is None: + ssl_context = ssl.create_default_context() + if server_hostname is None: + server_hostname = wsuri.host + sock.settimeout(deadline.timeout()) + sock = ssl_context.wrap_socket(sock, server_hostname=server_hostname) + sock.settimeout(None) + + # Initialize WebSocket connection + + protocol = ClientProtocol( + wsuri, + origin=origin, + extensions=extensions, + subprotocols=subprotocols, + state=CONNECTING, + max_size=max_size, + logger=logger, + ) + + # Initialize WebSocket protocol + + connection = create_connection( + sock, + protocol, + close_timeout=close_timeout, + ) + # On failure, handshake() closes the socket and raises an exception. + connection.handshake( + additional_headers, + user_agent_header, + deadline.timeout(), + ) + + except Exception: + if sock is not None: + sock.close() + raise + + return connection + + +def unix_connect( + path: Optional[str] = None, + uri: Optional[str] = None, + **kwargs: Any, +) -> ClientConnection: + """ + Connect to a WebSocket server listening on a Unix socket. + + This function is identical to :func:`connect`, except for the additional + ``path`` argument. It's only available on Unix. + + It's mainly useful for debugging servers listening on Unix sockets. + + Args: + path: File system path to the Unix socket. + uri: URI of the WebSocket server. ``uri`` defaults to + ``ws://localhost/`` or, when a ``ssl_context`` is provided, to + ``wss://localhost/``. + + """ + if uri is None: + if kwargs.get("ssl_context") is None: + uri = "ws://localhost/" + else: + uri = "wss://localhost/" + return connect(uri=uri, unix=True, path=path, **kwargs) diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/sync/connection.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/sync/connection.py new file mode 100644 index 0000000000..4a8879e370 --- /dev/null +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/sync/connection.py @@ -0,0 +1,773 @@ +from __future__ import annotations + +import contextlib +import logging +import random +import socket +import struct +import threading +import uuid +from types import TracebackType +from typing import Any, Dict, Iterable, Iterator, Mapping, Optional, Type, Union + +from ..exceptions import ConnectionClosed, ConnectionClosedOK, ProtocolError +from ..frames import DATA_OPCODES, BytesLike, CloseCode, Frame, Opcode, prepare_ctrl +from ..http11 import Request, Response +from ..protocol import CLOSED, OPEN, Event, Protocol, State +from ..typing import Data, LoggerLike, Subprotocol +from .messages import Assembler +from .utils import Deadline + + +__all__ = ["Connection"] + +logger = logging.getLogger(__name__) + + +class Connection: + """ + Threaded implementation of a WebSocket connection. + + :class:`Connection` provides APIs shared between WebSocket servers and + clients. + + You shouldn't use it directly. Instead, use + :class:`~websockets.sync.client.ClientConnection` or + :class:`~websockets.sync.server.ServerConnection`. + + """ + + recv_bufsize = 65536 + + def __init__( + self, + socket: socket.socket, + protocol: Protocol, + *, + close_timeout: Optional[float] = 10, + ) -> None: + self.socket = socket + self.protocol = protocol + self.close_timeout = close_timeout + + # Inject reference to this instance in the protocol's logger. + self.protocol.logger = logging.LoggerAdapter( + self.protocol.logger, + {"websocket": self}, + ) + + # Copy attributes from the protocol for convenience. + self.id: uuid.UUID = self.protocol.id + """Unique identifier of the connection. Useful in logs.""" + self.logger: LoggerLike = self.protocol.logger + """Logger for this connection.""" + self.debug = self.protocol.debug + + # HTTP handshake request and response. + self.request: Optional[Request] = None + """Opening handshake request.""" + self.response: Optional[Response] = None + """Opening handshake response.""" + + # Mutex serializing interactions with the protocol. + self.protocol_mutex = threading.Lock() + + # Assembler turning frames into messages and serializing reads. + self.recv_messages = Assembler() + + # Whether we are busy sending a fragmented message. + self.send_in_progress = False + + # Deadline for the closing handshake. + self.close_deadline: Optional[Deadline] = None + + # Mapping of ping IDs to pong waiters, in chronological order. + self.pings: Dict[bytes, threading.Event] = {} + + # Receiving events from the socket. + self.recv_events_thread = threading.Thread(target=self.recv_events) + self.recv_events_thread.start() + + # Exception raised in recv_events, to be chained to ConnectionClosed + # in the user thread in order to show why the TCP connection dropped. + self.recv_events_exc: Optional[BaseException] = None + + # Public attributes + + @property + def local_address(self) -> Any: + """ + Local address of the connection. + + For IPv4 connections, this is a ``(host, port)`` tuple. + + The format of the address depends on the address family. + See :meth:`~socket.socket.getsockname`. + + """ + return self.socket.getsockname() + + @property + def remote_address(self) -> Any: + """ + Remote address of the connection. + + For IPv4 connections, this is a ``(host, port)`` tuple. + + The format of the address depends on the address family. + See :meth:`~socket.socket.getpeername`. + + """ + return self.socket.getpeername() + + @property + def subprotocol(self) -> Optional[Subprotocol]: + """ + Subprotocol negotiated during the opening handshake. + + :obj:`None` if no subprotocol was negotiated. + + """ + return self.protocol.subprotocol + + # Public methods + + def __enter__(self) -> Connection: + return self + + def __exit__( + self, + exc_type: Optional[Type[BaseException]], + exc_value: Optional[BaseException], + traceback: Optional[TracebackType], + ) -> None: + if exc_type is None: + self.close() + else: + self.close(CloseCode.INTERNAL_ERROR) + + def __iter__(self) -> Iterator[Data]: + """ + Iterate on incoming messages. + + The iterator calls :meth:`recv` and yields messages in an infinite loop. + + It exits when the connection is closed normally. It raises a + :exc:`~websockets.exceptions.ConnectionClosedError` exception after a + protocol error or a network failure. + + """ + try: + while True: + yield self.recv() + except ConnectionClosedOK: + return + + def recv(self, timeout: Optional[float] = None) -> Data: + """ + Receive the next message. + + When the connection is closed, :meth:`recv` raises + :exc:`~websockets.exceptions.ConnectionClosed`. Specifically, it raises + :exc:`~websockets.exceptions.ConnectionClosedOK` after a normal closure + and :exc:`~websockets.exceptions.ConnectionClosedError` after a protocol + error or a network failure. This is how you detect the end of the + message stream. + + If ``timeout`` is :obj:`None`, block until a message is received. If + ``timeout`` is set and no message is received within ``timeout`` + seconds, raise :exc:`TimeoutError`. Set ``timeout`` to ``0`` to check if + a message was already received. + + If the message is fragmented, wait until all fragments are received, + reassemble them, and return the whole message. + + Returns: + A string (:class:`str`) for a Text_ frame or a bytestring + (:class:`bytes`) for a Binary_ frame. + + .. _Text: https://www.rfc-editor.org/rfc/rfc6455.html#section-5.6 + .. _Binary: https://www.rfc-editor.org/rfc/rfc6455.html#section-5.6 + + Raises: + ConnectionClosed: When the connection is closed. + RuntimeError: If two threads call :meth:`recv` or + :meth:`recv_streaming` concurrently. + + """ + try: + return self.recv_messages.get(timeout) + except EOFError: + raise self.protocol.close_exc from self.recv_events_exc + except RuntimeError: + raise RuntimeError( + "cannot call recv while another thread " + "is already running recv or recv_streaming" + ) from None + + def recv_streaming(self) -> Iterator[Data]: + """ + Receive the next message frame by frame. + + If the message is fragmented, yield each fragment as it is received. + The iterator must be fully consumed, or else the connection will become + unusable. + + :meth:`recv_streaming` raises the same exceptions as :meth:`recv`. + + Returns: + An iterator of strings (:class:`str`) for a Text_ frame or + bytestrings (:class:`bytes`) for a Binary_ frame. + + .. _Text: https://www.rfc-editor.org/rfc/rfc6455.html#section-5.6 + .. _Binary: https://www.rfc-editor.org/rfc/rfc6455.html#section-5.6 + + Raises: + ConnectionClosed: When the connection is closed. + RuntimeError: If two threads call :meth:`recv` or + :meth:`recv_streaming` concurrently. + + """ + try: + yield from self.recv_messages.get_iter() + except EOFError: + raise self.protocol.close_exc from self.recv_events_exc + except RuntimeError: + raise RuntimeError( + "cannot call recv_streaming while another thread " + "is already running recv or recv_streaming" + ) from None + + def send(self, message: Union[Data, Iterable[Data]]) -> None: + """ + Send a message. + + A string (:class:`str`) is sent as a Text_ frame. A bytestring or + bytes-like object (:class:`bytes`, :class:`bytearray`, or + :class:`memoryview`) is sent as a Binary_ frame. + + .. _Text: https://www.rfc-editor.org/rfc/rfc6455.html#section-5.6 + .. _Binary: https://www.rfc-editor.org/rfc/rfc6455.html#section-5.6 + + :meth:`send` also accepts an iterable of strings, bytestrings, or + bytes-like objects to enable fragmentation_. Each item is treated as a + message fragment and sent in its own frame. All items must be of the + same type, or else :meth:`send` will raise a :exc:`TypeError` and the + connection will be closed. + + .. _fragmentation: https://www.rfc-editor.org/rfc/rfc6455.html#section-5.4 + + :meth:`send` rejects dict-like objects because this is often an error. + (If you really want to send the keys of a dict-like object as fragments, + call its :meth:`~dict.keys` method and pass the result to :meth:`send`.) + + When the connection is closed, :meth:`send` raises + :exc:`~websockets.exceptions.ConnectionClosed`. Specifically, it + raises :exc:`~websockets.exceptions.ConnectionClosedOK` after a normal + connection closure and + :exc:`~websockets.exceptions.ConnectionClosedError` after a protocol + error or a network failure. + + Args: + message: Message to send. + + Raises: + ConnectionClosed: When the connection is closed. + RuntimeError: If a connection is busy sending a fragmented message. + TypeError: If ``message`` doesn't have a supported type. + + """ + # Unfragmented message -- this case must be handled first because + # strings and bytes-like objects are iterable. + + if isinstance(message, str): + with self.send_context(): + if self.send_in_progress: + raise RuntimeError( + "cannot call send while another thread " + "is already running send" + ) + self.protocol.send_text(message.encode("utf-8")) + + elif isinstance(message, BytesLike): + with self.send_context(): + if self.send_in_progress: + raise RuntimeError( + "cannot call send while another thread " + "is already running send" + ) + self.protocol.send_binary(message) + + # Catch a common mistake -- passing a dict to send(). + + elif isinstance(message, Mapping): + raise TypeError("data is a dict-like object") + + # Fragmented message -- regular iterator. + + elif isinstance(message, Iterable): + chunks = iter(message) + try: + chunk = next(chunks) + except StopIteration: + return + + try: + # First fragment. + if isinstance(chunk, str): + text = True + with self.send_context(): + if self.send_in_progress: + raise RuntimeError( + "cannot call send while another thread " + "is already running send" + ) + self.send_in_progress = True + self.protocol.send_text( + chunk.encode("utf-8"), + fin=False, + ) + elif isinstance(chunk, BytesLike): + text = False + with self.send_context(): + if self.send_in_progress: + raise RuntimeError( + "cannot call send while another thread " + "is already running send" + ) + self.send_in_progress = True + self.protocol.send_binary( + chunk, + fin=False, + ) + else: + raise TypeError("data iterable must contain bytes or str") + + # Other fragments + for chunk in chunks: + if isinstance(chunk, str) and text: + with self.send_context(): + assert self.send_in_progress + self.protocol.send_continuation( + chunk.encode("utf-8"), + fin=False, + ) + elif isinstance(chunk, BytesLike) and not text: + with self.send_context(): + assert self.send_in_progress + self.protocol.send_continuation( + chunk, + fin=False, + ) + else: + raise TypeError("data iterable must contain uniform types") + + # Final fragment. + with self.send_context(): + self.protocol.send_continuation(b"", fin=True) + self.send_in_progress = False + + except RuntimeError: + # We didn't start sending a fragmented message. + raise + + except Exception: + # We're half-way through a fragmented message and we can't + # complete it. This makes the connection unusable. + with self.send_context(): + self.protocol.fail( + CloseCode.INTERNAL_ERROR, + "error in fragmented message", + ) + raise + + else: + raise TypeError("data must be bytes, str, or iterable") + + def close(self, code: int = CloseCode.NORMAL_CLOSURE, reason: str = "") -> None: + """ + Perform the closing handshake. + + :meth:`close` waits for the other end to complete the handshake, for the + TCP connection to terminate, and for all incoming messages to be read + with :meth:`recv`. + + :meth:`close` is idempotent: it doesn't do anything once the + connection is closed. + + Args: + code: WebSocket close code. + reason: WebSocket close reason. + + """ + try: + # The context manager takes care of waiting for the TCP connection + # to terminate after calling a method that sends a close frame. + with self.send_context(): + if self.send_in_progress: + self.protocol.fail( + CloseCode.INTERNAL_ERROR, + "close during fragmented message", + ) + else: + self.protocol.send_close(code, reason) + except ConnectionClosed: + # Ignore ConnectionClosed exceptions raised from send_context(). + # They mean that the connection is closed, which was the goal. + pass + + def ping(self, data: Optional[Data] = None) -> threading.Event: + """ + Send a Ping_. + + .. _Ping: https://www.rfc-editor.org/rfc/rfc6455.html#section-5.5.2 + + A ping may serve as a keepalive or as a check that the remote endpoint + received all messages up to this point + + Args: + data: Payload of the ping. A :class:`str` will be encoded to UTF-8. + If ``data`` is :obj:`None`, the payload is four random bytes. + + Returns: + An event that will be set when the corresponding pong is received. + You can ignore it if you don't intend to wait. + + :: + + pong_event = ws.ping() + pong_event.wait() # only if you want to wait for the pong + + Raises: + ConnectionClosed: When the connection is closed. + RuntimeError: If another ping was sent with the same data and + the corresponding pong wasn't received yet. + + """ + if data is not None: + data = prepare_ctrl(data) + + with self.send_context(): + # Protect against duplicates if a payload is explicitly set. + if data in self.pings: + raise RuntimeError("already waiting for a pong with the same data") + + # Generate a unique random payload otherwise. + while data is None or data in self.pings: + data = struct.pack("!I", random.getrandbits(32)) + + pong_waiter = threading.Event() + self.pings[data] = pong_waiter + self.protocol.send_ping(data) + return pong_waiter + + def pong(self, data: Data = b"") -> None: + """ + Send a Pong_. + + .. _Pong: https://www.rfc-editor.org/rfc/rfc6455.html#section-5.5.3 + + An unsolicited pong may serve as a unidirectional heartbeat. + + Args: + data: Payload of the pong. A :class:`str` will be encoded to UTF-8. + + Raises: + ConnectionClosed: When the connection is closed. + + """ + data = prepare_ctrl(data) + + with self.send_context(): + self.protocol.send_pong(data) + + # Private methods + + def process_event(self, event: Event) -> None: + """ + Process one incoming event. + + This method is overridden in subclasses to handle the handshake. + + """ + assert isinstance(event, Frame) + if event.opcode in DATA_OPCODES: + self.recv_messages.put(event) + + if event.opcode is Opcode.PONG: + self.acknowledge_pings(bytes(event.data)) + + def acknowledge_pings(self, data: bytes) -> None: + """ + Acknowledge pings when receiving a pong. + + """ + with self.protocol_mutex: + # Ignore unsolicited pong. + if data not in self.pings: + return + # Sending a pong for only the most recent ping is legal. + # Acknowledge all previous pings too in that case. + ping_id = None + ping_ids = [] + for ping_id, ping in self.pings.items(): + ping_ids.append(ping_id) + ping.set() + if ping_id == data: + break + else: + raise AssertionError("solicited pong not found in pings") + # Remove acknowledged pings from self.pings. + for ping_id in ping_ids: + del self.pings[ping_id] + + def recv_events(self) -> None: + """ + Read incoming data from the socket and process events. + + Run this method in a thread as long as the connection is alive. + + ``recv_events()`` exits immediately when the ``self.socket`` is closed. + + """ + try: + while True: + try: + if self.close_deadline is not None: + self.socket.settimeout(self.close_deadline.timeout()) + data = self.socket.recv(self.recv_bufsize) + except Exception as exc: + if self.debug: + self.logger.debug("error while receiving data", exc_info=True) + # When the closing handshake is initiated by our side, + # recv() may block until send_context() closes the socket. + # In that case, send_context() already set recv_events_exc. + # Calling set_recv_events_exc() avoids overwriting it. + with self.protocol_mutex: + self.set_recv_events_exc(exc) + break + + if data == b"": + break + + # Acquire the connection lock. + with self.protocol_mutex: + # Feed incoming data to the connection. + self.protocol.receive_data(data) + + # This isn't expected to raise an exception. + events = self.protocol.events_received() + + # Write outgoing data to the socket. + try: + self.send_data() + except Exception as exc: + if self.debug: + self.logger.debug("error while sending data", exc_info=True) + # Similarly to the above, avoid overriding an exception + # set by send_context(), in case of a race condition + # i.e. send_context() closes the socket after recv() + # returns above but before send_data() calls send(). + self.set_recv_events_exc(exc) + break + + if self.protocol.close_expected(): + # If the connection is expected to close soon, set the + # close deadline based on the close timeout. + if self.close_deadline is None: + self.close_deadline = Deadline(self.close_timeout) + + # Unlock conn_mutex before processing events. Else, the + # application can't send messages in response to events. + + # If self.send_data raised an exception, then events are lost. + # Given that automatic responses write small amounts of data, + # this should be uncommon, so we don't handle the edge case. + + try: + for event in events: + # This may raise EOFError if the closing handshake + # times out while a message is waiting to be read. + self.process_event(event) + except EOFError: + break + + # Breaking out of the while True: ... loop means that we believe + # that the socket doesn't work anymore. + with self.protocol_mutex: + # Feed the end of the data stream to the connection. + self.protocol.receive_eof() + + # This isn't expected to generate events. + assert not self.protocol.events_received() + + # There is no error handling because send_data() can only write + # the end of the data stream here and it handles errors itself. + self.send_data() + + except Exception as exc: + # This branch should never run. It's a safety net in case of bugs. + self.logger.error("unexpected internal error", exc_info=True) + with self.protocol_mutex: + self.set_recv_events_exc(exc) + # We don't know where we crashed. Force protocol state to CLOSED. + self.protocol.state = CLOSED + finally: + # This isn't expected to raise an exception. + self.close_socket() + + @contextlib.contextmanager + def send_context( + self, + *, + expected_state: State = OPEN, # CONNECTING during the opening handshake + ) -> Iterator[None]: + """ + Create a context for writing to the connection from user code. + + On entry, :meth:`send_context` acquires the connection lock and checks + that the connection is open; on exit, it writes outgoing data to the + socket:: + + with self.send_context(): + self.protocol.send_text(message.encode("utf-8")) + + When the connection isn't open on entry, when the connection is expected + to close on exit, or when an unexpected error happens, terminating the + connection, :meth:`send_context` waits until the connection is closed + then raises :exc:`~websockets.exceptions.ConnectionClosed`. + + """ + # Should we wait until the connection is closed? + wait_for_close = False + # Should we close the socket and raise ConnectionClosed? + raise_close_exc = False + # What exception should we chain ConnectionClosed to? + original_exc: Optional[BaseException] = None + + # Acquire the protocol lock. + with self.protocol_mutex: + if self.protocol.state is expected_state: + # Let the caller interact with the protocol. + try: + yield + except (ProtocolError, RuntimeError): + # The protocol state wasn't changed. Exit immediately. + raise + except Exception as exc: + self.logger.error("unexpected internal error", exc_info=True) + # This branch should never run. It's a safety net in case of + # bugs. Since we don't know what happened, we will close the + # connection and raise the exception to the caller. + wait_for_close = False + raise_close_exc = True + original_exc = exc + else: + # Check if the connection is expected to close soon. + if self.protocol.close_expected(): + wait_for_close = True + # If the connection is expected to close soon, set the + # close deadline based on the close timeout. + + # Since we tested earlier that protocol.state was OPEN + # (or CONNECTING) and we didn't release protocol_mutex, + # it is certain that self.close_deadline is still None. + assert self.close_deadline is None + self.close_deadline = Deadline(self.close_timeout) + # Write outgoing data to the socket. + try: + self.send_data() + except Exception as exc: + if self.debug: + self.logger.debug("error while sending data", exc_info=True) + # While the only expected exception here is OSError, + # other exceptions would be treated identically. + wait_for_close = False + raise_close_exc = True + original_exc = exc + + else: # self.protocol.state is not expected_state + # Minor layering violation: we assume that the connection + # will be closing soon if it isn't in the expected state. + wait_for_close = True + raise_close_exc = True + + # To avoid a deadlock, release the connection lock by exiting the + # context manager before waiting for recv_events() to terminate. + + # If the connection is expected to close soon and the close timeout + # elapses, close the socket to terminate the connection. + if wait_for_close: + if self.close_deadline is None: + timeout = self.close_timeout + else: + # Thread.join() returns immediately if timeout is negative. + timeout = self.close_deadline.timeout(raise_if_elapsed=False) + self.recv_events_thread.join(timeout) + + if self.recv_events_thread.is_alive(): + # There's no risk to overwrite another error because + # original_exc is never set when wait_for_close is True. + assert original_exc is None + original_exc = TimeoutError("timed out while closing connection") + # Set recv_events_exc before closing the socket in order to get + # proper exception reporting. + raise_close_exc = True + with self.protocol_mutex: + self.set_recv_events_exc(original_exc) + + # If an error occurred, close the socket to terminate the connection and + # raise an exception. + if raise_close_exc: + self.close_socket() + self.recv_events_thread.join() + raise self.protocol.close_exc from original_exc + + def send_data(self) -> None: + """ + Send outgoing data. + + This method requires holding protocol_mutex. + + Raises: + OSError: When a socket operations fails. + + """ + assert self.protocol_mutex.locked() + for data in self.protocol.data_to_send(): + if data: + if self.close_deadline is not None: + self.socket.settimeout(self.close_deadline.timeout()) + self.socket.sendall(data) + else: + try: + self.socket.shutdown(socket.SHUT_WR) + except OSError: # socket already closed + pass + + def set_recv_events_exc(self, exc: Optional[BaseException]) -> None: + """ + Set recv_events_exc, if not set yet. + + This method requires holding protocol_mutex. + + """ + assert self.protocol_mutex.locked() + if self.recv_events_exc is None: + self.recv_events_exc = exc + + def close_socket(self) -> None: + """ + Shutdown and close socket. Close message assembler. + + Calling close_socket() guarantees that recv_events() terminates. Indeed, + recv_events() may block only on socket.recv() or on recv_messages.put(). + + """ + # shutdown() is required to interrupt recv() on Linux. + try: + self.socket.shutdown(socket.SHUT_RDWR) + except OSError: + pass # socket is already closed + self.socket.close() + self.recv_messages.close() diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/sync/messages.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/sync/messages.py new file mode 100644 index 0000000000..67a22313ca --- /dev/null +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/sync/messages.py @@ -0,0 +1,281 @@ +from __future__ import annotations + +import codecs +import queue +import threading +from typing import Iterator, List, Optional, cast + +from ..frames import Frame, Opcode +from ..typing import Data + + +__all__ = ["Assembler"] + +UTF8Decoder = codecs.getincrementaldecoder("utf-8") + + +class Assembler: + """ + Assemble messages from frames. + + """ + + def __init__(self) -> None: + # Serialize reads and writes -- except for reads via synchronization + # primitives provided by the threading and queue modules. + self.mutex = threading.Lock() + + # We create a latch with two events to ensure proper interleaving of + # writing and reading messages. + # put() sets this event to tell get() that a message can be fetched. + self.message_complete = threading.Event() + # get() sets this event to let put() that the message was fetched. + self.message_fetched = threading.Event() + + # This flag prevents concurrent calls to get() by user code. + self.get_in_progress = False + # This flag prevents concurrent calls to put() by library code. + self.put_in_progress = False + + # Decoder for text frames, None for binary frames. + self.decoder: Optional[codecs.IncrementalDecoder] = None + + # Buffer of frames belonging to the same message. + self.chunks: List[Data] = [] + + # When switching from "buffering" to "streaming", we use a thread-safe + # queue for transferring frames from the writing thread (library code) + # to the reading thread (user code). We're buffering when chunks_queue + # is None and streaming when it's a SimpleQueue. None is a sentinel + # value marking the end of the stream, superseding message_complete. + + # Stream data from frames belonging to the same message. + # Remove quotes around type when dropping Python < 3.9. + self.chunks_queue: Optional["queue.SimpleQueue[Optional[Data]]"] = None + + # This flag marks the end of the stream. + self.closed = False + + def get(self, timeout: Optional[float] = None) -> Data: + """ + Read the next message. + + :meth:`get` returns a single :class:`str` or :class:`bytes`. + + If the message is fragmented, :meth:`get` waits until the last frame is + received, then it reassembles the message and returns it. To receive + messages frame by frame, use :meth:`get_iter` instead. + + Args: + timeout: If a timeout is provided and elapses before a complete + message is received, :meth:`get` raises :exc:`TimeoutError`. + + Raises: + EOFError: If the stream of frames has ended. + RuntimeError: If two threads run :meth:`get` or :meth:``get_iter` + concurrently. + + """ + with self.mutex: + if self.closed: + raise EOFError("stream of frames ended") + + if self.get_in_progress: + raise RuntimeError("get or get_iter is already running") + + self.get_in_progress = True + + # If the message_complete event isn't set yet, release the lock to + # allow put() to run and eventually set it. + # Locking with get_in_progress ensures only one thread can get here. + completed = self.message_complete.wait(timeout) + + with self.mutex: + self.get_in_progress = False + + # Waiting for a complete message timed out. + if not completed: + raise TimeoutError(f"timed out in {timeout:.1f}s") + + # get() was unblocked by close() rather than put(). + if self.closed: + raise EOFError("stream of frames ended") + + assert self.message_complete.is_set() + self.message_complete.clear() + + joiner: Data = b"" if self.decoder is None else "" + # mypy cannot figure out that chunks have the proper type. + message: Data = joiner.join(self.chunks) # type: ignore + + assert not self.message_fetched.is_set() + self.message_fetched.set() + + self.chunks = [] + assert self.chunks_queue is None + + return message + + def get_iter(self) -> Iterator[Data]: + """ + Stream the next message. + + Iterating the return value of :meth:`get_iter` yields a :class:`str` or + :class:`bytes` for each frame in the message. + + The iterator must be fully consumed before calling :meth:`get_iter` or + :meth:`get` again. Else, :exc:`RuntimeError` is raised. + + This method only makes sense for fragmented messages. If messages aren't + fragmented, use :meth:`get` instead. + + Raises: + EOFError: If the stream of frames has ended. + RuntimeError: If two threads run :meth:`get` or :meth:``get_iter` + concurrently. + + """ + with self.mutex: + if self.closed: + raise EOFError("stream of frames ended") + + if self.get_in_progress: + raise RuntimeError("get or get_iter is already running") + + chunks = self.chunks + self.chunks = [] + self.chunks_queue = cast( + # Remove quotes around type when dropping Python < 3.9. + "queue.SimpleQueue[Optional[Data]]", + queue.SimpleQueue(), + ) + + # Sending None in chunk_queue supersedes setting message_complete + # when switching to "streaming". If message is already complete + # when the switch happens, put() didn't send None, so we have to. + if self.message_complete.is_set(): + self.chunks_queue.put(None) + + self.get_in_progress = True + + # Locking with get_in_progress ensures only one thread can get here. + yield from chunks + while True: + chunk = self.chunks_queue.get() + if chunk is None: + break + yield chunk + + with self.mutex: + self.get_in_progress = False + + assert self.message_complete.is_set() + self.message_complete.clear() + + # get_iter() was unblocked by close() rather than put(). + if self.closed: + raise EOFError("stream of frames ended") + + assert not self.message_fetched.is_set() + self.message_fetched.set() + + assert self.chunks == [] + self.chunks_queue = None + + def put(self, frame: Frame) -> None: + """ + Add ``frame`` to the next message. + + When ``frame`` is the final frame in a message, :meth:`put` waits until + the message is fetched, either by calling :meth:`get` or by fully + consuming the return value of :meth:`get_iter`. + + :meth:`put` assumes that the stream of frames respects the protocol. If + it doesn't, the behavior is undefined. + + Raises: + EOFError: If the stream of frames has ended. + RuntimeError: If two threads run :meth:`put` concurrently. + + """ + with self.mutex: + if self.closed: + raise EOFError("stream of frames ended") + + if self.put_in_progress: + raise RuntimeError("put is already running") + + if frame.opcode is Opcode.TEXT: + self.decoder = UTF8Decoder(errors="strict") + elif frame.opcode is Opcode.BINARY: + self.decoder = None + elif frame.opcode is Opcode.CONT: + pass + else: + # Ignore control frames. + return + + data: Data + if self.decoder is not None: + data = self.decoder.decode(frame.data, frame.fin) + else: + data = frame.data + + if self.chunks_queue is None: + self.chunks.append(data) + else: + self.chunks_queue.put(data) + + if not frame.fin: + return + + # Message is complete. Wait until it's fetched to return. + + assert not self.message_complete.is_set() + self.message_complete.set() + + if self.chunks_queue is not None: + self.chunks_queue.put(None) + + assert not self.message_fetched.is_set() + + self.put_in_progress = True + + # Release the lock to allow get() to run and eventually set the event. + self.message_fetched.wait() + + with self.mutex: + self.put_in_progress = False + + assert self.message_fetched.is_set() + self.message_fetched.clear() + + # put() was unblocked by close() rather than get() or get_iter(). + if self.closed: + raise EOFError("stream of frames ended") + + self.decoder = None + + def close(self) -> None: + """ + End the stream of frames. + + Callling :meth:`close` concurrently with :meth:`get`, :meth:`get_iter`, + or :meth:`put` is safe. They will raise :exc:`EOFError`. + + """ + with self.mutex: + if self.closed: + return + + self.closed = True + + # Unblock get or get_iter. + if self.get_in_progress: + self.message_complete.set() + if self.chunks_queue is not None: + self.chunks_queue.put(None) + + # Unblock put(). + if self.put_in_progress: + self.message_fetched.set() diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/sync/server.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/sync/server.py new file mode 100644 index 0000000000..14767968c9 --- /dev/null +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/sync/server.py @@ -0,0 +1,530 @@ +from __future__ import annotations + +import http +import logging +import os +import selectors +import socket +import ssl +import sys +import threading +from types import TracebackType +from typing import Any, Callable, Optional, Sequence, Type + +from websockets.frames import CloseCode + +from ..extensions.base import ServerExtensionFactory +from ..extensions.permessage_deflate import enable_server_permessage_deflate +from ..headers import validate_subprotocols +from ..http import USER_AGENT +from ..http11 import Request, Response +from ..protocol import CONNECTING, OPEN, Event +from ..server import ServerProtocol +from ..typing import LoggerLike, Origin, Subprotocol +from .connection import Connection +from .utils import Deadline + + +__all__ = ["serve", "unix_serve", "ServerConnection", "WebSocketServer"] + + +class ServerConnection(Connection): + """ + Threaded implementation of a WebSocket server connection. + + :class:`ServerConnection` provides :meth:`recv` and :meth:`send` methods for + receiving and sending messages. + + It supports iteration to receive messages:: + + for message in websocket: + process(message) + + The iterator exits normally when the connection is closed with close code + 1000 (OK) or 1001 (going away) or without a close code. It raises a + :exc:`~websockets.exceptions.ConnectionClosedError` when the connection is + closed with any other code. + + Args: + socket: Socket connected to a WebSocket client. + protocol: Sans-I/O connection. + close_timeout: Timeout for closing the connection in seconds. + + """ + + def __init__( + self, + socket: socket.socket, + protocol: ServerProtocol, + *, + close_timeout: Optional[float] = 10, + ) -> None: + self.protocol: ServerProtocol + self.request_rcvd = threading.Event() + super().__init__( + socket, + protocol, + close_timeout=close_timeout, + ) + + def handshake( + self, + process_request: Optional[ + Callable[ + [ServerConnection, Request], + Optional[Response], + ] + ] = None, + process_response: Optional[ + Callable[ + [ServerConnection, Request, Response], + Optional[Response], + ] + ] = None, + server_header: Optional[str] = USER_AGENT, + timeout: Optional[float] = None, + ) -> None: + """ + Perform the opening handshake. + + """ + if not self.request_rcvd.wait(timeout): + self.close_socket() + self.recv_events_thread.join() + raise TimeoutError("timed out during handshake") + + if self.request is None: + self.close_socket() + self.recv_events_thread.join() + raise ConnectionError("connection closed during handshake") + + with self.send_context(expected_state=CONNECTING): + self.response = None + + if process_request is not None: + try: + self.response = process_request(self, self.request) + except Exception as exc: + self.protocol.handshake_exc = exc + self.logger.error("opening handshake failed", exc_info=True) + self.response = self.protocol.reject( + http.HTTPStatus.INTERNAL_SERVER_ERROR, + ( + "Failed to open a WebSocket connection.\n" + "See server log for more information.\n" + ), + ) + + if self.response is None: + self.response = self.protocol.accept(self.request) + + if server_header is not None: + self.response.headers["Server"] = server_header + + if process_response is not None: + try: + response = process_response(self, self.request, self.response) + except Exception as exc: + self.protocol.handshake_exc = exc + self.logger.error("opening handshake failed", exc_info=True) + self.response = self.protocol.reject( + http.HTTPStatus.INTERNAL_SERVER_ERROR, + ( + "Failed to open a WebSocket connection.\n" + "See server log for more information.\n" + ), + ) + else: + if response is not None: + self.response = response + + self.protocol.send_response(self.response) + + if self.protocol.state is not OPEN: + self.recv_events_thread.join(self.close_timeout) + self.close_socket() + self.recv_events_thread.join() + + if self.protocol.handshake_exc is not None: + raise self.protocol.handshake_exc + + def process_event(self, event: Event) -> None: + """ + Process one incoming event. + + """ + # First event - handshake request. + if self.request is None: + assert isinstance(event, Request) + self.request = event + self.request_rcvd.set() + # Later events - frames. + else: + super().process_event(event) + + def recv_events(self) -> None: + """ + Read incoming data from the socket and process events. + + """ + try: + super().recv_events() + finally: + # If the connection is closed during the handshake, unblock it. + self.request_rcvd.set() + + +class WebSocketServer: + """ + WebSocket server returned by :func:`serve`. + + This class mirrors the API of :class:`~socketserver.BaseServer`, notably the + :meth:`~socketserver.BaseServer.serve_forever` and + :meth:`~socketserver.BaseServer.shutdown` methods, as well as the context + manager protocol. + + Args: + socket: Server socket listening for new connections. + handler: Handler for one connection. Receives the socket and address + returned by :meth:`~socket.socket.accept`. + logger: Logger for this server. + + """ + + def __init__( + self, + socket: socket.socket, + handler: Callable[[socket.socket, Any], None], + logger: Optional[LoggerLike] = None, + ): + self.socket = socket + self.handler = handler + if logger is None: + logger = logging.getLogger("websockets.server") + self.logger = logger + if sys.platform != "win32": + self.shutdown_watcher, self.shutdown_notifier = os.pipe() + + def serve_forever(self) -> None: + """ + See :meth:`socketserver.BaseServer.serve_forever`. + + This method doesn't return. Calling :meth:`shutdown` from another thread + stops the server. + + Typical use:: + + with serve(...) as server: + server.serve_forever() + + """ + poller = selectors.DefaultSelector() + poller.register(self.socket, selectors.EVENT_READ) + if sys.platform != "win32": + poller.register(self.shutdown_watcher, selectors.EVENT_READ) + + while True: + poller.select() + try: + # If the socket is closed, this will raise an exception and exit + # the loop. So we don't need to check the return value of select(). + sock, addr = self.socket.accept() + except OSError: + break + thread = threading.Thread(target=self.handler, args=(sock, addr)) + thread.start() + + def shutdown(self) -> None: + """ + See :meth:`socketserver.BaseServer.shutdown`. + + """ + self.socket.close() + if sys.platform != "win32": + os.write(self.shutdown_notifier, b"x") + + def fileno(self) -> int: + """ + See :meth:`socketserver.BaseServer.fileno`. + + """ + return self.socket.fileno() + + def __enter__(self) -> WebSocketServer: + return self + + def __exit__( + self, + exc_type: Optional[Type[BaseException]], + exc_value: Optional[BaseException], + traceback: Optional[TracebackType], + ) -> None: + self.shutdown() + + +def serve( + handler: Callable[[ServerConnection], None], + host: Optional[str] = None, + port: Optional[int] = None, + *, + # TCP/TLS — unix and path are only for unix_serve() + sock: Optional[socket.socket] = None, + ssl_context: Optional[ssl.SSLContext] = None, + unix: bool = False, + path: Optional[str] = None, + # WebSocket + origins: Optional[Sequence[Optional[Origin]]] = None, + extensions: Optional[Sequence[ServerExtensionFactory]] = None, + subprotocols: Optional[Sequence[Subprotocol]] = None, + select_subprotocol: Optional[ + Callable[ + [ServerConnection, Sequence[Subprotocol]], + Optional[Subprotocol], + ] + ] = None, + process_request: Optional[ + Callable[ + [ServerConnection, Request], + Optional[Response], + ] + ] = None, + process_response: Optional[ + Callable[ + [ServerConnection, Request, Response], + Optional[Response], + ] + ] = None, + server_header: Optional[str] = USER_AGENT, + compression: Optional[str] = "deflate", + # Timeouts + open_timeout: Optional[float] = 10, + close_timeout: Optional[float] = 10, + # Limits + max_size: Optional[int] = 2**20, + # Logging + logger: Optional[LoggerLike] = None, + # Escape hatch for advanced customization + create_connection: Optional[Type[ServerConnection]] = None, +) -> WebSocketServer: + """ + Create a WebSocket server listening on ``host`` and ``port``. + + Whenever a client connects, the server creates a :class:`ServerConnection`, + performs the opening handshake, and delegates to the ``handler``. + + The handler receives a :class:`ServerConnection` instance, which you can use + to send and receive messages. + + Once the handler completes, either normally or with an exception, the server + performs the closing handshake and closes the connection. + + :class:`WebSocketServer` mirrors the API of + :class:`~socketserver.BaseServer`. Treat it as a context manager to ensure + that it will be closed and call the :meth:`~WebSocketServer.serve_forever` + method to serve requests:: + + def handler(websocket): + ... + + with websockets.sync.server.serve(handler, ...) as server: + server.serve_forever() + + Args: + handler: Connection handler. It receives the WebSocket connection, + which is a :class:`ServerConnection`, in argument. + host: Network interfaces the server binds to. + See :func:`~socket.create_server` for details. + port: TCP port the server listens on. + See :func:`~socket.create_server` for details. + sock: Preexisting TCP socket. ``sock`` replaces ``host`` and ``port``. + You may call :func:`socket.create_server` to create a suitable TCP + socket. + ssl_context: Configuration for enabling TLS on the connection. + origins: Acceptable values of the ``Origin`` header, for defending + against Cross-Site WebSocket Hijacking attacks. Include :obj:`None` + in the list if the lack of an origin is acceptable. + extensions: List of supported extensions, in order in which they + should be negotiated and run. + subprotocols: List of supported subprotocols, in order of decreasing + preference. + select_subprotocol: Callback for selecting a subprotocol among + those supported by the client and the server. It receives a + :class:`ServerConnection` (not a + :class:`~websockets.server.ServerProtocol`!) instance and a list of + subprotocols offered by the client. Other than the first argument, + it has the same behavior as the + :meth:`ServerProtocol.select_subprotocol + <websockets.server.ServerProtocol.select_subprotocol>` method. + process_request: Intercept the request during the opening handshake. + Return an HTTP response to force the response or :obj:`None` to + continue normally. When you force an HTTP 101 Continue response, + the handshake is successful. Else, the connection is aborted. + process_response: Intercept the response during the opening handshake. + Return an HTTP response to force the response or :obj:`None` to + continue normally. When you force an HTTP 101 Continue response, + the handshake is successful. Else, the connection is aborted. + server_header: Value of the ``Server`` response header. + It defaults to ``"Python/x.y.z websockets/X.Y"``. Setting it to + :obj:`None` removes the header. + compression: The "permessage-deflate" extension is enabled by default. + Set ``compression`` to :obj:`None` to disable it. See the + :doc:`compression guide <../../topics/compression>` for details. + open_timeout: Timeout for opening connections in seconds. + :obj:`None` disables the timeout. + close_timeout: Timeout for closing connections in seconds. + :obj:`None` disables the timeout. + max_size: Maximum size of incoming messages in bytes. + :obj:`None` disables the limit. + logger: Logger for this server. + It defaults to ``logging.getLogger("websockets.server")``. See the + :doc:`logging guide <../../topics/logging>` for details. + create_connection: Factory for the :class:`ServerConnection` managing + the connection. Set it to a wrapper or a subclass to customize + connection handling. + """ + + # Process parameters + + if subprotocols is not None: + validate_subprotocols(subprotocols) + + if compression == "deflate": + extensions = enable_server_permessage_deflate(extensions) + elif compression is not None: + raise ValueError(f"unsupported compression: {compression}") + + if create_connection is None: + create_connection = ServerConnection + + # Bind socket and listen + + if sock is None: + if unix: + if path is None: + raise TypeError("missing path argument") + sock = socket.create_server(path, family=socket.AF_UNIX) + else: + sock = socket.create_server((host, port)) + else: + if path is not None: + raise TypeError("path and sock arguments are incompatible") + + # Initialize TLS wrapper + + if ssl_context is not None: + sock = ssl_context.wrap_socket( + sock, + server_side=True, + # Delay TLS handshake until after we set a timeout on the socket. + do_handshake_on_connect=False, + ) + + # Define request handler + + def conn_handler(sock: socket.socket, addr: Any) -> None: + # Calculate timeouts on the TLS and WebSocket handshakes. + # The TLS timeout must be set on the socket, then removed + # to avoid conflicting with the WebSocket timeout in handshake(). + deadline = Deadline(open_timeout) + + try: + # Disable Nagle algorithm + + if not unix: + sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, True) + + # Perform TLS handshake + + if ssl_context is not None: + sock.settimeout(deadline.timeout()) + assert isinstance(sock, ssl.SSLSocket) # mypy cannot figure this out + sock.do_handshake() + sock.settimeout(None) + + # Create a closure so that select_subprotocol has access to self. + + protocol_select_subprotocol: Optional[ + Callable[ + [ServerProtocol, Sequence[Subprotocol]], + Optional[Subprotocol], + ] + ] = None + + if select_subprotocol is not None: + + def protocol_select_subprotocol( + protocol: ServerProtocol, + subprotocols: Sequence[Subprotocol], + ) -> Optional[Subprotocol]: + # mypy doesn't know that select_subprotocol is immutable. + assert select_subprotocol is not None + # Ensure this function is only used in the intended context. + assert protocol is connection.protocol + return select_subprotocol(connection, subprotocols) + + # Initialize WebSocket connection + + protocol = ServerProtocol( + origins=origins, + extensions=extensions, + subprotocols=subprotocols, + select_subprotocol=protocol_select_subprotocol, + state=CONNECTING, + max_size=max_size, + logger=logger, + ) + + # Initialize WebSocket protocol + + assert create_connection is not None # help mypy + connection = create_connection( + sock, + protocol, + close_timeout=close_timeout, + ) + # On failure, handshake() closes the socket, raises an exception, and + # logs it. + connection.handshake( + process_request, + process_response, + server_header, + deadline.timeout(), + ) + + except Exception: + sock.close() + return + + try: + handler(connection) + except Exception: + protocol.logger.error("connection handler failed", exc_info=True) + connection.close(CloseCode.INTERNAL_ERROR) + else: + connection.close() + + # Initialize server + + return WebSocketServer(sock, conn_handler, logger) + + +def unix_serve( + handler: Callable[[ServerConnection], Any], + path: Optional[str] = None, + **kwargs: Any, +) -> WebSocketServer: + """ + Create a WebSocket server listening on a Unix socket. + + This function is identical to :func:`serve`, except the ``host`` and + ``port`` arguments are replaced by ``path``. It's only available on Unix. + + It's useful for deploying a server behind a reverse proxy such as nginx. + + Args: + handler: Connection handler. It receives the WebSocket connection, + which is a :class:`ServerConnection`, in argument. + path: File system path to the Unix socket. + + """ + return serve(handler, path=path, unix=True, **kwargs) diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/sync/utils.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/sync/utils.py new file mode 100644 index 0000000000..471f32e19d --- /dev/null +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/sync/utils.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +import time +from typing import Optional + + +__all__ = ["Deadline"] + + +class Deadline: + """ + Manage timeouts across multiple steps. + + Args: + timeout: Time available in seconds or :obj:`None` if there is no limit. + + """ + + def __init__(self, timeout: Optional[float]) -> None: + self.deadline: Optional[float] + if timeout is None: + self.deadline = None + else: + self.deadline = time.monotonic() + timeout + + def timeout(self, *, raise_if_elapsed: bool = True) -> Optional[float]: + """ + Calculate a timeout from a deadline. + + Args: + raise_if_elapsed (bool): Whether to raise :exc:`TimeoutError` + if the deadline lapsed. + + Raises: + TimeoutError: If the deadline lapsed. + + Returns: + Time left in seconds or :obj:`None` if there is no limit. + + """ + if self.deadline is None: + return None + timeout = self.deadline - time.monotonic() + if raise_if_elapsed and timeout <= 0: + raise TimeoutError("timed out") + return timeout diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/typing.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/typing.py index e672ba0069..cc3e3ec0d9 100644 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/typing.py +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/typing.py @@ -1,5 +1,6 @@ from __future__ import annotations +import http import logging from typing import List, NewType, Optional, Tuple, Union @@ -7,6 +8,7 @@ from typing import List, NewType, Optional, Tuple, Union __all__ = [ "Data", "LoggerLike", + "StatusLike", "Origin", "Subprotocol", "ExtensionName", @@ -30,6 +32,11 @@ LoggerLike = Union[logging.Logger, logging.LoggerAdapter] """Types accepted where a :class:`~logging.Logger` is expected.""" +StatusLike = Union[http.HTTPStatus, int] +""" +Types accepted where an :class:`~http.HTTPStatus` is expected.""" + + Origin = NewType("Origin", str) """Value of a ``Origin`` header.""" diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/uri.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/uri.py index fff0c38064..385090f66a 100644 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/uri.py +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/uri.py @@ -33,8 +33,8 @@ class WebSocketURI: port: int path: str query: str - username: Optional[str] - password: Optional[str] + username: Optional[str] = None + password: Optional[str] = None @property def resource_name(self) -> str: diff --git a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/version.py b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/version.py index c30bfd68f3..d1c99458e2 100644 --- a/testing/web-platform/tests/tools/third_party/websockets/src/websockets/version.py +++ b/testing/web-platform/tests/tools/third_party/websockets/src/websockets/version.py @@ -1,5 +1,7 @@ from __future__ import annotations +import importlib.metadata + __all__ = ["tag", "version", "commit"] @@ -18,7 +20,7 @@ __all__ = ["tag", "version", "commit"] released = True -tag = version = commit = "10.3" +tag = version = commit = "12.0" if not released: # pragma: no cover @@ -44,7 +46,11 @@ if not released: # pragma: no cover text=True, ).stdout.strip() # subprocess.run raises FileNotFoundError if git isn't on $PATH. - except (FileNotFoundError, subprocess.CalledProcessError): + except ( + FileNotFoundError, + subprocess.CalledProcessError, + subprocess.TimeoutExpired, + ): pass else: description_re = r"[0-9.]+-([0-9]+)-(g[0-9a-f]{7,}(?:-dirty)?)" @@ -56,8 +62,6 @@ if not released: # pragma: no cover # Read version from package metadata if it is installed. try: - import importlib.metadata # move up when dropping Python 3.7 - return importlib.metadata.version("websockets") except ImportError: pass |