""" 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, join_code=None): 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, join_code=None): 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"