path: root/packages/meshbay-cli/tests/test_e2e_local.py

"""
The real `meshbay-cli` against a real hub and a real node, both on loopback.

Opt-in (MESHBAY_CLI_E2E=1): it starts a throwaway hub (SQLite, a fresh key) and
a node in a home of its own, which takes a while and runs Argon2 at 128 MB
several times. Nothing reaches a real hub or the developer's node.

What it sets up, as an operator and a browser would:
- an account, activated directly in the throwaway hub's database (no mail);
- a group whose name holds a `/` and an `@`, which the hub allows and the path
  syntax has to carry;
- a node linked to the account, hosting the group over a folder whose names
  are the awkward ones real data has: spaces, accents, an apostrophe, a long
  name without spaces, and a name with an escape character in it;
- the account's first contact with the node — a member invitation code, then a
  keypair bundle left on the node — done here the way a browser does it,
  because the CLI does not join groups.

Then the CLI: login, ls of groups, of a root, recursive, JSON, a second run
from the identity it kept, and logout. And, where google-chrome is installed,
the application's group page linking itself as a device against the same node.

    MESHBAY_CLI_E2E=1 pytest packages/meshbay-cli/tests/test_e2e_local.py -v

MESHBAY_CLI_E2E_BIN names another meshbay-cli to run in place of the one in
this venv: a packaged build, through a wrapper that mounts it where it installs.
"""

import asyncio
import base64
import json
import os
import re
import secrets
import shutil
import socket
import sqlite3
import subprocess
import sys
import time
from pathlib import Path

import httpx
import pytest
from console import ctrl_c, interruptible

pytestmark = [
    pytest.mark.skipif(os.environ.get("MESHBAY_CLI_E2E") != "1",
                       reason="opt-in: set MESHBAY_CLI_E2E=1 (see the module docstring)"),
    pytest.mark.timeout(600),
]

USERNAME = "alice-e2e-cli"
PASSPHRASE = "a long enough test passphrase, nobody's"
GROUP_NAME = "Some Saga/Box @Set"
BIN = Path(sys.executable).parent
# The environment before conftest points HOME, USERPROFILE and LOCALAPPDATA at an
# empty directory for each test: Chrome on Windows does not start in one.
REAL_ENV = dict(os.environ)
# Windows refuses a control character in a file name, so a node there cannot
# hold the name with an escape character in it.
ESCAPE_NAME = sys.platform != "win32"


def _program(linux: str, *windows: str) -> str | None:
    """A program by its Linux name, or where Windows installs it."""
    found = shutil.which(linux)
    if found or sys.platform != "win32":
        return found
    for base in (os.environ.get("ProgramFiles"), os.environ.get("ProgramFiles(x86)")):
        candidate = os.path.join(base or "", *windows)
        if base and os.path.isfile(candidate):
            return candidate
    return None


CHROME = _program("google-chrome", "Google", "Chrome", "Application", "chrome.exe")
VLC = _program("cvlc", "VideoLAN", "VLC", "vlc.exe")


def _free_port() -> int:
    with socket.socket() as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


def _wait_http(url: str, timeout: float = 60) -> None:
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        try:
            httpx.get(url, timeout=2)
            return
        except httpx.HTTPError:
            time.sleep(0.3)
    raise AssertionError(f"{url} did not come up")


@pytest.fixture(scope="module")
def world(tmp_path_factory):
    root = tmp_path_factory.mktemp("e2e")
    procs: list[subprocess.Popen] = []
    try:
        yield _build(root, procs)
    finally:
        for p in procs:
            p.terminate()
        for p in procs:
            try:
                p.wait(10)
            except subprocess.TimeoutExpired:
                p.kill()


def _build(root: Path, procs: list) -> dict:
    from cryptography.hazmat.primitives import serialization
    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
    from meshbay_cli.client.keys import auth_key

    # ── The hub ──────────────────────────────────────────────────────────────
    hub_home = root / "hub-home"
    hub_home.mkdir()
    key = root / "hub_private.pem"
    key.write_bytes(Ed25519PrivateKey.generate().private_bytes(
        serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8,
        serialization.NoEncryption()))
    db = root / "hub.db"
    port = _free_port()
    hub_url = f"http://127.0.0.1:{port}"
    hub_env = {**os.environ, "HOME": str(hub_home),
               "MESHBAY_DATABASE_URL": f"sqlite+aiosqlite:///{db}",
               "MESHBAY_HUB_KEY": str(key), "MESHBAY_HUB_ID": "e2e-hub",
               "MESHBAY_CAPTCHA_ENABLED": "false"}
    hub_log = open(root / "hub.log", "w")
    procs.append(subprocess.Popen(
        [sys.executable, "-m", "uvicorn", "meshbay_hub.app:create_app", "--factory",
         "--host", "127.0.0.1", "--port", str(port)],
        env=hub_env, stdout=hub_log, stderr=subprocess.STDOUT))
    _wait_http(f"{hub_url}/v1/hub/version")

    ak = auth_key(USERNAME, PASSPHRASE)
    with httpx.Client(base_url=hub_url, timeout=60) as hub:
        r = hub.post("/v1/users/register", json={
            "username": USERNAME, "email": "alice@example.invalid", "auth_key": ak})
        assert r.status_code == 201, r.text
        with sqlite3.connect(db) as conn:
            conn.execute("UPDATE users SET status='active' WHERE username=?", (USERNAME,))
        r = hub.post("/v1/users/login", json={"username": USERNAME, "auth_key": ak})
        assert r.status_code == 200, r.text
        auth = {"Authorization": f"Bearer {r.json()['access_token']}"}
        r = hub.post("/v1/groups", headers=auth, json={
            "name": GROUP_NAME, "visibility": "private", "join_policy": "invite"})
        assert r.status_code == 201, r.text
        group_id = r.json()["group_id"]

        # ── The node ─────────────────────────────────────────────────────────
        node_home = root / "node-home"
        node_home.mkdir()
        share = root / "share"
        (share / "2024" / "été à la mer").mkdir(parents=True)
        (share / "2024" / "été à la mer" / "plage de l'ouest.jpg").write_bytes(
            secrets.token_bytes(3000))
        (share / "2024" / "A_Very_Long_Name_Without_Any_Spaces_In_It_1992-2000.flac")\
            .write_bytes(secrets.token_bytes(2_500_000))
        (share / "notes.txt").write_text("hello\n")
        if ESCAPE_NAME:
            (share / "evil\x1b[31mred.txt").write_text("escape\n")
        # USERPROFILE, LOCALAPPDATA and APPDATA: where the node's directories
        # are on Windows; HOME alone would leave it in the developer's own.
        node_env = {**os.environ, "HOME": str(node_home), "USERPROFILE": str(node_home),
                    "LOCALAPPDATA": str(node_home / "AppData" / "Local"),
                    "APPDATA": str(node_home / "AppData" / "Roaming"),
                    "XDG_CONFIG_HOME": str(node_home / ".config"),
                    "XDG_DATA_HOME": str(node_home / ".local/share"),
                    "XDG_STATE_HOME": str(node_home / ".local/state")}
        node = [str(BIN / "meshbay-node")]

        def run_node(*args: str) -> str:
            p = subprocess.run(node + list(args), env=node_env, capture_output=True,
                               text=True, timeout=120)
            assert p.returncode == 0, f"meshbay-node {args}: {p.stdout}\n{p.stderr}"
            return p.stdout

        out = run_node("init", "--hub-url", hub_url, "--username", USERNAME)
        node_pk = re.search(r"Node key: (\S+)", out).group(1)
        cfg = next(node_home.rglob("node.toml"))
        ui_port = _free_port()
        cfg.write_text(cfg.read_text().replace("ui_port   = 18000",
                                               f"ui_port   = {ui_port}"))
        r = hub.put("/v1/users/me/node_key", headers=auth,
                    json={"pk_node_ed25519": node_pk})
        assert r.status_code == 200, r.text

        node_log = open(root / "node.log", "w")
        procs.append(subprocess.Popen(node, env=node_env, stdout=node_log,
                                      stderr=subprocess.STDOUT))
        _wait_http(f"http://127.0.0.1:{ui_port}/", timeout=90)

        # Attaching reloads the node by itself; the key needs the group loaded.
        # The control API answers before the hub socket is up, later still now
        # that the connection processes start first.
        deadline = time.monotonic() + 90
        while "Hub not connected" in (out := _try(run_node, "group", "add", group_id,
                                                  "--dir", str(share))):
            assert time.monotonic() < deadline, f"group add never succeeded: {out}"
            time.sleep(1)
        assert "meshbay-node (" not in out, out
        deadline = time.monotonic() + 90
        while "GEK ready" not in (out := _try(run_node, "gek", "init", "--group", group_id)):
            assert time.monotonic() < deadline, f"gek init never succeeded: {out}"
            time.sleep(2)
        code = re.search(r"INVITATION CODE\s+(\S+)",
                         run_node("member", "invite", USERNAME, "--group", group_id)).group(1)

        # The node is online at the hub once it lists itself for the group.
        deadline = time.monotonic() + 90
        while not hub.get(f"/v1/groups/{group_id}/nodes", headers=auth).json()["nodes"]:
            assert time.monotonic() < deadline, "the node never reached the hub"
            time.sleep(1)

    app = App(hub_url, group_id)
    asyncio.run(app.first_contact(code))
    return {"root": root, "hub_url": hub_url, "group_id": group_id, "db": db,
            "share": share, "app": app, "run_node": run_node}


