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

"""
The local store, and the hub session it keeps.

The case that matters most is two runs renewing at once: the hub rotates refresh
tokens and treats a spent one as theft, so a second renewal with the token the
first has just spent would sign the account out everywhere.
"""

import asyncio
import os
import sys
import time

import keyring
import pytest
from fake_hub import PASSPHRASE, USERNAME, FakeHub
from keyring.backend import KeyringBackend
from meshbay_cli.client.account import Account, no_passphrase
from meshbay_cli.client.store import KEY_FILE, KEY_OS, Store
from meshbay_cli.errors import AUTH, CliError


class MemoryKeyring(KeyringBackend):
    priority = 1

    def __init__(self):
        super().__init__()
        self.items: dict[tuple[str, str], str] = {}

    def get_password(self, service, username):
        return self.items.get((service, username))

    def set_password(self, service, username, password):
        self.items[(service, username)] = password

    def delete_password(self, service, username):
        self.items.pop((service, username), None)


@pytest.fixture
def memory_keyring():
    backend = MemoryKeyring()
    previous = keyring.get_keyring()
    keyring.set_keyring(backend)
    yield backend
    keyring.set_keyring(previous)


def _saved(root, secrets=None, state=None) -> Store:
    store = Store(root)
    store.state = {"hub": "https://hub.test", "username": USERNAME, "user_id": "u",
                   **(state or {})}
    store.secrets = {"access_token": "old", "access_exp": 0, "refresh_token": "rt0",
                     **(secrets or {})}
    store.save(new=True)
    return store


# ── The store ────────────────────────────────────────────────────────────────

def test_without_key_storage_the_key_is_a_private_file(tmp_path):
    store = _saved(tmp_path / "s")
    assert store.state["key_store"] == KEY_FILE
    if sys.platform != "win32":
        for name in ("store.key", "secrets.enc", "state.json"):
            assert os.stat(tmp_path / "s" / name).st_mode & 0o077 == 0, name
    again = Store(tmp_path / "s")
    assert again.load() and again.secrets["refresh_token"] == "rt0"


def test_with_key_storage_the_key_is_kept_there(tmp_path, memory_keyring):
    a = _saved(tmp_path / "a")
    _saved(tmp_path / "b")
    assert a.state["key_store"] == KEY_OS and not (tmp_path / "a" / "store.key").exists()
    # One entry per store: a second store never overwrites the first's key.
    assert len(memory_keyring.items) == 2
    assert Store(tmp_path / "a").load() and Store(tmp_path / "b").load()
    a.erase()
    assert len(memory_keyring.items) == 1
    assert not Store(tmp_path / "a").exists()


def test_secrets_are_not_readable_in_the_file(tmp_path):
    _saved(tmp_path / "s", secrets={"refresh_token": "a-very-recognisable-token"})
    assert b"a-very-recognisable-token" not in (tmp_path / "s" / "secrets.enc").read_bytes()


def test_a_store_whose_key_storage_vanished_says_so(tmp_path, memory_keyring):
    _saved(tmp_path / "s")
    keyring.set_keyring(__import__("keyring.backends.fail", fromlist=["Keyring"]).Keyring())
    with pytest.raises(CliError) as e:
        Store(tmp_path / "s").load()
    assert e.value.code == AUTH and "OS key storage" in e.value.message


def test_peek_reads_the_state_without_the_key(tmp_path):
    _saved(tmp_path / "s")
    (tmp_path / "s" / "store.key").unlink()
    assert Store(tmp_path / "s").peek()["username"] == USERNAME


# ── Renewal ──────────────────────────────────────────────────────────────────

def _open(root, hub) -> Account:
    return Account.open(hub, None, no_passphrase, Store(root))


async def test_two_runs_renewing_at_once_spend_the_token_once(tmp_path):
    hub = FakeHub()
    first = hub._session()
    _saved(tmp_path / "s", secrets={"refresh_token": first["refresh_token"]})
    hub.refresh_delay = 0.2      # wide enough for the second run to arrive meanwhile

    a, b = _open(tmp_path / "s", hub), _open(tmp_path / "s", hub)
    ta, tb = await asyncio.gather(a.token(), b.token())

    assert not hub.reuse_detected
    assert hub.refresh_calls == 1
    assert ta == tb and ta in hub.access
    # And the stored token is the live one: a third run renews without incident.
    c = _open(tmp_path / "s", hub)
    c.secrets["access_exp"] = 0
    await c.token(rejected=c.secrets["access_token"])
    assert not hub.reuse_detected


async def test_a_token_still_fresh_is_used_as_it_is(tmp_path):
    hub = FakeHub()
    s = hub._session()
    _saved(tmp_path / "s", secrets={"access_token": s["access_token"],
                                    "access_exp": time.time() + 3600,
                                    "refresh_token": s["refresh_token"]})
    assert await _open(tmp_path / "s", hub).token() == s["access_token"]
    assert hub.refresh_calls == 0


async def test_a_refused_refresh_falls_back_to_the_device_key(tmp_path):
    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
    from meshbay_cli.client.keys import b64
    hub = FakeHub()
    sk = Ed25519PrivateKey.generate()
    hub.devices["dev-1"] = b64(sk.public_key().public_bytes_raw())
    _saved(tmp_path / "s", secrets={"refresh_token": "revoked",
                                    "device_sk": b64(sk.private_bytes_raw())})
    token = await _open(tmp_path / "s", hub).token()
    assert token in hub.access


