path: root/packages/meshbay-cli/src/meshbay_cli/client/account.py

"""
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 {},
            "account_pins": (previous.get("account_pins", {})
                             if previous.get("hub") == self.hub.url else {}),
        }
        if same_account and previous.get("browser_access"):
            state["browser_access"] = previous["browser_access"]
        secrets = {
            "access_token": token,
            "access_exp": time.time() + int(r.get("expires_in", 0)),
            "refresh_token": r["refresh_token"],
            "master": b64(master),
            "pepper_version": int(r.get("bundle_pepper_version") or 1),
            "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)

    # An invitation's code waiting for its node: kept when the node was not
    # reachable at `join`, sent by the next connection to it, dropped once used
    # or refused. A link's code names its node; an invitation's does not.

    def join_code(self, group_id: str, node_pk: str) -> str | None:
        c = self.secrets.get("join_codes", {}).get(group_id)
        if not c or (c.get("node_pk") and c["node_pk"] != node_pk):
            return None
        return c["code"]

    async def keep_join_code(self, group_id: str, code: str, node_pk: str | None) -> None:
        def put(state: dict, secrets: dict) -> None:
            secrets.setdefault("join_codes", {})[group_id] = {"code": code,
                                                              "node_pk": node_pk}
        await self.update(put)

    async def drop_join_code(self, group_id: str) -> None:
        def drop(state: dict, secrets: dict) -> None:
            secrets.get("join_codes", {}).pop(group_id, None)
        if group_id in self.secrets.get("join_codes", {}):
            await self.update(drop)

    @property
    def browser_access(self) -> str | None:
        """This machine's setting: "on", "off", or None (unset — leave the node
        as it is). Set only by `browser-access`; never from the hub."""
        return self.state.get("browser_access")

    @property
    def pepper_version(self) -> int:
        return int(self.secrets.get("pepper_version") or 1)

    # The device keys this machine accepted for each account, per node (Tier 2,
    # `roster.py`): public keys, kept with the other pins. Losing them costs one
    # "first sight", never a wrong answer.

    def account_pins(self, node_pk: str, user_id: str) -> list[str] | None:
        return self.state.get("account_pins", {}).get(node_pk, {}).get(user_id)

    async def set_account_pins(self, node_pk: str, user_id: str, keys: list[str]) -> None:
        def put(state: dict, secrets: dict) -> None:
            state.setdefault("account_pins", {}).setdefault(node_pk, {})[user_id] = keys
        await self.update(put)

    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