async def _signed_in(account) -> None:
    """Signed in, waiting out the hub's limit of 10 sign-ins a minute: this
    module signs in more often than that, as the application and as the CLI."""
    from meshbay_cli.errors import CliError
    deadline = time.monotonic() + 90
    while True:
        try:
            return await account.sign_in(remember=False)
        except CliError as e:
            if "Too many attempts" not in str(e) or time.monotonic() > deadline:
                raise
            await asyncio.sleep(5)


def _try(fn, *args) -> str:
    try:
        return fn(*args)
    except AssertionError as e:
        return str(e)


class App:
    """The application's side of the test: a device already linked to the node."""

    def __init__(self, hub_url: str, group_id: str):
        self.hub_url, self.group_id = hub_url, group_id
        self.ident = None
        self.passphrase = PASSPHRASE

    async def session(self, code: str = "", *, fetch: bool = False):
        """Signed in, handshake done, joined: (channel, close, node_pk, nonce_s, user_id).
        `fetch`: the node's copy of the account's keys, asked for in the handshake
        window as a browser asks, is left in `self.fetched`."""
        import os as _os

        from aiortc import (
            RTCConfiguration,
            RTCIceServer,
            RTCPeerConnection,
            RTCSessionDescription,
        )
        from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
        from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
        from meshbay_cli.client.account import Account
        from meshbay_cli.client.hub import Hub
        from meshbay_cli.client.keys import Identity, b64, unb64
        from meshbay_cli.client.session import STUN_SERVERS, Channel, _dtls_fingerprint
        from meshbay_common import MNP_VERSION
        from meshbay_common.crypto import unwrap_gek_aes
        from meshbay_common.handshake import MNP_MIN_SUPPORTED, make_proof, webrtc_binding
        from meshbay_common.join import join_transcript

        hub = Hub(self.hub_url)
        account = Account(hub, USERNAME, lambda: self.passphrase)
        await _signed_in(account)
        token = await account.token()
        node = (await hub.group_nodes(token, self.group_id))[0]
        mnp = await hub.mnp_token(token, node["pk_node"], self.group_id)
        pc = RTCPeerConnection(RTCConfiguration(
            iceServers=[RTCIceServer(urls=u) for u in STUN_SERVERS]))
        dc = pc.createDataChannel("mnp", ordered=True)
        ch = Channel(dc)
        opened = asyncio.Event()
        dc.on("open", opened.set)
        await pc.setLocalDescription(await pc.createOffer())
        offer = pc.localDescription.sdp
        answer = await hub.offer(token, node["node_id"], offer)
        await pc.setRemoteDescription(RTCSessionDescription(sdp=answer, type="answer"))
        await asyncio.wait_for(opened.wait(), 30)
        binding = webrtc_binding(_dtls_fingerprint(offer), _dtls_fingerprint(answer))
        nonce_c = _os.urandom(32)
        chal = await ch.request({"type": "handshake", "v": MNP_VERSION,
                                 "v_min": MNP_MIN_SUPPORTED, "token": mnp,
                                 "group_id": self.group_id, "nonce": b64(nonce_c)},
                                "handshake_challenge")
        nonce_s, node_pk = unb64(chal["nonce"]), chal["node_pk"]
        if fetch:
            self.fetched = await ch.request({"type": "keypair_bundle_fetch",
                                             "v": MNP_VERSION}, "keypair_bundle_resp")
        if self.ident is None:
            self.ident = Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate())
        ident, ts = self.ident, int(time.time())
        joined = await ch.request({
            "type": "join_request", "v": MNP_VERSION, "group_id": self.group_id,
            "pk_ed25519": ident.pk_ed, "pk_x25519": ident.pk_x, "code": code, "ts": ts,
            "sig": b64(ident.sk_ed.sign(join_transcript(
                node_pk_b64=node_pk, group_id=self.group_id, user_id=account.user_id,
                pk_ed25519_b64=ident.pk_ed, pk_x25519_b64=ident.pk_x,
                nonce_node=nonce_s, ts=ts)))}, "join_result")
        assert joined.get("ok"), joined
        gek = unwrap_gek_aes(joined, ident.sk_x.private_bytes_raw(),
                             ident.sk_x.public_key().public_bytes_raw())
        await ch.request({"type": "handshake_response", "v": MNP_VERSION,
                          "proof": b64(make_proof(gek, "client", self.group_id, nonce_c,
                                                  nonce_s, binding))}, "handshake_ack")

        async def close():
            await pc.close()
            await account.close()
            await hub.aclose()
        return ch, close, node_pk, nonce_s, account

    async def first_contact(self, code: str) -> None:
        """What a browser does the first time: join with the code, leave a bundle."""
        import os as _os

        from cryptography.hazmat.primitives.ciphers.aead import AESGCM
        from meshbay_cli.client.keys import node_bundle_key
        from meshbay_common import MNP_VERSION

        ch, close, node_pk, _, account = await self.session(code)
        # MBK3, as keyderive.js encryptBundle writes it.
        nonce = _os.urandom(12)
        aad = f"meshbay:bundle:v3|{account.user_id}|{node_pk}".encode()
        ct = AESGCM(node_bundle_key(account.master, node_pk)).encrypt(
            nonce, json.dumps(self.ident.to_pkcs8()).encode(), aad)
        bundle = base64.b64encode(b"MBK3" + bytes([1]) + nonce + ct).decode()
        stored = await ch.request({"type": "keypair_bundle_store", "v": MNP_VERSION,
                                   "bundle_enc": bundle}, "ack")
        assert stored.get("detail") == "keypair_bundle_stored", stored
        await close()

    async def leave_bundle(self, recovery: bytes | None = None) -> None:
        """Browser access: this device's keys left on the node under `M`, and a
        recovery copy beside it when the recovery key is held."""
        from meshbay_cli.client.keys import seal_bundle, seal_recovery
        from meshbay_common import MNP_VERSION
        ch, close, node_pk, _, account = await self.session()
        keys = self.ident.to_pkcs8()
        msg = {"type": "keypair_bundle_store", "v": MNP_VERSION,
               "bundle_enc": seal_bundle(keys, master=account.master, pepper_version=1,
                                         user_id=account.user_id, node_pk=node_pk)}
        if recovery is not None:
            msg["bundle_enc_recovery"] = seal_recovery(keys, rkey=recovery,
                                                       user_id=account.user_id,
                                                       node_pk=node_pk)
        assert (await ch.request(msg, "ack")).get("detail") == "keypair_bundle_stored"
        await close()

    async def held(self) -> dict:
        """What the node keeps of the account's keys, as a browser fetches it."""
        ch, close, node_pk, _, account = await self.session(fetch=True)
        found = self.fetched
        out = {"node_pk": node_pk, "user_id": account.user_id, "master": account.master,
               **{k: found.get(k) for k in ("bundle_enc", "bundle_enc_recovery")}}
        await close()
        return out

    async def withdraw_bundle(self) -> None:
        """Browser access turned off: the application withdraws the node's copy."""
        from meshbay_common import MNP_VERSION
        ch, close, *_ = await self.session()
        r = await ch.request({"type": "keypair_bundle_delete", "v": MNP_VERSION}, "ack")
        assert r.get("detail") == "keypair_bundle_deleted", r
        await close()

    async def revoke_the_others(self) -> None:
        """Retire every device of the account but this one: the node admits five,
        and the tests link more machines than that between them."""
        from meshbay_cli.client.keys import b64
        from meshbay_common import MNP_VERSION
        from meshbay_common.device import device_revoke_transcript
        ch, close, node_pk, nonce_s, account = await self.session()
        listed = await ch.request({"type": "device_list", "v": MNP_VERSION},
                                  "device_list_result")
        for d in listed["devices"]:
            if d["pk_ed25519"] == self.ident.pk_ed:
                continue
            ts = int(time.time())
            sig = self.ident.sk_ed.sign(device_revoke_transcript(
                node_pk_b64=node_pk, user_id=account.user_id,
                pk_ed25519_b64=d["pk_ed25519"], nonce_node=nonce_s, ts=ts))
            await ch.request({"type": "device_revoke", "v": MNP_VERSION,
                              "pk_ed25519": d["pk_ed25519"], "ts": ts, "sig": b64(sig)},
                             "device_add_ack")
        await close()

    async def approve(self, code: str) -> None:
        """Approve a device by the code it shows, as the group's Settings tab does."""
        from meshbay_cli.client.keys import b64
        from meshbay_common import MNP_VERSION
        from meshbay_common.device import device_add_transcript, device_code_hash
        ch, close, node_pk, nonce_s, account = await self.session()
        listed = await ch.request({"type": "device_lookup", "v": MNP_VERSION},
                                  "device_lookup_result")
        raw = code.replace("-", "").upper()
        match = next(r for r in listed["requests"]
                     if device_code_hash(raw, r["pk_ed25519"], r["pk_x25519"])
                     == r["code_hash"])
        ts = int(time.time())
        sig = self.ident.sk_ed.sign(device_add_transcript(
            node_pk_b64=node_pk, user_id=account.user_id,
            pk_ed25519_b64=match["pk_ed25519"], pk_x25519_b64=match["pk_x25519"],
            nonce_node=nonce_s, ts=ts))
        await ch.request({"type": "device_add", "v": MNP_VERSION,
                          "pk_ed25519": match["pk_ed25519"],
                          "pk_x25519": match["pk_x25519"],
                          "code_hash": match["code_hash"], "ts": ts, "sig": b64(sig)},
                         "device_add_ack")
        await close()


