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>

9 files changed, 961 insertions, 0 deletions

diff --git a/packages/meshbay-cli/src/meshbay_cli/commands/__init__.py b/packages/meshbay-cli/src/meshbay_cli/commands/__init__.py
new file mode 100644
index 0000000..313c146
--- /dev/null
+++ b/packages/meshbay-cli/src/meshbay_cli/commands/__init__.py
@@ -0,0 +1,4 @@
+"""
+One module per action. Each exposes `add_parser(subparsers, global_options)` and
+`async run(ctx)`.
+"""
diff --git a/packages/meshbay-cli/src/meshbay_cli/commands/context.py b/packages/meshbay-cli/src/meshbay_cli/commands/context.py
new file mode 100644
index 0000000..c11533b
--- /dev/null
+++ b/packages/meshbay-cli/src/meshbay_cli/commands/context.py
@@ -0,0 +1,82 @@
+"""What every action needs: the hub, the account name, the passphrase, the store."""
+
+import argparse
+import getpass
+import os
+import sys
+import time
+from pathlib import Path
+
+from ..client.account import Account, NoPassphrase
+from ..client.hub import DEFAULT_HUB, Hub
+from ..client.paths import handle
+from ..client.store import Store
+from ..errors import USAGE, CliError
+from ..output import safe
+
+
+class Context:
+    def __init__(self, args: argparse.Namespace):
+        self.args = args
+        self.store = Store()
+        stored = self.store.peek()
+        explicit_hub = args.hub or os.environ.get("MESHBAY_HUB")
+        self.hub_url = (explicit_hub or stored.get("hub") or DEFAULT_HUB).rstrip("/")
+        user = args.user or os.environ.get("MESHBAY_USER")
+        self.username = user.strip() if user else None
+        self.stored_username = stored.get("username") if stored.get("hub") == self.hub_url \
+            else None
+        self.hub = Hub(self.hub_url)
+        self._passphrase: str | None = None
+
+    def passphrase(self) -> str:
+        """--password, --password-file, MESHBAY_PASSWORD, then a prompt on a terminal."""
+        if self._passphrase is not None:
+            return self._passphrase
+        a = self.args
+        value = a.password
+        if value is None and a.password_file:
+            try:
+                text = sys.stdin.read() if a.password_file == "-" \
+                    else Path(a.password_file).read_text(encoding="utf-8")
+            except OSError as e:
+                raise CliError(USAGE, f"Cannot read {a.password_file}: {e.strerror}") from e
+            value = text.splitlines()[0] if text else ""
+        if value is None:
+            value = os.environ.get("MESHBAY_PASSWORD")
+        if value is None:
+            if not sys.stdin.isatty():
+                raise NoPassphrase("A passphrase is needed and there is no terminal to ask "
+                                   "on; pass --password-file or set MESHBAY_PASSWORD.")
+            value = getpass.getpass(f"Passphrase for {self.username or self.stored_username}"
+                                    f" on {self.hub_url}: ")
+        if not value:
+            raise NoPassphrase("No passphrase given.")
+        self._passphrase = value
+        return value
+
+    def account(self) -> Account:
+        account = Account.open(self.hub, self.username, self.passphrase, self.store)
+        account.on_device_code = self._device_code
+        return account
+
+    def _device_code(self, group: dict, pending, *, first: bool) -> bool:
+        """Show the code once. Wait only where somebody is there to approve it."""
+        waiting = sys.stderr.isatty() and not self.args.quiet
+        if first:
+            minutes = max(1, int((pending.expires_at - time.time()) // 60))
+            print(f"This machine is not yet a device of yours on the node hosting "
+                  f"{safe(handle(group))}.\n"
+                  "Approve it from the MeshBay application or a browser where you use "
+                  "this group:\n"
+                  "  the group's Settings → \"Your devices on this node\" → enter  "
+                  f"{pending.link_code}  → Approve\n"
+                  + (f"Waiting for the approval (the code expires in {minutes} min; "
+                     "Ctrl-C to stop)..." if waiting else
+                     f"Run the command again once it is approved (the code expires in "
+                     f"{minutes} min)."),
+                  file=sys.stderr, flush=True)
+        return waiting
+
+    async def aclose(self) -> None:
+        await self.hub.aclose()
diff --git a/packages/meshbay-cli/src/meshbay_cli/commands/get.py b/packages/meshbay-cli/src/meshbay_cli/commands/get.py
new file mode 100644
index 0000000..eec73c4
--- /dev/null
+++ b/packages/meshbay-cli/src/meshbay_cli/commands/get.py
@@ -0,0 +1,235 @@
+"""
+`get`: download files and folders from groups.
+
+The whole run is planned before the first byte: paths and patterns resolved
+against each group's index, local names made writable here, existing files
+found. A path that names nothing stops the run there (exit 3); a file that
+cannot be written is reported and the others go on (exit 4).
+"""
+
+import asyncio
+import os
+from dataclasses import dataclass
+from pathlib import Path
+
+from meshbay_common.paths import fold
+
+from ..client.groups import my_groups
+from ..client.names import UnsafeName, local_path
+from ..client.paths import handle, parse, resolve_group
+from ..client.pool import Pool
+from ..client.session import GroupSession
+from ..client.transfers import PART_SUFFIX, TransferError, download
+from ..client.tree import Tree
+from ..errors import NODE, NOT_FOUND, PARTIAL, USAGE, CliError
+from ..output import Progress, err, human_size, safe, sentence
+from .context import Context
+
+HELP = "download files, or folders with -R"
+PER_GROUP = 2      # the node's default lease cap per member and group (§5.5)
+
+
+def add_parser(sub, parent) -> None:
+    p = sub.add_parser("get", parents=[parent], help=HELP, description=sentence(HELP)
+                       + ". Wildcards (* ? [...]) are matched against the group, "
+                       "so quote them.")
+    p.add_argument("paths", nargs="+", metavar="PATH", help="group[@owner]/root/...")
+    p.add_argument("-R", dest="recursive", action="store_true", help="folders, recursively")
+    p.add_argument("-o", dest="dest", metavar="DEST",
+                   help="a folder (created when it ends with /), a file name for a "
+                        "single file, or - for stdout; default: the current folder")
+    exist = p.add_mutually_exclusive_group()
+    exist.add_argument("-f", dest="force", action="store_true",
+                       help="overwrite existing files")
+    exist.add_argument("-n", dest="skip", action="store_true",
+                       help="skip existing files silently")
+
+
+@dataclass
+class Job:
+    group: dict
+    session: GroupSession
+    entry: dict
+    remote: str            # as shown: handle/root/folders/name
+    target: Path | None    # None: stdout
+
+
+async def run(ctx: Context) -> int:
+    await ctx.hub.check_version()
+    account = ctx.account()
+    pool = Pool(account)
+    try:
+        return await _run(ctx, account, pool)
+    finally:
+        await pool.close()
+        await account.close()
+
+
+async def _run(ctx: Context, account, pool: Pool) -> int:
+    a = ctx.args
+    groups = await my_groups(account)
+    trees: dict[str, Tree] = {}
+
+    async def tree_of(group: dict) -> tuple[GroupSession, Tree]:
+        session = await pool.session(group)
+        if group["id"] not in trees:
+            trees[group["id"]] = Tree.from_index(await session.index())
+        return session, trees[group["id"]]
+
+    # ── What the paths name ──────────────────────────────────────────────────
+    named: list[tuple[dict, GroupSession, Tree, str]] = []   # group, session, tree, path
+    missing: list[str] = []
+    unreachable: list[str] = []
+    for text in a.paths:
+        remote = parse(text)
+        if remote is None:
+            raise CliError(USAGE, "Name a group, and what to fetch in it.")
+        group = resolve_group(groups, remote.group)
+        try:
+            session, tree = await tree_of(group)
+        except CliError as e:
+            # One group whose node is away does not cost the others their files.
+            if e.code != NODE:
+                raise
+            unreachable.append(f"{text}: {e.message}")
+            continue
+        found = tree.expand(remote.parts)
+        if not found and (remote.inner and (tree.is_dir(remote.inner)
+                                            or tree.file(remote.inner))):
+            found = [remote.inner]      # a literal name with brackets in it
+        if not found:
+            missing.append(text)
+        named += [(group, session, tree, p) for p in found]
+    if missing:
+        raise CliError(NOT_FOUND, "Nothing in your groups matches: "
+                                  + ", ".join(safe(m) for m in missing))
+
+    to_stdout = a.dest == "-"
+    files = [(g, s, t, p) for g, s, t, p in named if p and not t.is_dir(p)]
+    folders = [(g, s, t, p) for g, s, t, p in named if not p or t.is_dir(p)]
+    if unreachable and not named:
+        raise CliError(NODE, "\n".join(safe(u) for u in unreachable))
+    if folders and not a.recursive:
+        raise CliError(USAGE, "These are folders; use -R: "
+                       + ", ".join(safe(_shown(g, p)) for g, _, _, p in folders))
+    if to_stdout and (folders or len(files) != 1):
+        raise CliError(USAGE, "-o - writes exactly one file to stdout.")
+
+    # ── Where it goes ────────────────────────────────────────────────────────
+    single_file = not folders and len(files) == 1 and not to_stdout
+    dest, file_name = _destination(a.dest, single_file)
+
+    problems: list[str] = list(unreachable)
+    jobs: list[Job] = []
+    dirs: list[Path] = []
+    planned: set[str] = set()
+    renamed: list[tuple[str, Path]] = []
+
+    def plan_file(group, session, tree, path, parts):
+        entry = tree.file(path)
+        shown = _shown(group, path)
+        if not tree.root_available(path):
+            problems.append(f"{shown}: its folder on the node is unavailable")
+            return
+        if to_stdout:
+            jobs.append(Job(group, session, entry, shown, None))
+            return
+        try:
+            target, changed = local_path(dest, parts)
+        except UnsafeName as e:
+            problems.append(f"{shown}: {e}")
+            return
+        key = fold(os.path.normcase(str(target)))
+        if key in planned:
+            problems.append(f"{shown}: another file of this run has the same local name "
+                            f"{target}")
+            return
+        planned.add(key)
+        if changed:
+            renamed.append((shown, target))
+        if target.exists() and not target.with_name(target.name + PART_SUFFIX).exists():
+            if a.skip:
+                return
+            if not a.force:
+                problems.append(f"{target} already exists (-f to overwrite, -n to skip)")
+                return
+        jobs.append(Job(group, session, entry, shown, target))
+
+    for group, session, tree, path in files:
+        name = path.rsplit("/", 1)[-1]
+        plan_file(group, session, tree, path, [file_name or name])
+    for group, session, tree, path in folders:
+        top = path.rsplit("/", 1)[-1] if path else group["name"]
+        for folder, items in tree.walk(path):
+            rel = [top] + (folder[len(path):].strip("/").split("/")
+                           if folder != path else [])
+            rel = [r for r in rel if r]
+            try:
+                dirs.append(local_path(dest, rel)[0])
+            except UnsafeName as e:
+                problems.append(f"{_shown(group, folder)}: {e}")
+                continue
+            for item in items:
+                if not item.is_dir:
+                    plan_file(group, session, tree, item.path, rel + [item.name])
+
+    # ── Fetch ────────────────────────────────────────────────────────────────
+    progress = Progress("↓", quiet=a.quiet or to_stdout and not a.json, as_json=a.json)
+    for shown, target in renamed:
+        progress.note(f"meshbay-cli: {safe(shown)} is saved as {target.name} "
+                      "(its name cannot be written here as it is)")
+    for problem in problems:
+        progress.note(f"meshbay-cli: {safe(problem)}")
+    for d in dirs:
+        d.mkdir(parents=True, exist_ok=True)
+    progress.plan(len(jobs), sum(int(j.entry["size"]) for j in jobs))
+
+    failed = len(problems)
+    gates: dict[str, asyncio.Semaphore] = {}
+
+    async def one(job: Job) -> None:
+        nonlocal failed
+        gate = gates.setdefault(job.group["id"], asyncio.Semaphore(PER_GROUP))
+        async with gate:
+            progress.start(safe(job.entry["name"]))
+            try:
+                await download(job.session, job.entry, job.target, overwrite=a.force,
+                               progress=progress.advance,
+                               on_queued=lambda n: progress.queued(
+                                   safe(job.entry["name"]), n))
+                progress.finish(safe(job.entry["name"]), path=job.remote,
+                                to=str(job.target or "-"))
+            except TransferError as e:
+                failed += 1
+                progress.note(f"meshbay-cli: {safe(job.remote)}: {e.message}")
+
+    await asyncio.gather(*(one(j) for j in jobs))
+    progress.clear()
+
+    if not a.quiet and not to_stdout and not a.json:
+        done = len(jobs) - (failed - len(problems))
+        err(f"Downloaded {done} file{'s' if done != 1 else ''} "
+            f"({human_size(progress.bytes_done)})"
+            + (f"; {failed} failed." if failed else "."))
+    return PARTIAL if failed else 0
+
+
+def _shown(group: dict, path: str) -> str:
+    return f"{handle(group)}/{path}".rstrip("/")
+
+
+def _destination(dest: str | None, single_file: bool) -> tuple[Path, str | None]:
+    """The folder files land in, and a file name when -o names one."""
+    if dest is None or dest == "-":
+        return Path.cwd(), None
+    path = Path(dest).expanduser()
+    if dest.endswith(("/", os.sep)):
+        path.mkdir(parents=True, exist_ok=True)
+        return path, None
+    if path.is_dir():
+        return path, None
+    if single_file:
+        if not path.parent.is_dir():
+            raise CliError(USAGE, f"{path.parent} is not a folder.")
+        return path.parent, path.name
+    raise CliError(USAGE, f"{dest} is not a folder (end it with / to create it).")
diff --git a/packages/meshbay-cli/src/meshbay_cli/commands/groups.py b/packages/meshbay-cli/src/meshbay_cli/commands/groups.py
new file mode 100644
index 0000000..7595e2b
--- /dev/null
+++ b/packages/meshbay-cli/src/meshbay_cli/commands/groups.py
@@ -0,0 +1,69 @@
+"""
+`groups`: the groups this account belongs to, one line each.
+
+The hub answers alone (`/v1/groups/mine`); no node is dialled. Whether a node
+is up is what the hub sees of its socket, not a promise this machine can reach
+it. "owner" is the account that created the group on the hub — a node's
+operator is another notion.
+"""
+
+import json
+import sys
+from datetime import datetime
+
+from ..client.groups import my_groups
+from ..client.paths import handle
+from ..output import safe, sentence
+from .context import Context
+
+HELP = "list the groups you belong to"
+
+
+def add_parser(sub, parent) -> None:
+    sub.add_parser("groups", parents=[parent], help=HELP, description=sentence(HELP)
+                   + ": the node's state, your role, the visibility, the last activity. "
+                   "The GROUP column is the path `ls`, `get` and `put` take.")
+
+
+def node_state(g: dict) -> str:
+    if not g.get("hosted", True):
+        return "none"
+    return "up" if g.get("node_online") else "down"
+
+
+def _when(iso: str | None) -> str:
+    if not iso:
+        return ""
+    try:
+        return datetime.fromisoformat(iso).astimezone().strftime("%Y-%m-%d %H:%M")
+    except ValueError:
+        return ""
+
+
+async def run(ctx: Context) -> int:
+    await ctx.hub.check_version()
+    account = ctx.account()
+    try:
+        groups = await my_groups(account)
+    finally:
+        await account.close()
+    groups.sort(key=lambda g: (g["name"].lower(), g.get("owner_username") or ""))
+    if ctx.args.json:
+        json.dump([{"group": g["name"], "owner": g.get("owner_username"), "id": g["id"],
+                    "path": handle(g), "node": node_state(g),
+                    "role": "owner" if g.get("is_admin") else "member",
+                    "visibility": g.get("visibility"), "join_policy": g.get("join_policy"),
+                    "muted": bool(g.get("muted")), "description": g.get("description", ""),
+                    "last_activity_at": g.get("last_activity_at")}
+                   for g in groups], sys.stdout, indent=1)
+        print()
+        return 0
+    if not groups:
+        print("You belong to no group yet.", file=sys.stderr)
+        return 0
+    print(f"{'NODE':<6}{'ROLE':<8}{'VISIBILITY':<12}{'LAST ACTIVITY':<18}GROUP")
+    for g in groups:
+        role = "owner" if g.get("is_admin") else "member"
+        print(f"{node_state(g):<6}{role:<8}{safe(g.get('visibility') or ''):<12}"
+              f"{_when(g.get('last_activity_at')):<18}{safe(handle(g))}")
+    return 0
diff --git a/packages/meshbay-cli/src/meshbay_cli/commands/login.py b/packages/meshbay-cli/src/meshbay_cli/commands/login.py
new file mode 100644
index 0000000..cbd16ef
--- /dev/null
+++ b/packages/meshbay-cli/src/meshbay_cli/commands/login.py
@@ -0,0 +1,71 @@
+"""`login`: sign in with the passphrase, and remember the session."""
+
+import contextlib
+import json
+import sys
+
+from ..client.account import Account
+from ..client.groups import my_groups
+from ..client.store import FILE_KEY_NOTICE, KEY_FILE
+from ..errors import USAGE, CliError
+from ..output import err, safe, sentence
+from .context import Context
+from .logout import sign_out_stored
+
+HELP = "sign in, and remember the session so later runs need no passphrase"
+
+
+def add_parser(sub, parent) -> None:
+    p = sub.add_parser("login", parents=[parent], help=HELP, description=sentence(HELP) + ".")
+    p.add_argument("--no-remember", action="store_true",
+                   help="check the credentials and store nothing")
+
+
+async def run(ctx: Context) -> int:
+    username = ctx.username or ctx.stored_username
+    if not username:
+        if not sys.stdin.isatty():
+            raise CliError(USAGE, "Name the account with --user.")
+        username = input("Username: ").strip()
+        ctx.username = username
+    if not username:
+        raise CliError(USAGE, "No username given.")
+    ctx.username = username
+
+    notice = await ctx.hub.check_version()
+    remember = not ctx.args.no_remember
+
+    # Signing in again as the same account keeps its device key, its node
+    # identities and its pins; another account replaces the store.
+    previous = ctx.store.peek()
+    account = Account(ctx.hub, username, ctx.passphrase)
+    if previous.get("hub") == ctx.hub.url and previous.get("username") == username:
+        with contextlib.suppress(CliError):
+            ctx.store.load()
+            account.state, account.secrets = ctx.store.state, ctx.store.secrets
+    elif remember and previous:
+        await sign_out_stored(ctx)
+
+    await account.sign_in(remember=remember)
+    groups = await my_groups(account)
+
+    if remember:
+        kind = await account.remember(ctx.store)
+        if kind == KEY_FILE:
+            err(FILE_KEY_NOTICE)
+    else:
+        await account.close()
+
+    n = len(groups)
+    if ctx.args.json:
+        print(json.dumps({"username": username, "hub": ctx.hub.url, "groups": n,
+                          "remembered": remember,
+                          "key_store": ctx.store.peek().get("key_store") if remember
+                          else None}))
+    else:
+        verb = "Signed in" if remember else "Credentials accepted"
+        print(f"{verb}: {safe(username)} on {ctx.hub.url} — "
+              f"{n} group{'s' if n != 1 else ''}.")
+    if notice:
+        err(notice)
+    return 0
diff --git a/packages/meshbay-cli/src/meshbay_cli/commands/logout.py b/packages/meshbay-cli/src/meshbay_cli/commands/logout.py
new file mode 100644
index 0000000..7ee3aa2
--- /dev/null
+++ b/packages/meshbay-cli/src/meshbay_cli/commands/logout.py
@@ -0,0 +1,43 @@
+"""`logout`: end the session on the hub, retire this machine's key, erase the store."""
+
+import json
+
+from ..client.account import Account, no_passphrase
+from ..client.hub import Hub
+from ..errors import CliError
+from ..output import err, sentence
+from .context import Context
+
+HELP = "end the session, retire this machine's device key and erase what is stored"
+
+
+def add_parser(sub, parent) -> None:
+    sub.add_parser("logout", parents=[parent], help=HELP, description=sentence(HELP) + ".")
+
+
+async def sign_out_stored(ctx: Context) -> None:
+    """Whatever account the store holds, on whatever hub. Never asks anything."""
+    stored = ctx.store.peek()
+    hub = ctx.hub if stored.get("hub") == ctx.hub.url else Hub(stored.get("hub", ""))
+    try:
+        account = Account.open(hub, None, no_passphrase, ctx.store)
+        warnings = await account.sign_out()
+    except CliError as e:
+        warnings = [f"The saved session could not be opened ({e.message}); it is erased "
+                    "here, and ends on the hub when it expires."]
+        with ctx.store.locked():
+            ctx.store.erase()
+    finally:
+        if hub is not ctx.hub:
+            await hub.aclose()
+    for w in warnings:
+        err(f"meshbay-cli: warning: {w}")
+
+
+async def run(ctx: Context) -> int:
+    if not ctx.store.peek():
+        print(json.dumps({"signed_out": False}) if ctx.args.json else "Not signed in.")
+        return 0
+    await sign_out_stored(ctx)
+    print(json.dumps({"signed_out": True}) if ctx.args.json else "Signed out.")
+    return 0
diff --git a/packages/meshbay-cli/src/meshbay_cli/commands/ls.py b/packages/meshbay-cli/src/meshbay_cli/commands/ls.py
new file mode 100644
index 0000000..bdff3c0
--- /dev/null
+++ b/packages/meshbay-cli/src/meshbay_cli/commands/ls.py
@@ -0,0 +1,208 @@
+"""
+`ls`: your groups, or what a folder of a group holds — as the Files tab shows it.
+
+With no path the hub answers alone (`/v1/groups/mine`) and no node is dialled.
+Each group is its own top-level folder and nothing is merged across groups
+(docs/MESHBAY_DESIGN.md §9.11).
+"""
+
+import asyncio
+import json
+import sys
+
+from ..client.groups import my_groups
+from ..client.paths import encode_name, handle, parse, resolve_group
+from ..client.pool import Pool
+from ..client.tree import Item, Tree
+from ..errors import NODE, NOT_FOUND, PARTIAL, CliError
+from ..output import columns, date, err, human_size, is_tty, safe, sentence
+from .context import Context
+
+HELP = "list your groups, or a folder of a group"
+
+
+def add_parser(sub, parent) -> None:
+    p = sub.add_parser("ls", parents=[parent], help=HELP, description=sentence(HELP) + ".")
+    p.add_argument("path", nargs="?", default="",
+                   help="group[@owner]/root/folder; none lists your groups")
+    p.add_argument("-l", dest="long", action="store_true",
+                   help="long form: type, size, date added")
+    p.add_argument("-R", dest="recursive", action="store_true", help="recurse into folders")
+    p.add_argument("-a", dest="all", action="store_true",
+                   help="include the content of unavailable roots")
+    p.add_argument("--sort", choices=("name", "size", "date"), default="name",
+                   help="the order within folders and files (default: name)")
+    p.add_argument("-r", dest="reverse", action="store_true", help="reverse the order")
+
+
+def _sorted(items: list[Item], key: str, reverse: bool) -> list[Item]:
+    """Folders first, as the Files tab; the order applies within each."""
+    def k(i: Item):
+        if key == "size":
+            return (i.size, i.name.lower())
+        if key == "date":
+            return (i.added_at, i.name.lower())
+        return (i.name.lower(), i.name)
+    dirs = sorted((i for i in items if i.is_dir), key=k, reverse=reverse)
+    files = sorted((i for i in items if not i.is_dir), key=k, reverse=reverse)
+    return dirs + files
+
+
+def _flags(item: Item) -> str:
+    flags = []
+    if item.root is not None:
+        flags.append("writable" if item.root.get("writable") else "read-only")
+    if not item.available:
+        flags.append("unavailable")
+    return f"  ({', '.join(flags)})" if flags else ""
+
+
+def _line(item: Item, long: bool) -> str:
+    name = safe(item.name) + ("/" if item.is_dir else "")
+    if not long:
+        return name
+    kind = "root" if item.root is not None else ("dir" if item.is_dir else "file")
+    size = "" if item.is_dir else human_size(item.size)
+    return f"{kind:<5}{size:>10}  {date(item.added_at):<16}  {name}{_flags(item)}"
+
+
+def _print_items(items: list[Item], long: bool) -> None:
+    if long or not is_tty():
+        for i in items:
+            print(_line(i, long))
+    elif items:
+        print(columns([_line(i, False) for i in items]))
+
+
+def _json_item(group: dict, item: Item) -> dict:
+    base = handle(group)
+    out = {"group": group["name"], "owner": group.get("owner_username"),
+           "path": f"{base}/{'/'.join(encode_name(p) for p in item.path.split('/'))}",
+           "name": item.name, "available": item.available,
+           # `kind`, not `type`: an index entry's `type` is its media type.
+           "kind": "root" if item.root is not None else ("dir" if item.is_dir else "file")}
+    if item.root is not None:
+        out["writable"] = bool(item.root.get("writable"))
+    if item.entry is not None:
+        out.update({k: v for k, v in item.entry.items() if k not in ("name", "path")})
+    return out
+
+
+async def _list_groups(ctx: Context, account) -> int:
+    groups = sorted(await my_groups(account), key=lambda g: (g["name"].lower(),
+                                                             g.get("owner_username") or ""))
+    if ctx.args.json:
+        if ctx.args.recursive:
+            return await _every_group(ctx, account, groups)
+        json.dump([{"group": g["name"], "owner": g.get("owner_username"), "id": g["id"],
+                    "path": handle(g), "online": bool(g.get("node_online")),
+                    "visibility": g.get("visibility"), "admin": bool(g.get("is_admin"))}
+                   for g in groups], sys.stdout, indent=1)
+        print()
+        return 0
+    if ctx.args.recursive:
+        return await _every_group(ctx, account, groups)
+    if ctx.args.long or not is_tty():
+        for g in groups:
+            state = "up" if g.get("node_online") else "down"
+            print(f"{state:<6}{safe(handle(g))}" if ctx.args.long else safe(handle(g)))
+    else:
+        print(columns([safe(handle(g)) for g in groups]))
+    return 0
+
+
+async def _every_group(ctx: Context, account, groups: list[dict]) -> int:
+    """
+    Every group's tree. All are dialled at once through the pool, so a node that
+    does not answer costs its own deadline and nobody else's; a group whose node
+    the hub reports as down is not dialled at all. Printed in name order.
+    """
+    pool = Pool(account)
+
+    async def tree_of(group: dict):
+        if not group.get("node_online"):
+            raise CliError(NODE, "no node serving it is online")
+        session = await pool.session(group)
+        return Tree.from_index(await session.index())
+
+    try:
+        results = await asyncio.gather(*(tree_of(g) for g in groups), return_exceptions=True)
+    finally:
+        await pool.close()
+    failed = 0
+    listing_json = []
+    first = True
+    for group, result in zip(groups, results):
+        if isinstance(result, BaseException):
+            if not isinstance(result, CliError):
+                raise result
+            failed += 1
+            err(f"meshbay-cli: {safe(handle(group))}: {result.message}")
+            continue
+        if ctx.args.json:
+            listing_json += _json_listing(ctx, group, result, "")
+            continue
+        if not first:
+            print()
+        first = False
+        _list_in_group(ctx, group, result, "")
+    if ctx.args.json:
+        json.dump(listing_json, sys.stdout, indent=1)
+        print()
+    return PARTIAL if failed else 0
+
+
+async def run(ctx: Context) -> int:
+    await ctx.hub.check_version()
+    account = ctx.account()
+    try:
+        remote = parse(ctx.args.path)
+        if remote is None:
+            return await _list_groups(ctx, account)
+        group = resolve_group(await my_groups(account), remote.group)
+        pool = Pool(account)
+        try:
+            tree = Tree.from_index(await (await pool.session(group)).index())
+        finally:
+            await pool.close()
+        return _list_in_group(ctx, group, tree, remote.inner)
+    finally:
+        await account.close()
+
+
+def _list_in_group(ctx: Context, group: dict, tree: Tree, path: str) -> int:
+    a = ctx.args
+    if path and not tree.is_dir(path):
+        entry = tree.file(path)
+        if entry is None:
+            raise CliError(NOT_FOUND, f"No file or folder {path!r} in {handle(group)}.")
+        parent = path.rpartition("/")[0]
+        listing = [(None, [Item(entry["name"], path, False, entry=entry,
+                                available=tree.root_available(parent))])]
+    elif a.recursive:
+        listing = list(tree.walk(path, include_unavailable=a.all))
+    else:
+        listing = [(path, tree.children(path, include_unavailable=a.all))]
+
+    if a.json:
+        json.dump([_json_item(group, i) for _, items in listing for i in items],
+                  sys.stdout, indent=1)
+        print()
+        return 0
+    _print_listing(a, group, listing)
+    return 0
+
+
+def _json_listing(ctx: Context, group: dict, tree: Tree, path: str) -> list[dict]:
+    return [_json_item(group, i)
+            for _, items in tree.walk(path, include_unavailable=ctx.args.all) for i in items]
+
+
+def _print_listing(a, group: dict, listing) -> None:
+    for n, (folder, items) in enumerate(listing):
+        if a.recursive:
+            if n:
+                print()
+            shown = "/".join(encode_name(p) for p in folder.split("/")) if folder else ""
+            print(safe(f"{handle(group)}/{shown}".rstrip("/")) + ":")
+        _print_items(_sorted(items, a.sort, a.reverse), a.long)
diff --git a/packages/meshbay-cli/src/meshbay_cli/commands/put.py b/packages/meshbay-cli/src/meshbay_cli/commands/put.py
new file mode 100644
index 0000000..01a7233
--- /dev/null
+++ b/packages/meshbay-cli/src/meshbay_cli/commands/put.py
@@ -0,0 +1,195 @@
+"""
+`put`: upload files, or folders with -R, into a folder of a group.
+
+**Decided whole before anything is sent**, as a drop on the Files tab is
+(docs/MESHBAY_DESIGN.md §6.4): a name already in the destination — compared
+without case — or one the node does not accept refuses the run and lists every
+offender, so a folder never arrives with holes in it. `-n` skips the names
+already there instead. There is no overwrite: the node never replaces a file,
+and if one appears while an upload runs, the upload takes a free name and the
+CLI says which.
+"""
+
+import asyncio
+import contextlib
+import os
+import sys
+import tempfile
+import threading
+from pathlib import Path
+
+from ..client.groups import my_groups
+from ..client.paths import handle, parse, resolve_group
+from ..client.pool import Pool
+from ..client.transfers import TransferError, create_dir, upload
+from ..client.tree import Tree
+from ..client.upload_plan import UploadFile, plan_upload, refused_names, taken_names
+from ..errors import NOT_FOUND, PARTIAL, REFUSED, USAGE, CliError
+from ..output import Progress, err, human_size, safe, sentence
+from .context import Context
+
+HELP = "upload files, or folders with -R, into a folder of a group"
+PER_GROUP = 2
+
+
+def add_parser(sub, parent) -> None:
+    p = sub.add_parser("put", parents=[parent], help=HELP, description=sentence(HELP)
+                       + ". With - as the source, stdin is sent and DEST names the file.")
+    p.add_argument("sources", nargs="+", metavar="SOURCE",
+                   help="local files or folders, or - for stdin")
+    p.add_argument("dest", metavar="DEST",
+                   help="the destination folder: group[@owner]/root/folder; "
+                        "with - as the source, the file to write")
+    p.add_argument("-R", dest="recursive", action="store_true", help="folders, recursively")
+    p.add_argument("-n", dest="skip", action="store_true",
+                   help="skip names already in the destination instead of refusing")
+
+
+async def run(ctx: Context) -> int:
+    a = ctx.args
+    sources, dest_text = a.sources, a.dest
+    remote = parse(dest_text)
+    if remote is None or not remote.parts:
+        raise CliError(USAGE, "The destination is a folder of a group: "
+                              "group[@owner]/root/folder")
+    from_stdin = sources == ["-"]
+    if "-" in sources and not from_stdin:
+        raise CliError(USAGE, "- (stdin) is the only source when it is used.")
+
+    await ctx.hub.check_version()
+    account = ctx.account()
+    pool = Pool(account)
+    spool = None
+    try:
+        group = resolve_group(await my_groups(account), remote.group)
+        session = await pool.session(group)
+        tree = Tree.from_index(await session.index())
+
+        parts = remote.parts[:-1] if from_stdin else remote.parts
+        folder = "/".join(parts)
+        _check_folder(tree, group, folder)
+
+        if from_stdin:
+            spool = await asyncio.to_thread(_spool_stdin)
+            name = remote.parts[-1]
+            plan_dirs: list[list[str]] = []
+            files = [UploadFile(Path(spool.name), [name], os.path.getsize(spool.name))]
+            problems = refused_names([name], [])
+            problems += [f"{folder}/{t} is already there"
+                         for t in sorted(taken_names(tree, folder, [name]))]
+        else:
+            plan_dirs, files, problems = plan_upload(sources, a.recursive)
+            tops = sorted({tuple(d[:1]) for d in plan_dirs} | {tuple(f.rel[:1]) for f in files})
+            taken = taken_names(tree, folder, [t[0] for t in tops])
+            if taken and a.skip:
+                plan_dirs = [d for d in plan_dirs if d[0] not in taken]
+                files = [f for f in files if f.rel[0] not in taken]
+            elif taken:
+                problems += [f"{folder}/{t} is already there (-n to skip it)"
+                             for t in sorted(taken)]
+            problems += refused_names([f.rel[-1] for f in files], plan_dirs)
+        if problems:
+            raise CliError(REFUSED, "Nothing was sent:\n  "
+                           + "\n  ".join(safe(p) for p in problems))
+        return await _send(ctx, session, group, folder, plan_dirs, files,
+                           display_names=[remote.parts[-1]] if from_stdin else None)
+    finally:
+        if spool is not None:
+            with contextlib.suppress(OSError):
+                os.unlink(spool.name)
+        await pool.close()
+        await account.close()
+
+
+def _check_folder(tree: Tree, group: dict, folder: str) -> None:
+    shown = f"{handle(group)}/{folder}"
+    if not tree.is_dir(folder):
+        raise CliError(NOT_FOUND, f"No folder {safe(shown)}.")
+    root = tree.roots.get(folder.split("/", 1)[0], {})
+    if not root.get("available", True):
+        raise CliError(REFUSED, f"{safe(shown)} is unavailable on the node "
+                                "(its drive is disconnected).")
+    if not root.get("writable"):
+        raise CliError(REFUSED, f"{safe(shown)} is read-only: its operator does not "
+                                "accept uploads there.")
+
+
+def _spool_stdin():
+    """stdin to a file first: every chunk names the total, so the size comes first."""
+    f = tempfile.NamedTemporaryFile(prefix="meshbay-cli-", delete=False)
+    with f:
+        while block := sys.stdin.buffer.read(1024 * 1024):
+            f.write(block)
+    return f
+
+
+async def _send(ctx: Context, session, group: dict, folder: str, dirs: list[list[str]],
+                files: list[UploadFile], display_names: list[str] | None) -> int:
+    a = ctx.args
+    progress = Progress("↑", quiet=a.quiet, as_json=a.json)
+    progress.plan(len(files), sum(f.size for f in files))
+
+    # Folders first, parents before children, one at a time.
+    remote_dir: dict[tuple, str] = {(): 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:
+        progress.clear()
+        created = len(remote_dir) - 1
+        raise CliError(PARTIAL if created else e.code,
+                       f"Stopped: {e.message}"
+                       + (f" ({created} folder{'s' if created != 1 else ''} created)"
+                          if created else "")) from e
+
+    failed = 0
+    stop: TransferError | None = None
+    gate = asyncio.Semaphore(PER_GROUP)
+
+    async def one(job: UploadFile) -> None:
+        nonlocal failed, stop
+        async with gate:
+            if stop is not None:
+                return
+            name = job.rel[-1]
+            shown = f"{handle(group)}/{remote_dir[tuple(job.rel[:-1])]}/{name}"
+            progress.start(safe(name))
+            lock = threading.Lock()
+            try:
+                with open(job.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, job.size, name, remote_dir[tuple(job.rel[:-1])],
+                        progress=progress.advance,
+                        on_queued=lambda n: progress.queued(safe(name), n))
+            except OSError as e:
+                failed += 1
+                progress.note(f"meshbay-cli: {job.local}: {e.strerror}")
+                return
+            except TransferError as e:
+                failed += 1
+                progress.note(f"meshbay-cli: {safe(shown)}: {e.message}")
+                if e.fatal:
+                    stop = e
+                return
+            if stored != name:
+                progress.note(f"meshbay-cli: {safe(name)} was stored as {safe(stored)} "
+                              "(the name was taken meanwhile)")
+            progress.finish(safe(name), path=shown.rsplit("/", 1)[0] + "/" + stored,
+                            size=job.size)
+
+    await asyncio.gather(*(one(f) for f in files))
+    progress.clear()
+    if stop is not None:
+        skipped = len(files) - progress.files_done - failed
+        err(f"meshbay-cli: stopped: {stop.message}"
+            + (f" ({skipped} file{'s' if skipped != 1 else ''} not sent)" if skipped else ""))
+        failed += skipped
+    if not a.quiet and not a.json:
+        done = progress.files_done
+        err(f"Uploaded {done} file{'s' if done != 1 else ''} "
+            f"({human_size(progress.bytes_done)})" + (f"; {failed} failed." if failed else "."))
+    return PARTIAL if failed else 0
diff --git a/packages/meshbay-cli/src/meshbay_cli/commands/trust.py b/packages/meshbay-cli/src/meshbay_cli/commands/trust.py
new file mode 100644
index 0000000..f48a955
--- /dev/null
+++ b/packages/meshbay-cli/src/meshbay_cli/commands/trust.py
@@ -0,0 +1,54 @@
+"""
+`trust`: the node keys this machine pinned at first contact.
+
+A node presenting another key than the one pinned is refused outright (exit 6,
+docs/MESHBAY_DESIGN.md §5.2). `trust reset NODE` is the deliberate way out, for a
+rotation the node's operator has announced — the counterpart of the reset in
+the application's Settings. Nothing resets a pin by itself.
+"""
+
+import json
+import sys
+
+from ..client.account import Account, no_passphrase
+from ..errors import NOT_FOUND, USAGE, CliError
+from ..output import safe, sentence
+from .context import Context
+
+HELP = "the node keys pinned at first contact: list them, or forget one"
+
+
+def add_parser(sub, parent) -> None:
+    p = sub.add_parser("trust", parents=[parent], help=HELP, description=sentence(HELP)
+                       + ". A node whose key changed is refused until its pin is reset.")
+    p.add_argument("verb", choices=("list", "reset"))
+    p.add_argument("node", nargs="?", help="the node's id, as the refusal names it")
+
+
+async def run(ctx: Context) -> int:
+    a = ctx.args
+    if not ctx.store.peek():
+        pins = {}
+        account = None
+    else:
+        account = Account.open(ctx.hub, None, no_passphrase, ctx.store)
+        pins = account.state.get("pins", {}) if account.remembered else {}
+    if a.verb == "list":
+        if a.json:
+            json.dump(pins, sys.stdout, indent=1)
+            print()
+        else:
+            for node_id, pk in sorted(pins.items()):
+                print(f"{safe(node_id)}  {safe(pk)}")
+        return 0
+    if not a.node:
+        raise CliError(USAGE, "Name the node: meshbay-cli trust reset NODE")
+    if a.node not in pins:
+        raise CliError(NOT_FOUND, f"No key is pinned for the node {safe(a.node)}.")
+
+    def drop(state: dict, secrets: dict) -> None:
+        state.get("pins", {}).pop(a.node, None)
+    await account.update(drop)
+    print(f"Forgot the key pinned for {safe(a.node)}; the next connection pins the key "
+          "it presents.")
+    return 0