diff --git a/meshterm/core/channels.py b/meshterm/core/channels.py index 07ad9dc..e0bc29f 100644 --- a/meshterm/core/channels.py +++ b/meshterm/core/channels.py @@ -1,20 +1,27 @@ -"""Pure channel logic: secret derivation and MeshCore share-URL round-tripping. +"""Pure channel logic: secret derivation, MeshCore share-URL round-tripping, and decrypt. A MeshCore channel is a name plus a 16-byte shared secret. *Public* channels (whose name starts with ``#``) derive their secret deterministically from the name, so anyone naming the channel the same way lands on the same key; *private* channels carry a random secret shared out of band. This module holds that logic with no device or I/O dependency, so the create / join flows — and the ``meshcore://`` share links behind the QR codes — can be built and -validated (and unit-tested) without hardware. +validated (and unit-tested) without hardware. :func:`decrypt_channel_text` runs the same AES +key against an overheard, still-encrypted channel-text frame (the packet viewer's use of it), +so the app never needs the radio's own decode to read a channel it holds the key for. """ from __future__ import annotations import secrets +from dataclasses import dataclass +from datetime import datetime, timezone from hashlib import sha256 -from typing import Optional +from typing import Iterable, Optional from urllib.parse import parse_qs, quote, urlsplit +from Crypto.Cipher import AES +from Crypto.Hash import HMAC, SHA256 as _SHA256 + #: A channel shared secret is exactly 16 bytes (128-bit), per the MeshCore protocol. CHANNEL_SECRET_BYTES = 16 @@ -248,3 +255,76 @@ def parse_share_url(url: str) -> Optional[tuple[str, bytes]]: return names[0], normalize_secret(keys[0]) except ValueError: return None + + +@dataclass(slots=True) +class DecryptedText: + """A channel-text frame recovered from an overheard, encrypted packet. + + Attributes: + channel_name: The name of the channel whose key unlocked the frame. + text: The decrypted message body. + sent_at: The sender's own clock at the moment it composed the message, when + the embedded timestamp parses to a sane value. + attempt: The sender's resend counter for this message (0 = first try). + """ + + channel_name: str + text: str + sent_at: Optional[datetime] + attempt: int + + +def decrypt_channel_text( + chan_hash: str, + cipher_mac: str, + crypted: str, + channels: Iterable[tuple[str, bytes]], +) -> Optional[DecryptedText]: + """Recover a GRP_TXT frame's plaintext against a set of known channels. + + Mirrors the firmware's own decode of an overheard channel-text packet: the frame + names its channel only by :func:`channel_hash`'s one-byte fingerprint — several + channels can collide on it — so every same-fingerprint candidate is tried and its + 2-byte MAC checked before its key is trusted to decrypt anything. A frame from a + channel not in ``channels`` (or whose fingerprint matches but MAC doesn't — a + genuine collision) simply yields ``None``, same as firmware that doesn't know the + channel either. + + Args: + chan_hash: The frame's channel-hash fingerprint (2 hex chars). + cipher_mac: The frame's MAC, hex-encoded (2 bytes). + crypted: The frame's ciphertext, hex-encoded (a whole number of AES blocks). + channels: Candidate channels to try, as ``(name, secret)`` pairs. + + Returns: + The decrypted text and its channel, or ``None`` if no known channel's MAC + matches or the ciphertext is malformed. + """ + try: + mac = bytes.fromhex(cipher_mac) + msg = bytes.fromhex(crypted) + except ValueError: + return None + if not msg or len(msg) % AES.block_size: + return None # not a whole number of blocks: not a decryptable GRP_TXT body + for name, secret in channels: + key = effective_secret(name, secret) + if channel_hash(key) != chan_hash: + continue + mac_check = HMAC.new(key, digestmod=_SHA256) + mac_check.update(msg) + if mac_check.digest()[:2] != mac: + continue # fingerprint collided but the key doesn't actually match + plain = AES.new(key, AES.MODE_ECB).decrypt(msg) + timestamp = int.from_bytes(plain[0:4], "little") + attempt = plain[4] & 0x03 + text = plain[5:].strip(b"\x00").decode("utf-8", "ignore") + sent_at = None + if timestamp: + try: + sent_at = datetime.fromtimestamp(timestamp, tz=timezone.utc) + except (OverflowError, OSError, ValueError): + sent_at = None + return DecryptedText(channel_name=name, text=text, sent_at=sent_at, attempt=attempt) + return None diff --git a/meshterm/tools/channels.py b/meshterm/tools/channels.py index 6fb6ecc..181a241 100644 --- a/meshterm/tools/channels.py +++ b/meshterm/tools/channels.py @@ -83,10 +83,10 @@ class ChannelsTool(Tool): async def _cli_list(self, ctx: AppContext) -> ToolResult: """List the configured channel slots.""" - from ..ui.channels import _read_slots + from ..ui.channels import read_channel_slots device = await ctx.device() - slots = await _read_slots(device) + slots = await read_channel_slots(device) if not slots: ctx.ui.note("[muted]no channels configured[/muted]") return ToolResult(summary={"channels": 0}) @@ -146,11 +146,11 @@ class ChannelsTool(Tool): async def _cli_share(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: """Print a channel's share link and QR code.""" - from ..ui.channels import _read_slots + from ..ui.channels import read_channel_slots device = await ctx.device() idx = int(params["index"]) - slot = next((s for s in await _read_slots(device) if s.idx == idx), None) + slot = next((s for s in await read_channel_slots(device) if s.idx == idx), None) if slot is None: ctx.ui.note(f"[warn]slot {idx} is empty[/warn]") return ToolResult(summary={"index": idx, "shared": False}) diff --git a/meshterm/ui/braillechart.py b/meshterm/ui/braillechart.py index f6c2e96..dd57563 100644 --- a/meshterm/ui/braillechart.py +++ b/meshterm/ui/braillechart.py @@ -30,6 +30,10 @@ Beyond timelines, the module owns the app's two other braille conventions: ``width``-cell meter resolves ``2 × width`` levels. * :func:`axis_caption` — the ``oldest → now`` line under a timeline, which fills in intermediate marks whenever the chart is wide enough to fit them. +* :func:`axis_chart` — frames :func:`timeline_rows` output in a mirrored + numeric y-axis (the dashboard's activity chart, the Time Machine's per-day + and rhythm charts), so a chart quotes its scale on the gutters instead of + just naming the peak in its heading. """ from __future__ import annotations @@ -289,6 +293,72 @@ def axis_caption( return Text(label.ljust(chars), style=style) +def y_axis_labels(peak: float, rows: int) -> list[str]: + """Each chart row's top-edge value, top row first, dupes and zeros blanked. + + The top row always reads the peak; lower rows read the proportional values at + their upper edges, but a mark that would repeat the one above (a low peak makes + neighbouring rows round to the same value) or read zero is left blank, so the + scale never shows the same number twice. + """ + labels: list[str] = [] + seen: set[int] = set() + for i in range(rows): + value = round(peak * (rows - i) / rows) + if peak and value and value not in seen: + labels.append(str(value)) + seen.add(value) + else: + labels.append("") + return labels + + +def axis_chart( + chart_rows: list[Text], + peak: float, + chars: int, + label_at: Callable[[float], str], + *, + label_w: Optional[int] = None, + style: str = "muted", +) -> list[Text]: + """Frame :func:`timeline_rows` output with a mirrored y-axis and x-axis caption. + + Each row's top-edge value is mirrored on both gutters (blank where it would + repeat the mark above or read zero), closed with a boxed bottom border, and + followed by :func:`axis_caption`, indented to clear the gutter. + + Args: + chart_rows: The chart's rows, as returned by :func:`timeline_rows`. + peak: The value the top row's mark reads (the chart's tallest bar). + chars: The chart's width in character cells. + label_at: Maps a position fraction to the x-axis caption at that point. + label_w: The gutter's digit width, when several stacked charts must + share one width so their gutters line up; sized from ``peak`` alone + by default. + style: Style for the gutters, ticks, marks, and border. + + Returns: + ``len(chart_rows) + 2`` :class:`Text` lines: the decorated rows, the + bottom border, and the caption. + """ + label_w = label_w or max(1, len(str(round(peak)))) + marks = y_axis_labels(peak, len(chart_rows)) + out: list[Text] = [] + for mark, row in zip(marks, chart_rows): + line = Text(f"{mark:>{label_w}} " + ("┤" if mark else "│"), style=style) + line.append_text(row) + line.append("├" if mark else "│", style=style) + if mark: + line.append(f" {mark}", style=style) + out.append(line) + out.append(Text(" " * label_w + " └" + "─" * chars + "┘", style=style)) + caption = Text(" " * (label_w + 2)) + caption.append_text(axis_caption(chars, label_at)) + out.append(caption) + return out + + def _baseline_row(lo: float, hi: float, total: int) -> int: """The dot row (from the chart bottom) the zero baseline sits on. diff --git a/meshterm/ui/channels.py b/meshterm/ui/channels.py index 5b2f3bf..8ab49a6 100644 --- a/meshterm/ui/channels.py +++ b/meshterm/ui/channels.py @@ -152,7 +152,7 @@ async def manage_channels(ctx: "AppContext") -> int: highlight: Optional[object] = None while True: - slots = await _read_slots(device) + slots = await read_channel_slots(device) title, items = _menu_items(ctx, slots, capacity, stats) async def handle(choice: object) -> bool: @@ -248,7 +248,7 @@ async def _refresh_chat_channels(ctx: "AppContext") -> None: # --- reading ----------------------------------------------------------------- -async def _read_slots(device: Device) -> list[ChannelSlot]: +async def read_channel_slots(device: Device) -> list[ChannelSlot]: """Probe the channel slots and return the configured ones, in index order. The scan runs up to :data:`CHANNEL_SLOT_PROBE_CAP` and stops as soon as the firmware diff --git a/meshterm/ui/dashboard_screen.py b/meshterm/ui/dashboard_screen.py index a0c6421..e2cfdf9 100644 --- a/meshterm/ui/dashboard_screen.py +++ b/meshterm/ui/dashboard_screen.py @@ -22,7 +22,8 @@ stacks four reads of the mesh, coarsest first: opens full, then streamed live off the event hub. ``↑``/``↓`` walk the feed rows and Enter opens the highlighted packet in the shared :class:`~meshterm.ui.packet_viewer.PacketViewer` — which then pages through the - feed itself with PgUp/PgDn. + feed itself with the same ``↑``/``↓``, and, for an overheard channel-text packet + naming a channel we hold the key for, decrypts it. The screen holds no subscriptions of its own — the opener (:func:`open_dashboard`) wires the hub subscription, the device-stats poll, and the once-a-second repaint, and @@ -35,7 +36,7 @@ from __future__ import annotations import asyncio from collections import Counter, deque from statistics import median -from typing import TYPE_CHECKING, Any, Optional +from typing import TYPE_CHECKING, Any, Optional, Sequence from rich.console import Group, RenderableType from rich.table import Table @@ -44,7 +45,7 @@ from rich.text import Text from ..core.events import EventKind, MeshEvent from ..core.models import NODE_TYPE_REPEATER, Observation, utcnow from ..persistence.repository import ACTIVITY_WINDOW -from .braillechart import axis_caption, meter, timeline_rows +from .braillechart import axis_chart, meter, timeline_rows from .packet_viewer import KIND_STYLES, PacketEntry, PacketViewer, kind_icon, node_label from .theme import snr_style from .trace_screen import snr_bar @@ -99,31 +100,11 @@ def _span_label(minutes: int) -> str: return (text[:-2] if text.endswith(".0") else text) + " h" -def _axis_labels(peak: int, rows: int) -> list[str]: - """Each chart row's top-edge count, top row first, dupes and zeros blanked. - - The top row always reads the peak; lower rows read the proportional counts at - their upper edges, but a mark that would repeat the one above (a low peak makes - neighbouring rows round to the same value) or read zero is left blank, so the - scale never shows the same number twice. - """ - labels: list[str] = [] - seen: set[int] = set() - for i in range(rows): - value = round(peak * (rows - i) / rows) - if peak and value and value not in seen: - labels.append(str(value)) - seen.add(value) - else: - labels.append("") - return labels - - class DashboardScreen(Screen): """The live mesh overview. Renders state and scrolls; the opener feeds it.""" floating = False - footer_hint = "↑↓ packets · Enter open · PgUp/PgDn scroll · Esc back" + footer_hint = "↑↓ packets · Enter open · PgUp/PgDn/Home/End scroll · Esc back" def __init__( self, @@ -137,6 +118,7 @@ class DashboardScreen(Screen): hub_active: Any, prefix_bytes: int = 0, self_name: Optional[str] = None, + channels: Sequence[tuple[str, bytes]] = (), ) -> None: """Create the dashboard over its data feeds. @@ -151,6 +133,9 @@ class DashboardScreen(Screen): hub_active: Zero-arg callable: whether the event hub is pumping. prefix_bytes: Path-hash width to light in the packet viewer's hashes. self_name: Our own node's name, drawn white wherever it appears. + channels: The device's configured channels, as ``(name, secret)`` pairs, + handed to each opened :class:`~meshterm.ui.packet_viewer.PacketViewer` + so it can attempt to decrypt an overheard channel-text packet. """ super().__init__() self.title = "dashboard · mesh overview" @@ -162,6 +147,7 @@ class DashboardScreen(Screen): self._hub_active = hub_active self._prefix_bytes = prefix_bytes self._self_name = self_name + self._channels = channels #: The trailing window of observations (stored seed + live), oldest first. self._window: deque[Observation] = deque(window, maxlen=4000) #: The feed: latest events of every class as data, newest first — rendered @@ -280,6 +266,7 @@ class DashboardScreen(Screen): list(self._feed), self._selected, resolve=self._resolve, prefix_bytes=self._prefix_bytes, self_name=self._self_name, on_navigate=follow, + channels=self._channels, ) asyncio.ensure_future(self._session.run_screen(viewer)) @@ -342,26 +329,16 @@ class DashboardScreen(Screen): chart_rows = timeline_rows( list(reversed(shown)), rows=_CHART_ROWS, column_styles=styles ) - marks = _axis_labels(peak, len(chart_rows)) - out: list[RenderableType] = [heading] - for mark, row in zip(marks, chart_rows): - line = Text(f"{mark:>{label_w}} " + ("┤" if mark else "│"), style="muted") - line.append_text(row) - line.append("├" if mark else "│", style="muted") - if mark: - line.append(f" {mark}", style="muted") - out.append(line) - axis = Text(" " * label_w + " └" + "─" * chars + "┘", style="muted") - out.append(axis) def caption_at(frac: float) -> str: if frac >= 1.0: return "now" return "−" + _span_label(round(minutes * (1 - frac))) - caption = Text(" " * (label_w + 2)) - caption.append_text(axis_caption(chars, caption_at)) - out.append(caption) + out: list[RenderableType] = [ + heading, + *axis_chart(chart_rows, peak, chars, caption_at, label_w=label_w), + ] out.append(Text()) out.append(self._pulse_grid(shown, width)) return out @@ -607,6 +584,7 @@ async def open_dashboard(ctx: "AppContext") -> None: RuntimeError: If called outside the interactive menu (no full-screen session). """ from ..services import trace_runner + from .channels import read_channel_slots from .surface import TuiUi from .timemachine_screen import _routing_prefix_bytes @@ -616,11 +594,13 @@ async def open_dashboard(ctx: "AppContext") -> None: contacts = [] self_name: Optional[str] = None + channels: list[tuple[str, bytes]] = [] try: if ctx.is_connected or ctx.settings.connect_on_start: device = await ctx.device() contacts = await device.get_contacts() self_name = (await device.get_self_info()).get("name") or None + channels = [(s.name, s.secret) for s in await read_channel_slots(device)] except Exception: # noqa: BLE001 - the dashboard renders fine without contact names contacts = [] # Contacts first, every name the recorder ever overheard as the fallback — the @@ -639,6 +619,7 @@ async def open_dashboard(ctx: "AppContext") -> None: hub_active=lambda: ctx.events.active, prefix_bytes=prefix_bytes, self_name=self_name, + channels=channels, ) unsubscribe = ctx.events.subscribe(screen.on_event) diff --git a/meshterm/ui/packet_viewer.py b/meshterm/ui/packet_viewer.py index 68971ed..11f9005 100644 --- a/meshterm/ui/packet_viewer.py +++ b/meshterm/ui/packet_viewer.py @@ -5,27 +5,33 @@ the rows share this module's per-kind chrome — a colour and a two-cell icon pe class — and every row can open into the same viewer: a centered dialog over the list that lays the packet out in full, flavoured by kind. An advert shows the node's identity, type, and location; telemetry shows the node and its reported values; an -RX-logged packet shows the relay path it rode in on; a message shows the sender, the +RX-logged packet shows its parsed class and route plus the relay path it rode in on — +and, for an overheard channel-text frame naming a channel we hold the key for, the +decrypted text too (see :func:`~meshterm.core.channels.decrypt_channel_text`), even +though the radio itself never decoded it for us; a message shows the sender, the conversation, and the text; an ack shows its code. Common to all: the timestamp, the -node (name coloured by the app-wide palette, hash in the hash widget), and reception -quality on the shared SNR bar. +node (name coloured by the app-wide palette, hash in the hash widget), reception +quality on the shared SNR bar, and any leftover raw field the flavoured layout doesn't +already show. -When opened over a list the viewer pages through it in place — PgUp/PgDn step to the -newer/older packet, Home/End jump to the newest/oldest — so a burst can be read -packet by packet without bouncing back out to the list. ``↑``/``↓`` scroll a tall -packet inside the dialog; Esc closes it. +When opened over a list the viewer pages through it in place — ``↑``/``↓`` step to +the newer/older packet, Home/End jump to the newest/oldest, mirroring the keys the +opening list itself walks rows with — so a burst can be read packet by packet +without bouncing back out to the list. PgUp/PgDn scroll a tall packet inside the +dialog, matching what they do on every other screen; Esc closes it. """ from __future__ import annotations from dataclasses import dataclass from datetime import datetime -from typing import Callable, Optional +from typing import Callable, Optional, Sequence from rich.console import RenderableType from rich.table import Table from rich.text import Text +from ..core.channels import decrypt_channel_text from ..core.models import NODE_TYPE_LABELS, Observation from ..services.trace_runner import NodeResolver from .theme import name_style, snr_style @@ -34,6 +40,38 @@ from .tui.render import render_lines from .tui.screen import Screen from .widgets import _age_seconds, _format_age, highlighted_hash +#: Friendly gloss for a raw packet's parsed payload class (mirrors the meshcore +#: library's own ``PAYLOAD_TYPENAMES``), shown on a ``packet`` entry's "class" row. +_PAYLOAD_GLOSS = { + "REQ": "request", + "RESPONSE": "response", + "TEXT_MSG": "direct message", + "ACK": "ack", + "ADVERT": "advert", + "GRP_TXT": "channel text", + "GRP_DATA": "channel data", + "ANON_REQ": "anonymous request", + "PATH": "path", + "TRACE": "trace", + "MULTIPART": "multipart", + "CONTROL": "control", +} + +#: Raw-payload fields already folded into a flavoured row elsewhere in :meth:`_rows` +#: (frame plumbing parsed into "class"/"route", the channel crypto handled by +#: :meth:`PacketViewer._decrypt_rows`, or an advert field duplicating ``node``/ +#: ``lat``/``lon``/etc.) — skipped so the generic dump doesn't repeat them. +_RAW_ROW_SKIP = frozenset({ + "node", "name", "kind", "snr", "rssi", "lat", "lon", "path", "node_type", + "observed_at", "text", + "header", "payload_ver", "transport_code", "path_len", "path_hash_size", + "payload_type", "payload_typename", "route_type", "route_typename", + "pkt_payload", "pkt_hash", "raw_hex", "payload", "payload_length", "recv_time", + "chan_hash", "cipher_mac", "crypted", "message", "msg_hash", "sender_timestamp", + "attempt", "txt_type", + "adv_key", "adv_name", "adv_type", "adv_lat", "adv_lon", +}) + #: Display style per packet class, shared by every packet list and the viewer. KIND_STYLES = { "advert": "accent", @@ -157,9 +195,10 @@ class PacketViewer(Screen): """The floating packet dialog: one entry laid out in full, pageable over its list. A read-only popup. The opener hands it the list *as rendered* (newest first) and - the index of the row the user opened; PgUp/PgDn then walk toward newer/older - packets, Home/End jump to the ends, ``↑``/``↓`` scroll a tall body, Esc closes. - Opened over a single packet (a one-entry list) the paging keys simply do nothing. + the index of the row the user opened; ``↑``/``↓`` then walk toward newer/older + packets — the same keys the opening list itself uses — Home/End jump to the + ends, PgUp/PgDn scroll a tall body, Esc closes. Opened over a single packet (a + one-entry list) the paging keys simply do nothing. """ def __init__( @@ -171,6 +210,7 @@ class PacketViewer(Screen): prefix_bytes: int = 0, self_name: Optional[str] = None, on_navigate: Optional[Callable[[PacketEntry], None]] = None, + channels: Sequence[tuple[str, bytes]] = (), ) -> None: """Open the viewer over a packet list. @@ -183,6 +223,10 @@ class PacketViewer(Screen): self_name: Our own node's name, drawn white wherever it appears. on_navigate: Called with the newly shown entry whenever paging moves the view, so the opening list can walk its own highlight in step. + channels: The device's configured channels, as ``(name, secret)`` pairs — + tried against an overheard ``packet`` entry's channel-text frame (see + :func:`~meshterm.core.channels.decrypt_channel_text`), so a raw frame + the radio never decoded for us can still be read when we hold the key. """ super().__init__() self._entries = entries @@ -191,8 +235,9 @@ class PacketViewer(Screen): self._prefix_bytes = prefix_bytes self._self_name = self_name self._on_navigate = on_navigate + self._channels = channels if len(entries) > 1: - self.footer_hint = "PgUp/PgDn newer/older · Home/End ends · Esc close" + self.footer_hint = "↑↓ newer/older · PgUp/PgDn scroll · Home/End ends · Esc close" else: self.footer_hint = "Esc close" self._set_title() @@ -208,18 +253,18 @@ class PacketViewer(Screen): def handle(self, action: str, data: str = "") -> None: """Page through the list, scroll a tall entry, or dismiss.""" - if action == "pageup": + if action == "up": self._jump(self._index - 1) - elif action in ("pagedown", "space"): + elif action == "down": self._jump(self._index + 1) elif action in ("home", "ctrl_home"): self._jump(0) elif action in ("end", "ctrl_end"): self._jump(len(self._entries) - 1) - elif action == "up": - self.scroll_lines(-1) - elif action == "down": - self.scroll_lines(1) + elif action == "pageup": + self.scroll_pages(-1) + elif action in ("pagedown", "space"): + self.scroll_pages(1) elif action in ("escape", "enter"): self.resolve(None) @@ -277,11 +322,7 @@ class PacketViewer(Screen): if entry.snr is not None or entry.rssi is not None: rows.append(("snr", self._reception(entry))) if entry.kind == "packet": - rows.append(("via", self._path_text(entry))) - rows.append(( - "", Text("reception describes the last relay, not the origin", - style="faint"), - )) + rows.extend(self._packet_rows(entry)) if entry.lat is not None and entry.lon is not None: rows.append(("location", Text(f"{entry.lat:.5f}, {entry.lon:.5f}"))) @@ -324,6 +365,61 @@ class PacketViewer(Screen): text.append(hop, style="muted") return text + def _packet_rows(self, entry: PacketEntry) -> list[tuple[str, RenderableType]]: + """A raw ``packet`` entry's parsed class/route, relay path, and — for a + channel-text frame naming a channel we hold the key for — its plaintext. + """ + raw = entry.raw if isinstance(entry.raw, dict) else {} + rows: list[tuple[str, RenderableType]] = [] + typename = raw.get("payload_typename") + if typename: + rows.append(("class", Text(_PAYLOAD_GLOSS.get(typename, typename.lower())))) + route = raw.get("route_typename") + if route: + rows.append(("route", Text(route.replace("_", " ").lower()))) + rows.append(("via", self._path_text(entry))) + rows.append(( + "", Text("reception describes the last relay, not the origin", + style="faint"), + )) + if typename == "GRP_TXT": + rows.extend(self._decrypt_rows(raw)) + return rows + + def _decrypt_rows(self, raw: dict) -> list[tuple[str, RenderableType]]: + """Try every known channel's key against an overheard channel-text frame. + + The frame names its channel only by a one-byte hash fingerprint (several + channels can collide on it), so :func:`~meshterm.core.channels. + decrypt_channel_text` confirms the match by MAC before trusting a key to + decrypt anything — a channel we don't hold the key for, or a fingerprint we + can't confirm, is reported as such rather than left silently missing. + """ + 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): + return [] + decrypted = decrypt_channel_text(chan_hash, cipher_mac, crypted, self._channels) + if decrypted is None: + return [( + "channel", + Text(f"unknown (hash {chan_hash}) — can't decrypt", style="muted"), + )] + rows: list[tuple[str, RenderableType]] = [ + ("channel", Text(decrypted.channel_name, style="brand")), + ] + body = Text(decrypted.text) + if decrypted.sent_at is not None: + body.append( + f" · sent {decrypted.sent_at.astimezone().strftime('%H:%M:%S')}", + style="muted", + ) + if decrypted.attempt: + body.append(f" (resend #{decrypted.attempt})", style="muted") + rows.append(("text", body)) + return rows + def _raw_rows(self, entry: PacketEntry) -> list[tuple[str, RenderableType]]: """The raw payload's leftover fields, one labelled row each (telemetry's meat). @@ -334,13 +430,7 @@ class PacketViewer(Screen): raw = entry.raw if isinstance(entry.raw, dict) else None if not raw: return [] - shown = { - k: v - for k, v in raw.items() - if k not in ("node", "name", "kind", "snr", "rssi", "lat", "lon", "path", - "node_type", "observed_at", "text") - and v is not None - } + shown = {k: v for k, v in raw.items() if k not in _RAW_ROW_SKIP and v is not None} if not shown: return [] rows: list[tuple[str, RenderableType]] = [("", Text())] diff --git a/meshterm/ui/timemachine_screen.py b/meshterm/ui/timemachine_screen.py index 65903ce..0387f57 100644 --- a/meshterm/ui/timemachine_screen.py +++ b/meshterm/ui/timemachine_screen.py @@ -30,7 +30,7 @@ from rich.console import Group, RenderableType from rich.text import Text from ..core.models import utcnow -from .braillechart import axis_caption, chart_span, timeline_rows +from .braillechart import axis_caption, axis_chart, chart_span, timeline_rows from .theme import snr_style from .tui.render import render_lines from .tui.screen import Screen @@ -361,29 +361,29 @@ def _mesh_sections( Text("Press w to widen it.", style="muted"), ] - # Day-wide bars stretched to the chart's width; a deep history falls back to - # one dot column per day, the width capping how many trailing days fit. - chars = max(20, width - 2 * _GUTTER) + # The y-axis gutter is sized from the whole window's peaks (not just the + # visible slice), shared by both day charts so their gutters line up and + # neither chart's date range shifts relative to the other's. + label_w = max(len(str(max(d[1] for d in days))), len(str(max(d[2] for d in days)))) + chars = max(20, width - 2 * (label_w + 2)) shown = days[-chars * 2 :] out: list[RenderableType] = [] packets = [d[1] for d in shown] - out.append( - _heading("Packets per day", f"UTC days · peak {max(packets)} · newest at the right") - ) + out.append(_heading("Packets per day", "UTC days · newest at the right")) out.extend( - _chart_block( + axis_chart( timeline_rows(_day_columns(packets, chars), rows=_CHART_ROWS), - _day_axis(shown), chars, + max(packets), chars, _day_axis(shown), label_w=label_w, ) ) nodes = [d[2] for d in shown] out.append(Text()) - out.append(_heading("Nodes per day", f"distinct nodes heard · peak {max(nodes)}")) + out.append(_heading("Nodes per day", "distinct nodes heard")) out.extend( - _chart_block( + axis_chart( timeline_rows(_day_columns(nodes, chars), rows=_CHART_ROWS), - _day_axis(shown), chars, + max(nodes), chars, _day_axis(shown), label_w=label_w, ) ) @@ -397,7 +397,7 @@ def _mesh_sections( hours = [utc_hours[(h - offset) % 24] for h in range(24)] out.append(Text()) out.append(_heading("Rhythm", "packets by local hour of day")) - out.extend(_chart_block(timeline_rows(hours, rows=_CHART_ROWS), _hour_axis, 12)) + out.extend(axis_chart(timeline_rows(hours, rows=_CHART_ROWS), max(hours), 12, _hour_axis)) arrivals = ctx.repo.first_seen(since=since) out.append(Text()) diff --git a/pyproject.toml b/pyproject.toml index c59cce5..0cb7dc9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,6 +14,7 @@ keywords = ["meshcore", "lora", "mesh", "radio", "cli"] dependencies = [ "meshcore>=2.3", # companion-device protocol (serial/BLE/TCP) + "pycryptodome>=3.20", # AES/HMAC for decrypting overheard channel packets "pyserial>=3.5", # serial-port enumeration for device discovery "bleak>=0.22", # Bluetooth LE scanning + companion connection "typer>=0.12", # scripted CLI; subcommands mirror the menu diff --git a/tests/test_braillechart.py b/tests/test_braillechart.py index 24c69e2..563323e 100644 --- a/tests/test_braillechart.py +++ b/tests/test_braillechart.py @@ -6,7 +6,13 @@ the rendered braille bitmasks. from __future__ import annotations -from meshterm.ui.braillechart import activity_sparkline, chart_span, timeline_rows +from meshterm.ui.braillechart import ( + activity_sparkline, + axis_chart, + chart_span, + timeline_rows, + y_axis_labels, +) #: Dot bits for a column filled bottom-up to height 0–4 (left / right column). _L = (0x00, 0x40, 0x44, 0x46, 0x47) @@ -147,3 +153,42 @@ def test_sparkline_pads_and_crops_to_the_window() -> None: assert padded.plain == _BASE * 2 + _ch(_R[4], 0x40) cropped = activity_sparkline((0, 0, 21, 21, 9, 9), (1, 3, 8, 21), 2) assert cropped.plain == _BASE # only the two newest (silent) buckets survive + + +# --- y_axis_labels / axis_chart ------------------------------------------------------- + + +def test_y_axis_labels_blanks_a_silent_chart() -> None: + """A zero peak draws no marks at all.""" + assert y_axis_labels(0, 3) == ["", "", ""] + + +def test_y_axis_labels_top_row_reads_the_peak() -> None: + """The top row always reads the peak; lower rows are proportional.""" + assert y_axis_labels(10, 2) == ["10", "5"] + + +def test_y_axis_labels_blanks_a_repeated_mark() -> None: + """A low peak rounds a lower row to the same value as the one above: left blank.""" + assert y_axis_labels(1, 3) == ["1", "", ""] + + +def test_axis_chart_mirrors_marks_on_both_gutters() -> None: + """Each row's mark is framed by tick glyphs and repeated on both edges.""" + rows = timeline_rows([9] * 8, rows=2) + out = axis_chart(rows, 9, 4, lambda f: "now" if f >= 1.0 else "old") + assert len(out) == 4 # 2 chart rows + bottom border + caption + assert out[0].plain == "9 ┤⣿⣿⣿⣿├ 9" + assert out[1].plain == "4 ┤⣿⣿⣿⣿├ 4" + assert out[2].plain.startswith(" └") and out[2].plain.endswith("┘") + + +def test_axis_chart_honours_a_shared_label_width() -> None: + """An explicit label_w widens the gutter past the peak's own digit count. + + Lets several stacked charts (the Time Machine's per-day pair) share one gutter + width sized from the wider chart, so their marks line up column for column. + """ + rows = timeline_rows([9] * 8, rows=1) + out = axis_chart(rows, 9, 4, lambda f: "x", label_w=3) + assert out[0].plain.startswith(" 9 ┤") diff --git a/tests/test_channels.py b/tests/test_channels.py index efc8578..39123cf 100644 --- a/tests/test_channels.py +++ b/tests/test_channels.py @@ -19,6 +19,7 @@ from meshterm.core.channels import ( CHANNEL_SECRET_BYTES, DEFAULT_PUBLIC_SECRET, channel_hash, + decrypt_channel_text, derive_secret, full_channel_hash, normalize_secret, @@ -35,7 +36,7 @@ from meshterm.ui.channels import ( ChannelSlot, _apply_order, _next_free_slot, - _read_slots, + read_channel_slots, ) from meshterm.ui.qr import qr_text @@ -111,6 +112,88 @@ def test_parse_share_url_rejects_non_channel_links() -> None: assert parse_share_url("meshcore://channel/add?secret=" + "00" * 16) is None # no name +# -- channel-text decryption --------------------------------------------------- + + +def _grp_txt_frame( + name: str, secret: bytes, text: str, *, timestamp: int = 1_700_000_000, attempt: int = 0 +) -> tuple[str, str, str]: + """Build a firmware-shaped GRP_TXT frame (chan_hash, cipher_mac, crypted), all hex. + + Mirrors the encoding :func:`~meshterm.core.channels.decrypt_channel_text` reverses: + a 4-byte little-endian timestamp, an attempt/type byte, the text, zero-padded to a + whole AES block, encrypted ECB, then MAC'd with HMAC-SHA256 truncated to 2 bytes. + """ + from Crypto.Cipher import AES + from Crypto.Hash import HMAC, SHA256 + + from meshterm.core.channels import effective_secret + + key = effective_secret(name, secret) + plain = timestamp.to_bytes(4, "little") + bytes([attempt & 3]) + text.encode("utf-8") + plain += b"\x00" * (-len(plain) % 16) + crypted = AES.new(key, AES.MODE_ECB).encrypt(plain) + mac = HMAC.new(key, digestmod=SHA256) + mac.update(crypted) + return channel_hash(key), mac.digest()[:2].hex(), crypted.hex() + + +def test_decrypt_channel_text_round_trips_a_known_channel() -> None: + """A frame encrypted for a channel we hold the key for decodes cleanly.""" + name, secret = "#general", derive_secret("#general") + chash, mac, crypted = _grp_txt_frame(name, secret, "hello mesh") + result = decrypt_channel_text(chash, mac, crypted, [(name, secret)]) + assert result is not None + assert result.channel_name == name + assert result.text == "hello mesh" + assert result.attempt == 0 + assert result.sent_at is not None + + +def test_decrypt_channel_text_reports_a_resend_attempt() -> None: + """The sender's resend counter survives into the decrypted result.""" + name, secret = "Ops", random_secret() + chash, mac, crypted = _grp_txt_frame(name, secret, "again", attempt=2) + result = decrypt_channel_text(chash, mac, crypted, [(name, secret)]) + assert result is not None and result.attempt == 2 + + +def test_decrypt_channel_text_returns_none_for_an_unknown_channel() -> None: + """A frame from a channel we don't hold the key for decrypts to nothing.""" + name, secret = "#general", derive_secret("#general") + chash, mac, crypted = _grp_txt_frame(name, secret, "hello mesh") + other = ("Ops", random_secret()) + assert decrypt_channel_text(chash, mac, crypted, [other]) is None + assert decrypt_channel_text(chash, mac, crypted, []) is None + + +def test_decrypt_channel_text_rejects_a_fingerprint_collision() -> None: + """A candidate sharing the frame's hash byte but not its key fails the MAC check. + + The one-byte channel-hash fingerprint can collide between unrelated channels; + the MAC is what actually confirms the key, so a same-fingerprint wrong channel + must not be mistaken for a match, and decryption must fall through to try the + next (correct) candidate instead of stopping at the collision. + """ + name, secret = "#general", derive_secret("#general") + chash, mac, crypted = _grp_txt_frame(name, secret, "hello mesh") + # Brute-force a same-fingerprint, wrong-key decoy (a 1-byte hash, so cheap to find). + decoy = next( + s for s in (random_secret() for _ in range(10_000)) + if channel_hash(s) == chash and s != secret + ) + result = decrypt_channel_text( + chash, mac, crypted, [("decoy", decoy), (name, secret)] + ) + assert result is not None and result.text == "hello mesh" # falls through to the real key + + +def test_decrypt_channel_text_rejects_malformed_hex() -> None: + """Garbage hex fields fail closed instead of raising.""" + assert decrypt_channel_text("zz", "zzzz", "zzzzzzzzzzzzzzzz", [("x", random_secret())]) is None + assert decrypt_channel_text("00", "0000", "00" * 15, [("x", random_secret())]) is None # not block-aligned + + # -- QR rendering ------------------------------------------------------------- @@ -187,12 +270,12 @@ async def test_apply_order_relays_channels_into_new_positions(ctx: AppContext) - await device.set_channel(1, "#beta", None) # public, key derived from the name await device.set_channel(2, "Gamma", bytes(range(16, 32))) # private - slots = await _read_slots(device) + slots = await read_channel_slots(device) # Reverse the display order: the row that was third moves first, first moves last. writes = await _apply_order(device, slots, [2, 1, 0]) assert writes == 2 # the middle channel keeps its slot; the two ends swap - after = {s.idx: s for s in await _read_slots(device)} + after = {s.idx: s for s in await read_channel_slots(device)} assert after[0].name == "Gamma" and after[0].secret == bytes(range(16, 32)) assert after[1].name == "#beta" and after[1].is_public # public key re-derived in place assert after[2].name == "Alpha" and after[2].secret == bytes(range(16)) @@ -277,7 +360,7 @@ async def test_recreating_a_slot_refiles_messages_to_the_new_channel(ctx: AppCon ) await manage_channels(ctx) - ops = next(s for s in await _read_slots(device) if s.name == "Ops") + ops = next(s for s in await read_channel_slots(device) if s.name == "Ops") assert ops.idx == 0 # the new channel took the freed slot # A message now arrives on slot 0 — which is Ops, not Public. @@ -311,7 +394,7 @@ async def test_reorder_keeps_history_because_key_is_intrinsic(ctx: AppContext) - await chat.send_channel(0, "hello from alpha", label="Alpha") await chat.send_channel(1, "hello from beta", label="Beta") - slots = await _read_slots(device) + slots = await read_channel_slots(device) before = {s.name: s for s in slots} alpha_id, beta_id = before["Alpha"].identity, before["Beta"].identity @@ -320,7 +403,7 @@ async def test_reorder_keeps_history_because_key_is_intrinsic(ctx: AppContext) - assert writes == 2 await chat.refresh_channels() - after = {s.name: s for s in await _read_slots(device)} + after = {s.name: s for s in await read_channel_slots(device)} assert after["Alpha"].idx == 1 and after["Beta"].idx == 0 # slots actually swapped assert after["Alpha"].identity == alpha_id # identity is unchanged by the move @@ -399,7 +482,7 @@ async def test_channel_rows_carry_stats_unread_and_lanes(ctx: AppContext) -> Non device = await ctx.device() await device.set_channel(0, "Ops", bytes(range(16))) - slots = await _read_slots(device) + slots = await read_channel_slots(device) slot = slots[0] for text in ("one", "two"): ctx.repo.record_chat_message( @@ -462,7 +545,7 @@ async def test_detail_summary_reads_slot_totals_and_unread(ctx: AppContext) -> N device = await ctx.device() await device.set_channel(3, "Ops", bytes(range(16))) - slot = next(s for s in await _read_slots(device) if s.idx == 3) + slot = next(s for s in await read_channel_slots(device) if s.idx == 3) assert _detail_summary(ctx, slot, _LiveStats(ctx)) == "Slot 3 · no messages recorded yet" @@ -499,7 +582,7 @@ async def test_cli_add_private_generates_key_and_lists(ctx: AppContext) -> None: result = await tool.run(ctx, {"cli_action": "add", "index": 1, "name": "Ops", "secret": None}) assert result.summary == {"index": 1, "name": "Ops"} - slots = await _read_slots(await ctx.device()) + slots = await read_channel_slots(await ctx.device()) slot = next(s for s in slots if s.idx == 1) assert slot.name == "Ops" assert not slot.is_public # a random key, not derived from the name @@ -509,7 +592,7 @@ async def test_cli_add_public_derives_key(ctx: AppContext) -> None: """`add` with a #-name creates a public channel keyed from the name.""" tool = ChannelsTool() await tool.run(ctx, {"cli_action": "add", "index": 2, "name": "#general", "secret": None}) - slot = next(s for s in await _read_slots(await ctx.device()) if s.idx == 2) + slot = next(s for s in await read_channel_slots(await ctx.device()) if s.idx == 2) assert slot.name == "#general" and slot.is_public @@ -520,12 +603,12 @@ async def test_cli_join_and_import_round_trip(ctx: AppContext) -> None: await tool.run( ctx, {"cli_action": "join", "index": 3, "name": "Squad", "secret": secret.hex()} ) - slot = next(s for s in await _read_slots(await ctx.device()) if s.idx == 3) + slot = next(s for s in await read_channel_slots(await ctx.device()) if s.idx == 3) assert slot.name == "Squad" and slot.secret == secret # Importing the channel's own share link onto another slot reproduces it. await tool.run(ctx, {"cli_action": "import", "index": 4, "url": share_url("Squad", secret)}) - imported = next(s for s in await _read_slots(await ctx.device()) if s.idx == 4) + imported = next(s for s in await read_channel_slots(await ctx.device()) if s.idx == 4) assert imported.name == "Squad" and imported.secret == secret diff --git a/tests/test_packet_viewer.py b/tests/test_packet_viewer.py new file mode 100644 index 0000000..447242d --- /dev/null +++ b/tests/test_packet_viewer.py @@ -0,0 +1,80 @@ +"""Packet viewer tests: per-kind flavouring, the raw-field dedup, and channel decrypt. + +The viewer is pure Rich-in, ANSI-lines-out (``render_body``), so it's driven headless +against a synthetic :class:`PacketEntry` — the same approach as the dashboard tests. +""" + +from __future__ import annotations + +import re + +from Crypto.Cipher import AES +from Crypto.Hash import HMAC, SHA256 + +from meshterm.core.channels import channel_hash, derive_secret +from meshterm.core.models import utcnow +from meshterm.ui.packet_viewer import PacketEntry, PacketViewer + + +def _stripped(lines: list[str]) -> list[str]: + return [re.sub(r"\x1b\[[0-9;]*m", "", line) for line in lines] + + +def _plain(lines: list[str]) -> str: + return "\n".join(_stripped(lines)) + + +def _grp_txt_entry(raw_extra: dict) -> PacketEntry: + raw = {"payload_typename": "GRP_TXT", "route_typename": "FLOOD", **raw_extra} + return PacketEntry(when=utcnow(), kind="packet", path="a1b2c3d4e5f6", raw=raw) + + +def _viewer(entry: PacketEntry, channels=()) -> PacketViewer: + return PacketViewer([entry], 0, resolve=lambda h: "", channels=channels) + + +def _grp_txt_frame(secret: bytes, text: str) -> tuple[str, str, str]: + """A firmware-shaped GRP_TXT frame for ``text``, encrypted under ``secret``.""" + plain = (0).to_bytes(4, "little") + bytes([0]) + text.encode("utf-8") + plain += b"\x00" * (-len(plain) % 16) + crypted = AES.new(secret, AES.MODE_ECB).encrypt(plain) + mac = HMAC.new(secret, digestmod=SHA256) + mac.update(crypted) + return channel_hash(secret), mac.digest()[:2].hex(), crypted.hex() + + +def test_packet_viewer_shows_class_and_route() -> None: + """A raw packet's parsed payload class and route surface as labelled rows.""" + entry = _grp_txt_entry({"chan_hash": "ff", "cipher_mac": "0000", "crypted": "00" * 16}) + body = _plain(_viewer(entry).render_body(80)) + assert "channel text" in body + assert "flood" in body + + +def test_packet_viewer_decrypts_a_known_channel() -> None: + """A channel-text frame we hold the key for decrypts to its plaintext.""" + name, secret = "#general", derive_secret("#general") + chash, mac, crypted = _grp_txt_frame(secret, "hi mesh") + entry = _grp_txt_entry({"chan_hash": chash, "cipher_mac": mac, "crypted": crypted}) + body = _plain(_viewer(entry, channels=[(name, secret)]).render_body(80)) + assert "#general" in body + assert "hi mesh" in body + + +def test_packet_viewer_reports_an_unknown_channel() -> None: + """A frame from a channel we don't hold the key for says so instead of guessing.""" + entry = _grp_txt_entry({"chan_hash": "ab", "cipher_mac": "0000", "crypted": "00" * 16}) + body = _plain(_viewer(entry).render_body(80)) + assert "unknown" in body and "can't decrypt" in body + + +def test_packet_viewer_raw_dump_skips_fields_folded_into_flavoured_rows() -> None: + """Fields already shown as class/route/via/channel rows don't also dump generically.""" + entry = _grp_txt_entry({ + "chan_hash": "ab", "cipher_mac": "0000", "crypted": "00" * 16, + "path_len": 1, "path_hash_size": 1, "header": 5, "novel_field": "surprise", + }) + body = _plain(_viewer(entry).render_body(80)) + assert "path_len" not in body + assert "header" not in body + assert "novel_fi" in body # an unrecognized field still surfaces (label truncates to 8)