From 81301fea852cc703d1989429f7dc89f13c184152 Mon Sep 17 00:00:00 2001 From: Golumpa Date: Wed, 19 Aug 2026 10:51:12 +0100 Subject: [PATCH] Allow shifted views to omit protocol padding --- .../decoder/structured_protocol.py | 9 ++++++- tests/test_structured_protocol.py | 25 +++++++++++++++++++ 2 files changed, 33 insertions(+), 1 deletion(-) diff --git a/src/nte_history_exporter/decoder/structured_protocol.py b/src/nte_history_exporter/decoder/structured_protocol.py index e93f79e..a94a617 100644 --- a/src/nte_history_exporter/decoder/structured_protocol.py +++ b/src/nte_history_exporter/decoder/structured_protocol.py @@ -12,6 +12,9 @@ MONOPOLY_MARKER = b"FMonopolyLotteryRecordData" FORK_MARKER = b"FForkLotteryRecordData" MAX_ROWS_PER_BLOCK = 100 MAX_STRING_LENGTH = 256 +# Bit-packed UDP views can omit up to the final three protocol-padding bytes. +# The declared size covers that padding, while all declared rows remain intact. +MAX_SHIFTED_BLOCK_PADDING_SHORTFALL = 3 DOTNET_EPOCH_TICKS = 621_355_968_000_000_000 DOTNET_TICKS_PER_SECOND = 10_000_000 MIN_UNIX_SECONDS = 1_500_000_000 @@ -230,7 +233,11 @@ def _parse_block( pos += 12 if row_count > MAX_ROWS_PER_BLOCK: raise StructuredProtocolError(f"row count is too large: {row_count}") - if declared_size > len(data) - pos: + size_shortfall = declared_size - (len(data) - pos) + if size_shortfall > 0 and not ( + view_name.startswith("shift8:") + and size_shortfall <= MAX_SHIFTED_BLOCK_PADDING_SHORTFALL + ): raise StructuredProtocolError("declared block size exceeds payload") reader = _Reader(data, pos) diff --git a/tests/test_structured_protocol.py b/tests/test_structured_protocol.py index 3e1c690..adc1e8b 100644 --- a/tests/test_structured_protocol.py +++ b/tests/test_structured_protocol.py @@ -187,6 +187,31 @@ class StructuredProtocolTests(unittest.TestCase): self.assertEqual(rows[0].item_id, "1003") self.assertEqual(rows[0].protocol_view, "shift8:3") + def test_shifted_parser_allows_missing_declared_protocol_padding(self): + cases = ( + ("monopoly", monopoly_payload("1003,1"), 3, "1003"), + ("fork", fork_payload("fork_dustbin"), 2, "fork_dustbin"), + ) + for record_type, raw_payload, shift, expected_item_id in cases: + with self.subTest(record_type=record_type): + payload_with_declared_padding = bytearray(raw_payload) + marker = MONOPOLY_MARKER if record_type == "monopoly" else FORK_MARKER + declared_size_pos = len(marker) + 1 + 4 + declared_size = int.from_bytes( + payload_with_declared_padding[declared_size_pos : declared_size_pos + 4], + "little", + ) + payload_with_declared_padding[ + declared_size_pos : declared_size_pos + 4 + ] = (declared_size + 3).to_bytes(4, "little") + + packed = bit_pack_after_eight_byte_header( + bytes(payload_with_declared_padding), shift + ) + rows = parse_structured_records(packed, record_type) + + self.assertEqual([row.item_id for row in rows], [expected_item_id]) + def test_protocol_envelope_exposes_stream_and_segment_identity(self): payload = enveloped_monopoly_payload("1003", page_index=2, query_high=False)