Jean-Pierre MartineauandClaude Opus 5.5
5fc224476c
The same rewrite as for the core package. Three comments now agree with the code: the activity bucket of the header is one minute (ACTIVITY_BUCKET_S), the RX log holds packets, and a reference to a module note that does not exist is gone. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
458 lines
21 KiB
Python
458 lines
21 KiB
Python
# SPDX-License-Identifier: Apache-2.0
|
|
"""Find the paths on which a chat message arrived, from the stored packet log.
|
|
|
|
The RX log of the device records each packet that the device decodes, with the relay path
|
|
that the packet went through. The recorder always stored these packets as ``packet``
|
|
observations. Thus a message that was heard more than once left one logged packet for each
|
|
arrival, each with its own path and SNR. For example, several repeaters relayed a channel
|
|
broadcast, or a flood reached our node on two paths. This module matches these packets to
|
|
a chat message. Thus the chat screen can show, when the user asks, each path on which the
|
|
message reached our node:
|
|
|
|
* **Channel messages** match by content. A ``GRP_TXT`` packet identifies its channel only
|
|
by a one-byte hash. But MeshTerm has the key of the channel, so
|
|
:func:`~meshterm.core.channels.decrypt_channel_text` can confirm the channel by MAC and
|
|
recover the plaintext. A packet whose text is equal to the text of the message is one
|
|
arrival of that message. The comparison accepts the ``Name: `` sender prefix on either
|
|
side. Our own broadcasts are included, because the device overhears the relay of our
|
|
broadcast by a repeater and logs it like all other packets.
|
|
* **Direct messages** go in ECDH-encrypted ``TEXT_MSG`` packets that only the recipient
|
|
can decrypt. Thus a match by content is not possible from the log. But such a packet
|
|
carries its **addressing and its MAC** in clear text (:mod:`~meshterm.core.frames`): the
|
|
one-byte key hash of each end, then a two-byte tag over the encrypted body. That MAC
|
|
identifies the message itself: the copies of one message have the same MAC, and the
|
|
next message has a different MAC. Thus direct packets make groups as exactly as channel
|
|
packets do, and MeshTerm never decrypts them. Three filters apply, in this order: the
|
|
packet went **between the two ends of this conversation**, in the **direction in which
|
|
this message travelled** (our own sends are ``us → peer``, and a received message is
|
|
``peer → us``), and it carries **the same MAC** as the rest of its group.
|
|
|
|
Then the time tells apart the messages of a conversation: the MAC group nearest to the
|
|
time of a message is that message. One real gap remains. History that was stored before
|
|
MeshTerm kept the MAC has no MAC. For these messages, MeshTerm matches by address,
|
|
direction, and time, and it shows that it used this type of match. The reason is that
|
|
a packet between the correct pair in the correct window is probably this message, but
|
|
not certainly.
|
|
|
|
Nothing in this module transmits. It is a read-model over the repository.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Callable, Sequence
|
|
from dataclasses import dataclass, field, replace
|
|
from datetime import datetime, timedelta
|
|
from typing import TYPE_CHECKING
|
|
|
|
from ..core.channels import decrypt_channel_text, split_channel_sender
|
|
from ..core.frames import ENDPOINT_HASH_BYTES
|
|
from ..core.models import ChatMessage, Observation
|
|
|
|
if TYPE_CHECKING:
|
|
from ..core.regions import Scope
|
|
from ..persistence.repository import Repository
|
|
|
|
#: The span of the log that MeshTerm searches around a channel message. It is wide,
|
|
#: because an inbound message has the time of the sender's clock, and that clock can be
|
|
#: minutes different from our clock.
|
|
_CHANNEL_WINDOW = timedelta(minutes=15)
|
|
|
|
#: The span of the log that MeshTerm searches around a direct message, before the other
|
|
#: messages of the conversation make it narrower (refer to :func:`direct_window`). This
|
|
#: span alone is wide: a send is tried again for some time after it is composed, and an
|
|
#: inbound message has the time of the sender's clock.
|
|
_DIRECT_WINDOW = timedelta(seconds=90)
|
|
|
|
#: The minimum span on each side of a direct message, after its neighbours limit the
|
|
#: window. Two messages composed one second apart must each still have a span to search.
|
|
#: Without this minimum, the midpoint between them can leave one of them no span at all.
|
|
_MIN_HALF_WINDOW = timedelta(seconds=2)
|
|
|
|
#: Route types whose ``path`` field is an instruction for the route, not a record of where
|
|
#: the packet went. The sender writes the route, and the relays consume it. The library
|
|
#: gives names to the two route bits of the header (refer to its ``ROUTE_TYPENAMES``). The
|
|
#: flood types add to the path instead. Before, MeshTerm read each path here as a flood
|
|
#: path.
|
|
_ROUTED_TYPES = frozenset({"DIRECT", "TC_DIRECT"})
|
|
|
|
#: Packet classes that carry a direct (addressed) text message. This is the payload class
|
|
#: that the meshcore library names, spelled exactly as the library reports it (refer to
|
|
#: :data:`~meshterm.core.frames.ADDRESSED_CLASSES`, the lane of the packet viewer, and the
|
|
#: traffic keys of the dashboard). A name that is almost correct matches nothing here.
|
|
#: Then the mesh looks as if it never carries direct messages.
|
|
_DIRECT_TYPENAMES = frozenset({"TEXT_MSG"})
|
|
|
|
#: The number of hex digits in the endpoint hash of a packet. This hash is the first byte
|
|
#: of a public key. Both ends of a direct packet are addressed by this byte.
|
|
_HASH_CHARS = 2 * ENDPOINT_HASH_BYTES
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class Arrival:
|
|
"""One logged copy of a message that reached the device.
|
|
|
|
Attributes:
|
|
when: The time when the packet was heard.
|
|
hops: The relay path of the packet, in the order of propagation (empty means that
|
|
the packet arrived direct).
|
|
snr: The SNR of the reception in dB. It is the SNR from the last relay only, the
|
|
same as for each packet row.
|
|
resend: The resend counter of the sender for this copy (0 is the original send).
|
|
It comes from a decrypted channel packet. It is always 0 for direct packets.
|
|
routed: Whether the packet was **direct-routed** or flooded. This value decides
|
|
what :attr:`hops` means. A flooded packet adds to its path (each relay adds
|
|
itself), so the hops are the route on which the packet reached our node. A
|
|
direct-routed packet carries a route that its sender wrote and its relays
|
|
consume. Thus the hops are the part of the route that the packet still had to
|
|
go through. An empty path means that the packet used all of its route, not
|
|
that the packet arrived in a single hop. ``None`` for history that was stored
|
|
before MeshTerm kept the route type.
|
|
copies: The number of logged packets that this row represents. The matchers
|
|
always make it ``1``. It becomes more after :func:`collapse` combines a path
|
|
heard several times into one row. A direct send is tried again, and the device
|
|
overhears each retry from each repeater in range. Thus one message often
|
|
leaves a dozen packets on two paths. A list of each packet said "twelve
|
|
arrivals", but the truth was "two paths, six times each".
|
|
frame: The raw payload of the logged packet. :func:`message_scope` reads the scope
|
|
of the message from it. It is not part of the identity of an arrival (two
|
|
copies on one path are one row, whatever else their packets carry). Thus
|
|
equality does not use it.
|
|
"""
|
|
|
|
when: datetime
|
|
hops: tuple[str, ...]
|
|
snr: float | None
|
|
resend: int = 0
|
|
routed: bool | None = None
|
|
copies: int = 1
|
|
frame: dict | None = field(default=None, compare=False, repr=False)
|
|
|
|
@property
|
|
def route_known(self) -> bool:
|
|
"""Whether :attr:`hops` tells how this copy travelled.
|
|
|
|
False for a direct-routed packet that carries no path. The packet consumed its
|
|
route on the way. The field is empty because nothing of the route remains, not
|
|
because the packet went through no relays. MeshTerm once drew such a packet as a
|
|
zero-hop arrival, and thus claimed a neighbour that we do not have: the
|
|
"impossible direct path" (JP, 2026-09-02).
|
|
"""
|
|
return bool(self.hops) or not self.routed
|
|
|
|
|
|
def _frame_hops(observation: Observation) -> tuple[str, ...]:
|
|
"""The path field of an observation, as hop hashes.
|
|
|
|
Refer to :attr:`Arrival.routed` for what the hops mean.
|
|
"""
|
|
return tuple(h for h in (observation.path or "").split(",") if h)
|
|
|
|
|
|
def _frame_routed(raw: dict) -> bool | None:
|
|
"""Whether a packet was direct-routed, or ``None`` when its route type was not kept."""
|
|
name = str(raw.get("route_typename") or "").upper()
|
|
return name in _ROUTED_TYPES if name and name != "UNK" else None
|
|
|
|
|
|
def _texts_match(wire: str, stored: str) -> bool:
|
|
"""Whether a text from the air and a stored chat text are the same message body.
|
|
|
|
MeshTerm stores an inbound message exactly as it came from the air (with the sender
|
|
prefix), so an exact match covers it. MeshTerm stores our own outbound messages as
|
|
the user typed them, but the device adds the ``Name: `` sender prefix. Thus the text
|
|
from the air also matches when its body without the prefix is equal to the stored
|
|
text. The comparison ignores whitespace at the two ends, but it is otherwise exact.
|
|
Paths are evidence, and a fuzzy match can invent false evidence.
|
|
"""
|
|
wire, stored = wire.strip(), stored.strip()
|
|
if wire == stored:
|
|
return True
|
|
_sender, body = split_channel_sender(wire)
|
|
body = body.strip()
|
|
return bool(body) and body == stored
|
|
|
|
|
|
def channel_arrivals(
|
|
repo: Repository,
|
|
message: ChatMessage,
|
|
*,
|
|
channel_name: str,
|
|
secret: bytes,
|
|
) -> list[Arrival]:
|
|
"""All the logged arrivals of a channel message, matched by decrypted content.
|
|
|
|
Args:
|
|
repo: The repository that holds the packet log.
|
|
message: The chat message for which to find the arrivals (inbound or outbound).
|
|
channel_name: The channel name of the conversation (for the decryption).
|
|
secret: The 16-byte key of the channel.
|
|
|
|
Returns:
|
|
The matching arrivals, oldest first. Resends of the same message are included,
|
|
and each carries its ``resend`` counter.
|
|
"""
|
|
frames = repo.packet_frames_between(
|
|
message.created_at - _CHANNEL_WINDOW, message.created_at + _CHANNEL_WINDOW
|
|
)
|
|
arrivals: list[Arrival] = []
|
|
for frame in frames:
|
|
raw = frame.raw if isinstance(frame.raw, dict) else {}
|
|
if raw.get("payload_typename") != "GRP_TXT":
|
|
continue
|
|
chan_hash = raw.get("chan_hash")
|
|
cipher_mac = raw.get("cipher_mac")
|
|
crypted = raw.get("crypted")
|
|
if not (chan_hash and cipher_mac and crypted):
|
|
continue
|
|
decrypted = decrypt_channel_text(chan_hash, cipher_mac, crypted, [(channel_name, secret)])
|
|
if decrypted is None or not _texts_match(decrypted.text, message.text):
|
|
continue
|
|
arrivals.append(
|
|
Arrival(
|
|
when=frame.observed_at,
|
|
hops=_frame_hops(frame),
|
|
snr=frame.snr,
|
|
resend=decrypted.attempt,
|
|
routed=_frame_routed(raw),
|
|
frame=raw,
|
|
)
|
|
)
|
|
return arrivals
|
|
|
|
|
|
def _endpoint_hash(key: str | None) -> str:
|
|
"""The endpoint hash of a node: the first byte of its key, as a packet addresses it."""
|
|
text = (key or "").strip().lower()
|
|
return text[:_HASH_CHARS] if len(text) >= _HASH_CHARS else ""
|
|
|
|
|
|
def direct_window(repo: Repository, message: ChatMessage) -> tuple[datetime, datetime]:
|
|
"""The span of the log to search for the packets of one direct message.
|
|
|
|
The span is :data:`_DIRECT_WINDOW` on each side. Then the other messages of the
|
|
conversation limit it. Only the messages that travelled the **same way** limit it,
|
|
because two different clocks stamp the two directions (refer to
|
|
:meth:`~meshterm.persistence.repository.Repository
|
|
.direct_message_bounds`). Without a limit, the flat window is much too wide at the speed
|
|
of a conversation. Nine messages of one real exchange were all in a single ±90 s
|
|
window, so each of them claimed the packets of all nine messages.
|
|
|
|
The two directions get different limits, because we know very different things about
|
|
them:
|
|
|
|
* **Our own sends** get the time of our own clock when they leave. Thus no packet of a
|
|
send can come before it. The window opens at the message (minus a short time for
|
|
jitter) and goes to the next send. Retries come after the send, so all the span
|
|
goes forward.
|
|
* **A received message** has the time of the sender's clock, which can be before or
|
|
after the time when we heard it. Its window stays centred, and it goes halfway to
|
|
the messages before and after it.
|
|
|
|
The window is never narrower than :data:`_MIN_HALF_WINDOW` on each side. Thus two
|
|
messages one second apart still each have a span to search.
|
|
|
|
Args:
|
|
repo: The repository that holds the messages.
|
|
message: The message around which to limit a search.
|
|
|
|
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 own clock stamps it, so no packet of this send comes before it. All the
|
|
# 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 packet's ``(source, destination)`` endpoint hashes, in lower case (empty if 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, from the groups in the window.
|
|
|
|
Each group holds the packets of one message: the same MAC, thus the same encrypted
|
|
body. The log cannot tell which group is this message. Thus the clock decides, and it
|
|
decides well. The packets of a send start at the time when it was composed, and they
|
|
continue through the retries. Thus the correct group is the earliest one that did not
|
|
start before the message. The fallback to the nearest group covers an inbound message,
|
|
whose time comes from the sender's clock and can be a little after the packet that 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,
|
|
*,
|
|
self_key: str | None = None,
|
|
peer_key: str | None = None,
|
|
) -> tuple[list[Arrival], bool]:
|
|
"""All the logged packets of one direct message, and whether the match is exact.
|
|
|
|
Three filters apply (refer to the module docstring). The packet went between the two
|
|
ends of this conversation, in the direction in which this message travelled. Also,
|
|
where the log kept the MAC, the packet has the same MAC as the rest of its group. The
|
|
MAC lets MeshTerm separate the packets of one message from the packets of the next
|
|
message, without decryption of either.
|
|
|
|
**Direction is more important than it seems.** The hashes of both ends are on each
|
|
packet of a conversation, in both directions. Thus a match on the pair alone put our
|
|
own outgoing packets under a message that we received (not sent), and the opposite.
|
|
That caused the improbable one-hop rows. They were our own sends, overheard when the
|
|
two repeaters in range relayed them back. They correctly showed one hop, because they
|
|
had travelled only one hop. But they showed under an inbound message as if they were
|
|
its route to our node.
|
|
|
|
Args:
|
|
repo: The repository that holds the packet log.
|
|
message: The chat message for which to find the packets.
|
|
self_key: The key (or key prefix) of our node. It is one end of each packet here.
|
|
peer_key: The key (or key prefix) of the conversation peer: the other end.
|
|
|
|
Returns:
|
|
``(arrivals, exact)``: the packets oldest first, and whether a MAC identified them
|
|
as one message. If not, MeshTerm only correlated them by address and time.
|
|
"""
|
|
ours, theirs = _endpoint_hash(self_key), _endpoint_hash(peer_key)
|
|
# The direction of this message. Thus the packets of a send never show under a
|
|
# received message, and the packets of a received message never show under a send.
|
|
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
|
|
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,
|
|
routed=_frame_routed(raw),
|
|
frame=raw,
|
|
)
|
|
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 stored before MeshTerm kept the MAC: only the address, direction, and time
|
|
# are available. The caller says so, and does not claim more.
|
|
return loose, False
|
|
|
|
|
|
def collapse(arrivals: Sequence[Arrival]) -> list[Arrival]:
|
|
"""Combine the arrivals on the same path into one row for each path, and count the copies.
|
|
|
|
A direct send is tried again, and the device overhears each retry from each repeater in
|
|
range. Thus one message leaves a dozen packets on two paths. The graph always drew the
|
|
distinct paths. But the row list showed each packet, and that read as twelve arrivals,
|
|
when the truth was two paths heard six times each. Each row that remains keeps the time
|
|
of its **first** packet and its **best** SNR. The earliest time is when the path first
|
|
worked, and the strongest SNR is what the path can do.
|
|
|
|
Args:
|
|
arrivals: The matched packets, oldest first.
|
|
|
|
Returns:
|
|
One :class:`Arrival` for each distinct path, in the order first heard, each with
|
|
its :attr:`~Arrival.copies` count.
|
|
"""
|
|
# The key has the route type and the hops. The same hashes have two different meanings
|
|
# under the two types. If the two are combined, a route that the packet travelled
|
|
# merges with a route that the sender intended.
|
|
folded: dict[tuple, Arrival] = {}
|
|
for arrival in arrivals:
|
|
key = (arrival.hops, arrival.routed)
|
|
seen = folded.get(key)
|
|
if seen is None:
|
|
folded[key] = 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[key] = replace(seen, snr=best, copies=seen.copies + 1)
|
|
return list(folded.values())
|
|
|
|
|
|
def message_scope(
|
|
arrivals: Sequence[Arrival], scope_of: Callable[[dict | None], Scope | None]
|
|
) -> Scope | None:
|
|
"""The one scope under which a message was flooded, read from the packets of its arrivals.
|
|
|
|
A transport code is a keyed hash of the payload only. Neither the header nor the path
|
|
is in it (refer to :mod:`~meshterm.core.regions`). Thus the copy from each relay
|
|
carries the code of the sender without change. One message has **one** scope, however
|
|
many paths it arrived on. For this reason, the dialog shows the scope once, not for
|
|
each arrival row.
|
|
|
|
But the packet that gives the answer is still important, because not each copy is a
|
|
flood: a direct-routed copy has no scope at all. Thus the first scoped reading wins,
|
|
and a named one wins over one that no known region reproduces. (They are the same
|
|
code, but a name is the better answer when a copy gives one.) If there is no scoped
|
|
reading, any plain flood copy makes the message ``unscoped``. A message heard only as
|
|
direct-routed has no scope to show.
|
|
|
|
Args:
|
|
arrivals: The arrivals of the message, as the matchers (or :func:`collapse`)
|
|
return them. Each carries its logged packet.
|
|
scope_of: Reads the scope of a packet (``ctx.region_store.scope_of``).
|
|
|
|
Returns:
|
|
The scope of the message, or ``None`` when no copy was a flood.
|
|
"""
|
|
unscoped: Scope | None = None
|
|
unnamed: Scope | None = None
|
|
for arrival in arrivals:
|
|
scope = scope_of(arrival.frame)
|
|
if scope is None:
|
|
continue
|
|
if not scope.scoped:
|
|
unscoped = unscoped or scope
|
|
elif scope.region:
|
|
return scope
|
|
else:
|
|
unnamed = unnamed or scope
|
|
return unnamed or unscoped
|
|
|
|
|
|
def distinct_paths(arrivals: list[Arrival]) -> int:
|
|
"""The number of distinct relay paths that a set of arrivals covers."""
|
|
return len({a.hops for a in arrivals})
|