""" The account's keys on the nodes across a new passphrase, and what `account passphrase` does in which order: the copies first, the hub only then, the copies set back if the hub refuses — never an account whose nodes and hub disagree on the passphrase. """ import asyncio import base64 import types import pytest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey from fake_hub import PASSPHRASE, USER_ID, USERNAME, FakeHub from meshbay_cli import __main__ as cli from meshbay_cli.client import rewrap, session from meshbay_cli.client.account import Account from meshbay_cli.client.hub import HubError from meshbay_cli.client.keys import ( Identity, b64, bundle_master_key, opens_under, passphrase_half, seal_bundle, seal_recovery, unb64, ) from meshbay_cli.client.store import Store from meshbay_cli.commands import account as account_cmd from meshbay_cli.commands import context NODE = base64.b64encode(b"\x05" * 32).decode() M_OLD, M_NEW, M_ELSEWHERE = b"\x01" * 32, b"\x02" * 32, b"\x03" * 32 def _keys() -> dict: return Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate()).to_pkcs8() def _sealed(keys: dict, master: bytes) -> str: return seal_bundle(keys, master=master, pepper_version=1, user_id=USER_ID, node_pk=NODE) class Ch: def __init__(self): self.sent: list[dict] = [] async def request(self, msg, *expect, **kw): self.sent.append(msg) return {"type": "ack"} def _account(master: bytes, previous=(), setting=None) -> Account: acc = Account(None, USERNAME, lambda: PASSPHRASE) acc.state = {"user_id": USER_ID, **({"browser_access": setting} if setting else {})} acc.secrets = {"master": b64(master), "previous_masters": [b64(m) for m in previous]} return acc # ── What a connection leaves on the node ───────────────────────────────────── async def test_a_copy_under_a_passphrase_changed_here_is_carried_same_keys(): browsers, mine = _keys(), Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate()) ch = Ch() await session._settle_bundle(_account(M_NEW, [M_OLD]), ch, mine, NODE, _sealed(browsers, M_OLD)) assert len(ch.sent) == 1 assert opens_under(ch.sent[0]["bundle_enc"], M_NEW, user_id=USER_ID, node_pk=NODE) == browsers async def test_a_copy_under_a_passphrase_set_elsewhere_is_left_alone(caplog): """Rewriting it under this machine's older `M` would lock the browsers out.""" ch = Ch() mine = Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate()) await session._settle_bundle(_account(M_OLD, setting="on"), ch, mine, NODE, _sealed(_keys(), M_ELSEWHERE)) assert ch.sent == [] assert "does not know" in caplog.text @pytest.mark.parametrize("setting, bundle, expect", [ ("on", "", "keypair_bundle_store"), ("off", "x", "keypair_bundle_delete"), (None, "", None), ]) async def test_browser_access_still_decides_the_rest(setting, bundle, expect): ch = Ch() mine = Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate()) if bundle: bundle = _sealed(_keys(), M_NEW) await session._settle_bundle(_account(M_NEW, setting=setting), ch, mine, NODE, bundle) assert [m["type"] for m in ch.sent] == ([expect] if expect else []) # ── Carrying the copies ────────────────────────────────────────────────────── class FakeSession: def __init__(self, held: dict, identity: Identity | None = None): self.held, self.node_pk, self.ch = held, NODE, Ch() self.identity = identity or Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate()) async def close(self): pass def _hub(nodes_by_group: dict): async def group_nodes(token, gid): return nodes_by_group[gid] return types.SimpleNamespace(group_nodes=group_nodes) def _carry_world(monkeypatch, acc: Account, nodes_by_group: dict, sessions: dict): acc.hub = _hub(nodes_by_group) async def call(fn, *args): return await fn("t", *args) monkeypatch.setattr(acc, "call", call) opened = [] async def connect(account, gid, node, join_code=None, *, rewrap=None): opened.append(node["pk_node"]) s = sessions[node["pk_node"]] if isinstance(s, Exception): raise s return s monkeypatch.setattr(rewrap, "connect", connect) return opened async def test_each_copy_is_sealed_again_once_per_node(monkeypatch): browsers = _keys() acc = _account(M_OLD) s = FakeSession({"bundle_enc": _sealed(browsers, M_OLD)}) opened = _carry_world(monkeypatch, acc, { "g1": [{"pk_node": "n1"}], "g2": [{"pk_node": "n1"}], "g3": [], "g4": [{"pk_node": "n2"}]}, {"n1": s, "n2": session.NothingHere()}) groups = [{"id": g, "name": g, "owner_username": "bob"} for g in ("g1", "g2", "g3", "g4")] out = await rewrap.carry(acc, groups, master=M_NEW, pepper_version=1) assert opened == ["n1", "n2"] # n1 once, for two groups assert out.done == ["n1"] and out.unreachable == ["g3@bob"] and out.nothing == ["g4@bob"] sent = s.ch.sent[0] assert opens_under(sent["bundle_enc"], M_NEW, user_id=USER_ID, node_pk=NODE) == browsers assert "bundle_enc_recovery" not in sent # the node keeps the one it has async def test_after_a_reset_the_recovery_copy_gives_the_keys_back(monkeypatch): browsers, rkey = _keys(), b"\x09" * 32 acc = _account(M_NEW) # the reset's: opens nothing old s = FakeSession({"bundle_enc": _sealed(browsers, M_OLD), "bundle_enc_recovery": seal_recovery(browsers, rkey=rkey, user_id=USER_ID, node_pk=NODE)}) _carry_world(monkeypatch, acc, {"g1": [{"pk_node": "n1"}]}, {"n1": s}) out = await rewrap.carry(acc, [{"id": "g1", "name": "g1"}], master=M_NEW, pepper_version=1, recovery=rkey) assert out.done == ["n1"] sent = s.ch.sent[0] assert opens_under(sent["bundle_enc"], M_NEW, user_id=USER_ID, node_pk=NODE) == browsers assert sent["bundle_enc_recovery"] async def test_a_copy_nothing_opens_is_left_as_it_is_unless_this_machine_leaves_its_own( monkeypatch): acc = _account(M_NEW) held = {"bundle_enc": _sealed(_keys(), M_OLD)} s = FakeSession(dict(held)) _carry_world(monkeypatch, acc, {"g1": [{"pk_node": "n1"}]}, {"n1": s}) out = await rewrap.carry(acc, [{"id": "g1", "name": "g1"}], master=M_NEW, pepper_version=1) assert out.done == [] and "left as it is" in out.failed[0] and s.ch.sent == [] s = FakeSession(dict(held)) _carry_world(monkeypatch, acc, {"g1": [{"pk_node": "n1"}]}, {"n1": s}) out = await rewrap.carry(acc, [{"id": "g1", "name": "g1"}], master=M_NEW, pepper_version=1, own_keys=True) assert out.done == ["n1"] assert opens_under(s.ch.sent[0]["bundle_enc"], M_NEW, user_id=USER_ID, node_pk=NODE) == s.identity.to_pkcs8() # ── `account passphrase`: the order of things ──────────────────────────────── NEW = "a new passphrase that is long enough" @pytest.fixture def hub(monkeypatch): fake = FakeHub() calls: list[str] = [] fake.calls = calls fake.refuse_change = False async def bundle_pepper(token, ak): fake._auth(token) if ak != fake.ak: raise HubError(403, "Passphrase does not match") calls.append("pepper") return {"bundle_pepper": base64.b64encode(b"\x07" * 32).decode(), "bundle_pepper_version": 1} async def change_passphrase(token, old, new): fake._auth(token) calls.append("hub") if fake.refuse_change: raise HubError(429, "Too many requests") fake.ak = new return fake._session() async def my_groups(token): return [{"id": "g1", "name": "Family", "owner_username": USERNAME}] fake.bundle_pepper, fake.change_passphrase, fake.my_groups = ( bundle_pepper, change_passphrase, my_groups) monkeypatch.setattr(context, "Hub", lambda url: fake) return fake def _master(passphrase: str) -> bytes: return bundle_master_key(passphrase_half(USERNAME, passphrase), b"\x07" * 32, USER_ID) @pytest.fixture def carried(monkeypatch, hub): seen = [] async def carry(account, groups, *, master, pepper_version, only=None, **kw): hub.calls.append("carry") seen.append({"opens_with": account.master, "seals_with": master, "only": only}) return rewrap.Carried(done=["n1"]) monkeypatch.setattr(account_cmd, "carry", carry) return seen def _login(): assert cli.main(["--hub", "https://hub.test", "-u", USERNAME, "--password", PASSPHRASE, "login"]) == 0 async def _stored_secrets() -> dict: store = Store() store.load() return store.secrets def test_the_copies_move_first_and_the_hub_only_then(hub, carried, monkeypatch, capsys): _login() monkeypatch.setenv("MESHBAY_NEW_PASSWORD", NEW) assert cli.main(["--password", PASSPHRASE, "account", "passphrase"]) == 0 assert hub.calls == ["pepper", "carry", "hub"] assert carried[0]["opens_with"] == _master(PASSPHRASE) assert carried[0]["seals_with"] == _master(NEW) secrets = asyncio.run(_stored_secrets()) assert unb64(secrets["master"]) == _master(NEW) assert [unb64(m) for m in secrets["previous_masters"]] == [_master(PASSPHRASE)] assert "sealed again on 1 node" in capsys.readouterr().out def test_a_refused_change_sets_the_copies_back(hub, carried, monkeypatch, capsys): _login() hub.refuse_change = True monkeypatch.setenv("MESHBAY_NEW_PASSWORD", NEW) assert cli.main(["--password", PASSPHRASE, "account", "passphrase"]) == 8 assert hub.calls == ["pepper", "carry", "hub", "carry"] back = carried[1] assert back["opens_with"] == _master(NEW) and back["seals_with"] == _master(PASSPHRASE) assert back["only"] == {"n1"} assert unb64(asyncio.run(_stored_secrets())["master"]) == _master(PASSPHRASE) assert "nothing was changed" in capsys.readouterr().err @pytest.mark.parametrize("old, new, says", [ ("wrong passphrase entirely", NEW, "does not match"), (PASSPHRASE, "too short", "12 characters"), (PASSPHRASE, PASSPHRASE, "is the current one"), ]) def test_nothing_moves_on_a_wrong_or_weak_passphrase(hub, carried, monkeypatch, capsys, old, new, says): _login() monkeypatch.setenv("MESHBAY_NEW_PASSWORD", new) assert cli.main(["--password", old, "account", "passphrase"]) != 0 assert "carry" not in hub.calls and "hub" not in hub.calls assert says in capsys.readouterr().err