diff options
Diffstat (limited to 'packages/meshbay-cli/src/meshbay_cli')
32 files changed, 4787 insertions, 0 deletions
diff --git a/packages/meshbay-cli/src/meshbay_cli/__init__.py b/packages/meshbay-cli/src/meshbay_cli/__init__.py new file mode 100644 index 0000000..911e0e3 --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/__init__.py @@ -0,0 +1,3 @@ +"""MeshBay command-line client.""" + +__version__ = "0.19.0" diff --git a/packages/meshbay-cli/src/meshbay_cli/__main__.py b/packages/meshbay-cli/src/meshbay_cli/__main__.py new file mode 100644 index 0000000..2b0b349 --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/__main__.py @@ -0,0 +1,134 @@ +""" +meshbay-cli [global options] <action> [action options] [arguments] + +With no argument at all, the interactive mode opens. +""" + +import argparse +import asyncio +import logging +import os +import sys + +from . import __version__ +from .commands import get, groups, login, logout, ls, put, trust +from .commands.context import Context +from .errors import INTERRUPTED, USAGE, CliError + +ACTIONS = {"login": login, "logout": logout, "groups": groups, "ls": ls, "get": get, + "put": put, "trust": trust} + + +class _Parser(argparse.ArgumentParser): + """argparse exits 2 on a usage error, which here means a refused sign-in.""" + + def error(self, message: str): + self.print_usage(sys.stderr) + self.exit(USAGE, f"{self.prog}: error: {message}\n") + + +def _global_options(p: argparse.ArgumentParser, default) -> None: + p.add_argument("--hub", default=default, help="the hub (default: the one signed in to, " + "else https://meshbay.org); also MESHBAY_HUB") + p.add_argument("-u", "--user", default=default, help="account name; also MESHBAY_USER") + p.add_argument("-p", "--password", default=default, + help="the passphrase; visible to other users in the process list and " + "kept in shell history — --password-file - or MESHBAY_PASSWORD " + "are not") + p.add_argument("--password-file", metavar="FILE", default=default, + help="read the passphrase from FILE's first line (- for stdin)") + p.add_argument("--json", action="store_true", default=default or False, + help="machine-readable output") + p.add_argument("-q", "--quiet", action="store_true", default=default or False, + help="errors only") + p.add_argument("-v", "--verbose", action="count", default=default or 0, + help="-v: connection steps, -vv: protocol messages") + + +def build_parser() -> argparse.ArgumentParser: + """Global options are accepted before the action and after it.""" + p = _Parser( + prog="meshbay-cli", + description="A member's MeshBay files, from a terminal. Run with no argument " + "for the interactive mode.") + _global_options(p, None) + p.add_argument("--version", action="version", version=f"meshbay-cli {__version__}") + # SUPPRESS: an option the action's own parser did not see must not + # overwrite the value given before the action. + late = _Parser(add_help=False) + _global_options(late, argparse.SUPPRESS) + sub = p.add_subparsers(dest="action", metavar="<action>") + for module in ACTIONS.values(): + module.add_parser(sub, late) + return p + + +def _logging(verbose: int, quiet: bool) -> None: + handler = logging.StreamHandler(sys.stderr) + handler.setFormatter(logging.Formatter("meshbay-cli: %(levelname)s: %(message)s")) + root = logging.getLogger() + root.handlers[:] = [handler] + root.setLevel(logging.ERROR if quiet else logging.WARNING) + level = {0: logging.WARNING, 1: logging.INFO}.get(verbose, logging.DEBUG) + logging.getLogger("meshbay_cli").setLevel(logging.ERROR if quiet else level) + # aiortc and aioice are chatty at INFO; their story is told from -vv on. + for name in ("aiortc", "aioice", "httpx", "httpcore"): + logging.getLogger(name).setLevel(logging.DEBUG if verbose >= 3 else logging.WARNING) + + +async def _run(args: argparse.Namespace) -> int: + ctx = Context(args) + try: + if args.action is None: + from .tui.app import run_interactive + return await run_interactive(ctx) + return await ACTIONS[args.action].run(ctx) + finally: + await ctx.aclose() + + +def loop_factory(): + """aiortc's ICE needs a selector loop: Windows' default proactor never + completes a DataChannel handshake.""" + return asyncio.SelectorEventLoop if sys.platform == "win32" else None + + +def main(argv: list[str] | None = None) -> int: + argv = sys.argv[1:] if argv is None else argv + if not argv and os.environ.get("MESHBAY_CLI_SELFTEST") == "1": + from .selftest import run + return run() + parser = build_parser() + interactive = not argv + try: + args = parser.parse_args(argv) + except SystemExit as e: # --help, --version, or a usage error + return int(e.code or 0) + if args.action is None and not interactive: + parser.print_usage(sys.stderr) + return USAGE + _logging(args.verbose, args.quiet) + _utf8_output() + try: + return asyncio.run(_run(args), loop_factory=loop_factory()) + except CliError as e: + print(f"meshbay-cli: {e.message}", file=sys.stderr) + return e.code + except KeyboardInterrupt: + print("meshbay-cli: interrupted", file=sys.stderr) + return INTERRUPTED + + +def _utf8_output() -> None: + """Names are printed as they are; a Windows console's code page cannot hold most.""" + for stream in (sys.stdout, sys.stderr): + reconfigure = getattr(stream, "reconfigure", None) + if reconfigure is not None: + try: + reconfigure(encoding="utf-8", errors="replace") + except (ValueError, OSError): + pass + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/packages/meshbay-cli/src/meshbay_cli/client/account.py b/packages/meshbay-cli/src/meshbay_cli/client/account.py new file mode 100644 index 0000000..d620f6d --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/client/account.py @@ -0,0 +1,344 @@ +""" +A signed-in account: its hub session, its bundle key `M`, its node identities. + +Two lives. **Remembered** (after `login`): everything is in the store, and a +run needs no passphrase — an expired access token is renewed with the refresh +token, a refused one falls back to the device's hub key. **Ephemeral** (no +store, or another account named on the command line): signed in with the +passphrase for this run only, nothing written, the session ended on the hub +when the run ends. + +The refresh token rotates and a spent one presented again revokes the whole +family, so every renewal happens under the store's lock, after re-reading what +another run may have just written. +""" + +import asyncio +import contextlib +import logging +import socket +import time +from collections.abc import AsyncIterator, Callable +from dataclasses import dataclass + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from ..errors import AUTH, CliError +from .hub import Hub, HubError +from .keys import ( + Identity, + auth_key, + b64, + bundle_master_key, + passphrase_half, + unb64, +) +from .store import KEY_FILE, Store + +log = logging.getLogger(__name__) + +# Renew this long before the access token runs out, so a request never leaves +# with a token that expires on the way. +RENEW_MARGIN = 300 + +PassphraseSource = Callable[[], str] + + +class NoPassphrase(CliError): + """Raised by a passphrase source that has none to give.""" + + def __init__(self, message: str): + super().__init__(AUTH, message) + + +def no_passphrase() -> str: + """The source for an account that must never ask: one being signed out.""" + raise NoPassphrase("No passphrase is asked for to sign out.") + + +@dataclass +class Pending: + identity: Identity + code: str + expires_at: float + + +class Account: + def __init__(self, hub: Hub, username: str, passphrase: PassphraseSource, + store: Store | None = None): + self.hub = hub + self.username = username + self._passphrase = passphrase + self.store = store # None: ephemeral + # Called with (group, DeviceLinkPending, first=) when this machine waits + # for a device approval; returns whether to keep waiting. + self.on_device_code = None + self.state: dict = {} + self.secrets: dict = {} + + @property + def remembered(self) -> bool: + return self.store is not None + + @property + def user_id(self) -> str: + return self.state["user_id"] + + @property + def master(self) -> bytes | None: + m = self.secrets.get("master") + return unb64(m) if m else None + + @property + def legacy(self) -> bytes | None: + a = self.secrets.get("legacy") + return unb64(a) if a else None + + # ── Opening ────────────────────────────────────────────────────────────── + + @classmethod + def open(cls, hub: Hub, username: str | None, passphrase: PassphraseSource, + store: Store) -> "Account": + """The remembered account when it is the one asked for, else an ephemeral one.""" + stored = store.peek() + same_hub = stored.get("hub") == hub.url + same_user = username is None or username == stored.get("username") + if stored and same_hub and same_user and store.load(): + acc = cls(hub, store.state["username"], passphrase, store) + acc.state, acc.secrets = store.state, store.secrets + return acc + if not username: + raise CliError(AUTH, "Not signed in. Run `meshbay-cli login`, or pass --user.") + return cls(hub, username, passphrase) + + @contextlib.asynccontextmanager + async def _locked(self) -> AsyncIterator[None]: + """The store's lock, taken off the loop; re-reads what is stored.""" + if self.store is None: + yield + return + cm = self.store.locked() + await asyncio.to_thread(cm.__enter__) + try: + if not self.store.load(): + raise CliError(AUTH, "Signed out by another run; run `meshbay-cli login`.") + self.state, self.secrets = self.store.state, self.store.secrets + yield + finally: + cm.__exit__(None, None, None) + + def _save(self) -> None: + """Under the lock. `sign_in` replaces both dicts, so hand over the current ones.""" + self.store.state, self.store.secrets = self.state, self.secrets + self.store.save() + + async def update(self, mutate: Callable[[dict, dict], None]) -> None: + """Change state and secrets, and write them when remembered.""" + async with self._locked(): + mutate(self.state, self.secrets) + if self.store is not None: + self._save() + + # ── Signing in ─────────────────────────────────────────────────────────── + + async def sign_in(self, *, remember: bool) -> None: + """With the passphrase: tokens, `M`, and — remembered — the device's hub key.""" + passphrase = self._passphrase() + ak = await asyncio.to_thread(auth_key, self.username, passphrase) + previous = dict(self.state) + r = await self.hub.login(self.username, ak, previous.get("known_browser")) + token = r["access_token"] + me = await self.hub.me(token) + if not r.get("bundle_pepper"): + raise CliError(AUTH, "The hub did not provide the bundle pepper.") + a = await asyncio.to_thread(passphrase_half, self.username, passphrase) + master = bundle_master_key(a, unb64(r["bundle_pepper"]), me["user_id"]) + + same_account = previous.get("user_id") == me["user_id"] + state = { + "hub": self.hub.url, + "username": self.username, + "user_id": me["user_id"], + "known_browser": r.get("known_browser") or previous.get("known_browser"), + # A node's key does not depend on who signed in. + "pins": previous.get("pins", {}) if previous.get("hub") == self.hub.url else {}, + } + secrets = { + "access_token": token, + "access_exp": time.time() + int(r.get("expires_in", 0)), + "refresh_token": r["refresh_token"], + "master": b64(master), + "legacy": b64(a), + "identities": self.secrets.get("identities", {}) if same_account else {}, + } + if remember: + device = (self.secrets.get("device_sk") if same_account else None) + sk = (Ed25519PrivateKey.from_private_bytes(unb64(device)) if device + else Ed25519PrivateKey.generate()) + reg = await self.hub.register_device( + token, b64(sk.public_key().public_bytes_raw()), + f"meshbay-cli on {socket.gethostname()}", ak) + secrets["device_sk"] = b64(sk.private_bytes_raw()) + state["device_id"] = reg["id"] + self.state, self.secrets = state, secrets + + async def remember(self, store: Store) -> str | None: + """Write this account to `store`, replacing what it held. A notice, or None.""" + self.store = store + async with self._locked_fresh(): + store.state, store.secrets = self.state, self.secrets + store.save(new=True) + return KEY_FILE if store.state.get("key_store") == KEY_FILE else None + + @contextlib.asynccontextmanager + async def _locked_fresh(self) -> AsyncIterator[None]: + cm = self.store.locked() + await asyncio.to_thread(cm.__enter__) + try: + yield + finally: + cm.__exit__(None, None, None) + + # ── The access token ───────────────────────────────────────────────────── + + def _fresh(self) -> bool: + return bool(self.secrets.get("access_token")) and \ + self.secrets.get("access_exp", 0) - RENEW_MARGIN > time.time() + + async def token(self, *, rejected: str | None = None) -> str: + """A valid access token. `rejected`: the one the hub just refused.""" + if rejected is None and self._fresh(): + return self.secrets["access_token"] + if not self.remembered: + if not self.secrets: + await self.sign_in(remember=False) + else: + await self._renew() + return self.secrets["access_token"] + async with self._locked(): + # Another run may have renewed while this one waited for the lock; + # renewing again would present a refresh token it has just spent. + current = self.secrets.get("access_token") + if not self._fresh() or current == rejected: + await self._renew() + self._save() + return self.secrets["access_token"] + + async def _renew(self) -> None: + rt = self.secrets.get("refresh_token") + if rt: + try: + r = await self.hub.refresh(rt) + self._take(r) + return + except HubError as e: + log.debug("refresh refused: %s", e) + sk = self.secrets.get("device_sk") + if sk: + try: + r = await self.hub.device_auth( + self.username, Ed25519PrivateKey.from_private_bytes(unb64(sk))) + self._take(r) + return + except CliError as e: + log.debug("device sign-in refused: %s", e.message) + try: + await self.sign_in(remember=self.remembered) + except NoPassphrase as e: + if self.remembered: + raise CliError(AUTH, "The saved session has ended; run " + "`meshbay-cli login`, or pass --password.") from e + raise + + def _take(self, r: dict) -> None: + self.secrets["access_token"] = r["access_token"] + self.secrets["access_exp"] = time.time() + int(r.get("expires_in", 0)) + self.secrets["refresh_token"] = r["refresh_token"] + + async def call(self, fn, *args): + """A hub call with the access token, renewed once on a 401.""" + token = await self.token() + try: + return await fn(token, *args) + except HubError as e: + if e.status != 401: + raise + return await fn(await self.token(rejected=token), *args) + + # ── Identities and pins ────────────────────────────────────────────────── + + def identity(self, node_pk: str) -> Identity | None: + keys = self.secrets.get("identities", {}).get(node_pk) + return Identity.from_pkcs8(keys) if keys else None + + async def keep_identity(self, node_pk: str, identity: Identity) -> None: + def put(state: dict, secrets: dict) -> None: + secrets.setdefault("identities", {})[node_pk] = identity.to_pkcs8() + await self.update(put) + + async def forget_identity(self, node_pk: str) -> None: + def drop(state: dict, secrets: dict) -> None: + secrets.get("identities", {}).pop(node_pk, None) + await self.update(drop) + + # A device request this machine filed and that waits for an approval: the + # keys it named, and the code shown for them. Kept so a later run shows the + # same code and joins with the same keys once approved. + + def pending_identity(self, node_pk: str) -> "Pending | None": + p = self.secrets.get("pending", {}).get(node_pk) + if not p or p["expires_at"] < time.time(): + return None + return Pending(Identity.from_pkcs8(p["keys"]), p["code"], p["expires_at"]) + + async def keep_pending(self, node_pk: str, identity: Identity, code: str, + expires_at: float) -> None: + def put(state: dict, secrets: dict) -> None: + secrets.setdefault("pending", {})[node_pk] = { + "keys": identity.to_pkcs8(), "code": code, "expires_at": expires_at} + await self.update(put) + + async def adopt_pending(self, node_pk: str) -> None: + """The node admitted the pending keys: they are this machine's identity there.""" + def move(state: dict, secrets: dict) -> None: + p = secrets.get("pending", {}).pop(node_pk, None) + if p: + secrets.setdefault("identities", {})[node_pk] = p["keys"] + await self.update(move) + + def pin(self, node_id: str) -> str | None: + return self.state.get("pins", {}).get(node_id) + + async def set_pin(self, node_id: str, node_pk: str) -> None: + def put(state: dict, secrets: dict) -> None: + state.setdefault("pins", {})[node_id] = node_pk + await self.update(put) + + # ── Ending ─────────────────────────────────────────────────────────────── + + async def close(self) -> None: + """An ephemeral session ends on the hub with the run.""" + if not self.remembered and self.secrets.get("refresh_token"): + with contextlib.suppress(Exception): + await self.hub.logout(self.secrets["refresh_token"]) + + async def sign_out(self) -> list[str]: + """End the session, retire the device's hub key, erase the store. Warnings.""" + warnings = [] + device_id = self.state.get("device_id") + if device_id: + try: + await self.call(self.hub.delete_device, device_id) + except (HubError, CliError) as e: + warnings.append( + "The hub did not retire this machine's device key " + f"({getattr(e, 'detail', None) or getattr(e, 'message', e)}); remove " + f"\"meshbay-cli on {socket.gethostname()}\" from your account's devices.") + rt = self.secrets.get("refresh_token") + if rt: + with contextlib.suppress(Exception): + await self.hub.logout(rt) + if self.store is not None: + async with self._locked_fresh(): + self.store.erase() + return warnings diff --git a/packages/meshbay-cli/src/meshbay_cli/client/groups.py b/packages/meshbay-cli/src/meshbay_cli/client/groups.py new file mode 100644 index 0000000..ac738f3 --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/client/groups.py @@ -0,0 +1,68 @@ +"""Opening a group: which of the nodes serving it to dial, and in which order.""" + +import asyncio +import logging +import time + +from ..errors import NODE, CliError +from .account import Account +from .paths import handle +from .session import DeviceLinkPending, GroupSession, NodeKeyChanged, connect + +# How often a machine waiting for its approval tries the node again. Each try is +# one offer: well inside what the hub admits per account and per node. +LINK_POLL = 10 + +log = logging.getLogger(__name__) + + +async def my_groups(account: Account) -> list[dict]: + return await account.call(account.hub.my_groups) + + +async def open_group(account: Account, group: dict) -> GroupSession: + """ + A session with a node serving `group`: the pinned one first, then the others + in the hub's order. A node whose key changed stops everything — that is an + alarm, not a reason to quietly try its neighbour. + """ + nodes = await account.call(account.hub.group_nodes, group["id"]) + if not nodes: + raise CliError(NODE, f"No node serving {handle(group)} is online.") + nodes = sorted(nodes, key=lambda n: account.pin(n["node_id"]) is None) + failures = [] + for node in nodes: + try: + return await _connect_linking(account, group, node) + except (NodeKeyChanged, DeviceLinkPending): + raise + except CliError as e: + if e.code != NODE or len(nodes) == 1: + raise + log.info("node %s: %s", node["node_id"], e.message) + failures.append(f"{node['node_id']}: {e.message}") + raise CliError(NODE, f"No node serving {handle(group)} could be reached:\n " + + "\n ".join(failures)) + + +async def _connect_linking(account: Account, group: dict, node: dict) -> GroupSession: + """ + `connect`, and — when this machine is waiting to be approved as a device on + that node — the wait: the code is shown once (`account.on_device_code`), and + the node is tried again until the approval lands or the request expires. The + code names this node, so no other node is tried meanwhile. + """ + shown = False + while True: + try: + return await connect(account, group["id"], node) + except DeviceLinkPending as pending: + hook = account.on_device_code + wait = hook(group, pending, first=not shown) if hook else False + shown = True + if not wait: + raise + if pending.expires_at <= time.time(): + raise CliError(NODE, "The device request expired before it was approved; " + "run the command again for a new code.") from None + await asyncio.sleep(LINK_POLL) diff --git a/packages/meshbay-cli/src/meshbay_cli/client/hub.py b/packages/meshbay-cli/src/meshbay_cli/client/hub.py new file mode 100644 index 0000000..c8ce0c1 --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/client/hub.py @@ -0,0 +1,160 @@ +""" +The hub's API, as a member's client uses it. + +The hub signs in, lists groups and the nodes serving them, mints the token a +node is shown, and relays the WebRTC offer. Nothing else: no listing, no file +and no byte of content passes through it (docs/MESHBAY_DESIGN.md §1.3, §8.4). +""" + +import logging +import time + +import httpx +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from .. import __version__ +from ..errors import AUTH, NODE, NOT_FOUND, VERSION, CliError +from .keys import b64 + +log = logging.getLogger(__name__) + +DEFAULT_HUB = "https://meshbay.org" + + +class HubError(Exception): + def __init__(self, status: int, detail: str): + super().__init__(f"{status}: {detail}") + self.status = status + self.detail = detail + + +def _version_tuple(v: str) -> tuple[int, ...]: + out = [] + for part in v.split("."): + digits = "".join(c for c in part if c.isdigit()) + out.append(int(digits or 0)) + return tuple(out) + + +class Hub: + def __init__(self, url: str): + self.url = url.rstrip("/") + self._http = httpx.AsyncClient( + base_url=self.url, timeout=30, + headers={"User-Agent": f"meshbay-cli/{__version__}"}) + + async def aclose(self) -> None: + await self._http.aclose() + + async def _call(self, method: str, path: str, *, token: str | None = None, + json: dict | None = None) -> dict: + headers = {"Authorization": f"Bearer {token}"} if token else {} + try: + r = await self._http.request(method, path, json=json, headers=headers) + except httpx.HTTPError as e: + raise CliError(NODE, f"Cannot reach the hub at {self.url}: {e}") from e + if r.status_code >= 400: + try: + detail = r.json().get("detail", r.text) + except Exception: + detail = r.text + raise HubError(r.status_code, str(detail)) + return r.json() if r.content else {} + + # ── Version ────────────────────────────────────────────────────────────── + + async def check_version(self) -> str | None: + """Refuse a client below the hub's minimum. Returns a notice, or None.""" + try: + info = await self._call("GET", "/v1/hub/version") + except HubError as e: + raise CliError(NODE, f"{self.url} does not answer as a MeshBay hub " + f"({e.status})") from e + client = info.get("client") or {} + mine = _version_tuple(__version__) + minimum = client.get("minimum") + if minimum and mine < _version_tuple(minimum): + raise CliError(VERSION, f"This hub needs meshbay-cli {minimum} or later; " + f"this is {__version__}.") + recommended = client.get("recommended") + if recommended and mine < _version_tuple(recommended): + return f"meshbay-cli {recommended} is available; this is {__version__}." + return None + + # ── Signing in ─────────────────────────────────────────────────────────── + + async def login(self, username: str, auth_key: str, + known_browser: str | None) -> dict: + body = {"username": username, "auth_key": auth_key} + if known_browser: + body["known_browser"] = known_browser + try: + return await self._call("POST", "/v1/users/login", json=body) + except HubError as e: + raise _sign_in_refusal(e) from e + + async def register_device(self, token: str, pk: str, label: str, auth_key: str) -> dict: + return await self._call("POST", "/v1/users/devices", token=token, json={ + "pk_auth_ed25519": pk, "label": label[:64], "auth_key": auth_key}) + + async def delete_device(self, token: str, device_id: str) -> None: + await self._call("DELETE", f"/v1/users/devices/{device_id}", token=token) + + async def device_auth(self, username: str, sk: Ed25519PrivateKey) -> dict: + ts = int(time.time()) + sig = sk.sign(f"meshbay:user_auth:{username}:{ts}".encode()) + try: + return await self._call("POST", "/v1/users/auth", json={ + "username": username, "timestamp": ts, "signature": b64(sig)}) + except HubError as e: + raise _sign_in_refusal(e) from e + + async def refresh(self, refresh_token: str) -> dict: + return await self._call("POST", "/v1/users/token/refresh", + json={"refresh_token": refresh_token}) + + async def logout(self, refresh_token: str) -> None: + await self._call("POST", "/v1/users/logout", json={"refresh_token": refresh_token}) + + async def me(self, token: str) -> dict: + return await self._call("GET", "/v1/users/me", token=token) + + # ── Groups and nodes ───────────────────────────────────────────────────── + + async def my_groups(self, token: str) -> list[dict]: + return (await self._call("GET", "/v1/groups/mine", token=token))["groups"] + + async def group_nodes(self, token: str, group_id: str) -> list[dict]: + try: + return (await self._call("GET", f"/v1/groups/{group_id}/nodes", + token=token))["nodes"] + except HubError as e: + if e.status in (403, 404): + raise CliError(NOT_FOUND, f"The hub refused the group: {e.detail}") from e + raise + + async def mnp_token(self, token: str, node_pk: str, group_id: str) -> str: + """A token for one node and one group, never the hub session (E10).""" + r = await self._call("POST", "/v1/nodes/mnp-token", token=token, + json={"node_pk": node_pk, "group_id": group_id}) + return r["mnp_token"] + + async def offer(self, token: str, node_id: str, sdp: str) -> str: + try: + r = await self._call("POST", f"/v1/nodes/{node_id}/webrtc/offer", token=token, + json={"sdp": sdp, "ice_candidates": []}) + except HubError as e: + raise CliError(NODE, f"The node did not answer the offer: {e.detail}") from e + return r["sdp"] + + +def _sign_in_refusal(e: HubError) -> CliError: + if e.status == 401: + return CliError(AUTH, "Sign-in refused: wrong username or passphrase.") + if e.status == 403 and e.detail == "email_verification_required": + return CliError(AUTH, "This account's e-mail address is not verified yet.") + if e.status == 403: + return CliError(AUTH, f"Sign-in refused: {e.detail}.") + if e.status == 429: + return CliError(AUTH, f"Too many attempts: {e.detail}") + return CliError(AUTH, f"Sign-in failed ({e.status}): {e.detail}") diff --git a/packages/meshbay-cli/src/meshbay_cli/client/keys.py b/packages/meshbay-cli/src/meshbay_cli/client/keys.py new file mode 100644 index 0000000..faca01e --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/client/keys.py @@ -0,0 +1,137 @@ +# SPDX-License-Identifier: LGPL-3.0-or-later +""" +The keys a member derives from a passphrase, and the identities they open. + +Byte for byte what `static/keyderive.js` does (docs/MESHBAY_DESIGN.md §3.1, +§3.7); `test_bundle_kdf_parity.py` holds this module to it. A mismatch does not +look like an error — it looks like an account nobody can open. +""" + +import base64 +import hashlib +import json +from dataclasses import dataclass + +from argon2.low_level import Type, hash_secret_raw +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey +from cryptography.hazmat.primitives.ciphers.aead import AESGCM +from cryptography.hazmat.primitives.kdf.hkdf import HKDF +from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC + +PBKDF2_ITERATIONS = 600_000 +ARGON2_MEM_KIB = 131072 # 128 MB +ARGON2_TIME = 3 +ARGON2_LANES = 1 + +BUNDLE_MAGIC = b"MBK3" +LEGACY_MAGIC = b"MBK2" # TRANSITIONAL — read, never written + + +def b64(data: bytes) -> str: + return base64.b64encode(data).decode() + + +def unb64(text: str) -> bytes: + return base64.b64decode(text) + + +def hkdf(key: bytes, info: str) -> bytes: + return HKDF(algorithm=hashes.SHA256(), length=32, salt=None, + info=info.encode()).derive(key) + + +def auth_key(username: str, passphrase: str) -> str: + """What the hub checks at sign-in. The passphrase itself never leaves here.""" + salt = hashlib.sha256(f"meshbay:auth:v1:{username}".encode()).digest() + return b64(PBKDF2HMAC(algorithm=hashes.SHA512(), length=32, salt=salt, + iterations=PBKDF2_ITERATIONS).derive(passphrase.encode())) + + +def passphrase_half(username: str, passphrase: str) -> bytes: + """`A`: Argon2id of the passphrase, salted by the account name.""" + salt = hashlib.sha256(f"meshbay:bundle:v2:{username}".encode()).digest()[:16] + return hash_secret_raw(passphrase.encode(), salt, time_cost=ARGON2_TIME, + memory_cost=ARGON2_MEM_KIB, parallelism=ARGON2_LANES, + hash_len=32, type=Type.ID) + + +def bundle_master_key(a: bytes, pepper: bytes, user_id: str) -> bytes: + """`M`, from `A` and the pepper the hub hands out at sign-in.""" + return hkdf(a + pepper, f"meshbay:bundle-master:v3|{user_id}") + + +def node_bundle_key(master: bytes, node_pk: str) -> bytes: + return hkdf(master, f"meshbay:bundle:v3|node|{node_pk}") + + +class BundleError(Exception): + pass + + +def open_bundle(bundle_b64: str, *, master: bytes | None, legacy: bytes | None, + user_id: str, node_pk: str) -> dict: + """ + The `{skEd, skX}` sealed in a keypair bundle. + + `MBK3` opens under the node's key derived from `M`, with the account and the + node as associated data. `MBK2` opens under `A` alone; it is read so a member + is not locked out of a node still holding one, and never rewritten here — + the application or a browser replaces it. Anything older is refused by name. + """ + try: + raw = unb64(bundle_b64) + except Exception as e: + raise BundleError("the node's copy of your keys is unreadable") from e + try: + if raw[:4] == BUNDLE_MAGIC: + if master is None: + raise BundleError("no bundle key in this session") + aad = f"meshbay:bundle:v3|{user_id}|{node_pk}".encode() + plain = AESGCM(node_bundle_key(master, node_pk)).decrypt(raw[5:17], raw[17:], aad) + elif raw[:4] == LEGACY_MAGIC: + if legacy is None: + raise BundleError("this node holds an older copy of your keys; sign in " + "again with your passphrase to open it") + plain = AESGCM(legacy).decrypt(raw[4:16], raw[16:], None) + else: + raise BundleError("this node holds your keys in a format no longer read; " + "ask its operator to unpin you and send a fresh code") + except BundleError: + raise + except Exception as e: + raise BundleError("the node's copy of your keys does not open with this " + "passphrase") from e + return json.loads(plain) + + +@dataclass +class Identity: + """One account's keys on one node.""" + + sk_ed: Ed25519PrivateKey + sk_x: X25519PrivateKey + + @classmethod + def from_pkcs8(cls, keys: dict) -> "Identity": + sk_ed = serialization.load_der_private_key(unb64(keys["skEd"]), password=None) + sk_x = serialization.load_der_private_key(unb64(keys["skX"]), password=None) + if not isinstance(sk_ed, Ed25519PrivateKey) or not isinstance(sk_x, X25519PrivateKey): + raise BundleError("the node's copy of your keys holds the wrong kind of key") + return cls(sk_ed, sk_x) + + def to_pkcs8(self) -> dict: + def der(k) -> str: + return b64(k.private_bytes(serialization.Encoding.DER, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption())) + return {"skEd": der(self.sk_ed), "skX": der(self.sk_x)} + + @property + def pk_ed(self) -> str: + return b64(self.sk_ed.public_key().public_bytes_raw()) + + @property + def pk_x(self) -> str: + return b64(self.sk_x.public_key().public_bytes_raw()) diff --git a/packages/meshbay-cli/src/meshbay_cli/client/names.py b/packages/meshbay-cli/src/meshbay_cli/client/names.py new file mode 100644 index 0000000..f6d7d32 --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/client/names.py @@ -0,0 +1,69 @@ +""" +Names crossing between a group and this machine, in both directions. + +**Uploading**, the node decides: `SAFE_UPLOAD_NAME` and the Explorer-active types +in `meshbay_node/roots.py`. This is a copy of that rule, checked before anything +is sent so a folder holding one refused name is refused whole instead of +arriving with holes in it — as the Files tab does with its own copy. +`test_names.py` holds this copy to the node's answers. + +**Downloading**, a name from the index is someone else's text. It is made +writable on this platform (`meshbay_common.paths.sanitize_for_download`, the +same rule the browser applies), and the resolved path must stay under the +destination: checked after resolution, never by comparing strings. +""" + +import re +from pathlib import Path + +from meshbay_common.paths import fold, sanitize_for_download + +UPLOAD_NAME = re.compile( + r"^[^\W_]" + r"[\w .\-()\[\]'’,&+#@]{0,127}" + r"(?<![ .])$", + re.UNICODE) + +SHELL_ACTIVE_NAMES = frozenset({"desktop.ini"}) +SHELL_ACTIVE_SUFFIXES = (".lnk", ".url", ".scf", ".library-ms", ".searchconnector-ms") + + +def upload_name_problem(name: str, *, is_dir: bool = False) -> str | None: + if not UPLOAD_NAME.match(name): + return ("starts with something other than a letter or digit, ends with a space " + "or a dot, holds a character the node does not accept, or is longer " + "than 128 characters") + lower = name.lower() + if not is_dir and (lower in SHELL_ACTIVE_NAMES or lower.endswith(SHELL_ACTIVE_SUFFIXES)): + return "a file type Windows Explorer acts on by itself; no node accepts it" + return None + + +def same_name(a: str, b: str) -> bool: + """As the node and the Files tab compare names in a folder: without case.""" + return fold(a) == fold(b) + + +class UnsafeName(Exception): + pass + + +def local_path(dest: Path, parts: list[str]) -> tuple[Path, bool]: + """ + Where a remote file or folder lands under `dest`, and whether a name had to + be changed to be written here. + """ + renamed = False + out = dest + for part in parts: + safe = sanitize_for_download(part) + if safe in (".", ".."): + safe = "unnamed" + renamed |= safe != part + out = out / safe + base = dest.resolve() + try: + out.resolve().relative_to(base) + except ValueError: + raise UnsafeName(f"{'/'.join(parts)!r} would land outside {dest}") from None + return out, renamed diff --git a/packages/meshbay-cli/src/meshbay_cli/client/paths.py b/packages/meshbay-cli/src/meshbay_cli/client/paths.py new file mode 100644 index 0000000..1086085 --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/client/paths.py @@ -0,0 +1,95 @@ +""" +Remote paths: `<group>[@<owner>]/<root>/<folder>/.../<file>`. + +The same handle the application shows under every group and puts in its links +(docs/MESHBAY_DESIGN.md §8.4, `static/group-link.js`), without the `#/`. + +A group name may contain `/` and `@` (the hub refuses only control characters +and bidirectional overrides); a username contains neither. So the owner is +whatever follows the **last** `@` of the first segment, and a `/` inside a name +is written `%2F`. Only `%2F` and `%25` are decoded in a typed path — a file +called `100%.txt` is typed as it is. A link copied from the application +(anything containing `#/`) is decoded as the application encodes it, every +segment in full. +""" + +import unicodedata +from dataclasses import dataclass +from urllib.parse import unquote + +from ..errors import NOT_FOUND, USAGE, CliError + + +@dataclass +class RemotePath: + group: str # the first segment, decoded: a name, maybe with @owner + parts: list[str] # root, folders, file — decoded + from_link: bool = False + + @property + def inner(self) -> str: + return "/".join(self.parts) + + +def _typed(segment: str) -> str: + return segment.replace("%2F", "/").replace("%2f", "/").replace("%25", "%") + + +def parse(text: str) -> RemotePath | None: + """None for "all my groups" (empty, or `/`).""" + from_link = "#/" in text + if from_link: + text = text.split("#", 1)[1] + segments = [s for s in text.split("/") if s] + if not segments: + return None + decode = unquote if from_link else _typed + decoded = [unicodedata.normalize("NFC", decode(s)) for s in segments] + if any(s in (".", "..") for s in decoded[1:]): + raise CliError(USAGE, "A remote path cannot contain . or .. segments.") + return RemotePath(decoded[0], decoded[1:], from_link) + + +def encode_name(name: str) -> str: + """A group or file name as it is typed in a path.""" + return name.replace("%", "%25").replace("/", "%2F") + + +def handle(group: dict) -> str: + """`name@owner`, ready to be typed back.""" + return f"{encode_name(group['name'])}@{group.get('owner_username') or ''}" + + +def _by_owner(named: list[dict], owner: str) -> dict | None: + exact = [g for g in named if g.get("owner_username") == owner] + if exact: + return exact[0] + loose = [g for g in named if (g.get("owner_username") or "").lower() == owner.lower()] + return loose[0] if len(loose) == 1 else None + + +def resolve_group(groups: list[dict], spec: str) -> dict: + """ + The group `spec` names among `/v1/groups/mine` rows. + + Names compare without case, as the hub keeps them unique per owner. With an + `@`, `name@owner` is tried first; when it names nothing, the whole segment is + tried as a name, since a group name may contain an `@` of its own. + """ + def named(name: str) -> list[dict]: + n = name.lower() + return [g for g in groups if (g.get("name") or "").lower() == n] + + at = spec.rfind("@") + if 0 < at < len(spec) - 1: + found = _by_owner(named(spec[:at]), spec[at + 1:]) + if found is not None: + return found + candidates = named(spec) + if len(candidates) == 1: + return candidates[0] + if len(candidates) > 1: + options = ", ".join(sorted(handle(g) for g in candidates)) + raise CliError(USAGE, f"Several of your groups are called {spec!r}; " + f"name the owner: {options}") + raise CliError(NOT_FOUND, f"You are not a member of a group called {spec!r}.") diff --git a/packages/meshbay-cli/src/meshbay_cli/client/pool.py b/packages/meshbay-cli/src/meshbay_cli/client/pool.py new file mode 100644 index 0000000..00e69fb --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/client/pool.py @@ -0,0 +1,48 @@ +""" +One connection per group for the whole run, and a ceiling on how many are being +negotiated at once — the application's `ConnectionPool` rules +(docs/MESHBAY_DESIGN.md §9.11). + +The ceiling is on concurrency, never a batch: a group whose node does not +answer holds one place for its own deadline and delays nobody else. Six stays +far below what the hub admits per account (32 pending offers), even with the +person's application and a browser open at the same time. +""" + +import asyncio +import contextlib + +from .account import Account +from .groups import open_group +from .session import GroupSession + +MAX_IN_FLIGHT = 6 + + +class Pool: + def __init__(self, account: Account, max_in_flight: int = MAX_IN_FLIGHT): + self.account = account + self._gate = asyncio.Semaphore(max_in_flight) + self._sessions: dict[str, asyncio.Task] = {} + + async def session(self, group: dict) -> GroupSession: + """The group's session, negotiated once and shared by every caller.""" + task = self._sessions.get(group["id"]) + if task is None: + task = asyncio.ensure_future(self._open(group)) + self._sessions[group["id"]] = task + return await asyncio.shield(task) + + async def _open(self, group: dict) -> GroupSession: + async with self._gate: + return await open_group(self.account, group) + + async def close(self) -> None: + for task in self._sessions.values(): + if task.done() and not task.cancelled() and task.exception() is None: + await task.result().close() + else: + task.cancel() + with contextlib.suppress(BaseException): + await task + self._sessions.clear() diff --git a/packages/meshbay-cli/src/meshbay_cli/client/session.py b/packages/meshbay-cli/src/meshbay_cli/client/session.py new file mode 100644 index 0000000..1ea398a --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/client/session.py @@ -0,0 +1,443 @@ +""" +A member's connection to one node, for one group. + +The hub relays the WebRTC offer and the answer and nothing after; everything +from the handshake on goes straight to the node, sealed under the group key +(docs/MESHBAY_DESIGN.md §5.1, §5.2). The steps, in order: + +1. a token for this node and this group only (`/v1/nodes/mnp-token`, E10); +2. the DataChannel, and the channel binding from both DTLS fingerprints; +3. the handshake: the node signs its challenge with the key it announces, and + that key must be the one pinned for this node, if any (§5.2, TOFU); +4. this account's identity on the node — kept from an earlier run, or opened + from the bundle the node keeps for browsers; +5. the group key, wrapped by the node for that identity (`join_request`); +6. the mutual proof, the ack verified and unsealed — a payload that does not + open ends the session, never reads as an empty group (§4.4); +7. `device_hello`, so the node knows which of the account's devices this is. + +Every request carries a `req_id` and its reply is matched on it (§5.3): matching +by arrival order is a guess that fails silently, and against the unrelated +request. +""" + +import asyncio +import contextlib +import itertools +import logging +import os +import struct +import time +from datetime import UTC, datetime + +import msgpack +from aiortc import RTCConfiguration, RTCIceServer, RTCPeerConnection, RTCSessionDescription +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives.asymmetric.ed25519 import ( + Ed25519PrivateKey, + Ed25519PublicKey, +) +from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey +from meshbay_common import MNP_VERSION +from meshbay_common.crypto import unwrap_gek_aes +from meshbay_common.device import ( + device_code_hash, + device_hello_transcript, + device_request_transcript, +) +from meshbay_common.groupbox import PURPOSE_ACK, PURPOSE_INDEX, unseal +from meshbay_common.handshake import ( + MNP_MIN_SUPPORTED, + HandshakeError, + challenge_transcript, + check_version, + handshake_transcript, + make_proof, + verify_proof, + webrtc_binding, +) +from meshbay_common.join import join_transcript + +from ..errors import NODE, NODE_KEY, VERSION, CliError +from .account import Account +from .keys import BundleError, Identity, b64, open_bundle, unb64 + +log = logging.getLogger(__name__) + +# The browser's list (transport.js), two providers deep: one slow server must +# not add its whole timeout to every connection. +STUN_SERVERS = [ + "stun:stun.l.google.com:19302", + "stun:stun1.l.google.com:19302", + "stun:stun.cloudflare.com:3478", +] + +OPEN_TIMEOUT = 30 # ICE + DTLS + SCTP, measured at 0.6–7 s +REPLY_TIMEOUT = 60 + + +class NodeRefusal(CliError): + def __init__(self, detail: str, code: str = ""): + super().__init__(NODE, f"The node refused: {detail}") + self.detail = detail + self.refusal_code = code + + +class NodeKeyChanged(CliError): + def __init__(self, node_id: str, pinned: str, now: str): + super().__init__( + NODE_KEY, + f"The node {node_id} presents a different key than the one pinned at " + f"first contact.\n pinned: {pinned}\n now: {now}\n" + "If its operator announced a new key, run `meshbay-cli trust reset " + f"{node_id}`; otherwise do not connect.") + + +class Channel: + """MNP frames over a DataChannel: a 4-byte big-endian length, then msgpack.""" + + def __init__(self, dc): + self.dc = dc + self._buf = bytearray() + self._ids = itertools.count(1) + self._pending: dict[str, asyncio.Future] = {} + self._push_handlers: dict[str, list] = {} + self.closed = asyncio.Event() + dc.on("message", self._on_message) + dc.on("close", self._on_close) + + def on_push(self, msg_type: str, handler) -> None: + """Call `handler(msg)` for every unsolicited message of this type.""" + self._push_handlers.setdefault(msg_type, []).append(handler) + + def off_push(self, msg_type: str, handler) -> None: + with contextlib.suppress(KeyError, ValueError): + self._push_handlers[msg_type].remove(handler) + + def _on_close(self) -> None: + self.closed.set() + for fut in self._pending.values(): + if not fut.done(): + fut.set_exception(CliError(NODE, "The connection to the node closed.")) + + def _on_message(self, data) -> None: + self._buf.extend(data if isinstance(data, bytes) else data.encode()) + while len(self._buf) >= 4: + size = struct.unpack(">I", self._buf[:4])[0] + if len(self._buf) < 4 + size: + break + msg = msgpack.unpackb(bytes(self._buf[4:4 + size]), raw=False) + del self._buf[:4 + size] + self._route(msg) + + def _route(self, msg: dict) -> None: + # Type and id only: payloads carry sealed content and key material. + log.debug("← %s req_id=%s", msg.get("type"), msg.get("req_id")) + fut = self._pending.pop(str(msg.get("req_id", "")), None) + if fut is not None and not fut.done(): + fut.set_result(msg) + return + handlers = self._push_handlers.get(str(msg.get("type")), []) + if not handlers and msg.get("type") == "error": + # A refusal that names no request answers none of ours: logged, + # never handed to whichever request happens to be waiting. + log.debug("unattributed error from the node: %s", msg.get("detail")) + for handler in list(handlers): + try: + handler(msg) + except Exception: + log.exception("push handler for %s failed", msg.get("type")) + + def send(self, msg: dict) -> None: + log.debug("→ %s req_id=%s", msg.get("type"), msg.get("req_id")) + data = msgpack.packb(msg, use_bin_type=True) + self.dc.send(struct.pack(">I", len(data)) + data) + + async def request(self, msg: dict, *expect: str, timeout: float = REPLY_TIMEOUT) -> dict: + req_id = str(next(self._ids)) + fut = asyncio.get_running_loop().create_future() + self._pending[req_id] = fut + self.send({**msg, "req_id": req_id}) + try: + reply = await asyncio.wait_for(fut, timeout) + except TimeoutError: + raise CliError(NODE, f"The node did not answer {msg.get('type')} " + f"within {timeout:.0f} s.") from None + finally: + self._pending.pop(req_id, None) + if reply.get("type") == "error": + raise NodeRefusal(str(reply.get("detail") or "no reason given"), + str(reply.get("code") or "")) + if expect and reply.get("type") not in expect: + raise CliError(NODE, f"The node answered {msg.get('type')} with " + f"{reply.get('type')}.") + return reply + + +def _dtls_fingerprint(sdp: str) -> bytes: + for line in sdp.splitlines(): + if line.startswith("a=fingerprint:sha-256 "): + return bytes.fromhex(line.split(" ", 1)[1].strip().replace(":", "")) + raise CliError(NODE, "No DTLS fingerprint in the session description.") + + +class GroupSession: + """An authenticated connection to one node, for one group.""" + + def __init__(self, pc: RTCPeerConnection, ch: Channel, group_id: str, + node_id: str, node_pk: str, gek: bytes, ack: dict): + self.pc = pc + self.ch = ch + self.group_id = group_id + self.node_id = node_id + self.node_pk = node_pk + self.gek = gek + self.ack = ack + + async def index(self) -> dict: + """`{version, entries, dirs, roots}`, opened under the group key.""" + msg = await self.ch.request({"type": "index_sync", "v": MNP_VERSION}, "index_sync") + try: + return unseal(self.gek, PURPOSE_INDEX, "index_sync", self.group_id, msg) + except Exception as e: + raise CliError(NODE, "The group's index does not open under the group key; " + "the session is ended.") from e + + async def close(self) -> None: + with contextlib.suppress(Exception): + await self.pc.close() + + +async def connect(account: Account, group_id: str, node: dict) -> GroupSession: + """Open a session with `node` (a `/v1/groups/{id}/nodes` row) for `group_id`.""" + node_id, listed_pk = node["node_id"], node["pk_node"] + pinned = account.pin(node_id) + if pinned and pinned != listed_pk: + raise NodeKeyChanged(node_id, pinned, listed_pk) + + mnp_token = await account.call(account.hub.mnp_token, listed_pk, group_id) + + pc = RTCPeerConnection(RTCConfiguration( + iceServers=[RTCIceServer(urls=u) for u in STUN_SERVERS])) + try: + dc = pc.createDataChannel("mnp", ordered=True) + ch = Channel(dc) + opened = asyncio.Event() + dc.on("open", opened.set) + await pc.setLocalDescription(await pc.createOffer()) + offer_sdp = pc.localDescription.sdp + answer_sdp = await account.call(account.hub.offer, node_id, offer_sdp) + await pc.setRemoteDescription(RTCSessionDescription(sdp=answer_sdp, type="answer")) + try: + await asyncio.wait_for(opened.wait(), OPEN_TIMEOUT) + except TimeoutError: + raise CliError(NODE, f"No connection to the node within {OPEN_TIMEOUT} s " + "(it is online at the hub, but not reachable from " + "here).") from None + binding = webrtc_binding(_dtls_fingerprint(offer_sdp), _dtls_fingerprint(answer_sdp)) + return await _handshake(account, pc, ch, group_id, node_id, listed_pk, + mnp_token, binding) + except BaseException: + with contextlib.suppress(Exception): + await pc.close() + raise + + +async def _handshake(account: Account, pc, ch: Channel, group_id: str, node_id: str, + listed_pk: str, mnp_token: str, binding: bytes) -> GroupSession: + nonce_c = os.urandom(32) + challenge = await ch.request( + {"type": "handshake", "v": MNP_VERSION, "v_min": MNP_MIN_SUPPORTED, + "token": mnp_token, "group_id": group_id, "nonce": b64(nonce_c)}, + "handshake_challenge") + try: + check_version(challenge["v"], challenge.get("v_min", "")) + except HandshakeError as e: + raise CliError(VERSION, f"Protocol mismatch with the node: {e}") from e + nonce_s = unb64(challenge["nonce"]) + node_pk = challenge["node_pk"] + if node_pk != listed_pk: + raise CliError(NODE, "The node answering is not the one the hub listed.") + try: + Ed25519PublicKey.from_public_bytes(unb64(node_pk)).verify( + unb64(challenge.get("sig", "")), + challenge_transcript(group_id, nonce_c, nonce_s, binding)) + except (InvalidSignature, ValueError) as e: + raise CliError(NODE, "The node's challenge is not signed by the key it " + "announces.") from e + + identity, source = await _identity(account, ch, node_pk) + + ts = int(time.time()) + joined = await ch.request({ + "type": "join_request", "v": MNP_VERSION, "group_id": group_id, + "pk_ed25519": identity.pk_ed, "pk_x25519": identity.pk_x, "code": "", "ts": ts, + "sig": b64(identity.sk_ed.sign(join_transcript( + node_pk_b64=node_pk, group_id=group_id, user_id=account.user_id, + pk_ed25519_b64=identity.pk_ed, pk_x25519_b64=identity.pk_x, + nonce_node=nonce_s, ts=ts)))}, "join_result") + if not joined.get("ok") or not joined.get("gek"): + reason = str(joined.get("reason") or "") + if reason == "unknown_device" and source != "bundle": + await _link(account, ch, node_pk, nonce_s, identity, source) + raise NodeRefusal(JOIN_REFUSALS.get(reason, reason or "the group key was not given"), + reason) + if source == "pending": + await account.adopt_pending(node_pk) + gek = unwrap_gek_aes(joined, identity.sk_x.private_bytes_raw(), + identity.sk_x.public_key().public_bytes_raw()) + + ack = await ch.request({"type": "handshake_response", "v": MNP_VERSION, + "proof": b64(make_proof(gek, "client", group_id, nonce_c, + nonce_s, binding))}, + "handshake_ack") + if not verify_proof(gek, unb64(ack["proof"]), "node", group_id, nonce_c, nonce_s, + binding): + raise CliError(NODE, "The node does not hold the group key.") + if ack.get("node_pk") != node_pk: + raise CliError(NODE, "The node changed identity during the handshake.") + try: + Ed25519PublicKey.from_public_bytes(unb64(node_pk)).verify( + unb64(ack["sig"]), handshake_transcript("node", group_id, nonce_c, nonce_s, + binding)) + except (InvalidSignature, ValueError, KeyError) as e: + raise CliError(NODE, "The node's acknowledgement is not signed by its key.") from e + try: + config = unseal(gek, PURPOSE_ACK, "handshake_ack", group_id, ack) + except Exception as e: + raise CliError(NODE, "The node's configuration does not open under the group " + "key; the session is ended.") from e + + if account.pin(node_id) is None: + await account.set_pin(node_id, node_pk) + + await _device_hello(account, ch, identity, node_pk, group_id, nonce_s) + return GroupSession(pc, ch, group_id, node_id, node_pk, gek, config) + + +async def _identity(account: Account, ch: Channel, node_pk: str) -> tuple[Identity, str]: + """ + This account's keys on this node, and where they came from: kept from an + earlier run, waiting for a device approval, opened from the copy the node + keeps for browsers — or, failing all three, new, for device linking. + """ + kept = account.identity(node_pk) + if kept is not None: + return kept, "kept" + pending = account.pending_identity(node_pk) + if pending is not None: + return pending.identity, "pending" + found = await ch.request({"type": "keypair_bundle_fetch", "v": MNP_VERSION}, + "keypair_bundle_resp") + if not found.get("found"): + return Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate()), "new" + try: + keys = open_bundle(found["bundle_enc"], master=account.master, + legacy=account.legacy, user_id=account.user_id, node_pk=node_pk) + identity = Identity.from_pkcs8(keys) + except BundleError as e: + raise CliError(NODE, str(e)) from e + await account.keep_identity(node_pk, identity) + return identity, "bundle" + + +# What a refused join means, for the person reading it. +JOIN_REFUSALS = { + "code_required": "this node has not admitted you yet. Accept the invitation in the " + "MeshBay application and enter its code there; the command line " + "does not join groups", + "unknown_device": "the key this node holds for you is not one of your devices there " + "any more; ask its operator, or approve this machine from a device " + "already linked", + "key_changed": "your identity on this node no longer matches the one it pinned", + "not_authorized_for_group": "this node does not count you a member of the group", +} + +# The alphabet of every MeshBay code (Crockford base32, no I L O U), so the one +# this machine shows is typed like an invitation code, and needs no normalising. +CODE_ALPHABET = "0123456789ABCDEFGHJKMNPQRSTVWXYZ" + + +class DeviceLinkPending(CliError): + """This machine asked to be a device of the account on a node; it waits for an + approval from one already linked there.""" + + def __init__(self, link_code: str, expires_at: float): + super().__init__(NODE, f"Waiting for this machine to be approved, code {link_code}.") + # Not `code`: on a CliError that is the exit status. + self.link_code = link_code + self.expires_at = expires_at + + +async def _link(account: Account, ch: Channel, node_pk: str, nonce_s: bytes, + identity: Identity, source: str) -> None: + """ + File a device request bound to a code this machine shows (design §3.3), or + show again the one already filed. Always raises `DeviceLinkPending`. + + The code is hashed together with this machine's keys, so the device that + approves cannot be handed a substituted key; the code itself never reaches + the node. + """ + if not account.remembered: + raise CliError(NODE, "This machine is not yet a device of yours on this node. " + "Run `meshbay-cli login` first: linking it only makes sense " + "on a machine that keeps the identity it links.") + if source == "pending": + pending = account.pending_identity(node_pk) + raise DeviceLinkPending(pending.code, pending.expires_at) + if source == "kept": + # Revoked or unpinned on the node since this machine kept it. + await account.forget_identity(node_pk) + identity = Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate()) + raw = "".join(CODE_ALPHABET[b % 32] for b in os.urandom(8)) + code = f"{raw[:4]}-{raw[4:]}" + code_hash = device_code_hash(raw, identity.pk_ed, identity.pk_x) + ts = int(time.time()) + sig = identity.sk_ed.sign(device_request_transcript( + node_pk_b64=node_pk, user_id=account.user_id, pk_ed25519_b64=identity.pk_ed, + pk_x25519_b64=identity.pk_x, code_hash=code_hash, nonce_node=nonce_s, ts=ts)) + try: + ack = await ch.request({"type": "device_add_request", "v": MNP_VERSION, + "pk_ed25519": identity.pk_ed, "pk_x25519": identity.pk_x, + "code_hash": code_hash, "ts": ts, "sig": b64(sig)}, + "device_add_request_ack") + except NodeRefusal as e: + if e.detail != "Handshake required": + raise + # The node answers a device request only on a session that has proved + # the group key, which a machine with no identity there cannot do. + raise CliError(NODE, "This machine is not yet a device of yours on this node, and " + "the node does not accept a device request from a machine " + "that holds no key there. Turn on Browser access in the " + "MeshBay application's settings, then run the command " + "again.") from e + expires = _parse_time(ack.get("expires_at")) or time.time() + 3600 + await account.keep_pending(node_pk, identity, code, expires) + raise DeviceLinkPending(code, expires) + + +def _parse_time(text) -> float | None: + try: + moment = datetime.fromisoformat(str(text)) + except ValueError: + return None + if moment.tzinfo is None: + moment = moment.replace(tzinfo=UTC) + return moment.timestamp() + + +async def _device_hello(account: Account, ch: Channel, identity: Identity, node_pk: str, + group_id: str, nonce_s: bytes) -> None: + ts = int(time.time()) + sig = identity.sk_ed.sign(device_hello_transcript( + node_pk_b64=node_pk, group_id=group_id, user_id=account.user_id, + pk_ed25519_b64=identity.pk_ed, nonce_node=nonce_s, ts=ts)) + try: + await ch.request({"type": "device_hello", "v": MNP_VERSION, + "pk_ed25519": identity.pk_ed, "ts": ts, "sig": b64(sig)}, + "device_hello_ack") + except NodeRefusal as e: + # Listing does not depend on it; what is lost is the attribution of an + # upload to this device rather than to the account's oldest one. + log.warning("The node did not accept this device's hello: %s", e.detail) diff --git a/packages/meshbay-cli/src/meshbay_cli/client/store.py b/packages/meshbay-cli/src/meshbay_cli/client/store.py new file mode 100644 index 0000000..c48fe89 --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/client/store.py @@ -0,0 +1,251 @@ +""" +What the CLI remembers between runs. + +Two files in the data directory. `state.json` holds what is not secret: the hub, +the account, the device's id, each node's pinned key. `secrets.enc` holds the +rest — tokens, the device's hub key, the bundle key `M`, the identity on each +node — under AES-256-GCM with a 32-byte **store key**. + +The store key lives in the OS key storage (Windows Credential Manager, Secret +Service on a Linux desktop). One key there and a file beside it, rather than +every secret there: Credential Manager caps a credential at 2560 bytes, and the +identities grow with the number of nodes. Where there is no key storage — an +SSH session on a server — the key is kept in a file only this user can read, +and `login` says so once (`file_key_notice`). + +Every write goes through `locked()`: the hub rotates refresh tokens and revokes +the whole session when a spent one comes back, so two runs renewing at once +must take turns. +""" + +import contextlib +import hashlib +import json +import logging +import os +import secrets +import sys +import time +from collections.abc import Iterator +from pathlib import Path + +from cryptography.hazmat.primitives.ciphers.aead import AESGCM + +from ..errors import AUTH, CliError + +log = logging.getLogger(__name__) + +KEYRING_SERVICE = "meshbay-cli" +SECRETS_MAGIC = b"MBC1" +SECRETS_AAD = b"meshbay-cli:secrets:v1" + +KEY_OS = "os" +KEY_FILE = "file" + +FILE_KEY_NOTICE = ( + "No OS key storage here: the session is kept in a file only your user can " + "read (like an SSH key without a passphrase). `meshbay-cli logout` removes it." +) + + +def data_dir() -> Path: + """Never the node's directories (`~/.config/meshbay`, `%LOCALAPPDATA%\\meshbay`).""" + override = os.environ.get("MESHBAY_CLI_HOME") + if override: + return Path(override) + if sys.platform == "win32": + return Path(os.environ.get("LOCALAPPDATA") or Path.home() / "AppData" / "Local") \ + / "meshbay-cli" + base = os.environ.get("XDG_DATA_HOME") or Path.home() / ".local" / "share" + return Path(base) / "meshbay-cli" + + +def _write_private(path: Path, data: bytes) -> None: + """Atomically, readable by this user only.""" + tmp = path.with_name(path.name + ".tmp") + fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + try: + os.write(fd, data) + os.fsync(fd) + finally: + os.close(fd) + os.replace(tmp, path) + + +class _Keyring: + """ + The OS key storage. One entry per store directory, so two stores (another + `MESHBAY_CLI_HOME`) never overwrite each other's key. + """ + + def __init__(self, root: Path): + digest = hashlib.sha256(str(root.resolve()).encode()).hexdigest()[:16] + self.entry = f"store-key:{digest}" + + def usable(self) -> bool: + try: + import keyring + from keyring.backends import fail + backend = keyring.get_keyring() + if isinstance(backend, fail.Keyring): + return False + # A Secret Service that is installed but not running answers the + # first call with an exception rather than at selection time. + keyring.get_password(KEYRING_SERVICE, self.entry) + return True + except Exception as e: + log.debug("OS key storage unavailable: %s", e) + return False + + def get(self) -> bytes | None: + import keyring + value = keyring.get_password(KEYRING_SERVICE, self.entry) + return bytes.fromhex(value) if value else None + + def set(self, key: bytes) -> None: + import keyring + keyring.set_password(KEYRING_SERVICE, self.entry, key.hex()) + + def delete(self) -> None: + import keyring + with contextlib.suppress(Exception): + keyring.delete_password(KEYRING_SERVICE, self.entry) + + +class Store: + def __init__(self, root: Path | None = None): + self.root = root or data_dir() + self._keyring = _Keyring(self.root) + self.state: dict = {} + self.secrets: dict = {} + + @property + def _state_path(self) -> Path: + return self.root / "state.json" + + @property + def _secrets_path(self) -> Path: + return self.root / "secrets.enc" + + @property + def _key_path(self) -> Path: + return self.root / "store.key" + + def exists(self) -> bool: + return self._state_path.exists() and self._secrets_path.exists() + + # ── Locking ────────────────────────────────────────────────────────────── + + @contextlib.contextmanager + def locked(self) -> Iterator[None]: + """An exclusive lock on the store, across processes. Blocks.""" + self.root.mkdir(parents=True, exist_ok=True) + fd = os.open(self.root / "store.lock", os.O_RDWR | os.O_CREAT, 0o600) + try: + if sys.platform == "win32": + import msvcrt + while True: + try: + msvcrt.locking(fd, msvcrt.LK_NBLCK, 1) + break + except OSError: + time.sleep(0.1) + else: + import fcntl + fcntl.flock(fd, fcntl.LOCK_EX) + yield + finally: + if sys.platform == "win32": + import msvcrt + with contextlib.suppress(OSError): + os.lseek(fd, 0, os.SEEK_SET) + msvcrt.locking(fd, msvcrt.LK_UNLCK, 1) + os.close(fd) + + # ── The store key ──────────────────────────────────────────────────────── + + def _load_key(self) -> bytes: + kind = self.state.get("key_store") + if kind == KEY_FILE: + try: + return bytes.fromhex(self._key_path.read_text().strip()) + except OSError as e: + raise CliError(AUTH, f"The session's key file is missing ({e.strerror}); " + "run `meshbay-cli login`.") from e + if not self._keyring.usable(): + raise CliError(AUTH, "The session was saved in the OS key storage, which this " + "session cannot reach. Run `meshbay-cli login` here, or " + "pass --password.") + key = self._keyring.get() + if key is None: + raise CliError(AUTH, "The session's key is gone from the OS key storage; " + "run `meshbay-cli login`.") + return key + + def _new_key(self) -> tuple[bytes, str]: + key = secrets.token_bytes(32) + if self._keyring.usable(): + try: + self._keyring.set(key) + return key, KEY_OS + except Exception as e: + log.debug("OS key storage refused the key: %s", e) + _write_private(self._key_path, key.hex().encode()) + return key, KEY_FILE + + # ── Reading and writing ────────────────────────────────────────────────── + + def peek(self) -> dict: + """state.json alone — the hub and the account, which are not secret.""" + try: + return json.loads(self._state_path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return {} + + def load(self) -> bool: + """Read both files. False when nothing is stored.""" + if not self.exists(): + self.state, self.secrets = {}, {} + return False + self.state = json.loads(self._state_path.read_text(encoding="utf-8")) + raw = self._secrets_path.read_bytes() + if raw[:4] != SECRETS_MAGIC: + raise CliError(AUTH, "The saved session is unreadable; run `meshbay-cli login`.") + try: + plain = AESGCM(self._load_key()).decrypt(raw[4:16], raw[16:], SECRETS_AAD) + except CliError: + raise + except Exception as e: + raise CliError(AUTH, "The saved session does not open with its key; run " + "`meshbay-cli login`.") from e + self.secrets = json.loads(plain) + return True + + def save(self, *, new: bool = False) -> None: + """Write both files. `new` makes a fresh store key (a new sign-in).""" + self.root.mkdir(parents=True, exist_ok=True) + if new or "key_store" not in self.state: + self._forget_key() + key, kind = self._new_key() + self.state["key_store"] = kind + else: + key = self._load_key() + nonce = secrets.token_bytes(12) + ct = AESGCM(key).encrypt(nonce, json.dumps(self.secrets).encode(), SECRETS_AAD) + _write_private(self._secrets_path, SECRETS_MAGIC + nonce + ct) + _write_private(self._state_path, json.dumps(self.state, indent=1).encode()) + + def _forget_key(self) -> None: + kind = self.state.get("key_store") + if kind == KEY_OS: + self._keyring.delete() + with contextlib.suppress(FileNotFoundError): + self._key_path.unlink() + + def erase(self) -> None: + """Everything, key included.""" + self._forget_key() + for p in (self._secrets_path, self._state_path): + with contextlib.suppress(FileNotFoundError): + p.unlink() + self.state, self.secrets = {}, {} diff --git a/packages/meshbay-cli/src/meshbay_cli/client/transfers.py b/packages/meshbay-cli/src/meshbay_cli/client/transfers.py new file mode 100644 index 0000000..3f23c85 --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/client/transfers.py @@ -0,0 +1,382 @@ +""" +Downloads and uploads: one transfer lease each, then the chunks. + +**The unit is the lease** (docs/MESHBAY_DESIGN.md §5.5): the node's record that +this connection is transferring something, opened before the first chunk and +closed after the last, on every path out. A lease may be granted at once or +queued behind the member's other transfers; the grant then arrives as a push. + +Downloads are 1 MiB chunks, each encrypted under its own key derived from the +group key and the file's hash, and each independently decryptable — which is +what makes resuming a download cost nothing in the protocol. Uploads are 48 KB +chunks sealed under the group key, name and folder inside the seal; the node +keeps its own position, and a probe chunk asks for it, which is what resumes +an upload. +""" + +import asyncio +import contextlib +import json +import logging +import math +import os +import secrets +import sys +from collections.abc import Callable +from dataclasses import dataclass +from pathlib import Path + +from meshbay_common import MNP_VERSION +from meshbay_common.groupbox import PURPOSE_UPLOAD, unseal +from meshbay_common.paths import long_path +from meshbay_common.protocol import UPLOAD_PROBE_INDEX, file_upload_wire +from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes + +from ..errors import NODE, CliError +from .session import GroupSession, NodeRefusal + +log = logging.getLogger(__name__) + +DOWNLOAD_CHUNK = 1024 * 1024 # the node serves files in 1 MiB chunks +UPLOAD_CHUNK = 48 * 1024 # what fits one DataChannel message after sealing +DOWNLOAD_WINDOW = 8 # chunk requests in flight per file +UPLOAD_WINDOW = 32 # chunks sent ahead of the node's acks +CHUNK_TIMEOUT = 60 +# A queued lease waits behind the member's own transfers and other members'; +# a queue that has said nothing for this long is not moving. +QUEUE_SILENCE = 600 + +PART_SUFFIX = ".meshbay-part" + +Progress = Callable[[int], None] # bytes just completed + + +class TransferError(CliError): + """One file failed; the run goes on unless `fatal`.""" + + def __init__(self, message: str, *, fatal: bool = False, code: str = ""): + super().__init__(NODE, message) + self.fatal = fatal + self.refusal_code = code + + +# ── Leases ─────────────────────────────────────────────────────────────────── + +class Lease: + def __init__(self, session: GroupSession, kind: str): + self.session = session + self.kind = kind + self.tr = secrets.token_hex(16) + self.closed_by_node = "" + self._state: asyncio.Queue = asyncio.Queue() + + def _on_state(self, msg: dict) -> None: + if msg.get("tr") != self.tr: + return + if msg.get("state") == "closed": + self.closed_by_node = str(msg.get("reason") or "closed by the node") + self._state.put_nowait(msg) + + async def open(self, size: int, chunks: int, + on_queued: Callable[[int], None] | None = None) -> None: + ch = self.session.ch + ch.on_push("transfer_state", self._on_state) + reply = await ch.request({"type": "transfer_open", "v": MNP_VERSION, "tr": self.tr, + "kind": self.kind, "bytes": size, "chunks": chunks}, + "transfer_state") + state = reply.get("state") + while state == "queued": + if on_queued: + on_queued(int(reply.get("ahead") or 0)) + try: + reply = await asyncio.wait_for(self._state.get(), QUEUE_SILENCE) + except TimeoutError: + raise TransferError("The node's transfer queue did not move for " + f"{QUEUE_SILENCE // 60} minutes.") from None + state = reply.get("state") + if state != "granted": + raise TransferError(f"The node did not grant the transfer " + f"({reply.get('reason') or state}).") + + def check(self) -> None: + if self.closed_by_node: + raise TransferError(f"The node ended the transfer ({self.closed_by_node}).") + + async def close(self, reason: str) -> None: + ch = self.session.ch + ch.off_push("transfer_state", self._on_state) + if self.closed_by_node or ch.closed.is_set(): + return + with contextlib.suppress(CliError): + await ch.request({"type": "transfer_close", "v": MNP_VERSION, "tr": self.tr, + "reason": reason}, "transfer_state", timeout=10) + + +# ── Download ───────────────────────────────────────────────────────────────── + +@dataclass +class Resume: + """What a `.meshbay-part` beside the target already holds, if it is this file.""" + + part: Path + sidecar: Path + chunks_done: int + + @classmethod + def find(cls, target: Path, entry: dict) -> "Resume": + part = target.with_name(target.name + PART_SUFFIX) + sidecar = target.with_name(target.name + PART_SUFFIX + ".json") + done = 0 + try: + meta = json.loads(sidecar.read_text(encoding="utf-8")) + if meta.get("id") == entry["id"] and meta.get("size") == entry["size"]: + # Every chunk but the last is whole, so the complete ones are + # what the part's length says. + done = os.path.getsize(long_path(part)) // DOWNLOAD_CHUNK + except (OSError, ValueError): + done = 0 + return cls(part, sidecar, done) + + +async def download(session: GroupSession, entry: dict, target: Path | None, *, + overwrite: bool = False, progress: Progress | None = None, + on_queued: Callable[[int], None] | None = None) -> int: + """ + Fetch `entry` into `target` (None: stdout). Returns the bytes written. + + Written to `<name>.meshbay-part` and renamed at the end, so an interrupted run + never leaves a truncated file under the real name; the part is resumed from + its last whole chunk when the sidecar says it is this file. + """ + size = int(entry["size"]) + chunks = max(1, math.ceil(size / DOWNLOAD_CHUNK)) + file_hash = bytes.fromhex(entry["id"]) + + resume = None + if target is not None: + resume = Resume.find(target, entry) + start = min(resume.chunks_done, chunks) + await asyncio.to_thread(_prepare_part, resume, entry, start) + out = open(long_path(resume.part), "r+b") + out.seek(start * DOWNLOAD_CHUNK) + else: + start = 0 + out = sys.stdout.buffer + if progress and start: + progress(min(size, start * DOWNLOAD_CHUNK)) + + lease = Lease(session, "download") + reason = "error" + try: + await lease.open(size, chunks, on_queued) + written = await _fetch_chunks(session, lease, entry["id"], file_hash, start, chunks, + out, progress, streaming=target is None) + if target is not None: + await asyncio.to_thread(out.flush) + await asyncio.to_thread(os.fsync, out.fileno()) + total = start * DOWNLOAD_CHUNK + written + if total != size: + raise TransferError(f"The node sent {total} bytes of a {size}-byte file.") + reason = "done" + except asyncio.CancelledError: + reason = "cancelled" + raise + finally: + if target is not None: + out.close() + await lease.close(reason) + + if target is not None: + await asyncio.to_thread(_publish, resume, target, overwrite) + return size if target is not None else written + + +def _prepare_part(resume: Resume, entry: dict, start: int) -> None: + part = long_path(resume.part) + if start == 0: + with open(part, "wb"): + pass + else: + os.truncate(part, start * DOWNLOAD_CHUNK) + resume.sidecar.write_text(json.dumps({"id": entry["id"], "size": entry["size"]}), + encoding="utf-8") + + +def _publish(resume: Resume, target: Path, overwrite: bool) -> None: + part, dest = long_path(resume.part), long_path(target) + if overwrite: + os.replace(part, dest) + else: + # A file that appeared meanwhile is not replaced: a hard link refuses + # an existing target, where a rename would silently take its place. + try: + os.link(part, dest) + os.unlink(part) + except FileExistsError: + raise TransferError(f"{target} appeared while downloading; the download " + f"is kept as {resume.part.name}.") from None + except OSError: + if os.path.exists(dest): + raise TransferError(f"{target} already exists.") from None + os.replace(part, dest) + with contextlib.suppress(OSError): + resume.sidecar.unlink() + if sys.platform == "win32": + mark_of_the_web(target) + + +async def _fetch_chunks(session: GroupSession, lease: Lease, file_id: str, file_hash: bytes, + start: int, chunks: int, out, progress: Progress | None, + *, streaming: bool) -> int: + """Chunk requests `DOWNLOAD_WINDOW` ahead, written in order as they land.""" + ch, gek = session.ch, session.gek + inflight: dict[int, asyncio.Task] = {} + nxt, written = start, 0 + + def ask(i: int) -> asyncio.Task: + return asyncio.ensure_future(ch.request( + {"type": "file_req", "v": MNP_VERSION, "file_id": file_id, "chunk_index": i, + "tr": lease.tr}, "file_chunk", timeout=CHUNK_TIMEOUT)) + + try: + for i in range(start, chunks): + while nxt < chunks and nxt - i < DOWNLOAD_WINDOW: + inflight[nxt] = ask(nxt) + nxt += 1 + try: + msg = await inflight.pop(i) + except NodeRefusal as e: + raise TransferError(f"chunk {i}: {e.detail}", code=e.refusal_code) from e + except CliError as e: + raise TransferError(f"chunk {i}: {e.message}") from e + lease.check() + data = decrypt_chunk_aes(chunk_key_aes(gek, file_hash, i), msg["nonce"], msg["ct"]) + if streaming: + out.write(data) + out.flush() + else: + await asyncio.to_thread(out.write, data) + written += len(data) + if progress: + progress(len(data)) + finally: + for task in inflight.values(): + task.cancel() + for task in inflight.values(): + with contextlib.suppress(BaseException): + await task + return written + + +def mark_of_the_web(path: Path) -> None: + """ZoneId 3, as a browser marks a download, so SmartScreen and Protected View + apply when it is opened. FAT and exFAT have no alternate streams: nothing to do.""" + with contextlib.suppress(OSError): + with open(long_path(path) + ":Zone.Identifier", "w", encoding="ascii") as f: + f.write("[ZoneTransfer]\r\nZoneId=3\r\n") + + +# ── Upload ─────────────────────────────────────────────────────────────────── + +async def create_dir(session: GroupSession, parent: str, name: str) -> str: + """A folder `name` inside `parent` (root/folders). Returns its path.""" + try: + ack = await session.ch.request({"type": "dir_create", "v": MNP_VERSION, + "dir": parent, "name": name}, "dir_create_ack") + except NodeRefusal as e: + raise TransferError(f"{parent}/{name}: {e.detail}", code=e.refusal_code, + fatal=e.refusal_code in FATAL_UPLOAD_CODES) from e + return str(ack.get("dir") or f"{parent}/{name}") + + +# Refusals that will not get better for the next file: the run stops. +FATAL_UPLOAD_CODES = frozenset({"disk_full", "root_unavailable", "root_read_only", + "no_writable_root", "no_roots", "no_group_key"}) + + +async def upload(session: GroupSession, read: Callable[[int, int], bytes], size: int, + filename: str, folder: str, *, progress: Progress | None = None, + on_queued: Callable[[int], None] | None = None) -> str: + """ + Send `size` bytes, read by `read(offset, length)`, as `filename` in `folder`. + Returns the name the node stored it under — it never replaces a file, and + takes a free name when this one is taken by the time the upload ends. + """ + gek, gid, ch = session.gek, session.group_id, session.ch + chunks = max(1, math.ceil(size / UPLOAD_CHUNK)) + upload_id = secrets.token_hex(16) + + def chunk(i: int, data: bytes) -> dict: + return {**file_upload_wire(gek, gid, upload_id=upload_id, chunk_index=i, + total_chunks=chunks, filename=filename, data=data, + dir=folder), "tr": lease.tr} + + def refused(e: NodeRefusal) -> TransferError: + return TransferError(f"{filename}: {e.detail}", code=e.refusal_code, + fatal=e.refusal_code in FATAL_UPLOAD_CODES) + + lease = Lease(session, "upload") + reason = "error" + try: + await lease.open(size, chunks, on_queued) + # Where am I? Only the node knows how much of this file it already holds. + try: + probe = await ch.request(chunk(UPLOAD_PROBE_INDEX, b""), "file_upload_ack", + timeout=10) + start = int(_open_ack(gek, gid, probe).get("resume_from") or 0) + except NodeRefusal as e: + raise refused(e) from e + except CliError: + start = 0 + if not 0 < start < chunks: + start = 0 + if progress and start: + progress(min(size, start * UPLOAD_CHUNK)) + + pending: list[tuple[int, asyncio.Task]] = [] + last: dict = {} + try: + for i in range(start, chunks): + while len(pending) >= UPLOAD_WINDOW: + last = await _ack(pending.pop(0), refused, progress, size) + lease.check() + data = await asyncio.to_thread(read, i * UPLOAD_CHUNK, UPLOAD_CHUNK) + pending.append((i, asyncio.ensure_future( + ch.request(chunk(i, data), "file_upload_ack", timeout=CHUNK_TIMEOUT)))) + while pending: + last = await _ack(pending.pop(0), refused, progress, size) + finally: + for _, task in pending: + task.cancel() + for _, task in pending: + with contextlib.suppress(BaseException): + await task + reason = "done" + except asyncio.CancelledError: + reason = "cancelled" + raise + finally: + await lease.close(reason) + return str(_open_ack(gek, gid, last).get("stored_as") or filename) + + +async def _ack(item: tuple[int, asyncio.Task], refused, progress: Progress | None, + size: int) -> dict: + i, task = item + try: + msg = await task + except NodeRefusal as e: + raise refused(e) from e + except CliError as e: + raise TransferError(f"chunk {i}: {e.message}") from e + if progress: + progress(min(UPLOAD_CHUNK, size - i * UPLOAD_CHUNK)) + return msg + + +def _open_ack(gek: bytes, group_id: str, msg: dict) -> dict: + try: + return unseal(gek, PURPOSE_UPLOAD, "file_upload_ack", group_id, msg) + except Exception as e: + raise TransferError("The node's acknowledgement does not open under the group " + "key.", fatal=True) from e diff --git a/packages/meshbay-cli/src/meshbay_cli/client/tree.py b/packages/meshbay-cli/src/meshbay_cli/client/tree.py new file mode 100644 index 0000000..6e537ef --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/client/tree.py @@ -0,0 +1,119 @@ +""" +A group's index as the Files tab draws it: roots, then folders, then files. + +Index entries carry `path` = the root's name and the folders, and `name`; the +index also lists every directory (`dirs`, roots included), because an empty +folder is not an entry and would otherwise not exist; and each root's +`available` and `writable` (docs/MESHBAY_DESIGN.md §6.2, §9.5). +""" + +from dataclasses import dataclass, field +from fnmatch import fnmatchcase + +from ..errors import NOT_FOUND, CliError + +GLOB_CHARS = frozenset("*?[") + + +def is_pattern(segment: str) -> bool: + return any(c in GLOB_CHARS for c in segment) + + +@dataclass +class Item: + name: str + path: str # inside the group: root/folders/name + is_dir: bool + entry: dict | None = None # the index entry, for a file + root: dict | None = None # the root's state, for a root + available: bool = True + + @property + def size(self) -> int: + return int(self.entry.get("size", 0)) if self.entry else 0 + + @property + def added_at(self) -> int: + return int(self.entry.get("added_at", 0)) if self.entry else 0 + + +@dataclass +class Tree: + roots: dict[str, dict] + dirs: set[str] + files: dict[str, list[dict]] = field(default_factory=dict) # parent path → entries + + @classmethod + def from_index(cls, index: dict) -> "Tree": + roots = {r["name"]: r for r in index.get("roots", [])} + dirs = set(index.get("dirs", [])) + files: dict[str, list[dict]] = {} + for e in index.get("entries", []): + files.setdefault(e["path"], []).append(e) + # A file's folders exist even if the node's directory walk missed one. + parts = e["path"].split("/") + for i in range(1, len(parts) + 1): + dirs.add("/".join(parts[:i])) + dirs |= set(roots) + return cls(roots, dirs, files) + + def root_available(self, path: str) -> bool: + root = self.roots.get(path.split("/", 1)[0]) + return bool(root.get("available", True)) if root else True + + def is_dir(self, path: str) -> bool: + return path in self.dirs + + def file(self, path: str) -> dict | None: + parent, _, name = path.rpartition("/") + for e in self.files.get(parent, []): + if e["name"] == name: + return e + return None + + def children(self, path: str, *, include_unavailable: bool = False) -> list[Item]: + """What a folder holds — at the top, the roots. Unavailable roots' content + is left out unless asked for, as the Files tab does.""" + if not path: + return [Item(name, name, True, root=r, available=bool(r.get("available", True))) + for name, r in self.roots.items()] + if not self.is_dir(path): + raise CliError(NOT_FOUND, f"No folder {path!r} in this group.") + available = self.root_available(path) + if not available and not include_unavailable: + return [] + prefix = path + "/" + items = [Item(d[len(prefix):], d, True, available=available) + for d in self.dirs + if d.startswith(prefix) and "/" not in d[len(prefix):]] + items += [Item(e["name"], f"{path}/{e['name']}", False, entry=e, available=available) + for e in self.files.get(path, [])] + return items + + def walk(self, path: str, *, include_unavailable: bool = False): + """(folder, children) for `path` and every folder below it, depth first.""" + kids = self.children(path, include_unavailable=include_unavailable) + yield path, kids + for item in sorted((k for k in kids if k.is_dir), key=lambda k: k.name.lower()): + yield from self.walk(item.path, include_unavailable=include_unavailable) + + def expand(self, parts: list[str]) -> list[str]: + """ + The paths `parts` names, wildcards (`*`, `?`, `[...]`, one segment each) + matched against the index rather than by a shell, so a pattern means the + same in bash and in PowerShell. Matching is case-sensitive, as paths are. + """ + paths = [""] + for seg in parts: + found = [] + for base in paths: + if base and not self.is_dir(base): + continue + if is_pattern(seg): + found += [i.path for i in self.children(base) if fnmatchcase(i.name, seg)] + else: + p = f"{base}/{seg}" if base else seg + if self.is_dir(p) or self.file(p) is not None: + found.append(p) + paths = found + return sorted(set(paths)) diff --git a/packages/meshbay-cli/src/meshbay_cli/client/upload_plan.py b/packages/meshbay-cli/src/meshbay_cli/client/upload_plan.py new file mode 100644 index 0000000..070e686 --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/client/upload_plan.py @@ -0,0 +1,92 @@ +""" +What an upload will do, decided before anything is sent — shared by `put` and +the interactive mode, so both refuse the same names and the same collisions. + +A name already in the destination folder, compared without case as the node and +the Files tab compare it, or a name the node does not accept (`names.py`), is +found here; what to do about it (refuse the run, skip it, ask) is the caller's. +""" + +import os +import stat +from dataclasses import dataclass +from pathlib import Path + +from meshbay_common.paths import fold + +from ..errors import USAGE, CliError +from ..output import safe +from .names import same_name, upload_name_problem +from .tree import Tree + + +@dataclass +class UploadFile: + local: Path + rel: list[str] # remote parts below the destination folder + size: int + + +def plan_upload(sources: list[str], recursive: bool): + """Folders to create (parents first) and files to send, as remote parts.""" + dirs: list[list[str]] = [] + files: list[UploadFile] = [] + problems: list[str] = [] + seen: dict[str, str] = {} + for src in sources: + path = Path(src).expanduser() + try: + st = path.stat() + except OSError as e: + problems.append(f"{src}: {e.strerror}") + continue + top = path.resolve().name if src.rstrip("/\\") in (".", "..") else path.name + key = fold(top) + if key in seen: + problems.append(f"{src} and {seen[key]} would have the same name there") + continue + seen[key] = src + if stat.S_ISREG(st.st_mode): + files.append(UploadFile(path, [top], st.st_size)) + elif stat.S_ISDIR(st.st_mode): + if not recursive: + raise CliError(USAGE, f"{safe(src)} is a folder; use -R.") + for here, subdirs, names in os.walk(path): + rel = [top] + list(Path(here).relative_to(path).parts) + dirs.append(rel) + for d in list(subdirs): + if (Path(here) / d).is_symlink(): + subdirs.remove(d) + problems.append(f"{Path(here) / d} is a link to a folder; not " + "followed") + for n in names: + p = Path(here) / n + try: + st_n = p.stat() + except OSError as e: + problems.append(f"{p}: {e.strerror}") + continue + if stat.S_ISREG(st_n.st_mode): + files.append(UploadFile(p, rel + [n], st_n.st_size)) + subdirs.sort() + else: + problems.append(f"{src} is neither a file nor a folder") + return dirs, files, problems + + +def taken_names(tree: Tree, folder: str, names: list[str]) -> set[str]: + """The names among `names` already in `folder`, compared without case.""" + there = [i.name for i in tree.children(folder, include_unavailable=True)] + return {n for n in names if any(same_name(n, t) for t in there)} + + +def refused_names(file_names: list[str], dirs: list[list[str]]) -> list[str]: + """Names the node would refuse — the client's copy of its rule.""" + out = [] + for name in file_names: + if (why := upload_name_problem(name)): + out.append(f"{name}: {why}") + for d in dirs: + if (why := upload_name_problem(d[-1], is_dir=True)): + out.append(f"{'/'.join(d)}/: {why}") + return out 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 diff --git a/packages/meshbay-cli/src/meshbay_cli/errors.py b/packages/meshbay-cli/src/meshbay_cli/errors.py new file mode 100644 index 0000000..319645e --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/errors.py @@ -0,0 +1,24 @@ +""" +Every way a run can end badly, as an exit code a script can act on. + +The library raises `CliError` with one of these codes; only `__main__` turns it +into a message and an exit status. +""" + +USAGE = 1 +AUTH = 2 # sign-in refused, no session and no passphrase, locked out +NOT_FOUND = 3 # group, path, or a pattern that matches nothing +PARTIAL = 4 # some items failed +NODE = 5 # node unreachable, or it refused the connection +NODE_KEY = 6 # a node's key changed since it was pinned +VERSION = 7 # this client, or the protocol, is too old or too new +REFUSED = 8 # refused before anything was sent: a name taken or not accepted, + # a folder read-only or unavailable +INTERRUPTED = 130 + + +class CliError(Exception): + def __init__(self, code: int, message: str): + super().__init__(message) + self.code = code + self.message = message diff --git a/packages/meshbay-cli/src/meshbay_cli/manpage.py b/packages/meshbay-cli/src/meshbay_cli/manpage.py new file mode 100644 index 0000000..0633904 --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/manpage.py @@ -0,0 +1,344 @@ +""" +The man page, generated from the parser: `python -m meshbay_cli.manpage`. + +Every action and option comes from `build_parser()`, so the page cannot list an +option the program lacks or miss one it has; the prose the parser cannot carry +(remote paths, the store, exit codes, environment) is written here. The +packages build it at build time and install it in /usr/share/man/man1. +""" + +import argparse +import sys + +from . import __version__, errors +from .__main__ import ACTIONS, build_parser + +EXIT_CODES = { + 0: "Success.", + errors.USAGE: "Usage error: an unknown option, a missing argument.", + errors.AUTH: "Sign-in refused, no stored session and no passphrase to sign in " + "with, or the account is locked out.", + errors.NOT_FOUND: "A group, a path, or a pattern that matches nothing.", + errors.PARTIAL: "Some items failed; the others were done. Each failure is " + "named on stderr.", + errors.NODE: "No node serving the group could be reached, or the node refused " + "the connection. Also: this machine is waiting for device " + "approval (the code is on stderr).", + errors.NODE_KEY: "A node presented another key than the one pinned at first " + "contact. Nothing was sent. See \\fBtrust\\fR.", + errors.VERSION: "This client, or the protocol it speaks, is too old or too new " + "for the hub or the node.", + errors.REFUSED: "Refused before anything was sent: a name already taken or not " + "accepted by the node, a folder read-only or unavailable.", + errors.INTERRUPTED: "Interrupted (Ctrl+C). A download resumes where it stopped " + "on the next run.", +} + +DESCRIPTION = r""" +.B meshbay-cli +is a member's view of their MeshBay groups from a terminal: sign in, list the +groups and their folders, download and upload files. +It talks to the hub only to sign in and to find a group's nodes; files go +directly between this machine and the node, end-to-end encrypted, never +through the hub. +.PP +Run with no argument at all, it opens the interactive mode (below). +Global options are accepted before the action and after it. +.PP +Administering a node is not this program's business: see +.BR meshbay-node (1). +""" + +PATHS = r""" +A remote path names a group, then a root the node shares, then folders and a +file: +.PP +.RS +.EX +group[@owner]/root/folder/file +.EE +.RE +.PP +.B @owner +is needed only when two of your groups share a name; group names compare +without regard to case. +A group name may itself contain +.B / +or +.BR @ : +type them as +.B %2F +and +.B %40 +(and a literal +.B % +as +.BR %25 ). +The GROUP column of +.B meshbay-cli groups +prints each group in this form. +A link copied from the application's address bar (containing +.BR #/ ) +is accepted as it is. +.PP +.B get +matches the wildcards +.BR * , +.B ? +and +.B [...] +against the group's index, not against local files: quote them. +""" + +INTERACTIVE = r""" +.B meshbay-cli +with no argument opens two panes: this machine's folders on the left, your +groups on the right, and the transfers below. +A sign-in screen comes first when no session is stored. +.TP +.B Tab +switch pane +.TP +.B Enter +open a folder or a group +.TP +.B Backspace +go up +.TP +.B Space +select an entry (several can be) +.TP +.B F5 +copy the selection, or the entry under the cursor, to the other pane's folder; +a name already there is asked about once for the whole batch +.TP +.B / +filter by name; +.B Esc +clears the filter +.TP +.B s +cycle the sort: name, size, date +.TP +.B r +refresh +.TP +.B q +quit +.PP +The right pane follows the node: a file added by someone else appears there. +""" + +SESSION = r""" +.B login +signs in with the account's passphrase and, unless +.B \-\-no\-remember +is given, keeps the session so that later runs need no passphrase. +The passphrase is read from, in order: +.BR \-\-password\-file , +.BR \-\-password , +.BR MESHBAY_PASSWORD , +then a prompt when there is a terminal. +.B \-\-password +puts it in the process list and the shell history; prefer +.B \-\-password\-file \- +(stdin). +.PP +The session and this machine's keys are kept encrypted in the data directory +(see FILES), under a key held by the system's key storage (Secret Service, +macOS Keychain, Windows Credential Manager). +Where there is none, such as a server with no desktop session, the key is kept +in a file only your user can read, like an SSH key without a passphrase; +.B login +says so once. +.B logout +ends the session on the hub, retires this machine's device key and erases +everything stored. +.PP +The first time this machine opens a group whose node holds no copy of your +keys (the application's "Browser access" turned off), it asks to be linked as +a new device and prints a code. +Approve that code from the application on a device already linked (the +group's Settings); with a terminal, +.B meshbay-cli +waits and goes on by itself, otherwise it exits with status 5 and the next run +goes on once the code is approved. +.PP +Each node's key is pinned at first contact. +A node presenting another key is refused, with status 6, until +.B meshbay-cli trust reset +.I NODE +forgets the pin; do that only for a key change the node's operator announced. +""" + +FILES = r""" +.TP +.I $XDG_DATA_HOME/meshbay\-cli/ \fR(default \fI~/.local/share/meshbay\-cli/\fR) +On Windows, +.IR %LOCALAPPDATA%\emeshbay\-cli\e . +Holds +.B state.json +(the hub, the account name, pinned node keys), +.B secrets.enc +(the session and keys, encrypted), +.B store.key +only where the system has no key storage, and +.B interactive.log +(the interactive mode's log). +.TP +\fIname\fB.meshbay\-part\fR, \fIname\fB.meshbay\-part.json\fR +A download in progress, next to where the file lands. +A later +.B get +of the same file resumes from it; the finished file replaces it. +""" + +ENVIRONMENT = r""" +.TP +.B MESHBAY_HUB +The hub, when +.B \-\-hub +is not given (default: the one signed in to, else https://meshbay.org). +.TP +.B MESHBAY_USER +The account name, when +.B \-u +is not given. +.TP +.B MESHBAY_PASSWORD +The passphrase, when neither +.B \-\-password\-file +nor +.B \-\-password +is given. +.TP +.B MESHBAY_CLI_HOME +The data directory, instead of the default (see FILES). +.TP +.B NO_COLOR +The interactive mode draws without colour. +""" + +EXAMPLES = r""" +.EX +meshbay\-cli login \-u alice +meshbay\-cli groups +meshbay\-cli ls \-l Family/photos/2024 +meshbay\-cli get \-R Family/photos/2024 \-o ~/Pictures/ +meshbay\-cli get 'Family/music/*.flac' +meshbay\-cli put \-R ~/scans Family/documents +tar c notes | meshbay\-cli put \- Family/documents/notes.tar +meshbay\-cli ls \-\-json \-R / | jq '.[] | .path' +.EE +""" + + +def _roff(text: str) -> str: + """ + Plain text into roff: no control line, no escape, no hyphen taken for a + dash, and anything past ASCII named, which troff reads whatever the locale. + """ + text = text.replace("\\", "\\e").replace("-", "\\-") + text = "".join(c if ord(c) < 128 else f"\\[u{ord(c):04X}]" for c in text) + return "\n".join("\\&" + line if line[:1] in (".", "'") else line + for line in text.splitlines()) + + +def _option_lines(action: argparse.Action) -> str: + if action.option_strings: + names = ", ".join(f"\\fB{_roff(s)}\\fR" for s in action.option_strings) + if action.choices: + names += " " + "|".join(f"\\fB{_roff(c)}\\fR" for c in action.choices) + elif action.nargs != 0: + names += f" \\fI{_roff(action.metavar or action.dest.upper())}\\fR" + else: + names = f"\\fI{_roff(action.metavar or action.dest.upper())}\\fR" + if action.choices: + names = " | ".join(f"\\fB{_roff(c)}\\fR" for c in action.choices) + help_text = action.help or "" + return f".TP\n{names}\n{_roff(help_text)}\n" + + +def _own_options(parser: argparse.ArgumentParser) -> list[argparse.Action]: + """The action's own arguments; the global ones are listed once, above.""" + late = {s for a in _globals(build_parser()) for s in a.option_strings} + return [a for a in parser._actions + if not isinstance(a, argparse._HelpAction) + and not (set(a.option_strings) & late)] + + +def _globals(parser: argparse.ArgumentParser) -> list[argparse.Action]: + return [a for a in parser._actions + if a.option_strings and not isinstance(a, argparse._SubParsersAction)] + + +def _subparsers(parser: argparse.ArgumentParser) -> dict[str, argparse.ArgumentParser]: + sub = next(a for a in parser._actions if isinstance(a, argparse._SubParsersAction)) + return dict(sub.choices) + + +def _synopsis(name: str, parser: argparse.ArgumentParser) -> str: + """`meshbay-cli ls [-l] ... [path]`, without the global options.""" + parts = [] + own = _own_options(parser) + for a in sorted(own, key=lambda a: not a.option_strings): # options first + if a.option_strings: + s = a.option_strings[0] + if a.choices: + s += " " + "|".join(a.choices) + elif a.nargs != 0: + s += f" {a.metavar or a.dest.upper()}" + parts.append(f"[{s}]") + elif a.choices: + parts.append(" | ".join(a.choices)) + else: + meta = a.metavar or a.dest.upper() + parts.append({"?": f"[{meta}]", "+": f"{meta}...", "*": f"[{meta}...]"} + .get(a.nargs, meta)) + return _roff(" ".join([name] + parts)) + + +def render() -> str: + parser = build_parser() + out = [ + f'.TH MESHBAY\\-CLI 1 "" "meshbay\\-cli {__version__}" "User Commands"', + ".nh", ".ad l", + ".SH NAME", + "meshbay\\-cli \\- a member's MeshBay groups and files, from a terminal", + ".SH SYNOPSIS", + ".B meshbay\\-cli", + ".br", + ".B meshbay\\-cli", + "[\\fIglobal options\\fR] \\fIaction\\fR [\\fIaction options\\fR] " + "[\\fIarguments\\fR]", + ".SH DESCRIPTION", + DESCRIPTION.strip(), + ".SH GLOBAL OPTIONS", + ] + out += [_option_lines(a).rstrip() for a in _globals(parser)] + out.append(".SH ACTIONS") + subs = _subparsers(parser) + for name in ACTIONS: + sp = subs[name] + out += [f".SS {_roff(name)}", f".B {_synopsis(name, sp)}", ".PP", + _roff(sp.description or "")] + out += [_option_lines(a).rstrip() for a in _own_options(sp)] + out += [".SH REMOTE PATHS", PATHS.strip(), + ".SH INTERACTIVE MODE", INTERACTIVE.strip(), + ".SH SESSION, DEVICES AND NODE KEYS", SESSION.strip(), + ".SH EXIT STATUS"] + for code, text in sorted(EXIT_CODES.items()): + out += [".TP", f".B {code}", text] + out += [".SH ENVIRONMENT", ENVIRONMENT.strip(), + ".SH FILES", FILES.strip(), + ".SH EXAMPLES", EXAMPLES.strip(), + ".SH SEE ALSO", ".BR meshbay\\-node (1)", ""] + return "\n".join(out) + + +def main() -> None: + sys.stdout.write(render()) + + +if __name__ == "__main__": + main() diff --git a/packages/meshbay-cli/src/meshbay_cli/output.py b/packages/meshbay-cli/src/meshbay_cli/output.py new file mode 100644 index 0000000..75a54ff --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/output.py @@ -0,0 +1,140 @@ +""" +Writing to the terminal. + +**Names are someone else's text.** A file or group name is chosen by whoever put +it in the group, and a raw ESC in it is an escape sequence the reader's +terminal executes; a bidirectional override makes it display as another name. +Every name printed goes through `safe()`. +""" + +import json +import shutil +import sys +import time +import unicodedata + +_BIDI = frozenset("") + + +def safe(text: str) -> str: + out = [] + for c in text: + if unicodedata.category(c) == "Cc" or c in _BIDI: + out.append(f"\\x{ord(c):02x}" if ord(c) < 0x100 else f"\\u{ord(c):04x}") + else: + out.append(c) + return "".join(out) + + +def human_size(n: int) -> str: + size = float(n) + for unit in ("B", "KB", "MB", "GB", "TB"): + if size < 1024 or unit == "TB": + return f"{size:.0f} {unit}" if unit == "B" else f"{size:.1f} {unit}" + size /= 1024 + return f"{n} B" + + +def date(ts: int) -> str: + return time.strftime("%Y-%m-%d %H:%M", time.localtime(ts)) if ts else "" + + +def columns(names: list[str]) -> str: + """`ls`'s short form on a terminal: as many columns as fit, filled down.""" + if not names: + return "" + width = shutil.get_terminal_size((80, 24)).columns + col = max(len(n) for n in names) + 2 + ncols = max(1, width // col) + nrows = -(-len(names) // ncols) + lines = [] + for r in range(nrows): + row = [names[c * nrows + r] for c in range(ncols) if c * nrows + r < len(names)] + lines.append("".join(n.ljust(col) for n in row).rstrip()) + return "\n".join(lines) + + +def is_tty() -> bool: + return sys.stdout.isatty() + + +def err(message: str) -> None: + print(message, file=sys.stderr) + + +class Progress: + """ + One status line on stderr for a run of transfers, redrawn at most ten times + a second; JSON Lines with --json; nothing with -q or when stderr is not a + terminal. + """ + + def __init__(self, arrow: str, *, quiet: bool = False, as_json: bool = False): + self.arrow = arrow + self.live = sys.stderr.isatty() and not quiet and not as_json + self.as_json = as_json and not quiet + self.files_total = 0 + self.files_done = 0 + self.bytes_total = 0 + self.bytes_done = 0 + self.current = "" + self._start = time.monotonic() + self._drawn = 0.0 + + def plan(self, files: int, nbytes: int) -> None: + self.files_total, self.bytes_total = files, nbytes + + def event(self, **fields) -> None: + if self.as_json: + print(json.dumps(fields, ensure_ascii=False), file=sys.stderr, flush=True) + + def start(self, name: str) -> None: + self.current = name + self.event(event="start", name=name) + self._draw(force=True) + + def advance(self, nbytes: int) -> None: + self.bytes_done += nbytes + self._draw() + + def queued(self, name: str, ahead: int) -> None: + self.current = f"{name} (waiting for a free slot on the node, {ahead} ahead)" + self.event(event="queued", name=name, ahead=ahead) + self._draw(force=True) + + def finish(self, name: str, **fields) -> None: + self.files_done += 1 + self.event(event="done", name=name, **fields) + self._draw(force=True) + + def note(self, message: str) -> None: + """A line of its own, above the status line.""" + self.clear() + if not self.as_json: + print(message, file=sys.stderr) + self._draw(force=True) + + def _draw(self, force: bool = False) -> None: + if not self.live: + return + now = time.monotonic() + if not force and now - self._drawn < 0.1: + return + self._drawn = now + rate = self.bytes_done / max(now - self._start, 0.001) + line = (f"{self.arrow} {self.files_done}/{self.files_total} " + f"{human_size(self.bytes_done)} / {human_size(self.bytes_total)} " + f"{human_size(int(rate))}/s {self.current}") + width = shutil.get_terminal_size((80, 24)).columns + sys.stderr.write("\r\x1b[K" + line[:width - 1]) + sys.stderr.flush() + + def clear(self) -> None: + if self.live: + sys.stderr.write("\r\x1b[K") + sys.stderr.flush() + + +def sentence(text: str) -> str: + """The first letter up, the rest untouched: `str.capitalize` turns -R into -r.""" + return text[:1].upper() + text[1:] diff --git a/packages/meshbay-cli/src/meshbay_cli/selftest.py b/packages/meshbay-cli/src/meshbay_cli/selftest.py new file mode 100644 index 0000000..81453c4 --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/selftest.py @@ -0,0 +1,61 @@ +""" +What a frozen build must be able to load, checked with no hub and no node. + +A PyInstaller build carries only the modules its analysis found, and a module +it missed fails at run time only — in the interactive mode, or the first time +the store asks for the system's key storage. The Windows build runs this +(`MESHBAY_CLI_SELFTEST=1 meshbay-cli`, see packaging/win/build-node-runtime.ps1) +on the frozen executable; nothing here is reachable through the command line. +""" + +import asyncio +import os +import sys +import tempfile + + +async def _interactive() -> None: + """The sign-in screen, drawn off-screen: Textual's widgets, CSS and Rich.""" + from .__main__ import build_parser + from .commands.context import Context + from .tui.app import MeshBayApp + from .tui.screens import SignInScreen + + app = MeshBayApp(Context(build_parser().parse_args(["--hub", "http://127.0.0.1:1"]))) + async with app.run_test(size=(100, 30)) as pilot: + for _ in range(100): + if isinstance(app.screen, SignInScreen): + break + await pilot.pause(0.05) + else: + raise RuntimeError("the sign-in screen never showed") + app.screen.query_one("#user") + + +async def _webrtc() -> None: + from aiortc import RTCPeerConnection + pc = RTCPeerConnection() + pc.createDataChannel("selftest") + await pc.setLocalDescription(await pc.createOffer()) + await pc.close() + + +def run() -> int: + with tempfile.TemporaryDirectory() as home: + os.environ["MESHBAY_CLI_HOME"] = home # nothing stored: the sign-in screen + # keyring finds its backends through its own entry points: without its + # metadata in the build, Windows Credential Manager is simply not + # there and the store falls back to a key file on every machine. + import keyring.backend + backends = [type(b).__module__ for b in keyring.backend.get_all_keyring()] + if sys.platform == "win32" and "keyring.backends.Windows" not in backends: + raise RuntimeError(f"no Windows Credential Manager backend in {backends}") + + from argon2.low_level import Type, hash_secret_raw + hash_secret_raw(b"x", b"saltsalt", 1, 8, 1, 16, Type.ID) + + from .__main__ import loop_factory + asyncio.run(_webrtc(), loop_factory=loop_factory()) + asyncio.run(_interactive(), loop_factory=loop_factory()) + print(f"self-test ok (keyring: {', '.join(backends)})") + return 0 diff --git a/packages/meshbay-cli/src/meshbay_cli/tui/__init__.py b/packages/meshbay-cli/src/meshbay_cli/tui/__init__.py new file mode 100644 index 0000000..becd195 --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/tui/__init__.py @@ -0,0 +1 @@ +"""The interactive mode: `meshbay-cli` with no argument.""" diff --git a/packages/meshbay-cli/src/meshbay_cli/tui/app.py b/packages/meshbay-cli/src/meshbay_cli/tui/app.py new file mode 100644 index 0000000..dbb773c --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/tui/app.py @@ -0,0 +1,76 @@ +""" +`meshbay-cli` with no argument: a full-screen terminal interface. + +Textual, because it runs on the same asyncio loop as aiortc and httpx rather +than beside them, and on Linux terminals and Windows' consoles alike. Signed in +already, it opens on the two panes; otherwise on the sign-in screen first. + +Nothing is printed while it runs — a line on stderr would tear the screen — so +the log goes to a file in the CLI's data directory instead. +""" + +import logging + +from textual.app import App + +from ..client.account import Account, no_passphrase +from ..client.hub import Hub +from ..client.store import data_dir +from ..commands.context import Context +from ..errors import CliError +from .screens import DeviceCode, MainScreen, SignInScreen + + +class MeshBayApp(App): + TITLE = "MeshBay" + + def __init__(self, ctx: Context): + super().__init__() + self.ctx = ctx + self.hub = ctx.hub + + async def use_hub(self, url: str) -> None: + if url != self.hub.url: + await self.hub.aclose() + self.hub = Hub(url) + self.ctx.hub = self.hub + self.ctx.hub_url = url + + def on_mount(self) -> None: + account = None + if self.ctx.store.peek().get("hub") == self.hub.url: + try: + account = Account.open(self.hub, None, no_passphrase, self.ctx.store) + except CliError as e: + self.notify(e.message, severity="warning", markup=False) + if account is not None and account.remembered: + self.start(account) + else: + self.push_screen(SignInScreen()) + + def start(self, account: Account) -> None: + account.on_device_code = DeviceCode(self) + if isinstance(self.screen, SignInScreen): + self.switch_screen(MainScreen(account)) + else: + self.push_screen(MainScreen(account)) + + +def _log_to_file() -> None: + root = logging.getLogger() + for h in list(root.handlers): + root.removeHandler(h) + try: + directory = data_dir() + directory.mkdir(parents=True, exist_ok=True) + handler = logging.FileHandler(directory / "interactive.log", encoding="utf-8") + except OSError: + handler = logging.NullHandler() + handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s")) + root.addHandler(handler) + + +async def run_interactive(ctx: Context) -> int: + _log_to_file() + await MeshBayApp(ctx).run_async() + return 0 diff --git a/packages/meshbay-cli/src/meshbay_cli/tui/panes.py b/packages/meshbay-cli/src/meshbay_cli/tui/panes.py new file mode 100644 index 0000000..d1e3dd0 --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/tui/panes.py @@ -0,0 +1,372 @@ +""" +The two panes: this machine's folders on the left, the groups on the right. + +Both draw rows the same way — folders first, then files, each sorted by the +pane's key — and both mark a selection with Space. Names are someone else's +text on the right and may be on the left: every name is escaped (`safe`) and +handed to the table as plain text, never as markup. +""" + +import asyncio +import os +import stat +from dataclasses import dataclass, field +from pathlib import Path + +from rich.text import Text +from textual.app import ComposeResult +from textual.binding import Binding +from textual.containers import Vertical +from textual.widgets import DataTable, Input, Static + +from ..client.groups import my_groups +from ..client.paths import handle +from ..client.tree import Tree +from ..errors import CliError +from ..output import date, human_size, safe + +PARENT = ".." +SORTS = ("name", "size", "date") +SIZE_WIDTH, DATE_WIDTH = 9, 16 +# Cell padding on both sides of three columns, and the vertical scrollbar. +CHROME = 2 * 3 + 2 + + +@dataclass +class Row: + key: str + name: str + is_dir: bool + size: int | None = None + when: float | None = None + note: str = "" # "down", "read-only", ... + dim: bool = False + data: dict = field(default_factory=dict) + + +class Pane(Vertical, can_focus=False): + """A titled table with marks, a sort key and a filter.""" + + BINDINGS = [ + Binding("space", "mark", "Select"), + Binding("backspace", "up", "Up"), + Binding("slash", "filter", "Filter"), + Binding("s", "sort", "Sort"), + Binding("r", "refresh", "Refresh"), + ] + + DEFAULT_CSS = """ + Pane { width: 1fr; border: round $panel; } + Pane:focus-within { border: round $accent; } + Pane > .title { height: 1; padding: 0 1; background: $panel; } + Pane > DataTable { height: 1fr; } + Pane > Input { display: none; } + Pane > Input.shown { display: block; } + """ + + def __init__(self, **kwargs): + super().__init__(**kwargs) + self.rows: list[Row] = [] + self.marks: set[str] = set() + self.sort = "name" + self.filter_text = "" + self.name_width = 30 + self.show_date = True + + def compose(self) -> ComposeResult: + yield Static("", classes="title", markup=False) + table = DataTable(cursor_type="row", zebra_stripes=False) + self._add_columns(table) + yield table + yield Input(placeholder="filter — Enter to apply, Esc to clear") + + def _add_columns(self, table: DataTable) -> None: + table.add_column("Name", key="name", width=self.name_width) + table.add_column(Text("Size", justify="right"), key="size", width=SIZE_WIDTH) + if self.show_date: + table.add_column("Date", key="date", width=DATE_WIDTH) + + def on_resize(self) -> None: + """ + The name takes what the size and the date leave, never more: a long name + with no spaces in it must not push them out of the pane. In a pane too + narrow for all three (two panes in an 80-column terminal), the date goes. + """ + self._draw_title() + inner = self.table.size.width + show_date = inner - SIZE_WIDTH - DATE_WIDTH - CHROME >= 16 + width = max(8, inner - SIZE_WIDTH - (DATE_WIDTH if show_date else 0) + - (CHROME if show_date else CHROME - 2)) + if (width, show_date) != (self.name_width, self.show_date): + self.name_width, self.show_date = width, show_date + self.table.clear(columns=True) + self._add_columns(self.table) + self.redraw() + + @property + def table(self) -> DataTable: + return self.query_one(DataTable) + + def set_title(self, text: str) -> None: + self._title = text + self._draw_title() + + def _draw_title(self) -> None: + """One line; a long path loses its beginning, which is the part known.""" + text = getattr(self, "_title", "") + room = max(8, self.size.width - 4) + if len(text) > room: + text = "…" + text[-(room - 1):] + self.query_one(".title", Static).update(Text(text, no_wrap=True)) + + # ── Drawing ────────────────────────────────────────────────────────────── + + def show(self, rows: list[Row], *, keep_cursor: str | None = None) -> None: + self.rows = rows + self.marks &= {r.key for r in rows} + self.redraw(keep_cursor=keep_cursor) + + def _ordered(self) -> list[Row]: + def k(r: Row): + if self.sort == "size": + return (-(r.size or 0), r.name.lower()) + if self.sort == "date": + return (-(r.when or 0), r.name.lower()) + return (r.name.lower(), r.name) + visible = [r for r in self.rows if r.key != PARENT + and self.filter_text.lower() in r.name.lower()] + parent = [r for r in self.rows if r.key == PARENT] + return (parent + sorted((r for r in visible if r.is_dir), key=k) + + sorted((r for r in visible if not r.is_dir), key=k)) + + def redraw(self, *, keep_cursor: str | None = None) -> None: + table = self.table + current = keep_cursor or self.cursor_key() + table.clear() + for r in self._ordered(): + mark = "* " if r.key in self.marks else " " + name = Text(mark + safe(r.name) + ("/" if r.is_dir and r.key != PARENT else ""), + style="dim" if r.dim else ("bold" if r.key in self.marks else ""), + no_wrap=True, overflow="ellipsis") + if r.note: + name.append(f" ({r.note})", style="italic dim") + size = Text("" if r.is_dir or r.size is None else human_size(r.size), + justify="right") + cells = [name, size] + if self.show_date: + cells.append(Text(date(int(r.when)) if r.when else "")) + table.add_row(*cells, key=r.key) + keys = [r.key for r in self._ordered()] + if current in keys: + table.move_cursor(row=keys.index(current)) + + def cursor_key(self) -> str | None: + table = self.table + if table.row_count == 0: + return None + try: + return table.coordinate_to_cell_key(table.cursor_coordinate).row_key.value + except Exception: + return None + + def row(self, key: str | None) -> Row | None: + return next((r for r in self.rows if r.key == key), None) + + def selection(self) -> list[Row]: + """The marked rows, or the one under the cursor.""" + if self.marks: + return [r for r in self.rows if r.key in self.marks] + r = self.row(self.cursor_key()) + return [r] if r and r.key != PARENT else [] + + # ── Actions ────────────────────────────────────────────────────────────── + + def action_mark(self) -> None: + key = self.cursor_key() + if key is None or key == PARENT: + return + self.marks ^= {key} + row = self.table.cursor_row + self.redraw(keep_cursor=key) + self.table.move_cursor(row=min(row + 1, self.table.row_count - 1)) + + def action_sort(self) -> None: + self.sort = SORTS[(SORTS.index(self.sort) + 1) % len(SORTS)] + self.app.notify(f"Sorted by {self.sort}", timeout=2) + self.redraw() + + def action_filter(self) -> None: + box = self.query_one(Input) + box.add_class("shown") + box.value = self.filter_text + box.focus() + + def on_input_changed(self, event: Input.Changed) -> None: + self.filter_text = event.value + self.redraw() + + def on_input_submitted(self, event: Input.Submitted) -> None: + self.table.focus() + + def key_escape(self) -> None: + box = self.query_one(Input) + if box.has_class("shown"): + box.value = "" + box.remove_class("shown") + self.filter_text = "" + self.redraw() + self.table.focus() + + def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None: + row = self.row(event.row_key.value) + if row is None: + return + if row.key == PARENT: + self.action_up() + elif row.is_dir: + self.enter(row) + + # Subclasses + def enter(self, row: Row) -> None: ... + def action_up(self) -> None: ... + def action_refresh(self) -> None: ... + + +class LocalPane(Pane): + def __init__(self, start: Path, **kwargs): + super().__init__(**kwargs) + self.path = start.resolve() + + def on_mount(self) -> None: + self.load() + + def load(self, keep: str | None = None) -> None: + self.set_title(f"Local: {self.path}") + self.run_worker(self._load(keep), exclusive=True, group="local") + + async def _load(self, keep: str | None) -> None: + try: + rows = await asyncio.to_thread(_scan, self.path) + except OSError as e: + self.app.notify(f"{self.path}: {e.strerror}", severity="error", markup=False) + rows = [] + if self.path.parent != self.path: + rows.insert(0, Row(PARENT, PARENT, True)) + self.show(rows, keep_cursor=keep) + + def enter(self, row: Row) -> None: + self.path = self.path / row.name + self.marks.clear() + self.load() + + def action_up(self) -> None: + if self.path.parent != self.path: + came_from = self.path.name + self.path = self.path.parent + self.marks.clear() + self.load(keep=came_from) + + def action_refresh(self) -> None: + # No position captured now: the cursor may move while the folder is read, + # and the redraw keeps wherever it is when the rows arrive. + self.load() + + +def _scan(path: Path) -> list[Row]: + rows = [] + with os.scandir(path) as it: + for e in it: + try: + st = e.stat() + except OSError: + continue + is_dir = stat.S_ISDIR(st.st_mode) + if not is_dir and not stat.S_ISREG(st.st_mode): + continue + rows.append(Row(e.name, e.name, is_dir, None if is_dir else st.st_size, + st.st_mtime)) + return rows + + +class RemotePane(Pane): + """Groups, then a group's roots and folders — the Files tab, never merged.""" + + def __init__(self, screen, **kwargs): + super().__init__(**kwargs) + self.main = screen + self.groups: list[dict] = [] + self.group: dict | None = None # None: the list of groups + self.path = "" # inside the group: root/folders + + def on_mount(self) -> None: + self.load() + + @property + def location(self) -> str: + if self.group is None: + return "Your groups" + return f"{handle(self.group)}/{self.path}".rstrip("/") + + def load(self, keep: str | None = None) -> None: + self.set_title(f"{safe(self.location)} (loading…)") + self.run_worker(self._load(keep), exclusive=True, group="remote") + + async def _load(self, keep: str | None) -> None: + try: + if self.group is None: + self.groups = await my_groups(self.main.account) + rows = [Row(g["id"], handle(g), True, + note="" if g.get("node_online") else "down", + dim=not g.get("node_online"), data=g) + for g in self.groups] + else: + tree = await self.main.tree(self.group) + rows = [Row(PARENT, PARENT, True)] + _tree_rows(tree, self.path) + except CliError as e: + self.set_title(safe(self.location)) + self.app.notify(e.message, severity="error", markup=False, timeout=8) + if self.group is not None: + self.group, self.path = None, "" + self.load() + return + self.set_title(safe(self.location)) + self.show(rows, keep_cursor=keep) + + def enter(self, row: Row) -> None: + self.marks.clear() + if self.group is None: + self.group, self.path = row.data, "" + else: + self.path = row.data["path"] + self.load() + + def action_up(self) -> None: + if self.group is None: + return + self.marks.clear() + if not self.path: + came_from, self.group = self.group["id"], None + else: + came_from = self.path + self.path = self.path.rpartition("/")[0] + self.load(keep=came_from) + + def action_refresh(self) -> None: + if self.group is not None: + self.main.forget_tree(self.group) + self.load() + + +def _tree_rows(tree: Tree, path: str) -> list[Row]: + rows = [] + for item in tree.children(path): + note = "" + if item.root is not None: + note = "" if item.root.get("writable") else "read-only" + if not item.available: + note = "unavailable" + rows.append(Row(item.path, item.name, item.is_dir, + None if item.is_dir else item.size, + item.added_at or None, note=note, dim=not item.available, + data={"path": item.path, "entry": item.entry})) + return rows 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 diff --git a/packages/meshbay-cli/src/meshbay_cli/tui/widgets.py b/packages/meshbay-cli/src/meshbay_cli/tui/widgets.py new file mode 100644 index 0000000..66fca4e --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/tui/widgets.py @@ -0,0 +1,109 @@ +"""A question with a few answers, and the strip that shows transfers.""" + +import time +from dataclasses import dataclass + +from rich.text import Text +from textual.app import ComposeResult +from textual.containers import Horizontal, Vertical +from textual.screen import ModalScreen +from textual.widgets import Button, Static + +from ..output import human_size, safe + + +class Choice(ModalScreen[str]): + """One question, asked once for a whole batch — never once per file.""" + + DEFAULT_CSS = """ + Choice { align: center middle; } + Choice > Vertical { width: 72; height: auto; max-height: 80%; + border: thick $accent; background: $surface; padding: 1 2; } + Choice Static { height: auto; margin-bottom: 1; } + Choice Horizontal { height: auto; align: center middle; } + Choice Button { margin: 0 1; } + """ + BINDINGS = [("escape", "dismiss('cancel')", "Cancel")] + + def __init__(self, message: str, options: list[tuple[str, str]]): + super().__init__() + self.message = message + self.options = options + + def compose(self) -> ComposeResult: + with Vertical(): + yield Static(Text(self.message)) + with Horizontal(): + for value, label in self.options: + yield Button(label, id=f"choice-{value}") + + def on_mount(self) -> None: + self.query(Button).first().focus() + + def on_button_pressed(self, event: Button.Pressed) -> None: + self.dismiss(event.button.id.removeprefix("choice-")) + + +@dataclass +class Job: + arrow: str + name: str + size: int + done: int = 0 + state: str = "waiting" # waiting, queued, running, done, failed + note: str = "" + started: float = 0.0 + + +class TransferStrip(Static): + """The running transfers, then the latest finished ones.""" + + DEFAULT_CSS = """ + TransferStrip { height: 6; border-top: solid $panel; padding: 0 1; } + """ + SHOWN = 5 + + def __init__(self, **kwargs): + super().__init__("", markup=False, **kwargs) + self.jobs: list[Job] = [] + + def on_mount(self) -> None: + self.refresh_text() + + def add(self, arrow: str, name: str, size: int) -> Job: + job = Job(arrow, name, size) + self.jobs.append(job) + self.refresh_text() + return job + + def update_job(self, job: Job, **fields) -> None: + for k, v in fields.items(): + setattr(job, k, v) + if job.state == "running" and not job.started: + job.started = time.monotonic() + self.refresh_text() + + def refresh_text(self) -> None: + active = [j for j in self.jobs if j.state in ("running", "queued", "waiting")] + ended = [j for j in self.jobs if j.state in ("done", "failed")] + lines = [self._line(j) for j in active[:self.SHOWN]] + if len(active) > self.SHOWN: + lines.append(f" … and {len(active) - self.SHOWN} more waiting") + for j in reversed(ended[-(self.SHOWN - len(lines)):] if len(lines) < self.SHOWN + else []): + lines.append(self._line(j)) + self.update(Text("\n".join(lines) or "No transfers.")) + + def _line(self, j: Job) -> str: + name = safe(j.name) + if j.state == "running": + pct = 100 * j.done // j.size if j.size else 100 + rate = j.done / max(time.monotonic() - j.started, 0.001) + return f"{j.arrow} {name} {pct}% {human_size(int(rate))}/s" + if j.state == "queued": + return f"{j.arrow} {name} waiting for a free slot on the node {j.note}" + if j.state == "waiting": + return f"{j.arrow} {name} waiting" + if j.state == "done": + return f"{j.arrow} {name} done{(' — ' + j.note) if j.note else ''}" + return f"{j.arrow} {name} FAILED: {j.note}" |