diff options
Diffstat (limited to 'packages/meshbay-cli/tests')
| -rw-r--r-- | packages/meshbay-cli/tests/fake_node.py | 51 | ||||
| -rw-r--r-- | packages/meshbay-cli/tests/test_e2e_local.py | 51 | ||||
| -rw-r--r-- | packages/meshbay-cli/tests/test_interactive.py | 79 | ||||
| -rw-r--r-- | packages/meshbay-cli/tests/test_roster.py | 100 |
4 files changed, 277 insertions, 4 deletions
diff --git a/packages/meshbay-cli/tests/fake_node.py b/packages/meshbay-cli/tests/fake_node.py index f5fe631..8f4356b 100644 --- a/packages/meshbay-cli/tests/fake_node.py +++ b/packages/meshbay-cli/tests/fake_node.py @@ -32,6 +32,7 @@ from meshbay_common.device import ( from meshbay_common.groupbox import ( PURPOSE_CHAT_KEYS, PURPOSE_INDEX, + PURPOSE_ROSTER, PURPOSE_UPLOAD, seal, unseal, @@ -69,6 +70,12 @@ class FakeNode: self.ack: dict = {"chat_epoch": 1} # the sealed configuration, opened self.chat_epochs = {1: secrets.token_bytes(32)} self.chat: list[dict] = [] # stored rows, oldest first + # group_roster_resp's devices; this machine's own key is a root of its account. + self.roster: list[dict] = [{ + "user_id": USER_ID, "username": "alice-test", "pk_ed25519": self.identity.pk_ed, + "pk_x25519": self.identity.pk_x, "added_by_pk": "", "add_sig": "", + "add_nonce": "", "add_ts": 0}] + self.members: dict[str, Identity] = {} # sender → the device it signs with self.admin_ops: dict[str, dict] = {} def add(self, path: str, name: str, data: bytes, uploader_id: str = "") -> dict: @@ -86,17 +93,47 @@ class FakeNode: return GroupSession(None, FakeChannel(self), GROUP_ID, "node-1", NODE_PK, self.gek, dict(self.ack), self.identity, self.nonce, self.role) + def list_device(self, sender: str, device: Identity, *, + countersigned_by: Identity | None = None, proof: bool = True) -> None: + """A roster row for `sender`'s device: a root, or countersigned by another + of its devices — with the signature, or (`proof=False`) naming the + countersigner and carrying nothing, as a lying node would.""" + row = {"user_id": "u-" + sender, "username": sender, "pk_ed25519": device.pk_ed, + "pk_x25519": device.pk_x, "added_by_pk": "", "add_sig": "", + "add_nonce": "", "add_ts": 0} + if countersigned_by is not None: + row["added_by_pk"] = countersigned_by.pk_ed + if proof: + nonce, ts = secrets.token_bytes(32), 1_700_000_000 + row.update(add_nonce=base64.b64encode(nonce).decode(), add_ts=ts, + add_sig=base64.b64encode(countersigned_by.sk_ed.sign( + device_add_transcript( + node_pk_b64=NODE_PK, user_id="u-" + sender, + pk_ed25519_b64=device.pk_ed, + pk_x25519_b64=device.pk_x, + nonce_node=nonce, ts=ts))).decode()) + self.roster.append(row) + def say(self, text: str, *, sender: str = "bob-test", epoch: int = 1, - forge: bool = False) -> None: - """Another member's message, sealed and signed as a client does; `forge` - signs it with a key other than the device it names.""" - other = Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate()) + forge: bool = False, device: Identity | None = None) -> Identity: + """Another member's message, sealed and signed as a client does, from + `device` (default: the sender's usual one, listed in the roster as a + root the first time); `forge` signs it with a key other than the device + it names.""" + if device is None: + if sender not in self.members: + self.members[sender] = Identity(Ed25519PrivateKey.generate(), + X25519PrivateKey.generate()) + self.list_device(sender, self.members[sender]) + device = self.members[sender] + other = device sealed = chatbox.seal(self.chat_epochs[epoch], GROUP_ID, epoch, other.pk_ed, base64.b64decode(other.pk_ed), other.sk_ed, {"text": text, "sender_name": sender}) if forge: sealed["sig"] = Ed25519PrivateKey.generate().sign(b"something else") self._store(base64.b64decode(other.pk_ed), epoch, sealed, "u-" + sender) + return other def _store(self, device: bytes, epoch: int, sealed: dict, sender_id: str) -> None: self.chat.append({"id": len(self.chat) + 1, "sender_id": sender_id, @@ -236,6 +273,12 @@ class FakeChannel: return {"type": "chat_keys_resp", "group_id": GROUP_ID, **seal(n.gek, PURPOSE_CHAT_KEYS, "chat_keys_resp", GROUP_ID, payload)} + def _group_roster_req(self, msg): + n = self.node + return {"type": "group_roster_resp", "group_id": GROUP_ID, + **seal(n.gek, PURPOSE_ROSTER, "group_roster_resp", GROUP_ID, + {"devices": n.roster, "node_pk": NODE_PK})} + def _chat_hist(self, msg): rows = self.node.chat if msg.get("before"): diff --git a/packages/meshbay-cli/tests/test_e2e_local.py b/packages/meshbay-cli/tests/test_e2e_local.py index 19f5ca9..cf6a3b9 100644 --- a/packages/meshbay-cli/tests/test_e2e_local.py +++ b/packages/meshbay-cli/tests/test_e2e_local.py @@ -1017,3 +1017,54 @@ def test_chat_between_the_command_line_and_the_application(world): 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 diff --git a/packages/meshbay-cli/tests/test_interactive.py b/packages/meshbay-cli/tests/test_interactive.py index 49180d5..fa7e96c 100644 --- a/packages/meshbay-cli/tests/test_interactive.py +++ b/packages/meshbay-cli/tests/test_interactive.py @@ -206,3 +206,82 @@ async def test_delete_removes_what_is_yours_and_asks_first(world): await pilot.press("delete") await pilot.pause(0.2) assert any("select them on the right" in n.message for n in app._notifications) + + +def _log_text(screen) -> str: + from textual.widgets import RichLog + return "\n".join("".join(seg.text for seg in strip) + for strip in screen.query_one(RichLog).lines) + + +async def test_the_interactive_chat_joins_talks_leaves_and_comes_back(world, monkeypatch): + from meshbay_cli.tui.chat import ChatGroupsScreen, ChatRoomScreen + from textual.widgets import DataTable, Input + hub, node, _ = world + hub.groups.append({"id": "g9", "name": "Quiet", "owner_username": "bob-test", + "node_online": True, "is_admin": False}) + quiet = FakeNode() + quiet.ack["enabled_apps"] = ["files"] + sessions = {} + + async def open_group(account, group): + target = quiet if group["id"] == "g9" else node + sessions[group["id"]] = target.session() + return sessions[group["id"]] + monkeypatch.setattr(pool, "open_group", open_group) + node.say("earlier, from bob") + assert await asyncio.to_thread(cli.main, ["--hub", "https://hub.test", "-u", USERNAME, + "--password", PASSPHRASE, "login"]) == 0 + + app = MeshBayApp(context.Context(cli.build_parser().parse_args( + ["--hub", "https://hub.test"])), ChatGroupsScreen) + async with app.run_test(size=(120, 30)) as pilot: + await _until(pilot, lambda: isinstance(app.screen, ChatGroupsScreen)) + table = app.screen.query_one(DataTable) + await _until(pilot, lambda: table.row_count == 2) + + # A group whose chat is off is not entered. + table.move_cursor(row=1) # Quiet@bob-test + await pilot.press("enter") + await _until(pilot, lambda: any("chat turned off" in n.message + for n in app._notifications)) + assert isinstance(app.screen, ChatGroupsScreen) + + table.move_cursor(row=0) # Family@alice-test + await pilot.press("enter") + await _until(pilot, lambda: isinstance(app.screen, ChatRoomScreen)) + room = app.screen + await _until(pilot, lambda: "<bob-test> earlier, from bob" in _log_text(room)) + + # Posting: sealed, signed, stored on the node, shown here at once. + room.query_one(Input).value = "hello from the room" + await pilot.press("enter") + await _until(pilot, lambda: "<alice-test> hello from the room" in _log_text(room)) + assert len(node.chat) == 2 + + # What the node pushes appears as it arrives. + bob = node.say("live, from bob") + sessions["g1"].ch._emit({**node.chat[-1], "type": "chat_msg"}) + await _until(pilot, lambda: "<bob-test> live, from bob" in _log_text(room)) + assert bob.pk_ed + + # /part goes back; entering again shows the conversation once, and a + # pushed message once — the room left behind no handler. + room.query_one(Input).value = "/part" + await pilot.press("enter") + await _until(pilot, lambda: isinstance(app.screen, ChatGroupsScreen)) + await pilot.press("enter") + await _until(pilot, lambda: isinstance(app.screen, ChatRoomScreen)) + room = app.screen + await _until(pilot, lambda: "live, from bob" in _log_text(room)) + node.say("once") + sessions["g1"].ch._emit({**node.chat[-1], "type": "chat_msg"}) + await _until(pilot, lambda: "<bob-test> once" in _log_text(room)) + await pilot.pause(0.2) + assert _log_text(room).count("<bob-test> once") == 1 + + # Esc goes back too; q quits from the list. + await pilot.press("escape") + await _until(pilot, lambda: isinstance(app.screen, ChatGroupsScreen)) + await pilot.press("q") + assert app.return_code in (0, None) diff --git a/packages/meshbay-cli/tests/test_roster.py b/packages/meshbay-cli/tests/test_roster.py new file mode 100644 index 0000000..cdaf90b --- /dev/null +++ b/packages/meshbay-cli/tests/test_roster.py @@ -0,0 +1,100 @@ +""" +Tier 2 for chat (docs/MESHBAY_DESIGN.md §3.3): whose device signed a message, +checked against the group's roster and the keys this machine accepted before — +through `main()`, against the in-memory node, with real countersignatures. +""" + +import json + +import pytest +from fake_hub import PASSPHRASE, USERNAME, FakeHub +from fake_node import FakeNode +from meshbay_cli import __main__ as cli +from meshbay_cli.client.keys import Identity +from meshbay_cli.client.roster import RosterCheck, judge, verify_roster +from meshbay_cli.commands import chat, context + + +def _new() -> Identity: + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey + return Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate()) + + +@pytest.fixture +def node(monkeypatch, capsys): + hub, fake = FakeHub(), FakeNode() + monkeypatch.setattr(context, "Hub", lambda url: hub) + + async def open_group(account, group): + return fake.session() + monkeypatch.setattr(chat, "open_group", open_group) + assert cli.main(["--hub", hub.url, "-u", USERNAME, "--password", PASSPHRASE, + "login"]) == 0 + capsys.readouterr() + return fake + + +def _last(capsys, n=1) -> list[dict]: + assert cli.main(["--json", "chat", f"-{n}", "Family"]) == 0 + return json.loads(capsys.readouterr().out) + + +def test_each_verdict_on_a_message_signing_device(node, capsys): + phone = node.say("from bob's phone") + assert _last(capsys)[0]["trust"] == "first" # first sight: pinned + assert _last(capsys)[0]["trust"] == "pinned" + + # A second device, countersigned by the first: no notice. + laptop = _new() + node.list_device("bob-test", laptop, countersigned_by=phone) + node.say("from bob's laptop", device=laptop) + assert _last(capsys)[0]["trust"] == "linked" + + # Named as countersigned and carrying no proof, as a lying node would list it. + planted = _new() + node.list_device("bob-test", planted, countersigned_by=phone, proof=False) + node.say("hello, it's bob", device=planted) + assert _last(capsys)[0]["trust"] == "changed" + # Not in the roster at all. + node.say("hello again", device=_new()) + assert _last(capsys)[0]["trust"] == "changed" + + # Said in plain text under the message, and nowhere else. + assert cli.main(["chat", "-5", "Family"]) == 0 + out = capsys.readouterr().out.splitlines() + assert sum("key you have not seen before" in line for line in out) == 2 + assert out[-1].lstrip().startswith("! This account") + assert out[-3].lstrip().startswith("! This account") + + # Another account, first sight. + node.say("hi from carol", sender="carol-test") + assert _last(capsys)[0]["trust"] == "first" + + +async def test_a_device_listed_after_the_roster_was_read_is_not_a_substitution(node): + """The roster is read once per connection; a member's new device added since + is read again before "changed" is said.""" + from meshbay_cli.client.account import Account + from meshbay_cli.client.store import Store + store = Store() + account = Account.open(context.Hub("x"), None, lambda: PASSPHRASE, store) + session = node.session() + phone = node.say("one") + check = RosterCheck(account, session) + assert await check.status("u-bob-test", phone.pk_ed) == "first" + tablet = _new() + node.list_device("bob-test", tablet, countersigned_by=phone) # after the read + assert await check.status("u-bob-test", tablet.pk_ed) == "linked" + + +def test_a_root_is_a_device_naming_no_countersigner(): + a, b = _new(), _new() + rows = [{"user_id": "u", "pk_ed25519": a.pk_ed, "pk_x25519": a.pk_x, + "added_by_pk": ""}, + {"user_id": "u", "pk_ed25519": b.pk_ed, "pk_x25519": b.pk_x, + "added_by_pk": a.pk_ed, "add_sig": "", "add_nonce": "", "add_ts": 0}] + got = verify_roster({"devices": rows, "node_pk": "pk"}, "pk")["u"] + assert got.verified == [a.pk_ed] and got.unevidenced == [b.pk_ed] + assert judge([a.pk_ed], got, b.pk_ed) == ("changed", None) + assert judge(None, got, b.pk_ed)[0] == "first" |