From 1a4df1a151e0b6d1f7be7faf196fcc0c56f1de9b Mon Sep 17 00:00:00 2001 From: agessaman Date: Thu, 16 Jul 2026 14:28:55 -0700 Subject: [PATCH 1/6] fix(room_server): enforce firmware post text budget in bytes The room post limit was 160 Python characters; firmware caps post text at MAX_POST_TEXT_LEN = 160-9 = 151 bytes (the 160-byte encrypted text budget minus the timestamp/flags/author-prefix header). Store posts truncated to 151 UTF-8 bytes at a codepoint boundary, and clamp again when pushing so previously stored oversized posts cannot produce oversized frames. --- repeater/handler_helpers/room_server.py | 36 +++++-- tests/test_handler_helpers_room_server.py | 110 +++++++++++++++++++++- 2 files changed, 139 insertions(+), 7 deletions(-) diff --git a/repeater/handler_helpers/room_server.py b/repeater/handler_helpers/room_server.py index 8139a33..3680dbc 100644 --- a/repeater/handler_helpers/room_server.py +++ b/repeater/handler_helpers/room_server.py @@ -27,7 +27,10 @@ PUSH_TIMEOUT_BASE_MS = 4000 PUSH_ACK_TIMEOUT_FACTOR_MS = 2000 # Safety limits and protections -MAX_MESSAGE_LENGTH = 160 # Match C++ MAX_POST_TEXT_LEN (151 bytes for text) +# Match C++ MAX_POST_TEXT_LEN from examples/simple_room_server/MyMesh.h: +# #define MAX_POST_TEXT_LEN (160-9) -- 160-byte encrypted text budget minus the +# 9-byte prefix (4-byte timestamp + 1-byte flags/attempt + 4-byte author pubkey prefix). +MAX_POST_TEXT_LEN = 151 MAX_POSTS_PER_CLIENT_PER_MINUTE = 10 # Prevent spam MAX_CLIENTS_PER_ROOM = 50 # From ACL default MAX_PUSH_FAILURES = 3 # Evict after this many consecutive failures @@ -47,6 +50,19 @@ _global_push_lock = asyncio.Lock() GLOBAL_MIN_GAP_BETWEEN_MESSAGES = 1.1 # 1.1s minimum gap between transmissions +def _truncate_utf8(text: str, max_bytes: int) -> str: + """Truncate ``text`` to at most ``max_bytes`` of its UTF-8 encoding. + + Cuts at a codepoint boundary so a partial multi-byte sequence is never + emitted (matches firmware's byte-length text budget, but stays + UTF-8-safe rather than the firmware's raw ``strncpy``). + """ + encoded = text.encode("utf-8") + if len(encoded) <= max_bytes: + return text + return encoded[:max_bytes].decode("utf-8", errors="ignore") + + class GlobalRateLimiter: def __init__(self, min_gap_seconds: float = 0.1): self.min_gap = min_gap_seconds # Minimum gap between consecutive messages @@ -234,13 +250,15 @@ class RoomServer: ) -> bool: try: - # SAFETY: Validate message length - if len(message_text) > MAX_MESSAGE_LENGTH: + # SAFETY: Validate message length (byte length, matching firmware's + # MAX_POST_TEXT_LEN text budget), cutting at a UTF-8 codepoint boundary. + encoded_len = len(message_text.encode("utf-8")) + if encoded_len > MAX_POST_TEXT_LEN: logger.warning( f"Room '{self.room_name}': Message from {client_pubkey[:4].hex()} " - f"exceeds max length ({len(message_text)} > {MAX_MESSAGE_LENGTH}), truncating" + f"exceeds max length ({encoded_len} > {MAX_POST_TEXT_LEN} bytes), truncating" ) - message_text = message_text[:MAX_MESSAGE_LENGTH] + message_text = _truncate_utf8(message_text, MAX_POST_TEXT_LEN) # SAFETY: Rate limit per client client_key = client_pubkey.hex() @@ -362,7 +380,13 @@ class RoomServer: author_prefix = author_pubkey[:4] # Plaintext: timestamp(4) + flags(1) + author_prefix(4) + text - message_bytes = post["message_text"].encode("utf-8") + # SAFETY: Clamp at push time too, in case a legacy stored post + # (from before this limit was enforced on write) exceeds + # MAX_POST_TEXT_LEN bytes -- truncate at a UTF-8 codepoint boundary. + message_text = post["message_text"] + if len(message_text.encode("utf-8")) > MAX_POST_TEXT_LEN: + message_text = _truncate_utf8(message_text, MAX_POST_TEXT_LEN) + message_bytes = message_text.encode("utf-8") plaintext = ( timestamp.to_bytes(4, "little") + bytes([flags]) + author_prefix + message_bytes ) diff --git a/tests/test_handler_helpers_room_server.py b/tests/test_handler_helpers_room_server.py index 683daca..e1ee422 100644 --- a/tests/test_handler_helpers_room_server.py +++ b/tests/test_handler_helpers_room_server.py @@ -5,10 +5,12 @@ from unittest.mock import AsyncMock, MagicMock, patch import pytest from repeater.handler_helpers.room_server import ( + MAX_POST_TEXT_LEN, MAX_UNSYNCED_POSTS, TXT_TYPE_PLAIN, TXT_TYPE_SIGNED_PLAIN, RoomServer, + _truncate_utf8, ) @@ -91,7 +93,7 @@ async def test_room_server_add_post_truncates_and_rate_limits_client(): first_ok = await rs.add_post(client_key, long_msg, sender_timestamp=5) assert first_ok is True args = db.insert_room_message.call_args.kwargs - assert len(args["message_text"]) == 160 + assert len(args["message_text"].encode("utf-8")) == MAX_POST_TEXT_LEN # Force client to appear at post-per-minute limit. rs.client_post_times[client_key.hex()] = [time.time() - 1] * 10 @@ -99,6 +101,71 @@ async def test_room_server_add_post_truncates_and_rate_limits_client(): assert second_ok is False +@pytest.mark.asyncio +async def test_room_server_add_post_ascii_over_limit_stores_exactly_max_bytes(): + db = _FakeDB() + rs = _make_room_server(db=db) + + long_msg = "a" * 200 # ASCII: 1 byte per char, well over MAX_POST_TEXT_LEN + ok = await rs.add_post(b"G" * 32, long_msg, sender_timestamp=1) + assert ok is True + + stored = db.insert_room_message.call_args.kwargs["message_text"] + assert len(stored.encode("utf-8")) == MAX_POST_TEXT_LEN + assert stored == "a" * MAX_POST_TEXT_LEN + + +@pytest.mark.asyncio +async def test_room_server_add_post_multibyte_utf8_truncates_on_codepoint_boundary(): + db = _FakeDB() + rs = _make_room_server(db=db) + + # Each emoji is 4 bytes in UTF-8; padding forces the cut to land mid-emoji + # if truncation were byte-naive instead of codepoint-aware. + padding = "a" * (MAX_POST_TEXT_LEN - 2) + msg = padding + "\U0001f600\U0001f600\U0001f600" # grinning face emoji x3 + assert len(msg.encode("utf-8")) > MAX_POST_TEXT_LEN + + ok = await rs.add_post(b"H" * 32, msg, sender_timestamp=2) + assert ok is True + + stored = db.insert_room_message.call_args.kwargs["message_text"] + encoded = stored.encode("utf-8") + assert len(encoded) <= MAX_POST_TEXT_LEN + # Must decode cleanly (no partial multi-byte sequence) and round-trip. + assert encoded.decode("utf-8") == stored + + +@pytest.mark.asyncio +async def test_room_server_add_post_exact_limit_text_untouched(): + db = _FakeDB() + rs = _make_room_server(db=db) + + msg = "y" * MAX_POST_TEXT_LEN + ok = await rs.add_post(b"I" * 32, msg, sender_timestamp=3) + assert ok is True + + stored = db.insert_room_message.call_args.kwargs["message_text"] + assert stored == msg + assert len(stored.encode("utf-8")) == MAX_POST_TEXT_LEN + + +def test_truncate_utf8_helper_boundary_cases(): + # Under the limit: untouched. + assert _truncate_utf8("short", 151) == "short" + + # Exactly at the limit: untouched. + exact = "z" * 151 + assert _truncate_utf8(exact, 151) == exact + + # Multi-byte straddling the boundary: cuts cleanly, decodes, stays <= limit. + text = ("b" * 149) + "\U0001f600\U0001f600" # 149 + 4 + 4 = 157 bytes + result = _truncate_utf8(text, 151) + encoded = result.encode("utf-8") + assert len(encoded) <= 151 + assert encoded.decode("utf-8") == result + + @pytest.mark.asyncio async def test_room_server_add_post_returns_false_on_db_insert_failure(): db = _FakeDB() @@ -192,6 +259,47 @@ async def test_room_server_push_post_to_client_success_direct_route_sets_path_an rs.global_limiter.release.assert_called_once() +@pytest.mark.asyncio +async def test_room_server_push_post_to_client_clamps_oversized_legacy_stored_text(): + """A post stored before the write-time limit was enforced (or otherwise + over MAX_POST_TEXT_LEN bytes) must still be clamped at push time so the + outgoing frame's text portion never exceeds the firmware's budget.""" + db = _FakeDB() + db.get_client_sync.return_value = {"push_failures": 0} + injector = AsyncMock(return_value=True) + rs = _make_room_server(db=db, injector=injector) + rs.global_limiter = SimpleNamespace(acquire=AsyncMock(), release=MagicMock()) + rs._handle_ack_received = AsyncMock() + + client = _FakeClient(pubkey=b"E" * 32, out_path=b"\xaa\xbb", out_path_len=2) + oversized_text = "q" * (MAX_POST_TEXT_LEN + 50) + post = { + "author_pubkey": (b"F" * 32).hex(), + "message_text": oversized_text, + "post_timestamp": 1234.5, + } + + packet = SimpleNamespace(path=bytearray(), path_len=0) + with ( + patch( + "repeater.handler_helpers.room_server.CryptoUtils.sha256", + return_value=b"\x01\x02\x03\x04abcd", + ), + patch( + "repeater.handler_helpers.room_server.PacketBuilder.create_datagram", + return_value=packet, + ) as create_datagram, + ): + ok = await rs.push_post_to_client(client, post) + + assert ok is True + plaintext = create_datagram.call_args.kwargs["plaintext"] + # Prefix is timestamp(4) + flags(1) + author_prefix(4) = 9 bytes. + text_portion = plaintext[9:] + assert len(text_portion) == MAX_POST_TEXT_LEN + assert text_portion.decode("utf-8") == "q" * MAX_POST_TEXT_LEN + + @pytest.mark.asyncio async def test_room_server_push_expected_ack_matches_firmware_signed_ack(): """The pending ACK CRC must match what a signed-plain receiver sends back. From 5a1822e6421cfdf2fa4c039ee5d379d10e3bb936 Mon Sep 17 00:00:00 2001 From: agessaman Date: Thu, 16 Jul 2026 20:28:04 -0700 Subject: [PATCH 2/6] test(engine): stop building reserved-version packets in fixtures Firmware's Dispatcher::tryParsePacket drops any payload version above 0 at parse, and core's Packet.read_from now enforces the same rule, so a version-1 routed ACK can never reach the relay; the wire vector exercising that unreachable input is removed (header regeneration is still covered by the remaining vectors, including transport-direct). The advert fixture also OR'd the version-1 bit into its header; it now builds the version-0 header that PacketBuilder.create_advert emits. --- tests/test_engine.py | 13 ++++--------- tests/test_path_hash_mode_advert.py | 4 ++-- 2 files changed, 6 insertions(+), 11 deletions(-) diff --git a/tests/test_engine.py b/tests/test_engine.py index 0ea4315..f334a0f 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -692,8 +692,10 @@ class TestForwardRoutedAck: # Literal MeshCore-format routed-ACK frames (Packet::writeTo layout). MeshCore's # routeDirectRecvAcks re-emits through createAck: the header is rebuilt from -# scratch (payload type ACK, route DIRECT, version bits cleared, transport codes -# dropped) and this node is removed from the path. Independent wire fixtures — +# scratch (payload type ACK, route DIRECT, transport codes dropped) and this +# node is removed from the path. Non-zero-version inputs are not represented +# here: firmware's Dispatcher::tryParsePacket (and our Packet.read_from) drops +# them at parse, so they can never reach the relay. Independent wire fixtures — # do NOT rebuild them through Packet.write_to()/PathUtils. FIRMWARE_ROUTED_ACK_VECTORS = ( # header 0x0E = ACK<<2 | ROUTE_DIRECT; path_len 0x02 = 1-byte/2-hop; @@ -719,13 +721,6 @@ FIRMWARE_ROUTED_ACK_VECTORS = ( bytes.fromhex("0E01CC4DABAF95"), id="transport-direct-1-byte-2-hop", ), - # header 0x4E = version bit set on a direct ACK; createAck clears it. - pytest.param( - bytes([0xAB]), - bytes.fromhex("4E02ABCC4DABAF95"), - bytes.fromhex("0E01CC4DABAF95"), - id="direct-version-bits-cleared", - ), ) diff --git a/tests/test_path_hash_mode_advert.py b/tests/test_path_hash_mode_advert.py index 9b68487..65007a0 100644 --- a/tests/test_path_hash_mode_advert.py +++ b/tests/test_path_hash_mode_advert.py @@ -34,8 +34,8 @@ class MockRadio: def _make_advert_packet(): """Build a 0-hop flood ADVERT packet (same shape as PacketBuilder.create_advert).""" pkt = Packet() - # Version 1, ROUTE_TYPE_FLOOD, PAYLOAD_TYPE_ADVERT - pkt.header = (1 << 6) | (PAYLOAD_TYPE_ADVERT << PH_TYPE_SHIFT) | ROUTE_TYPE_FLOOD + # Version 0 (the only version firmware accepts), ROUTE_TYPE_FLOOD, PAYLOAD_TYPE_ADVERT + pkt.header = (PAYLOAD_TYPE_ADVERT << PH_TYPE_SHIFT) | ROUTE_TYPE_FLOOD pkt.path_len = 0 pkt.path = bytearray() pkt.payload = bytearray(b"minimal_advert_payload") From ef234f143b35f1167f3a5d3b48f438a2ba42c53e Mon Sep 17 00:00:00 2001 From: agessaman Date: Thu, 16 Jul 2026 20:28:13 -0700 Subject: [PATCH 3/6] fix(room_server): hold the global transmit lock across the whole send GlobalRateLimiter.acquire() enforced the inter-message gap inside an async-with that released the lock before returning, so the caller's radio transmission ran with no lock held and every room's sync loop could push concurrently. acquire() now returns with the lock held (released via release() in a finally, including on cancellation during the gap wait), uses a monotonic clock so wall-clock jumps cannot skip or stretch the gap, and push_post_to_client acquires it only around the sync-state write plus the blocking send, so the ACK-timeout clock starts when the transmission actually begins. Regression tests cover two rooms sharing one limiter (no concurrent sends, gap enforced) and lock release when the send raises. --- repeater/handler_helpers/room_server.py | 89 ++++++++++++------- tests/test_handler_helpers_room_server.py | 103 ++++++++++++++++++++++ 2 files changed, 158 insertions(+), 34 deletions(-) diff --git a/repeater/handler_helpers/room_server.py b/repeater/handler_helpers/room_server.py index 3680dbc..eda2e66 100644 --- a/repeater/handler_helpers/room_server.py +++ b/repeater/handler_helpers/room_server.py @@ -67,23 +67,35 @@ class GlobalRateLimiter: def __init__(self, min_gap_seconds: float = 0.1): self.min_gap = min_gap_seconds # Minimum gap between consecutive messages self.lock = asyncio.Lock() # Only one transmission at a time - self.last_release_time = 0 + self.last_release_time = 0.0 async def acquire(self): + """Acquire the global transmit lock and enforce the inter-message gap. - async with self.lock: - # Enforce minimum gap between consecutive transmissions - now = time.time() - time_since_last = now - self.last_release_time + The lock is **held on return** and must be released with ``release()`` + (call it in a ``finally``). The previous implementation released the + lock inside an ``async with`` before returning, so the caller's radio + transmission ran with no lock held and multiple room loops could push + concurrently. A monotonic clock is used so a wall-clock jump cannot + skip or extend the gap. + """ + await self.lock.acquire() + try: + time_since_last = time.monotonic() - self.last_release_time if time_since_last < self.min_gap: wait_time = self.min_gap - time_since_last logger.debug(f"Global rate limiter: waiting {wait_time * 1000:.0f}ms") await asyncio.sleep(wait_time) - # Lock is now held - caller can transmit - # Will be released when context exits + except BaseException: + # Never leak the lock if the gap wait is cancelled. + self.lock.release() + raise def release(self): - self.last_release_time = time.time() + """Record the transmission time and release the transmit lock.""" + self.last_release_time = time.monotonic() + if self.lock.locked(): + self.lock.release() class RoomServer: @@ -338,10 +350,6 @@ class RoomServer: async def push_post_to_client(self, client_info, post: Dict) -> bool: try: - # SAFETY: Global transmission lock - only ONE message on radio at a time - # This is critical because LoRa is serial (0.5-9s airtime per message) - await self.global_limiter.acquire() - # SAFETY: Check client failure backoff sync_state = self.db.get_client_sync( room_hash=f"0x{self.room_hash:02X}", @@ -436,34 +444,47 @@ class RoomServer: PUSH_TIMEOUT_BASE_MS + PUSH_ACK_TIMEOUT_FACTOR_MS * (path_len + 1) ) / 1000.0 - # Update client sync state with pending ACK current_sync_since = ( sync_state.get("sync_since", 0) if sync_state else getattr(client_info, "sync_since", 0) ) - self.db.upsert_client_sync( - room_hash=f"0x{self.room_hash:02X}", - client_pubkey=client_info.id.get_public_key().hex(), - sync_since=current_sync_since, - pending_ack_crc=expected_ack_crc, - push_post_timestamp=post["post_timestamp"], - ack_timeout_time=time.time() + ack_timeout, - last_activity=time.time(), - ) - # Send and wait for the client's delivery ACK. The injector must be - # told the crypto ACK CRC we computed above — its default - # (packet.get_crc()) is a packet-hash CRC no client ever sends. - # This blocks for the entire transmission duration (0.5-9 seconds) - success = await self.packet_injector( - packet, - wait_for_ack=True, - expected_crc=expected_ack_crc, - ack_timeout_s=ack_timeout, - ) - # SAFETY: Release transmission lock AFTER send completes - self.global_limiter.release() + # SAFETY: Global transmission lock - only ONE message on the radio at + # a time, enforced across the entire send. LoRa is serial (0.5-9s + # airtime per message), and every room has its own sync loop sharing + # this limiter, so the lock must be held from before we start the ACK + # timeout clock through the blocking send. Released in `finally` so + # backoff/exception paths above (which returned before acquiring) do + # not, and the send path always does, free it exactly once. + await self.global_limiter.acquire() + try: + # Update client sync state with pending ACK. Done under the lock + # so the ACK-timeout clock starts when the send actually begins, + # not before the inter-message gap has elapsed. + self.db.upsert_client_sync( + room_hash=f"0x{self.room_hash:02X}", + client_pubkey=client_info.id.get_public_key().hex(), + sync_since=current_sync_since, + pending_ack_crc=expected_ack_crc, + push_post_timestamp=post["post_timestamp"], + ack_timeout_time=time.time() + ack_timeout, + last_activity=time.time(), + ) + # Send and wait for the client's delivery ACK. The injector must + # be told the crypto ACK CRC we computed above — its default + # (packet.get_crc()) is a packet-hash CRC no client ever sends. + # This blocks for the entire transmission duration (0.5-9 seconds) + success = await self.packet_injector( + packet, + wait_for_ack=True, + expected_crc=expected_ack_crc, + ack_timeout_s=ack_timeout, + ) + finally: + # SAFETY: Release the transmission lock on every path (ACK, + # timeout, or exception during the send). + self.global_limiter.release() if success: # ACK received! Update sync state diff --git a/tests/test_handler_helpers_room_server.py b/tests/test_handler_helpers_room_server.py index e1ee422..7908783 100644 --- a/tests/test_handler_helpers_room_server.py +++ b/tests/test_handler_helpers_room_server.py @@ -1,3 +1,4 @@ +import asyncio import time from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch @@ -9,6 +10,7 @@ from repeater.handler_helpers.room_server import ( MAX_UNSYNCED_POSTS, TXT_TYPE_PLAIN, TXT_TYPE_SIGNED_PLAIN, + GlobalRateLimiter, RoomServer, _truncate_utf8, ) @@ -509,3 +511,104 @@ async def test_room_server_start_and_stop_are_idempotent(): # Ensure task is cleaned up. if first_task: assert first_task.cancelled() or first_task.done() + + +def _push_post(author=b"F" * 32): + return { + "author_pubkey": author.hex(), + "message_text": "payload", + "post_timestamp": 1234.5, + } + + +@pytest.mark.asyncio +async def test_global_rate_limiter_serializes_concurrent_room_pushes(): + """Two room loops sharing the limiter must not transmit concurrently, and + the inter-message gap must be enforced between their sends. + + Regression: acquire() used to release the lock before returning, so + both pushes ran their radio send with no lock held. + """ + limiter = GlobalRateLimiter(min_gap_seconds=0.05) + + active = 0 + max_active = 0 + starts = [] + + async def injector(packet, wait_for_ack=True, expected_crc=0, ack_timeout_s=0.0): + nonlocal active, max_active + starts.append(time.monotonic()) + active += 1 + max_active = max(max_active, active) + try: + await asyncio.sleep(0.02) # simulate airtime + finally: + active -= 1 + return True + + rooms = [] + for i in range(2): + db = _FakeDB() + db.get_client_sync.return_value = {"push_failures": 0} + rs = _make_room_server(db=db, injector=injector) + rs.global_limiter = limiter # share one real limiter across both rooms + rs._handle_ack_received = AsyncMock() + rooms.append(rs) + + client = _FakeClient(pubkey=b"E" * 32) + packet = SimpleNamespace(path=bytearray(), path_len=0) + with ( + patch( + "repeater.handler_helpers.room_server.CryptoUtils.sha256", + return_value=b"\x01\x02\x03\x04abcd", + ), + patch( + "repeater.handler_helpers.room_server.PacketBuilder.create_datagram", + return_value=packet, + ), + ): + results = await asyncio.gather( + *(rs.push_post_to_client(client, _push_post()) for rs in rooms) + ) + + assert results == [True, True] + assert max_active == 1 # never two transmissions in flight at once + assert len(starts) == 2 + # The second send started at least min_gap after the first began. + assert starts[1] - starts[0] >= 0.05 + assert not limiter.lock.locked() # fully released at the end + + +@pytest.mark.asyncio +async def test_global_rate_limiter_releases_lock_on_send_exception(): + """If the radio send raises, the limiter lock must still be released so the + next push is not deadlocked.""" + limiter = GlobalRateLimiter(min_gap_seconds=0.0) + + async def failing_injector(packet, **kwargs): + raise RuntimeError("radio boom") + + db = _FakeDB() + db.get_client_sync.return_value = {"push_failures": 0} + rs = _make_room_server(db=db, injector=failing_injector) + rs.global_limiter = limiter + + client = _FakeClient(pubkey=b"E" * 32) + packet = SimpleNamespace(path=bytearray(), path_len=0) + with ( + patch( + "repeater.handler_helpers.room_server.CryptoUtils.sha256", + return_value=b"\x01\x02\x03\x04abcd", + ), + patch( + "repeater.handler_helpers.room_server.PacketBuilder.create_datagram", + return_value=packet, + ), + ): + ok = await rs.push_post_to_client(client, _push_post()) + + assert ok is False + assert not limiter.lock.locked() + # The lock is reusable immediately afterwards. + await limiter.acquire() + limiter.release() From 377b054ab4682708f50e624d7fdbd517d5db3eb5 Mon Sep 17 00:00:00 2001 From: agessaman Date: Thu, 16 Jul 2026 20:37:30 -0700 Subject: [PATCH 4/6] fix(room_server): push the stored post timestamp with a random attempt nonce The pushed frame serialized the delivery wall-clock time instead of the post's own stored timestamp, and left the flag byte's low 2 attempt bits always zero. Firmware sends the stored post timestamp and ORs in a fresh random attempt so a retried post produces a different packet hash and ACK; with a fixed timestamp two retries within the same second would be byte-identical and dropped as mesh duplicates. The serialized timestamp now also equals the push_post_timestamp sync watermark, matching the firmware invariant. --- repeater/handler_helpers/room_server.py | 10 +- tests/test_handler_helpers_room_server.py | 117 ++++++++++++++++++++++ 2 files changed, 124 insertions(+), 3 deletions(-) diff --git a/repeater/handler_helpers/room_server.py b/repeater/handler_helpers/room_server.py index eda2e66..de3b85d 100644 --- a/repeater/handler_helpers/room_server.py +++ b/repeater/handler_helpers/room_server.py @@ -379,9 +379,13 @@ class RoomServer: ) return False # Skip this client for now - # Build message payload - timestamp = int(time.time()) - flags = TXT_TYPE_SIGNED_PLAIN << 2 # Include author prefix + # Build message payload. + # Timestamp is the post's own stored timestamp, not the delivery + # time (MyMesh.cpp:56 serializes the stored post's timestamp). + timestamp = int(post["post_timestamp"]) + # Low 2 bits carry a random attempt nonce so retries of the same + # post get a different packet hash/ACK (MyMesh.cpp:59-60). + flags = (TXT_TYPE_SIGNED_PLAIN << 2) | (secrets.randbits(8) & 0x03) # Author prefix (first 4 bytes of pubkey) author_pubkey = bytes.fromhex(post["author_pubkey"]) diff --git a/tests/test_handler_helpers_room_server.py b/tests/test_handler_helpers_room_server.py index 7908783..2754113 100644 --- a/tests/test_handler_helpers_room_server.py +++ b/tests/test_handler_helpers_room_server.py @@ -383,6 +383,123 @@ async def test_room_server_push_expected_ack_matches_firmware_signed_ack(): assert client_ack_crc == expected_ack_crc +@pytest.mark.asyncio +async def test_push_serializes_stored_post_timestamp_not_walltime(): + """The pushed frame's timestamp must be the post's own stored timestamp, + not the delivery wall-clock time (MyMesh.cpp:56 serializes the stored + post's timestamp, not `now`).""" + db = _FakeDB() + db.get_client_sync.return_value = {"push_failures": 0} + injector = AsyncMock(return_value=True) + rs = _make_room_server(db=db, injector=injector) + rs.global_limiter = SimpleNamespace(acquire=AsyncMock(), release=MagicMock()) + rs._handle_ack_received = AsyncMock() + + client = _FakeClient(pubkey=b"E" * 32) + post = { + "author_pubkey": (b"F" * 32).hex(), + "message_text": "payload", + "post_timestamp": 1000.0, + } + + packet = SimpleNamespace(path=bytearray(), path_len=0) + with ( + patch( + "repeater.handler_helpers.room_server.CryptoUtils.sha256", + return_value=b"\x01\x02\x03\x04abcd", + ), + patch( + "repeater.handler_helpers.room_server.PacketBuilder.create_datagram", + return_value=packet, + ) as create_datagram, + ): + ok = await rs.push_post_to_client(client, post) + + assert ok is True + plaintext = create_datagram.call_args.kwargs["plaintext"] + assert int.from_bytes(plaintext[0:4], "little") == 1000 + + +@pytest.mark.asyncio +async def test_push_attempt_bits_randomized_across_retries(): + """Each push must OR a fresh random byte's low 2 bits into flags so a + retried post gets a different packet hash (and ACK), per MyMesh.cpp:59-60 + ("so packet hash (and ACK) will be different").""" + db = _FakeDB() + db.get_client_sync.return_value = {"push_failures": 0} + injector = AsyncMock(return_value=True) + rs = _make_room_server(db=db, injector=injector) + rs.global_limiter = SimpleNamespace(acquire=AsyncMock(), release=MagicMock()) + rs._handle_ack_received = AsyncMock() + + client = _FakeClient(pubkey=b"E" * 32) + post = { + "author_pubkey": (b"F" * 32).hex(), + "message_text": "payload", + "post_timestamp": 1234.5, + } + + plaintexts = [] + acks = [] + with patch("repeater.handler_helpers.room_server.secrets.randbits", side_effect=[1, 2]): + for _ in range(2): + packet = SimpleNamespace(path=bytearray(), path_len=0) + with patch( + "repeater.handler_helpers.room_server.PacketBuilder.create_datagram", + return_value=packet, + ) as create_datagram: + ok = await rs.push_post_to_client(client, post) + assert ok is True + plaintexts.append(create_datagram.call_args.kwargs["plaintext"]) + acks.append(db.upsert_client_sync.call_args.kwargs["pending_ack_crc"]) + + flags = [pt[4] for pt in plaintexts] + assert (flags[0] >> 2) & 0x3F == TXT_TYPE_SIGNED_PLAIN + assert (flags[1] >> 2) & 0x3F == TXT_TYPE_SIGNED_PLAIN + assert flags[0] & 0x03 != flags[1] & 0x03 + assert plaintexts[0][0:4] == plaintexts[1][0:4] # timestamp unchanged + assert plaintexts[0][5:] == plaintexts[1][5:] # author prefix + text unchanged + assert acks[0] != acks[1] + + +@pytest.mark.asyncio +async def test_push_watermark_equals_serialized_timestamp(): + """The db sync watermark update must use the same timestamp value that + was serialized into the pushed plaintext.""" + db = _FakeDB() + db.get_client_sync.return_value = {"push_failures": 0} + injector = AsyncMock(return_value=True) + rs = _make_room_server(db=db, injector=injector) + rs.global_limiter = SimpleNamespace(acquire=AsyncMock(), release=MagicMock()) + rs._handle_ack_received = AsyncMock() + + client = _FakeClient(pubkey=b"E" * 32) + post = { + "author_pubkey": (b"F" * 32).hex(), + "message_text": "payload", + "post_timestamp": 555.0, + } + + packet = SimpleNamespace(path=bytearray(), path_len=0) + with ( + patch( + "repeater.handler_helpers.room_server.CryptoUtils.sha256", + return_value=b"\x01\x02\x03\x04abcd", + ), + patch( + "repeater.handler_helpers.room_server.PacketBuilder.create_datagram", + return_value=packet, + ) as create_datagram, + ): + ok = await rs.push_post_to_client(client, post) + + assert ok is True + plaintext = create_datagram.call_args.kwargs["plaintext"] + serialized_ts = int.from_bytes(plaintext[0:4], "little") + watermark = db.upsert_client_sync.call_args.kwargs["push_post_timestamp"] + assert int(watermark) == serialized_ts + + @pytest.mark.asyncio async def test_room_server_push_post_to_client_backoff_skip_and_timeout_path(): db = _FakeDB() From 1687c299d10af5cc9c4ae824bbf37048d96f14a6 Mon Sep 17 00:00:00 2001 From: agessaman Date: Thu, 16 Jul 2026 22:39:15 -0700 Subject: [PATCH 5/6] refactor(daemon): drop the neutralized own-packet stub --- repeater/main.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/repeater/main.py b/repeater/main.py index 4739095..e48e894 100644 --- a/repeater/main.py +++ b/repeater/main.py @@ -377,8 +377,6 @@ class RepeaterDaemon: # Load additional identities from config (e.g., room servers) await self._load_additional_identities() - self.dispatcher._is_own_packet = lambda pkt: False - self.repeater_handler = RepeaterHandler( self.config, self.dispatcher, From 90eb3cae07d809ab4e4fc33ee2699e4c43083e40 Mon Sep 17 00:00:00 2001 From: agessaman Date: Thu, 16 Jul 2026 22:42:34 -0700 Subject: [PATCH 6/6] fix(acl): keep the session replay watermark monotonic --- repeater/handler_helpers/acl.py | 5 ++++- tests/test_handler_helpers_acl_advert.py | 23 +++++++++++++++++++++++ 2 files changed, 27 insertions(+), 1 deletion(-) diff --git a/repeater/handler_helpers/acl.py b/repeater/handler_helpers/acl.py index 6d38bce..4fb1437 100644 --- a/repeater/handler_helpers/acl.py +++ b/repeater/handler_helpers/acl.py @@ -64,7 +64,10 @@ class ACL: sync_since: int = None, ) -> None: now = int(time.time()) - client.last_timestamp = timestamp + # Monotonic: the replay watermark must never move backwards, even if + # another accepted request advanced it between the replay check and + # this write. + client.last_timestamp = max(client.last_timestamp, timestamp) client.last_activity = now client.last_login_success = now client.shared_secret = shared_secret diff --git a/tests/test_handler_helpers_acl_advert.py b/tests/test_handler_helpers_acl_advert.py index 6af33a3..0c26932 100644 --- a/tests/test_handler_helpers_acl_advert.py +++ b/tests/test_handler_helpers_acl_advert.py @@ -142,6 +142,29 @@ def test_acl_admin_login_sets_client_state_and_replay_protection(): assert replay_perms == 0 +def test_acl_touch_client_session_watermark_is_monotonic(): + """The replay watermark never moves backwards: if another accepted request + advanced it between the replay check and the session touch, an older (but + already-validated) timestamp must not regress it.""" + identity = _FakeIdentity(b"B" * 32) + acl = ACL(max_clients=5, admin_password="top-secret", guest_password="guest") + + ok, _ = acl.authenticate_client( + client_identity=identity, + shared_secret=b"k" * 32, + password="top-secret", + timestamp=2000, + ) + assert ok is True + client = acl.get_client(b"B" * 40) + + acl._touch_client_session(client, b"k" * 32, timestamp=1500) + assert client.last_timestamp == 2000 + + acl._touch_client_session(client, b"k" * 32, timestamp=2500) + assert client.last_timestamp == 2500 + + def test_acl_max_clients_invalid_password_and_remove_client_paths(): acl = ACL(max_clients=1, admin_password="a", guest_password="g") id_a = _FakeIdentity(b"C" * 32)