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/repeater/handler_helpers/room_server.py b/repeater/handler_helpers/room_server.py index 8139a33..de3b85d 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,27 +50,52 @@ _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 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: @@ -234,13 +262,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() @@ -320,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}", @@ -353,16 +379,26 @@ 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"]) 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 ) @@ -412,34 +448,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/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, 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_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) diff --git a/tests/test_handler_helpers_room_server.py b/tests/test_handler_helpers_room_server.py index 683daca..2754113 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 @@ -5,10 +6,13 @@ 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, + GlobalRateLimiter, RoomServer, + _truncate_utf8, ) @@ -91,7 +95,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 +103,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 +261,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. @@ -273,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() @@ -401,3 +628,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() 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")