async def test_a_session_that_ended_asks_for_login(tmp_path):
    hub = FakeHub()
    _saved(tmp_path / "s", secrets={"refresh_token": "revoked"})
    with pytest.raises(CliError) as e:
        await _open(tmp_path / "s", hub).token()
    assert e.value.code == AUTH and "meshbay-cli login" in e.value.message


async def test_a_401_renews_once_and_replays(tmp_path):
    hub = FakeHub()
    s = hub._session()
    _saved(tmp_path / "s", secrets={"access_token": "revoked-by-the-hub",
                                    "access_exp": time.time() + 3600,
                                    "refresh_token": s["refresh_token"]})
    groups = await _open(tmp_path / "s", hub).call(hub.my_groups)
    assert groups[0]["name"] == "Family" and hub.refresh_calls == 1


# ── Signing in ───────────────────────────────────────────────────────────────

async def test_sign_in_derives_the_keys_and_registers_one_device(tmp_path):
    hub = FakeHub()
    acc = Account(hub, USERNAME, lambda: PASSPHRASE)
    await acc.sign_in(remember=True)
    assert acc.master and acc.legacy and acc.user_id
    assert len(hub.devices) == 1 and acc.state["device_id"] == "dev-1"
    # Signing in again keeps the same device key rather than adding one.
    await acc.sign_in(remember=True)
    assert len(hub.devices) == 1


async def test_a_wrong_passphrase_is_an_auth_error():
    with pytest.raises(CliError) as e:
        await Account(FakeHub(), USERNAME, lambda: "wrong").sign_in(remember=False)
    assert e.value.code == AUTH


async def test_a_renewal_by_passphrase_is_what_gets_stored(tmp_path):
    """The fallback replaces the session wholesale; the store must get the new one."""
    hub = FakeHub()
    _saved(tmp_path / "s", secrets={"refresh_token": "revoked"})
    acc = Account.open(hub, None, lambda: PASSPHRASE, Store(tmp_path / "s"))
    await acc.token()
    again = Store(tmp_path / "s")
    again.load()
    assert again.secrets["refresh_token"] == acc.secrets["refresh_token"]
    assert again.secrets["refresh_token"] in hub.refresh_tokens
    assert again.secrets.get("master")


async def test_a_run_signed_out_meanwhile_says_so(tmp_path):
    hub = FakeHub()
    _saved(tmp_path / "s")
    acc = _open(tmp_path / "s", hub)
    Store(tmp_path / "s").erase()
    with pytest.raises(CliError) as e:
        await acc.token()
    assert e.value.code == AUTH and "another run" in e.value.message


async def test_a_pending_device_request_is_kept_then_adopted(tmp_path):
    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
    from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
    from meshbay_cli.client.keys import Identity
    hub = FakeHub()
    _saved(tmp_path / "s")
    acc = _open(tmp_path / "s", hub)
    ident = Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate())
    await acc.keep_pending("node-pk", ident, "ABCD-EFGH", time.time() + 600)

    # A later run shows the same code and joins with the same keys.
    again = _open(tmp_path / "s", hub)
    pending = again.pending_identity("node-pk")
    assert pending.code == "ABCD-EFGH" and pending.identity.pk_ed == ident.pk_ed
    assert again.identity("node-pk") is None
    await again.adopt_pending("node-pk")
    assert again.pending_identity("node-pk") is None
    assert _open(tmp_path / "s", hub).identity("node-pk").pk_ed == ident.pk_ed


async def test_an_expired_device_request_is_not_reused(tmp_path):
    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
    from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
    from meshbay_cli.client.keys import Identity
    hub = FakeHub()
    _saved(tmp_path / "s")
    acc = _open(tmp_path / "s", hub)
    ident = Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate())
    await acc.keep_pending("node-pk", ident, "ABCD-EFGH", time.time() - 1)
    assert _open(tmp_path / "s", hub).pending_identity("node-pk") is None


async def test_a_machine_waiting_for_approval_retries_until_it_is_approved(monkeypatch):
    from meshbay_cli.client import groups
    from meshbay_cli.client.session import DeviceLinkPending
    monkeypatch.setattr(groups, "LINK_POLL", 0)
    tries, shown = [], []

    async def connect(account, group_id, node):
        tries.append(node["node_id"])
        if len(tries) < 3:
            raise DeviceLinkPending("ABCD-EFGH", time.time() + 600)
        return "session"
    monkeypatch.setattr(groups, "connect", connect)

    class Acc:
        def on_device_code(self, group, pending, *, first):
            shown.append((pending.link_code, first))
            return True
    got = await groups._connect_linking(Acc(), {"id": "g"}, {"node_id": "n1"})
    assert got == "session" and tries == ["n1", "n1", "n1"]
    assert shown == [("ABCD-EFGH", True), ("ABCD-EFGH", False)]


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):
        raise DeviceLinkPending("ABCD-EFGH", time.time() + 600)
    monkeypatch.setattr(groups, "connect", connect)

    class Acc:
        def on_device_code(self, group, pending, *, first):
            return False
    with pytest.raises(DeviceLinkPending) as e:
        await groups._connect_linking(Acc(), {"id": "g"}, {"node_id": "n1"})
    assert e.value.code == 5 and e.value.link_code == "ABCD-EFGH"