def cli(world: dict, *args: str, stdin: str | None = None,
        home: str = "cli", env_extra: dict | None = None) -> subprocess.CompletedProcess:
    env = {**os.environ, "MESHBAY_CLI_HOME": str(world["root"] / home),
           "PYTHON_KEYRING_BACKEND": "keyring.backends.fail.Keyring"}
    env.pop("MESHBAY_PASSWORD", None)
    env.update(env_extra or {})
    exe = os.environ.get("MESHBAY_CLI_E2E_BIN") or str(BIN / "meshbay-cli")
    return subprocess.run([exe, *args], env=env, input=stdin,
                          capture_output=True, text=True, encoding="utf-8", timeout=180)


def _handle() -> str:
    return GROUP_NAME.replace("%", "%25").replace("/", "%2F") + "@" + USERNAME


def test_the_cli_against_a_real_hub_and_node(world):
    hub = world["hub_url"]

    r = cli(world, "--hub", hub, "-u", USERNAME, "login", "--password-file", "-",
            stdin=PASSPHRASE + "\n")
    assert r.returncode == 0, r.stderr
    assert "Signed in" in r.stdout and "1 group" in r.stdout
    # No key storage in this environment: the file fallback, said once.
    assert "No OS key storage here" in r.stderr

    # The groups, from the hub alone; the stored session needs no passphrase.
    r = cli(world, "ls")
    assert r.returncode == 0, r.stderr
    assert r.stdout.strip() == _handle()
    r = cli(world, "groups")
    assert r.returncode == 0, r.stderr
    print(r.stdout)
    [line] = r.stdout.splitlines()[1:]
    assert line.split()[:3] == ["up", "owner", "private"] and line.endswith(_handle())

    # One group, by its handle: its root.
    r = cli(world, "-vv", "ls", "-l", _handle())
    assert r.returncode == 0, r.stderr
    assert "keypair_bundle_fetch" in r.stderr     # first contact: the node's copy
    assert re.search(r"^root\s.*share/\s+\(writable\)$", r.stdout, re.M), r.stdout

    # Recursive: the awkward names arrive, the escape character is escaped.
    r = cli(world, "-vv", "ls", "-R", "-l", f"{_handle()}/share")
    assert r.returncode == 0, r.stderr
    assert "keypair_bundle_fetch" not in r.stderr  # the identity was kept
    out = r.stdout
    assert "plage de l'ouest.jpg" in out
    assert "A_Very_Long_Name_Without_Any_Spaces_In_It_1992-2000.flac" in out
    assert "\x1b" not in out
    assert not ESCAPE_NAME or "evil\\x1b[31mred.txt" in out
    assert "été à la mer/" in out

    # By name alone, case-insensitively, and as JSON.
    r = cli(world, "--json", "ls", "some saga%2Fbox @set/share/2024")
    assert r.returncode == 0, r.stderr
    items = json.loads(r.stdout)
    names = {i["name"]: i for i in items}
    assert names["été à la mer"]["kind"] == "dir"
    flac = names["A_Very_Long_Name_Without_Any_Spaces_In_It_1992-2000.flac"]
    assert flac["kind"] == "file" and flac["type"] == "audio"
    assert flac["size"] == 2_500_000 and len(flac["id"]) == 64

    # Not found is exit 3; an unknown group too.
    assert cli(world, "ls", f"{_handle()}/share/nope").returncode == 3
    assert cli(world, "ls", "No Such Group").returncode == 3

    # ── get ──────────────────────────────────────────────────────────────────
    out_dir = world["root"] / "downloads"
    flac = "A_Very_Long_Name_Without_Any_Spaces_In_It_1992-2000.flac"
    r = cli(world, "get", f"{_handle()}/share/2024/{flac}", "-o", f"{out_dir}/")
    assert r.returncode == 0, r.stderr
    assert (out_dir / flac).read_bytes() == (world["share"] / "2024" / flac).read_bytes()
    r = cli(world, "get", "-R", f"{_handle()}/share", "-o", f"{out_dir}/tree/")
    assert r.returncode == 0, r.stderr
    got = out_dir / "tree" / "share"
    assert (got / "2024" / "été à la mer" / "plage de l'ouest.jpg").read_bytes() == \
        (world["share"] / "2024" / "été à la mer" / "plage de l'ouest.jpg").read_bytes()
    # The escape character could not be written as it is: renamed, and said so.
    if ESCAPE_NAME:
        assert "evil\\x1b[31mred.txt is saved as evil_[31mred.txt" in r.stderr
        assert (got / "evil_[31mred.txt").exists()
    r = cli(world, "get", f"{_handle()}/share/notes.txt", "-o", "-")
    assert r.returncode == 0 and r.stdout == "hello\n"

    # ── put ──────────────────────────────────────────────────────────────────
    local = world["root"] / "to-send"
    (local / "Scans" / "2024").mkdir(parents=True)
    scan = secrets.token_bytes(300_000)
    (local / "Scans" / "2024" / "page 1.pdf").write_bytes(scan)
    (local / "Scans" / "empty").mkdir()
    (local / "Notes.TXT").write_bytes(b"clash")
    r = cli(world, "put", "-R", str(local / "Scans"), f"{_handle()}/share/2024")
    assert r.returncode == 0, r.stderr
    assert (world["share"] / "2024" / "Scans" / "2024" / "page 1.pdf").read_bytes() == scan
    assert (world["share"] / "2024" / "Scans" / "empty").is_dir()
    # notes.txt is there already, whatever the case: nothing sent.
    r = cli(world, "put", str(local / "Notes.TXT"), f"{_handle()}/share")
    assert r.returncode == 8 and "already there" in r.stderr
    # By listing, not exists(): on Windows that finds notes.txt under any case.
    assert "Notes.TXT" not in os.listdir(world["share"])
    r = cli(world, "put", "-", f"{_handle()}/share/piped.txt", stdin="through stdin")
    assert r.returncode == 0, r.stderr
    assert (world["share"] / "piped.txt").read_text() == "through stdin"
    # And what was sent can be fetched back, once the node has indexed it.
    deadline = time.monotonic() + 60
    while True:
        r = cli(world, "get", f"{_handle()}/share/2024/Scans/2024/page 1.pdf", "-o",
                str(out_dir / "back.pdf"))
        if r.returncode == 0 or time.monotonic() > deadline:
            break
        time.sleep(1)
    assert r.returncode == 0, r.stderr
    assert (out_dir / "back.pdf").read_bytes() == scan

    # ── ls -R over every group ───────────────────────────────────────────────
    deadline = time.monotonic() + 60
    while True:
        r = cli(world, "ls", "-R", "/")
        assert r.returncode == 0, r.stderr
        if "piped.txt" in r.stdout or time.monotonic() > deadline:
            break
        time.sleep(1)       # the node indexes what it received on its own schedule
    assert "page 1.pdf" in r.stdout and "piped.txt" in r.stdout, r.stdout

    # The device key was registered, and logout retires it.
    with sqlite3.connect(world["db"]) as conn:
        assert conn.execute("SELECT count(*) FROM user_devices").fetchone()[0] == 1
    r = cli(world, "logout")
    assert r.returncode == 0, r.stderr
    assert "Signed out." in r.stdout
    with sqlite3.connect(world["db"]) as conn:
        assert conn.execute("SELECT count(*) FROM user_devices").fetchone()[0] == 0
    assert not (world["root"] / "cli" / "secrets.enc").exists()

    # Signed out: a run needs credentials again, and says so.
    r = cli(world, "ls")
    assert r.returncode == 2 and "Not signed in" in r.stderr


def test_a_changed_node_key_is_refused_until_reset(world):
    def run(*args, stdin=None):
        return cli(world, *args, stdin=stdin, home="cli-trust")

    r = run("--hub", world["hub_url"], "-u", USERNAME, "login", "--password-file", "-",
            stdin=PASSPHRASE + "\n")
    assert r.returncode == 0, r.stderr
    assert run("ls", _handle()).returncode == 0

    r = run("--json", "trust", "list")
    [(node_id, pk)] = json.loads(r.stdout).items()
    state_path = world["root"] / "cli-trust" / "state.json"
    state = json.loads(state_path.read_text())
    state["pins"][node_id] = base64.b64encode(b"\x01" * 32).decode()
    state_path.write_text(json.dumps(state))
    r = run("ls", _handle())
    assert r.returncode == 6 and "trust reset" in r.stderr, r.stderr
    assert run("trust", "reset", "no-such-node").returncode == 3
    r = run("trust", "reset", node_id)
    assert r.returncode == 0, r.stderr
    assert run("ls", _handle()).returncode == 0
    assert json.loads(run("--json", "trust", "list").stdout) == {node_id: pk}
    assert run("logout").returncode == 0


def test_a_machine_with_no_identity_there_is_linked_as_a_device(world):
    """Browser access off: the node keeps no copy, so a new machine asks to be
    approved, shows a code, and joins with its own keys once approved."""
    app = world["app"]
    asyncio.run(app.withdraw_bundle())

    def run(*args, stdin=None):
        return cli(world, *args, stdin=stdin, home="cli2")

    r = run("--hub", world["hub_url"], "-u", USERNAME, "login", "--password-file", "-",
            stdin=PASSPHRASE + "\n")
    assert r.returncode == 0, r.stderr

    # No terminal here: the code is shown, and the run ends rather than waits.
    r = run("ls", _handle())
    assert r.returncode == 5, r.stderr
    found = re.search(r"enter\s+([0-9A-Z]{4}-[0-9A-Z]{4})", r.stderr)
    assert found, r.stderr
    code = found.group(1)
    # A second run shows the same code: the request it filed is still pending.
    r = run("ls", _handle())
    assert r.returncode == 5 and code in r.stderr, r.stderr

    asyncio.run(app.approve(code))
    r = run("ls", "-l", _handle())
    assert r.returncode == 0, r.stderr
    assert "share/" in r.stdout

    assert run("logout").returncode == 0


