From 5e0deb8a46875f02ffbb3098c1be187056e2502f Mon Sep 17 00:00:00 2001 From: Afri <220195275+l5yth@users.noreply.github.com> Date: Thu, 8 Oct 2026 00:33:48 +0200 Subject: [PATCH] data: check MeshCore RX-log adverts against their signatures (#996) --- CHANGELOG.md | 1 + data/mesh_ingestor/CONTRACTS.md | 4 +- .../protocols/meshcore/advert_replay.py | 216 ++++ .../protocols/meshcore/advert_signature.py | 156 +++ .../protocols/meshcore/handlers.py | 68 +- data/requirements.txt | 2 +- tests/conftest.py | 17 + tests/ed25519_oracle.py | 131 +++ tests/meshcore_frames.py | 197 ++++ tests/test_meshcore_advert_signature.py | 936 ++++++++++++++++++ tests/test_provider_unit.py | 180 ++-- 11 files changed, 1834 insertions(+), 74 deletions(-) create mode 100644 data/mesh_ingestor/protocols/meshcore/advert_replay.py create mode 100644 data/mesh_ingestor/protocols/meshcore/advert_signature.py create mode 100644 tests/ed25519_oracle.py create mode 100644 tests/test_meshcore_advert_signature.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 260a73ff..d64e950b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,7 @@ * Web: a MeshCore channel message stays on its sender's node and no longer moves to, or revives, a same-name node; a sender named by a stale roster is credited to the current key (#883) * Data: the MeshCore ingestor no longer posts placeholder nodes for chat senders or mentions, so hidden-channel names stay hidden; the web app ignores such posts from older ingestors (#883) * Data: a MeshCore sender name shared by several roster contacts resolves to the most recently advertised one (#883) +* Data: the MeshCore ingestor posts an advert from the radio's RX log only when its signature verifies, refusing small-order keys and replayed adverts, so a node's SNR, RSSI and hops now come from the first copy heard; roster contacts are unchanged (SPEC SG1-SG5) * Web: live refreshes keep unchanged nodes-table rows, so an open `+` row, focus, a text selection, an overlay opened from a badge and the row under a scrolled page stay put; a moved overlay's close button now works (#881, SPEC DR1) * Matrix: the bridge crate declares its crates.io package metadata, and its README screenshot no longer links outside the crate (#879) * Data: `TRANSPORT=udp` listens on both Meshtastic multicast groups by default, `239.0.0.69` (firmware 2.8+) and `224.0.0.69`; `MESH_UDP_GROUP` takes a comma-separated list. If you set `MESH_UDP_GROUP=224.0.0.69` before, remove it to also hear firmware 2.8+ nodes (#903) diff --git a/data/mesh_ingestor/CONTRACTS.md b/data/mesh_ingestor/CONTRACTS.md index 6d10fcde..a8dcfd0d 100644 --- a/data/mesh_ingestor/CONTRACTS.md +++ b/data/mesh_ingestor/CONTRACTS.md @@ -167,9 +167,9 @@ MeshCore advert sourcing (capturing adverts from other nodes). A MeshCore node a - *Auto-update re-fetch (freshness).* The provider sets `mc.auto_update_contacts = True`, so the meshcore library re-fetches changed contacts (incrementally, by `lastmod`) whenever an `ADVERTISEMENT` / `PATH_UPDATE` push arrives. A re-advert from a known node therefore refreshes its `last_advert` / position without waiting for a reconnect. - *Bare advert (reach).* The `ADVERTISEMENT` (pubkey-only) push is also handled directly: for a public key not in the contact roster it upserts a minimal "heard now" node (`lastHeard`, `protocol`, `user.shortName`/`publicKey` only - no name/type/position), so radios running with auto-add off still register the advertiser. Known keys are skipped (the auto-update path keeps them fresh). The Ruby web app preserves an existing long name on conflict, so this placeholder never clobbers a richer record, and a later full contact advertisement reconciles it. Reconciliation does not depend on timestamp ordering: the contact record carries `lastHeard = last_advert` (the sender-stamped advert-creation time), which is always older than the placeholder's wall-clock stamp - the web app's node upsert therefore fills identity fields (name, role, public key, …) that are still NULL even from an older-stamped record, while timestamps/telemetry stay freshness-guarded (ACCEPTANCE GH-A1). A record under another key than the row's fills nothing (SPEC NI2). -- *RX-log advert (full identity + signal, roster-independent - SPEC RF3).* Companion firmware ≥ 1.16 pushes every received RF frame (`RX_LOG_DATA`) while a client is connected; the library parses `ADVERT` frames completely (full public key, name, type, optional lat/lon). The ingestor converts these to full node upserts carrying per-reception `snr` / `rssi` / `hopsAway`, so node identity and signal metrics no longer depend on the radio's contact roster at all - including when the roster is full. Absent RX-log frames (older/other builds) are never an error; the three paths above still function. Non-`ADVERT` RX-log frames are not ingested: decrypted `GRP_TXT` copies only route channel messages (see `POST /api/messages`), and the DEBUG-only capture drops their decrypted text. +- *RX-log advert (full identity + signal, roster-independent - SPEC RF3).* Companion firmware ≥ 1.16 pushes every received RF frame (`RX_LOG_DATA`) while a client is connected; the library parses `ADVERT` frames completely (full public key, name, type, optional lat/lon). The radio pushes each frame before the firmware checks it, and the library does not check the advert's signature, so the ingestor does (SPEC SG1): it posts an RX-log advert, node and position, only when the Ed25519 signature in the raw payload (`pkt_payload`) verifies under the advert's own key over `pub_key + timestamp + app_data`, the bytes `Mesh::onRecvPacket` checks. It also refuses app data longer than the 32 bytes the firmware signs, a parsed `adv_key` that is not the key in the signed bytes, and a key of small order, which no private key stands behind (SPEC SG4). Like the firmware's roster, it then posts an advert only when its signed timestamp is newer than the newest one accepted for that key, or reported by a roster contact as its `last_advert` (SPEC SG5): a replay, or the same advert heard again over another path, posts nothing. That memory is per process and mirrors the radio's roster. The first `CONTACTS` listing of a connection, the whole roster, sets which keys are roster keys; a later listing, the auto-update re-fetch of changed contacts, adds the contacts it names; `NEW_CONTACT`, which the radio pushes for an advert from a key it did not add, and `NEXT_CONTACT` only raise a roster key's timestamp. Adverts from new keys cannot push a roster key out. A key the next full listing lacks joins the other keys, of which the memory keeps the 4096 most recently accepted. A key the radio adds is protected once a listing names it, and if auto-add lets the radio's own roster be flooded, the protection is the radio's: a key stays protected while it is in the radio's roster. After a restart a key outside the roster passes its first advert, and roster keys are protected again once the connection's first listing has run. A refused advert posts nothing and logs one debug line (`context=meshcore.rx_advert`, with `node_id` and `reason`). The three paths above are not checked again: the firmware verified each advert they carry before it added the contact or pushed `ADVERTISEMENT`, and a contact added over the companion link itself is the operator's own. The ingestor converts verified adverts to full node upserts carrying the `snr` / `rssi` / `hopsAway` of each advert's first reception, so node identity and signal metrics no longer depend on the radio's contact roster at all - including when the roster is full. Absent RX-log frames (older/other builds) are never an error; the three paths above still function. Non-`ADVERT` RX-log frames are not ingested: decrypted `GRP_TXT` copies only route channel messages (see `POST /api/messages`), and the DEBUG-only capture drops their decrypted text. - *Position anchoring (SPEC MR5).* One advert reaches the radio several times over different flood paths, each copy carrying its own receiver-side `recv_time`. The position derived from an RX-log advert is therefore keyed on the advert's sender-side `adv_timestamp` - both for the `POST /api/positions` record and for the node row's `position.time` - falling back to `recv_time` only when the parser reported no usable value. `lastHeard` stays receiver-side. Because `_store_meshcore_position` derives its row id from `(node_id, position_time)`, every copy of one advert - across flood paths and across co-operating ingestors - collapses to a single position row. New protocols whose position data rides on rebroadcast beacons SHOULD likewise anchor on a sender-side timestamp. + *Position anchoring (SPEC MR5).* One advert reaches the radio several times over different flood paths, each copy carrying its own receiver-side `recv_time`. The position derived from an RX-log advert is therefore keyed on the advert's sender-side `adv_timestamp` - both for the `POST /api/positions` record and for the node row's `position.time` - falling back to `recv_time` only when the parser reported no usable value. `lastHeard` stays receiver-side. Because `_store_meshcore_position` derives its row id from `(node_id, position_time)`, every copy of one advert - across flood paths and across co-operating ingestors - collapses to a single position row. One ingestor posts only the first copy it hears (SPEC SG5); the anchor collapses the copies of co-operating ingestors. New protocols whose position data rides on rebroadcast beacons SHOULD likewise anchor on a sender-side timestamp. *Key binding (SPEC NI3).* An advert, from any of the four paths, whose full public key differs from the key the row is bound to refreshes only `last_heard`, telemetry and the signal fields; the names, role, key and position stay those of the first key (see "Collision trade-off" above). diff --git a/data/mesh_ingestor/protocols/meshcore/advert_replay.py b/data/mesh_ingestor/protocols/meshcore/advert_replay.py new file mode 100644 index 00000000..6338ad4c --- /dev/null +++ b/data/mesh_ingestor/protocols/meshcore/advert_replay.py @@ -0,0 +1,216 @@ +# Copyright © 2025-26 l5yth & contributors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Refuse a MeshCore RX-log advert that is not newer than the last one accepted. + +The firmware's roster refuses an advert whose signed timestamp is not newer +than the last one it accepted from that contact (``BaseChatMesh::onAdvertRecv``, +``src/helpers/BaseChatMesh.cpp:124``). The RX log carries every received +copy of every advert, replays included, so the ingestor applies the same rule +to it (SPEC SG5): per process and per public key it remembers the newest +signed timestamp it accepted, or that the radio's roster reported as a +contact's ``last_advert``, and posts an RX-log advert only when the advert's +own signed timestamp is strictly newer. A replay, and the same advert heard +again over another path, post nothing. + +The memory has two pools. The roster pool mirrors the radio's roster, and +only the radio's ``CONTACTS`` listings change who is in it +(:func:`_take_roster_listing`): the first listing of a connection is the whole +roster and replaces the membership, and a later listing (the library's +auto-update re-fetch, which asks only for contacts changed since the newest +``lastmod`` it saw) adds the contacts it names. ``NEW_CONTACT`` and +``NEXT_CONTACT`` only raise a member's timestamp (:func:`_raise_roster_advert`): +the radio pushes ``NEW_CONTACT`` for an advert from a key it did not add, so a +flood of new keys never enters the roster pool through it; a key the radio +auto-adds enters through the next listing, as a spare the cap evicts first. +Every other key sits in the +RX-log pool of :data:`_REPLAY_MEMORY_CAP` keys, which forgets the least +recently accepted first. + +Limits: a key the radio adds between listings is protected once the next +listing names it, and if auto-add lets the radio's own roster be flooded, the +protection is the radio's: a key stays protected while it is in the radio's +roster. A restart empties both pools until the connection's first listing +refills the roster pool; after that only a key known solely from the RX log +passes its first advert again. +""" + +from __future__ import annotations + +import threading +from collections import OrderedDict + +_REPLAY_MEMORY_CAP = 4096 +"""Most keys the RX-log pool remembers, least recently accepted forgotten +first. + +More than ten times the largest companion roster (``MAX_CONTACTS=350`` in the +firmware's variants), at about 860 KiB when full.""" + +_ROSTER_MEMORY_CAP = 1024 +"""Backstop cap on the roster pool, which the radio's own table bounds. + +About three times the largest companion roster. Past it the pool hands the +least recently refreshed key that the latest full listing lacks to the RX-log +pool; it never evicts a key of the latest full listing.""" + +_heard: OrderedDict[str, int] = OrderedDict() +"""Newest accepted signed timestamp per lowercase hex key outside the radio's +roster, least recently accepted first.""" + +_roster: OrderedDict[str, int] = OrderedDict() +"""Newest signed timestamp per lowercase hex key of the radio's roster, least +recently refreshed first.""" + +_listed: set[str] = set() +"""Keys of the latest full listing: the ones :data:`_ROSTER_MEMORY_CAP` never +evicts.""" + +_lock = threading.Lock() +"""Serialises the memory: a reconnect starts a new MeshCore event-loop thread +while the old one may still be finishing its last events.""" + + +def _store( + pool: OrderedDict[str, int], cap: int, pub_key_hex: str, timestamp: int +) -> None: + """Store *timestamp* as the newest for *pub_key_hex* in *pool*, cap it. + + The caller holds :data:`_lock`. + + Parameters: + pool: :data:`_heard` or :data:`_roster`. + cap: Most keys *pool* keeps; the least recent beyond it are forgotten. + pub_key_hex: Lowercase hex public key, made the most recent. + timestamp: Signed advert timestamp to remember. + """ + pool[pub_key_hex] = timestamp + pool.move_to_end(pub_key_hex) + while len(pool) > cap: + pool.popitem(last=False) + + +def _leave_roster(pub_key_hex: str) -> None: + """Move a key out of the roster pool into the RX-log pool, timestamp kept. + + It becomes an ordinary RX-log-pool entry, as if just accepted, so a + replay of its last advert is still refused until that pool forgets it. + The caller holds :data:`_lock`. + + Parameters: + pub_key_hex: A key of :data:`_roster`. + """ + _store(_heard, _REPLAY_MEMORY_CAP, pub_key_hex, _roster.pop(pub_key_hex)) + + +def _accept_advert_timestamp(pub_key_hex: str, timestamp: int) -> bool: + """Accept a verified RX-log advert's timestamp when it is strictly newer. + + Call it only for an advert whose signature holds, so that no forged + timestamp is ever remembered: a forged far-future one would otherwise + lock the genuine node out. The key stays in the pool it is in. + + Parameters: + pub_key_hex: The advert's public key as lowercase hex (``adv_key``). + timestamp: The advert's signed sender-side timestamp. + + Returns: + ``True`` when the timestamp is newer than any remembered for the key, + which then becomes the newest; ``False`` for a replay or a repeat. + """ + with _lock: + if pub_key_hex in _roster: + pool, cap = _roster, _ROSTER_MEMORY_CAP + else: + pool, cap = _heard, _REPLAY_MEMORY_CAP + newest = pool.get(pub_key_hex) + if newest is not None and timestamp <= newest: + return False + _store(pool, cap, pub_key_hex, timestamp) + return True + + +def _take_roster_listing(last_adverts: dict[str, object], *, full: bool) -> None: + """Make the roster pool follow one ``CONTACTS`` listing from the radio. + + The firmware verified the advert each listed contact carries, and its + ``last_advert`` is that advert's signed sender-side timestamp + (``ContactInfo::last_advert_timestamp``), the clock RX-log adverts carry. + Each listed key joins or stays in the roster pool with the newest of its + ``last_advert`` and what was remembered for it, never lowered. + + A full listing (*full*: the first of a connection, asked with ``since = 0``) + is the whole roster: its keys replace the membership, and a member it + lacks leaves for the RX-log pool (:func:`_leave_roster`). A later listing + (the auto-update re-fetch) names only contacts changed since the newest + ``lastmod`` the library saw, so a member it lacks is unchanged and stays. + + Parameters: + last_adverts: Each listed contact's hex public key and ``last_advert``; + a contact whose ``last_advert`` is not an ``int`` is skipped. + full: Whether the listing is the radio's whole roster. + """ + listed = { + key.lower(): stamp + for key, stamp in last_adverts.items() + if isinstance(stamp, int) + } + with _lock: + if full: + for key in [key for key in _roster if key not in listed]: + _leave_roster(key) + _listed.clear() + _listed.update(listed) + for key, stamp in listed.items(): + newest = max(stamp, _roster.get(key, stamp), _heard.pop(key, stamp)) + _roster[key] = newest + _roster.move_to_end(key) + # The radio's table bounds the pool; the cap only backs it up, and it + # never evicts a key of the latest full listing. + spare = [key for key in _roster if key not in _listed] + while len(_roster) > _ROSTER_MEMORY_CAP and spare: + _leave_roster(spare.pop(0)) + + +def _raise_roster_advert(pub_key_hex: str, last_advert: object) -> None: + """Raise a roster member's timestamp to a contact event's ``last_advert``. + + ``NEW_CONTACT`` and ``NEXT_CONTACT`` call this. It never adds a key: the + radio pushes ``NEW_CONTACT`` for an advert from a key it did not add to + its roster (``BaseChatMesh::onAdvertRecv``, ``MyMesh::onDiscoveredContact``), + so only a ``CONTACTS`` listing makes a key a member. + + Parameters: + pub_key_hex: The contact's public key as hex. + last_advert: The contact's ``last_advert``; anything but an ``int`` is + ignored, as is a value not newer than the remembered one. + """ + if not isinstance(last_advert, int): + return + key = pub_key_hex.lower() + with _lock: + if key in _roster and last_advert > _roster[key]: + _roster[key] = last_advert + _roster.move_to_end(key) + + +def _reset_replay_memory() -> None: + """Forget every remembered timestamp and the roster, as a restart does. + + Provided for test isolation. + """ + with _lock: + _heard.clear() + _roster.clear() + _listed.clear() diff --git a/data/mesh_ingestor/protocols/meshcore/advert_signature.py b/data/mesh_ingestor/protocols/meshcore/advert_signature.py new file mode 100644 index 00000000..14362170 --- /dev/null +++ b/data/mesh_ingestor/protocols/meshcore/advert_signature.py @@ -0,0 +1,156 @@ +# Copyright © 2025-26 l5yth & contributors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Check the Ed25519 signature of a MeshCore advert heard in the RX log. + +The companion radio pushes every frame it receives to the RX log before the +firmware checks it (``Dispatcher::checkRecv`` calls ``logRxRaw`` before it +parses the packet), and the ``meshcore`` library's parser extracts an +advert's signature without checking it. The firmware verifies an advert only +on its way into the contact roster (``Mesh::onRecvPacket``), so roster +contacts and ``ADVERTISEMENT`` pushes arrive verified and an RX-log advert +does not. :func:`_rx_advert_signature_problem` repeats the firmware's check +on the raw payload, so the ingestor posts an RX-log advert only when its +signature holds (SPEC SG1) under a key that is not of small order (SG4). +:func:`_rx_advert_signed_timestamp` reads the signed timestamp the replay +check compares (SG5, :mod:`.advert_replay`). +""" + +from __future__ import annotations + +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey + +_PUB_KEY_SIZE = 32 +"""Bytes of the advertiser's Ed25519 public key (firmware ``PUB_KEY_SIZE``).""" + +_SIGNATURE_OFFSET = _PUB_KEY_SIZE + 4 +"""Offset of the signature in an ``ADVERT`` payload: after the public key and +the 4-byte sender-side timestamp.""" + +_APP_DATA_OFFSET = _SIGNATURE_OFFSET + 64 +"""Offset of the app data (flags, position, name): after the 64-byte +signature (firmware ``SIGNATURE_SIZE``).""" + +_MAX_ADVERT_DATA_SIZE = 32 +"""Most app-data bytes an advert carries and signs (firmware +``MAX_ADVERT_DATA_SIZE``; ``Mesh::createAdvert`` refuses more).""" + +# libsodium's small-order blocklist, ge25519_has_small_order() in +# src/libsodium/crypto_core/ed25519/ref10/ed25519_ref10.c, compared as there: +# the first 31 bytes exactly, byte 31 with its top bit (the sign of x) masked. +# The seven values are every encoding of the eight points of order 1, 2, 4 and +# 8 on the RFC 8032 curve (section 5.1), non-canonical y = p and y = p + 1 +# included; tests/ed25519_oracle.py derives them from the RFC's constants. +_SMALL_ORDER_KEYS = frozenset( + bytes.fromhex(encoding) + for encoding in ( + "00" * 32, # 0 (order 4) + "01" + "00" * 31, # 1 (order 1) + "26e8958fc2b227b045c3f489f2ef98f0d5dfac05d3c63339b13802886d53fc05", # order 8 + "c7176a703d4dd84fba3c0b760d10670f2a2053fa2c39ccc64ec7fd7792ac037a", # order 8 + "ec" + "ff" * 30 + "7f", # p - 1 (order 2) + "ed" + "ff" * 30 + "7f", # p (= 0, order 4) + "ee" + "ff" * 30 + "7f", # p + 1 (= 1, order 1) + ) +) +"""Public-key encodings of small order, the sign bit cleared (SPEC SG4).""" + + +def _has_small_order(pub_key: bytes) -> bool: + """Return whether a 32-byte public key encodes a point of small order. + + No private key stands behind such a key and trivial signatures verify + under it, so its signature proves nothing. The sign bit is masked as + libsodium masks it, so the sign-bit variants match too, among them the + three that OpenSSL accepts and the firmware's decoder refuses (``x = 0`` + with the sign bit set): ``01 00..00 80``, ``ec ff..ff ff`` and + ``ee ff..ff ff``. + + Parameters: + pub_key: The advert's raw public key. + + Returns: + ``True`` when the key is on :data:`_SMALL_ORDER_KEYS`. + """ + return pub_key[:31] + bytes([pub_key[31] & 0x7F]) in _SMALL_ORDER_KEYS + + +def _rx_advert_signature_problem(frame: dict) -> str | None: + """Return why an RX-log advert's signature does not hold, or ``None``. + + Checks what ``Mesh::onRecvPacket`` checks, on the raw payload the parser + read its ``adv_*`` fields from: an Ed25519 signature by the advert's own + public key over ``pub_key + timestamp + app_data`` (``Mesh::createAdvert``). + The payload is laid out ``pub_key(32) + timestamp(4) + signature(64) + + app_data``. Three rules go past the firmware's, so that every field the + ingestor posts is signed by a key some private key stands behind: + + * App data longer than 32 bytes is refused. The firmware never sends + more and verifies only the first 32 bytes of a longer frame, while the + parser reads the name to the end of the frame. + * The parsed ``adv_key``, which names the posted node, must be the key + in the signed bytes. + * A key of small order is refused (:func:`_has_small_order`, SPEC SG4). + + Parameters: + frame: Parsed RX-log ``ADVERT`` frame. Reads ``pkt_payload`` (the raw + packet payload as ``bytes`` or ``bytearray``) and ``adv_key``. + + Returns: + ``None`` when the signature holds, else the reason it does not: + ``"no raw payload"``, ``"truncated signature"``, ``"app data over 32 + bytes"``, ``"key mismatch"``, ``"small-order key"`` or ``"invalid + signature"``. + """ + raw = frame.get("pkt_payload") + if not isinstance(raw, (bytes, bytearray)): + return "no raw payload" + raw = bytes(raw) + if len(raw) < _APP_DATA_OFFSET: + return "truncated signature" + if len(raw) - _APP_DATA_OFFSET > _MAX_ADVERT_DATA_SIZE: + return "app data over 32 bytes" + pub_key = raw[:_PUB_KEY_SIZE] + if frame.get("adv_key") != pub_key.hex(): + return "key mismatch" + if _has_small_order(pub_key): + return "small-order key" + # The signed message skips the signature itself: key, timestamp, app data. + message = raw[:_SIGNATURE_OFFSET] + raw[_APP_DATA_OFFSET:] + try: + Ed25519PublicKey.from_public_bytes(pub_key).verify( + raw[_SIGNATURE_OFFSET:_APP_DATA_OFFSET], message + ) + except InvalidSignature: + return "invalid signature" + return None + + +def _rx_advert_signed_timestamp(frame: dict) -> int: + """Return the sender-side timestamp an RX-log advert's signature covers. + + Reads the four little-endian bytes after the public key in ``pkt_payload``, + the bytes the parser reports as ``adv_timestamp``. Call it only for a + frame whose signature holds (:func:`_rx_advert_signature_problem` returned + ``None``): only then is the timestamp the advertiser's own. + + Parameters: + frame: Parsed RX-log ``ADVERT`` frame whose signature holds. + + Returns: + The signed ``uint32`` advert timestamp. + """ + raw = bytes(frame["pkt_payload"]) + return int.from_bytes(raw[_PUB_KEY_SIZE:_SIGNATURE_OFFSET], "little") diff --git a/data/mesh_ingestor/protocols/meshcore/handlers.py b/data/mesh_ingestor/protocols/meshcore/handlers.py index f643d06b..c8b3e0b5 100644 --- a/data/mesh_ingestor/protocols/meshcore/handlers.py +++ b/data/mesh_ingestor/protocols/meshcore/handlers.py @@ -20,6 +20,12 @@ import sys import time from ... import config, ingestors as _ingestors +from .advert_replay import ( + _accept_advert_timestamp, + _raise_roster_advert, + _take_roster_listing, +) +from .advert_signature import _rx_advert_signature_problem, _rx_advert_signed_timestamp from .decode import ( _advert_to_node_dict, _contact_to_node_dict, @@ -108,20 +114,32 @@ def _process_self_info( def _process_contacts( - contacts: dict, iface: _MeshcoreInterface, handlers: object + contacts: dict, + iface: _MeshcoreInterface, + handlers: object, + *, + full_listing: bool = False, ) -> None: """Apply a bulk ``CONTACTS`` payload: update the local snapshot and upsert nodes. + The listing also sets which keys the RX-log replay memory treats as the + radio's roster (SPEC SG5, :func:`.advert_replay._take_roster_listing`). + Parameters: contacts: Mapping of full ``public_key`` hex strings to contact dicts. iface: Active interface whose contact snapshot will be updated. handlers: Module reference for :func:`~data.mesh_ingestor.handlers`. + full_listing: ``True`` when *contacts* is the radio's whole roster + (the first listing of a connection), ``False`` for a listing of + changed contacts only (the auto-update re-fetch). """ + last_adverts: dict[str, object] = {} for pub_key, contact in contacts.items(): node_id = _meshcore_node_id(pub_key) if node_id is None: continue iface._update_contact(contact) + last_adverts[pub_key] = contact.get("last_advert") handlers.upsert_node(node_id, _contact_to_node_dict(contact)) lat = contact.get("adv_lat") lon = contact.get("adv_lon") @@ -139,6 +157,10 @@ def _process_contacts( pub_key, rx_time=last_advert, ) + # The firmware verified each listed contact's last advert: the listing + # sets the roster pool, whose timestamps refuse an RX-log copy of those + # adverts, or older ones (SPEC SG5). + _take_roster_listing(last_adverts, full=full_listing) # Companion-link roster fetch, not over-air frames: clock only (Model A). handlers._mark_packet_activity() @@ -158,6 +180,9 @@ def _process_contact_update( if node_id is None: return iface._update_contact(contact) + # Raises a roster member's timestamp only, never adds a member: the radio + # pushes NEW_CONTACT for an advert from a key it did not add (SPEC SG5). + _raise_roster_advert(pub_key, contact.get("last_advert")) handlers.upsert_node(node_id, _contact_to_node_dict(contact)) lat = contact.get("adv_lat") lon = contact.get("adv_lon") @@ -215,8 +240,18 @@ def _make_event_handlers(iface: _MeshcoreInterface, target: str | None) -> dict: async def on_self_info(evt) -> None: _process_self_info(evt.payload or {}, iface, _handlers) + # The first CONTACTS listing of a connection is the radio's whole roster: + # each connection builds a new MeshCore, whose first fetch asks with + # since = 0. Later listings come from the auto-update re-fetch, which asks + # only for contacts changed since the newest lastmod (SPEC SG5). + full_listing_seen = False + async def on_contacts(evt) -> None: - _process_contacts(evt.payload or {}, iface, _handlers) + nonlocal full_listing_seen + _process_contacts( + evt.payload or {}, iface, _handlers, full_listing=not full_listing_seen + ) + full_listing_seen = True async def on_contact_update(evt) -> None: _process_contact_update(evt.payload or {}, iface, _handlers) @@ -363,8 +398,9 @@ def _make_event_handlers(iface: _MeshcoreInterface, target: str | None) -> dict: # "what is in the air" — while the decoded high-level events # (CHANNEL_MSG_RECV, CONTACT_MSG_RECV, ADVERTISEMENT, telemetry # responses) are duplicates of these same frames and only advance the - # reconnect clock. Counting precedes the DEBUG-capture drop and the - # malformed-advert skip, so no received frame — ignored, errored, or + # reconnect clock. Counting precedes the DEBUG-capture drop, the + # malformed-advert skip and the signature and replay checks, so no + # received frame — ignored, errored, forged, replayed, or # unimplemented — is under-reported. _handlers._mark_packet_seen() # Remember each decrypted channel-message copy so on_channel_msg can @@ -395,6 +431,30 @@ def _make_event_handlers(iface: _MeshcoreInterface, target: str | None) -> dict: ) return + # The radio logs every frame before the firmware checks it, and the + # library parses an advert without checking its signature, so an + # RX-log advert posts its node and position only when its signature + # holds under a key not of small order (SPEC SG1, SG4) and its signed + # timestamp is newer than the last one accepted for that key (SG5): + # a replay, or the same advert over another path, posts nothing. The + # replay check runs only on a verified advert, so a forged timestamp + # is never remembered. The drop logs at debug level: a forged, broken + # or repeated frame is per-packet noise a warning would flood the log + # with. + reason = _rx_advert_signature_problem(payload) + if reason is None and not _accept_advert_timestamp( + pub_key, _rx_advert_signed_timestamp(payload) + ): + reason = "replayed advert" + if reason is not None: + config._debug_log( + "Unverified RX-log advert dropped", + context="meshcore.rx_advert", + node_id=node_id, + reason=reason, + ) + return + _handlers.upsert_node(node_id, _rx_advert_to_node_dict(payload)) lat = payload.get("adv_lat") lon = payload.get("adv_lon") diff --git a/data/requirements.txt b/data/requirements.txt index ce5dfdc9..bc781dee 100644 --- a/data/requirements.txt +++ b/data/requirements.txt @@ -9,5 +9,5 @@ protobuf==7.36.2 # armv7 images compile cryptography from source with the Rust of the base # image's Alpine (3.24: Rust 1.96); a release needing a newer Rust breaks that # build (DK-A2), so check its rust-version before bumping. -cryptography==50.0.2 # AES-CTR decryption for the passive UDP transport +cryptography==50.0.2 # UDP transport AES-CTR, MeshCore RX-log advert signatures rns==1.5.6 # Reticulum announce listener (PROTOCOL=reticulum) diff --git a/tests/conftest.py b/tests/conftest.py index 0b72c065..f0e7756c 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -26,6 +26,23 @@ if str(REPO_ROOT) not in sys.path: import data.mesh_ingestor.config as config +_ADVERT_REPLAY_MODULE = "data.mesh_ingestor.protocols.meshcore.advert_replay" +"""Module holding the per-process MeshCore advert replay memory (SPEC SG5).""" + + +@pytest.fixture(autouse=True) +def forget_meshcore_advert_timestamps(): + """Start every test with an empty MeshCore advert replay memory (SPEC SG5). + + The memory lives for the whole process, so an advert one test hears would + otherwise refuse the same advert in a later test. Nothing is imported: a + run that never loaded the MeshCore package has no memory to clear. + """ + + replay = sys.modules.get(_ADVERT_REPLAY_MODULE) + if replay is not None: + replay._reset_replay_memory() + @pytest.fixture def permit_tx(monkeypatch): diff --git a/tests/ed25519_oracle.py b/tests/ed25519_oracle.py new file mode 100644 index 00000000..bbe5fc22 --- /dev/null +++ b/tests/ed25519_oracle.py @@ -0,0 +1,131 @@ +# Copyright © 2025-26 l5yth & contributors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""RFC 8032 Ed25519 arithmetic: a test oracle for the small-order blocklist. + +Derives, from the curve constants of RFC 8032 section 5.1 alone, every 32-byte +encoding of a point of small order (order 1, 2, 4 or 8), so the ingestor's +blocklist (libsodium's ``ge25519_has_small_order`` list, SPEC SG4) is checked +against first principles rather than copied from it. Pure Python affine +arithmetic, slow and not constant-time: for tests only. +""" + +from __future__ import annotations + +P = 2**255 - 19 +"""The field prime ``p`` (RFC 8032 section 5.1).""" + +D = -121665 * pow(121666, P - 2, P) % P +"""The curve constant ``d = -121665/121666`` (RFC 8032 section 5.1).""" + +L = 2**252 + 27742317777372353535851937790883648493 +"""The prime order of the base point; the curve's order is ``8 * L``.""" + +SQRT_M1 = pow(2, (P - 1) // 4, P) +"""A square root of ``-1`` modulo ``p`` (RFC 8032 section 5.1.3).""" + +IDENTITY = (0, 1) +"""The neutral point ``(x, y) = (0, 1)``.""" + + +def point_add(a: tuple[int, int], b: tuple[int, int]) -> tuple[int, int]: + """Add two points of ``-x^2 + y^2 = 1 + d x^2 y^2`` (RFC 8032 section 5.1.4). + + Parameters: + a: First point ``(x, y)``. + b: Second point ``(x, y)``. + + Returns: + The sum ``a + b``. + """ + (x1, y1), (x2, y2) = a, b + t = D * x1 * x2 * y1 * y2 % P + x3 = (x1 * y2 + y1 * x2) * pow(1 + t, P - 2, P) % P + y3 = (y1 * y2 + x1 * x2) * pow(1 - t, P - 2, P) % P + return x3, y3 + + +def scalar_mult(k: int, point: tuple[int, int]) -> tuple[int, int]: + """Return ``[k] point`` by double-and-add. + + Parameters: + k: Non-negative scalar. + point: Point ``(x, y)``. + + Returns: + The multiple. + """ + result, addend = IDENTITY, point + while k: + if k & 1: + result = point_add(result, addend) + addend = point_add(addend, addend) + k >>= 1 + return result + + +def curve_x(y: int) -> int | None: + """Return an ``x`` with ``(x, y)`` on the curve (RFC 8032 section 5.1.3). + + Parameters: + y: Coordinate below ``p``. + + Returns: + One of the two square roots, or ``None`` when no point has this ``y``. + """ + u, v = (y * y - 1) % P, (D * y * y + 1) % P + x = u * pow(v, 3, P) * pow(u * pow(v, 7, P), (P - 5) // 8, P) % P + if v * x * x % P == u: + return x + if v * x * x % P == -u % P: + return x * SQRT_M1 % P + return None + + +def small_order_points() -> set[tuple[int, int]]: + """Return the eight points of order dividing 8. + + The group is cyclic of order 8 times the prime ``L``, so ``[L] Q`` has order + dividing 8 for every point ``Q``; one of order 8 yields all eight as its + multiples. + + Returns: + The eight small-order points. + """ + points: set[tuple[int, int]] = set() + y = 2 + while len(points) < 8: + x = curve_x(y) + if x is not None: + torsion = scalar_mult(L, (x, y)) + points |= {scalar_mult(k, torsion) for k in range(8)} + y += 1 + return points + + +def small_order_encodings() -> frozenset[bytes]: + """Return every encoding of a small-order point, with the sign bit cleared. + + Covers each point's canonical ``y`` and, where ``y + p`` still fits the + 255-bit field, its non-canonical twin, in the little-endian encoding of + RFC 8032 section 5.1.2 with the top bit (the sign of ``x``) cleared. + + Returns: + The 32-byte encodings. + """ + encodings = set() + for _x, y in small_order_points(): + for value in (y, y + P): + if value < 2**255: + encodings.add(value.to_bytes(32, "little")) + return frozenset(encodings) diff --git a/tests/meshcore_frames.py b/tests/meshcore_frames.py index c7a4ebff..da374c0f 100644 --- a/tests/meshcore_frames.py +++ b/tests/meshcore_frames.py @@ -23,6 +23,13 @@ flood-scope helpers are an independent oracle of the firmware rules (``TransportKey::calcTransportCode`` and ``RegionMap::getTransportKeysFor``), deliberately not shared with the code under test. Seeded from the #765 scoping probes. + +The advert helpers build ``ADVERT`` payloads the way ``Mesh::createAdvert`` and +``AdvertDataBuilder::encodeTo`` do, signed with Ed25519 keys derived from fixed +seeds when a test runs, and parse them with the pinned library's packet parser +as the reader does for an RX-log push (SPEC SG1). They are a second +independent oracle: the signed bytes are rebuilt here, not taken from the code +under test. """ from __future__ import annotations @@ -30,10 +37,14 @@ from __future__ import annotations import asyncio import hashlib import hmac +import struct import types import meshcore from Crypto.Cipher import AES +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat +from meshcore.meshcore_parser import MeshcorePacketParser CHANNEL_NAME = "#test" """Hashtag channel the frames are encrypted for (index 0).""" @@ -231,6 +242,192 @@ def message_hash(sender_ts: int = SENDER_TS, text: str = TEXT) -> int: return int.from_bytes(digest.digest()[:4], "little") +ADV_TYPE_NONE = 0 +"""Firmware ``ADV_TYPE_NONE``: an advert that names no node type.""" + +ADV_TYPE_CHAT = 1 +"""Firmware ``ADV_TYPE_CHAT``: a companion (chat) node.""" + +ADV_TYPE_REPEATER = 2 +"""Firmware ``ADV_TYPE_REPEATER``: a repeater.""" + +ADV_LATLON_MASK = 0x10 +"""Firmware ``ADV_LATLON_MASK``: the app data carries a latitude and longitude.""" + +ADV_NAME_MASK = 0x80 +"""Firmware ``ADV_NAME_MASK``: the app data carries a node name.""" + +MAX_ADVERT_DATA_SIZE = 32 +"""Firmware ``MAX_ADVERT_DATA_SIZE``: the most app-data bytes an advert signs.""" + + +def advert_key(seed: str) -> Ed25519PrivateKey: + """Return a deterministic Ed25519 key for one test advertiser. + + The key is derived when the test runs, with ``SHA256(seed)`` as the + RFC 8032 private-key seed, so no private key is stored in a fixture. + + Parameters: + seed: Fixed label naming the advertiser. + + Returns: + The advertiser's private key. + """ + return Ed25519PrivateKey.from_private_bytes( + hashlib.sha256(seed.encode("utf-8")).digest() + ) + + +def advert_public_key(key: Ed25519PrivateKey) -> bytes: + """Return the 32 raw public-key bytes an advert carries for *key*. + + Parameters: + key: Advertiser key from :func:`advert_key`. + + Returns: + The raw Ed25519 public key. + """ + return key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw) + + +def advert_node_id(key: Ed25519PrivateKey) -> str: + """Return the node id the ingestor posts *key* under. + + Parameters: + key: Advertiser key from :func:`advert_key`. + + Returns: + ``!`` followed by the public key's first four bytes in lowercase hex. + """ + return "!" + advert_public_key(key)[:4].hex() + + +def advert_app_data( + adv_type: int = ADV_TYPE_NONE, + *, + name: str | None = None, + lat_e6: int | None = None, + lon_e6: int | None = None, +) -> bytes: + """Encode an advert's app data as ``AdvertDataBuilder::encodeTo`` does. + + Parameters: + adv_type: ``ADV_TYPE_*`` value for the low four flag bits. + name: Node name, cut to fit the 32-byte app data as the firmware cuts + it; ``None`` or empty leaves the name out. + lat_e6: Latitude in millionths of a degree; used with *lon_e6*. + lon_e6: Longitude in millionths of a degree; used with *lat_e6*. + + Returns: + ``flags(1) + [lat(4) + lon(4)] + [name]``, integers little-endian. + """ + flags = adv_type + body = b"" + if lat_e6 is not None and lon_e6 is not None: + flags |= ADV_LATLON_MASK + body += struct.pack(" bytes: + """Return the bytes ``Mesh::createAdvert`` signs for one advert. + + Parameters: + public_key: The advertiser's raw public key (:func:`advert_public_key`). + timestamp: Sender-side advert time. + app_data: Encoded app data from :func:`advert_app_data`. + + Returns: + ``pub_key(32) + timestamp(4) + app_data``. + """ + return public_key + struct.pack(" bytes: + """Lay out an ``ADVERT`` packet payload for any public-key bytes. + + For keys no private key stands behind, such as the small-order encodings; + :func:`advert_payload` signs with a test key instead. + + Parameters: + public_key: The 32 bytes to send as the advertiser's key. + timestamp: Sender-side advert time, sent as a little-endian ``uint32``. + signature: The 64 bytes to send as the signature. + app_data: Encoded app data from :func:`advert_app_data`. + + Returns: + ``pub_key(32) + timestamp(4) + signature(64) + app_data``. + """ + return public_key + struct.pack(" bytes: + """Build an ``ADVERT`` packet payload as ``Mesh::createAdvert`` does. + + Parameters: + key: Advertiser key from :func:`advert_key`. + timestamp: Sender-side advert time, sent as a little-endian ``uint32``. + app_data: Encoded app data from :func:`advert_app_data`. + signature: Bytes to send in the signature's place, to forge or cut + one; by default *key* signs :func:`advert_signed_message`. + + Returns: + ``pub_key(32) + timestamp(4) + signature(64) + app_data``. + """ + public_key = advert_public_key(key) + if signature is None: + signature = key.sign(advert_signed_message(public_key, timestamp, app_data)) + return advert_payload_bytes(public_key, timestamp, signature, app_data) + + +def rx_log_advert( + payload: bytes, + *, + recv_time: int, + snr: float | None = None, + rssi: int | None = None, + hops: int = 0, +) -> dict: + """Return the ``RX_LOG_DATA`` event payload the reader dispatches for an advert. + + Wraps *payload* in a flood packet that travelled *hops* repeaters and + parses it with the pinned library's ``MeshcorePacketParser``, as + ``MessageReader`` does for a ``LOG_RX_DATA`` push. The reader's + ``raw_hex`` and ``payload`` hex copies are left out: the ingestor reads + the parsed fields and ``pkt_payload`` only. + + Parameters: + payload: Advert payload from :func:`advert_payload`. + recv_time: Receiver-side reception time. + snr: Reception SNR in dB, or ``None`` to leave it out. + rssi: Reception RSSI in dBm, or ``None`` to leave it out. + hops: Repeaters the copy travelled, as one-byte hashes ``01``, ``02``... + + Returns: + The parsed RX-log frame; ``pkt_payload`` holds *payload*. + """ + log: dict = {"recv_time": recv_time} + if snr is not None: + log["snr"] = snr + if rssi is not None: + log["rssi"] = rssi + raw = raw_packet( + payload, path=bytes(range(1, hops + 1)), payload_type=PAYLOAD_TYPE_ADVERT + ) + return asyncio.run(MeshcorePacketParser().parsePacketPayload(raw, log)) + + class FakeConnection: """Serial-link stand-in for ``meshcore.MeshCore``; it never sends a byte.""" diff --git a/tests/test_meshcore_advert_signature.py b/tests/test_meshcore_advert_signature.py new file mode 100644 index 00000000..61b59327 --- /dev/null +++ b/tests/test_meshcore_advert_signature.py @@ -0,0 +1,936 @@ +# Copyright © 2025-26 l5yth & contributors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Regression guard: an RX-log advert is ingested only when its signature holds. + +The companion radio pushes every frame it receives to the RX log before the +firmware checks it (``Dispatcher::checkRecv`` calls ``logRxRaw`` before +``tryParsePacket``), and the pinned ``meshcore`` parser reads an advert's +signature without checking it. The firmware verifies an advert only on its +way into the contact roster (``Mesh::onRecvPacket``), and before this check +the ingestor posted every parsed RX-log advert. It now posts one only when +its signature holds (SPEC SG1-SG3), under a key not of small order (SG4), and +when its signed timestamp is newer than the last one accepted for its key +(SG5). + +Each case builds an on-air advert with :mod:`meshcore_frames` from a key +derived from a fixed seed, parses it with the pinned library's +``MeshcorePacketParser`` and hands it to the ingestor's ``RX_LOG_DATA`` +handler with the real ``handlers`` module; only the HTTP queue is replaced. +The small-order cases send keys no private key stands behind, and their +blocklist is checked against :mod:`ed25519_oracle`, which derives it from the +RFC 8032 curve constants. ``tests/conftest.py`` empties the per-process replay +memory before every test. +""" + +from __future__ import annotations + +import asyncio +import sys +import types +from pathlib import Path + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey + +REPO_ROOT = Path(__file__).resolve().parents[1] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +import ed25519_oracle # noqa: E402 - pytest puts tests/ on sys.path +import meshcore_frames as frames # noqa: E402 - pytest puts tests/ on sys.path + +import data.mesh_ingestor.activity as activity # noqa: E402 +import data.mesh_ingestor.config as config # noqa: E402 +import data.mesh_ingestor.queue as queue # noqa: E402 +from data.mesh_ingestor.protocols.meshcore import ( # noqa: E402 + _MeshcoreInterface, + _make_event_handlers, +) + +VICTIM = frames.advert_key("potato-mesh test advertiser: victim") +"""Key of the node a forged advert claims to be (derived per run, never stored).""" + +VICTIM_ID = frames.advert_node_id(VICTIM) +"""Canonical node id the victim's adverts post under.""" + +VICTIM_KEY = frames.advert_public_key(VICTIM) +"""The victim's raw public key.""" + +VICTIM_KEY_HEX = VICTIM_KEY.hex() +"""The victim's full public key as the parser reports it (``adv_key``).""" + +OTHER_KEY_HEX = frames.advert_public_key( + frames.advert_key("potato-mesh test advertiser: other") +).hex() +"""A second node's full public key.""" + +THIRD_KEY_HEX = frames.advert_public_key( + frames.advert_key("potato-mesh test advertiser: third") +).hex() +"""A third node's full public key.""" + +SENT_AT = 1_791_288_000 +"""Sender-side advert time: 2026-10-06T12:00:00Z.""" + +BERLIN = {"lat_e6": 52_520_000, "lon_e6": 13_405_000} +"""Advertised position of every case: 52.52 N, 13.405 E.""" + +LONG_NAME = "Alice of the Tiergarten" +"""A 23-byte name: with the flags and the position the app data fills 32 bytes.""" + +DROPPED = "Unverified RX-log advert dropped" +"""Log message of a dropped RX-log advert.""" + + +@pytest.fixture +def posts(monkeypatch): + """Capture every queued POST, with MeshCore as the active protocol. + + Returns: + A list that collects ``(path, payload)`` for each queued request. + """ + sent: list = [] + monkeypatch.setattr(config, "PROTOCOL", "meshcore") + monkeypatch.setattr( + queue, + "_queue_post_json", + lambda path, payload, *_args, **_kwargs: sent.append((path, payload)), + ) + return sent + + +@pytest.fixture +def logged(monkeypatch): + """Capture the RX-log advert log lines instead of printing them. + + Returns: + A list that collects each ``meshcore.rx_advert`` line's message, + severity and metadata. + """ + lines: list = [] + + def record(message, *, context=None, severity="debug", **metadata): + """Keep the RX-log advert lines and drop every other log line.""" + if context == "meshcore.rx_advert": + lines.append({"message": message, "severity": severity, **metadata}) + + monkeypatch.setattr(config, "_debug_log", record) + return lines + + +@pytest.fixture +def replay(): + """Return the replay-memory module (SPEC SG5), imported when a test needs it. + + Imported here rather than at the top so that the handler-level cases + still run, and fail on their own assertions, against code without it. + + Returns: + :mod:`data.mesh_ingestor.protocols.meshcore.advert_replay`. + """ + from data.mesh_ingestor.protocols.meshcore import advert_replay + + return advert_replay + + +def _hear(frame: dict) -> None: + """Hand one parsed RX-log frame to the ingestor's ``RX_LOG_DATA`` handler. + + Parameters: + frame: RX-log payload as the reader dispatches it. + """ + hmap = _make_event_handlers(_MeshcoreInterface(target=None), "/dev/ttyUSB0") + asyncio.run(hmap["RX_LOG_DATA"](types.SimpleNamespace(payload=frame))) + + +def _app_data(name: str) -> bytes: + """Encode the app data of a companion at :data:`BERLIN` named *name*. + + Parameters: + name: Advertised node name. + + Returns: + The encoded app data. + """ + return frames.advert_app_data(frames.ADV_TYPE_CHAT, name=name, **BERLIN) + + +def _signature(app_data: bytes, timestamp: int = SENT_AT) -> bytes: + """Return the victim's own signature over its advert carrying *app_data*. + + Parameters: + app_data: Encoded app data. + timestamp: Sender-side advert time the signature covers. + + Returns: + The 64-byte Ed25519 signature. + """ + return VICTIM.sign(frames.advert_signed_message(VICTIM_KEY, timestamp, app_data)) + + +def _victim_frame( + app_data: bytes, + *, + signature: bytes | None = None, + timestamp: int = SENT_AT, + hops: int = 2, + snr: float = 9.0, + rssi: int = -80, +) -> dict: + """Return the parsed RX-log copy of an advert carrying the victim's key. + + Parameters: + app_data: Encoded app data the frame carries. + signature: Signature bytes the frame carries; the victim's genuine + signature over this frame by default. + timestamp: Sender-side advert time the frame carries. + hops: Repeaters this copy travelled. + snr: Reception SNR of this copy in dB. + rssi: Reception RSSI of this copy in dBm. + + Returns: + The parsed frame. + """ + payload = frames.advert_payload(VICTIM, timestamp, app_data, signature=signature) + return frames.rx_log_advert( + payload, recv_time=SENT_AT + 5, snr=snr, rssi=rssi, hops=hops + ) + + +def _dropped(reason: str, node_id: str = VICTIM_ID) -> list: + """Return the one log line a dropped advert writes. + + Parameters: + reason: Why the signature check failed. + node_id: Node id the advert claimed. + + Returns: + The expected :func:`logged` entries. + """ + return [ + {"message": DROPPED, "severity": "debug", "node_id": node_id, "reason": reason} + ] + + +FORGED = { + # An all-zero signature. + "zeroed": lambda: _victim_frame(_app_data("Mallory"), signature=bytes(64)), + # The victim's signature over other bytes: its own advert naming Alice. + "other-bytes": lambda: _victim_frame( + _app_data("Mallory"), signature=_signature(_app_data("Alice")) + ), + # A signature one byte short: the parser reads 64 bytes regardless, so the + # first app-data byte ends the signature. + "truncated": lambda: _victim_frame( + _app_data("Alice"), signature=_signature(_app_data("Alice"))[:63] + ), +} +"""Adverts carrying the victim's key whose signature does not hold, by case.""" + + +@pytest.mark.parametrize("case", sorted(FORGED)) +def test_rx_log_advert_with_an_invalid_signature_is_not_ingested(case, posts, logged): + """A zeroed signature, one over other bytes or a cut one posts neither a + node nor a position, and the drop is logged at debug level with the node + id and the reason.""" + + frame = FORGED[case]() + assert frame["adv_key"] == VICTIM_KEY_HEX + + _hear(frame) + + assert posts == [], f"{case} advert posted {posts}" + assert logged == _dropped("invalid signature") + + +def _heard_before(frame: dict) -> dict: + """Hear *frame* once, so that hearing it again is a replay (SPEC SG5). + + Parameters: + frame: Parsed RX-log frame of a genuine advert. + + Returns: + The same frame. + """ + _hear(frame) + return frame + + +COUNTED = { + # Refused by the signature check: the zeroed signature of FORGED. + "refused": (FORGED["zeroed"], []), + # Refused by the replay check: a genuine advert heard once already. + "replayed": (lambda: _heard_before(_victim_frame(_app_data("Alice"))), []), + # Let through: the victim's own advert. + "verified": ( + lambda: _victim_frame(_app_data("Alice")), + ["/api/nodes", "/api/positions"], + ), +} +"""Adverts the checks refuse and one they let through, by case: frame, routes.""" + + +@pytest.mark.parametrize("case", sorted(COUNTED)) +def test_rx_log_advert_counts_as_air_traffic_once_whether_refused_or_verified( + case, posts +): + """Every RX-log advert counts once toward the activity total, before the + signature and replay checks run (SPEC SG2, SG5, MA1): a refused or + replayed frame still counts and posts nothing, and a verified one counts + once and posts as before.""" + + build, routes = COUNTED[case] + frame = build() + activity.take_packet_count() # drain what earlier tests and the set-up counted + posts.clear() + + _hear(frame) + + assert activity.take_packet_count() == 1 + assert [path for path, _payload in posts] == routes + + +def test_rx_log_advert_with_a_valid_signature_is_ingested(posts, logged): + """A genuinely signed advert posts its node and its position, as before.""" + + _hear(_victim_frame(_app_data("Alice"))) + + assert [path for path, _payload in posts] == ["/api/nodes", "/api/positions"] + user = posts[0][1][VICTIM_ID]["user"] + assert (user["longName"], user["publicKey"]) == ("Alice", VICTIM_KEY_HEX) + position = posts[1][1] + assert ( + position["node_id"], + position["latitude"], + position["longitude"], + position["position_time"], + ) == (VICTIM_ID, 52.52, 13.405, SENT_AT) + assert [line["message"] for line in logged] == ["MeshCore RX-log advert"] + + +def test_rx_log_advert_with_app_data_past_the_signed_32_bytes_is_not_ingested( + posts, logged +): + """App data past 32 bytes is outside the signature, so the advert is dropped. + + The firmware never sends more than 32 bytes of app data and verifies only + the first 32 of a longer frame (``Mesh::onRecvPacket``), while the pinned + parser reads the name to the end of the frame: the bytes past 32 would + reach the posted name unsigned. + """ + + app_data = _app_data(LONG_NAME) + assert len(app_data) == frames.MAX_ADVERT_DATA_SIZE + signature = _signature(app_data) + frame = _victim_frame(app_data + b" (not signed)", signature=signature) + # The firmware's own check passes: it stops reading at 32 bytes. + VICTIM.public_key().verify( + signature, frames.advert_signed_message(VICTIM_KEY, SENT_AT, app_data) + ) + assert frame["adv_name"] == LONG_NAME + " (not signed)" + + _hear(frame) + + assert posts == [], f"advert with unsigned app data posted {posts}" + assert logged == _dropped("app data over 32 bytes") + + +def _without(frame: dict, key: str) -> dict: + """Return a copy of *frame* without *key*. + + Parameters: + frame: Parsed RX-log frame. + key: Field to leave out. + + Returns: + The reduced copy. + """ + return {name: value for name, value in frame.items() if name != key} + + +def _with(frame: dict, **fields) -> dict: + """Return a copy of *frame* with *fields* replaced. + + Parameters: + frame: Parsed RX-log frame. + **fields: Fields to set. + + Returns: + The changed copy. + """ + return {**frame, **fields} + + +UNVERIFIABLE = { + # No raw bytes to check the signature against. + "no-raw-payload": ( + lambda frame: _without(frame, "pkt_payload"), + "no raw payload", + VICTIM_ID, + ), + # Raw bytes in a form the check does not read. + "hex-raw-payload": ( + lambda frame: _with(frame, pkt_payload=frame["pkt_payload"].hex()), + "no raw payload", + VICTIM_ID, + ), + # Raw bytes that end inside the signature. + "cut-in-signature": ( + lambda frame: _with(frame, pkt_payload=frame["pkt_payload"][:99]), + "truncated signature", + VICTIM_ID, + ), + # A parsed key that is not the key the signed bytes carry. + "other-key": ( + lambda frame: _with(frame, adv_key=OTHER_KEY_HEX), + "key mismatch", + "!" + OTHER_KEY_HEX[:8], + ), +} +"""Parsed fields the signed bytes do not back, by case: change, reason, node id.""" + + +@pytest.mark.parametrize("case", sorted(UNVERIFIABLE)) +def test_rx_log_advert_whose_fields_the_signature_does_not_cover_is_not_ingested( + case, posts, logged +): + """An advert is posted only when its signature covers the fields posted: + a frame missing its raw bytes, or whose key differs from theirs, is + dropped, logged, and raises nothing.""" + + change, reason, node_id = UNVERIFIABLE[case] + + _hear(change(_victim_frame(_app_data("Alice")))) + + assert posts == [], f"{case} advert posted {posts}" + assert logged == _dropped(reason, node_id) + + +LIMITS = { + # The most app data the firmware signs: 32 bytes, as parsed. + "app-data-at-32-bytes": lambda frame: frame, + # The same raw bytes handed over as a bytearray rather than bytes. + "bytearray-payload": lambda frame: _with( + frame, pkt_payload=bytearray(frame["pkt_payload"]) + ), +} +"""Genuine adverts at the edges of what the check reads, by case: change.""" + + +@pytest.mark.parametrize("case", sorted(LIMITS)) +def test_rx_log_advert_at_the_limits_of_the_check_is_ingested(case, posts, logged): + """A genuine advert with 32 bytes of app data verifies and posts, whether + its raw bytes arrive as bytes or as a bytearray.""" + + _hear(LIMITS[case](_victim_frame(_app_data(LONG_NAME)))) + + assert [path for path, _payload in posts] == ["/api/nodes", "/api/positions"] + assert posts[0][1][VICTIM_ID]["user"]["longName"] == LONG_NAME + assert [line["message"] for line in logged] == ["MeshCore RX-log advert"] + + +CUT_SHORT = { + "cut-in-key": 20, + "cut-in-signature": 70, + "no-app-data": 100, +} +"""Length a raw advert payload is cut to, by case: inside the 32-byte key, +inside the 64-byte signature, and right after it with no flags byte.""" + + +@pytest.mark.parametrize("case", sorted(CUT_SHORT)) +def test_rx_log_advert_cut_short_is_skipped_without_raising(case, posts): + """An advert cut short parses without a key and is skipped without + raising, as before the check (RF3).""" + + payload = frames.advert_payload(VICTIM, SENT_AT, _app_data("Alice")) + frame = frames.rx_log_advert(payload[: CUT_SHORT[case]], recv_time=SENT_AT) + assert "adv_key" not in frame + + _hear(frame) + + assert posts == [] + + +# --------------------------------------------------------------------------- +# SPEC SG4: no key of small order +# --------------------------------------------------------------------------- + +SMALL_ORDER = { + "zero": "00" * 32, + "one": "01" + "00" * 31, + "order-8-26e8": "26e8958fc2b227b045c3f489f2ef98f0d5dfac05d3c63339b13802886d53fc05", + "order-8-c717": "c7176a703d4dd84fba3c0b760d10670f2a2053fa2c39ccc64ec7fd7792ac037a", + "p-minus-1": "ec" + "ff" * 30 + "7f", + "p": "ed" + "ff" * 30 + "7f", + "p-plus-1": "ee" + "ff" * 30 + "7f", +} +"""libsodium's small-order blocklist (``ge25519_has_small_order``), by case.""" + + +def _sign_bit_set(key: bytes) -> bytes: + """Return *key* with the top bit of byte 31, the sign of x, set. + + Parameters: + key: A 32-byte public-key encoding. + + Returns: + The sign-bit variant. + """ + return key[:31] + bytes([key[31] | 0x80]) + + +SMALL_ORDER_KEYS = { + **{name: bytes.fromhex(key) for name, key in SMALL_ORDER.items()}, + **{ + f"{name}-sign-bit": _sign_bit_set(bytes.fromhex(key)) + for name, key in SMALL_ORDER.items() + }, +} +"""Every listed encoding and its sign-bit variant, by case: 14 keys.""" + +IDENTITY_SIGNATURE = b"\x01" + bytes(31) + bytes(32) +"""The signature ``R = identity, S = 0``, which small-order keys can verify.""" + + +def _small_order_frame(key: bytes) -> dict: + """Return the parsed RX-log copy of an advert naming Mallory under *key*. + + Parameters: + key: The 32 bytes the frame carries as its public key. + + Returns: + The parsed frame, signed with :data:`IDENTITY_SIGNATURE`. + """ + payload = frames.advert_payload_bytes( + key, SENT_AT, IDENTITY_SIGNATURE, _app_data("Mallory") + ) + return frames.rx_log_advert( + payload, recv_time=SENT_AT + 5, snr=9.0, rssi=-80, hops=2 + ) + + +def test_small_order_blocklist_is_every_encoding_of_the_eight_torsion(): + """The blocklist holds libsodium's seven values, and they are exactly the + encodings of the eight points of order dividing 8 that the RFC 8032 + curve constants give, sign bit cleared (SPEC SG4).""" + from data.mesh_ingestor.protocols.meshcore.advert_signature import ( + _SMALL_ORDER_KEYS, + ) + + listed = {bytes.fromhex(key) for key in SMALL_ORDER.values()} + assert _SMALL_ORDER_KEYS == listed == ed25519_oracle.small_order_encodings() + + +@pytest.mark.parametrize("case", sorted(SMALL_ORDER_KEYS)) +def test_rx_log_advert_with_a_small_order_key_is_not_ingested(case, posts, logged): + """A key of small order posts nothing, whichever listed encoding and sign + bit it carries, and logs reason ``small-order key`` (SPEC SG4).""" + + key = SMALL_ORDER_KEYS[case] + + _hear(_small_order_frame(key)) + + assert posts == [], f"{case} advert posted {posts}" + assert logged == _dropped("small-order key", "!" + key[:4].hex()) + + +@pytest.mark.parametrize("case", ["one", "one-sign-bit"]) +def test_rx_log_advert_small_order_key_is_refused_though_openssl_accepts_it( + case, posts, logged +): + """OpenSSL verifies the identity key's trivial signature over any message, + also as ``01 00..00 80``, which the firmware's decoder refuses: without + SG4 both adverts would post.""" + + key = SMALL_ORDER_KEYS[case] + message = frames.advert_signed_message(key, SENT_AT, _app_data("Mallory")) + Ed25519PublicKey.from_public_bytes(key).verify(IDENTITY_SIGNATURE, message) + + _hear(_small_order_frame(key)) + + assert posts == [] + assert logged == _dropped("small-order key", "!01000000") + + +def test_rx_log_advert_with_an_ordinary_key_passes_the_small_order_check(posts, logged): + """An ordinary key passes, even with its sign bit set, as the victim's is: + the masked comparison matches only the seven listed values (SPEC SG4).""" + + assert VICTIM_KEY[31] & 0x80 + + _hear(_victim_frame(_app_data("Alice"))) + + assert [path for path, _payload in posts] == ["/api/nodes", "/api/positions"] + assert [line["message"] for line in logged] == ["MeshCore RX-log advert"] + + +# --------------------------------------------------------------------------- +# SPEC SG5: newer than the last advert accepted for the key +# --------------------------------------------------------------------------- + +REPLAYS = { + # The same advert again, over four repeaters instead of two. + "same-advert-other-path": (SENT_AT, 4, []), + # An older advert of the same node. + "older-advert": (SENT_AT - 60, 2, []), + # A newer advert of the same node. + "newer-advert": (SENT_AT + 60, 2, ["/api/nodes", "/api/positions"]), +} +"""A second advert after one accepted at SENT_AT, by case: time, hops, routes.""" + + +@pytest.mark.parametrize("case", sorted(REPLAYS)) +def test_replay_check_needs_a_timestamp_newer_than_the_last_accepted( + case, posts, logged +): + """After one advert is accepted, a copy of it over another path or an + older advert posts nothing and logs reason ``replayed advert``; a newer + advert posts as usual (SPEC SG5).""" + + timestamp, hops, routes = REPLAYS[case] + _hear(_victim_frame(_app_data("Alice"))) + posts.clear() + logged.clear() + + _hear(_victim_frame(_app_data("Alice"), timestamp=timestamp, hops=hops)) + + assert [path for path, _payload in posts] == routes + if routes: + assert [line["message"] for line in logged] == ["MeshCore RX-log advert"] + else: + assert logged == _dropped("replayed advert") + + +def test_replay_check_keeps_the_first_receptions_signal(posts, logged): + """Of two copies of one advert only the first reaches the web, so the + node keeps that copy's SNR, RSSI and hop count (SPEC SG5).""" + + _hear(_victim_frame(_app_data("Alice"), hops=1, snr=9.0, rssi=-80)) + _hear(_victim_frame(_app_data("Alice"), hops=5, snr=-3.5, rssi=-118)) + + nodes = [payload[VICTIM_ID] for path, payload in posts if path == "/api/nodes"] + assert [(n["snr"], n["rssi"], n["hopsAway"]) for n in nodes] == [(9.0, -80, 1)] + assert logged[-1] == _dropped("replayed advert")[0] + + +def test_refused_rx_log_advert_leaves_the_replay_memory_alone(posts, logged): + """Only a verified advert is remembered: a forged one with a far-future + timestamp cannot refuse the genuine advert that follows (SPEC SG5).""" + + _hear( + _victim_frame( + _app_data("Mallory"), timestamp=SENT_AT + 10**8, signature=bytes(64) + ) + ) + _hear(_victim_frame(_app_data("Alice"))) + + assert [path for path, _payload in posts] == ["/api/nodes", "/api/positions"] + assert [line.get("reason") for line in logged] == ["invalid signature", None] + + +def test_replay_memory_is_empty_after_a_restart(posts, logged, replay): + """The memory lives in the process: after a restart the first advert of a + key heard only in the RX log passes again, and a repeat of it is refused + again (SPEC SG5). A roster contact's key is reloaded by the startup + roster fetch instead (:func:`test_replay_memory_takes_a_listed_contacts_last_advert`). + """ + + frame = _victim_frame(_app_data("Alice")) + _hear(frame) + replay._reset_replay_memory() # what a restart does + _hear(frame) + _hear(frame) + + assert [path for path, _payload in posts] == ["/api/nodes", "/api/positions"] * 2 + assert [line.get("reason") for line in logged] == [None, None, "replayed advert"] + + +def test_replay_memory_holds_4096_keys(replay): + """The RX-log cap is pinned: more than ten times the largest companion + roster.""" + + assert replay._REPLAY_MEMORY_CAP == 4096 + + +def test_replay_memory_holds_1024_roster_keys(replay): + """The roster cap is pinned: about three times the largest companion + roster.""" + + assert replay._ROSTER_MEMORY_CAP == 1024 + + +def test_replay_memory_forgets_the_least_recently_accepted_key_past_its_cap( + monkeypatch, replay +): + """Past its cap the RX-log pool forgets the least recently accepted key, + whose next advert passes as after a restart (SPEC SG5).""" + + monkeypatch.setattr(replay, "_REPLAY_MEMORY_CAP", 2) + accept = replay._accept_advert_timestamp + + assert accept("aa", 10) and accept("bb", 10) + assert accept("aa", 11) # aa is now the most recently accepted + assert accept("cc", 10) # over the cap: bb, the least recent, is forgotten + assert accept("bb", 10) # bb passes again, and aa is forgotten in turn + assert not accept("cc", 10) + assert accept("aa", 5) + + +def test_replay_memory_backstop_evicts_only_keys_the_full_listing_lacks( + monkeypatch, replay +): + """Past its backstop cap the roster pool hands the least recently + refreshed key that the latest full listing lacks to the RX-log pool, and + never a key of that listing, even when the listing alone exceeds the cap; + a key that leaves keeps its timestamp (SPEC SG5).""" + + monkeypatch.setattr(replay, "_ROSTER_MEMORY_CAP", 2) + listing = replay._take_roster_listing + + listing({"aa": 10}, full=True) + listing({"bb": 10}, full=False) + listing({"cc": 10}, full=False) # over the cap: bb leaves, listed aa stays + assert list(replay._roster) == ["aa", "cc"] + assert dict(replay._heard) == {"bb": 10} + listing({"dd": 10, "ee": 10, "ff": 10}, full=True) # larger than the cap + assert list(replay._roster) == ["dd", "ee", "ff"] + assert not replay._accept_advert_timestamp("aa", 10) # left, time kept + assert not replay._accept_advert_timestamp("dd", 10) + + +def _connect() -> dict: + """Return the event-handler map of a new connection, as the runner builds it. + + Returns: + The handler map; its first ``CONTACTS`` listing counts as the full roster. + """ + return _make_event_handlers(_MeshcoreInterface(target=None), "/dev/ttyUSB0") + + +def _contact( + last_advert: object, key_hex: str = VICTIM_KEY_HEX, name: str = "Alice" +) -> dict: + """Return a roster contact as the meshcore library reports it. + + Parameters: + last_advert: The contact's ``last_advert`` field. + key_hex: The contact's public key as hex. + name: The contact's advertised name. + + Returns: + The contact dict. + """ + return { + "public_key": key_hex, + "type": frames.ADV_TYPE_CHAT, + "adv_name": name, + "last_advert": last_advert, + } + + +def _list_contacts(hmap: dict, *contacts: dict) -> None: + """Deliver one ``CONTACTS`` listing of *contacts* to a connection's handlers. + + Parameters: + hmap: The connection's handler map from :func:`_connect`. + *contacts: The listed contacts. + """ + payload = {contact["public_key"]: contact for contact in contacts} + asyncio.run(hmap["CONTACTS"](types.SimpleNamespace(payload=payload))) + + +def _push(hmap: dict, event: str, contact: dict) -> None: + """Deliver one ``NEW_CONTACT`` or ``NEXT_CONTACT`` event to a connection. + + Parameters: + hmap: The connection's handler map from :func:`_connect`. + event: ``"NEW_CONTACT"`` or ``"NEXT_CONTACT"``. + contact: The contact the event carries. + """ + asyncio.run(hmap[event](types.SimpleNamespace(payload=contact))) + + +LISTED = { + # The connection's first listing: the radio's whole roster. + "first-listing": lambda hmap: _list_contacts(hmap, _contact(SENT_AT)), + # A later listing, the auto-update re-fetch of changed contacts. + "later-listing": lambda hmap: ( + _list_contacts(hmap, _contact(SENT_AT, OTHER_KEY_HEX, "Bob")), + _list_contacts(hmap, _contact(SENT_AT)), + ), +} +"""Ways the radio lists the victim as a roster contact, by case.""" + + +@pytest.mark.parametrize("case", sorted(LISTED)) +def test_replay_memory_takes_a_listed_contacts_last_advert(case, posts, logged): + """A contact the radio lists, in a connection's first listing or a later + one, refuses an RX-log copy of its last advert, also right after a + restart, and lets a newer advert through (SPEC SG5).""" + + LISTED[case](_connect()) + posts.clear() + + _hear(_victim_frame(_app_data("Alice"))) + _hear(_victim_frame(_app_data("Alice"), timestamp=SENT_AT + 60)) + + assert [path for path, _payload in posts] == ["/api/nodes", "/api/positions"] + assert [line.get("reason") for line in logged] == ["replayed advert", None] + + +def test_replay_memory_takes_no_key_from_a_new_contact_push(posts, replay): + """The radio pushes ``NEW_CONTACT`` for an advert from a key it did not + add to its roster, so the push adds nothing to the roster pool, and that + key's RX-log advert, new to the memory, posts (SPEC SG5).""" + + hmap = _connect() + _list_contacts(hmap) # the connection's first listing: an empty roster + _push(hmap, "NEW_CONTACT", _contact(SENT_AT)) + posts.clear() + + _hear(_victim_frame(_app_data("Alice"))) + + assert VICTIM_KEY_HEX not in replay._roster + assert [path for path, _payload in posts] == ["/api/nodes", "/api/positions"] + + +def test_replay_memory_keeps_a_roster_key_through_a_new_contact_flood( + posts, logged, replay +): + """1025 new keys, each delivered as the radio delivers an advert it does + not add (``RX_LOG_DATA``, then ``NEW_CONTACT``), leave a listed contact's + key in the roster pool, so a replay of its advert is refused (SPEC SG5).""" + + hmap = _connect() + _list_contacts(hmap, _contact(SENT_AT)) + for n in range(replay._ROSTER_MEMORY_CAP + 1): + key = frames.advert_key(f"potato-mesh test advertiser: flood {n}") + payload = frames.advert_payload(key, SENT_AT, _app_data(f"Flood {n}")) + _hear(frames.rx_log_advert(payload, recv_time=SENT_AT + 5)) + key_hex = frames.advert_public_key(key).hex() + _push(hmap, "NEW_CONTACT", _contact(SENT_AT, key_hex, f"Flood {n}")) + posts.clear() + logged.clear() + + _hear(_victim_frame(_app_data("Alice"))) + + assert posts == [], f"replay posted {posts}" + assert logged == _dropped("replayed advert") + # The flood keys stayed in the RX-log pool; the roster pool holds the victim. + assert list(replay._roster) == [VICTIM_KEY_HEX] + assert len(replay._heard) == replay._ROSTER_MEMORY_CAP + 1 + + +def test_replay_memory_drops_a_key_the_next_full_listing_lacks(posts, logged, replay): + """A connection's first listing replaces the membership: a key it lacks + leaves the roster pool and becomes an ordinary RX-log-pool entry that + keeps its timestamp. A later listing that omits a member keeps it, as + such a listing names only the contacts changed since (SPEC SG5).""" + + first = _connect() + _list_contacts(first, _contact(SENT_AT), _contact(SENT_AT, OTHER_KEY_HEX, "Bob")) + _list_contacts(first, _contact(SENT_AT, THIRD_KEY_HEX, "Carol")) + assert set(replay._roster) == {VICTIM_KEY_HEX, OTHER_KEY_HEX, THIRD_KEY_HEX} + + second = _connect() # a reconnect: its first listing is the whole roster + _list_contacts(second, _contact(SENT_AT, OTHER_KEY_HEX, "Bob")) + assert set(replay._roster) == {OTHER_KEY_HEX} + assert replay._heard[VICTIM_KEY_HEX] == SENT_AT + posts.clear() + + _hear(_victim_frame(_app_data("Alice"))) + + assert posts == [] + assert logged == _dropped("replayed advert") + + +def test_replay_memory_timestamps_only_rise(posts, logged, replay): + """Listings, ``NEW_CONTACT``, ``NEXT_CONTACT`` and accepted RX-log adverts + only ever raise a key's remembered timestamp (SPEC SG5).""" + + hmap = _connect() + _list_contacts(hmap, _contact(SENT_AT + 60)) + _push(hmap, "NEXT_CONTACT", _contact(SENT_AT)) # older: no change + _list_contacts(hmap, _contact(SENT_AT)) # an older later listing: no change + assert replay._roster[VICTIM_KEY_HEX] == SENT_AT + 60 + _push(hmap, "NEW_CONTACT", _contact(SENT_AT + 120)) # newer: raised + assert replay._roster[VICTIM_KEY_HEX] == SENT_AT + 120 + + _hear(_victim_frame(_app_data("Alice"), timestamp=SENT_AT + 90)) + _hear(_victim_frame(_app_data("Alice"), timestamp=SENT_AT + 180)) + _list_contacts(_connect(), _contact(SENT_AT + 120)) # the next connection + + assert replay._roster[VICTIM_KEY_HEX] == SENT_AT + 180 + assert [line.get("reason") for line in logged] == ["replayed advert", None] + + +@pytest.mark.parametrize("last_advert", [None, "1891288000"], ids=["none", "string"]) +def test_replay_memory_ignores_a_listed_contact_without_an_integer_time( + last_advert, posts, replay +): + """A listed contact without an integer ``last_advert`` joins nothing and + a contact event carrying one raises nothing, so the next RX-log advert + passes and nothing raises (SPEC SG5).""" + + hmap = _connect() + _list_contacts(hmap, _contact(last_advert)) + _push(hmap, "NEXT_CONTACT", _contact(last_advert)) + posts.clear() + + _hear(_victim_frame(_app_data("Alice"))) + + assert VICTIM_KEY_HEX not in replay._roster + assert [path for path, _payload in posts] == ["/api/nodes", "/api/positions"] + + +FLOODED = { + # Listed by the radio: the flood cannot push the key out, so a replay of + # the advert is refused. + "roster-contact": (lambda: _list_contacts(_connect(), _contact(SENT_AT)), []), + # Heard only in the RX log: the flood pushes the key out, so a replay of + # the advert posts once (the limit SPEC SG5 states). + "rx-log-only": ( + lambda: _hear(_victim_frame(_app_data("Alice"))), + ["/api/nodes", "/api/positions"], + ), +} +"""How the victim's advert at SENT_AT was remembered, by case: how, and the +routes a replay of it posts after the flood.""" + + +@pytest.mark.parametrize("case", sorted(FLOODED)) +def test_replay_memory_through_a_flood_of_adverts_from_new_keys( + case, posts, logged, replay +): + """4097 verified adverts from new keys overflow the RX-log pool: a key the + radio listed stays, so a replay of its advert is refused, while a silent + node known only from the RX log is forgotten and a replay of its advert + posts (SPEC SG5). The flood goes straight to the memory, as the handler + hands it each verified advert.""" + + remember, routes = FLOODED[case] + remember() + for n in range(replay._REPLAY_MEMORY_CAP + 1): + assert replay._accept_advert_timestamp(f"{n:064x}", SENT_AT) + posts.clear() + logged.clear() + + _hear(_victim_frame(_app_data("Alice"))) + + assert [path for path, _payload in posts] == routes + if routes: + assert [line["message"] for line in logged] == ["MeshCore RX-log advert"] + else: + assert logged == _dropped("replayed advert") diff --git a/tests/test_provider_unit.py b/tests/test_provider_unit.py index bc10e2dd..ee84f571 100644 --- a/tests/test_provider_unit.py +++ b/tests/test_provider_unit.py @@ -47,6 +47,8 @@ REPO_ROOT = Path(__file__).resolve().parents[1] if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) +import meshcore_frames as frames # noqa: E402 - pytest puts tests/ on sys.path + from data.mesh_ingestor import daemon # noqa: E402 - path setup from data.mesh_ingestor.mesh_protocol import MeshProtocol # noqa: E402 - path setup from data.mesh_ingestor.protocols.meshtastic import ( # noqa: E402 - path setup @@ -2807,6 +2809,45 @@ def test_rx_advert_to_node_dict_minimal_frame(): assert isinstance(node["lastHeard"], int) +def _signed_rx_advert( + seed, app_data, *, adv_timestamp, recv_time, snr=None, rssi=None, hops=0 +): + """Return a parsed RX-log ADVERT signed with the test key derived from *seed*. + + The ingestor posts an RX-log advert only when its signature holds (SPEC + SG1), so every advert fixture is a real on-air payload signed with a key + derived from a fixed seed when the test runs; no private key is stored. + + Parameters: + seed: Fixed label the advertiser's key is derived from. + app_data: Encoded app data (:func:`meshcore_frames.advert_app_data`). + adv_timestamp: Sender-side advert time the payload signs. + recv_time: Receiver-side reception time. + snr: Reception SNR in dB, or ``None``. + rssi: Reception RSSI in dBm, or ``None``. + hops: Repeaters the copy travelled. + + Returns: + The frame as the pinned library's packet parser reports it. + """ + payload = frames.advert_payload(frames.advert_key(seed), adv_timestamp, app_data) + return frames.rx_log_advert( + payload, recv_time=recv_time, snr=snr, rssi=rssi, hops=hops + ) + + +_BER_SEED = "potato-mesh test advertiser: BER Drachentoeter" +"""Seed of the repeater key the BER Drachentoeter advert fixtures sign with.""" + +_BER_APP_DATA = frames.advert_app_data( + frames.ADV_TYPE_REPEATER, + name="BER Drachentoeter", + lat_e6=52_516_274, + lon_e6=13_405_612, +) +"""App data of the BER Drachentoeter repeater: name, role and position.""" + + def test_on_rx_log_data_advert_upserts_node_and_position(monkeypatch): """An RX-log ADVERT frame upserts the full node and stores its position.""" import asyncio @@ -2820,34 +2861,26 @@ def test_on_rx_log_data_advert_upserts_node_and_position(monkeypatch): lambda *args: positions.append(args), ) - pub_key = "511617e3" + "00" * 28 - asyncio.run( - hmap["RX_LOG_DATA"]( - _FakeEvt( - { - "payload_typename": "ADVERT", - "adv_key": pub_key, - "adv_name": "BER Drachentoeter", - "adv_type": 2, - "adv_lat": 52.516274, - "adv_lon": 13.405612, - "recv_time": 1_758_000_021, - "snr": 11.5, - "rssi": -70, - "path_len": 3, - } - ) - ) + frame = _signed_rx_advert( + _BER_SEED, + _BER_APP_DATA, + adv_timestamp=1_758_000_000, + recv_time=1_758_000_021, + snr=11.5, + rssi=-70, + hops=3, ) + asyncio.run(hmap["RX_LOG_DATA"](_FakeEvt(frame))) + expected_id = frames.advert_node_id(frames.advert_key(_BER_SEED)) assert captured == [] # adverts never produce message packets assert len(upserted) == 1 node_id, node = upserted[0] - assert node_id == "!511617e3" + assert node_id == expected_id assert node["user"]["longName"] == "BER Drachentoeter" assert node["snr"] == 11.5 and node["rssi"] == -70 and node["hopsAway"] == 3 assert len(positions) == 1 - assert positions[0][0] == "!511617e3" + assert positions[0][0] == expected_id assert positions[0][1] == 52.516274 and positions[0][2] == 13.405612 @@ -2867,23 +2900,14 @@ def test_on_rx_log_data_advert_position_rx_time_is_now(monkeypatch): lambda route, payload, **_k: posted.append((route, payload)), ) - before = int(_time.time()) - asyncio.run( - hmap["RX_LOG_DATA"]( - _FakeEvt( - { - "payload_typename": "ADVERT", - "adv_key": "511617e3" + "00" * 28, - "adv_name": "BER Drachentoeter", - "adv_type": 2, - "adv_lat": 52.516274, - "adv_lon": 13.405612, - "adv_timestamp": 1_758_000_000, - "recv_time": 1_758_000_021, - } - ) - ) + frame = _signed_rx_advert( + _BER_SEED, + _BER_APP_DATA, + adv_timestamp=1_758_000_000, + recv_time=1_758_000_021, ) + before = int(_time.time()) + asyncio.run(hmap["RX_LOG_DATA"](_FakeEvt(frame))) after = int(_time.time()) pos = [p for r, p in posted if r == "/api/positions"][0] @@ -2904,17 +2928,14 @@ def test_on_rx_log_data_advert_without_position_skips_position_store(monkeypatch lambda *args: positions.append(args), ) - asyncio.run( - hmap["RX_LOG_DATA"]( - _FakeEvt( - { - "payload_typename": "ADVERT", - "adv_key": "aabbccdd" + "00" * 28, - "snr": 4.0, - } - ) - ) + frame = _signed_rx_advert( + "potato-mesh test advertiser: unnamed", + frames.advert_app_data(), + adv_timestamp=1_758_000_000, + recv_time=1_758_000_022, + snr=4.0, ) + asyncio.run(hmap["RX_LOG_DATA"](_FakeEvt(frame))) assert len(upserted) == 1 assert positions == [] @@ -2989,23 +3010,31 @@ def test_on_rx_log_data_malformed_advert_tolerated(monkeypatch): # --------------------------------------------------------------------------- -def _rx_advert_frame(**overrides): - """Build a full RX-log ADVERT frame with sender-side ``adv_timestamp``.""" - frame = { - "payload_typename": "ADVERT", - "adv_key": "ae46e493" + "00" * 28, - "adv_name": "Flood Node", - "adv_type": 1, - "adv_lat": 52.498481, - "adv_lon": 13.475442, - "adv_timestamp": 1_784_803_284, - "recv_time": 1_784_803_284, - "snr": -11.5, - "rssi": -124, - "path_len": 5, - } - frame.update(overrides) - return frame +def _rx_advert_frame(*, adv_timestamp=1_784_803_284, recv_time=1_784_803_284): + """Build a full, signed RX-log ADVERT frame with sender-side ``adv_timestamp``. + + Parameters: + adv_timestamp: Sender-side advert time the payload signs. + recv_time: Receiver-side reception time of this copy. + + Returns: + The parsed frame of a five-hop flood copy. + """ + app_data = frames.advert_app_data( + frames.ADV_TYPE_CHAT, + name="Flood Node", + lat_e6=52_498_481, + lon_e6=13_475_442, + ) + return _signed_rx_advert( + "potato-mesh test advertiser: Flood Node", + app_data, + adv_timestamp=adv_timestamp, + recv_time=recv_time, + snr=-11.5, + rssi=-124, + hops=5, + ) def test_on_rx_log_data_advert_position_time_uses_sender_adv_timestamp(monkeypatch): @@ -3013,10 +3042,13 @@ def test_on_rx_log_data_advert_position_time_uses_sender_adv_timestamp(monkeypat Every flood copy of one advert carries the same ``adv_timestamp`` but its own receiver-side ``recv_time``; keying the position on ``recv_time`` mints - one row per copy (and per ingestor) instead of deduplicating (MR-A3). + one row per copy (and per ingestor) instead of deduplicating (MR-A3). One + ingestor posts only the first copy it hears (SPEC SG5), so here four + ingestors each hear one copy, each with its own empty replay memory. """ import asyncio import data.mesh_ingestor.protocols.meshcore.handlers as _handlers_mod + from data.mesh_ingestor.protocols.meshcore import advert_replay _captured, _upserted, _iface, hmap = _setup_channel_msg_handlers(monkeypatch) positions: list = [] @@ -3027,6 +3059,7 @@ def test_on_rx_log_data_advert_position_time_uses_sender_adv_timestamp(monkeypat ) for offset in (0, 3, 5, 17): + advert_replay._reset_replay_memory() # the next ingestor's process asyncio.run( hmap["RX_LOG_DATA"]( _FakeEvt(_rx_advert_frame(recv_time=1_784_803_284 + offset)) @@ -3040,9 +3073,15 @@ def test_on_rx_log_data_advert_position_time_uses_sender_adv_timestamp(monkeypat def test_on_rx_log_data_advert_flood_copies_collapse_to_one_position_id(monkeypatch): - """Four flood copies of one advert must queue a single position identity.""" + """Four flood copies of one advert must queue a single position identity. + + One ingestor posts only the first copy it hears (SPEC SG5), so here four + ingestors each hear one copy, each with its own empty replay memory, and + their four position posts must share one id. + """ import asyncio import data.mesh_ingestor.protocols.meshcore as _mod + from data.mesh_ingestor.protocols.meshcore import advert_replay _captured, _upserted, _iface, hmap = _setup_channel_msg_handlers(monkeypatch) posted: list = [] @@ -3053,20 +3092,23 @@ def test_on_rx_log_data_advert_flood_copies_collapse_to_one_position_id(monkeypa ) for offset in (0, 3, 5, 17): + advert_replay._reset_replay_memory() # the next ingestor's process asyncio.run( hmap["RX_LOG_DATA"]( _FakeEvt(_rx_advert_frame(recv_time=1_784_803_284 + offset)) ) ) - position_ids = {p["id"] for route, p in posted if route == "/api/positions"} - assert len(position_ids) == 1 + position_posts = [p for route, p in posted if route == "/api/positions"] + assert len(position_posts) == 4 + assert len({p["id"] for p in position_posts}) == 1 def test_on_rx_log_data_advert_position_falls_back_to_recv_time(monkeypatch): """Absent or zero ``adv_timestamp`` degrades to the receiver-side time.""" import asyncio import data.mesh_ingestor.protocols.meshcore.handlers as _handlers_mod + from data.mesh_ingestor.protocols.meshcore import advert_replay _captured, _upserted, _iface, hmap = _setup_channel_msg_handlers(monkeypatch) positions: list = [] @@ -3079,6 +3121,10 @@ def test_on_rx_log_data_advert_position_falls_back_to_recv_time(monkeypatch): frame = _rx_advert_frame(recv_time=1_784_803_300) del frame["adv_timestamp"] asyncio.run(hmap["RX_LOG_DATA"](_FakeEvt(frame))) + # The zero-timestamp advert is older than the first (signed 1_784_803_284), + # so the same ingestor would refuse it (SPEC SG5); another ingestor, with + # its own replay memory, hears it. + advert_replay._reset_replay_memory() asyncio.run( hmap["RX_LOG_DATA"]( _FakeEvt(_rx_advert_frame(adv_timestamp=0, recv_time=1_784_803_301))