jmri_server.auth.store

  1"""Load / save the ``users.yaml`` file at ``<settings-root>/users.yaml``.
  2
  3YAML shape::
  4
  5    users:
  6      - username: admin
  7        password_hash: "$2b$12$..."
  8        role: admin
  9        token: "8f3a..."   # optional; empty string when not present
 10      - username: alice
 11        password_hash: "$2b$12$..."
 12        role: operator
 13        token: ""
 14
 15Load / save round-trip preserves user order so a human editor doesn't
 16see their carefully-organised file reshuffle on the first save. Atomic
 17write (temp file + replace) so a crashed save can't leave a truncated
 18file.
 19"""
 20from __future__ import annotations
 21
 22import logging
 23import os
 24import pathlib
 25import tempfile
 26from typing import Any
 27
 28import yaml
 29from pydantic import BaseModel, Field
 30
 31from jmri_server.auth.models import User
 32
 33_log = logging.getLogger(__name__)
 34
 35
 36class UsersFile(BaseModel):
 37    """Top-level ``users.yaml`` shape."""
 38
 39    users: list[User] = Field(default_factory=list)
 40
 41    def by_username(self, username: str) -> User | None:
 42        """Case-sensitive lookup. Returns ``None`` for unknown user.
 43
 44        JMRI usernames are treated as opaque identifiers; we do not
 45        lowercase for lookup because doing so silently hides the
 46        real difference between ``Admin`` and ``admin`` and
 47        confuses log-line audits."""
 48        for u in self.users:
 49            if u.username == username:
 50                return u
 51        return None
 52
 53    def by_token(self, token: str) -> User | None:
 54        """Constant-time-ish lookup by API token.
 55
 56        Bounded by user count and skips empty tokens (a user with
 57        no token gets no matches). ``hmac.compare_digest`` guards
 58        against timing side channels on the compare — matters when
 59        the caller can flood the endpoint with guesses."""
 60        if not token:
 61            return None
 62        import hmac  # noqa: PLC0415
 63        for u in self.users:
 64            if u.token and hmac.compare_digest(u.token, token):
 65                return u
 66        return None
 67
 68
 69def load_users_file(path: str | pathlib.Path) -> UsersFile:
 70    """Load *path* into a :class:`UsersFile`. Missing → empty file.
 71
 72    Malformed YAML raises ``yaml.YAMLError``; the caller decides
 73    whether to keep starting (empty users file → bootstrap admin)
 74    or crash out (edited file corrupt → refuse to start with auth
 75    on, since silently ignoring would be worse than a loud stop).
 76    """
 77    p = pathlib.Path(path)
 78    if not p.is_file():
 79        return UsersFile(users=[])
 80    raw: Any = yaml.safe_load(p.read_text(encoding="utf-8"))
 81    if raw is None:
 82        return UsersFile(users=[])
 83    if not isinstance(raw, dict):
 84        raise ValueError(
 85            f"{path}: expected a mapping at the top level, got {type(raw).__name__}"
 86        )
 87    return UsersFile.model_validate(raw)
 88
 89
 90def save_users_file(path: str | pathlib.Path, file: UsersFile) -> None:
 91    """Atomically write *file* to *path* with mode 0600 on POSIX.
 92
 93    Uses temp file + rename so a crashed save can't leave the
 94    file half-written. On Windows, the ``chmod`` is a no-op (the
 95    OS ACL isn't representable through the same mode bits); the
 96    settings root should be user-only anyway.
 97    """
 98    p = pathlib.Path(path)
 99    p.parent.mkdir(parents=True, exist_ok=True)
100    # ``mode="json"`` converts the ``Role`` enum to its string value —
101    # pyyaml's safe_dump can't represent StrEnum instances directly,
102    # and even if it could, we want the on-disk shape to be plain
103    # strings so a human editor doesn't see ``!!python/object:``
104    # references.
105    payload = {"users": [u.model_dump(mode="json") for u in file.users]}
106    fd, tmp = tempfile.mkstemp(
107        prefix=".users.yaml.", dir=str(p.parent),
108    )
109    try:
110        with os.fdopen(fd, "w", encoding="utf-8") as f:
111            yaml.safe_dump(payload, f, sort_keys=False)
112        # Best-effort mode 0600 — matters on multi-user hosts, no-op
113        # on Windows where the ACL model differs.
114        try:
115            os.chmod(tmp, 0o600)
116        except (OSError, PermissionError) as e:  # pragma: no cover
117            _log.debug("chmod on %s ignored: %s", tmp, e)
118        os.replace(tmp, p)
119    except Exception:
120        try:
121            os.unlink(tmp)
122        except OSError:
123            pass
124        raise