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

"""
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