diff --git a/meshterm/core/frames.py b/meshterm/core/frames.py index 088c65f..e7bc401 100644 --- a/meshterm/core/frames.py +++ b/meshterm/core/frames.py @@ -97,6 +97,9 @@ def frame_addressing(payload: Mapping[str, Any]) -> dict[str, str]: * ``dest_hash`` — the recipient's one-byte key hash (every addressed class). * ``src_hash`` — the sender's one-byte key hash (the two-hash classes). + * ``cipher_mac`` — the two-byte MAC that follows them: a tag over the encrypted + message, and so a fingerprint identifying *which* message a frame carries without + being able to read it. * ``src_key`` — the sender's *whole* public key, which an anonymous request carries instead of a hash (it has no shared secret to be recognised by yet). * ``ack_crc`` — an ack's four-byte checksum of the message it acknowledges, hex in @@ -137,6 +140,13 @@ def frame_addressing(payload: Mapping[str, Any]) -> dict[str, str]: return { "dest_hash": body[:hash_w].hex(), "src_hash": body[hash_w : hash_w * 2].hex(), + # The MAC an addressed frame carries is a tag over *this* message's plaintext, + # so two frames sharing one (between the same pair) are copies of the same + # message — a fingerprint that needs no key to compare. It is what lets the + # message-paths view group a direct message's own retransmissions and tell them + # from the next message's, which content-matching does for a channel frame we + # can decrypt (see :mod:`~meshterm.services.message_paths`). + "cipher_mac": body[hash_w * 2 : hash_w * 2 + _MAC].hex(), } if typename == "ANON_REQ": if len(body) < hash_w + _KEY + _MAC: diff --git a/meshterm/persistence/repository.py b/meshterm/persistence/repository.py index 47a761d..b124c03 100644 --- a/meshterm/persistence/repository.py +++ b/meshterm/persistence/repository.py @@ -14,7 +14,7 @@ from datetime import datetime, timedelta from pathlib import Path from typing import TYPE_CHECKING, Any, Optional, Sequence -from ..core.frames import CHANNEL_CLASSES, ENDPOINT_HASH_BYTES +from ..core.frames import ADDRESSED_CLASSES, CHANNEL_CLASSES, ENDPOINT_HASH_BYTES from ..core.models import ( PATH_TRACE_TARGET, ChatMessage, @@ -88,11 +88,12 @@ def _packet_raw(row: sqlite3.Row) -> Optional[dict]: """ typename = _row_value(row, "payload_typename") chan_hash = row["chan_hash"] + cipher_mac = _row_value(row, "cipher_mac") dest = _row_value(row, "dest") src = _row_value(row, "src") tag = _row_value(row, "tag") trace_snrs = _row_value(row, "trace_snrs") - if not any((typename, chan_hash, dest, src, tag, trace_snrs)): + if not any((typename, chan_hash, cipher_mac, dest, src, tag, trace_snrs)): return None raw: dict = {} if typename: @@ -102,6 +103,10 @@ def _packet_raw(row: sqlite3.Row) -> Optional[dict]: raw["chan_hash"] = chan_hash raw["cipher_mac"] = row["cipher_mac"] raw["crypted"] = row["crypted"] + if cipher_mac and not chan_hash: + # An addressed frame keeps only its MAC (no channel envelope), so it is restored + # here rather than in the chan_hash branch above. + raw["cipher_mac"] = cipher_mac if dest: raw["dest_hash"] = dest if src: @@ -114,6 +119,16 @@ def _packet_raw(row: sqlite3.Row) -> Optional[dict]: return raw +def _as_when(iso: Any) -> Optional[datetime]: + """Parse a stored ISO timestamp, or ``None`` when absent/unparseable.""" + if not iso: + return None + try: + return datetime.fromisoformat(str(iso)) + except (TypeError, ValueError): + return None + + def _row_value(row: sqlite3.Row, key: str) -> Any: """Read ``key`` from a row, tolerating a query that didn't select it (returns ``None``).""" try: @@ -992,6 +1007,13 @@ class Repository: chan_hash = raw.get("chan_hash") cipher_mac = raw.get("cipher_mac") crypted = raw.get("crypted") + elif typename in ADDRESSED_CLASSES: + # An addressed frame's MAC, in the same column its channel sibling uses: a tag + # over the encrypted body, so copies of one message share it and the next + # message's do not. The ciphertext itself is *not* kept — we could never read + # it, and only the fingerprint is needed to tell one message's frames apart + # from another's (see :mod:`~meshterm.services.message_paths`). + cipher_mac = raw.get("cipher_mac") # One column each for the three shapes of addressing: the sender is a hash or a # whole key (its length says which), the token an ack's checksum or a trace's tag # (the payload class says which) — so neither needs a column of its own. @@ -1929,6 +1951,56 @@ class Repository: ).fetchall() return {row["peer"]: datetime.fromisoformat(row["last"]) for row in rows} + def direct_message_bounds( + self, peer: Optional[str], when: datetime, *, outbound: bool + ) -> "tuple[Optional[datetime], Optional[datetime]]": + """The times of the direct messages either side of ``when`` in one conversation. + + What bounds the message-paths search for a direct message (see + :mod:`~meshterm.services.message_paths`). A direct frame is encrypted, so the log + cannot say which message it carried; a fixed window around the message is therefore + the only bound available — and at conversational pace it is far too wide. Nine + messages of one real exchange fell inside a single ±90 s window, so every one of + them listed all nine messages' frames as its own. + + The messages around it are the natural edges: a frame logged after the *next* + message was composed belongs to that one, not this one. The caller clamps its window + to the midpoint of each gap, so the bound tightens exactly as the conversation + speeds up. + + Only messages travelling the **same way** count as neighbours, because the two + directions are stamped by two different clocks: our own sends carry ours, taken as + they leave, while a received message carries the *sender's*, which may drift from + ours by minutes. Ordering an inbound message against an outbound one therefore + compares two clocks and can bound a message by an edge that, in its own clock, + hasn't happened yet — which is exactly how the first cut of this clamp gave a + received message an empty window. Within one direction the stamps are consistent, + so the neighbours mean what they say. + + Args: + peer: The contact's key prefix, as :meth:`record_chat_message` stores it. + when: The message's own time. + outbound: The direction to look along — the message's own. + + Returns: + ``(previous, next)`` same-direction message times, either of which is ``None`` + when this message is the first or last of its side of the conversation. + """ + key = (peer or "").lower() + args = (key, int(outbound), when.isoformat()) + row = self._conn.execute( + "SELECT MAX(created_at) AS edge FROM messages " + "WHERE is_channel = 0 AND peer = ? AND outbound = ? AND created_at < ?", + args, + ).fetchone() + prev = _as_when(row["edge"] if row else None) + row = self._conn.execute( + "SELECT MIN(created_at) AS edge FROM messages " + "WHERE is_channel = 0 AND peer = ? AND outbound = ? AND created_at > ?", + args, + ).fetchone() + return prev, _as_when(row["edge"] if row else None) + def channel_stats(self) -> dict[str, ChannelStats]: """Aggregate stored channel messages into per-channel statistics. diff --git a/meshterm/services/message_paths.py b/meshterm/services/message_paths.py index 3daf8e8..c38fd6a 100644 --- a/meshterm/services/message_paths.py +++ b/meshterm/services/message_paths.py @@ -14,21 +14,28 @@ chat message so the chat screen can show, on demand, every way the message reach ``Name: `` sender prefix convention on either side) is one arrival of that message — our own broadcasts included, since a repeater's rebroadcast of us is overheard and logged like anything else. -* **Direct messages** ride ECDH-encrypted ``TEXT_MSG`` frames that only the recipient - can decrypt, so no content match is possible from the log. What such a frame does - carry in the clear is its *addressing* — the one-byte key hash of each end - (:mod:`~meshterm.core.frames`) — so the window narrows to the frames travelling - between this conversation's two ends, in either direction, and what is left is billed - as matched by address and time. Honest evidence, not a claim: a one-byte hash - collides, and a frame between the right pair in the right ninety seconds is still - only *probably* this message. +* **Direct messages** ride ECDH-encrypted ``TEXT_MSG`` frames that only the recipient can + decrypt, so no *content* match is possible from the log. What such a frame carries in + the clear is its **addressing and its MAC** (:mod:`~meshterm.core.frames`): the one-byte + key hash of each end, then a two-byte tag over the encrypted body. That MAC is a + fingerprint of the message itself — copies of one message share it and the next message's + do not — so direct frames group as exactly as channel frames do, without ever reading + them. Three filters, in order: the frame ran **between this conversation's two ends**, + in the **direction this message travelled** (our own sends are ``us → peer``; a received + message is ``peer → us``), and it carries **the same MAC** as the rest of its group. + + A conversation's messages are then told apart by which MAC group sits nearest the + message's own time. That leaves one honest gap: history recorded before the MAC was kept + has none, and those messages fall back to matching by address, direction and time — + billed as such, because a frame between the right pair in the right window is only + *probably* this one. Nothing here transmits; it is a read-model over the repository. """ from __future__ import annotations -from dataclasses import dataclass +from dataclasses import dataclass, replace from datetime import datetime, timedelta from typing import TYPE_CHECKING, Optional, Sequence @@ -43,10 +50,17 @@ if TYPE_CHECKING: #: message is stamped with the *sender's* clock, which may drift from ours by minutes. _CHANNEL_WINDOW = timedelta(minutes=15) -#: How far around a direct message the log is searched. Tight, because the frame's -#: addressing and the clock are the only evidence tying it to the message. +#: How far around a direct message the log is searched, before the conversation's own +#: messages narrow it further (see :func:`direct_window`). Generous on its own: a send is +#: retried for a while after it is composed, and an inbound message is stamped with the +#: sender's clock. _DIRECT_WINDOW = timedelta(seconds=90) +#: The least room left either side of a direct message once its neighbours clamp the +#: window. Two messages composed a second apart still each need somewhere to look, and the +#: midpoint between them would otherwise leave one of them nothing at all. +_MIN_HALF_WINDOW = timedelta(seconds=2) + #: Frame classes that carry a direct (addressed) text message — the payload class the #: meshcore library names, spelled exactly as it reports it (see #: :data:`~meshterm.core.frames.ADDRESSED_CLASSES`, the packet viewer's lane, the @@ -69,12 +83,19 @@ class Arrival: snr: Reception SNR in dB — of the *last relay*, as with every packet row. resend: The sender's resend counter for this copy (0 = the original send), recovered from a decrypted channel frame; always 0 for direct frames. + copies: How many logged frames this row stands for — always ``1`` as the matchers + produce them, and more once :func:`collapse` folds a path heard several times + into one row. A direct send is retried, and each retry is overheard off every + repeater in earshot, so one message routinely leaves a dozen frames on two + paths; listing them individually said "twelve arrivals" where the truth was + "two paths, six times each". """ when: datetime hops: tuple[str, ...] snr: Optional[float] resend: int = 0 + copies: int = 1 def _frame_hops(observation: Observation) -> tuple[str, ...]: @@ -154,51 +175,178 @@ def _endpoint_hash(key: Optional[str]) -> str: return text[:_HASH_CHARS] if len(text) >= _HASH_CHARS else "" -def direct_frames_near( +def direct_window( + repo: "Repository", message: ChatMessage +) -> tuple[datetime, datetime]: + """The span of log to search for one direct message's frames. + + :data:`_DIRECT_WINDOW` either side, then clamped by the conversation's own messages — + only those travelling the **same way**, because the two directions are stamped by two + different clocks (see :meth:`~meshterm.persistence.repository.Repository + .direct_message_bounds`). Without any clamp the flat window is wildly too wide at + conversational pace: nine messages of one real exchange fell inside a single ±90 s + window, so each of them claimed all nine messages' frames as its own. + + The two directions are clamped differently, because we know very different things about + them: + + * **Our own sends** are stamped as they leave, by our own clock, so no frame of one can + predate it — the window opens *at* the message (less a moment for jitter) and runs to + the next send. Retries follow the send, so all the room goes forward. + * **A received message** carries the sender's clock, which may sit either side of when + we actually heard it. Its window stays centred, reaching halfway to the messages + before and after it. + + Never narrower than :data:`_MIN_HALF_WINDOW` either side, so two messages a second apart + still each have somewhere to look. + + Args: + repo: The repository holding the messages. + message: The message to bound a search around. + + Returns: + ``(start, end)`` for :meth:`~meshterm.persistence.repository.Repository + .packet_frames_between`. + """ + at = message.created_at + start, end = at - _DIRECT_WINDOW, at + _DIRECT_WINDOW + previous, following = repo.direct_message_bounds( + message.peer, at, outbound=message.outbound + ) + if message.outbound: + # Our clock, so nothing of this send predates it; the whole window goes forward, + # up to the next send. + start = at - _MIN_HALF_WINDOW + if following is not None: + end = min(end, max(at + _MIN_HALF_WINDOW, following)) + else: + if previous is not None: + start = max(start, min(at - _MIN_HALF_WINDOW, previous + (at - previous) / 2)) + if following is not None: + end = min(end, max(at + _MIN_HALF_WINDOW, at + (following - at) / 2)) + return start, end + + +def _addressed(raw: dict) -> tuple[str, str]: + """A frame's ``(source, destination)`` endpoint hashes, lowercased (empty when absent).""" + return ( + str(raw.get("src_hash") or "").lower(), + str(raw.get("dest_hash") or "").lower(), + ) + + +def _pick_group( + groups: "dict[str, list[Arrival]]", message: ChatMessage +) -> list[Arrival]: + """The MAC group that is this message, out of the ones in the window. + + Each group is one message's frames — same MAC, so the same encrypted body. Which of + them is *this* message is the one thing the log cannot say, so the clock decides, and it + decides well: a send's frames start at the moment it was composed and run on through the + retries, so the group to take is the earliest one that did not start before the message + did. Falling back to the nearest group covers an inbound message, whose stamp is the + sender's clock and may sit slightly after the frame we heard. + """ + if not groups: + return [] + at = message.created_at + started = {mac: min(a.when for a in arrivals) for mac, arrivals in groups.items()} + after = [mac for mac, first in started.items() if first >= at] + if after: + return groups[min(after, key=lambda mac: started[mac])] + return groups[min(started, key=lambda mac: abs(started[mac] - at))] + + +def direct_arrivals( repo: "Repository", message: ChatMessage, *, - ends: Sequence[Optional[str]] = (), -) -> list[Arrival]: - """Direct-message frames logged around ``message``, matched by address and time. + self_key: Optional[str] = None, + peer_key: Optional[str] = None, +) -> tuple[list[Arrival], bool]: + """Every logged frame of one direct message, and whether they were matched exactly. - Direct frames are encrypted to their recipient, so the log can't confirm which - message a frame carried — the caller must present these as correlated evidence, not - a claim (see the module docstring). Their *addressing* is in the clear, though, so - naming the conversation's two ends narrows the window to the frames that ran between - those two nodes, in either direction, instead of every direct frame the radio - happened to overhear. A frame whose addressing wasn't recovered can't be shown to - belong, so it drops out — evidence, not guesswork. + Three filters (see the module docstring): the frame ran between this conversation's two + ends, in the direction this message travelled, and — where the log kept it — it shares + a MAC with the rest of its group, which is what makes one message's frames separable + from the next message's without decrypting either. + + **Direction matters more than it looks.** Both ends' hashes appear on every frame of a + conversation, whichever way it went, so matching on the pair alone put our own outgoing + frames into the view of a message we *received* and vice versa. That is where the + implausible one-hop rows came from: they were our own sends, overheard coming back off + the two repeaters in earshot, correctly showing one hop because one hop is all they had + travelled — but shown under an inbound message as if they were its route to us. Args: repo: The repository holding the packet log. - message: The chat message to search around. - ends: Keys (or key prefixes) of the conversation's two ends — the peer and our - own node. Each contributes its endpoint hash; whichever are given must - *both* appear on a frame for it to count. Empty falls back to every direct - frame in the window, matched by time alone. + message: The chat message whose frames to find. + self_key: Our own node's key (or prefix) — one end of every frame here. + peer_key: The conversation peer's key (or prefix) — the other end. Returns: - The matching frames as arrivals, oldest first. + ``(arrivals, exact)`` — the frames oldest first, and whether a MAC identified them + as one message rather than merely correlating them by address and time. """ - wanted = {h for h in (_endpoint_hash(end) for end in ends) if h} - frames = repo.packet_frames_between( - message.created_at - _DIRECT_WINDOW, message.created_at + _DIRECT_WINDOW - ) - arrivals: list[Arrival] = [] - for frame in frames: + ours, theirs = _endpoint_hash(self_key), _endpoint_hash(peer_key) + # Which way this message travelled, so a send's own frames and a received message's + # never appear under each other. + if message.outbound: + want_src, want_dest = ours, theirs + else: + want_src, want_dest = theirs, ours + + start, end = direct_window(repo, message) + groups: dict[str, list[Arrival]] = {} + loose: list[Arrival] = [] + for frame in repo.packet_frames_between(start, end): raw = frame.raw if isinstance(frame.raw, dict) else {} if raw.get("payload_typename") not in _DIRECT_TYPENAMES: continue - if wanted: - addressed = { - str(raw.get("dest_hash") or "").lower(), - str(raw.get("src_hash") or "").lower(), - } - if not wanted <= addressed: - continue - arrivals.append(Arrival(when=frame.observed_at, hops=_frame_hops(frame), snr=frame.snr)) - return arrivals + src, dest = _addressed(raw) + if (want_src and src != want_src) or (want_dest and dest != want_dest): + continue + arrival = Arrival(when=frame.observed_at, hops=_frame_hops(frame), snr=frame.snr) + mac = str(raw.get("cipher_mac") or "").lower() + if mac: + groups.setdefault(mac, []).append(arrival) + else: + loose.append(arrival) + if groups: + return _pick_group(groups, message), True + # History recorded before the MAC was kept: address, direction and time are all there + # is, and the caller says so rather than claiming more. + return loose, False + + +def collapse(arrivals: Sequence[Arrival]) -> list[Arrival]: + """Fold arrivals that rode the same path into one row apiece, counting the copies. + + A direct send is retried and each retry is overheard off every repeater in earshot, so + one message leaves a dozen frames on two paths. The graph has always drawn the *distinct* + paths; the row list showed every frame, which read as twelve arrivals where the truth was + two paths heard six times each. Each surviving row keeps its **first** sighting's time + and its **best** SNR — the earliest is when the path first worked, and the strongest is + what the path is capable of. + + Args: + arrivals: The matched frames, oldest first. + + Returns: + One :class:`Arrival` per distinct path, in first-heard order, each carrying its + :attr:`~Arrival.copies` count. + """ + folded: dict[tuple[str, ...], Arrival] = {} + for arrival in arrivals: + seen = folded.get(arrival.hops) + if seen is None: + folded[arrival.hops] = replace(arrival, copies=1) + continue + best = seen.snr if arrival.snr is None else ( + arrival.snr if seen.snr is None else max(seen.snr, arrival.snr) + ) + folded[arrival.hops] = replace(seen, snr=best, copies=seen.copies + 1) + return list(folded.values()) def distinct_paths(arrivals: list[Arrival]) -> int: diff --git a/meshterm/ui/chat.py b/meshterm/ui/chat.py index ee4f041..304c653 100644 --- a/meshterm/ui/chat.py +++ b/meshterm/ui/chat.py @@ -1182,7 +1182,8 @@ async def _make_paths_presenter( from ..services import trace_runner from ..services.message_paths import ( channel_arrivals, - direct_frames_near, + collapse, + direct_arrivals, distinct_paths, ) from .message_paths_screen import MessagePathsScreen @@ -1201,7 +1202,7 @@ async def _make_paths_presenter( info = await ctx.devstate.self_info() self_name = str(info.get("name") or "") or None # Our own key names one end of every direct frame in this conversation; the - # contact's key names the other (see :func:`direct_frames_near`). + # contact's key names the other (see :func:`direct_arrivals`). self_key = str(info.get("public_key") or "") or None except Exception: # noqa: BLE001 - a nameless self just skips the white highlight self_name = None @@ -1242,13 +1243,27 @@ async def _make_paths_presenter( peer_key = conversation.peer or ( (contact.public_key or contact.key_prefix) if contact else None ) - arrivals = direct_frames_near(ctx.repo, message, ends=(peer_key, self_key)) - matched = False - summary = ( - "direct frames are encrypted — matched by address and time (±90 s)" - if peer_key and self_key - else "direct frames are encrypted — matched by time alone (±90 s)" + arrivals, exact = direct_arrivals( + ctx.repo, message, self_key=self_key, peer_key=peer_key ) + # An exact match is the same claim the channel view makes: these frames are + # this message. It is earned by the frames' shared MAC, so it survives being + # unable to read a word of them. + matched = exact + arrivals = collapse(arrivals) + heard = sum(a.copies for a in arrivals) + if exact: + summary = ( + f"heard {heard} time{'s' if heard != 1 else ''}" + f" · {len(arrivals)} distinct path{'s' if len(arrivals) != 1 else ''}" + if arrivals else "no copies in the packet log" + ) + else: + summary = ( + "direct frames are encrypted — matched by address and time" + if peer_key and self_key + else "direct frames are encrypted — matched by time alone" + ) # The graph's left endpoint: who the message set out from. Our own sends are # us; an inbound channel message names its sender on the wire; a direct chat's # origin is the conversation's peer (billed as time-matched by the summary). diff --git a/meshterm/ui/message_paths_screen.py b/meshterm/ui/message_paths_screen.py index 8a3e552..8b9ba2c 100644 --- a/meshterm/ui/message_paths_screen.py +++ b/meshterm/ui/message_paths_screen.py @@ -286,10 +286,19 @@ class MessagePathsScreen(Screen): The hop count leads (:func:`~meshterm.ui.pathline.hops_atom`): it is the one fact about the route the line above encodes without stating, and the number two arrivals of the same message are compared on before anything else. + + A row stands for a *path*, not a frame (see + :func:`~meshterm.services.message_paths.collapse`), so a path heard more than once + says how many times and the time reads as the first of them — a direct send is + retried and every retry is overheard off every repeater in earshot, which used to + fill the list with a dozen identical lines. The SNR is the best that path managed, + which is what it is capable of. """ text = hops_atom(len(arrival.hops)) text.append(" ") text.append(arrival.when.astimezone().strftime("%H:%M:%S"), style="muted") + if arrival.copies > 1: + text.append(f" ×{arrival.copies}", style="muted") if arrival.snr is not None: text.append(" ") text.append(f"{arrival.snr:+.1f} dB", style=snr_style(arrival.snr)) diff --git a/tests/test_frames.py b/tests/test_frames.py index f397d4e..53ae41e 100644 --- a/tests/test_frames.py +++ b/tests/test_frames.py @@ -29,15 +29,22 @@ def _frame(typename: str, body: bytes, **extra) -> dict: return {"payload_typename": typename, "pkt_payload": body, **extra} -def test_addressed_classes_name_both_ends() -> None: - """A direct message, request, response or returned path opens with dest then src.""" +def test_addressed_classes_name_both_ends_and_their_mac() -> None: + """A direct message, request, response or returned path opens dest, src, then MAC. + + The MAC is a tag over the encrypted body, so it fingerprints *which* message a frame + carries without the frame ever being read — which is how the message-paths view groups + a direct message's own retransmissions and tells them from the next message's (see + :mod:`~meshterm.services.message_paths`). + """ body = bytes.fromhex("3da1") + b"\xab\xcd" + b"\x00" * 16 for typename in ("TEXT_MSG", "REQ", "RESPONSE", "PATH"): assert frame_addressing(_frame(typename, body)) == { - "dest_hash": "3d", "src_hash": "a1", + "dest_hash": "3d", "src_hash": "a1", "cipher_mac": "abcd", } + def test_anonymous_request_carries_its_senders_whole_key() -> None: """No shared secret yet, so the sender identifies itself in full — 32 bytes of key.""" key = bytes(range(32)) diff --git a/tests/test_message_paths.py b/tests/test_message_paths.py index f287a23..5c5c586 100644 --- a/tests/test_message_paths.py +++ b/tests/test_message_paths.py @@ -19,7 +19,8 @@ from meshterm.core.models import ChatMessage, Observation, utcnow from meshterm.persistence.repository import Repository from meshterm.services.message_paths import ( channel_arrivals, - direct_frames_near, + collapse, + direct_arrivals, distinct_paths, ) @@ -113,63 +114,169 @@ def test_channel_arrivals_ignore_frames_outside_the_window(tmp_path: Path) -> No repo.close() -def _direct_raw(dest: str = "", src: str = "") -> dict: +def _direct_raw(dest: str = "", src: str = "", mac: str = "") -> dict: """A direct-message frame's raw payload, spelled as the meshcore library reports it.""" raw: dict = {"payload_typename": "TEXT_MSG"} if dest: raw["dest_hash"] = dest if src: raw["src_hash"] = src + if mac: + raw["cipher_mac"] = mac return raw -def test_direct_frames_match_by_time_only(tmp_path: Path) -> None: - """With neither end named, every direct frame in the tight window is the evidence.""" +def test_direct_frames_without_a_mac_fall_back_to_address_and_time(tmp_path: Path) -> None: + """History recorded before the MAC was kept still correlates — and says that it did. + + The fallback is honest evidence, not a claim: a frame between the right pair in the + right window is only *probably* this message, and ``exact`` comes back ``False`` so the + view can bill it that way. + """ repo, run = _repo(tmp_path) now = utcnow() - _record_frame(repo, run, when=now + timedelta(seconds=3), path="3d63", raw=_direct_raw()) + _record_frame( + repo, run, when=now + timedelta(seconds=3), path="3d63", + raw=_direct_raw(dest="d4", src="a1"), + ) _record_frame( # a channel frame in the window is not direct-message evidence repo, run, when=now + timedelta(seconds=4), path="", raw=_grp_txt_raw(SECRET, "Alice: hi"), ) - _record_frame( # a direct frame far outside the tight window doesn't correlate - repo, run, when=now + timedelta(minutes=10), path="", raw=_direct_raw(), - ) - message = ChatMessage(text="see you at 8", peer="d4e5", created_at=now) - arrivals = direct_frames_near(repo, message) - assert len(arrivals) == 1 and arrivals[0].hops == ("3d63",) - repo.close() - - -def test_direct_frames_narrow_to_the_conversation_ends(tmp_path: Path) -> None: - """Naming both ends keeps only the frames that ran between them, either direction.""" - repo, run = _repo(tmp_path) - now = utcnow() - # Us (a1…) and the peer (d4…), each way round: both are this conversation's traffic. - _record_frame( - repo, run, when=now + timedelta(seconds=1), path="3d63", + _record_frame( # a direct frame far outside the window doesn't correlate + repo, run, when=now + timedelta(minutes=10), path="", raw=_direct_raw(dest="d4", src="a1"), ) - _record_frame( - repo, run, when=now + timedelta(seconds=2), path="c0", - raw=_direct_raw(dest="a1", src="d4"), + message = ChatMessage( + text="see you at 8", outbound=True, peer="d4e5", created_at=now ) - # Someone else's direct message, overheard in the same window. - _record_frame( - repo, run, when=now + timedelta(seconds=3), path="3d63", - raw=_direct_raw(dest="7f", src="c0"), + arrivals, exact = direct_arrivals( + repo, message, self_key="a1" + "0" * 62, peer_key="d4e5" ) - # Addressed to us, but by a third party — not this conversation. - _record_frame( - repo, run, when=now + timedelta(seconds=4), path="", - raw=_direct_raw(dest="a1", src="7f"), - ) - # Its addressing was never recovered, so it can't be shown to belong. - _record_frame(repo, run, when=now + timedelta(seconds=5), path="", raw=_direct_raw()) - - message = ChatMessage(text="see you at 8", peer="d4e5f6a7", created_at=now) - arrivals = direct_frames_near( - repo, message, ends=("d4e5f6a7", "a1" + "0" * 62), - ) - assert [a.hops for a in arrivals] == [("3d63",), ("c0",)] + assert [a.hops for a in arrivals] == [("3d63",)] + assert exact is False repo.close() + + +def test_direct_frames_keep_only_the_direction_the_message_travelled(tmp_path: Path) -> None: + """A send shows our outgoing frames; a received message shows the incoming ones. + + Both ends' hashes sit on every frame of a conversation whichever way it went, so + matching on the pair alone put our own sends into a received message's view. That is + where the implausible one-hop rows came from (JP, 2026-09-02): our transmissions heard + coming back off the repeaters in earshot, correctly one hop, shown as if they were an + inbound route to us. + """ + repo, run = _repo(tmp_path) + now = utcnow() + ours = dict(dest="d4", src="a1", mac="beef") # us -> peer + theirs = dict(dest="a1", src="d4", mac="f00d") # peer -> us + _record_frame(repo, run, when=now + timedelta(seconds=1), path="3d63", + raw=_direct_raw(**ours)) + _record_frame(repo, run, when=now + timedelta(seconds=2), path="c0", + raw=_direct_raw(**theirs)) + # Someone else's traffic, overheard in the same window. + _record_frame(repo, run, when=now + timedelta(seconds=3), path="3d63", + raw=_direct_raw(dest="7f", src="c0", mac="dead")) + _record_frame(repo, run, when=now + timedelta(seconds=4), path="", + raw=_direct_raw(dest="a1", src="7f", mac="cafe")) + + sent = ChatMessage(text="see you at 8", outbound=True, peer="d4e5f6a7", created_at=now) + got = ChatMessage(text="ok", outbound=False, peer="d4e5f6a7", created_at=now) + keys = dict(self_key="a1" + "0" * 62, peer_key="d4e5f6a7") + + assert [a.hops for a in direct_arrivals(repo, sent, **keys)[0]] == [("3d63",)] + assert [a.hops for a in direct_arrivals(repo, got, **keys)[0]] == [("c0",)] + repo.close() + + +def test_a_shared_mac_separates_one_message_from_the_next(tmp_path: Path) -> None: + """Two sends seconds apart keep their own frames — the MAC is the fingerprint. + + This is the direct-message counterpart of the channel view's content match: the frames + cannot be read, but copies of one message carry the same MAC over its ciphertext and the + next message's carry a different one. + """ + repo, run = _repo(tmp_path) + now = utcnow() + # First send, retried twice and heard off two repeaters. + for offset, path in ((1, "3d63"), (1, "27d4"), (6, "3d63"), (6, "27d4")): + _record_frame(repo, run, when=now + timedelta(seconds=offset), path=path, + raw=_direct_raw(dest="d4", src="a1", mac="beef")) + # Second send, twelve seconds later, same pair and same paths. + for offset, path in ((13, "3d63"), (13, "27d4")): + _record_frame(repo, run, when=now + timedelta(seconds=offset), path=path, + raw=_direct_raw(dest="d4", src="a1", mac="f00d")) + + keys = dict(self_key="a1" + "0" * 62, peer_key="d4e5f6a7") + first = ChatMessage(text="one", outbound=True, peer="d4e5f6a7", created_at=now) + second = ChatMessage( + text="two", outbound=True, peer="d4e5f6a7", + created_at=now + timedelta(seconds=12), + ) + got, exact = direct_arrivals(repo, first, **keys) + assert exact is True + assert len(got) == 4, "the first send's four frames, and not the second's" + assert len(direct_arrivals(repo, second, **keys)[0]) == 2 + repo.close() + + +def test_the_window_clamps_to_the_messages_going_the_same_way(tmp_path: Path) -> None: + """Each direction is bounded by its own neighbours, and bounded differently. + + Without a clamp a fast exchange put every message's frames in every other's view — nine + messages of one real conversation fell inside a single flat window. With the wrong + clamp, a *received* message got an empty one: its stamp is the sender's clock, so + ordering it against our own sends compares two clocks and can bound it by an edge that, + in its own clock, hasn't happened yet (JP, 2026-09-02). + """ + from meshterm.services.message_paths import _DIRECT_WINDOW, direct_window + + repo, run = _repo(tmp_path) + now = utcnow() + sent = [now, now + timedelta(seconds=20)] + received = [now - timedelta(seconds=30), now + timedelta(seconds=50)] + for at in sent: + repo.record_chat_message( + ChatMessage(text="ours", outbound=True, peer="d4e5f6a7", created_at=at) + ) + for at in received: + repo.record_chat_message( + ChatMessage(text="theirs", outbound=False, peer="d4e5f6a7", created_at=at) + ) + + # Our own send: our clock, so nothing of it predates it — the window opens at the + # message and runs forward to the next send, where retries actually live. + ours = ChatMessage(text="ours", outbound=True, peer="d4e5f6a7", created_at=now) + start, end = direct_window(repo, ours) + assert (now - start) < timedelta(seconds=5), "no room behind a send" + assert end == sent[1], "forward to the next send, not to the next received message" + + # A received message keeps a centred window, reaching halfway to the received messages + # either side of it — and is untouched by our own sends in between. + theirs = ChatMessage( + text="theirs", outbound=False, peer="d4e5f6a7", created_at=now + timedelta(seconds=50) + ) + start, end = direct_window(repo, theirs) + assert start == theirs.created_at - timedelta(seconds=40), "halfway back to the last one" + assert end == theirs.created_at + _DIRECT_WINDOW, "nothing follows, so that side stands" + repo.close() + + + +def test_collapse_folds_a_repeated_path_into_one_counted_row(tmp_path: Path) -> None: + """One row per path, carrying its copy count, first sighting and best SNR.""" + from meshterm.services.message_paths import Arrival + + base = utcnow() + arrivals = [ + Arrival(when=base, hops=("3d63",), snr=10.0), + Arrival(when=base + timedelta(seconds=1), hops=("27d4",), snr=5.0), + Arrival(when=base + timedelta(seconds=5), hops=("3d63",), snr=13.5), + ] + folded = collapse(arrivals) + assert [a.hops for a in folded] == [("3d63",), ("27d4",)], "first-heard order" + assert folded[0].copies == 2 and folded[1].copies == 1 + assert folded[0].when == base, "the first sighting is when the path first worked" + assert folded[0].snr == 13.5, "the best reading is what the path can do" +