async def test_the_interactive_mode_against_the_real_node(world, monkeypatch, tmp_path):
    """The panes, opened on a real group: F5 bringing a file back, F5 sending one,
    and Delete taking that one away again."""
    from meshbay_cli import __main__ as main_module
    from meshbay_cli.commands.context import Context
    from meshbay_cli.tui.app import MeshBayApp
    from meshbay_cli.tui.panes import LocalPane, RemotePane
    from meshbay_cli.tui.screens import MainScreen
    from meshbay_cli.tui.widgets import Choice

    # Browser access off, whichever test ran before: this machine must be linked.
    await world["app"].withdraw_bundle()
    r = await asyncio.to_thread(cli, world, "--hub", world["hub_url"], "-u", USERNAME,
                                "login", "--password-file", "-",
                                stdin=PASSPHRASE + "\n", home="cli-tui")
    assert r.returncode == 0, r.stderr
    monkeypatch.setenv("MESHBAY_CLI_HOME", str(world["root"] / "cli-tui"))
    monkeypatch.chdir(tmp_path)
    args = main_module.build_parser().parse_args(["--hub", world["hub_url"]])
    app = MeshBayApp(Context(args))

    def names(pane):
        return [pane.table.get_row_at(i)[0].plain.split("  (")[0].strip()
                for i in range(pane.table.row_count)]

    async def until(pilot, cond, timeout=60):
        deadline = time.monotonic() + timeout
        while not cond():
            shown = (names(app.screen.query_one(RemotePane))
                     if isinstance(app.screen, MainScreen) else None)
            assert time.monotonic() < deadline, (
                f"timed out on {type(app.screen).__name__}; "
                f"notified: {[n.message for n in app._notifications]}; remote: {shown}")
            await pilot.pause(0.1)

    async with app.run_test(size=(140, 40)) as pilot:
        await until(pilot, lambda: isinstance(app.screen, MainScreen))
        remote = app.screen.query_one(RemotePane)
        name = _handle() + "/"
        await until(pilot, lambda: name in names(remote))
        remote.table.move_cursor(row=names(remote).index(name))
        remote.table.focus()
        await pilot.press("enter")
        # The node keeps no copy of the identity: this machine is a new device,
        # shows its code, and waits while it is approved elsewhere — then the
        # group opens by itself.
        await until(pilot, lambda: any("enter " in n.message for n in app._notifications))
        note = next(n.message for n in app._notifications if "enter " in n.message)
        code = re.search(r"enter ([0-9A-Z]{4}-[0-9A-Z]{4})", note).group(1)
        await world["app"].approve(code)
        await until(pilot, lambda: "share/" in names(remote))
        remote.table.move_cursor(row=names(remote).index("share/"))
        await pilot.press("enter")
        await until(pilot, lambda: "notes.txt" in names(remote))
        remote.table.move_cursor(row=names(remote).index("notes.txt"))
        await pilot.press("f5")
        await until(pilot, lambda: (tmp_path / "notes.txt").exists()
                    and (tmp_path / "notes.txt").read_text() == "hello\n")

        # Sent from the left, then deleted from the right: it was this
        # account's upload, so the node lets it go, from its disk too.
        (tmp_path / "tui-upload.txt").write_text("sent from the panes\n")
        local = app.screen.query_one(LocalPane)
        local.action_refresh()
        await until(pilot, lambda: "tui-upload.txt" in names(local))
        local.table.move_cursor(row=names(local).index("tui-upload.txt"))
        local.table.focus()
        await pilot.press("f5")
        await until(pilot, lambda: "tui-upload.txt" in names(remote))
        assert (world["share"] / "tui-upload.txt").exists()
        remote.table.move_cursor(row=names(remote).index("tui-upload.txt"))
        remote.table.focus()
        await pilot.press("delete")
        await until(pilot, lambda: isinstance(app.screen, Choice))
        await pilot.click("#choice-delete")
        await until(pilot, lambda: "tui-upload.txt" not in names(remote))
        assert not (world["share"] / "tui-upload.txt").exists()
        assert (world["share"] / "notes.txt").exists()


# ── The application's half of linking, in a real Chrome ──────────────────────
#
# Here because this is where a real hub and node stand: the group page, served
# by the hub, linking a device in the desktop application — the request filed
# before the proof (MNP 6.3) on the connection the node refused, then the group
# opening by itself once a linked device approves. The application's key store
# (keyring.js, in its main process) is stood in for by a page script that keeps
# the identity in memory and signs through the page's own transcripts.

_KEYS_STUB = r"""
(() => {
  const ids = {};
  const b64 = (b) => btoa(String.fromCharCode(...new Uint8Array(b)));
  const k = (u, n) => `${u}|${n}`;
  window.__minted = 0;
  const keys = {
    available: async () => true,
    hasSession: async () => true,
    identity: async (u, n) => (ids[k(u, n)] ? { ...ids[k(u, n)].pub, sealedWith: null } : null),
    async mint(u, n) {
      window.__minted += 1;
      const ed = await crypto.subtle.generateKey({ name: 'Ed25519' }, true, ['sign', 'verify']);
      const x = await crypto.subtle.generateKey({ name: 'X25519' }, true, ['deriveBits']);
      const pub = { pkEdB64: b64(await crypto.subtle.exportKey('raw', ed.publicKey)),
                    pkXB64: b64(await crypto.subtle.exportKey('raw', x.publicKey)) };
      ids[k(u, n)] = { skEd: b64(await crypto.subtle.exportKey('pkcs8', ed.privateKey)),
                       skX: x.privateKey, pub };
      return pub;
    },
    async sign(u, n, kind, fields) {
      const id = ids[k(u, n)];
      return window.MeshBayKeys.signBytes(
        id.skEd, window.MeshBayCrypto.transcriptFor(kind, fields, id.pub));
    },
    async shared(u, n, peerB64) {
      const raw = Uint8Array.from(atob(peerB64), (c) => c.charCodeAt(0));
      const pk = await crypto.subtle.importKey('raw', raw, { name: 'X25519' }, false, []);
      return b64(await crypto.subtle.deriveBits(
        { name: 'X25519', public: pk }, ids[k(u, n)].skX, 256));
    },
    browserAccess: async () => false,
    fingerprint: async () => 'fp',
    openBundle: async () => { throw new Error('the node keeps no bundle here'); },
    commitPending: async () => true, dropPending: async () => true,
    forgetSession: async () => true, createdHere: async () => true,
  };
  window.meshbay = { hubBase: () => location.origin, setHubBase: async () => true, keys };
  localStorage.setItem('mb_auth', JSON.stringify(__AUTH__));
})();
"""

_LINK_FORM = r"""(() => {
  const page = document.body ? document.body.innerText : '';   // null mid-navigation
  const form = [...document.querySelectorAll('.invite-form')]
    .find((f) => f.querySelector('button, p') && /linked/i.test(f.innerText));
  const code = form && [...form.querySelectorAll('p')].map((p) => p.innerText.trim())
    .find((t) => /^[0-9A-Z]{4}-[0-9A-Z]{4}$/.test(t));
  return { form: Boolean(form), button: Boolean(form && form.querySelector('button')),
           code: code || '', minted: window.__minted,
           error: (document.querySelector('.error-msg') || {}).innerText || '',
           // The group open on its Chat tab, the composer ready.
           opened: Boolean(document.querySelector('.chat-input:not([disabled])')),
           page: page.slice(-400) };
})()"""


class _Chrome:
    """Chrome, headless, driven over the DevTools protocol."""

    def __init__(self, init_script: str):
        import tempfile

        from websockets.sync.client import connect
        self.profile = tempfile.TemporaryDirectory(prefix="meshbay-e2e-chrome-",
                                                   ignore_cleanup_errors=True)
        port = _free_port()
        self.proc = subprocess.Popen(
            [CHROME, "--headless=new", "--disable-gpu", "--no-sandbox",
             f"--remote-debugging-port={port}", f"--user-data-dir={self.profile.name}",
             "--lang=en-US", "about:blank"],
            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env=REAL_ENV)
        deadline = time.monotonic() + 30
        while True:
            try:
                pages = httpx.get(f"http://127.0.0.1:{port}/json/list").json()
                target = next(p for p in pages if p["type"] == "page")
                break
            except (httpx.HTTPError, StopIteration):
                if time.monotonic() > deadline:
                    # Not left running: nobody holds this object to close it.
                    self.proc.kill()
                    self.proc.wait(10)
                    raise AssertionError("Chrome did not come up") from None
                time.sleep(0.2)
        self.ws = connect(target["webSocketDebuggerUrl"], max_size=None)
        self.n = 0
        self.call("Page.enable")
        self.call("Page.addScriptToEvaluateOnNewDocument", source=init_script)

    def call(self, method: str, **params):
        self.n += 1
        self.ws.send(json.dumps({"id": self.n, "method": method, "params": params}))
        while True:
            msg = json.loads(self.ws.recv())
            if msg.get("id") == self.n:
                assert "error" not in msg, msg
                return msg.get("result", {})

    def js(self, expr: str):
        r = self.call("Runtime.evaluate", expression=expr, awaitPromise=True,
                      returnByValue=True)
        assert "exceptionDetails" not in r, r["exceptionDetails"]
        return r["result"].get("value")

    def until(self, expr: str, pred, timeout: float = 90):
        deadline = time.monotonic() + timeout
        while not pred(state := self.js(expr)):
            assert time.monotonic() < deadline, f"timed out: {state}"
            time.sleep(0.5)
        return state

    def close(self):
        self.ws.close()
        self.proc.terminate()
        self.proc.wait(10)
        self.profile.cleanup()


