diff --git a/meshtools/core/channels.py b/meshtools/core/channels.py new file mode 100644 index 0000000..a595ab5 --- /dev/null +++ b/meshtools/core/channels.py @@ -0,0 +1,157 @@ +"""Pure channel logic: secret derivation and MeshCore share-URL round-tripping. + +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. +""" + +from __future__ import annotations + +import secrets +from hashlib import sha256 +from typing import Optional +from urllib.parse import parse_qs, quote, urlsplit + +#: A channel shared secret is exactly 16 bytes (128-bit), per the MeshCore protocol. +CHANNEL_SECRET_BYTES = 16 + +#: The default public channel every MeshCore device ships with on slot 0. +DEFAULT_PUBLIC_NAME = "public" + + +def is_public_name(name: str) -> bool: + """Whether a channel name is a public one whose key derives from the name. + + Args: + name: The channel name. + + Returns: + ``True`` for a ``#``-prefixed name (the firmware's convention for a name-derived + key), ``False`` otherwise. + """ + return name.startswith("#") + + +def derive_secret(name: str) -> bytes: + """Derive a channel's 16-byte secret from its name, exactly as the firmware does. + + A ``#``-prefixed (public) channel — or any channel created without an explicit key — + uses ``sha256(name)[:16]``, so everyone who types the same name shares the same key. + + Args: + name: The channel name (including any leading ``#``). + + Returns: + The derived 16-byte secret. + """ + return sha256(name.encode("utf-8")).digest()[:CHANNEL_SECRET_BYTES] + + +def random_secret() -> bytes: + """Generate a fresh cryptographically-random 16-byte secret for a private channel. + + Returns: + 16 random bytes. + """ + return secrets.token_bytes(CHANNEL_SECRET_BYTES) + + +def normalize_secret(text: str) -> bytes: + """Parse a user-supplied secret (32 hex chars / 16 bytes). + + Tolerates surrounding whitespace, an ``0x`` prefix, and internal spaces so a key + pasted in any common shape is accepted. + + Args: + text: The secret as a hex string. + + Returns: + The 16-byte secret. + + Raises: + ValueError: If ``text`` is not exactly 16 bytes of hexadecimal. + """ + cleaned = text.strip().removeprefix("0x").replace(" ", "").replace(":", "") + data = bytes.fromhex(cleaned) # raises ValueError on odd length / non-hex + if len(data) != CHANNEL_SECRET_BYTES: + raise ValueError("Channel secret must be exactly 16 bytes (32 hex characters).") + return data + + +def full_channel_hash(secret: bytes) -> str: + """Return the full ``sha256(secret)`` hex digest (64 chars) for a channel. + + Its leading byte is the short :func:`channel_hash` MeshCore reports; the rest is shown + only for a complete fingerprint. + + Args: + secret: The channel's 16-byte secret. + + Returns: + A 64-character lowercase hex string. + """ + return sha256(bytes(secret)).hexdigest() + + +def channel_hash(secret: bytes) -> str: + """Return the 2-hex-character channel hash MeshCore shows for a secret. + + This is the leading byte of ``sha256(secret)``, the same short fingerprint the + companion reports for a channel — handy for telling two channels apart at a glance. + + Args: + secret: The channel's 16-byte secret. + + Returns: + A two-character lowercase hex string. + """ + return full_channel_hash(secret)[:2] + + +def share_url(name: str, secret: bytes) -> str: + """Build the MeshCore ``meshcore://channel/add`` share URL for a channel. + + This is the exact scheme the MeshCore mobile apps scan, so a QR code rendered from the + returned string adds the channel directly on another device. + + Args: + name: The channel name. + secret: The channel's 16-byte secret. + + Returns: + A ``meshcore://channel/add?name=…&secret=…`` URL. + """ + return f"meshcore://channel/add?name={quote(name, safe='')}&secret={bytes(secret).hex()}" + + +def parse_share_url(url: str) -> Optional[tuple[str, bytes]]: + """Parse a ``meshcore://channel/add`` link into ``(name, secret)``. + + Args: + url: A channel-add share URL (as scanned from a QR code or pasted in). + + Returns: + The channel name and its 16-byte secret, or ``None`` if the string is not a valid + MeshCore channel-add link. + """ + try: + parts = urlsplit(url.strip()) + except ValueError: + return None + if parts.scheme != "meshcore": + return None + # The path splits as netloc="channel", path="/add"; recombine and compare on the route. + if f"{parts.netloc}{parts.path}".strip("/") != "channel/add": + return None + query = parse_qs(parts.query) + names = query.get("name") + keys = query.get("secret") + if not names or not keys: + return None + try: + return names[0], normalize_secret(keys[0]) + except ValueError: + return None diff --git a/meshtools/tools/base.py b/meshtools/tools/base.py index b4cbf92..35b56e4 100644 --- a/meshtools/tools/base.py +++ b/meshtools/tools/base.py @@ -131,6 +131,19 @@ class Tool(ABC): _REGISTRY: dict[str, Tool] = {} +#: Explicit menu ordering for tool categories. Messaging (Chat + Channels) leads because it +#: holds the first-class, everyday features; the rest follow in a natural workflow order. +#: Categories not listed here sort last, alphabetically, so a new category still appears. +_CATEGORY_ORDER = ["Messaging", "Device", "Diagnostics", "Optimization", "Data"] + + +def _category_rank(category: str) -> int: + """Return a category's sort rank (its position in ``_CATEGORY_ORDER``, else last).""" + try: + return _CATEGORY_ORDER.index(category) + except ValueError: + return len(_CATEGORY_ORDER) + def register(cls: type[Tool]) -> type[Tool]: """Class decorator that instantiates a tool and adds it to the registry. @@ -154,12 +167,19 @@ def register(cls: type[Tool]) -> type[Tool]: def all_tools() -> list[Tool]: - """Return all registered tools sorted by category then order then name. + """Return all registered tools sorted by category rank, then order, then name. + + Categories are ordered by :data:`_CATEGORY_ORDER` (Messaging first), so the first-class + Chat and Channels tools lead the menu; within a category, ``order`` then ``name`` break + ties. Returns: The registered tool instances in stable menu order. """ - return sorted(_REGISTRY.values(), key=lambda t: (t.category, t.order, t.name)) + return sorted( + _REGISTRY.values(), + key=lambda t: (_category_rank(t.category), t.category, t.order, t.name), + ) def get_tool(name: str) -> Optional[Tool]: diff --git a/meshtools/tools/channels.py b/meshtools/tools/channels.py new file mode 100644 index 0000000..853579d --- /dev/null +++ b/meshtools/tools/channels.py @@ -0,0 +1,217 @@ +"""The ``channels`` tool: create, join, and manage mesh channels. + +Interactively it opens a full-screen channel manager (see :mod:`meshtools.ui.channels`) — a +first-class, phone-app-style experience for creating private channels, adding public ``#`` +channels, joining with a key, importing a scanned ``meshcore://`` link, and sharing any +channel as a QR code. On the CLI it exposes ``list``, ``add``, ``join``, ``import``, and +``share`` subcommands for scripted use. + +Channels are also *listed* by the ``chat`` tool for picking a conversation; this tool owns +everything to do with configuring the slots themselves. +""" + +from __future__ import annotations + +from typing import Any, Optional + +import typer +from rich.console import Group +from rich.table import Table +from rich.text import Text + +from ..context import AppContext +from ..core.channels import ( + derive_secret, + normalize_secret, + parse_share_url, + random_secret, + share_url, +) +from .base import Tool, ToolResult, register + + +@register +class ChannelsTool(Tool): + """Create, join, share, and manage the device's mesh channels.""" + + name = "channels" + help = "Create, join, and share mesh channels (with QR codes)." + category = "Messaging" + order = 20 + + async def run(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: + """Open the channel manager (menu) or run a scripted action (CLI). + + Args: + ctx: Shared application context. + params: A ``cli_action`` with its arguments (CLI), or empty for the menu. + + Returns: + A :class:`ToolResult` summarizing what happened. + """ + action = params.get("cli_action") + if action is not None: + return await self._run_cli(ctx, action, params) + + from ..ui.channels import manage_channels + + changes = await manage_channels(ctx) + message = ( + f"[ok]✓[/ok] applied [brand]{changes}[/brand] channel change(s)" + if changes + else None + ) + return ToolResult(summary={"changes": changes}, message=message) + + # -- CLI -------------------------------------------------------------------- + + async def _run_cli( + self, ctx: AppContext, action: str, params: dict[str, Any] + ) -> ToolResult: + """Dispatch a scripted CLI action.""" + if action == "add": + return await self._cli_add(ctx, params) + if action == "join": + return await self._cli_join(ctx, params) + if action == "import": + return await self._cli_import(ctx, params) + if action == "share": + return await self._cli_share(ctx, params) + return await self._cli_list(ctx) + + async def _cli_list(self, ctx: AppContext) -> ToolResult: + """List the configured channel slots.""" + from ..ui.channels import _read_slots + + device = await ctx.device() + slots = await _read_slots(device) + if not slots: + ctx.ui.note("[muted]no channels configured[/muted]") + return ToolResult(summary={"channels": 0}) + table = Table(title="Channels", border_style="muted", expand=False) + table.add_column("Slot", justify="right") + table.add_column("Name") + table.add_column("Type") + table.add_column("Hash") + for slot in slots: + table.add_row( + str(slot.idx), + Text(slot.name, style="brand"), + "public" if slot.is_public else "private", + slot.hash, + ) + ctx.ui.show(table) + return ToolResult(summary={"channels": len(slots)}) + + async def _cli_add(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: + """Add a public, private-random, or explicitly-keyed channel on a slot.""" + device = await ctx.device() + idx = int(params["index"]) + name = str(params["name"]) + if name.startswith("#"): # public: firmware derives the key from the name + await device.set_channel(idx, name, None) + secret = derive_secret(name) + else: + secret = ( + normalize_secret(params["secret"]) if params.get("secret") else random_secret() + ) + await device.set_channel(idx, name, secret) + ctx.ui.note(f"[ok]✓[/ok] channel [brand]{name}[/brand] set on slot {idx}") + self._print_share(ctx, name, secret) + return ToolResult(summary={"index": idx, "name": name}) + + async def _cli_join(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: + """Join a channel from a name and an explicit key.""" + device = await ctx.device() + idx = int(params["index"]) + name = str(params["name"]) + secret = normalize_secret(str(params["secret"])) + await device.set_channel(idx, name, secret) + ctx.ui.note(f"[ok]✓[/ok] joined [brand]{name}[/brand] on slot {idx}") + return ToolResult(summary={"index": idx, "name": name}) + + async def _cli_import(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: + """Import a channel from a ``meshcore://channel/add`` link.""" + parsed = parse_share_url(str(params["url"])) + if parsed is None: + raise typer.BadParameter("not a valid meshcore:// channel link") + name, secret = parsed + device = await ctx.device() + idx = int(params["index"]) + await device.set_channel(idx, name, secret) + ctx.ui.note(f"[ok]✓[/ok] imported [brand]{name}[/brand] on slot {idx}") + return ToolResult(summary={"index": idx, "name": name}) + + 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 + + device = await ctx.device() + idx = int(params["index"]) + slot = next((s for s in await _read_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}) + self._print_share(ctx, slot.name, slot.secret) + return ToolResult(summary={"index": idx, "shared": True}) + + @staticmethod + def _print_share(ctx: AppContext, name: str, secret: bytes) -> None: + """Render a channel's QR code and share URL into the output surface.""" + from ..ui.qr import qr_text + + url = share_url(name, secret) + ctx.ui.show(Group(qr_text(url), Text(""), Text(url, style="accent"))) + + def register_cli(self, app: typer.Typer) -> None: + """Register the ``channels`` subcommand group. + + Args: + app: The Typer application. + """ + from ..cli import run_tool_command + + channels_app = typer.Typer( + help=self.help, no_args_is_help=True, rich_markup_mode="rich" + ) + + @channels_app.command("list", help="List the configured channel slots.") + def _list_cmd() -> None: + run_tool_command(self, {"cli_action": "list"}) + + @channels_app.command("add", help="Add a channel (# name = public; else private).") + def _add_cmd( + index: int = typer.Argument(..., help="Channel slot index."), + name: str = typer.Argument(..., help="Channel name (leading # = public)."), + secret: Optional[str] = typer.Option( + None, "--secret", help="32-hex-char key (private only; random if omitted)." + ), + ) -> None: + run_tool_command( + self, {"cli_action": "add", "index": index, "name": name, "secret": secret} + ) + + @channels_app.command("join", help="Join a channel with its name and key.") + def _join_cmd( + index: int = typer.Argument(..., help="Channel slot index."), + name: str = typer.Argument(..., help="Channel name."), + secret: str = typer.Argument(..., help="32-hex-char (16-byte) key."), + ) -> None: + run_tool_command( + self, {"cli_action": "join", "index": index, "name": name, "secret": secret} + ) + + @channels_app.command("import", help="Import a meshcore:// channel link.") + def _import_cmd( + index: int = typer.Argument(..., help="Channel slot index."), + url: str = typer.Argument(..., help="A meshcore://channel/add link."), + ) -> None: + run_tool_command(self, {"cli_action": "import", "index": index, "url": url}) + + @channels_app.command("share", help="Print a channel's share link and QR code.") + def _share_cmd( + index: int = typer.Argument(..., help="Channel slot index."), + ) -> None: + run_tool_command(self, {"cli_action": "share", "index": index}) + + app.add_typer(channels_app, name=self.name) diff --git a/meshtools/tools/chat.py b/meshtools/tools/chat.py index 95e3b8b..edcd535 100644 --- a/meshtools/tools/chat.py +++ b/meshtools/tools/chat.py @@ -3,7 +3,7 @@ Interactively it opens a conversation picker and then a live, full-screen chat (see :mod:`meshtools.ui.chat`) where sent and received messages stream together. On the CLI it exposes ``send``, ``history``, and ``list`` subcommands for scripted use. Channels are only -*listed* here for picking; creating and editing channel slots lives in the ``config`` tool. +*listed* here for picking; creating and editing channel slots lives in the ``channels`` tool. Every inbound message is recorded to history by the always-on :class:`~meshtools.services.chat_service.ChatService`, and outbound messages are recorded on @@ -383,10 +383,10 @@ async def _read_channels(device: Device) -> list[Conversation]: if channel: name = str(channel.get("channel_name") or idx) conversations.append( - Conversation(label=f"#{name}", is_channel=True, channel_idx=idx) + Conversation(label=name, is_channel=True, channel_idx=idx) ) if not any(c.channel_idx == 0 for c in conversations): - conversations.insert(0, Conversation(label="#public", is_channel=True, channel_idx=0)) + conversations.insert(0, Conversation(label="Public", is_channel=True, channel_idx=0)) return conversations diff --git a/meshtools/ui/channels.py b/meshtools/ui/channels.py new file mode 100644 index 0000000..3459e14 --- /dev/null +++ b/meshtools/ui/channels.py @@ -0,0 +1,477 @@ +"""The interactive channel manager, rendered in the full-screen session. + +This is the user-facing channel-management experience for the ``channels`` tool: a live +list of the device's channel slots with, for each, a detail view that shows the sharable +QR code and key, renames or re-keys it, opens it in chat, or clears it. New channels are +created four ways — a fresh private channel (random key), a public ``#`` channel (key +derived from the name), joining by pasting a key, or importing a scanned ``meshcore://`` +link. Every change is written to the device immediately (like a phone app), so the list you +see always reflects the radio. + +The module sits in the UI layer but, like the config editor and chat screen, is allowed to +depend on the context and services; it owns no persistence of its own. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING, Optional + +from rich.console import Group +from rich.table import Table +from rich.text import Text + +from ..core.channels import ( + channel_hash, + derive_secret, + full_channel_hash, + is_public_name, + normalize_secret, + parse_share_url, + random_secret, + share_url, +) +from ..core.connection import Device +from ..core.models import Conversation +from .qr import qr_text +from .tui import Choice, Separator + +if TYPE_CHECKING: + from ..context import AppContext + +#: How many channel slots the firmware exposes (0-based); matches the chat/backup probes. +_MAX_CHANNELS = 8 + +# Top-menu action sentinels (distinct from a plain slot index, which selects that channel). +_CREATE = "__create__" +_PUBLIC = "__public__" +_JOIN = "__join__" +_IMPORT = "__import__" +_REORDER = "__reorder__" +_BACK = "__back__" + +# Channel-detail action sentinels. +_QR = "qr" +_KEY = "key" +_CHAT = "chat" +_EDIT = "edit" +_CLEAR = "clear" + + +@dataclass(slots=True) +class ChannelSlot: + """One configured channel slot read back from the device. + + Attributes: + idx: The 0-based slot index. + name: The channel's name. + secret: The channel's 16-byte shared secret. + """ + + idx: int + name: str + secret: bytes + + @property + def is_public(self) -> bool: + """Whether this channel's key is derived from its name (a public channel).""" + return is_public_name(self.name) or self.secret == derive_secret(self.name) + + @property + def hash(self) -> str: + """The channel's two-character hash fingerprint (the leading byte MeshCore shows).""" + return channel_hash(self.secret) + + @property + def full_hash(self) -> str: + """The channel's complete ``sha256(secret)`` digest; its first byte is :attr:`hash`.""" + return full_channel_hash(self.secret) + + @property + def conversation(self) -> Conversation: + """A :class:`~meshtools.core.models.Conversation` for opening this channel in chat.""" + return Conversation(label=self.name, is_channel=True, channel_idx=self.idx) + + +async def manage_channels(ctx: "AppContext") -> int: + """Run the interactive channel manager until the user backs out. + + Args: + ctx: Shared application context (provides the connected device and UI surface). + + Returns: + The number of channels created, changed, or cleared during the session. + """ + device = await ctx.device() + changes = 0 + highlight: Optional[object] = None + + while True: + slots = await _read_slots(device) + choice = await _main_menu(ctx, slots, default=highlight) + if choice in (None, _BACK): + return changes + highlight = choice + if choice == _CREATE: + changes += await _create_private(ctx, device, slots) + elif choice == _PUBLIC: + changes += await _add_public(ctx, device, slots) + elif choice == _JOIN: + changes += await _join_with_key(ctx, device, slots) + elif choice == _IMPORT: + changes += await _import_link(ctx, device, slots) + elif choice == _REORDER: + changes += await _reorder_channels(ctx, device, slots) + else: # an existing slot index + slot = next((s for s in slots if s.idx == choice), None) + if slot is not None: + changes += await _channel_detail(ctx, device, slot) + # unreachable + + +# --- reading ----------------------------------------------------------------- + + +async def _read_slots(device: Device) -> list[ChannelSlot]: + """Probe the channel slots and return the configured ones, in index order. + + Args: + device: The connected device to query. + + Returns: + One :class:`ChannelSlot` per configured slot (an empty slot is skipped). + """ + slots: list[ChannelSlot] = [] + for idx in range(_MAX_CHANNELS): + try: + payload = await device.get_channel(idx) + except Exception: # noqa: BLE001 - firmware may not support channel reads + break + if payload and payload.get("channel_name"): + slots.append( + ChannelSlot( + idx=idx, + name=str(payload["channel_name"]), + secret=bytes(payload.get("channel_secret") or b"\x00" * 16), + ) + ) + return slots + + +def _next_free_slot(slots: list[ChannelSlot]) -> Optional[int]: + """Return the lowest unused slot index, or ``None`` when every slot is full.""" + used = {s.idx for s in slots} + return next((i for i in range(_MAX_CHANNELS) if i not in used), None) + + +def _slot_label(slot: ChannelSlot) -> str: + """Format a channel for a menu row: name, public/private, and hash (no slot index).""" + lock = "＃ public " if slot.is_public else "🔒 private" + return f"{slot.name:<18.18} {lock} hash {slot.hash}" + + +# --- menus ------------------------------------------------------------------- + + +async def _main_menu( + ctx: "AppContext", slots: list[ChannelSlot], *, default: object = None +) -> object: + """Show the channel list and the add-a-channel actions; return the chosen value.""" + items: list = [Separator("── Channels ──")] + if slots: + for slot in slots: + items.append(Choice(title=_slot_label(slot), value=slot.idx)) + else: + items.append(Separator(" (no channels configured yet)")) + + if len(slots) > 1: + items.append(Separator("── Channel list ──")) + items.append(Choice(title="↕ Reorder channels", value=_REORDER)) + + items.append(Separator("── Add a channel ──")) + items.append(Choice(title="＋ New private channel (random key)", value=_CREATE)) + items.append(Choice(title="＃ Public channel (key from its name)", value=_PUBLIC)) + items.append(Choice(title="🔑 Join a channel with its key", value=_JOIN)) + items.append(Choice(title="🔗 Import a meshcore:// link", value=_IMPORT)) + items.append(Separator(" ")) + items.append(Choice(title="Back", value=_BACK)) + + choice = await ctx.ui.select("Channels", items, default=default) + return _BACK if choice is None else choice + + +async def _channel_detail(ctx: "AppContext", device: Device, slot: ChannelSlot) -> int: + """Show one channel's actions (QR, key, chat, rename, clear); return changes made.""" + kind = "public" if slot.is_public else "private" + while True: + choice = await ctx.ui.select( + f"{slot.name} ({kind}, hash {slot.hash})", + [ + Choice(title="📱 Show QR code (share)", value=_QR), + Choice(title="🔑 Show key", value=_KEY), + Choice(title="💬 Open in chat", value=_CHAT), + Choice(title="✎ Rename / change key", value=_EDIT), + Choice(title="🗑 Clear this slot", value=_CLEAR), + Choice(title="Back", value=_BACK), + ], + ) + if choice in (None, _BACK): + return 0 + if choice == _QR: + await _show_share(ctx, slot.name, slot.secret) + elif choice == _KEY: + await _show_key(ctx, slot) + elif choice == _CHAT: + await _open_chat(ctx, slot) + elif choice == _EDIT: + if await _edit(ctx, device, slot): + return 1 + elif choice == _CLEAR: + if await _clear(ctx, device, slot): + return 1 + + +# --- create / join flows ----------------------------------------------------- + + +async def _create_private( + ctx: "AppContext", device: Device, slots: list[ChannelSlot] +) -> int: + """Create a private channel with a fresh random key on the next free slot.""" + idx = await _pick_free_slot(ctx, slots) + if idx is None: + return 0 + name = await ctx.ui.text("Channel name:", validate=_nonblank) + if not name: + return 0 + secret = random_secret() + await device.set_channel(idx, name.strip(), secret) + ctx.ui.note(f"[ok]✓[/ok] created private channel [brand]{name.strip()}[/brand]") + await _show_share(ctx, name.strip(), secret, intro="Share this channel:") + return 1 + + +async def _add_public(ctx: "AppContext", device: Device, slots: list[ChannelSlot]) -> int: + """Create a public channel whose key is derived from its (``#``-prefixed) name.""" + idx = await _pick_free_slot(ctx, slots) + if idx is None: + return 0 + raw = await ctx.ui.text( + "Public channel name:", + help_text="A leading # is added automatically; the key is derived from the name.", + validate=_nonblank, + ) + if not raw: + return 0 + name = raw.strip() + if not name.startswith("#"): + name = f"#{name}" + secret = derive_secret(name) # what the firmware will compute; kept for the QR/share + await device.set_channel(idx, name, None) # None => firmware derives the key from name + ctx.ui.note(f"[ok]✓[/ok] added public channel [brand]{name}[/brand]") + await _show_share(ctx, name, secret, intro="Share this channel:") + return 1 + + +async def _join_with_key( + ctx: "AppContext", device: Device, slots: list[ChannelSlot] +) -> int: + """Join an existing private channel by entering its name and 16-byte key.""" + idx = await _pick_free_slot(ctx, slots) + if idx is None: + return 0 + name = await ctx.ui.text("Channel name:", validate=_nonblank) + if not name: + return 0 + key = await ctx.ui.text( + "Channel key (32 hex characters / 16 bytes):", validate=_valid_secret + ) + if not key: + return 0 + secret = normalize_secret(key) + await device.set_channel(idx, name.strip(), secret) + ctx.ui.note(f"[ok]✓[/ok] joined [brand]{name.strip()}[/brand]") + return 1 + + +async def _import_link( + ctx: "AppContext", device: Device, slots: list[ChannelSlot] +) -> int: + """Import a channel from a pasted ``meshcore://channel/add`` link.""" + idx = await _pick_free_slot(ctx, slots) + if idx is None: + return 0 + url = await ctx.ui.text( + "Paste a meshcore:// channel link:", validate=_valid_link + ) + if not url: + return 0 + parsed = parse_share_url(url) + if parsed is None: # pragma: no cover - guarded by the validator + ctx.ui.note("[err]not a valid channel link[/err]") + return 0 + name, secret = parsed + await device.set_channel(idx, name, secret) + ctx.ui.note(f"[ok]✓[/ok] imported [brand]{name}[/brand]") + return 1 + + +async def _edit(ctx: "AppContext", device: Device, slot: ChannelSlot) -> bool: + """Rename and/or re-key an existing channel; return whether it changed.""" + name = await ctx.ui.text("Channel name:", default=slot.name, validate=_nonblank) + if name is None: + return False + name = name.strip() + key = await ctx.ui.text( + "Channel key (32 hex chars; blank to derive from the name):", + default="" if slot.is_public else slot.secret.hex(), + validate=_optional_secret, + ) + if key is None: + return False + secret = normalize_secret(key) if key.strip() else None + await device.set_channel(slot.idx, name, secret) + ctx.ui.note(f"[ok]✓[/ok] updated channel [brand]{name}[/brand]") + return True + + +async def _clear(ctx: "AppContext", device: Device, slot: ChannelSlot) -> bool: + """Clear a channel slot after confirmation; return whether it was cleared.""" + if not await ctx.ui.confirm( + f"Clear [brand]{slot.name}[/brand]? This removes the channel from this device.", + default=False, + ): + return False + await device.set_channel(slot.idx, "", None) # empty name => the slot reads as unused + ctx.ui.note(f"[warn]cleared channel {slot.name}[/warn]") + return True + + +# --- reordering -------------------------------------------------------------- + + +async def _write_slot(device: Device, idx: int, slot: ChannelSlot) -> None: + """Write ``slot``'s contents into slot ``idx`` (public channels re-derive their key).""" + secret = None if slot.is_public else slot.secret + await device.set_channel(idx, slot.name, secret) + + +async def _apply_order( + device: Device, slots: list[ChannelSlot], order: list[int] +) -> int: + """Rewrite the channel slots so they display in ``order``; return the writes made. + + ``slots`` is the current list in slot-index order and ``order`` is a permutation of its + positions (as returned by the reorder screen). The channels are re-laid across the same + physical slot indices, lowest first, so their on-device order matches the new display + order. Only slots whose occupant actually changes are written. Reads come from the + in-memory snapshot, so the interleaved writes never clobber a not-yet-placed channel. + """ + indices = sorted(s.idx for s in slots) # the physical slots to fill, ascending + changes = 0 + for target_idx, pos in zip(indices, order, strict=True): + slot = slots[pos] + if slot.idx == target_idx: + continue # already in place; no write needed + await _write_slot(device, target_idx, slot) + changes += 1 + return changes + + +async def _reorder_channels( + ctx: "AppContext", device: Device, slots: list[ChannelSlot] +) -> int: + """Let the user drag channels into a new order with the arrows; return writes made.""" + if len(slots) < 2: # pragma: no cover - the menu only offers reorder with 2+ channels + return 0 + labels = [_slot_label(slot) for slot in slots] + order = await ctx.ui.reorder("Reorder channels", labels) + if not order or order == list(range(len(slots))): + return 0 # cancelled or left unchanged + return await _apply_order(device, slots, order) + + +# --- shared views ------------------------------------------------------------ + + +async def _show_share( + ctx: "AppContext", name: str, secret: bytes, *, intro: str = "Scan to add this channel:" +) -> None: + """Show the channel's QR code and its share URL in a dismissable window.""" + url = share_url(name, secret) + body = Group( + Text(intro, style="muted"), + Text(""), + qr_text(url), + Text(""), + Text(url, style="accent"), + ) + await ctx.ui.view(body, title=f"Share {name}", footer_hint="Esc back") + + +def _hash_text(full_hash: str) -> Text: + """Render a full channel hash with its first byte (the MeshCore fingerprint) highlighted.""" + text = Text(no_wrap=True) + text.append(full_hash[:2], style="brand") # the leading byte the companion app reports + text.append(full_hash[2:], style="muted") + return text + + +async def _show_key(ctx: "AppContext", slot: ChannelSlot) -> None: + """Show a channel's name, key, hash, and share URL in a small table.""" + table = Table(border_style="muted", expand=False, show_header=False) + table.add_column("", style="muted") + table.add_column("") + table.add_row("Name", Text(slot.name, style="brand")) + table.add_row("Type", "public (key from name)" if slot.is_public else "private") + table.add_row("Hash", _hash_text(slot.full_hash)) + table.add_row("Key", Text(slot.secret.hex(), style="warn")) + table.add_row("Link", Text(share_url(slot.name, slot.secret), style="accent")) + await ctx.ui.view(table, title=f"Key — {slot.name}", footer_hint="Esc back") + + +async def _open_chat(ctx: "AppContext", slot: ChannelSlot) -> None: + """Open this channel in the live chat screen.""" + from .chat import open_chat + + await open_chat(ctx, slot.conversation) + + +async def _pick_free_slot(ctx: "AppContext", slots: list[ChannelSlot]) -> Optional[int]: + """Return the next free slot, warning (and returning ``None``) if all are full.""" + idx = _next_free_slot(slots) + if idx is None: + await ctx.ui.view( + Text.from_markup( + "[warn]All channel slots are full.[/warn] Clear one first, then try again." + ), + title="No free slot", + ) + return idx + + +# --- validators -------------------------------------------------------------- + + +def _nonblank(text: str) -> bool | str: + """Require a non-empty name.""" + return bool(text.strip()) or "Enter a channel name." + + +def _valid_secret(text: str) -> bool | str: + """Require a valid 16-byte hex key.""" + try: + normalize_secret(text) + return True + except ValueError as exc: + return str(exc) + + +def _optional_secret(text: str) -> bool | str: + """Accept a blank key (derive from name) or a valid 16-byte hex key.""" + return True if not text.strip() else _valid_secret(text) + + +def _valid_link(text: str) -> bool | str: + """Require a parseable ``meshcore://channel/add`` link.""" + return parse_share_url(text) is not None or "Not a valid meshcore:// channel link." diff --git a/meshtools/ui/config_editor.py b/meshtools/ui/config_editor.py index 5d07501..e0af88f 100644 --- a/meshtools/ui/config_editor.py +++ b/meshtools/ui/config_editor.py @@ -2,8 +2,9 @@ Drives the menu flow for the ``config`` tool: it presents the device's settings (each row showing its current value and any staged change), lets the user stage changes setting-by- -setting with type-aware prompts and validation, and handles custom variables, channels, -backup/restore and the destructive "danger zone". It returns an operation list for +setting with type-aware prompts and validation, and handles custom variables, +backup/restore and the destructive "danger zone". (Channels have their own first-class +manager — see the ``channels`` tool.) It returns an operation list for :class:`~meshtools.tools.config.ConfigTool` to execute and log; it performs no device writes itself (except danger-zone actions, which run immediately and show their result at once). """ @@ -30,7 +31,6 @@ from .tui import Choice, Separator # Menu action sentinels (distinct from setting keys, which are plain strings). _PRESETS = "__presets__" _CUSTOM = "__custom__" -_CHANNELS = "__channels__" _BACKUP = "__backup__" _RESTORE = "__restore__" _DANGER = "__danger__" @@ -75,8 +75,6 @@ async def edit_config(ctx: AppContext) -> Optional[list[tuple]]: await _stage_preset(ctx, pending) elif choice == _CUSTOM: await _stage_custom_var(ctx, custom, extra_ops) - elif choice == _CHANNELS: - await _stage_channel(ctx, extra_ops) elif choice == _BACKUP: await _stage_backup(ctx, extra_ops) elif choice == _RESTORE: @@ -168,7 +166,6 @@ async def _main_menu( items.append(Choice(title="View current config (full table)", value=_VIEW)) items.append(Choice(title="Radio presets (standard configs)", value=_PRESETS)) items.append(Choice(title="Custom / experimental vars", value=_CUSTOM)) - items.append(Choice(title="Channels", value=_CHANNELS)) items.append(Choice(title="Backup to file", value=_BACKUP)) items.append(Choice(title="Restore from file", value=_RESTORE)) items.append(Choice(title="⚠ Danger zone", value=_DANGER)) @@ -267,24 +264,6 @@ async def _stage_custom_var( extra_ops.append(("set_custom", key.strip(), value)) -async def _stage_channel(ctx: AppContext, extra_ops: list[tuple]) -> None: - """Prompt for a channel slot's name/secret and stage a set operation.""" - idx_raw = await ctx.ui.text("Channel index:", default="0", validate=_is_int) - if idx_raw is None: - return - name = await ctx.ui.text("Channel name (leading # derives the secret from the name):") - if not name: - return - secret_hex = await ctx.ui.text( - "Secret (32 hex chars / 16 bytes; blank to derive from name):", - validate=_is_optional_secret, - ) - if secret_hex is None: - return - secret = bytes.fromhex(secret_hex) if secret_hex else None - extra_ops.append(("set_channel", int(idx_raw), name, secret)) - - async def _stage_backup(ctx: AppContext, extra_ops: list[tuple]) -> None: """Prompt for a backup destination path and stage the operation.""" path = await ctx.ui.path("Write backup to:", default="meshtools-config.toml") @@ -372,15 +351,6 @@ async def _confirm_typed(ctx: AppContext, word: str, *, warning: str = "") -> bo # --- validators -------------------------------------------------------------- -def _is_int(text: str) -> bool | str: - """Validate that ``text`` is an integer.""" - try: - int(text) - return True - except ValueError: - return "Enter a whole number." - - def _is_hex(text: str) -> bool | str: """Validate that ``text`` is a hex string.""" try: @@ -388,13 +358,3 @@ def _is_hex(text: str) -> bool | str: return True except ValueError: return "Enter hex characters only." - - -def _is_optional_secret(text: str) -> bool | str: - """Validate an optional 16-byte hex channel secret.""" - if text == "": - return True - try: - return len(bytes.fromhex(text)) == 16 or "Secret must be exactly 16 bytes (32 hex chars)." - except ValueError: - return "Enter hex characters only." diff --git a/meshtools/ui/qr.py b/meshtools/ui/qr.py new file mode 100644 index 0000000..098ef56 --- /dev/null +++ b/meshtools/ui/qr.py @@ -0,0 +1,56 @@ +"""Render a string as a scannable QR code in the terminal, using half-block characters. + +Each character cell stacks two vertical modules — ``▀`` (upper only), ``▄`` (lower only), +``█`` (both), space (neither) — so a code renders at half the row height of a full-block +one and comfortably fits an interactive window. Modules are always drawn black on a white +background (with the mandatory quiet-zone border), so a phone camera reads the code no +matter what colour theme the terminal itself is using. +""" + +from __future__ import annotations + +from rich.text import Text + +#: Quiet-zone width (modules) around the code. Scanners need at least 4 to lock on. +_BORDER = 4 + +#: Half-block glyph for each (top-module-dark, bottom-module-dark) pair. Drawn as black +#: ink, so a dark module is black and a light module shows the white cell background. +_GLYPH = { + (True, True): "█", # █ full block + (True, False): "▀", # ▀ upper half + (False, True): "▄", # ▄ lower half + (False, False): " ", +} + +#: Constant style for every glyph: black ink on a white field. Naming both ends means the +#: contrast holds regardless of the surrounding terminal palette. +_STYLE = "black on white" + + +def qr_text(data: str, *, error: str = "m") -> Text: + """Render ``data`` as a QR code built from half-block characters. + + Args: + data: The string to encode (e.g. a ``meshcore://channel/add`` share URL). + error: QR error-correction level — ``l``/``m``/``q``/``h`` (default ``m``). + + Returns: + A Rich :class:`~rich.text.Text` (no-wrap) whose lines draw the code, ready to hand + to ``ctx.ui.view`` / ``ctx.ui.show``. + """ + import segno + + qr = segno.make(data, error=error) + rows = [[bool(v) for v in row] for row in qr.matrix_iter(border=_BORDER)] + # Pair rows top-to-bottom; pad an odd final row with light (white) modules so the last + # half-block cell renders cleanly. + if len(rows) % 2: + rows.append([False] * len(rows[0])) + + text = Text(no_wrap=True) + for top, bottom in zip(rows[0::2], rows[1::2], strict=True): + line = "".join(_GLYPH[(t, b)] for t, b in zip(top, bottom, strict=True)) + text.append(line, style=_STYLE) + text.append("\n") + return text diff --git a/meshtools/ui/surface.py b/meshtools/ui/surface.py index 21d219e..748d503 100644 --- a/meshtools/ui/surface.py +++ b/meshtools/ui/surface.py @@ -57,6 +57,10 @@ class Ui: """Prompt the user to choose one item; return its value or ``None`` if cancelled.""" raise NotImplementedError + async def reorder(self, title: str, labels: list[str]) -> list[int]: + """Let the user rearrange rows with the arrows; return the new order of row indices.""" + raise NotImplementedError + async def text( self, title: str, @@ -129,6 +133,10 @@ class PlainUi(Ui): """Unsupported in scripted CLI mode.""" raise self._no_prompt() + async def reorder(self, title: str, labels: list[str]) -> list[int]: + """Unsupported in scripted CLI mode.""" + raise self._no_prompt() + async def text( self, title: str, @@ -239,6 +247,10 @@ class TuiUi(Ui): """Delegate to the session's select screen.""" return await self.session.select(title, items, default=default) + async def reorder(self, title: str, labels: list[str]) -> list[int]: + """Delegate to the session's reorder screen.""" + return await self.session.reorder(title, labels) + async def text( self, title: str, diff --git a/meshtools/ui/tui/select.py b/meshtools/ui/tui/select.py index e1acd52..1c3d365 100644 --- a/meshtools/ui/tui/select.py +++ b/meshtools/ui/tui/select.py @@ -159,6 +159,82 @@ class SelectScreen(Screen): self._index = 0 +class ReorderScreen(Screen): + """A list whose rows the user rearranges in place with the arrow keys. + + Move the cursor with ↑/↓; press Enter to *grab* the highlighted row, then ↑/↓ carry it + up and down the list; press Enter again to *drop* it. The screen stays open until Esc, + at which point it resolves with the final order as a list of the original row indices + (so ``[2, 0, 1]`` means "the row that started third is now first"). + """ + + def __init__(self, title: str, labels: list[str]) -> None: + """Build a reorder screen. + + Args: + title: Heading shown above the list. + labels: The row labels, in their current order. + """ + super().__init__() + self.title = title + self._labels = list(labels) + # order[position] == the label's original index; a moved row carries its index along. + self._order = list(range(len(labels))) + self._index = 0 + self._grabbed = False + + @property + def footer_hint(self) -> str: # type: ignore[override] + """Key hint, phrased for whether a row is currently grabbed.""" + if self._grabbed: + return "↑↓ move row · Enter drop · Esc done" + return "↑↓ choose · Enter grab · Esc done" + + def render_body(self, width: int) -> list[str]: + """Render each row, marking the cursor (and, when grabbed, the moving row).""" + lines: list[str] = [] + for pos, orig in enumerate(self._order): + is_cursor = pos == self._index + if is_cursor and self._grabbed: + pointer, style = "▸ ", "brand" + elif is_cursor: + pointer, style = "❯ ", "brand" + else: + pointer, style = " ", "" + text = Text(pointer + self._labels[orig], style=style, no_wrap=True, + overflow="ellipsis") + text.truncate(width) + lines.append(render_to_ansi(text, width)) + self._cursor = self._index + return lines + + def cursor_line(self) -> Optional[int]: + """Return the body line index of the cursor row, so the session keeps it in view.""" + return getattr(self, "_cursor", None) + + def handle(self, action: str, data: str = "") -> None: + """Move the cursor, carry a grabbed row, toggle grab, or finish on Esc.""" + n = len(self._order) + if action == "up": + if self._grabbed and self._index > 0: + self._order[self._index - 1], self._order[self._index] = ( + self._order[self._index], self._order[self._index - 1]) + self._index -= 1 + elif not self._grabbed and n: + self._index = (self._index - 1) % n + elif action == "down": + if self._grabbed and self._index < n - 1: + self._order[self._index + 1], self._order[self._index] = ( + self._order[self._index], self._order[self._index + 1]) + self._index += 1 + elif not self._grabbed and n: + self._index = (self._index + 1) % n + elif action == "enter": + self._grabbed = not self._grabbed + elif action == "escape": + self.resolve(list(self._order)) + + def _cursor_line(rows: list, selected: Optional[Choice], filtered: bool) -> Optional[int]: """Compute the rendered body line index of the selected row. diff --git a/meshtools/ui/tui/session.py b/meshtools/ui/tui/session.py index 489e81b..00ff57a 100644 --- a/meshtools/ui/tui/session.py +++ b/meshtools/ui/tui/session.py @@ -28,7 +28,7 @@ from . import frame from .progress import TuiProgress from .prompt import AutocompleteScreen, ConfirmScreen, TextScreen, Validator from .screen import CANCEL, Screen, ScrollScreen -from .select import Choice, SelectScreen, Separator +from .select import Choice, ReorderScreen, SelectScreen, Separator #: Maps prompt_toolkit keys to the normalized action names screens understand. _KEY_ACTIONS: dict[Any, str] = { @@ -125,6 +125,11 @@ class TuiSession: result = await self.run_screen(SelectScreen(title, items, default=default)) return None if result is CANCEL else result + async def reorder(self, title: str, labels: list[str]) -> list[int]: + """Show a drag-with-arrows reorder screen; return the final order of row indices.""" + result = await self.run_screen(ReorderScreen(title, labels)) + return list(range(len(labels))) if result is CANCEL else result + async def text( self, title: str, @@ -312,5 +317,6 @@ __all__ = [ "Choice", "Separator", "SelectScreen", + "ReorderScreen", "ScrollScreen", ] diff --git a/pyproject.toml b/pyproject.toml index 39f8e8e..16553c3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -19,6 +19,7 @@ dependencies = [ "rich>=13.7", # modern console rendering (progress, tables, panels) "prompt_toolkit>=3.0", # full-screen interactive TUI (terminal, input, resize) "plotly>=5.20", # interactive self-contained HTML charts + "segno>=1.6", # pure-Python QR codes (channel share links) "pyvis>=0.3.2", # interactive HTML mesh graphs "networkx>=3.2", # graph model + path scoring "tomli>=2.0; python_version < '3.11'", # TOML config reader on 3.10 diff --git a/tests/test_channels.py b/tests/test_channels.py new file mode 100644 index 0000000..3957eeb --- /dev/null +++ b/tests/test_channels.py @@ -0,0 +1,226 @@ +"""Tests for the channel-management feature: pure logic, QR rendering, and the tool. + +The logic (secret derivation, share-URL round-tripping) is hardware-free; the tool actions +run against the :class:`MockDevice` simulator and a temporary database. +""" + +from __future__ import annotations + +import re +from hashlib import sha256 +from pathlib import Path + +import pytest +from rich.console import Console + +from meshtools.context import AppContext +from meshtools.core.admin_store import AdminStore +from meshtools.core.channels import ( + CHANNEL_SECRET_BYTES, + channel_hash, + derive_secret, + full_channel_hash, + normalize_secret, + parse_share_url, + random_secret, + share_url, +) +from meshtools.core.config import Settings +from meshtools.core.device_store import DeviceStore +from meshtools.persistence.repository import Repository +from meshtools.tools.channels import ChannelsTool +from meshtools.ui.channels import ChannelSlot, _apply_order, _next_free_slot, _read_slots +from meshtools.ui.qr import qr_text + +_ANSI = re.compile(r"\x1b\[[0-9;]*m") + + +# -- pure channel logic ------------------------------------------------------- + + +def test_derive_secret_matches_firmware_formula() -> None: + """A derived key is sha256(name)[:16] — the firmware's public-channel scheme.""" + assert derive_secret("#public") == sha256(b"#public").digest()[:16] + assert len(derive_secret("#anything")) == CHANNEL_SECRET_BYTES + assert derive_secret("#a") == derive_secret("#a") # deterministic + + +def test_random_secret_is_16_unique_bytes() -> None: + """A random private key is 16 bytes and (practically) never repeats.""" + a, b = random_secret(), random_secret() + assert len(a) == CHANNEL_SECRET_BYTES and len(b) == CHANNEL_SECRET_BYTES + assert a != b + + +def test_normalize_secret_accepts_common_shapes() -> None: + """A pasted key is accepted with 0x, spaces, or colons; length is enforced.""" + raw = "000102030405060708090a0b0c0d0e0f" + expected = bytes(range(16)) + assert normalize_secret(raw) == expected + assert normalize_secret("0x" + raw) == expected + assert normalize_secret("0001 0203 0405 0607 0809 0a0b 0c0d 0e0f") == expected + assert normalize_secret("00:01:02:03:04:05:06:07:08:09:0a:0b:0c:0d:0e:0f") == expected + + +@pytest.mark.parametrize("bad", ["", "abcd", "zz" * 16, "00" * 15, "00" * 17]) +def test_normalize_secret_rejects_bad_keys(bad: str) -> None: + """Non-hex or wrong-length keys raise, so the prompt can re-ask.""" + with pytest.raises(ValueError): + normalize_secret(bad) + + +def test_channel_hash_is_two_hex_chars() -> None: + """The channel hash is the leading byte of sha256(secret), as the companion reports.""" + secret = bytes(range(16)) + assert channel_hash(secret) == sha256(secret).hexdigest()[:2] + assert len(channel_hash(secret)) == 2 + + +def test_full_channel_hash_is_the_whole_digest_led_by_the_short_hash() -> None: + """The full hash is the complete sha256(secret) digest, opening with the short hash.""" + secret = bytes(range(16)) + full = full_channel_hash(secret) + assert full == sha256(secret).hexdigest() + assert len(full) == 64 + assert full[:2] == channel_hash(secret) # the highlighted first byte + + +def test_share_url_round_trips_with_encoding() -> None: + """A share URL builds and parses back to the same name and secret, encoding included.""" + name, secret = "Ops Team #1", bytes(range(16)) + url = share_url(name, secret) + assert url.startswith("meshcore://channel/add?") + assert "Ops%20Team" in url # the space is percent-encoded + parsed = parse_share_url(url) + assert parsed == (name, secret) + + +def test_parse_share_url_rejects_non_channel_links() -> None: + """Anything that isn't a valid channel-add link parses to None.""" + assert parse_share_url("https://example.com") is None + assert parse_share_url("meshcore://contact/add?name=x&public_key=ab") is None + assert parse_share_url("meshcore://channel/add?name=x") is None # missing secret + assert parse_share_url("meshcore://channel/add?name=x&secret=nothex") is None + assert parse_share_url("meshcore://channel/add?secret=" + "00" * 16) is None # no name + + +# -- QR rendering ------------------------------------------------------------- + + +def test_qr_text_renders_square_with_quiet_zone() -> None: + """qr_text produces block rows framed by a light quiet zone, wider than the data.""" + console = Console(force_terminal=True, width=120, file=__import__("io").StringIO()) + console.print(qr_text("meshcore://channel/add?name=Test&secret=" + "ab" * 16)) + plain = _ANSI.sub("", console.file.getvalue()).rstrip("\n") + lines = plain.splitlines() + assert lines, "QR produced no output" + # A version-appropriate QR for this URL is at least ~25 modules wide plus an 8-module + # quiet zone; the finder pattern makes the code non-trivial. + assert len(lines[0]) >= 30 + assert any("█" in line for line in lines) # dark modules were drawn + + +# -- channel slot model ------------------------------------------------------- + + +def test_channel_slot_classifies_public_and_private() -> None: + """A #-named or name-derived slot reads as public; a random-key slot as private.""" + public = ChannelSlot(idx=0, name="#public", secret=derive_secret("#public")) + assert public.is_public + derived = ChannelSlot(idx=1, name="general", secret=derive_secret("general")) + assert derived.is_public # key derived from the name, even without a leading # + private = ChannelSlot(idx=2, name="Ops", secret=bytes(range(16))) + assert not private.is_public + assert private.conversation.is_channel and private.conversation.channel_idx == 2 + assert private.conversation.label == "Ops" # raw name, no forced leading '#' + + +def test_next_free_slot_finds_gaps_and_full() -> None: + """The next free slot skips used indices and is None when every slot is taken.""" + slots = [ChannelSlot(idx=0, name="a", secret=b"\x00" * 16), + ChannelSlot(idx=2, name="c", secret=b"\x00" * 16)] + assert _next_free_slot(slots) == 1 + full = [ChannelSlot(idx=i, name=str(i), secret=b"\x00" * 16) for i in range(8)] + assert _next_free_slot(full) is None + + +async def test_apply_order_relays_channels_into_new_positions(ctx: AppContext) -> None: + """Applying a new order re-lays channels across the same slots in display order.""" + device = await ctx.device() + await device.set_channel(0, "Alpha", bytes(range(16))) # private + 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) + # 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)} + 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)) + + +# -- the tool against the simulator ------------------------------------------- + + +@pytest.fixture() +def ctx(tmp_path: Path) -> AppContext: + """A mock-backed application context with the plain (console) UI surface.""" + settings = Settings(config_dir=tmp_path, db_path=tmp_path / "chan.db") + context = AppContext( + console=Console(file=__import__("io").StringIO()), + settings=settings, + repo=Repository(settings.db_path), + device_store=DeviceStore(tmp_path / "devices.json"), + admin_store=AdminStore(tmp_path / "admin.json"), + mock=True, + ) + yield context + context.repo.close() + + +async def test_cli_add_private_generates_key_and_lists(ctx: AppContext) -> None: + """`add` with no secret creates a private channel with a random key; `list` sees it.""" + tool = ChannelsTool() + 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()) + 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 + + +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) + assert slot.name == "#general" and slot.is_public + + +async def test_cli_join_and_import_round_trip(ctx: AppContext) -> None: + """A channel joined by key can be shared and re-imported to the same secret.""" + tool = ChannelsTool() + secret = bytes(range(16)) + 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) + 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) + assert imported.name == "Squad" and imported.secret == secret + + +async def test_cli_import_rejects_bad_link(ctx: AppContext) -> None: + """Importing a non-channel link is a clean parameter error, not a crash.""" + import typer + + tool = ChannelsTool() + with pytest.raises(typer.BadParameter): + await tool.run(ctx, {"cli_action": "import", "index": 5, "url": "https://nope"})