MeshTerm/meshterm/core/admin_store.py

# SPDX-License-Identifier: Apache-2.0
"""Storage for the admin passwords of remote nodes that MeshTerm remembers.

The remote-admin optimizer logs in to a repeater to adjust its TX power. When MeshTerm
stores the password, the user does not get a prompt at each run. This store is a small
JSON file in the config directory (``<config_dir>/admin.json``), the same as the
remembered device (:mod:`meshterm.core.device_store`). It is not in the SQLite database
of each invocation, because admin credentials are global state of the machine.

The store reads the file into :class:`AdminCredential` records, and it writes the file
from these records. It never writes back the raw JSON that it read. Thus a field that no
record knows is gone after the next write (refer to :data:`AdminCredential.RETIRED`).

The passwords are stored in plaintext. Thus the store writes the file with owner-only
permissions where the platform supports them. This store is a local tool for a user who
adjusts their own mesh. It is not a secret store for many users. Handle the file as a
local file that contains plaintext passwords.
"""

from __future__ import annotations

import json
from dataclasses import dataclass
from pathlib import Path
from typing import ClassVar

from .atomicwrite import write_atomically
from .models import Contact, LoginResult, utcnow


def admin_key(node: Contact) -> str:
    """Return the lookup key for the stored password of a node.

    The function uses the full public key first, because it is stable and has only one
    meaning. If there is no public key, it uses the key prefix, and then the name. Thus
    MeshTerm can also remember a contact that has no key.

    Args:
        node: The contact for the lookup key.

    Returns:
        A normalized identifier in lower case.
    """
    key = node.public_key or node.key_prefix or node.name
    return key.lower().removeprefix("0x")


@dataclass(slots=True)
class AdminCredential:
    """A remembered admin password for one remote node.

    Attributes:
        key: The :func:`admin_key` under which the password is stored. It is the key of
            the record in the file, not one of its fields.
        password: The admin password of the node (plaintext).
        label: A node name to show to the user.
        last_used: The ISO-8601 timestamp of the last successful login.
    """

    key: str
    password: str
    label: str
    last_used: str

    #: The fields that this record held before. It must never hold them again with another
    #: meaning (refer to :data:`meshterm.core.preferences.RETIRED` for the reason that a
    #: name is never used again).
    RETIRED: ClassVar[frozenset[str]] = frozenset()

    @classmethod
    def read(cls, key: str, raw: object) -> AdminCredential | None:
        """Make a record from one file entry, or ``None`` when the entry has no password."""
        if not isinstance(raw, dict) or raw.get("password") is None:
            return None
        return cls(
            key=key,
            password=str(raw["password"]),
            label=str(raw.get("label") or ""),
            last_used=str(raw.get("last_used") or ""),
        )

    def to_json(self) -> dict:
        """The record as the store writes it: its own fields, and no other data read with it."""
        return {"password": self.password, "label": self.label, "last_used": self.last_used}


class AdminStore:
    """Reads and writes the admin passwords of remote nodes that MeshTerm remembers."""

    def __init__(self, path: Path) -> None:
        """Open the store on the location of a JSON file.

        Args:
            path: The path to the JSON state file. The store makes the file at the first
                write.
        """
        self._path = path

    def _load(self) -> dict[str, AdminCredential]:
        """Read all the credentials, by :func:`admin_key`.

        The result is empty if the file is missing or corrupt.
        """
        try:
            data = json.loads(self._path.read_text(encoding="utf-8"))
        except (OSError, ValueError):
            return {}
        if not isinstance(data, dict):
            return {}
        records = {}
        for key, raw in data.items():
            record = AdminCredential.read(str(key), raw)
            if record is not None:
                records[record.key] = record
        return records

    def get(self, node: Contact) -> str | None:
        """Return the remembered password for ``node``, or ``None`` if it is not stored.

        Args:
            node: The contact to find.

        Returns:
            The stored password, or ``None``.
        """
        record = self._load().get(admin_key(node))
        return record.password if record is not None else None

    def remember(self, node: Contact, password: str) -> None:
        """Store (or change) the admin password for ``node``.

        Args:
            node: The contact that the password is for.
            password: The admin password to remember.
        """
        records = self._load()
        key = admin_key(node)
        records[key] = AdminCredential(
            key=key, password=password, label=node.name, last_used=utcnow().isoformat()
        )
        self._write(records)

    def record(self, node: Contact, password: str, outcome: LoginResult) -> None:
        """Change what MeshTerm remembers about ``node``, from the result of a login.

        This method is the only credential policy. It is in one place because it was wrong
        in five places. The store remembers a password when the node accepts it, and it
        forgets the password when the node rejects it. When no reply came back, the store
        keeps the password exactly as it is. No reply is not a refusal. Before, each caller
        changed "refused" and "no reply" into the same ``False``, and forgot the password
        in both cases. Thus, when the user opened the admin flow on a repeater that was
        down at that time, MeshTerm deleted its password. The node sent nothing at all, and
        MeshTerm read that silence as a refusal.

        The callers still write their own messages (the sweep says "sweep again", the CLI
        says "re-run"), but no caller makes this decision.

        Args:
            node: The contact that the login was for.
            password: The password that was tried.
            outcome: What the device reported.
        """
        if outcome is LoginResult.ACCEPTED:
            self.remember(node, password)
        elif outcome is LoginResult.REFUSED:
            self.forget(node)

    def forget(self, node: Contact) -> None:
        """Delete the remembered password for ``node``, if there is one.

        For example, use this method after the password stops working.

        Args:
            node: The contact to forget.
        """
        records = self._load()
        if records.pop(admin_key(node), None) is not None:
            self._write(records)

    def _write(self, records: dict[str, AdminCredential]) -> None:
        """Store ``records`` atomically, with owner-only permissions where they are supported."""
        data = {key: record.to_json() for key, record in records.items()}
        write_atomically(self._path, json.dumps(data, indent=2), owner_only=True)