def _session_for_the_page(world: dict) -> dict:
    """What the page keeps in `mb_auth` after signing in, for the account."""
    from meshbay_cli.client.keys import auth_key
    hub_url = world["hub_url"]
    deadline = time.monotonic() + 90
    while (r := httpx.post(f"{hub_url}/v1/users/login", json={
            "username": USERNAME, "auth_key": auth_key(USERNAME, PASSPHRASE)})
           ).status_code == 429 and time.monotonic() < deadline:
        time.sleep(5)             # the hub's 10 sign-ins a minute (see _signed_in)
    assert r.status_code == 200, r.text
    tokens = r.json()
    me = httpx.get(f"{hub_url}/v1/users/me",
                   headers={"Authorization": f"Bearer {tokens['access_token']}"}).json()
    return {"username": USERNAME, "userId": me["user_id"], "role": "user",
            "token": tokens["access_token"], "refreshToken": tokens.get("refresh_token")}


def _linked_group_page(world: dict) -> "_Chrome":
    """The group page in Chrome as the desktop application, linked as a new
    device by its own "Get a linking code" and open. The caller closes it."""
    app = world["app"]
    asyncio.run(app.withdraw_bundle())
    chrome = _Chrome(_KEYS_STUB.replace("__AUTH__", json.dumps(_session_for_the_page(world))))
    try:
        chrome.call("Page.navigate",
                    url=f"{world['hub_url']}/#/group/{world['group_id']}")
        # Refused as an unknown device: the form, and nothing shown as an error.
        s = chrome.until(_LINK_FORM, lambda s: s["button"])
        assert not s["error"], s
        chrome.js("""[...document.querySelectorAll('.invite-form button')]
                     .find((b) => /linking code/i.test(b.innerText)).click()""")
        s = chrome.until(_LINK_FORM, lambda s: s["code"] or s["error"])
        assert s["code"] and not s["error"], s
        asyncio.run(app.approve(s["code"]))
        # Nobody clicks anything: the page tries again and the group opens.
        s = chrome.until(_LINK_FORM, lambda s: s["opened"] and not s["form"], timeout=60)
        assert not s["error"], s
        assert s["minted"] == 1, "the identity linked is not the one kept"
        return chrome
    except BaseException:
        chrome.close()
        raise


@pytest.mark.skipif(not CHROME, reason="Chrome is not available")
def test_the_application_links_itself_from_its_group_page(world):
    _linked_group_page(world).close()


# ── A second account: invited, joined, its devices, browser access, leaving ──

BOB = "bob-e2e-cli"
BOB_PASSPHRASE = "another long test passphrase, bob's"


def _register(world: dict, username: str, passphrase: str) -> None:
    from meshbay_cli.client.keys import auth_key
    r = httpx.post(f"{world['hub_url']}/v1/users/register", json={
        "username": username, "email": f"{username}@example.invalid",
        "auth_key": auth_key(username, passphrase)})
    assert r.status_code == 201, r.text
    with sqlite3.connect(world["db"]) as conn:
        conn.execute("UPDATE users SET status='active' WHERE username=?", (username,))


def bob(world: dict, home: str, *args: str) -> subprocess.CompletedProcess:
    """`meshbay-cli` as bob, on the machine `home` names; no terminal."""
    return cli(world, *args, home=home)


def _bob_login(world: dict, home: str) -> None:
    deadline = time.monotonic() + 90
    while True:
        r = cli(world, "--hub", world["hub_url"], "-u", BOB, "login", "--password-file",
                "-", stdin=BOB_PASSPHRASE + "\n", home=home)
        if r.returncode == 0:
            return
        assert "Too many attempts" in r.stderr and time.monotonic() < deadline, r.stderr
        time.sleep(5)                     # the hub's 10 sign-ins a minute


def test_a_member_joins_links_devices_shares_with_browsers_and_leaves(world):
    _register(world, BOB, BOB_PASSPHRASE)
    group = _handle()
    # The owner invites bob on the node, which also tells the hub.
    code = re.search(r"INVITATION CODE\s+(\S+)",
                     world["run_node"]("member", "invite", BOB, "--group",
                                       world["group_id"])).group(1)

    _bob_login(world, "bob1")
    r = bob(world, "bob1", "groups")
    assert r.returncode == 0 and "1 invitation waiting" in r.stderr, r.stderr
    r = bob(world, "bob1", "invitations")
    assert r.returncode == 0 and r.stdout.splitlines()[1].endswith(group), r.stdout

    # Accepted on the hub; the node wants the code, and there is no terminal.
    r = bob(world, "bob1", "join", group)
    assert r.returncode == 5 and "--code CODE" in r.stderr, r.stderr
    r = bob(world, "bob1", "join", group, "--code", code)
    assert r.returncode == 0, r.stderr
    assert "Joined" in r.stdout and "browser-access on" in r.stderr
    r = bob(world, "bob1", "ls", f"{group}/share")
    assert r.returncode == 0 and "notes.txt" in r.stdout, r.stderr

    # Browser access on: the next connection leaves a copy, which a machine
    # with nothing of its own then opens — the same identity, no approval.
    assert bob(world, "bob1", "browser-access", "on").returncode == 0
    assert bob(world, "bob1", "ls", group).returncode == 0
    _bob_login(world, "bob2")
    r = bob(world, "bob2", "ls", group)
    assert r.returncode == 0 and "share/" in r.stdout, r.stderr
    r = bob(world, "bob1", "--json", "devices", "list", group)
    assert r.returncode == 0, r.stderr
    assert len(json.loads(r.stdout)["devices"]) == 1

    # Off: the copy is withdrawn, so a third machine is a new device, approved
    # from the first by its code, and revoked from it.
    assert bob(world, "bob1", "browser-access", "off").returncode == 0
    assert bob(world, "bob1", "ls", group).returncode == 0
    _bob_login(world, "bob3")
    r = bob(world, "bob3", "ls", group)
    assert r.returncode == 5, r.stderr
    link_code = re.search(r"enter\s+([0-9A-Z]{4}-[0-9A-Z]{4})", r.stderr).group(1)
    assert f"meshbay-cli devices approve {group} {link_code}" in r.stderr
    r = bob(world, "bob1", "devices", "approve", group, link_code)
    assert r.returncode == 0, r.stderr
    assert bob(world, "bob3", "ls", group).returncode == 0
    listed = json.loads(bob(world, "bob1", "--json", "devices", "list", group).stdout)
    new = next(d for d in listed["devices"] if not d["is_this_one"])
    assert len(listed["devices"]) == 2
    r = bob(world, "bob1", "devices", "revoke", group, new["pk_ed25519"][:12])
    assert r.returncode == 1 and "--yes" in r.stderr           # no terminal to ask on
    r = bob(world, "bob1", "devices", "revoke", group, new["pk_ed25519"][:12], "--yes")
    assert r.returncode == 0, r.stderr
    assert bob(world, "bob3", "ls", group).returncode == 5      # a stranger again

    # rm: what bob uploaded, from another of his machines too; nothing of the
    # owner's, whose share bob did not upload.
    up = world["root"] / "bob-upload.txt"
    up.write_text("from bob\n")
    assert bob(world, "bob1", "put", str(up), f"{group}/share").returncode == 0
    deadline = time.monotonic() + 30          # indexed once the node has written it
    while "bob-upload.txt" not in bob(world, "bob1", "ls", f"{group}/share").stdout:
        assert time.monotonic() < deadline, "the upload never reached the index"
        time.sleep(1)
    r = bob(world, "bob1", "rm", f"{group}/share/bob-upload.txt")
    assert r.returncode == 1 and "--yes" in r.stderr             # no terminal to ask on
    r = bob(world, "bob1", "rm", "--yes", f"{group}/share/notes.txt")
    assert r.returncode == 8 and "did not upload" in r.stderr, r.stderr
    r = bob(world, "bob2", "rm", "--yes", f"{group}/share/bob-upload.txt")
    assert r.returncode == 0, r.stderr
    assert "bob-upload.txt" not in bob(world, "bob1", "ls", f"{group}/share").stdout
    assert (world["share"] / "notes.txt").exists()

    # Leaving: asked first, then the group is gone from the account.
    assert bob(world, "bob1", "leave", group).returncode == 1
    r = bob(world, "bob1", "leave", "--yes", group)
    assert r.returncode == 0, r.stderr
    assert group not in bob(world, "bob1", "groups").stdout
    assert bob(world, "bob1", "ls", group).returncode == 3

    # Back through an invitation link, pasted as it was sent.
    out = world["run_node"]("member", "invite", "--link", "--group", world["group_id"])
    link = re.search(r"INVITATION LINK\s+(\S+)", out).group(1)
    r = bob(world, "bob1", "join", link)
    assert r.returncode == 0, r.stderr
    assert bob(world, "bob1", "ls", f"{group}/share").returncode == 0
    for home in ("bob1", "bob2", "bob3"):
        assert bob(world, home, "logout").returncode == 0


