diff options
Diffstat (limited to 'packages/meshbay-cli/src/meshbay_cli/client/account.py')
| -rw-r--r-- | packages/meshbay-cli/src/meshbay_cli/client/account.py | 344 |
1 files changed, 344 insertions, 0 deletions
diff --git a/packages/meshbay-cli/src/meshbay_cli/client/account.py b/packages/meshbay-cli/src/meshbay_cli/client/account.py new file mode 100644 index 0000000..d620f6d --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/client/account.py @@ -0,0 +1,344 @@ +""" +A signed-in account: its hub session, its bundle key `M`, its node identities. + +Two lives. **Remembered** (after `login`): everything is in the store, and a +run needs no passphrase — an expired access token is renewed with the refresh +token, a refused one falls back to the device's hub key. **Ephemeral** (no +store, or another account named on the command line): signed in with the +passphrase for this run only, nothing written, the session ended on the hub +when the run ends. + +The refresh token rotates and a spent one presented again revokes the whole +family, so every renewal happens under the store's lock, after re-reading what +another run may have just written. +""" + +import asyncio +import contextlib +import logging +import socket +import time +from collections.abc import AsyncIterator, Callable +from dataclasses import dataclass + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from ..errors import AUTH, CliError +from .hub import Hub, HubError +from .keys import ( + Identity, + auth_key, + b64, + bundle_master_key, + passphrase_half, + unb64, +) +from .store import KEY_FILE, Store + +log = logging.getLogger(__name__) + +# Renew this long before the access token runs out, so a request never leaves +# with a token that expires on the way. +RENEW_MARGIN = 300 + +PassphraseSource = Callable[[], str] + + +class NoPassphrase(CliError): + """Raised by a passphrase source that has none to give.""" + + def __init__(self, message: str): + super().__init__(AUTH, message) + + +def no_passphrase() -> str: + """The source for an account that must never ask: one being signed out.""" + raise NoPassphrase("No passphrase is asked for to sign out.") + + +@dataclass +class Pending: + identity: Identity + code: str + expires_at: float + + +class Account: + def __init__(self, hub: Hub, username: str, passphrase: PassphraseSource, + store: Store | None = None): + self.hub = hub + self.username = username + self._passphrase = passphrase + self.store = store # None: ephemeral + # Called with (group, DeviceLinkPending, first=) when this machine waits + # for a device approval; returns whether to keep waiting. + self.on_device_code = None + self.state: dict = {} + self.secrets: dict = {} + + @property + def remembered(self) -> bool: + return self.store is not None + + @property + def user_id(self) -> str: + return self.state["user_id"] + + @property + def master(self) -> bytes | None: + m = self.secrets.get("master") + return unb64(m) if m else None + + @property + def legacy(self) -> bytes | None: + a = self.secrets.get("legacy") + return unb64(a) if a else None + + # ── Opening ────────────────────────────────────────────────────────────── + + @classmethod + def open(cls, hub: Hub, username: str | None, passphrase: PassphraseSource, + store: Store) -> "Account": + """The remembered account when it is the one asked for, else an ephemeral one.""" + stored = store.peek() + same_hub = stored.get("hub") == hub.url + same_user = username is None or username == stored.get("username") + if stored and same_hub and same_user and store.load(): + acc = cls(hub, store.state["username"], passphrase, store) + acc.state, acc.secrets = store.state, store.secrets + return acc + if not username: + raise CliError(AUTH, "Not signed in. Run `meshbay-cli login`, or pass --user.") + return cls(hub, username, passphrase) + + @contextlib.asynccontextmanager + async def _locked(self) -> AsyncIterator[None]: + """The store's lock, taken off the loop; re-reads what is stored.""" + if self.store is None: + yield + return + cm = self.store.locked() + await asyncio.to_thread(cm.__enter__) + try: + if not self.store.load(): + raise CliError(AUTH, "Signed out by another run; run `meshbay-cli login`.") + self.state, self.secrets = self.store.state, self.store.secrets + yield + finally: + cm.__exit__(None, None, None) + + def _save(self) -> None: + """Under the lock. `sign_in` replaces both dicts, so hand over the current ones.""" + self.store.state, self.store.secrets = self.state, self.secrets + self.store.save() + + async def update(self, mutate: Callable[[dict, dict], None]) -> None: + """Change state and secrets, and write them when remembered.""" + async with self._locked(): + mutate(self.state, self.secrets) + if self.store is not None: + self._save() + + # ── Signing in ─────────────────────────────────────────────────────────── + + async def sign_in(self, *, remember: bool) -> None: + """With the passphrase: tokens, `M`, and — remembered — the device's hub key.""" + passphrase = self._passphrase() + ak = await asyncio.to_thread(auth_key, self.username, passphrase) + previous = dict(self.state) + r = await self.hub.login(self.username, ak, previous.get("known_browser")) + token = r["access_token"] + me = await self.hub.me(token) + if not r.get("bundle_pepper"): + raise CliError(AUTH, "The hub did not provide the bundle pepper.") + a = await asyncio.to_thread(passphrase_half, self.username, passphrase) + master = bundle_master_key(a, unb64(r["bundle_pepper"]), me["user_id"]) + + same_account = previous.get("user_id") == me["user_id"] + state = { + "hub": self.hub.url, + "username": self.username, + "user_id": me["user_id"], + "known_browser": r.get("known_browser") or previous.get("known_browser"), + # A node's key does not depend on who signed in. + "pins": previous.get("pins", {}) if previous.get("hub") == self.hub.url else {}, + } + secrets = { + "access_token": token, + "access_exp": time.time() + int(r.get("expires_in", 0)), + "refresh_token": r["refresh_token"], + "master": b64(master), + "legacy": b64(a), + "identities": self.secrets.get("identities", {}) if same_account else {}, + } + if remember: + device = (self.secrets.get("device_sk") if same_account else None) + sk = (Ed25519PrivateKey.from_private_bytes(unb64(device)) if device + else Ed25519PrivateKey.generate()) + reg = await self.hub.register_device( + token, b64(sk.public_key().public_bytes_raw()), + f"meshbay-cli on {socket.gethostname()}", ak) + secrets["device_sk"] = b64(sk.private_bytes_raw()) + state["device_id"] = reg["id"] + self.state, self.secrets = state, secrets + + async def remember(self, store: Store) -> str | None: + """Write this account to `store`, replacing what it held. A notice, or None.""" + self.store = store + async with self._locked_fresh(): + store.state, store.secrets = self.state, self.secrets + store.save(new=True) + return KEY_FILE if store.state.get("key_store") == KEY_FILE else None + + @contextlib.asynccontextmanager + async def _locked_fresh(self) -> AsyncIterator[None]: + cm = self.store.locked() + await asyncio.to_thread(cm.__enter__) + try: + yield + finally: + cm.__exit__(None, None, None) + + # ── The access token ───────────────────────────────────────────────────── + + def _fresh(self) -> bool: + return bool(self.secrets.get("access_token")) and \ + self.secrets.get("access_exp", 0) - RENEW_MARGIN > time.time() + + async def token(self, *, rejected: str | None = None) -> str: + """A valid access token. `rejected`: the one the hub just refused.""" + if rejected is None and self._fresh(): + return self.secrets["access_token"] + if not self.remembered: + if not self.secrets: + await self.sign_in(remember=False) + else: + await self._renew() + return self.secrets["access_token"] + async with self._locked(): + # Another run may have renewed while this one waited for the lock; + # renewing again would present a refresh token it has just spent. + current = self.secrets.get("access_token") + if not self._fresh() or current == rejected: + await self._renew() + self._save() + return self.secrets["access_token"] + + async def _renew(self) -> None: + rt = self.secrets.get("refresh_token") + if rt: + try: + r = await self.hub.refresh(rt) + self._take(r) + return + except HubError as e: + log.debug("refresh refused: %s", e) + sk = self.secrets.get("device_sk") + if sk: + try: + r = await self.hub.device_auth( + self.username, Ed25519PrivateKey.from_private_bytes(unb64(sk))) + self._take(r) + return + except CliError as e: + log.debug("device sign-in refused: %s", e.message) + try: + await self.sign_in(remember=self.remembered) + except NoPassphrase as e: + if self.remembered: + raise CliError(AUTH, "The saved session has ended; run " + "`meshbay-cli login`, or pass --password.") from e + raise + + def _take(self, r: dict) -> None: + self.secrets["access_token"] = r["access_token"] + self.secrets["access_exp"] = time.time() + int(r.get("expires_in", 0)) + self.secrets["refresh_token"] = r["refresh_token"] + + async def call(self, fn, *args): + """A hub call with the access token, renewed once on a 401.""" + token = await self.token() + try: + return await fn(token, *args) + except HubError as e: + if e.status != 401: + raise + return await fn(await self.token(rejected=token), *args) + + # ── Identities and pins ────────────────────────────────────────────────── + + def identity(self, node_pk: str) -> Identity | None: + keys = self.secrets.get("identities", {}).get(node_pk) + return Identity.from_pkcs8(keys) if keys else None + + async def keep_identity(self, node_pk: str, identity: Identity) -> None: + def put(state: dict, secrets: dict) -> None: + secrets.setdefault("identities", {})[node_pk] = identity.to_pkcs8() + await self.update(put) + + async def forget_identity(self, node_pk: str) -> None: + def drop(state: dict, secrets: dict) -> None: + secrets.get("identities", {}).pop(node_pk, None) + await self.update(drop) + + # A device request this machine filed and that waits for an approval: the + # keys it named, and the code shown for them. Kept so a later run shows the + # same code and joins with the same keys once approved. + + def pending_identity(self, node_pk: str) -> "Pending | None": + p = self.secrets.get("pending", {}).get(node_pk) + if not p or p["expires_at"] < time.time(): + return None + return Pending(Identity.from_pkcs8(p["keys"]), p["code"], p["expires_at"]) + + async def keep_pending(self, node_pk: str, identity: Identity, code: str, + expires_at: float) -> None: + def put(state: dict, secrets: dict) -> None: + secrets.setdefault("pending", {})[node_pk] = { + "keys": identity.to_pkcs8(), "code": code, "expires_at": expires_at} + await self.update(put) + + async def adopt_pending(self, node_pk: str) -> None: + """The node admitted the pending keys: they are this machine's identity there.""" + def move(state: dict, secrets: dict) -> None: + p = secrets.get("pending", {}).pop(node_pk, None) + if p: + secrets.setdefault("identities", {})[node_pk] = p["keys"] + await self.update(move) + + def pin(self, node_id: str) -> str | None: + return self.state.get("pins", {}).get(node_id) + + async def set_pin(self, node_id: str, node_pk: str) -> None: + def put(state: dict, secrets: dict) -> None: + state.setdefault("pins", {})[node_id] = node_pk + await self.update(put) + + # ── Ending ─────────────────────────────────────────────────────────────── + + async def close(self) -> None: + """An ephemeral session ends on the hub with the run.""" + if not self.remembered and self.secrets.get("refresh_token"): + with contextlib.suppress(Exception): + await self.hub.logout(self.secrets["refresh_token"]) + + async def sign_out(self) -> list[str]: + """End the session, retire the device's hub key, erase the store. Warnings.""" + warnings = [] + device_id = self.state.get("device_id") + if device_id: + try: + await self.call(self.hub.delete_device, device_id) + except (HubError, CliError) as e: + warnings.append( + "The hub did not retire this machine's device key " + f"({getattr(e, 'detail', None) or getattr(e, 'message', e)}); remove " + f"\"meshbay-cli on {socket.gethostname()}\" from your account's devices.") + rt = self.secrets.get("refresh_token") + if rt: + with contextlib.suppress(Exception): + await self.hub.logout(rt) + if self.store is not None: + async with self._locked_fresh(): + self.store.erase() + return warnings |