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

"""
What the CLI remembers between runs.

Two files in the data directory. `state.json` holds what is not secret: the hub,
the account, the device's id, each node's pinned key. `secrets.enc` holds the
rest — tokens, the device's hub key, the bundle key `M`, the identity on each
node — under AES-256-GCM with a 32-byte **store key**.

The store key lives in the OS key storage (Windows Credential Manager, Secret
Service on a Linux desktop). One key there and a file beside it, rather than
every secret there: Credential Manager caps a credential at 2560 bytes, and the
identities grow with the number of nodes. Where there is no key storage — an
SSH session on a server — the key is kept in a file only this user can read,
and `login` says so once (`file_key_notice`).

Every write goes through `locked()`: the hub rotates refresh tokens and revokes
the whole session when a spent one comes back, so two runs renewing at once
must take turns.
"""

import contextlib
import hashlib
import json
import logging
import os
import secrets
import sys
import time
from collections.abc import Iterator
from pathlib import Path

from cryptography.hazmat.primitives.ciphers.aead import AESGCM

from ..errors import AUTH, CliError

log = logging.getLogger(__name__)

KEYRING_SERVICE = "meshbay-cli"
SECRETS_MAGIC = b"MBC1"
SECRETS_AAD = b"meshbay-cli:secrets:v1"

KEY_OS = "os"
KEY_FILE = "file"

FILE_KEY_NOTICE = (
    "No OS key storage here: the session is kept in a file only your user can "
    "read (like an SSH key without a passphrase). `meshbay-cli logout` removes it."
)


def data_dir() -> Path:
    """Never the node's directories (`~/.config/meshbay`, `%LOCALAPPDATA%\\meshbay`)."""
    override = os.environ.get("MESHBAY_CLI_HOME")
    if override:
        return Path(override)
    if sys.platform == "win32":
        return Path(os.environ.get("LOCALAPPDATA") or Path.home() / "AppData" / "Local") \
            / "meshbay-cli"
    base = os.environ.get("XDG_DATA_HOME") or Path.home() / ".local" / "share"
    return Path(base) / "meshbay-cli"


def _write_private(path: Path, data: bytes) -> None:
    """Atomically, readable by this user only."""
    tmp = path.with_name(path.name + ".tmp")
    fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    try:
        os.write(fd, data)
        os.fsync(fd)
    finally:
        os.close(fd)
    os.replace(tmp, path)


class _Keyring:
    """
    The OS key storage. One entry per store directory, so two stores (another
    `MESHBAY_CLI_HOME`) never overwrite each other's key.
    """

    def __init__(self, root: Path):
        digest = hashlib.sha256(str(root.resolve()).encode()).hexdigest()[:16]
        self.entry = f"store-key:{digest}"

    def usable(self) -> bool:
        try:
            import keyring
            from keyring.backends import fail
            backend = keyring.get_keyring()
            if isinstance(backend, fail.Keyring):
                return False
            # A Secret Service that is installed but not running answers the
            # first call with an exception rather than at selection time.
            keyring.get_password(KEYRING_SERVICE, self.entry)
            return True
        except Exception as e:
            log.debug("OS key storage unavailable: %s", e)
            return False

    def get(self) -> bytes | None:
        import keyring
        value = keyring.get_password(KEYRING_SERVICE, self.entry)
        return bytes.fromhex(value) if value else None

    def set(self, key: bytes) -> None:
        import keyring
        keyring.set_password(KEYRING_SERVICE, self.entry, key.hex())

    def delete(self) -> None:
        import keyring
        with contextlib.suppress(Exception):
            keyring.delete_password(KEYRING_SERVICE, self.entry)