def test_search_finds_a_file_and_get_takes_its_path(world, tmp_path):
    """The path `search` prints is the one `get` takes, accents, spaces, an
    apostrophe and a group name holding a / and an @ included."""
    deadline = time.monotonic() + 90
    while True:
        r = cli(world, "--hub", world["hub_url"], "-u", USERNAME, "login",
                "--password-file", "-", stdin=PASSPHRASE + "\n", home="cli-search")
        if r.returncode == 0:
            break
        assert "Too many attempts" in r.stderr and time.monotonic() < deadline, r.stderr
        time.sleep(5)
    r = cli(world, "search", "PLAGE", "ouest", home="cli-search")
    # After a test that withdrew the node's copy, this machine is a new device:
    # search names the group and its code instead of waiting (exit 4).
    if r.returncode == 4 and "waits to be approved" in r.stderr:
        code = re.search(r"\(([0-9A-Z]{4}-[0-9A-Z]{4})\)", r.stderr).group(1)
        asyncio.run(world["app"].approve(code))
        r = cli(world, "search", "PLAGE", "ouest", home="cli-search")
    assert r.returncode == 0, r.stderr
    [path] = r.stdout.splitlines()
    assert path == f"{_handle()}/share/2024/été à la mer/plage de l'ouest.jpg"
    r = cli(world, "get", path, "-o", str(tmp_path) + "/", home="cli-search")
    assert r.returncode == 0, r.stderr
    assert (tmp_path / "plage de l'ouest.jpg").stat().st_size == 3000
    r = cli(world, "search", "no-such-thing-anywhere", home="cli-search")
    assert r.returncode == 3 and "Nothing found" in r.stderr


def _cli_signed_in(world: dict, home: str) -> None:
    deadline = time.monotonic() + 90
    while True:
        r = cli(world, "--hub", world["hub_url"], "-u", USERNAME, "login",
                "--password-file", "-", stdin=PASSPHRASE + "\n", home=home)
        if r.returncode == 0:
            return
        assert "Too many attempts" in r.stderr and time.monotonic() < deadline, r.stderr
        time.sleep(5)


@pytest.mark.skipif(not CHROME, reason="Chrome is not available")
def test_chat_between_the_command_line_and_the_application(world):
    """Posted here, read on the group page in Chrome; answered there, read here —
    the Python seal and the page's crypto.js, on a real node."""
    group = _handle()
    asyncio.run(world["app"].revoke_the_others())    # room for this machine and the page
    _cli_signed_in(world, "cli-chat")
    r = cli(world, "chat", group, "-m", "hello from the terminal 1", home="cli-chat")
    if r.returncode == 5:                         # a new device here: approved first
        found = re.search(r"enter\s+([0-9A-Z]{4}-[0-9A-Z]{4})", r.stderr)
        assert found, r.stderr
        code = found.group(1)
        asyncio.run(world["app"].approve(code))
        r = cli(world, "chat", group, "-m", "hello from the terminal 1", home="cli-chat")
    assert r.returncode == 0, r.stderr
    for i in range(2, 13):
        r = cli(world, "chat", group, "-m", f"hello from the terminal {i}", home="cli-chat")
        assert r.returncode == 0, r.stderr

    r = cli(world, "chat", group, home="cli-chat")
    assert r.returncode == 0, r.stderr
    lines = r.stdout.splitlines()
    assert len(lines) == 10
    assert lines[0].endswith(f"{USERNAME}: hello from the terminal 3")
    assert lines[-1].endswith(f"{USERNAME}: hello from the terminal 12")
    r = cli(world, "chat", "-100", group, home="cli-chat")
    assert len(r.stdout.splitlines()) == 12, r.stdout

    chrome = _linked_group_page(world)
    try:
        # The page opens the terminal's messages: signature, epoch key, content.
        chrome.until("document.body.innerText",
                     lambda text: "hello from the terminal 12" in (text or ""))
        assert "cannot be read" not in chrome.js("document.body.innerText").lower()
        chrome.js("""(() => {
            const box = document.querySelector('.chat-input');
            box.value = 'hello from the browser';
            box.dispatchEvent(new Event('input', { bubbles: true }));
        })()""")
        chrome.until("!document.querySelector('.chat-send').disabled", bool)
        chrome.js("document.querySelector('.chat-send').click()")
        chrome.until("document.querySelector('.chat-input').value === ''", bool)
    finally:
        chrome.close()

    deadline = time.monotonic() + 30
    while True:
        r = cli(world, "chat", "-1", group, home="cli-chat")
        if r.stdout.strip().endswith(f"{USERNAME}: hello from the browser"):
            break
        assert time.monotonic() < deadline, (r.stdout, r.stderr)
        time.sleep(1)


async def test_the_interactive_chat_against_the_real_node(world, monkeypatch, tmp_path):
    """The room open on a real group: a message posted from another run arrives
    live, and one typed in the room is read back by `chat`. Runs after the
    chat test above, on the machine it linked."""
    from meshbay_cli import __main__ as main_module
    from meshbay_cli.commands.context import Context
    from meshbay_cli.tui.app import MeshBayApp
    from meshbay_cli.tui.chat import ChatGroupsScreen, ChatRoomScreen
    from textual.widgets import DataTable, Input, RichLog

    group = _handle()
    r = await asyncio.to_thread(cli, world, "chat", "-1", group, home="cli-chat")
    assert r.returncode == 0, "the chat test above did not leave this machine linked"
    monkeypatch.setenv("MESHBAY_CLI_HOME", str(world["root"] / "cli-chat"))
    app = MeshBayApp(Context(main_module.build_parser().parse_args(
        ["--hub", world["hub_url"]])), ChatGroupsScreen)

    def text(screen) -> str:
        return "\n".join("".join(s.text for s in strip)
                         for strip in screen.query_one(RichLog).lines)

    async def until(pilot, cond, timeout=60):
        deadline = time.monotonic() + timeout
        while not cond():
            assert time.monotonic() < deadline, [n.message for n in app._notifications]
            await pilot.pause(0.1)

    async with app.run_test(size=(140, 40)) as pilot:
        await until(pilot, lambda: isinstance(app.screen, ChatGroupsScreen)
                    and app.screen.query_one(DataTable).row_count >= 1)
        await pilot.press("enter")
        await until(pilot, lambda: isinstance(app.screen, ChatRoomScreen))
        room = app.screen
        await until(pilot, lambda: "hello from the browser" in text(room))
        assert "cannot be read" not in text(room) and "not seen before" not in text(room)

        r = await asyncio.to_thread(cli, world, "chat", group, "-m",
                                    "live from another run", home="cli-chat")
        assert r.returncode == 0, r.stderr
        await until(pilot, lambda: f"<{USERNAME}> live from another run" in text(room))

        room.query_one(Input).value = "typed in the room"
        await pilot.press("enter")
        await until(pilot, lambda: f"<{USERNAME}> typed in the room" in text(room))
        await pilot.press("escape")
        await until(pilot, lambda: isinstance(app.screen, ChatGroupsScreen))

    r = await asyncio.to_thread(cli, world, "chat", "-1", group, home="cli-chat")
    assert r.stdout.strip().endswith(f"{USERNAME}: typed in the room"), r.stdout


async def test_put_and_get_in_the_interactive_chat_against_the_real_node(
        world, monkeypatch, tmp_path):
    """/put sends a file into the chat's folder on a real node and names it in
    the chat, as the page's paperclip does: `chat` and the group page show it as
    a file. /get saves a file of that folder into the current directory."""
    from meshbay_cli import __main__ as main_module
    from meshbay_cli.commands.context import Context
    from meshbay_cli.tui.app import MeshBayApp
    from meshbay_cli.tui.chat import ChatGroupsScreen, ChatLog, ChatRoomScreen
    from textual.widgets import DataTable, Input

    group = _handle()
    r = await asyncio.to_thread(cli, world, "chat", "-1", group, home="cli-chat")
    assert r.returncode == 0, "the chat tests above did not leave this machine linked"
    monkeypatch.setenv("MESHBAY_CLI_HOME", str(world["root"] / "cli-chat"))
    here = tmp_path / "here"
    here.mkdir()
    monkeypatch.chdir(here)
    sent = secrets.token_bytes(3 * 1024 * 1024 + 17)          # several chunks
    (here / "from the room.bin").write_bytes(sent)
    app = MeshBayApp(Context(main_module.build_parser().parse_args(
        ["--hub", world["hub_url"]])), ChatGroupsScreen)

    def said(screen) -> str:
        return "\n".join(t for *_, t in screen.query_one(ChatLog)._blocks)

    async def until(pilot, cond, timeout=60):
        deadline = time.monotonic() + timeout
        while not cond():
            assert time.monotonic() < deadline, [n.message for n in app._notifications]
            await pilot.pause(0.1)

    async with app.run_test(size=(140, 40)) as pilot:
        await until(pilot, lambda: isinstance(app.screen, ChatGroupsScreen)
                    and app.screen.query_one(DataTable).row_count >= 1)
        await pilot.press("enter")
        await until(pilot, lambda: isinstance(app.screen, ChatRoomScreen))
        room = app.screen
        await until(pilot, lambda: "typed in the room" in said(room))

        # No folder chosen for the chat: the first writable root, `share`.
        room.query_one(Input).value = "/put from the room.bin"
        await pilot.press("enter")
        await until(pilot, lambda: "[file: from the room.bin, 3.0 MB]" in said(room))
        assert (world["share"] / "from the room.bin").read_bytes() == sent

        room.query_one(Input).value = "/get notes.txt"
        await pilot.press("enter")
        await until(pilot, lambda: "Saved " in said(room) or "! " in said(room))
        assert (here / "notes.txt").read_text() == "hello\n", said(room)
        await pilot.press("escape")

    r = await asyncio.to_thread(cli, world, "chat", "-1", group, home="cli-chat")
    assert r.stdout.strip().endswith(f"{USERNAME}: [file: from the room.bin, 3.0 MB]"), \
        r.stdout
    r = await asyncio.to_thread(cli, world, "--json", "chat", "-1", group, home="cli-chat")
    assert json.loads(r.stdout)[0]["attachment"] == {
        "filename": "from the room.bin", "size": len(sent), "type": "file"}

    if CHROME:
        await asyncio.to_thread(_the_page_shows_the_file, world, "from the room.bin")


