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, 195 insertions, 0 deletions

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