From f45b84b15cfcf9a187fc0bbb0bc8a3a17634d636 Mon Sep 17 00:00:00 2001 From: Jack Kingsman Date: Wed, 8 Jul 2026 15:16:21 -0700 Subject: [PATCH] Add more aggressive packet validity checking. Closes #315. --- app/decoder.py | 52 ++++++++++++++++++++++++ app/packet_processor.py | 15 +++++++ tests/test_decoder.py | 60 ++++++++++++++++++++++++++++ tests/test_packet_pipeline.py | 74 +++++++++++++++++++++++++++++++---- 4 files changed, 194 insertions(+), 7 deletions(-) diff --git a/app/decoder.py b/app/decoder.py index 06c028a..705a360 100644 --- a/app/decoder.py +++ b/app/decoder.py @@ -11,6 +11,8 @@ from enum import IntEnum import nacl.bindings from Crypto.Cipher import AES +from nacl.exceptions import BadSignatureError +from nacl.signing import VerifyKey logger = logging.getLogger(__name__) @@ -377,6 +379,56 @@ def parse_advertisement( ) +# Advertisement layout constants, mirroring firmware MeshCore.h / Mesh.cpp. +# pub_key(32) || timestamp(4) || signature(64) || app_data(<= MAX_ADVERT_DATA_SIZE) +_ADVERT_PUB_KEY_SIZE = 32 +_ADVERT_TIMESTAMP_SIZE = 4 +_ADVERT_SIGNATURE_SIZE = 64 +_ADVERT_MAX_APP_DATA_SIZE = 32 # firmware MAX_ADVERT_DATA_SIZE +_ADVERT_HEADER_SIZE = _ADVERT_PUB_KEY_SIZE + _ADVERT_TIMESTAMP_SIZE + _ADVERT_SIGNATURE_SIZE # 100 + + +def verify_advert_signature(payload: bytes) -> bool: + """Verify a MeshCore advertisement's Ed25519 signature. + + Mirrors the firmware check in ``Mesh.cpp`` ``onRecvPacket()`` for + ``PAYLOAD_TYPE_ADVERT``: the signed message is + ``pub_key(32) || timestamp(4) || app_data``, where ``app_data`` is + everything following the 64-byte signature, clamped to + ``MAX_ADVERT_DATA_SIZE`` (32) bytes. The signature is verified against the + public key embedded in the advert itself, so a bit-flip in the key, the + signed body, or the signature all cause rejection. + + ``payload`` must be the advert payload starting at the public key (i.e. + ``PacketInfo.payload`` for an ADVERT packet). Returns ``True`` only when the + signature is valid; any malformed input or verification failure returns + ``False`` so callers fail closed and never ingest a forged advert. + """ + if len(payload) < _ADVERT_HEADER_SIZE: + return False + + public_key = payload[0:_ADVERT_PUB_KEY_SIZE] + signature = payload[_ADVERT_PUB_KEY_SIZE + _ADVERT_TIMESTAMP_SIZE : _ADVERT_HEADER_SIZE] + app_data_len = min(len(payload) - _ADVERT_HEADER_SIZE, _ADVERT_MAX_APP_DATA_SIZE) + + # pub_key || timestamp is the contiguous slice payload[0:36]; append app_data. + message = ( + payload[0 : _ADVERT_PUB_KEY_SIZE + _ADVERT_TIMESTAMP_SIZE] + + payload[_ADVERT_HEADER_SIZE : _ADVERT_HEADER_SIZE + app_data_len] + ) + + try: + VerifyKey(public_key).verify(message, signature) + return True + except BadSignatureError: + return False + except Exception: + # Malformed/non-canonical public key or other unexpected failure — treat + # as invalid rather than propagating into the packet pipeline. + logger.warning("Advert signature verification errored; treating as invalid", exc_info=True) + return False + + # ============================================================================= # Direct Message (TEXT_MESSAGE) Decryption # ============================================================================= diff --git a/app/packet_processor.py b/app/packet_processor.py index 1b9f5d8..79210a5 100644 --- a/app/packet_processor.py +++ b/app/packet_processor.py @@ -27,6 +27,7 @@ from app.decoder import ( try_decrypt_dm, try_decrypt_packet_with_channel_key, try_decrypt_path, + verify_advert_signature, ) from app.keystore import get_private_key, get_public_key, has_private_key from app.models import ( @@ -533,6 +534,20 @@ async def _process_advertisement( logger.debug("Failed to parse advertisement payload") return + # Reject adverts whose Ed25519 signature does not verify against the embedded + # public key. MeshCore firmware (Mesh.cpp onRecvPacket) drops forged/corrupted + # adverts at exactly this check; without it a bit-flipped advert would be + # ingested as a phantom contact with a mangled public key (issue #315). The raw + # packet is already stored (see process_raw_packet) and still surfaces in the + # debug feed — only contact creation/update is gated here, matching firmware. + if not verify_advert_signature(packet_info.payload): + logger.warning( + "Dropping advertisement with invalid signature from %s (packet %s)", + advert.public_key[:12], + raw_bytes.hex().upper(), + ) + return + new_path_len = packet_info.path_length new_path_hex = packet_info.path.hex() if packet_info.path else "" diff --git a/tests/test_decoder.py b/tests/test_decoder.py index 141ad66..34e55b8 100644 --- a/tests/test_decoder.py +++ b/tests/test_decoder.py @@ -675,6 +675,66 @@ class TestAdvertisementParsing: assert result is None +class TestVerifyAdvertSignature: + """Ed25519 advertisement signature verification (mirrors firmware Mesh.cpp). + + Uses a real captured advert as the golden compatibility vector: it was signed + by a genuine MeshCore device and accepted by firmware, so our verifier must + accept it. Corrupting any signed region (pubkey, body, or signature) must be + rejected — this is the phantom-contact bug in issue #315. + """ + + # Real advert packet (name decodes to "Lightless" + emoji), same vector used + # by test_parse_advertisement_extracts_public_key above. + REAL_ADVERT_PACKET = bytes.fromhex( + "1100AE92564C5C9884854F04F469BBB2BAB8871A078053AF6CF4AA2C014B18CE8A83" + "2DBF6669128E9476F36320F21D1B37FF1CF31680F50F4B17EDABCC7CF8C47D3C5E1D" + "F3AFD0C8721EA06A8078462EF241DEF80AD6922751F206E3BB121DFB604F4146D60D" + "913628D902602DB5F8466C696768746C657373F09FA59D" + ) + + def _payload(self) -> bytes: + from app.decoder import parse_packet + + info = parse_packet(self.REAL_ADVERT_PACKET) + assert info is not None + return info.payload + + def test_accepts_real_advert(self): + from app.decoder import verify_advert_signature + + assert verify_advert_signature(self._payload()) is True + + def test_rejects_corrupted_public_key(self): + from app.decoder import verify_advert_signature + + payload = bytearray(self._payload()) + payload[5] ^= 0x01 # flip a bit in the public key (bytes 0-31) + assert verify_advert_signature(bytes(payload)) is False + + def test_rejects_corrupted_signed_body(self): + from app.decoder import verify_advert_signature + + payload = bytearray(self._payload()) + payload[110] ^= 0x01 # flip a bit in the app-data (after the signature) + assert verify_advert_signature(bytes(payload)) is False + + def test_rejects_corrupted_signature(self): + from app.decoder import verify_advert_signature + + payload = bytearray(self._payload()) + payload[40] ^= 0x01 # flip a bit inside the 64-byte signature (bytes 36-99) + assert verify_advert_signature(bytes(payload)) is False + + def test_rejects_truncated_payload(self): + from app.decoder import verify_advert_signature + + # Anything shorter than pubkey(32)+timestamp(4)+signature(64)=100 bytes + # cannot be verified and must fail closed. + assert verify_advert_signature(b"") is False + assert verify_advert_signature(self._payload()[:99]) is False + + class TestPublicKeyDerivation: """Test deriving Ed25519 public key from MeshCore private key.""" diff --git a/tests/test_packet_pipeline.py b/tests/test_packet_pipeline.py index 3e58f8c..2aff82c 100644 --- a/tests/test_packet_pipeline.py +++ b/tests/test_packet_pipeline.py @@ -369,7 +369,10 @@ class TestAdvertisementPipeline: short_packet_info.payload = b"" with patch("app.packet_processor.broadcast_event", mock_broadcast): - with patch("app.packet_processor.parse_advertisement") as mock_parse: + with ( + patch("app.packet_processor.parse_advertisement") as mock_parse, + patch("app.packet_processor.verify_advert_signature", return_value=True), + ): mock_parse.return_value = ParsedAdvertisement( public_key=test_pubkey, name="TestNode", @@ -387,7 +390,10 @@ class TestAdvertisementPipeline: long_packet_info.payload = b"" with patch("app.packet_processor.broadcast_event", mock_broadcast): - with patch("app.packet_processor.parse_advertisement") as mock_parse: + with ( + patch("app.packet_processor.parse_advertisement") as mock_parse, + patch("app.packet_processor.verify_advert_signature", return_value=True), + ): mock_parse.return_value = ParsedAdvertisement( public_key=test_pubkey, name="TestNode", @@ -439,7 +445,10 @@ class TestAdvertisementPipeline: skewed_shorter_packet_info.payload = b"" with patch("app.packet_processor.broadcast_event", mock_broadcast): - with patch("app.packet_processor.parse_advertisement") as mock_parse: + with ( + patch("app.packet_processor.parse_advertisement") as mock_parse, + patch("app.packet_processor.verify_advert_signature", return_value=True), + ): mock_parse.return_value = ParsedAdvertisement( public_key=test_pubkey, name="TestNode", @@ -453,7 +462,10 @@ class TestAdvertisementPipeline: ) with patch("app.packet_processor.broadcast_event", mock_broadcast): - with patch("app.packet_processor.parse_advertisement") as mock_parse: + with ( + patch("app.packet_processor.parse_advertisement") as mock_parse, + patch("app.packet_processor.verify_advert_signature", return_value=True), + ): mock_parse.return_value = ParsedAdvertisement( public_key=test_pubkey, name="TestNode", @@ -501,7 +513,10 @@ class TestAdvertisementPipeline: packet_info.path_hash_size = 1 with patch("app.packet_processor.broadcast_event", mock_broadcast): - with patch("app.packet_processor.parse_advertisement") as mock_parse: + with ( + patch("app.packet_processor.parse_advertisement") as mock_parse, + patch("app.packet_processor.verify_advert_signature", return_value=True), + ): mock_parse.return_value = ParsedAdvertisement( public_key=test_pubkey, name="TestNode", @@ -552,7 +567,10 @@ class TestAdvertisementPipeline: long_packet_info.payload = b"" with patch("app.packet_processor.broadcast_event", mock_broadcast): - with patch("app.packet_processor.parse_advertisement") as mock_parse: + with ( + patch("app.packet_processor.parse_advertisement") as mock_parse, + patch("app.packet_processor.verify_advert_signature", return_value=True), + ): mock_parse.return_value = ParsedAdvertisement( public_key=test_pubkey, name="TestNode", @@ -572,6 +590,45 @@ class TestAdvertisementPipeline: ("aa", 1), ] + @pytest.mark.asyncio + async def test_advertisement_with_forged_signature_creates_no_contact( + self, test_db, captured_broadcasts + ): + """A corrupted/forged advert must not be ingested as a phantom contact (#315). + + The raw packet is still stored (so the debug feed sees it), but no contact + row is created and no `contact` event is broadcast — mirroring firmware, + which drops adverts that fail signature verification. + """ + from app.decoder import parse_packet + from app.packet_processor import process_raw_packet + + fixture = FIXTURES["advertisement_chat_node"] + packet_bytes = bytearray(bytes.fromhex(fixture["raw_packet_hex"])) + expected_pubkey = fixture["expected_ws_event"]["data"]["public_key"] + + # Flip a bit inside the public key (payload byte 5) to simulate an + # over-the-air corrupted advert. Parsing still succeeds structurally, but + # the signature no longer matches the (now-mangled) key. + payload_start = len(packet_bytes) - len(parse_packet(bytes(packet_bytes)).payload) + packet_bytes[payload_start + 5] ^= 0x01 + + broadcasts, mock_broadcast = captured_broadcasts + + with patch("app.packet_processor.broadcast_event", mock_broadcast): + result = await process_raw_packet(bytes(packet_bytes), timestamp=1700000000) + + # No contact created — neither under the original key nor the mangled one. + assert await ContactRepository.get_by_key_prefix(expected_pubkey[:12]) is None + all_contacts = await ContactRepository.get_all() + assert all_contacts == [] + + # No contact broadcast emitted. + assert [b for b in broadcasts if b["type"] == "contact"] == [] + + # The raw packet is still stored for the debug feed. + assert result["packet_id"] is not None + class TestPathPacketPipeline: """Test PATH packet learning and bundled ACK handling.""" @@ -2010,7 +2067,10 @@ class TestProcessRawPacketIntegration: raw = b"\x11\x00" + b"\xee" * 30 with patch("app.packet_processor.broadcast_event", mock_broadcast): - with patch("app.packet_processor.parse_advertisement", return_value=advert): + with ( + patch("app.packet_processor.parse_advertisement", return_value=advert), + patch("app.packet_processor.verify_advert_signature", return_value=True), + ): # First arrival: long path with patch("app.packet_processor.parse_packet", return_value=long_path_info): await process_raw_packet(raw, timestamp=5000)