feat(cli): meshbay-cli, a command-line client for members

Signs in to the hub, lists the account's groups and their folders, and
downloads and uploads files directly with each group's node, end-to-end
encrypted. With no argument it opens a two-pane interactive mode (Textual).
Session and device keys are kept encrypted under the system's key storage,
or a 0600 key file where there is none; node keys are pinned at first
contact; a new machine links as a device.

Packaged on its own on Linux (deb and rpm, on meshbay-common's venv, with a
man page generated from the parser) and frozen next to meshbay-node in the
full Windows installer, not the light one, with a self-test the build runs.
client/keys.py ports keyderive.js and is LGPL like it.

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

1 files changed, 354 insertions, 0 deletions

diff --git a/packages/meshbay-cli/src/meshbay_cli/tui/screens.py b/packages/meshbay-cli/src/meshbay_cli/tui/screens.py
new file mode 100644
index 0000000..9d19431
--- /dev/null
+++ b/packages/meshbay-cli/src/meshbay_cli/tui/screens.py
@@ -0,0 +1,354 @@
+"""
+The sign-in screen, and the main screen: two panes and the transfers.
+
+F5 copies the selection to the other pane's folder — a download or an upload,
+with the rules `get` and `put` follow: an upload is decided whole before it
+starts (names the node refuses stop it, names already there are asked about
+once for the batch), nothing on either side is replaced without being asked,
+and a name from a group is made writable here before it is written.
+"""
+
+import asyncio
+import contextlib
+import threading
+from pathlib import Path
+
+from rich.text import Text
+from textual.app import ComposeResult
+from textual.binding import Binding
+from textual.containers import Horizontal, Vertical
+from textual.screen import Screen
+from textual.widgets import Button, Footer, Input, Static
+
+from ..client.account import Account
+from ..client.names import UnsafeName, local_path
+from ..client.paths import handle
+from ..client.pool import Pool
+from ..client.store import FILE_KEY_NOTICE, KEY_FILE
+from ..client.transfers import PART_SUFFIX, TransferError, create_dir, download, upload
+from ..client.tree import Tree
+from ..client.upload_plan import plan_upload, refused_names, taken_names
+from ..errors import CliError
+from ..output import safe
+from .panes import LocalPane, RemotePane
+from .widgets import Choice, TransferStrip
+
+PER_GROUP = 2
+
+
+class SignInScreen(Screen):
+    DEFAULT_CSS = """
+    SignInScreen { align: center middle; }
+    SignInScreen > Vertical { width: 64; height: auto; border: thick $accent;
+                              padding: 1 2; background: $surface; }
+    SignInScreen Input { margin-bottom: 1; }
+    SignInScreen #error { color: $error; height: auto; }
+    """
+
+    def compose(self) -> ComposeResult:
+        ctx = self.app.ctx
+        with Vertical():
+            yield Static(Text("Sign in to MeshBay"))
+            yield Input(value=ctx.hub_url, placeholder="hub", id="hub")
+            yield Input(value=ctx.username or ctx.stored_username or "",
+                        placeholder="username", id="user")
+            yield Input(placeholder="passphrase", password=True, id="pass")
+            yield Button("Sign in", id="go", variant="primary")
+            yield Static("", id="error", markup=False)
+
+    def on_mount(self) -> None:
+        empty = next((i for i in self.query(Input) if not i.value), None)
+        (empty or self.query_one("#pass", Input)).focus()
+
+    def on_input_submitted(self, event: Input.Submitted) -> None:
+        self._submit()
+
+    def on_button_pressed(self, event: Button.Pressed) -> None:
+        self._submit()
+
+    def _submit(self) -> None:
+        hub = self.query_one("#hub", Input).value.strip().rstrip("/")
+        user = self.query_one("#user", Input).value.strip()
+        passphrase = self.query_one("#pass", Input).value
+        if not (hub and user and passphrase):
+            self.query_one("#error", Static).update("Hub, username and passphrase.")
+            return
+        self.query_one("#error", Static).update("Signing in…")
+        self.run_worker(self._sign_in(hub, user, passphrase), exclusive=True)
+
+    async def _sign_in(self, hub_url: str, user: str, passphrase: str) -> None:
+        app = self.app
+        try:
+            await app.use_hub(hub_url)
+            account = Account(app.hub, user, lambda: passphrase)
+            await account.hub.check_version()
+            await account.sign_in(remember=True)
+            kind = await account.remember(app.ctx.store)
+        except CliError as e:
+            self.query_one("#error", Static).update(e.message)
+            return
+        if kind == KEY_FILE:
+            app.notify(FILE_KEY_NOTICE, timeout=10, markup=False)
+        app.start(account)
+
+
+class MainScreen(Screen):
+    BINDINGS = [
+        Binding("f5", "copy", "Copy"),
+        Binding("q", "app.quit", "Quit"),
+        Binding("tab", "app.focus_next", "Switch", show=False),
+    ]
+
+    DEFAULT_CSS = """
+    MainScreen > Horizontal { height: 1fr; }
+    """
+
+    def __init__(self, account: Account):
+        super().__init__()
+        self.account = account
+        self.pool = Pool(account)
+        self._trees: dict[str, Tree] = {}
+        self._gates: dict[str, asyncio.Semaphore] = {}
+        self._watching: set[str] = set()
+
+    def compose(self) -> ComposeResult:
+        with Horizontal():
+            yield LocalPane(Path.cwd(), id="local")
+            yield RemotePane(self, id="remote")
+        yield TransferStrip(id="transfers")
+        yield Footer()
+
+    def on_mount(self) -> None:
+        self.query_one(LocalPane).table.focus()
+
+    @property
+    def local(self) -> LocalPane:
+        return self.query_one(LocalPane)
+
+    @property
+    def remote(self) -> RemotePane:
+        return self.query_one(RemotePane)
+
+    @property
+    def strip(self) -> TransferStrip:
+        return self.query_one(TransferStrip)
+
+    # ── Groups ───────────────────────────────────────────────────────────────
+
+    async def tree(self, group: dict) -> Tree:
+        """The group's index, asked once and then kept current by the node's pushes."""
+        if group["id"] not in self._trees:
+            session = await self.pool.session(group)
+            self._trees[group["id"]] = Tree.from_index(await session.index())
+            if group["id"] not in self._watching:
+                self._watching.add(group["id"])
+                session.ch.on_push("index_delta", lambda _msg, g=group: self._changed(g))
+        return self._trees[group["id"]]
+
+    def forget_tree(self, group: dict) -> None:
+        self._trees.pop(group["id"], None)
+
+    def _changed(self, group: dict) -> None:
+        """Something changed in the group: the index is asked for again, and the
+        pane redrawn if it is showing that group."""
+        self.forget_tree(group)
+        remote = self.remote
+        if remote.group is not None and remote.group["id"] == group["id"]:
+            remote.load()
+
+    # ── F5 ───────────────────────────────────────────────────────────────────
+
+    def action_copy(self) -> None:
+        focused = self.app.focused
+        if focused is not None and self.remote in focused.ancestors_with_self:
+            self.run_worker(self._download(), group="copy")
+        else:
+            self.run_worker(self._upload(), group="copy")
+
+    def _gate(self, group: dict) -> asyncio.Semaphore:
+        return self._gates.setdefault(group["id"], asyncio.Semaphore(PER_GROUP))
+
+    async def _ask(self, message: str, options: list[tuple[str, str]]) -> str:
+        return await self.app.push_screen(Choice(message, options), wait_for_dismiss=True)
+
+    async def _download(self) -> None:
+        remote, local = self.remote, self.local
+        rows = remote.selection()
+        if remote.group is None or not rows:
+            self.app.notify("Open a group and choose what to download.", timeout=4)
+            return
+        group, dest = remote.group, local.path
+        try:
+            tree = await self.tree(group)
+        except CliError as e:
+            self.app.notify(e.message, severity="error", markup=False)
+            return
+        plan: list[tuple[dict, Path]] = []
+        dirs: list[Path] = []
+        problems: list[str] = []
+        for row in rows:
+            path = row.data["path"]
+            top = path.rsplit("/", 1)[-1]
+            if not tree.root_available(path):
+                problems.append(f"{top}: its folder on the node is unavailable")
+                continue
+            if not row.is_dir:
+                try:
+                    plan.append((row.data["entry"], local_path(dest, [top])[0]))
+                except UnsafeName as e:
+                    problems.append(str(e))
+                continue
+            for folder, items in tree.walk(path):
+                rel = [top] + [p for p in folder[len(path):].strip("/").split("/") if p]
+                try:
+                    dirs.append(local_path(dest, rel)[0])
+                    plan += [(i.entry, local_path(dest, rel + [i.name])[0])
+                             for i in items if not i.is_dir]
+                except UnsafeName as e:
+                    problems.append(str(e))
+        for p in problems:
+            self.app.notify(p, severity="warning", markup=False)
+        existing = [t for _, t in plan if t.exists()
+                    and not t.with_name(t.name + PART_SUFFIX).exists()]
+        overwrite = False
+        if existing:
+            names = ", ".join(safe(t.name) for t in existing[:5]) + (
+                f" and {len(existing) - 5} more" if len(existing) > 5 else "")
+            answer = await self._ask(
+                f"{len(existing)} file(s) already in {dest}: {names}.",
+                [("skip", "Skip them"), ("overwrite", "Overwrite"), ("cancel", "Cancel")])
+            if answer == "cancel":
+                return
+            overwrite = answer == "overwrite"
+            if not overwrite:
+                plan = [(e, t) for e, t in plan if t not in existing]
+        for d in dirs:
+            await asyncio.to_thread(d.mkdir, parents=True, exist_ok=True)
+        remote.marks.clear()
+        remote.redraw()
+        session = await self.pool.session(group)
+        await asyncio.gather(*(self._one_download(group, session, e, t, overwrite)
+                               for e, t in plan))
+        local.action_refresh()
+
+    async def _one_download(self, group, session, entry: dict, target: Path,
+                            overwrite: bool) -> None:
+        job = self.strip.add("↓", entry["name"], int(entry["size"]))
+        async with self._gate(group):
+            self.strip.update_job(job, state="running")
+            try:
+                await download(
+                    session, entry, target, overwrite=overwrite,
+                    progress=lambda n: self.strip.update_job(job, done=job.done + n),
+                    on_queued=lambda n: self.strip.update_job(job, state="queued",
+                                                              note=f"({n} ahead)"))
+                self.strip.update_job(job, state="done",
+                                      note="" if target.name == entry["name"]
+                                      else f"saved as {target.name}")
+            except TransferError as e:
+                self.strip.update_job(job, state="failed", note=e.message)
+
+    async def _upload(self) -> None:
+        remote, local = self.remote, self.local
+        rows = local.selection()
+        if not rows:
+            return
+        if remote.group is None or not remote.path:
+            self.app.notify("Open a folder of a group on the right to upload into.",
+                            timeout=4)
+            return
+        group, folder = remote.group, remote.path
+        try:
+            tree = await self.tree(group)
+        except CliError as e:
+            self.app.notify(e.message, severity="error", markup=False)
+            return
+        root = tree.roots.get(folder.split("/", 1)[0], {})
+        if not root.get("writable") or not root.get("available", True):
+            self.app.notify(f"{safe(handle(group))}/{safe(folder)} does not accept "
+                            "uploads (read-only or unavailable).", severity="error",
+                            markup=False)
+            return
+        try:
+            dirs, files, problems = plan_upload([str(local.path / r.name) for r in rows],
+                                                recursive=True)
+        except CliError as e:
+            self.app.notify(e.message, severity="error", markup=False)
+            return
+        problems += refused_names([f.rel[-1] for f in files], dirs)
+        if problems:
+            await self._ask("Nothing was sent:\n" + "\n".join(safe(p) for p in problems),
+                            [("ok", "OK")])
+            return
+        tops = sorted({d[0] for d in dirs} | {f.rel[0] for f in files})
+        taken = taken_names(tree, folder, tops)
+        if taken:
+            answer = await self._ask(
+                f"Already in {safe(folder)}: {', '.join(safe(t) for t in sorted(taken))}. "
+                "The node never replaces a file.",
+                [("skip", "Skip them"), ("cancel", "Cancel")])
+            if answer != "skip":
+                return
+            dirs = [d for d in dirs if d[0] not in taken]
+            files = [f for f in files if f.rel[0] not in taken]
+        local.marks.clear()
+        local.redraw()
+        session = await self.pool.session(group)
+        remote_dir = {(): folder}
+        try:
+            for d in dirs:
+                remote_dir[tuple(d)] = await create_dir(session, remote_dir[tuple(d[:-1])],
+                                                        d[-1])
+        except TransferError as e:
+            self.app.notify(e.message, severity="error", markup=False)
+            return
+        await asyncio.gather(*(self._one_upload(group, session, f,
+                                                remote_dir[tuple(f.rel[:-1])])
+                               for f in files))
+        self._changed(group)
+
+    async def _one_upload(self, group, session, item, folder: str) -> None:
+        name = item.rel[-1]
+        job = self.strip.add("↑", name, item.size)
+        async with self._gate(group):
+            self.strip.update_job(job, state="running")
+            lock = threading.Lock()
+            try:
+                with open(item.local, "rb") as fh:
+                    def read(offset: int, length: int) -> bytes:
+                        with lock:
+                            fh.seek(offset)
+                            return fh.read(length)
+                    stored = await upload(
+                        session, read, item.size, name, folder,
+                        progress=lambda n: self.strip.update_job(job, done=job.done + n),
+                        on_queued=lambda n: self.strip.update_job(job, state="queued",
+                                                                  note=f"({n} ahead)"))
+                self.strip.update_job(job, state="done", note="" if stored == name
+                                      else f"stored as {stored}")
+            except OSError as e:
+                self.strip.update_job(job, state="failed", note=e.strerror or str(e))
+            except TransferError as e:
+                self.strip.update_job(job, state="failed", note=e.message)
+
+    async def on_unmount(self) -> None:
+        with contextlib.suppress(Exception):
+            await self.pool.close()
+
+
+class DeviceCode:
+    """`account.on_device_code` for the interactive mode: the code in a
+    notification that stays until the approval lands; the node is retried meanwhile."""
+
+    def __init__(self, app):
+        self.app = app
+
+    def __call__(self, group: dict, pending, *, first: bool) -> bool:
+        if first:
+            self.app.notify(
+                f"This machine is not yet a device of yours on the node hosting "
+                f"{safe(handle(group))}. Approve it from the MeshBay application or a "
+                f"browser: the group's Settings → \"Your devices on this node\" → "
+                f"enter {pending.link_code}. Waiting…",
+                title="Approve this machine", timeout=600, markup=False)
+        return True