Punch-list pass: dashboard hints, packet paging, decrypt, Time Machine axes

Four small UX gaps closed together:

- Dashboard's footer hint was missing Home/End even though the screen
  already handled them.
- The packet viewer's paging keys were reversed relative to the list
  that opens it: PgUp/PgDn walked packets while the dashboard feed
  walks with up/down. Swapped so up/down page through packets (matching
  the feed) and PgUp/PgDn scroll a tall body (matching every other
  screen's scroll keys).
- A raw "packet" entry (an overheard RX_LOG_DATA frame) now shows its
  parsed payload class and route, and — for a channel-text frame naming
  a channel we hold the key for — its decrypted text. The AES/HMAC
  logic lives in core/channels.py as pure, unit-tested crypto
  (decrypt_channel_text), independent of the vendored meshcore
  library's own reader internals; the dashboard fetches the device's
  channel secrets once at open time and hands them down to each
  PacketViewer it opens. Raw-field dumps now skip fields already folded
  into a flavoured row instead of repeating them.
- The Time Machine's whole-mesh charts (packets/day, nodes/day, rhythm)
  gained the same mirrored numeric y-axis the dashboard's activity
  chart already had. That gutter-drawing logic moved into
  braillechart.axis_chart, a shared widget, and the dashboard's own
  activity chart was refactored to use it too instead of hand-rolling
  its own copy.

meshterm/ui/channels.py's _read_slots is now public (read_channel_slots)
since the dashboard opener needs it too.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

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

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})

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.
 

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

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)

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())]

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())

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

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

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
 
 

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)