class Store:
    def __init__(self, root: Path | None = None):
        self.root = root or data_dir()
        self._keyring = _Keyring(self.root)
        self.state: dict = {}
        self.secrets: dict = {}

    @property
    def _state_path(self) -> Path:
        return self.root / "state.json"

    @property
    def _secrets_path(self) -> Path:
        return self.root / "secrets.enc"

    @property
    def _key_path(self) -> Path:
        return self.root / "store.key"

    def exists(self) -> bool:
        return self._state_path.exists() and self._secrets_path.exists()

    # ── Locking ──────────────────────────────────────────────────────────────

    @contextlib.contextmanager
    def locked(self) -> Iterator[None]:
        """An exclusive lock on the store, across processes. Blocks."""
        self.root.mkdir(parents=True, exist_ok=True)
        fd = os.open(self.root / "store.lock", os.O_RDWR | os.O_CREAT, 0o600)
        try:
            if sys.platform == "win32":
                import msvcrt
                while True:
                    try:
                        msvcrt.locking(fd, msvcrt.LK_NBLCK, 1)
                        break
                    except OSError:
                        time.sleep(0.1)
            else:
                import fcntl
                fcntl.flock(fd, fcntl.LOCK_EX)
            yield
        finally:
            if sys.platform == "win32":
                import msvcrt
                with contextlib.suppress(OSError):
                    os.lseek(fd, 0, os.SEEK_SET)
                    msvcrt.locking(fd, msvcrt.LK_UNLCK, 1)
            os.close(fd)

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

    def _load_key(self) -> bytes:
        kind = self.state.get("key_store")
        if kind == KEY_FILE:
            try:
                return bytes.fromhex(self._key_path.read_text().strip())
            except OSError as e:
                raise CliError(AUTH, f"The session's key file is missing ({e.strerror}); "
                                     "run `meshbay-cli login`.") from e
        if not self._keyring.usable():
            raise CliError(AUTH, "The session was saved in the OS key storage, which this "
                                 "session cannot reach. Run `meshbay-cli login` here, or "
                                 "pass --password.")
        key = self._keyring.get()
        if key is None:
            raise CliError(AUTH, "The session's key is gone from the OS key storage; "
                                 "run `meshbay-cli login`.")
        return key

    def _new_key(self) -> tuple[bytes, str]:
        key = secrets.token_bytes(32)
        if self._keyring.usable():
            try:
                self._keyring.set(key)
                return key, KEY_OS
            except Exception as e:
                log.debug("OS key storage refused the key: %s", e)
        _write_private(self._key_path, key.hex().encode())
        return key, KEY_FILE

    # ── Reading and writing ──────────────────────────────────────────────────

    def peek(self) -> dict:
        """state.json alone — the hub and the account, which are not secret."""
        try:
            return json.loads(self._state_path.read_text(encoding="utf-8"))
        except (OSError, ValueError):
            return {}

    def load(self) -> bool:
        """Read both files. False when nothing is stored."""
        if not self.exists():
            self.state, self.secrets = {}, {}
            return False
        self.state = json.loads(self._state_path.read_text(encoding="utf-8"))
        raw = self._secrets_path.read_bytes()
        if raw[:4] != SECRETS_MAGIC:
            raise CliError(AUTH, "The saved session is unreadable; run `meshbay-cli login`.")
        try:
            plain = AESGCM(self._load_key()).decrypt(raw[4:16], raw[16:], SECRETS_AAD)
        except CliError:
            raise
        except Exception as e:
            raise CliError(AUTH, "The saved session does not open with its key; run "
                                 "`meshbay-cli login`.") from e
        self.secrets = json.loads(plain)
        return True

    def save(self, *, new: bool = False) -> None:
        """Write both files. `new` makes a fresh store key (a new sign-in)."""
        self.root.mkdir(parents=True, exist_ok=True)
        if new or "key_store" not in self.state:
            self._forget_key()
            key, kind = self._new_key()
            self.state["key_store"] = kind
        else:
            key = self._load_key()
        nonce = secrets.token_bytes(12)
        ct = AESGCM(key).encrypt(nonce, json.dumps(self.secrets).encode(), SECRETS_AAD)
        _write_private(self._secrets_path, SECRETS_MAGIC + nonce + ct)
        _write_private(self._state_path, json.dumps(self.state, indent=1).encode())

    def _forget_key(self) -> None:
        kind = self.state.get("key_store")
        if kind == KEY_OS:
            self._keyring.delete()
        with contextlib.suppress(FileNotFoundError):
            self._key_path.unlink()

    def erase(self) -> None:
        """Everything, key included."""
        self._forget_key()
        for p in (self._secrets_path, self._state_path):
            with contextlib.suppress(FileNotFoundError):
                p.unlink()
        self.state, self.secrets = {}, {}