From 422647e7898b8df67cd11f8fb060aec23dec84aa Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Wed, 14 Oct 2026 22:20:15 +0200 Subject: 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 --- .../src/meshbay_cli/commands/__init__.py | 4 + .../src/meshbay_cli/commands/context.py | 82 +++++++ .../meshbay-cli/src/meshbay_cli/commands/get.py | 235 +++++++++++++++++++++ .../meshbay-cli/src/meshbay_cli/commands/groups.py | 69 ++++++ .../meshbay-cli/src/meshbay_cli/commands/login.py | 71 +++++++ .../meshbay-cli/src/meshbay_cli/commands/logout.py | 43 ++++ .../meshbay-cli/src/meshbay_cli/commands/ls.py | 208 ++++++++++++++++++ .../meshbay-cli/src/meshbay_cli/commands/put.py | 195 +++++++++++++++++ .../meshbay-cli/src/meshbay_cli/commands/trust.py | 54 +++++ 9 files changed, 961 insertions(+) create mode 100644 packages/meshbay-cli/src/meshbay_cli/commands/__init__.py create mode 100644 packages/meshbay-cli/src/meshbay_cli/commands/context.py create mode 100644 packages/meshbay-cli/src/meshbay_cli/commands/get.py create mode 100644 packages/meshbay-cli/src/meshbay_cli/commands/groups.py create mode 100644 packages/meshbay-cli/src/meshbay_cli/commands/login.py create mode 100644 packages/meshbay-cli/src/meshbay_cli/commands/logout.py create mode 100644 packages/meshbay-cli/src/meshbay_cli/commands/ls.py create mode 100644 packages/meshbay-cli/src/meshbay_cli/commands/put.py create mode 100644 packages/meshbay-cli/src/meshbay_cli/commands/trust.py (limited to 'packages/meshbay-cli/src/meshbay_cli/commands') 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 -- cgit v1.2.3