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

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

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

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

12 files changed, 2208 insertions, 0 deletions

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