diff --git a/.audit/oberstet_fix_1986.md b/.audit/oberstet_fix_1986.md new file mode 100644 index 000000000..6906bfb79 --- /dev/null +++ b/.audit/oberstet_fix_1986.md @@ -0,0 +1,8 @@ +- [ ] I did **not** use any AI-assistance tools to help create this pull request. +- [x] I **did** use AI-assistance tools to *help* create this pull request. +- [x] I have read, understood and followed the project's AI_POLICY.md when creating code, documentation etc. for this pull request. + +Submitted by: @oberstet +Date: 2026-10-17 +Related issue(s): #1986 +Branch: oberstet:fix_1986 diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index 625342102..3b57ebe33 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -128,6 +128,78 @@ jobs: - name: Unit test - asyncio / pytest - WITHOUT NVX run: just test-asyncio cpy314 0 + # The unit test suites on every supported Python run-time, with all four + # WebSocket compression extensions installed (#1986). quality-checks above + # additionally runs them on cpy314 with and without NVX, plus coverage. + unit-tests: + name: Unit Tests + needs: identifiers + runs-on: ubuntu-24.04 + + env: + BASE_REPO: ${{ needs.identifiers.outputs.base_repo }} + BASE_BRANCH: ${{ needs.identifiers.outputs.base_branch }} + PR_NUMBER: ${{ needs.identifiers.outputs.pr_number }} + PR_REPO: ${{ needs.identifiers.outputs.pr_repo }} + PR_BRANCH: ${{ needs.identifiers.outputs.pr_branch }} + + strategy: + fail-fast: false + matrix: + python-env: [cpy314, cpy313, cpy312, cpy311, pypy311] + + steps: + - name: Checkout code + uses: actions/checkout@v4 + with: + submodules: recursive + + - name: Install Just + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + run: | + curl --proto '=https' --tlsv1.2 -sSf https://just.systems/install.sh | bash -s -- --to ~/bin + echo "$HOME/bin" >> $GITHUB_PATH + + - name: Install uv + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + run: | + curl -LsSf https://astral.sh/uv/install.sh | sh + source $HOME/.cargo/env + echo "$HOME/.cargo/bin" >> $GITHUB_PATH + + - name: Verify toolchain installation + run: | + just --version + uv --version + + - name: Setup uv cache + uses: actions/cache@v4 + with: + path: ${{ env.UV_CACHE_DIR }} + key: + uv-cache-ubuntu-unit-${{ matrix.python-env }}-${{ + hashFiles('pyproject.toml') }} + restore-keys: | + uv-cache-ubuntu-unit-${{ matrix.python-env }}- + uv-cache-ubuntu-unit- + + - name: Create Python environment and install tools + run: | + just create ${{ matrix.python-env }} + just install-tools ${{ matrix.python-env }} + + - name: Unit test - Twisted / trial + run: just test-twisted ${{ matrix.python-env }} + + - name: Unit test - asyncio / pytest + run: just test-asyncio ${{ matrix.python-env }} + + # Last: its "install" dependency replaces the editable install. + - name: All four WebSocket compression extensions available + run: just check-compressors ${{ matrix.python-env }} + test-serdes: name: SerDes Conformance Tests needs: identifiers diff --git a/docs/changelog.rst b/docs/changelog.rst index bed171a7b..01e4b9766 100644 --- a/docs/changelog.rst +++ b/docs/changelog.rst @@ -8,8 +8,17 @@ Changelog 26.9.1 ------ +**WebSocket** + +* Make all four per-message compression extensions (``permessage-deflate``, ``-bzip2``, ``-brotli``, ``-snappy``) enforce the decompression budget the same way: decompression stops near the remaining ``maxMessagePayloadSize`` budget instead of decompressing a whole frame first, and "over budget" as well as "budget already used up" raise ``PayloadExceededError``. brotli now bounds its output with ``Decompressor.process(output_buffer_limit=...)``, snappy is decompressed one framing chunk at a time, and deflate no longer lets one octet through once the budget is used up (#1986) +* Add ``autobahn.exception.PayloadCorruptError``: every compression extension now raises it for input it cannot decompress (corrupt, truncated, or not the extension's format), instead of the library's own exception (``zlib.error``, ``OSError``, ``brotli.error``, ...), and snappy no longer accepts such input silently. The protocol fails the connection with close code 1007 (invalid frame payload data), the same for every extension and on both Twisted and asyncio (#1986) +* Fix ``permessage-brotli`` failing on the second message of a connection when context takeover was negotiated (the default): the compressor and decompressor were reused after their brotli stream had finished. Each message is now a new brotli stream, which is the wire format brotli used for the first message already, and for no context takeover (#1986) +* Fix an asyncio WebSocket connection being left open, but no longer serviced, when an exception escaped the processing of received data: the connection is now dropped, as on Twisted (#1986) +* Add a documentation page on WebSocket compression (``docs/websocket/compression.rst``): for each of the four extensions, which Python package and which compression library (and bundled version) it uses on CPython and on PyPy, plus context takeover, decompression limits and the handling of invalid data. The old, unpublished page under ``docs/work/`` did not cover ``permessage-brotli`` (#1986) + **Build & CI/CD** +* Run the unit tests (Twisted/trial and asyncio/pytest) in CI on every supported Python: CPython 3.11–3.14 and PyPy 3.11, instead of on CPython 3.14 only, and with all four compression extensions installed (``python-snappy`` is added to the ``dev`` extra; ``just check-compressors`` verifies all four are available) (#1986) * Adopt the contribution workflow shared by all WAMP projects: ``CONTRIBUTING.md``, the pull request template and ``.audit/README.md`` are now deployed byte-identically from wamp-cicd (GitHub issue first, red → green tests, AI-assistance disclosure) and kept in sync by a CI drift check; everything specific to Autobahn|Python moved to a new ``DEVELOPMENT.md`` (also included in the docs). ``.cicd`` is pinned to wamp-cicd ``38ded6a``, the same commit across the WAMP fleet (#1956) * Add an ``import-smoke-lowest`` CI job (``just test-imports-lowest``) that runs the import smoke test on CPython 3.11 with every direct dependency resolved to the lowest version ``pyproject.toml`` allows (``uv --resolution lowest-direct``), so stale version floors are caught. The import smoke test now also covers the Twisted and asyncio bindings, each module imported in a fresh interpreter (#1955) @@ -19,6 +28,7 @@ Changelog **Dependencies** +* Raise the ``compress`` extra's floors to ``brotli>=1.2.0`` (CPython) and ``brotlicffi>=1.2.0`` (PyPy), the first versions with ``Decompressor.process(output_buffer_limit=...)`` (#1986) * Fix ``import autobahn.twisted`` failing on CPython 3.11–3.13 with ``TypeError: unsupported operand type(s) for |: 'InterfaceClass' and 'InterfaceClass'`` when ``zope.interface`` < 6.2 is installed: ``autobahn.twisted.util`` and ``autobahn.twisted.wamp`` now use ``from __future__ import annotations``, and the ``twisted`` extra's ``zope.interface`` floor is raised from ``>=5.2.0`` to ``>=6.2``, the first release whose interfaces support ``|`` (#1955) * Split the ``cbor2`` dependency by Python runtime: CPython uses the current ``>=6.1.0`` line (Rust/pyo3, binary wheels), while PyPy is pinned to ``cbor2==5.9.0`` — the final pure-Python release — deliberately keeping PyPy out of the native-extension matrix. This removes the ``cbor2`` Rust source-compile that was timing out the PyPy-aarch64 wheel build under QEMU emulation. The two implementations are verified wire-equivalent by a new cross-version CBOR conformance test (``examples/serdes/tests/test_cbor_cross_version.py``) that runs on both CPython and PyPy in CI and asserts each encodes/decodes the wamp-proto canonical CBOR vectors byte-for-byte identically. ``cbor2==5.9.0`` also carries the ``max_depth`` decoder hardening for CVE-2026-26209 (#1947) * Raise the ``cryptography`` floor to ``>=50`` so PyPy (including PyPy-aarch64) installs a prebuilt wheel instead of source-compiling Rust under QEMU emulation, and refresh ``pip``/``packaging`` in ``install-build-tools`` before installing so newer platform wheel tags (e.g. ``manylinux_2_34`` PyPy-aarch64) resolve to wheels. ``>=50`` is also the first release clear of the known 2026 ``cryptography`` advisories — CVE-2026-26007 (fixed 46.0.5), CVE-2026-39892 (fixed 46.0.7) and the PKCS#7 Bleichenbacher oracle (fixed 50.0.0); since ``cryptography`` is Rust-only with no pure-Python fallback it cannot be capped, so the floor is raised to the security-clean version (#1947) diff --git a/docs/programming-guide/websocket/index.rst b/docs/programming-guide/websocket/index.rst index ca38118a7..01ecafcda 100644 --- a/docs/programming-guide/websocket/index.rst +++ b/docs/programming-guide/websocket/index.rst @@ -7,5 +7,6 @@ This section covers WebSocket programming with Autobahn|Python. :maxdepth: 2 /websocket/programming + /websocket/compression /websocket/examples /websocket/conformance diff --git a/docs/websocket/compression.rst b/docs/websocket/compression.rst new file mode 100644 index 000000000..58cf0ddbb --- /dev/null +++ b/docs/websocket/compression.rst @@ -0,0 +1,139 @@ +WebSocket Compression +===================== + +Autobahn|Python implements WebSocket per-message compression (`RFC 7692 +`__) with four compression +extensions. Only ``permessage-deflate`` is standardized; the other three are +Autobahn|Python specific, and both peers must be Autobahn|Python (or another +implementation of the same extension) to use them. + +The compression itself is done by a library for each extension, which is +either part of Python or a separate package. This page lists which library, +in which version, each extension uses, on CPython and on PyPy. + +Extensions and Libraries +------------------------ + +.. list-table:: + :header-rows: 1 + :widths: 16 14 30 40 + + * - Extension + - Installed by + - Python package + - Compression library used + * - ``permessage-deflate`` + - always available + - ``zlib`` (Python standard library) + - `zlib `__ - the zlib the Python interpreter was built with + * - ``permessage-bzip2`` + - always available [1]_ + - ``bz2`` (Python standard library) + - `bzip2 `__ (libbz2) - the libbz2 the Python interpreter was built with + * - ``permessage-brotli`` + - ``compress`` extra + - CPython: `brotli `__ ``>= 1.2.0`` + (C extension, from the `google/brotli `__ + repository itself) + + PyPy: `brotlicffi `__ ``>= 1.2.0`` + (CFFI, `python-hyper/brotlicffi `__) + - `google/brotli `__ (libbrotli), bundled + in the package: brotli 1.2.0, brotlicffi 1.2.0.0 and brotlicffi 1.2.0.2 + all bundle libbrotli **1.2.0** + * - ``permessage-snappy`` + - installed separately (``pip install python-snappy``) + - `python-snappy `__ (tested with + 0.7.3), a pure Python package on top of + `cramjam `__ (Rust, PyO3) + - The Rust `snap `__ crate, an + independent implementation of the Snappy format, bundled in cramjam + (cramjam 2.14.0 bundles snap **1.1.1**). **Not** `google/snappy + `__ (libsnappy). + +.. [1] Unless the Python interpreter was built without ``bz2``; the extension + is then not available. + +Notes: + +* ``brotli`` ``>= 1.2.0`` / ``brotlicffi`` ``>= 1.2.0`` is required: older + versions lack ``Decompressor.process(output_buffer_limit=...)``, which + ``permessage-brotli`` needs to bound decompression (see below). +* Before version 0.7, python-snappy was a C extension linking google/snappy + (libsnappy). Autobahn|Python is tested with python-snappy 0.7.3 only. +* python-snappy declares no minimum version for cramjam. +* On PyPy, cramjam (a PyO3 extension) runs through PyPy's emulation of the + CPython C API (cpyext), unlike brotlicffi, which uses CFFI. +* python-snappy is not part of the ``compress`` extra; install it yourself + where you need ``permessage-snappy``. +* zlib and libbz2 come with the Python interpreter, so their version depends + on how it was built - not on Autobahn|Python. For zlib, check with + + .. code-block:: python + + import zlib + print(zlib.ZLIB_RUNTIME_VERSION) + + For example, the PyPy 3.11 Linux binaries that ``uv`` installs ship their + own ``libz.so.1``. + +To list the compression extensions available in an environment: + +.. code-block:: python + + from autobahn.websocket.compress import PERMESSAGE_COMPRESSION_EXTENSION + + print(sorted(PERMESSAGE_COMPRESSION_EXTENSION)) + +In a development checkout, ``just check-compressors `` does the same +and fails unless all four are available; CI runs it on every supported +Python. + +Context Takeover +---------------- + +With context takeover (the default), a compressor keeps its state from one +message to the next, so later messages can refer back to earlier ones. + +* ``permessage-deflate`` takes over context as negotiated + (``client_no_context_takeover`` / ``server_no_context_takeover``). +* ``permessage-snappy`` negotiates the same parameters. The Snappy framing + format has no back-references across chunks, so context takeover makes no + difference to the compression ratio. +* ``permessage-bzip2`` and ``permessage-brotli`` never take over context: + each message is one complete bzip2 or brotli stream. ``permessage-brotli`` + still negotiates the parameters. + +Decompression Limits and Invalid Data +------------------------------------- + +``maxMessagePayloadSize`` (see :doc:`programming`) bounds the size of a +message *after* decompression. All four extensions decompress incrementally +and stop once the remaining budget of the message is exceeded, rather than +decompressing a whole frame first: + +* ``permessage-deflate`` and ``permessage-bzip2`` stop one octet over the + budget, +* ``permessage-brotli`` stops within one output buffer of the budget, +* ``permessage-snappy`` stops within one Snappy chunk (at most 64 KiB) of + the budget. + +A message over the limit raises ``autobahn.exception.PayloadExceededError`` +and fails the connection with close code 1009 (message too big). This is the +same for all four extensions, on both Twisted and asyncio. + +Data an extension cannot decompress - corrupt, truncated, or not in the +extension's format - raises ``autobahn.exception.PayloadCorruptError`` and +fails the connection with close code 1007 (invalid frame payload data). This +is the same for all four extensions, on both Twisted and asyncio. + +Without a limit (``maxMessagePayloadSize = 0``, the default), a small +compressed message can decompress to a very large one. Set a limit when +accepting compressed messages from peers you do not trust. + +Example +------- + +A complete example is in the repository: +`examples/twisted/websocket/echo_compressed +`__. diff --git a/pyproject.toml b/pyproject.toml index c28c5de25..3cf056275 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -102,8 +102,10 @@ accelerate = [ # non-standard WebSocket compression support # Note: zlib/deflate is always available (Python stdlib) compress = [ - "brotli>=1.0.0; platform_python_implementation == 'CPython'", # CPyExt for CPython - "brotlicffi>=1.0.0; platform_python_implementation != 'CPython'", # CFFI for PyPy + # >=1.2.0: Decompressor.process(output_buffer_limit=...), which the + # permessage-brotli receive path needs to bound decompression (#1986) + "brotli>=1.2.0; platform_python_implementation == 'CPython'", # CPyExt for CPython + "brotlicffi>=1.2.0; platform_python_implementation != 'CPython'", # CFFI for PyPy # python-snappy is optional - only available if installed separately # Users who need snappy: pip install python-snappy ] @@ -178,6 +180,10 @@ dev = [ "pytest-asyncio>=0.14.0", "pytest-runner>=2.11.1", "pytest>=3.4.2", + # Test-only: python-snappy is an optional, user-installed dependency of + # permessage-snappy (see the "compress" extra), installed here so the + # snappy test cases run in CI too (#1986). + "python-snappy>=0.7.3", "pyyaml>=4.2b4", "qualname>=0.1.0", "ruff>=0.12.1", diff --git a/src/autobahn/asyncio/test/test_aio_websocket.py b/src/autobahn/asyncio/test/test_aio_websocket.py index ad22a5f65..99eb0c3e3 100644 --- a/src/autobahn/asyncio/test/test_aio_websocket.py +++ b/src/autobahn/asyncio/test/test_aio_websocket.py @@ -81,3 +81,87 @@ async def on_connect(req): assert len(values) == 1 assert values[0] == num * num + + +@pytest.mark.skipif( + not os.environ.get("USE_ASYNCIO", False), reason="test runs on asyncio only" +) +def test_corrupt_compressed_frame_closes_connection(): + # A compressed frame that does not decompress must fail the connection + # (#1986). It must not stop the asyncio receive loop and leave the + # connection open. + from autobahn.websocket.compress import ( + PerMessageDeflateOffer, + PerMessageDeflateOfferAccept, + ) + + def accept(offers): + for offer in offers: + if isinstance(offer, PerMessageDeflateOffer): + return PerMessageDeflateOfferAccept(offer) + + loop = asyncio.new_event_loop() + try: + factory = WebSocketServerFactory(loop=loop) + factory.setProtocolOptions(perMessageCompressionAccept=accept) + server = factory() + transport = Mock() + server.connection_made(transport) + + server.data_received( + b"\r\n".join( + [ + b"GET /ws HTTP/1.1", + b"Host: www.example.com", + b"Sec-WebSocket-Version: 13", + b"Sec-WebSocket-Extensions: permessage-deflate", + b"Sec-WebSocket-Key: tXAxWFUqnhi86Ajj7dRY5g==", + b"Connection: keep-alive, Upgrade", + b"Upgrade: websocket", + b"\r\n", # last string doesn't get a \r\n from join() + ] + ) + ) + loop.run_until_complete(asyncio.sleep(0.1)) + assert server.state == server.STATE_OPEN + assert server._perMessageCompress is not None + + # masked, FIN, RSV1 (compressed), binary frame with a corrupt body + body = b"\xff" * 64 + mask = b"\x11\x22\x33\x44" + frame = bytes([0x80 | 0x40 | 0x02, 0x80 | len(body)]) + mask + frame += bytes(b ^ mask[i % 4] for i, b in enumerate(body)) + server.data_received(frame) + loop.run_until_complete(asyncio.sleep(0.1)) + + assert transport.close.called or transport.abort.called, ( + "connection left open after a corrupt compressed frame" + ) + finally: + loop.close() + + +@pytest.mark.skipif( + not os.environ.get("USE_ASYNCIO", False), reason="test runs on asyncio only" +) +def test_unhandled_error_in_received_data_drops_connection(): + # Any exception escaping the protocol's data processing must drop the + # connection, as on Twisted (#1986) - not stop the asyncio receive loop + # and leave the connection open. + loop = asyncio.new_event_loop() + try: + factory = WebSocketServerFactory(loop=loop) + server = factory() + transport = Mock() + server.connection_made(transport) + + def broken(data): + raise RuntimeError("bug in data processing") + + server._dataReceived = broken + server.data_received(b"anything") + loop.run_until_complete(asyncio.sleep(0.1)) + + assert transport.abort.called + finally: + loop.close() diff --git a/src/autobahn/asyncio/test/test_event_loop_1965.py b/src/autobahn/asyncio/test/test_event_loop_1965.py index f1f12fd2e..a191e3ad1 100644 --- a/src/autobahn/asyncio/test/test_event_loop_1965.py +++ b/src/autobahn/asyncio/test/test_event_loop_1965.py @@ -99,8 +99,19 @@ def test_run_from_sync_code(code): env = { k: v for k, v in os.environ.items() if k not in ("USE_TWISTED", "USE_ASYNCIO") } + # Deprecation warnings are errors in the child, except the one passlib + # triggers on Python < 3.13 by importing the stdlib "crypt" module (a later + # -W option takes precedence). result = subprocess.run( - [sys.executable, "-W", "error::DeprecationWarning", "-c", code], + [ + sys.executable, + "-W", + "error::DeprecationWarning", + "-W", + "ignore:'crypt' is deprecated:DeprecationWarning", + "-c", + code, + ], capture_output=True, text=True, env=env, diff --git a/src/autobahn/asyncio/websocket.py b/src/autobahn/asyncio/websocket.py index 8cb100013..9d0ad0d85 100644 --- a/src/autobahn/asyncio/websocket.py +++ b/src/autobahn/asyncio/websocket.py @@ -112,7 +112,19 @@ def process(_): while self.receive_queue: data = self.receive_queue.popleft() if self.transport: - self._dataReceived(data) + try: + self._dataReceived(data) + except Exception: + # As Twisted does for an exception raised in + # dataReceived(): log it and drop the connection. + # Otherwise this receive loop would stop, and leave + # the connection open but unserviced. + self.log.error( + "dropping connection: unhandled error while processing received data:\n{tb}", + tb=txaio.failure_format_traceback(txaio.create_failure()), + ) + self.receive_queue.clear() + self._closeConnection(abort=True) self._consume() self.waiter.add_done_callback(process) diff --git a/src/autobahn/exception.py b/src/autobahn/exception.py index 0263f936b..23446ef64 100644 --- a/src/autobahn/exception.py +++ b/src/autobahn/exception.py @@ -26,7 +26,7 @@ from autobahn.util import public -__all__ = ("PayloadExceededError",) +__all__ = ("Disconnected", "PayloadCorruptError", "PayloadExceededError") @public @@ -37,6 +37,15 @@ class PayloadExceededError(RuntimeError): """ +@public +class PayloadCorruptError(RuntimeError): + """ + Exception raised when a compressed payload (eg of a WebSocket per-message + compression extension) cannot be decompressed, because it is corrupt or + truncated. + """ + + @public class Disconnected(RuntimeError): """ diff --git a/src/autobahn/websocket/compress_base.py b/src/autobahn/websocket/compress_base.py index 627aba019..22b5a88f2 100644 --- a/src/autobahn/websocket/compress_base.py +++ b/src/autobahn/websocket/compress_base.py @@ -73,9 +73,13 @@ class PerMessageCompress: ``max_output_len`` octets of decompressed output and raises :class:`autobahn.exception.PayloadExceededError` if the input would produce more - it never silently truncates. ``max_output_len=None`` (the default) - leaves decompression unbounded. Backends whose underlying library exposes an - incremental output limit (deflate, bzip2) enforce the bound before fully - inflating a frame; backends without one (snappy, brotli) inflate the frame - (already bounded on the wire by ``maxFramePayloadSize``) and then check, - a weaker but still-clean per-frame guarantee. + leaves decompression unbounded. Every backend stops decompressing near the + budget, instead of inflating a whole frame first: deflate and bzip2 at + exactly one octet over it, brotli at its library's output buffer + granularity, and snappy within one framing chunk (at most 64 KiB). + + Input that cannot be decompressed (corrupt, truncated, or not the + extension's format) raises :class:`autobahn.exception.PayloadCorruptError` + from ``decompress_message_data`` or ``end_decompress_message``, never the + library's own exception. """ diff --git a/src/autobahn/websocket/compress_brotli.py b/src/autobahn/websocket/compress_brotli.py index 4ca3a87f8..dceece417 100644 --- a/src/autobahn/websocket/compress_brotli.py +++ b/src/autobahn/websocket/compress_brotli.py @@ -31,7 +31,7 @@ except ImportError: import brotlicffi as brotli -from autobahn.exception import PayloadExceededError +from autobahn.exception import PayloadCorruptError, PayloadExceededError from autobahn.websocket.compress_base import ( PerMessageCompress, PerMessageCompressOffer, @@ -474,7 +474,16 @@ def compress_message_data(self, data): return self._compressor.process(data) def end_compress_message(self): - return self._compressor.finish() + data = self._compressor.finish() + + # Each message is one complete brotli stream, and a finished + # compressor must not be reused (as with bzip2). So there is no + # context takeover across messages, whatever was negotiated: the + # next message starts a new stream - which is also what the wire + # format with no context takeover is. + self._compressor = None + + return data def start_decompress_message(self): if self._is_server: @@ -485,13 +494,25 @@ def start_decompress_message(self): self._decompressor = brotli.Decompressor() def decompress_message_data(self, data, max_output_len=None): - # brotli's Decompressor.process() has no output-length argument, so the - # frame is decompressed in full (bounded on the wire by - # maxFramePayloadSize) and then checked. This is a weaker, - # per-frame-granular guarantee than the incremental cap deflate/bzip2 - # provide, but it still rejects an over-budget message cleanly. - data = self._decompressor.process(data) - if max_output_len is not None and len(data) > max_output_len: + try: + if max_output_len is None: + return self._decompressor.process(data) + # Cap at one octet over the budget (brotli/brotlicffi >= 1.2.0), + # like bzip2. brotlicffi stops exactly there; CPython's brotli + # grows its output buffer block-wise and may return more, but + # either way decompression stops near the budget, instead of + # inflating the whole frame first. If the cap was hit, the + # decoder holds back the rest (can_accept_more_data() is False): + # the message is over budget either way. + data = self._decompressor.process( + data, output_buffer_limit=max(max_output_len, 0) + 1 + ) + except brotli.error as e: + raise PayloadCorruptError(f"invalid permessage-brotli data: {e}") from e + if ( + len(data) > max_output_len + or not self._decompressor.can_accept_more_data() + ): raise PayloadExceededError( "WebSocket message exceeds decompression limit of " f"{max_output_len} octets" @@ -499,4 +520,9 @@ def decompress_message_data(self, data, max_output_len=None): return data def end_decompress_message(self): - pass + # Each message is one complete brotli stream (end_compress_message() + # finishes it), and a finished decompressor must not be reused. + complete = self._decompressor.is_finished() + self._decompressor = None + if not complete: + raise PayloadCorruptError("truncated permessage-brotli data") diff --git a/src/autobahn/websocket/compress_bzip2.py b/src/autobahn/websocket/compress_bzip2.py index fa30afd37..4eddd120c 100644 --- a/src/autobahn/websocket/compress_bzip2.py +++ b/src/autobahn/websocket/compress_bzip2.py @@ -26,7 +26,7 @@ import bz2 -from autobahn.exception import PayloadExceededError +from autobahn.exception import PayloadCorruptError, PayloadExceededError from autobahn.websocket.compress_base import ( PerMessageCompress, PerMessageCompressOffer, @@ -524,15 +524,20 @@ def start_decompress_message(self): self._decompressor = bz2.BZ2Decompressor() def decompress_message_data(self, data, max_output_len=None): - if max_output_len is None: - return self._decompressor.decompress(data) - # BZ2Decompressor.decompress(data, max_length) returns at most - # max_length bytes and buffers any excess internally. Cap at one octet - # over the budget: if that yields more than max_output_len octets the - # message is over budget and is rejected (matching deflate's - # strictly-greater boundary). (needs_input is unreliable here - it also - # goes False at end-of-stream, i.e. for an under-budget message.) - data = self._decompressor.decompress(data, max(max_output_len, 0) + 1) + try: + if max_output_len is None: + return self._decompressor.decompress(data) + # BZ2Decompressor.decompress(data, max_length) returns at most + # max_length bytes and buffers any excess internally. Cap at one + # octet over the budget: if that yields more than max_output_len + # octets the message is over budget and is rejected (matching + # deflate's strictly-greater boundary). (needs_input is unreliable + # here - it also goes False at end-of-stream, i.e. for an + # under-budget message.) + data = self._decompressor.decompress(data, max(max_output_len, 0) + 1) + except (OSError, EOFError) as e: + # OSError: invalid data; EOFError: data after the end of the stream + raise PayloadCorruptError(f"invalid permessage-bzip2 data: {e}") from e if len(data) > max_output_len: raise PayloadExceededError( "WebSocket message exceeds decompression limit of " @@ -541,4 +546,8 @@ def decompress_message_data(self, data, max_output_len=None): return data def end_decompress_message(self): + # Each message is one complete bzip2 stream. + complete = self._decompressor.eof self._decompressor = None + if not complete: + raise PayloadCorruptError("truncated permessage-bzip2 data") diff --git a/src/autobahn/websocket/compress_deflate.py b/src/autobahn/websocket/compress_deflate.py index 65c8d32bd..c38b309c8 100644 --- a/src/autobahn/websocket/compress_deflate.py +++ b/src/autobahn/websocket/compress_deflate.py @@ -26,7 +26,7 @@ import zlib -from autobahn.exception import PayloadExceededError +from autobahn.exception import PayloadCorruptError, PayloadExceededError from autobahn.util import public from autobahn.websocket.compress_base import ( PerMessageCompress, @@ -826,23 +826,27 @@ def decompress_message_data(self, data, max_output_len=None): limits.append(self.max_message_size - self._decompress_message_size) if max_output_len is not None: limits.append(max_output_len) - if not limits: - return self._decompressor.decompress(data) - - # Cap output at the tighter remaining budget. zlib treats a max_length - # of 0 as "unlimited", so once the budget is exhausted we cap the next - # call at 1 byte: any further real output then lands in unconsumed_tail - # and triggers the clean rejection below. - limit = min(limits) - data = self._decompressor.decompress(data, limit if limit > 0 else 1) + try: + if not limits: + return self._decompressor.decompress(data) + + # Cap output at the tighter remaining budget. zlib treats a + # max_length of 0 as "unlimited", so once the budget is exhausted + # we cap the next call at 1 byte: any real output then exceeds the + # budget, and triggers the clean rejection below. + limit = min(limits) + data = self._decompressor.decompress(data, limit if limit > 0 else 1) + except zlib.error as e: + raise PayloadCorruptError(f"invalid permessage-deflate data: {e}") from e self._decompress_message_size += len(data) - # A non-empty unconsumed_tail means more output was available than the - # budget allowed, i.e. the message exceeds the limit. Reject cleanly - # instead of silently truncating - truncation both drops application - # data and corrupts the deflate stream, so the subsequent - # end_decompress_message() would raise "zlib error -3" on the trailer. - if self._decompressor.unconsumed_tail: + # More output than the budget allowed (a non-empty unconsumed_tail, or + # the 1 byte let through by an exhausted budget) means the message + # exceeds the limit. Reject cleanly instead of silently truncating - + # truncation both drops application data and corrupts the deflate + # stream, so the subsequent end_decompress_message() would raise + # "zlib error -3" on the trailer. + if self._decompressor.unconsumed_tail or len(data) > limit: raise PayloadExceededError( "WebSocket message exceeds decompression limit " f"(max_message_size={self.max_message_size}, " @@ -853,4 +857,7 @@ def decompress_message_data(self, data, max_output_len=None): def end_decompress_message(self): # Eat stripped LEN and NLEN field of a non-compressed block added # for Z_SYNC_FLUSH. - self._decompressor.decompress(b"\x00\x00\xff\xff") + try: + self._decompressor.decompress(b"\x00\x00\xff\xff") + except zlib.error as e: + raise PayloadCorruptError(f"invalid permessage-deflate data: {e}") from e diff --git a/src/autobahn/websocket/compress_snappy.py b/src/autobahn/websocket/compress_snappy.py index b95d370e5..5983206d7 100644 --- a/src/autobahn/websocket/compress_snappy.py +++ b/src/autobahn/websocket/compress_snappy.py @@ -26,7 +26,7 @@ import snappy -from autobahn.exception import PayloadExceededError +from autobahn.exception import PayloadCorruptError, PayloadExceededError from autobahn.websocket.compress_base import ( PerMessageCompress, PerMessageCompressOffer, @@ -35,6 +35,12 @@ PerMessageCompressResponseAccept, ) +# Snappy framing format: a stream identifier chunk (type 0xff) has length 6, +# a data chunk (type 0x00 compressed, 0x01 uncompressed) is at most +# 4 + 76490 octets long (CRC, plus the maximum compressed size of the 64 KiB +# a chunk may hold), and chunk types 0x02-0x7f are reserved, not skippable. +_SNAPPY_MAX_DATA_CHUNK_LENGTH = 4 + 76490 + __all__ = ( "PerMessageSnappy", "PerMessageSnappyMixin", @@ -444,6 +450,8 @@ def __init__( self._compressor = None self._decompressor = None + # received input not yet fed to the decompressor (an incomplete chunk) + self._decompress_pending = bytearray() def __json__(self): return { @@ -475,23 +483,60 @@ def start_decompress_message(self): if self._is_server: if self._decompressor is None or self.client_no_context_takeover: self._decompressor = snappy.StreamDecompressor() + self._decompress_pending = bytearray() else: if self._decompressor is None or self.server_no_context_takeover: self._decompressor = snappy.StreamDecompressor() + self._decompress_pending = bytearray() def decompress_message_data(self, data, max_output_len=None): # python-snappy's StreamDecompressor has no output-length argument, so - # the frame is decompressed in full (bounded on the wire by - # maxFramePayloadSize) and then checked. This is a weaker, - # per-frame-granular guarantee than the incremental cap deflate/bzip2 - # provide, but it still rejects an over-budget message cleanly. - data = self._decompressor.decompress(data) - if max_output_len is not None and len(data) > max_output_len: - raise PayloadExceededError( - "WebSocket message exceeds decompression limit of " - f"{max_output_len} octets" - ) - return data + # it is fed one framing chunk (4-octet header: type, 24-bit length) at + # a time. A chunk decompresses to at most 64 KiB (the decoder rejects + # larger ones), so decompression stops within 64 KiB of the budget, + # instead of inflating the whole frame first. + # (A bytearray, so input trickling in over many calls is appended, + # not copied anew each time.) + buf = self._decompress_pending + buf += data + pos = 0 + out = [] + size = 0 + while len(buf) - pos >= 4: + kind = buf[pos] + length = int.from_bytes(buf[pos + 1 : pos + 4], "little") + # reject a bad chunk header right away, instead of buffering a + # chunk that cannot be valid + if ( + (kind == 0xFF and length != 6) + or (kind <= 0x01 and length > _SNAPPY_MAX_DATA_CHUNK_LENGTH) + or 0x02 <= kind <= 0x7F + ): + raise PayloadCorruptError( + "invalid permessage-snappy data: bad chunk header " + f"(type 0x{kind:02x}, length {length})" + ) + end = pos + 4 + length + if end > len(buf): + break + try: + chunk = self._decompressor.decompress(bytes(buf[pos:end])) + except (OSError, snappy.UncompressError) as e: + raise PayloadCorruptError(f"invalid permessage-snappy data: {e}") from e + pos = end + size += len(chunk) + if max_output_len is not None and size > max_output_len: + raise PayloadExceededError( + "WebSocket message exceeds decompression limit of " + f"{max_output_len} octets" + ) + out.append(chunk) + del buf[:pos] + return b"".join(out) def end_decompress_message(self): - pass + # A message ends on a chunk boundary: anything left over is a + # truncated chunk, or not snappy framing at all. + if self._decompress_pending: + self._decompress_pending = bytearray() + raise PayloadCorruptError("truncated permessage-snappy data") diff --git a/src/autobahn/websocket/protocol.py b/src/autobahn/websocket/protocol.py index cfa7fb2bc..05af6e136 100755 --- a/src/autobahn/websocket/protocol.py +++ b/src/autobahn/websocket/protocol.py @@ -46,7 +46,11 @@ import txaio from autobahn import __version__ -from autobahn.exception import Disconnected, PayloadExceededError +from autobahn.exception import ( + Disconnected, + PayloadCorruptError, + PayloadExceededError, +) from autobahn.util import ( ObservableMixin, Stopwatch, @@ -1862,12 +1866,12 @@ def onFrameData(self, payload: bytes) -> bool | None: # length to message_data_total_length, so subtracting it back # out yields the uncompressed total of the preceding frames; the # remainder up to maxMessagePayloadSize is what this frame may - # inflate to. Passing it as max_output_len lets backends with an - # incremental cap (deflate, bzip2) stop inflating at the limit - # instead of expanding the whole frame into memory first. A - # PayloadExceededError means the message exceeds the limit; the - # post-inflation check below is the backstop for backends that - # can only bound per-frame (snappy, brotli). + # inflate to. Passing it as max_output_len lets every backend + # stop inflating near the limit, instead of expanding the whole + # frame into memory first. A PayloadExceededError means the + # message exceeds the limit; the post-inflation check below is + # the backstop. A PayloadCorruptError means the frame cannot + # be decompressed. if self.maxMessagePayloadSize > 0: max_output_len = self.maxMessagePayloadSize - ( self.message_data_total_length - compressedLen @@ -1887,6 +1891,14 @@ def onFrameData(self, payload: bytes) -> bool | None: f"received WebSocket message exceeds payload limit of {self.maxMessagePayloadSize} octets after decompression", ) return False + except PayloadCorruptError as e: + # the same for every compression backend: fail the + # connection with 1007 (invalid frame payload data) + if not self.failedByMe: + self._invalid_payload( + f"received WebSocket message cannot be decompressed: {e}" + ) + return False uncompressedLen = len(payload) else: l = len(payload) @@ -1955,7 +1967,14 @@ def onFrameEnd(self) -> bool | None: # handle end of compressed message # if self._isMessageCompressed: - self._perMessageCompress.end_decompress_message() + try: + self._perMessageCompress.end_decompress_message() + except PayloadCorruptError as e: + if not self.failedByMe: + self._invalid_payload( + f"received WebSocket message cannot be decompressed: {e}" + ) + return False # verify UTF8 has actually ended # diff --git a/src/autobahn/websocket/test/test_websocket_compress.py b/src/autobahn/websocket/test/test_websocket_compress.py index f3fff9866..2173d8938 100644 --- a/src/autobahn/websocket/test/test_websocket_compress.py +++ b/src/autobahn/websocket/test/test_websocket_compress.py @@ -29,14 +29,20 @@ # permessage-compress backends are shared code, so bounded-decompression must # behave identically regardless of the networking backend. +import random import unittest import zlib -from autobahn.exception import PayloadExceededError +try: + import tracemalloc +except ImportError: # PyPy has no tracemalloc + tracemalloc = None + +from autobahn.exception import PayloadCorruptError, PayloadExceededError from autobahn.websocket.compress_deflate import PerMessageDeflate -def _make_compressor_pair(name): +def _make_compressor_pair(name, no_context_takeover=False): """ Build a (client_compressor, server_decompressor) pair for the named permessage-compress extension, or return None if its optional dependency @@ -44,13 +50,20 @@ def _make_compressor_pair(name): extension-level max_message_size, so the bounded-decompress behaviour under test is driven purely by the ``max_output_len`` argument. + ``no_context_takeover`` applies to both directions (permessage-bzip2 has + no such parameter: it never takes over context). + Concrete per-codec imports (rather than the union-typed extension registry) keep the constructor call sites resolvable for the static type checker. """ if name == "permessage-deflate": return ( - PerMessageDeflate(False, False, False, 15, 15, 8), - PerMessageDeflate(True, False, False, 15, 15, 8), + PerMessageDeflate( + False, no_context_takeover, no_context_takeover, 15, 15, 8 + ), + PerMessageDeflate( + True, no_context_takeover, no_context_takeover, 15, 15, 8 + ), ) if name == "permessage-bzip2": try: @@ -64,8 +77,8 @@ def _make_compressor_pair(name): except ImportError: return None return ( - PerMessageSnappy(False, False, False), - PerMessageSnappy(True, False, False), + PerMessageSnappy(False, no_context_takeover, no_context_takeover), + PerMessageSnappy(True, no_context_takeover, no_context_takeover), ) if name == "permessage-brotli": try: @@ -73,8 +86,8 @@ def _make_compressor_pair(name): except ImportError: return None return ( - PerMessageBrotli(False, False, False), - PerMessageBrotli(True, False, False), + PerMessageBrotli(False, no_context_takeover, no_context_takeover), + PerMessageBrotli(True, no_context_takeover, no_context_takeover), ) return None @@ -171,6 +184,240 @@ def test_unbounded_default_roundtrips(self): decompressor.end_decompress_message() +class DecompressBudgetUniformTests(unittest.TestCase): + """ + Every compression backend must enforce the ``max_output_len`` budget the + same way (#1986): "over budget" and "budget already 0" both raise + ``PayloadExceededError``, and the backend must stop inflating at the + budget instead of inflating a whole frame first and checking afterwards. + """ + + BUDGET = 64 * 1024 + # A run of zeros inflates far beyond BUDGET from a few bytes on the wire + # (except snappy, whose format caps expansion at about 21:1). + BOMB = b"\0" * (32 * 1024 * 1024) + # Peak allocation allowed while rejecting BOMB, on top of one copy of the + # input (snappy buffers it): generous against BUDGET (library buffers, + # CPython brotli's block-granular output limit, one snappy framing chunk), + # and still a 32x margin below inflating BOMB. + PEAK_LIMIT = 1024 * 1024 + + @staticmethod + def _compress(compressor, payload): + compressor.start_compress_message() + body = compressor.compress_message_data(payload) + body += compressor.end_compress_message() + return body + + def _pairs(self): + for codec in _CODECS: + pair = _make_compressor_pair(codec) + if pair is not None: + yield codec, pair + + def test_budget_zero_rejects(self): + # No budget left: any further output is over budget - including a + # single octet, which a 1-octet output cap would let through. + for codec in _CODECS: + for size in (1, 16): + with self.subTest(codec=codec, size=size): + pair = _make_compressor_pair(codec) + if pair is None: + continue + compressor, decompressor = pair + body = self._compress(compressor, b"x" * size) + decompressor.start_decompress_message() + self.assertRaises( + PayloadExceededError, + decompressor.decompress_message_data, + body, + max_output_len=0, + ) + + def test_one_octet_over_budget_rejects(self): + payload = random.Random(1986).randbytes(4096) + for codec, (compressor, decompressor) in self._pairs(): + with self.subTest(codec=codec): + body = self._compress(compressor, payload) + decompressor.start_decompress_message() + self.assertRaises( + PayloadExceededError, + decompressor.decompress_message_data, + body, + max_output_len=len(payload) - 1, + ) + + def test_over_budget_across_frames_rejects(self): + # The protocol passes the REMAINING budget for each frame of a message. + payload = b"x" * 8192 + budget = 4096 + for codec, (compressor, decompressor) in self._pairs(): + with self.subTest(codec=codec): + body = self._compress(compressor, payload) + half = len(body) // 2 + decompressor.start_decompress_message() + got = 0 + with self.assertRaises(PayloadExceededError): + for frame in (body[:half], body[half:]): + out = decompressor.decompress_message_data( + frame, max_output_len=budget - got + ) + got += len(out) + self.assertLessEqual(got, budget) + + @unittest.skipIf(tracemalloc is None, "tracemalloc not available (PyPy)") + def test_bomb_inflation_bounded_by_budget(self): + # Black-box check, independent of how a backend bounds its output: + # rejecting an over-budget frame must not allocate the inflated frame. + for codec, (compressor, decompressor) in self._pairs(): + with self.subTest(codec=codec): + body = self._compress(compressor, self.BOMB) + decompressor.start_decompress_message() + tracemalloc.start() + try: + with self.assertRaises(PayloadExceededError): + decompressor.decompress_message_data( + body, max_output_len=self.BUDGET + ) + peak = tracemalloc.get_traced_memory()[1] + finally: + tracemalloc.stop() + self.assertLess( + peak, + self.PEAK_LIMIT + len(body), + f"{codec}: {len(body)} octets on the wire allocated " + f"{peak} octets against a budget of {self.BUDGET}", + ) + + +class DecompressCorruptInputUniformTests(unittest.TestCase): + """ + Every compression backend must report input it cannot decompress the + same way (#1986): as ``PayloadCorruptError``, never as the library's own + exception (``zlib.error``, ``OSError``, ``brotli.error``, ...), and never + by silently returning nothing. + """ + + @staticmethod + def _compress(compressor, payload): + compressor.start_compress_message() + body = compressor.compress_message_data(payload) + body += compressor.end_compress_message() + return body + + @staticmethod + def _decompress(decompressor, frames, max_output_len=None): + decompressor.start_decompress_message() + out = b"".join( + decompressor.decompress_message_data(f, max_output_len=max_output_len) + for f in frames + ) + decompressor.end_decompress_message() + return out + + def test_corrupt_input_raises_payload_corrupt(self): + for codec in _CODECS: + for max_output_len in (None, 1024 * 1024): + with self.subTest(codec=codec, max_output_len=max_output_len): + pair = _make_compressor_pair(codec) + if pair is None: + continue + _, decompressor = pair + # raised when the data is processed, or - for a backend + # that buffers incomplete input (snappy) - at message end + self.assertRaises( + PayloadCorruptError, + self._decompress, + decompressor, + [b"\xff" * 64], + max_output_len, + ) + + def test_truncated_message_raises_payload_corrupt(self): + # bzip2, brotli and snappy send each message as complete streams or + # chunks, so a truncated message is detectable at message end. A + # deflate message ends with a sync flush, not a stream end, so + # truncation is not detectable there in general. + payload = random.Random(1986).randbytes(8192) + for codec in ("permessage-bzip2", "permessage-brotli", "permessage-snappy"): + for max_output_len in (None, 1024 * 1024): + with self.subTest(codec=codec, max_output_len=max_output_len): + pair = _make_compressor_pair(codec) + if pair is None: + continue + compressor, decompressor = pair + body = self._compress(compressor, payload) + self.assertRaises( + PayloadCorruptError, + self._decompress, + decompressor, + [body[:-8]], + max_output_len, + ) + + def test_split_frames_roundtrip(self): + # A message split over two frames at any octet, also inside a snappy + # chunk header, must decompress byte-exact. + payload = random.Random(1986).randbytes(2 * 65536 + 123) + for codec in _CODECS: + pair = _make_compressor_pair(codec) + if pair is None: + continue + compressor, _ = pair + body = self._compress(compressor, payload) + for split in (1, 3, 4, 5, 11, 12, 13, len(body) // 2, len(body) - 1): + for max_output_len in (None, len(payload)): + with self.subTest( + codec=codec, split=split, max_output_len=max_output_len + ): + _, decompressor = _make_compressor_pair(codec) + out = self._decompress( + decompressor, + [body[:split], body[split:]], + max_output_len, + ) + self.assertEqual(out, payload) + + +class MultiMessageRoundtripTests(unittest.TestCase): + """ + One compressor / decompressor pair must carry any number of messages, + with and without context takeover (the default is context takeover). + The other tests in this module build a fresh pair per message, which + hides a backend that cannot be reused after its first message. + """ + + def test_messages_roundtrip_on_one_pair(self): + rng = random.Random(1986) + payloads = [ + b"Hello, world! " * 100, + rng.randbytes(5000), + b"", + b"x" * 70000, + rng.randbytes(3), + ] + for codec in _CODECS: + for no_context_takeover in (False, True): + for max_output_len in (None, 1024 * 1024): + with self.subTest( + codec=codec, + no_context_takeover=no_context_takeover, + max_output_len=max_output_len, + ): + pair = _make_compressor_pair(codec, no_context_takeover) + if pair is None: + continue + compressor, decompressor = pair + for i, payload in enumerate(payloads): + body = DecompressCorruptInputUniformTests._compress( + compressor, payload + ) + out = DecompressCorruptInputUniformTests._decompress( + decompressor, [body], max_output_len + ) + self.assertEqual(out, payload, f"message {i}") + + class PerMessageDeflateMaxMessageSizeTests(unittest.TestCase): """ The permessage-deflate extension-level ``max_message_size`` cap (relocated diff --git a/src/autobahn/websocket/test/test_websocket_compress_corrupt.py b/src/autobahn/websocket/test/test_websocket_compress_corrupt.py new file mode 100644 index 000000000..8f4cb8068 --- /dev/null +++ b/src/autobahn/websocket/test/test_websocket_compress_corrupt.py @@ -0,0 +1,97 @@ +############################################################################### +# +# The MIT License (MIT) +# +# Copyright (c) typedef int GmbH +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +# THE SOFTWARE. +# +############################################################################### + +# Backend-neutral: this module is NOT gated on USE_TWISTED, so it runs under +# both the Twisted (trial) and asyncio (pytest) coverage phases, exercising the +# shared WebSocket protocol handling on both networking backends. + +import unittest + +from autobahn.websocket.protocol import WebSocketProtocol +from autobahn.websocket.test.test_websocket_max_message_size import ( + _CODECS, + _build_compressed_body_frame, + _make_compressor_pair, + _make_server, +) + +# Not a valid compressed stream for any of the backends. +_CORRUPT_BODY = b"\xff" * 64 + + +class CorruptCompressedFrameTests(unittest.TestCase): + """ + A compressed frame that does not decompress must fail the WebSocket + connection cleanly, the same way for every compression backend (#1986): + no decoder exception escapes the protocol, the connection is failed with + close code 1007 (invalid frame payload data), and nothing is delivered to + the application. + """ + + LIMIT = 1024 * 1024 + + def _assert_failed_cleanly(self, streaming, fail_by_drop=True): + for codec in _CODECS: + with self.subTest(codec=codec): + pair = _make_compressor_pair(codec) + if pair is None: + continue + _, server_decompressor = pair + proto = _make_server( + server_decompressor, self.LIMIT, streaming=streaming + ) + proto.failByDrop = fail_by_drop + # no closing-handshake timer: it would outlive the test + proto.closeHandshakeTimeout = 0 + try: + proto._dataReceived(_build_compressed_body_frame(_CORRUPT_BODY)) + except Exception as e: + self.fail( + f"{codec}: decoder exception escaped the protocol: " + f"{type(e).__module__}.{type(e).__name__}: {e}" + ) + self.assertTrue(proto.failedByMe, f"{codec}: connection not failed") + if fail_by_drop: + self.assertTrue(proto.was_dropped, f"{codec}: connection not dropped") + else: + self.assertEqual( + proto.localCloseCode, + WebSocketProtocol.CLOSE_STATUS_CODE_INVALID_PAYLOAD, + f"{codec}: wrong close code", + ) + self.assertEqual(proto.delivered, [], f"{codec}: message was delivered") + self.assertEqual(proto.streamed, [], f"{codec}: frame was delivered") + + def test_message_mode_corrupt_frame_fails_connection(self): + self._assert_failed_cleanly(streaming=False) + + def test_streaming_mode_corrupt_frame_fails_connection(self): + self._assert_failed_cleanly(streaming=True) + + def test_corrupt_frame_closes_with_1007(self): + # failByDrop=False: the connection is failed with a closing handshake, + # which carries the close code. + self._assert_failed_cleanly(streaming=False, fail_by_drop=False) diff --git a/src/autobahn/websocket/test/test_websocket_max_message_size.py b/src/autobahn/websocket/test/test_websocket_max_message_size.py index 40564c723..8401bed8e 100644 --- a/src/autobahn/websocket/test/test_websocket_max_message_size.py +++ b/src/autobahn/websocket/test/test_websocket_max_message_size.py @@ -95,7 +95,14 @@ def _build_compressed_frame(client_compressor, payload, opcode=0x02): client_compressor.start_compress_message() body = client_compressor.compress_message_data(payload) body += client_compressor.end_compress_message() + return _build_compressed_body_frame(body, opcode) + +def _build_compressed_body_frame(body, opcode=0x02): + """ + Encode a single masked, FIN client->server frame with RSV1 (compressed) + set, carrying `body` as is - e.g. a corrupt compressed body. + """ mask = b"\x11\x22\x33\x44" masked = bytes(b ^ mask[i % 4] for i, b in enumerate(body)) n = len(body) @@ -214,8 +221,10 @@ class WebSocketMaxMessagePayloadSizeTests(unittest.TestCase): streaming processing modes. """ - LIMIT = 128 - BIG = b"x" * 4096 # inflates well beyond LIMIT, compresses to a few bytes + # The compressed frame must be under LIMIT, and its inflated size over it: + # BIG compresses to a few octets (snappy, at most about 21:1: 222 octets). + LIMIT = 1024 + BIG = b"x" * 4096 def _run(self, codec, streaming): pair = _make_compressor_pair(codec)