diff options
Diffstat (limited to 'packages/meshbay-cli/tests')
| -rw-r--r-- | packages/meshbay-cli/tests/fake_hub.py | 70 | ||||
| -rw-r--r-- | packages/meshbay-cli/tests/fake_node.py | 86 | ||||
| -rw-r--r-- | packages/meshbay-cli/tests/test_cli_actions.py | 192 | ||||
| -rw-r--r-- | packages/meshbay-cli/tests/test_e2e_local.py | 107 | ||||
| -rw-r--r-- | packages/meshbay-cli/tests/test_manpage.py | 2 | ||||
| -rw-r--r-- | packages/meshbay-cli/tests/test_membership.py | 78 | ||||
| -rw-r--r-- | packages/meshbay-cli/tests/test_store_and_session.py | 4 |
7 files changed, 534 insertions, 5 deletions
diff --git a/packages/meshbay-cli/tests/fake_hub.py b/packages/meshbay-cli/tests/fake_hub.py index ec1a0c0..15cbdf7 100644 --- a/packages/meshbay-cli/tests/fake_hub.py +++ b/packages/meshbay-cli/tests/fake_hub.py @@ -34,6 +34,9 @@ class FakeHub: self.refresh_delay = 0.0 self.groups = [{"id": "g1", "name": "Family", "owner_username": USERNAME, "node_online": True, "visibility": "private", "is_admin": True}] + self.pending: list[dict] = [] + self.links: dict[str, dict] = {} + self.prefs: dict[str, str] = {} def _session(self, family: str | None = None) -> dict: family = family or secrets.token_hex(4) @@ -106,3 +109,70 @@ class FakeHub: async def my_groups(self, token): self._auth(token) return self.groups + + # ── Joining and leaving ────────────────────────────────────────────────── + # + # `pending` holds invitations (group rows with `invited_by`); `links` maps a + # ticket to the group it admits to, and is spent by the first redeem. + + def _group_row(self, g: dict) -> dict: + return {k: v for k, v in g.items() if k not in ("invited_by", "created_at")} + + async def invitations(self, token): + self._auth(token) + return [{"group_id": g["id"], "name": g["name"], + "owner_username": g["owner_username"], "invited_by": g["invited_by"], + "created_at": "2026-10-14T09:30:00+00:00"} for g in self.pending] + + async def accept_invitation(self, token, group_id): + self._auth(token) + found = [g for g in self.pending if g["id"] == group_id] + if not found: + raise HubError(404, "No such invitation") + self.pending.remove(found[0]) + self.groups.append(self._group_row(found[0])) + return {"status": "joined", "group_id": group_id} + + async def decline_invitation(self, token, group_id): + self._auth(token) + found = [g for g in self.pending if g["id"] == group_id] + if not found: + raise HubError(404, "No such invitation") + self.pending.remove(found[0]) + return {"status": "declined", "group_id": group_id} + + async def preview_link(self, token, ticket): + self._auth(token) + g = self.links.get(ticket) + if g is None: + raise HubError(404, "invite_not_valid") + return {"group_id": g["id"], "group_name": g["name"], "inviter": "bob-test", + "expires_at": "2026-10-21T00:00:00+00:00", + "already_member": any(m["id"] == g["id"] for m in self.groups)} + + async def redeem_link(self, token, ticket): + self._auth(token) + g = self.links.pop(ticket, None) + if g is None: + raise HubError(404, "invite_not_valid") + self.groups.append(self._group_row(g)) + return {"group_id": g["id"], "group_name": g["name"]} + + async def leave_group(self, token, group_id): + self._auth(token) + g = next((g for g in self.groups if g["id"] == group_id), None) + if g is None: + raise HubError(404, "You are not a member of this group") + if g.get("is_admin"): + raise HubError(409, "You own this group, so you cannot leave it") + self.groups.remove(g) + return {"status": "left", "group_id": group_id} + + async def preferences(self, token): + self._auth(token) + return dict(self.prefs) + + async def set_preference(self, token, key, value): + self._auth(token) + self.prefs[key] = value + return {"key": key, "value": value} diff --git a/packages/meshbay-cli/tests/fake_node.py b/packages/meshbay-cli/tests/fake_node.py index e733c56..f53eb01 100644 --- a/packages/meshbay-cli/tests/fake_node.py +++ b/packages/meshbay-cli/tests/fake_node.py @@ -9,16 +9,30 @@ the node's code: a node the CLI must work against. """ import asyncio +import base64 import hashlib import math import secrets +from cryptography.hazmat.primitives.asymmetric.ed25519 import ( + Ed25519PrivateKey, + Ed25519PublicKey, +) +from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey +from fake_hub import USER_ID +from meshbay_cli.client.keys import Identity from meshbay_cli.client.session import GroupSession, NodeRefusal +from meshbay_common.device import ( + device_add_transcript, + device_code_hash, + device_revoke_transcript, +) from meshbay_common.groupbox import PURPOSE_INDEX, PURPOSE_UPLOAD, seal, unseal from meshbay_common.protocol import UPLOAD_PROBE_INDEX, file_upload_ack_wire from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes GROUP_ID = "g1" +NODE_PK = base64.b64encode(b"\x05" * 32).decode() CHUNK = 1024 * 1024 @@ -35,6 +49,14 @@ class FakeNode: self.refuse_upload_code = "" self.chunk_requests = 0 self.chunk_delay = 0.0 + # This machine's key here, the account's other devices, and the requests + # waiting for an approval — each `{pk_ed25519, pk_x25519, code_hash}`. + self.identity = Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate()) + self.nonce = secrets.token_bytes(32) + self.devices = [{"pk_ed25519": self.identity.pk_ed, "pk_x25519": self.identity.pk_x, + "label": "", "pinned_at": "2026-10-01T10:00:00+00:00", + "pinned_via": "code"}] + self.requests: list[dict] = [] def add(self, path: str, name: str, data: bytes) -> dict: fid = hashlib.blake2b(data + path.encode() + name.encode(), digest_size=32).hexdigest() @@ -47,7 +69,17 @@ class FakeNode: return entry def session(self) -> GroupSession: - return GroupSession(None, FakeChannel(self), GROUP_ID, "node-1", "pk", self.gek, {}) + return GroupSession(None, FakeChannel(self), GROUP_ID, "node-1", NODE_PK, self.gek, + {}, self.identity, self.nonce) + + def request_device(self, code: str) -> Identity: + """Another machine files a request with `code`, as it would before its proof.""" + other = Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate()) + self.requests.append({"pk_ed25519": other.pk_ed, "pk_x25519": other.pk_x, + "created_at": "2026-10-14T10:00:00+00:00", + "code_hash": device_code_hash(code.replace("-", ""), + other.pk_ed, other.pk_x)}) + return other def stored(self, path: str) -> bytes | None: parent, _, name = path.rpartition("/") @@ -162,5 +194,57 @@ class FakeChannel: dir=folder) + # ── Devices ────────────────────────────────────────────────────────────── + + def _device_list(self, msg): + n = self.node + return {"type": "device_list_result", "pending": len(n.requests), + "devices": [{**d, "is_this_one": d["pk_ed25519"] == n.identity.pk_ed} + for d in n.devices]} + + def _device_lookup(self, msg): + return {"type": "device_lookup_result", "requests": list(self.node.requests)} + + def _device_add(self, msg): + n = self.node + transcript = device_add_transcript( + node_pk_b64=NODE_PK, user_id=USER_ID, pk_ed25519_b64=msg["pk_ed25519"], + pk_x25519_b64=msg["pk_x25519"], nonce_node=n.nonce, ts=msg["ts"]) + if not _signed_by_a_device(n, transcript, msg["sig"]): + return {"type": "error", "detail": "Not signed by a device already paired here"} + req = next((r for r in n.requests if r["code_hash"] == msg["code_hash"] + and r["pk_ed25519"] == msg["pk_ed25519"]), None) + if req is None: + return {"type": "error", "detail": "No pending request"} + n.requests.remove(req) + n.devices.append({"pk_ed25519": req["pk_ed25519"], "pk_x25519": req["pk_x25519"], + "label": "", "pinned_at": "2026-10-14T10:01:00+00:00", + "pinned_via": "device"}) + return {"type": "device_add_ack", "pk_ed25519": req["pk_ed25519"]} + + def _device_revoke(self, msg): + n = self.node + transcript = device_revoke_transcript( + node_pk_b64=NODE_PK, user_id=USER_ID, pk_ed25519_b64=msg["pk_ed25519"], + nonce_node=n.nonce, ts=msg["ts"]) + if not _signed_by_a_device(n, transcript, msg["sig"]): + return {"type": "error", "detail": "Not signed by a device already paired here"} + if len(n.devices) <= 1: + return {"type": "error", "detail": "This is your only device here"} + n.devices = [d for d in n.devices if d["pk_ed25519"] != msg["pk_ed25519"]] + return {"type": "device_add_ack", "revoked": msg["pk_ed25519"]} + + +def _signed_by_a_device(node: FakeNode, transcript: bytes, sig: str) -> bool: + for d in node.devices: + try: + Ed25519PublicKey.from_public_bytes(base64.b64decode(d["pk_ed25519"])).verify( + base64.b64decode(sig), transcript) + return True + except Exception: + continue + return False + + def chunks_of(size: int, chunk: int) -> int: return max(1, math.ceil(size / chunk)) diff --git a/packages/meshbay-cli/tests/test_cli_actions.py b/packages/meshbay-cli/tests/test_cli_actions.py index b88d93c..4fffc4e 100644 --- a/packages/meshbay-cli/tests/test_cli_actions.py +++ b/packages/meshbay-cli/tests/test_cli_actions.py @@ -233,6 +233,198 @@ def test_groups_lists_each_group_with_its_state(hub, capsys): assert rows[0]["path"] == "Club%2FPhotos@bob-test" +class _Terminal(io.StringIO): + """stdin at a terminal, with the answers a person types.""" + + def isatty(self): + return True + + +def _signed_in(hub, capsys): + assert main("--hub", hub.url, "-u", USERNAME, "--password", PASSPHRASE, "login") == 0 + capsys.readouterr() + + +CLUB = {"id": "g2", "name": "Club/Photos", "owner_username": "bob-test", + "node_online": True, "visibility": "private", "is_admin": False, "hosted": True} + + +@pytest.fixture +def joined(monkeypatch): + """`join`'s node half, recorded: the code and the node each try was given.""" + from meshbay_cli.client.session import NodeRefusal + from meshbay_cli.commands import join + tries = [] + state = {"needs_code": False, "away": False} + + async def open_group(account, group, *, join_code=None, node_pk=None): + tries.append({"group": group["id"], "code": join_code, "node_pk": node_pk}) + if state["away"]: + from meshbay_cli.errors import NODE, CliError + raise CliError(NODE, "No node serving it is online.") + if state["needs_code"] and not join_code: + raise NodeRefusal("this node has not admitted you yet", "code_required") + return FakeSession() + monkeypatch.setattr(join, "open_group", open_group) + return tries, state + + +def test_an_invitation_is_listed_joined_and_named_in_groups(hub, joined, capsys): + tries, _ = joined + hub.pending.append({**CLUB, "invited_by": "bob-test"}) + _signed_in(hub, capsys) + + assert main("groups") == 0 + assert "1 invitation waiting for your answer" in capsys.readouterr().err + assert main("invitations") == 0 + header, line = capsys.readouterr().out.splitlines() + assert header.split() == ["INVITED", "BY", "DATE", "GROUP"] + assert line.startswith("bob-test") and line.endswith("Club%2FPhotos@bob-test") + + assert main("join", "Club%2FPhotos") == 0 + out = capsys.readouterr() + assert "Joined Club%2FPhotos@bob-test" in out.out + assert tries == [{"group": "g2", "code": None, "node_pk": None}] + # The hub keeps no word on browser access: said once, nothing done. + assert "meshbay-cli browser-access on" in out.err + assert [g["id"] for g in hub.groups] == ["g1", "g2"] and not hub.pending + + assert main("invitations") == 0 + assert "No invitation is waiting" in capsys.readouterr().err + assert main("join", "Nothing") == 3 + + +def test_a_node_that_asks_for_a_code_gets_it_from_the_option_or_the_terminal( + hub, joined, monkeypatch, capsys): + tries, state = joined + state["needs_code"] = True + hub.pending.append({**CLUB, "invited_by": "bob-test"}) + _signed_in(hub, capsys) + + monkeypatch.setattr("sys.stdin.isatty", lambda: False) + assert main("join", "Club%2FPhotos") == 5 # a member now, not admitted + err = capsys.readouterr().err + assert "join Club%2FPhotos@bob-test --code CODE" in err + + assert main("join", "Club%2FPhotos", "--code", "k7p2-9wqx") == 0 + assert tries[-1]["code"] == "K7P2-9WQX" + + monkeypatch.setattr("sys.stdin", _Terminal("ab12-cd34\n")) + assert main("join", "Club%2FPhotos@bob-test") == 0 + assert tries[-1]["code"] == "AB12-CD34" + + +def test_an_invitation_link_goes_to_its_hub_and_its_code_to_its_node(hub, joined, capsys): + tries, state = joined + ticket, node = "AbCdEfGhIjKlMnOpQr-_12", "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8" + link = (f"{hub.url}/#/invite?v=1&g=0f8fad5b-d9cb-469f-a165-70867728950e&t={ticket}" + f"&n={node}&c=k7p2-9wqx") + hub.links[ticket] = {**CLUB, "id": "0f8fad5b-d9cb-469f-a165-70867728950e"} + _signed_in(hub, capsys) + + assert main("join", link.replace(hub.url, "https://elsewhere.test")) == 1 + assert "--hub https://elsewhere.test" in capsys.readouterr().err + + state["away"] = True # redeemed, but the node is not there + assert main("join", link) == 5 + assert "The code is kept" in capsys.readouterr().err + assert tries[-1] == {"group": "0f8fad5b-d9cb-469f-a165-70867728950e", + "code": "K7P2-9WQX", + "node_pk": "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="} + from meshbay_cli.client.store import Store + store = Store() + store.load() + assert store.secrets["join_codes"]["0f8fad5b-d9cb-469f-a165-70867728950e"] == { + "code": "K7P2-9WQX", "node_pk": "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="} + + assert main("join", link) == 3 # spent: the hub says so + assert "no longer works" in capsys.readouterr().err + + +def test_declining_and_leaving_ask_first(hub, monkeypatch, capsys): + hub.pending.append({**CLUB, "invited_by": "bob-test"}) + hub.groups.append({**CLUB, "id": "g3", "name": "Choir"}) + _signed_in(hub, capsys) + + monkeypatch.setattr("sys.stdin.isatty", lambda: False) + assert main("decline", "Club%2FPhotos") == 1 # no terminal, no --yes + assert "pass --yes" in capsys.readouterr().err and hub.pending + assert main("leave", "Choir") == 1 + assert any(g["id"] == "g3" for g in hub.groups) + assert main("leave", "Family") == 8 # the owner, before any question + assert "You own" in capsys.readouterr().err + + monkeypatch.setattr("sys.stdin", _Terminal("n\n")) + assert main("leave", "Choir") == 1 + assert any(g["id"] == "g3" for g in hub.groups) + monkeypatch.setattr("sys.stdin", _Terminal("y\n")) + assert main("leave", "Choir") == 0 + assert "Left Choir@bob-test" in capsys.readouterr().out + assert not any(g["id"] == "g3" for g in hub.groups) + + assert main("decline", "--yes", "Club%2FPhotos") == 0 + assert not hub.pending and not any(g["id"] == "g2" for g in hub.groups) + + +def test_devices_are_listed_approved_by_code_and_revoked(node, monkeypatch, capsys): + from meshbay_cli.client.groups import open_group # noqa: F401 (patched below) + from meshbay_cli.commands import devices + + async def open_group_(account, group): + return node.session() + monkeypatch.setattr(devices, "open_group", open_group_) + + other = node.request_device("ZX4P-7QRT") + assert main("devices", "list", "Family") == 0 + out = capsys.readouterr() + assert out.out.splitlines()[1].startswith(f"* {node.identity.pk_ed[:16]}") + assert "1 device waiting for approval" in out.err + + assert main("devices", "approve", "Family", "zx4p-7qrz") == 3 # no request has it + assert main("devices", "approve", "Family", "ZX4P 7QRT") == 0 + assert other.pk_ed in [d["pk_ed25519"] for d in node.devices] + + # The same code filed twice: neither is approved. + node.request_device("AAAA-BBBB") + node.request_device("AAAA-BBBB") + assert main("devices", "approve", "Family", "AAAA-BBBB") == 3 + assert "none was approved" in capsys.readouterr().err + assert len(node.devices) == 2 + + monkeypatch.setattr("sys.stdin.isatty", lambda: False) + assert main("devices", "revoke", "Family", node.identity.pk_ed[:12], "--yes") == 8 + assert main("devices", "revoke", "Family", other.pk_ed[:12]) == 1 # not confirmed + assert main("devices", "revoke", "Family", other.pk_ed[:12], "--yes") == 0 + assert [d["pk_ed25519"] for d in node.devices] == [node.identity.pk_ed] + assert main("devices", "approve", "Family") == 1 + + +def test_browser_access_is_set_here_and_mirrored_never_taken_from_the_hub( + hub, joined, capsys): + hub.prefs["browser_access"] = "on" # anyone with a session can write it + _signed_in(hub, capsys) + assert main("--json", "browser-access") == 0 + assert json.loads(capsys.readouterr().out) == {"this_machine": None, "hub": "on"} + from meshbay_cli.client.store import Store + store = Store() + store.load() + assert "browser_access" not in store.state + + assert main("browser-access", "off") == 0 + assert hub.prefs["browser_access"] == "off" + store.load() + assert store.state["browser_access"] == "off" + # Set here: `join` has nothing to point out. + hub.pending.append({**CLUB, "invited_by": "bob-test"}) + assert main("join", "Club%2FPhotos") == 0 + assert "browser-access" not in capsys.readouterr().err + + # Kept through a new sign-in of the same account. + _signed_in(hub, capsys) + store.load() + assert store.state["browser_access"] == "off" + + def test_every_action_is_run_here(): """Runs last in this module: every action the parser knows was run above.""" assert RUN == set(cli.ACTIONS), f"never run: {set(cli.ACTIONS) - RUN}" diff --git a/packages/meshbay-cli/tests/test_e2e_local.py b/packages/meshbay-cli/tests/test_e2e_local.py index e2306d2..9cac392 100644 --- a/packages/meshbay-cli/tests/test_e2e_local.py +++ b/packages/meshbay-cli/tests/test_e2e_local.py @@ -185,7 +185,7 @@ def _build(root: Path, procs: list) -> dict: 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} + "share": share, "app": app, "run_node": run_node} async def _signed_in(account) -> None: @@ -743,3 +743,108 @@ def test_the_application_links_itself_from_its_group_page(world): assert s["minted"] == 1, "the identity linked is not the one kept" finally: chrome.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 + + # 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 diff --git a/packages/meshbay-cli/tests/test_manpage.py b/packages/meshbay-cli/tests/test_manpage.py index ebd71c8..4a43e5d 100644 --- a/packages/meshbay-cli/tests/test_manpage.py +++ b/packages/meshbay-cli/tests/test_manpage.py @@ -34,7 +34,7 @@ def test_every_action_and_option_is_in_its_section(): sub = next(a for a in parser._actions if isinstance(a, argparse._SubParsersAction)) assert list(sub.choices) == list(cli.ACTIONS) for name, sp in sub.choices.items(): - own = _section(page, f".SS {name}") + own = _section(page, f".SS {manpage._roff(name)}") for a in sp._actions: if isinstance(a, argparse._HelpAction): continue diff --git a/packages/meshbay-cli/tests/test_membership.py b/packages/meshbay-cli/tests/test_membership.py new file mode 100644 index 0000000..30fe680 --- /dev/null +++ b/packages/meshbay-cli/tests/test_membership.py @@ -0,0 +1,78 @@ +""" +The pieces `join`, `devices` and `browser-access` stand on, held to the +application's own rules: the invitation link's shape (`static/invite-link.js`), +how a typed code is folded (`crypto.js` `normalizeCode`), and which bundle on a +node is current enough to be left alone. +""" + +import base64 + +import pytest +from meshbay_cli.client.devices import find_device, normalize_code +from meshbay_cli.client.invite import parse_link +from meshbay_cli.client.keys import bundle_is_current, open_bundle, seal_bundle +from meshbay_cli.errors import NOT_FOUND, USAGE, CliError + +GROUP = "0f8fad5b-d9cb-469f-a165-70867728950e" +TICKET = "AbCdEfGhIjKlMnOpQr-_12" +NODE_URLSAFE = base64.urlsafe_b64encode(bytes(range(250, 256)) * 5 + b"\xfe\xff").decode() + + +def _link(**over) -> str: + q = {"v": "1", "g": GROUP, "t": TICKET, "n": NODE_URLSAFE.rstrip("="), "c": "k7p2-9wqx", + **over} + return "https://hub.test/#/invite?" + "&".join(f"{k}={v}" for k, v in q.items()) + + +def test_a_link_gives_its_hub_group_ticket_node_and_code(): + link = parse_link(_link()) + assert (link.origin, link.group_id, link.ticket, link.code) == ( + "https://hub.test", GROUP, TICKET, "K7P2-9WQX") + # The node key as the hub lists nodes: standard base64, padded. + assert base64.b64decode(link.node_pk) == base64.urlsafe_b64decode(NODE_URLSAFE) + assert "+" in link.node_pk or "/" in link.node_pk + # Only the fragment, as somebody may paste it. + assert parse_link(_link().split("#", 1)[1]).origin == "" + + +@pytest.mark.parametrize("bad", [{"v": "2"}, {"g": "not-a-group"}, {"t": "short"}, + {"n": "x" * 42}, {"c": "K7P2-9WQ"}, {"c": "K7P2+9WQX"}]) +def test_a_malformed_link_is_no_link(bad): + assert parse_link(_link(**bad)) is None + + +def test_text_without_a_link_is_no_link(): + assert parse_link("Family@alice") is None + + +def test_a_code_is_folded_as_the_application_folds_it(): + assert normalize_code("k7p2-9wqx") == "K7P29WQX" + assert normalize_code(" ilou 0O1V ") == "110V001V" + + +def test_a_device_is_named_by_an_unambiguous_prefix(): + devices = [{"pk_ed25519": "ABCDEFGH1234"}, {"pk_ed25519": "ABCDEFGH5678"}] + assert find_device(devices, "ABCDEFGH5")["pk_ed25519"] == "ABCDEFGH5678" + with pytest.raises(CliError) as e: + find_device(devices, "ABCDEFGH") + assert e.value.code == USAGE + with pytest.raises(CliError) as e: + find_device(devices, "ABCD") + assert e.value.code == USAGE + with pytest.raises(CliError) as e: + find_device(devices, "ZZZZZZZZ") + assert e.value.code == NOT_FOUND + + +def test_a_bundle_sealed_here_opens_and_is_current_only_under_its_own_key(): + keys = {"skEd": base64.b64encode(b"ed").decode(), "skX": base64.b64encode(b"x").decode()} + m, other = b"\x01" * 32, b"\x02" * 32 + sealed = seal_bundle(keys, master=m, pepper_version=1, user_id="u", node_pk="pk") + assert base64.b64decode(sealed)[:5] == b"MBK3\x01" + assert open_bundle(sealed, master=m, legacy=None, user_id="u", node_pk="pk") == keys + assert bundle_is_current(sealed, master=m, user_id="u", node_pk="pk") + # Sealed under an earlier passphrase, or for another node: sealed again. + assert not bundle_is_current(sealed, master=other, user_id="u", node_pk="pk") + assert not bundle_is_current(sealed, master=m, user_id="u", node_pk="pk2") + legacy = base64.b64encode(b"MBK2" + b"\x00" * 40).decode() + assert not bundle_is_current(legacy, master=m, user_id="u", node_pk="pk") diff --git a/packages/meshbay-cli/tests/test_store_and_session.py b/packages/meshbay-cli/tests/test_store_and_session.py index 42458e2..07ba77e 100644 --- a/packages/meshbay-cli/tests/test_store_and_session.py +++ b/packages/meshbay-cli/tests/test_store_and_session.py @@ -244,7 +244,7 @@ async def test_a_machine_waiting_for_approval_retries_until_it_is_approved(monke monkeypatch.setattr(groups, "LINK_POLL", 0) tries, shown = [], [] - async def connect(account, group_id, node): + async def connect(account, group_id, node, join_code=None): tries.append(node["node_id"]) if len(tries) < 3: raise DeviceLinkPending("ABCD-EFGH", time.time() + 600) @@ -264,7 +264,7 @@ async def test_without_a_terminal_the_code_is_shown_and_the_run_ends(monkeypatch from meshbay_cli.client import groups from meshbay_cli.client.session import DeviceLinkPending - async def connect(account, group_id, node): + async def connect(account, group_id, node, join_code=None): raise DeviceLinkPending("ABCD-EFGH", time.time() + 600) monkeypatch.setattr(groups, "connect", connect) |