From 20767269fd27dffd48ea16c9768e57ba1707c15a Mon Sep 17 00:00:00 2001 From: Dominik Buszowiecki Date: Tue, 1 Sep 2026 13:40:23 -0400 Subject: [PATCH 1/4] fix(replays): Bound recording segment decompression at ingest Recording segments were inflated with an uncapped `zlib.decompress` before being stored. Relay bounds the compressed payload at 10 MiB, but only applies its uncompressed bound for orgs with recording scrubbing and a non-empty PII config, so a highly compressible segment could be stored with no bound on its inflated size. Decompress incrementally against a byte cap instead, and drop segments that exceed it. The limit comes from a new `replay.consumer.max-segment-decompressed-size` option (default 100 MiB) so it can be tuned without a deploy; drops are counted under `replays.consumer.recording.oversized_segment`. A follow-up applies the same bound on the read path. Refs REPLAY-994 --- src/sentry/options/defaults.py | 8 ++ src/sentry/replays/consumers/recording.py | 45 +++++++++-- .../integration/consumers/test_recording.py | 6 +- .../replays/unit/consumers/test_recording.py | 81 ++++++++++++++++--- 4 files changed, 123 insertions(+), 17 deletions(-) diff --git a/src/sentry/options/defaults.py b/src/sentry/options/defaults.py index 0f4e31a6346e..9485cb21e10f 100644 --- a/src/sentry/options/defaults.py +++ b/src/sentry/options/defaults.py @@ -528,6 +528,14 @@ default=False, flags=FLAG_AUTOMATOR_MODIFIABLE, ) +# Maximum number of bytes a single recording segment may decompress to at ingest. Segments +# which inflate past this limit are dropped instead of stored. +register( + "replay.consumer.max-segment-decompressed-size", + type=Int, + default=100 * 1024 * 1024, + flags=FLAG_AUTOMATOR_MODIFIABLE, +) # Enable new database query caching. register( "replay.consumer.enable_new_query_caching_system", diff --git a/src/sentry/replays/consumers/recording.py b/src/sentry/replays/consumers/recording.py index a0a93df035e8..09d33ed0526e 100644 --- a/src/sentry/replays/consumers/recording.py +++ b/src/sentry/replays/consumers/recording.py @@ -1,6 +1,6 @@ import logging import zlib -from typing import cast +from typing import NoReturn, cast import sentry_sdk from sentry_kafka_schemas.codecs import Codec, ValidationError @@ -37,7 +37,10 @@ class DropSilently(Exception): @trace def process_message(message: bytes) -> ProcessedEvent | None: try: - recording_event = parse_recording_event(message) + recording_event = parse_recording_event( + message, + max_segment_size=options.get("replay.consumer.max-segment-decompressed-size"), + ) set_tag("org_id", recording_event["context"]["org_id"]) sentry_sdk.set_attribute("org_id", recording_event["context"]["org_id"]) set_tag("project_id", recording_event["context"]["project_id"]) @@ -54,10 +57,10 @@ def process_message(message: bytes) -> ProcessedEvent | None: @trace -def parse_recording_event(message: bytes) -> Event: +def parse_recording_event(message: bytes, max_segment_size: int) -> Event: recording = parse_request_message(message) segment_id, payload = parse_headers(cast(bytes, recording["payload"]), recording["replay_id"]) - compressed, decompressed = decompress_segment(payload) + compressed, decompressed = decompress_segment(payload, max_segment_size) replay_event_json = recording.get("replay_event") if replay_event_json: @@ -111,16 +114,46 @@ def parse_request_message(message: bytes) -> ReplayRecording: @trace -def decompress_segment(segment: bytes) -> tuple[bytes, bytes]: +def decompress_segment(segment: bytes, max_size: int) -> tuple[bytes, bytes]: + """Return the compressed and decompressed forms of a recording segment. + + Decompression is bounded to `max_size` bytes. Stored segments are inflated again on every + read, so a segment with an implausible compression ratio is a durable, repeatable cost on + the read path -- we drop it here rather than persist it. + """ try: - return (segment, zlib.decompress(segment)) + # Ask for one byte past the limit so an oversized segment is detectable without ever + # materializing more than `max_size + 1` bytes. + decompressor = zlib.decompressobj() + decompressed = decompressor.decompress(segment, max_size + 1) + + # The incremental API returns partial output where `zlib.decompress` raised on a + # truncated stream. Normalize so malformed bodies take the branch below. + if not decompressor.eof and not decompressor.unconsumed_tail: + raise zlib.error("incomplete or truncated stream") except zlib.error: if segment and segment[0] == ord("["): + if len(segment) > max_size: + _drop_oversized_segment(len(segment), max_size) return (zlib.compress(segment), segment) else: logger.exception("Invalid recording body.") raise DropSilently() + if len(decompressed) > max_size: + _drop_oversized_segment(len(segment), max_size) + + return (segment, decompressed) + + +def _drop_oversized_segment(compressed_size: int, max_size: int) -> NoReturn: + metrics.incr("replays.consumer.recording.oversized_segment") + logger.warning( + "Dropped a recording segment which decompressed past the size limit.", + extra={"compressed_size": compressed_size, "max_size": max_size}, + ) + raise DropSilently() + @trace def parse_headers(recording: bytes, replay_id: str) -> tuple[int, bytes]: diff --git a/tests/sentry/replays/integration/consumers/test_recording.py b/tests/sentry/replays/integration/consumers/test_recording.py index c83b8a1f9159..9ead133c5f18 100644 --- a/tests/sentry/replays/integration/consumers/test_recording.py +++ b/tests/sentry/replays/integration/consumers/test_recording.py @@ -9,10 +9,14 @@ from sentry.replays.tasks import process_replay_recording from sentry.utils import json +# Options read on this path are boolean toggles we want enabled, except for the segment size +# limit which needs a real byte count. +_OPTIONS: dict[str, object] = {"replay.consumer.max-segment-decompressed-size": 100 * 1024 * 1024} + @mock.patch("sentry.options.get") def test_recording_task(options_get: MagicMock) -> None: - options_get.return_value = True + options_get.side_effect = lambda key, *args, **kwargs: _OPTIONS.get(key, True) headers = json.dumps({"segment_id": 42}).encode() recording_payload = headers + b"\n" + zlib.compress(json.dumps(MOCK_EVENTS).encode()) diff --git a/tests/sentry/replays/unit/consumers/test_recording.py b/tests/sentry/replays/unit/consumers/test_recording.py index d16a11adcfb5..972eaa8dcf45 100644 --- a/tests/sentry/replays/unit/consumers/test_recording.py +++ b/tests/sentry/replays/unit/consumers/test_recording.py @@ -15,20 +15,23 @@ from sentry.replays.usecases.ingest import ProcessedEvent from sentry.replays.usecases.ingest.event_parser import ParsedEventMeta from sentry.replays.usecases.pack import pack +from sentry.testutils.helpers import override_options from sentry.testutils.pytest.fixtures import django_db_all from sentry.utils import json +MAX_SIZE = 100 * 1024 * 1024 + def test_decompress_segment_success() -> None: """Test successful decompression of segment""" data = b"[hello, world!]" compressed_data = zlib.compress(data) - compressed, decompressed = decompress_segment(compressed_data) + compressed, decompressed = decompress_segment(compressed_data, MAX_SIZE) assert compressed == compressed_data assert decompressed == data - compressed, decompressed = decompress_segment(data) + compressed, decompressed = decompress_segment(data, MAX_SIZE) assert compressed == compressed_data assert decompressed == data @@ -38,7 +41,7 @@ def test_decompress_segment_already_decompressed() -> None: data = b"[hello, world!]" compressed_data = zlib.compress(data) - compressed, decompressed = decompress_segment(data) + compressed, decompressed = decompress_segment(data, MAX_SIZE) assert compressed == compressed_data assert decompressed == data @@ -46,13 +49,71 @@ def test_decompress_segment_already_decompressed() -> None: def test_decompress_segment_unexpected_start_character() -> None: """Test handling of invalid data that can't be decompressed""" with pytest.raises(DropSilently): - decompress_segment(b"hello, world!") + decompress_segment(b"hello, world!", MAX_SIZE) def test_decompress_segment_empty_data() -> None: """Test handling of empty data""" with pytest.raises(DropSilently): - decompress_segment(b"") + decompress_segment(b"", MAX_SIZE) + + +def test_decompress_segment_exceeds_max_size() -> None: + """Test a segment which inflates past the limit is dropped""" + payload = b"[" + b" " * (1024 * 1024) + b"]" + bomb = zlib.compress(payload, 9) + # A kilobyte of stored input, a megabyte of output on every read. + assert len(payload) // len(bomb) > 500 + + with pytest.raises(DropSilently): + decompress_segment(bomb, 1024) + + +def test_decompress_segment_at_max_size() -> None: + """Test a segment which inflates to exactly the limit is accepted""" + data = b"[hello, world!]" + compressed_data = zlib.compress(data) + + compressed, decompressed = decompress_segment(compressed_data, len(data)) + assert compressed == compressed_data + assert decompressed == data + + with pytest.raises(DropSilently): + decompress_segment(compressed_data, len(data) - 1) + + +def test_decompress_segment_uncompressed_exceeds_max_size() -> None: + """Test an uncompressed segment larger than the limit is dropped""" + data = b"[" + b" " * 1024 + b"]" + + with pytest.raises(DropSilently): + decompress_segment(data, 512) + + +@django_db_all +def test_process_message_oversized_segment() -> None: + """Test "process_message" function drops a segment which inflates past the limit.""" + bomb = zlib.compress(b"[" + b" " * (1024 * 1024) + b"]", 9) + message = { + "type": "replay_recording_not_chunked", + "org_id": 3, + "project_id": 4, + "replay_id": "1", + "received": 2, + "retention_days": 30, + "payload": json.dumps({"segment_id": 42}).encode() + b"\n" + bomb, + "key_id": 1, + "replay_event": b"{}", + "replay_video": b"", + "version": 0, + } + + kafka_message = make_kafka_message(message) + with override_options({"replay.consumer.max-segment-decompressed-size": 1024}): + assert process_message(kafka_message) is None + + # The same message is accepted under the default limit. + assert process_message(kafka_message) is not None def test_parse_headers_success() -> None: @@ -143,7 +204,7 @@ def test_parse_recording_event_success() -> None: "version": 0, } - result = parse_recording_event(msgpack.packb(message)) + result = parse_recording_event(msgpack.packb(message), MAX_SIZE) expected = { "context": { @@ -191,7 +252,7 @@ def test_parse_recording_event_with_replay_event() -> None: "version": 0, } - result = parse_recording_event(msgpack.packb(message)) + result = parse_recording_event(msgpack.packb(message), MAX_SIZE) expected = { "context": { @@ -226,7 +287,7 @@ def test_parse_recording_event_missing_payload() -> None: "version": 0, } with pytest.raises(DropSilently): - parse_recording_event(msgpack.packb(message)) + parse_recording_event(msgpack.packb(message), MAX_SIZE) def test_parse_recording_event_invalid_compression() -> None: @@ -242,7 +303,7 @@ def test_parse_recording_event_invalid_compression() -> None: "version": 0, } with pytest.raises(DropSilently): - parse_recording_event(msgpack.packb(message)) + parse_recording_event(msgpack.packb(message), MAX_SIZE) def test_parse_recording_event_invalid_headers() -> None: @@ -258,7 +319,7 @@ def test_parse_recording_event_invalid_headers() -> None: "version": 0, } with pytest.raises(DropSilently): - parse_recording_event(msgpack.packb(message)) + parse_recording_event(msgpack.packb(message), MAX_SIZE) @django_db_all From 3b58adffe2bff609ca9181253d439ba45b9b5a55 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Wed, 28 Oct 2026 15:44:24 +0000 Subject: [PATCH 2/4] ref(replays): Read segment size limit inside decompress_segment Read the option where it is used instead of threading it through process_message and parse_recording_event. Refs REPLAY-994 Co-Authored-By: Dominik Buszowiecki <44422760+DominikB2014@users.noreply.github.com> --- src/sentry/replays/consumers/recording.py | 14 +++---- .../replays/unit/consumers/test_recording.py | 40 ++++++++++--------- 2 files changed, 27 insertions(+), 27 deletions(-) diff --git a/src/sentry/replays/consumers/recording.py b/src/sentry/replays/consumers/recording.py index f05efcbe0a48..c6e9f5de7bba 100644 --- a/src/sentry/replays/consumers/recording.py +++ b/src/sentry/replays/consumers/recording.py @@ -36,10 +36,7 @@ class DropSilently(Exception): @traces.trace def process_message(message: bytes) -> ProcessedEvent | None: try: - recording_event = parse_recording_event( - message, - max_segment_size=options.get("replay.consumer.max-segment-decompressed-size"), - ) + recording_event = parse_recording_event(message) set_tag("org_id", recording_event["context"]["org_id"]) sentry_sdk.set_attribute("org_id", recording_event["context"]["org_id"]) set_tag("project_id", recording_event["context"]["project_id"]) @@ -56,10 +53,10 @@ def process_message(message: bytes) -> ProcessedEvent | None: @traces.trace -def parse_recording_event(message: bytes, max_segment_size: int) -> Event: +def parse_recording_event(message: bytes) -> Event: recording = parse_request_message(message) segment_id, payload = parse_headers(cast(bytes, recording["payload"]), recording["replay_id"]) - compressed, decompressed = decompress_segment(payload, max_segment_size) + compressed, decompressed = decompress_segment(payload) replay_event_json = recording.get("replay_event") if replay_event_json: @@ -113,13 +110,14 @@ def parse_request_message(message: bytes) -> ReplayRecording: @traces.trace -def decompress_segment(segment: bytes, max_size: int) -> tuple[bytes, bytes]: +def decompress_segment(segment: bytes) -> tuple[bytes, bytes]: """Return the compressed and decompressed forms of a recording segment. - Decompression is bounded to `max_size` bytes. Stored segments are inflated again on every + Decompression is bounded by `replay.consumer.max-segment-decompressed-size`. Stored segments are inflated again on every read, so a segment with an implausible compression ratio is a durable, repeatable cost on the read path -- we drop it here rather than persist it. """ + max_size = options.get("replay.consumer.max-segment-decompressed-size") try: # Ask for one byte past the limit so an oversized segment is detectable without ever # materializing more than `max_size + 1` bytes. diff --git a/tests/sentry/replays/unit/consumers/test_recording.py b/tests/sentry/replays/unit/consumers/test_recording.py index 9b273fa4e5a0..7e64af9b5dea 100644 --- a/tests/sentry/replays/unit/consumers/test_recording.py +++ b/tests/sentry/replays/unit/consumers/test_recording.py @@ -22,19 +22,17 @@ from sentry.testutils.pytest.fixtures import django_db_all from sentry.utils import json -MAX_SIZE = 100 * 1024 * 1024 - def test_decompress_segment_success() -> None: """Test successful decompression of segment""" data = b"[hello, world!]" compressed_data = zlib.compress(data) - compressed, decompressed = decompress_segment(compressed_data, MAX_SIZE) + compressed, decompressed = decompress_segment(compressed_data) assert compressed == compressed_data assert decompressed == data - compressed, decompressed = decompress_segment(data, MAX_SIZE) + compressed, decompressed = decompress_segment(data) assert compressed == compressed_data assert decompressed == data @@ -44,7 +42,7 @@ def test_decompress_segment_already_decompressed() -> None: data = b"[hello, world!]" compressed_data = zlib.compress(data) - compressed, decompressed = decompress_segment(data, MAX_SIZE) + compressed, decompressed = decompress_segment(data) assert compressed == compressed_data assert decompressed == data @@ -52,13 +50,13 @@ def test_decompress_segment_already_decompressed() -> None: def test_decompress_segment_unexpected_start_character() -> None: """Test handling of invalid data that can't be decompressed""" with pytest.raises(DropSilently): - decompress_segment(b"hello, world!", MAX_SIZE) + decompress_segment(b"hello, world!") def test_decompress_segment_empty_data() -> None: """Test handling of empty data""" with pytest.raises(DropSilently): - decompress_segment(b"", MAX_SIZE) + decompress_segment(b"") def test_decompress_segment_exceeds_max_size() -> None: @@ -68,8 +66,9 @@ def test_decompress_segment_exceeds_max_size() -> None: # A kilobyte of stored input, a megabyte of output on every read. assert len(payload) // len(bomb) > 500 - with pytest.raises(DropSilently): - decompress_segment(bomb, 1024) + with override_options({"replay.consumer.max-segment-decompressed-size": 1024}): + with pytest.raises(DropSilently): + decompress_segment(bomb) def test_decompress_segment_at_max_size() -> None: @@ -77,20 +76,23 @@ def test_decompress_segment_at_max_size() -> None: data = b"[hello, world!]" compressed_data = zlib.compress(data) - compressed, decompressed = decompress_segment(compressed_data, len(data)) + with override_options({"replay.consumer.max-segment-decompressed-size": len(data)}): + compressed, decompressed = decompress_segment(compressed_data) assert compressed == compressed_data assert decompressed == data - with pytest.raises(DropSilently): - decompress_segment(compressed_data, len(data) - 1) + with override_options({"replay.consumer.max-segment-decompressed-size": len(data) - 1}): + with pytest.raises(DropSilently): + decompress_segment(compressed_data) def test_decompress_segment_uncompressed_exceeds_max_size() -> None: """Test an uncompressed segment larger than the limit is dropped""" data = b"[" + b" " * 1024 + b"]" - with pytest.raises(DropSilently): - decompress_segment(data, 512) + with override_options({"replay.consumer.max-segment-decompressed-size": 512}): + with pytest.raises(DropSilently): + decompress_segment(data) @django_db_all @@ -207,7 +209,7 @@ def test_parse_recording_event_success() -> None: "version": 0, } - result = parse_recording_event(msgpack.packb(message), MAX_SIZE) + result = parse_recording_event(msgpack.packb(message)) expected = { "context": { @@ -255,7 +257,7 @@ def test_parse_recording_event_with_replay_event() -> None: "version": 0, } - result = parse_recording_event(msgpack.packb(message), MAX_SIZE) + result = parse_recording_event(msgpack.packb(message)) expected = { "context": { @@ -290,7 +292,7 @@ def test_parse_recording_event_missing_payload() -> None: "version": 0, } with pytest.raises(DropSilently): - parse_recording_event(msgpack.packb(message), MAX_SIZE) + parse_recording_event(msgpack.packb(message)) def test_parse_recording_event_invalid_compression() -> None: @@ -306,7 +308,7 @@ def test_parse_recording_event_invalid_compression() -> None: "version": 0, } with pytest.raises(DropSilently): - parse_recording_event(msgpack.packb(message), MAX_SIZE) + parse_recording_event(msgpack.packb(message)) def test_parse_recording_event_invalid_headers() -> None: @@ -322,7 +324,7 @@ def test_parse_recording_event_invalid_headers() -> None: "version": 0, } with pytest.raises(DropSilently): - parse_recording_event(msgpack.packb(message), MAX_SIZE) + parse_recording_event(msgpack.packb(message)) @django_db_all From 55632f7efac33e44a7ace3ae877954eb97f38191 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Thu, 29 Oct 2026 20:13:38 +0000 Subject: [PATCH 3/4] test(replays): Enable DB for tests that read the segment size option decompress_segment reads replay.consumer.max-segment-decompressed-size from the options store, which falls through to the database on a cache miss. The unit tests calling it without a DB only passed when another test had already warmed the options cache, and failed when run on their own. Refs REPLAY-994 Co-Authored-By: Dominik Buszowiecki <44422760+DominikB2014@users.noreply.github.com> --- tests/sentry/replays/unit/consumers/test_recording.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/tests/sentry/replays/unit/consumers/test_recording.py b/tests/sentry/replays/unit/consumers/test_recording.py index 7e64af9b5dea..b48452c1deae 100644 --- a/tests/sentry/replays/unit/consumers/test_recording.py +++ b/tests/sentry/replays/unit/consumers/test_recording.py @@ -23,6 +23,7 @@ from sentry.utils import json +@django_db_all def test_decompress_segment_success() -> None: """Test successful decompression of segment""" data = b"[hello, world!]" @@ -37,6 +38,7 @@ def test_decompress_segment_success() -> None: assert decompressed == data +@django_db_all def test_decompress_segment_already_decompressed() -> None: """Test handling of already decompressed JSON data""" data = b"[hello, world!]" @@ -47,12 +49,14 @@ def test_decompress_segment_already_decompressed() -> None: assert decompressed == data +@django_db_all def test_decompress_segment_unexpected_start_character() -> None: """Test handling of invalid data that can't be decompressed""" with pytest.raises(DropSilently): decompress_segment(b"hello, world!") +@django_db_all def test_decompress_segment_empty_data() -> None: """Test handling of empty data""" with pytest.raises(DropSilently): @@ -184,6 +188,7 @@ def test_parse_request_message_validation_error() -> None: parse_request_message(msgpack.packb(b"invalid")) +@django_db_all def test_parse_recording_event_success() -> None: """Test successful parsing of recording event end-to-end""" # Create real compressed data @@ -231,6 +236,7 @@ def test_parse_recording_event_success() -> None: assert result == expected +@django_db_all def test_parse_recording_event_with_replay_event() -> None: """Test parsing recording event with replay_event JSON""" # Create real compressed data @@ -295,6 +301,7 @@ def test_parse_recording_event_missing_payload() -> None: parse_recording_event(msgpack.packb(message)) +@django_db_all def test_parse_recording_event_invalid_compression() -> None: """Test that invalid compression in payload causes DropSilently""" message = { From c6616a4e2174d9460d44b72061aaa8cd1c931737 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Tue, 3 Nov 2026 09:37:34 +0000 Subject: [PATCH 4/4] ref(replays): Pass segment size limit into decompress_segment Read replay.consumer.max-segment-decompressed-size in parse_recording_event and pass it to decompress_segment, so the helper no longer depends on the options store and its unit tests no longer need the database. Add regression tests for truncated zlib streams and corrupted checksums, since decompressobj returns partial output on truncation where zlib.decompress raised. Refs REPLAY-994 --- src/sentry/replays/consumers/recording.py | 9 ++-- .../replays/unit/consumers/test_recording.py | 52 ++++++++++++------- 2 files changed, 37 insertions(+), 24 deletions(-) diff --git a/src/sentry/replays/consumers/recording.py b/src/sentry/replays/consumers/recording.py index c6e9f5de7bba..db4f2e75bc91 100644 --- a/src/sentry/replays/consumers/recording.py +++ b/src/sentry/replays/consumers/recording.py @@ -56,7 +56,9 @@ def process_message(message: bytes) -> ProcessedEvent | None: def parse_recording_event(message: bytes) -> Event: recording = parse_request_message(message) segment_id, payload = parse_headers(cast(bytes, recording["payload"]), recording["replay_id"]) - compressed, decompressed = decompress_segment(payload) + compressed, decompressed = decompress_segment( + payload, max_size=options.get("replay.consumer.max-segment-decompressed-size") + ) replay_event_json = recording.get("replay_event") if replay_event_json: @@ -110,14 +112,13 @@ def parse_request_message(message: bytes) -> ReplayRecording: @traces.trace -def decompress_segment(segment: bytes) -> tuple[bytes, bytes]: +def decompress_segment(segment: bytes, max_size: int) -> tuple[bytes, bytes]: """Return the compressed and decompressed forms of a recording segment. - Decompression is bounded by `replay.consumer.max-segment-decompressed-size`. Stored segments are inflated again on every + Decompression is bounded by `max_size` bytes. Stored segments are inflated again on every read, so a segment with an implausible compression ratio is a durable, repeatable cost on the read path -- we drop it here rather than persist it. """ - max_size = options.get("replay.consumer.max-segment-decompressed-size") try: # Ask for one byte past the limit so an oversized segment is detectable without ever # materializing more than `max_size + 1` bytes. diff --git a/tests/sentry/replays/unit/consumers/test_recording.py b/tests/sentry/replays/unit/consumers/test_recording.py index b48452c1deae..20b619910c16 100644 --- a/tests/sentry/replays/unit/consumers/test_recording.py +++ b/tests/sentry/replays/unit/consumers/test_recording.py @@ -22,45 +22,61 @@ from sentry.testutils.pytest.fixtures import django_db_all from sentry.utils import json +MAX_SIZE = 100 * 1024 * 1024 + -@django_db_all def test_decompress_segment_success() -> None: """Test successful decompression of segment""" data = b"[hello, world!]" compressed_data = zlib.compress(data) - compressed, decompressed = decompress_segment(compressed_data) + compressed, decompressed = decompress_segment(compressed_data, MAX_SIZE) assert compressed == compressed_data assert decompressed == data - compressed, decompressed = decompress_segment(data) + compressed, decompressed = decompress_segment(data, MAX_SIZE) assert compressed == compressed_data assert decompressed == data -@django_db_all def test_decompress_segment_already_decompressed() -> None: """Test handling of already decompressed JSON data""" data = b"[hello, world!]" compressed_data = zlib.compress(data) - compressed, decompressed = decompress_segment(data) + compressed, decompressed = decompress_segment(data, MAX_SIZE) assert compressed == compressed_data assert decompressed == data -@django_db_all def test_decompress_segment_unexpected_start_character() -> None: """Test handling of invalid data that can't be decompressed""" with pytest.raises(DropSilently): - decompress_segment(b"hello, world!") + decompress_segment(b"hello, world!", MAX_SIZE) -@django_db_all def test_decompress_segment_empty_data() -> None: """Test handling of empty data""" with pytest.raises(DropSilently): - decompress_segment(b"") + decompress_segment(b"", MAX_SIZE) + + +def test_decompress_segment_truncated_stream() -> None: + """Test a truncated zlib stream is dropped rather than returning partial output""" + compressed_data = zlib.compress(b"[" + b"hello, world!" * 100 + b"]") + + for cut in (1, 4, len(compressed_data) // 2): + with pytest.raises(DropSilently): + decompress_segment(compressed_data[:-cut], MAX_SIZE) + + +def test_decompress_segment_invalid_checksum() -> None: + """Test a zlib stream with a corrupted Adler-32 checksum is dropped""" + compressed_data = bytearray(zlib.compress(b"[hello, world!]")) + compressed_data[-1] ^= 0xFF + + with pytest.raises(DropSilently): + decompress_segment(bytes(compressed_data), MAX_SIZE) def test_decompress_segment_exceeds_max_size() -> None: @@ -70,9 +86,8 @@ def test_decompress_segment_exceeds_max_size() -> None: # A kilobyte of stored input, a megabyte of output on every read. assert len(payload) // len(bomb) > 500 - with override_options({"replay.consumer.max-segment-decompressed-size": 1024}): - with pytest.raises(DropSilently): - decompress_segment(bomb) + with pytest.raises(DropSilently): + decompress_segment(bomb, 1024) def test_decompress_segment_at_max_size() -> None: @@ -80,23 +95,20 @@ def test_decompress_segment_at_max_size() -> None: data = b"[hello, world!]" compressed_data = zlib.compress(data) - with override_options({"replay.consumer.max-segment-decompressed-size": len(data)}): - compressed, decompressed = decompress_segment(compressed_data) + compressed, decompressed = decompress_segment(compressed_data, len(data)) assert compressed == compressed_data assert decompressed == data - with override_options({"replay.consumer.max-segment-decompressed-size": len(data) - 1}): - with pytest.raises(DropSilently): - decompress_segment(compressed_data) + with pytest.raises(DropSilently): + decompress_segment(compressed_data, len(data) - 1) def test_decompress_segment_uncompressed_exceeds_max_size() -> None: """Test an uncompressed segment larger than the limit is dropped""" data = b"[" + b" " * 1024 + b"]" - with override_options({"replay.consumer.max-segment-decompressed-size": 512}): - with pytest.raises(DropSilently): - decompress_segment(data) + with pytest.raises(DropSilently): + decompress_segment(data, 512) @django_db_all