Add more aggressive packet validity checking. Closes #315.

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
 # =============================================================================

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 ""
 

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."""
 

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)