def _the_page_shows_the_file(world: dict, name: str) -> None:
    # Not revoke_the_others: `chat --follow` below runs on this machine's key.
    chrome = _linked_group_page(world)
    try:
        files = chrome.until(
            "[...document.querySelectorAll('.chat-attachment')].map(a => a.innerText)",
            lambda found: any(name in f for f in found or []))
        assert any("3" in f and "MB" in f for f in files if name in f), files
        assert '"attachment"' not in chrome.js("document.body.innerText")
    finally:
        chrome.close()


def test_chat_follow_against_the_real_node(world):
    """`chat --follow` in its own process: messages posted from another run
    arrive as the node pushes them, across more than one liveness ping, and
    Ctrl+C ends it. Runs on the machine the chat tests above linked."""
    group = _handle()
    env = {**os.environ, "MESHBAY_CLI_HOME": str(world["root"] / "cli-chat"),
           "PYTHON_KEYRING_BACKEND": "keyring.backends.fail.Keyring"}
    exe = os.environ.get("MESHBAY_CLI_E2E_BIN") or str(BIN / "meshbay-cli")
    follower = interruptible([exe, "--json", "chat", "-1", group, "--follow"], env)
    try:
        first = json.loads(follower.stdout.readline())          # the last one, at once
        assert first["text"] or first.get("attachment")       # /put's, above
        for text, wait in (("followed one", 0), ("followed two, after a ping", 36)):
            time.sleep(wait)
            r = cli(world, "chat", group, "-m", text, home="cli-chat")
            assert r.returncode == 0, r.stderr
            got = json.loads(follower.stdout.readline())
            assert got["text"] == text and got["sender"] == USERNAME, got
            assert got["trust"] in ("pinned", "first", "linked")
    finally:
        ctrl_c(follower)
        _, err = follower.communicate(timeout=20)
    assert follower.returncode == 130
    assert "reconnecting" not in err, err


@pytest.mark.skipif(not (shutil.which("ffmpeg") and shutil.which("ffprobe")
                         and VLC),
                    reason="ffmpeg, ffprobe and VLC are needed")
def test_stream_a_film_to_real_players(world, tmp_path):
    """A film put on the real node, then streamed: ffprobe reads it, ffmpeg seeks
    near its end, a foreign Host is refused, Ctrl+C stops it; then VLC started
    with --player plays it to the end and, closing, ends the stream."""
    _cli_signed_in(world, "cli-stream")
    film = tmp_path / "Night Drive.mkv"
    subprocess.run(["ffmpeg", "-v", "error", "-f", "lavfi", "-i",
                    "testsrc=duration=12:size=640x360:rate=25", "-f", "lavfi", "-i",
                    "sine=frequency=440:duration=12", "-vf", "noise=alls=30:allf=t",
                    "-c:v", "libx264", "-b:v", "3M",
                    "-g", "50", "-c:a", "aac", "-shortest", str(film)], check=True)
    assert film.stat().st_size > 3 * 1024 * 1024        # several of the node's chunks
    r = cli(world, "put", str(film), f"{_handle()}/share", home="cli-stream")
    # After a test that withdrew the node's copy of the keys, this machine is a
    # new device: approved as the application would approve it.
    if r.returncode == 5 and "Waiting for this machine to be approved" in r.stderr:
        code = re.search(r"code ([0-9A-Z]{4}-[0-9A-Z]{4})", r.stderr).group(1)
        asyncio.run(world["app"].approve(code))
        r = cli(world, "put", str(film), f"{_handle()}/share", home="cli-stream")
    assert r.returncode == 0, r.stderr
    path = f"{_handle()}/share/Night Drive.mkv"
    deadline = time.monotonic() + 60                    # once the node has indexed it
    while "Night Drive.mkv" not in cli(world, "ls", f"{_handle()}/share",
                                       home="cli-stream").stdout:
        assert time.monotonic() < deadline, "the film was never indexed"
        time.sleep(1)

    env = {**os.environ, "MESHBAY_CLI_HOME": str(world["root"] / "cli-stream"),
           "PYTHON_KEYRING_BACKEND": "keyring.backends.fail.Keyring"}
    exe = os.environ.get("MESHBAY_CLI_E2E_BIN") or str(BIN / "meshbay-cli")
    streamer = interruptible([exe, "--json", "stream", path], env)
    try:
        line = streamer.stdout.readline()
        assert line, streamer.stderr.read()
        shown = json.loads(line)
        url = shown["url"]
        assert shown["size"] == film.stat().st_size
        probe = subprocess.run(["ffprobe", "-v", "error", "-show_entries",
                                "format=duration:stream=codec_name", "-of", "json", url],
                               capture_output=True, text=True, timeout=60)
        assert probe.returncode == 0, probe.stderr
        info = json.loads(probe.stdout)
        assert abs(float(info["format"]["duration"]) - 12) < 0.5
        assert {s["codec_name"] for s in info["streams"]} == {"h264", "aac"}
        frame = tmp_path / "frame.png"
        seek = subprocess.run(["ffmpeg", "-v", "error", "-ss", "10", "-i", url,
                               "-frames:v", "1", str(frame)],
                              capture_output=True, text=True, timeout=60)
        assert seek.returncode == 0 and frame.stat().st_size > 0, seek.stderr
        port = url.split("/")[2].split(":")[1]
        foreign = httpx.get(url, headers={"Host": f"rebound.example:{port}"}, timeout=10,
                            trust_env=False)
        assert foreign.status_code == 403
    finally:
        ctrl_c(streamer)
        _, err = streamer.communicate(timeout=20)
    assert streamer.returncode == 130, err
    with pytest.raises(httpx.HTTPError):
        httpx.get(url, timeout=5, trust_env=False)

    log = tmp_path / "vlc.log"
    flags = "--play-and-exit --intf dummy --vout dummy --aout dummy"
    if sys.platform == "win32":
        player = tmp_path / "vlc-to-the-end.cmd"
        player.write_text(f'@"{VLC}" {flags} "%~1" >"{log}" 2>&1\r\n')
    else:
        player = tmp_path / "vlc-to-the-end"
        player.write_text(f"#!/bin/sh\nexec {VLC} {flags} \"$1\" >{log} 2>&1\n")
        player.chmod(0o755)
    started = time.monotonic()
    r = subprocess.run([exe, "stream", "--player", str(player), path], env=env,
                       capture_output=True, text=True, encoding="utf-8", timeout=120)
    assert r.returncode == 0, r.stderr + log.read_text()
    assert "The player was closed" in r.stderr
    assert time.monotonic() - started > 10, "VLC played the film, in real time"


def test_mkdir_and_rm_r_against_the_real_node(world, tmp_path):
    """mkdir --parents, a file put in the new folder, then rm -r of it all from
    the operator's key — paired as an operator pairs a browser."""
    group = _handle()
    r = cli(world, "rm", "-r", "--yes", f"{group}/share/2024", home="cli-chat")
    assert r.returncode == 8 and "operator" in r.stderr, r.stderr      # not yet

    code = re.search(r"PAIRING CODE\s+(\S+)", world["run_node"]("operator", "pair")).group(1)
    r = cli(world, "join", group, "--code", code, home="cli-chat")
    assert r.returncode == 0, r.stderr

    r = cli(world, "mkdir", "--parents", f"{group}/share/Album/Disc 1",
            home="cli-chat")
    assert r.returncode == 0, r.stderr
    assert (world["share"] / "Album" / "Disc 1").is_dir()
    (tmp_path / "track notes.txt").write_text("liner notes\n")
    r = cli(world, "put", str(tmp_path / "track notes.txt"), f"{group}/share/Album/Disc 1",
            home="cli-chat")
    assert r.returncode == 0, r.stderr
    assert (world["share"] / "Album" / "Disc 1" / "track notes.txt").is_file()
    r = cli(world, "mkdir", f"{group}/share/Album/Empty", home="cli-chat")
    assert r.returncode == 0, r.stderr
    deadline = time.monotonic() + 30
    while "track notes.txt" not in (r := cli(world, "ls", "-R", f"{group}/share/Album",
                                           home="cli-chat")).stdout:
        assert time.monotonic() < deadline, r.stdout        # the node indexes it first
        time.sleep(1)

    r = cli(world, "--json", "rm", "-r", "--yes", f"{group}/share/Album", home="cli-chat")
    assert r.returncode == 0, r.stderr
    out = json.loads(r.stdout)
    assert len(out["deleted"]) == 1 and len(out["folders_removed"]) == 3, out
    assert not (world["share"] / "Album").exists()


