goober/nte-exporter
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Allow shifted views to omit protocol padding

Browse source
This commit is contained in:
Golumpa 2026-08-19 10:51:12 +01:00
parent e44c5eecbf
commit 81301fea85
2 changed files with 33 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

9
src/nte_history_exporter/decoder/structured_protocol.py
View file

@ -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)

25
tests/test_structured_protocol.py
View file

@ -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)