""" 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 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 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") (share / "evil\x1b[31mred.txt").write_text("escape\n") node_env = {**os.environ, "HOME": str(node_home), "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. run_node("group", "add", group_id, "--dir", str(share)) 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 async def session(self, code: str = ""): """Signed in, handshake done, joined: (channel, close, node_pk, nonce_s, user_id).""" 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: 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 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 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") -> subprocess.CompletedProcess: env = {**os.environ, "MESHBAY_CLI_HOME": str(world["root"] / home), "PYTHON_KEYRING_BACKEND": "keyring.backends.fail.Keyring"} env.pop("MESHBAY_PASSWORD", None) 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, 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 and "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. 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 assert not (world["share"] / "Notes.TXT").exists() 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( ["google-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) 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): assert time.monotonic() < deadline, "Chrome did not come up" 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 shutil.which("google-chrome"), reason="google-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 shutil.which("google-chrome"), reason="google-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)