def test_tab_completes_remote_paths_from_the_real_node(world):
    """The candidates come from the hub and the node, for the machine signed in;
    bash's own completion function, and PowerShell's when pwsh is installed."""
    group = _handle()

    def complete(*words: str) -> list[str]:
        r = cli(world, "__complete", "bash", str(len(words)), "meshbay-cli", *words,
                home="cli-chat")
        assert r.returncode == 0 and not r.stderr, r.stderr
        return r.stdout.splitlines()

    assert complete("ls", group[:4]) == [group + "/"]
    assert complete("ls", f"{group}/sh") == [f"{group}/share/"]
    assert complete("get", f"{group}/share/20") == [f"{group}/share/2024/"]
    inside = complete("get", f"{group}/share/2024/")
    assert f"{group}/share/2024/été à la mer/" in inside
    assert f"{group}/share/2024/A_Very_Long_Name_Without_Any_Spaces_In_It_1992-2000.flac" \
        in inside
    folders = complete("mkdir", f"{group}/share/2024/")     # a destination: folders only
    assert f"{group}/share/2024/été à la mer/" in folders
    assert all(f.endswith("/") for f in folders), folders
    assert complete("chat", group[:4]) == [group]
    assert complete("put", "nothing-here", f"{group}/sh")[0] == "__files__"
    assert complete("ls", "No such group/") == []            # nothing, quietly

    # bash itself: the function the script installs, on a word typed with its
    # spaces escaped, as bash hands it over. On Windows, where Git's is on PATH.
    if shutil.which("bash"):
        script = cli(world, "completion", "bash").stdout
        escaped = group[:10].replace(" ", "\\ ")
        bash = subprocess.run(
            ["bash", "-c", script + "\nCOMP_WORDS=(meshbay-cli ls " + escaped + ")\n"
             "COMP_CWORD=2\n_meshbay_cli\nprintf '%s\\n' \"${COMPREPLY[@]}\"\n"],
            env={**os.environ, "MESHBAY_CLI_HOME": str(world["root"] / "cli-chat"),
                 "PYTHON_KEYRING_BACKEND": "keyring.backends.fail.Keyring",
                 "PATH": f"{BIN}{os.pathsep}{os.environ['PATH']}"},
            capture_output=True, text=True, encoding="utf-8", timeout=60)
        assert bash.stdout.splitlines() == [group + "/"], (bash.stdout, bash.stderr)

    # PowerShell 7, or Windows PowerShell 5.1, which decodes a program's output
    # with the console's code page unless told otherwise: an accented name is the
    # test. The matches come back as code points, past PowerShell's own output.
    pwsh = shutil.which("pwsh") or (sys.platform == "win32" and shutil.which("powershell"))
    if pwsh:
        ps = cli(world, "completion", "powershell").stdout
        for line, want in ((f"meshbay-cli ls '{group}/sh", f"'{group}/share/'"),
                           (f"meshbay-cli ls '{group}/share/2024/é",
                            f"'{group}/share/2024/été à la mer/'")):
            r = subprocess.run(
                [pwsh, "-NoProfile", "-Command",
                 ps + f"\n$r = TabExpansion2 -inputScript \"{line}\" -cursorColumn {len(line)}"
                 "\n$r.CompletionMatches | ForEach-Object {"
                 " ($_.CompletionText.ToCharArray() | ForEach-Object { [int]$_ }) -join ',' }"],
                env={**os.environ, "MESHBAY_CLI_HOME": str(world["root"] / "cli-chat"),
                     "PYTHON_KEYRING_BACKEND": "keyring.backends.fail.Keyring",
                     "PATH": f"{BIN}{os.pathsep}{os.environ['PATH']}"},
                capture_output=True, text=True, timeout=120)
            got = ["".join(map(chr, map(int, m.split(",")))) for m in r.stdout.split()]
            assert got == [want], (got, r.stderr)


def _cli(world: dict, *args: str, new: str | None = None, **kw):
    """`cli`, on this world's hub even from a machine that never signed in."""
    extra = {"MESHBAY_HUB": world["hub_url"]}
    if new is not None:
        extra["MESHBAY_NEW_PASSWORD"] = new
    return cli(world, *args, env_extra=extra, **kw)


def _code(world: dict, purpose: str) -> str:
    """The code the hub would have mailed: this throwaway hub mails nothing."""
    with sqlite3.connect(world["db"]) as conn:
        row = conn.execute("SELECT code FROM email_verifications WHERE purpose = ? "
                           "AND verified_at IS NULL ORDER BY created_at DESC LIMIT 1",
                           (purpose,)).fetchone()
    assert row, f"no {purpose} code"
    return row[0]


def test_an_account_registered_verified_changed_and_erased_from_the_cli(world):
    """register → verify → sign in → show → new address → sign out everywhere →
    delete, each against the real hub; the e-mailed codes read from its database."""
    name, email, new_email = "carol-e2e-cli", "carol@example.invalid", "carol2@example.invalid"
    phrase = "carol's own long passphrase, 2026"
    r = _cli(world, "account", "register", "-u", name, "--email", email,
            "--password-file", "-", stdin=phrase + "\n", home="cli-carol")
    assert r.returncode == 0, r.stderr
    mnemonic = re.search(r"^ {4}((?:[A-Z2-7]{4} ){12}[A-Z2-7]{4})$", r.stdout, re.M)
    assert mnemonic, r.stdout
    assert "meshbay-cli account verify CODE" in r.stderr

    # Not usable before the address is verified.
    r = _cli(world, "login", "-u", name, "--password-file", "-", stdin=phrase + "\n",
            home="cli-carol")
    assert r.returncode == 2 and "not verified" in r.stderr, r.stderr
    r = _cli(world, "account", "verify", _code(world, "registration"), "--email", email,
            home="cli-carol")
    assert r.returncode == 0 and "Verified" in r.stdout, r.stderr
    r = _cli(world, "login", "-u", name, "--password-file", "-", stdin=phrase + "\n",
            home="cli-carol")
    assert r.returncode == 0, r.stderr

    r = _cli(world, "--json", "account", home="cli-carol")
    assert json.loads(r.stdout)["email"] == email, r.stdout
    r = _cli(world, "account", "email", new_email, "--password-file", "-",
            stdin=phrase + "\n", home="cli-carol")
    assert r.returncode == 0 and "--code CODE" in r.stdout, (r.stdout, r.stderr)
    r = _cli(world, "account", "email", "--code", _code(world, "email_change"),
            home="cli-carol")
    assert r.returncode == 0, r.stderr
    r = _cli(world, "--json", "account", home="cli-carol")
    assert json.loads(r.stdout)["email"] == new_email, r.stdout

    r = _cli(world, "--json", "account", "sign-out-everywhere", home="cli-carol")
    assert r.returncode == 0 and json.loads(r.stdout)["revoked"] >= 1, r.stderr
    # This machine signs back in with its device key, without the passphrase.
    r = _cli(world, "--json", "account", home="cli-carol")
    assert r.returncode == 0, r.stderr

    r = _cli(world, "account", "delete", home="cli-carol")
    assert r.returncode == 1 and "--yes" in r.stderr, r.stderr      # never unasked
    r = _cli(world, "account", "delete", "--yes", "--password-file", "-",
            stdin=phrase + "\n", home="cli-carol")
    assert r.returncode == 0 and "erased" in r.stdout, r.stderr
    r = _cli(world, "login", "-u", name, "--no-remember", "--password-file", "-",
            stdin=phrase + "\n", home="cli-carol-2")
    assert r.returncode == 2, r.stderr


def test_a_new_passphrase_carries_the_browsers_keys_and_a_reset_restores_them(world):
    """
    The browsers' copy of the account's keys on the node — another device's
    keys than this machine's — sealed again under the new passphrase, the same
    keys; then, the passphrase forgotten, a machine that keeps nothing restores
    them from the recovery copy. Last: it changes the account's passphrase.
    """
    from meshbay_cli.client.keys import (
        generate_recovery,
        open_recovery,
        opens_under,
        recovery_key,
    )
    app = world["app"]
    raw, mnemonic = generate_recovery()
    rkey = recovery_key(raw, USERNAME)
    asyncio.run(app.leave_bundle(recovery=rkey))
    browsers = app.ident.to_pkcs8()

    new = "a brand new passphrase for alice, 2026"
    r = _cli(world, "chat", "-1", _handle(), home="cli-chat")
    assert r.returncode == 0, "the chat tests above did not leave this machine linked"
    r = _cli(world, "account", "passphrase", "--password-file", "-",
            stdin=PASSPHRASE + "\n", home="cli-chat", new=new)
    assert r.returncode == 0, r.stderr
    assert "sealed again on 1 node" in r.stdout, r.stdout

    app.passphrase = new
    held = asyncio.run(app.held())
    keys = opens_under(held["bundle_enc"], held["master"], user_id=held["user_id"],
                       node_pk=held["node_pk"])
    assert keys == browsers, "the browsers' keys were replaced, not carried"
    # The recovery copy, left as it was, still opens.
    assert open_recovery(held["bundle_enc_recovery"], rkey=rkey, user_id=held["user_id"],
                         node_pk=held["node_pk"]) == browsers
    r = _cli(world, "login", "--no-remember", "-u", USERNAME, "--password-file", "-",
            stdin=PASSPHRASE + "\n", home="cli-old")
    assert r.returncode == 2, "the old passphrase still signs in"
    # This machine goes on with the new one, without being asked for it.
    r = _cli(world, "ls", _handle(), home="cli-chat")
    assert r.returncode == 0, r.stderr

    # Forgotten: a code, then a machine that keeps nothing and the recovery key.
    reset = "the passphrase after the reset, 2026"
    r = _cli(world, "account", "reset", "-u", USERNAME, "--email", "alice@example.invalid",
            home="cli-reset")
    assert r.returncode == 0 and "--code CODE" in r.stderr, r.stderr
    key_file = world["root"] / "recovery.txt"
    key_file.write_text(mnemonic + "\n")
    r = _cli(world, "account", "reset", "-u", USERNAME, "--code",
            _code(world, "password_reset"), "--recovery-key-file", str(key_file),
            "--password-file", "-", stdin=reset + "\n", home="cli-reset")
    assert r.returncode == 0, r.stderr
    assert "sealed again on 1 node" in r.stdout, (r.stdout, r.stderr)

    app.passphrase = reset
    held = asyncio.run(app.held())
    assert opens_under(held["bundle_enc"], held["master"], user_id=held["user_id"],
                       node_pk=held["node_pk"]) == browsers
    assert open_recovery(held["bundle_enc_recovery"], rkey=rkey, user_id=held["user_id"],
                         node_pk=held["node_pk"]) == browsers
    r = _cli(world, "ls", _handle(), home="cli-reset")
    assert r.returncode == 0, r.stderr