jmri_server.auth.manager

  1"""``AuthManager`` — the runtime auth surface.
  2
  3Wraps the :class:`~jmri_server.auth.store.UsersFile` behind a small
  4API the middleware + FastAPI routes call into. Adds:
  5
  6* Password login (returns matched user or None).
  7* Bearer-token login (constant-time compare against every user's
  8  personal token).
  9* Signed-session-cookie mint + verify — uses
 10  :class:`~itsdangerous.URLSafeTimedSerializer` so a tampered
 11  cookie can't impersonate a user and expired cookies fall out on
 12  verify. The signing key is derived from the settings-root path
 13  (stable across restarts) unless the operator sets
 14  ``JMRI_AUTH_SECRET`` explicitly.
 15* Bootstrap: with auth enabled + no ``users.yaml`` present, mint an
 16  ``admin`` account with a random password + token, persist the
 17  password to ``<settings-root>/admin_password.txt`` (mode 0600),
 18  and log a WARNING pointing the operator at the file. This is the
 19  path a fresh install takes — subsequent starts leave it alone.
 20
 21The manager is deliberately synchronous: password hashing is CPU
 22work, not I/O, and putting it under ``async`` would just hide the
 23CPU spin behind a threadpool anyway. Callers await into the
 24middleware, then the middleware calls into these plain methods.
 25"""
 26from __future__ import annotations
 27
 28import logging
 29import os
 30import pathlib
 31import secrets
 32from dataclasses import dataclass
 33
 34from itsdangerous import BadSignature, SignatureExpired, URLSafeTimedSerializer
 35
 36from jmri_server.auth.models import Role, User
 37from jmri_server.auth.passwords import hash_password, verify_password
 38from jmri_server.auth.store import UsersFile, load_users_file, save_users_file
 39
 40_log = logging.getLogger(__name__)
 41
 42USERS_FILE_NAME = "users.yaml"
 43ADMIN_PASSWORD_FILE = "admin_password.txt"
 44ADMIN_TOKEN_ENV = "JMRI_AUTH_ADMIN_TOKEN"
 45SECRET_ENV = "JMRI_AUTH_SECRET"
 46EPHEMERAL_SECRET_ENV = "JMRI_AUTH_EPHEMERAL_SECRET"
 47
 48# One week — matches JMRI mobile clients that stay logged in
 49# between operating sessions but forces re-auth after long gaps.
 50SESSION_MAX_AGE_SECONDS = 7 * 24 * 3600
 51
 52
 53@dataclass(frozen=True, slots=True)
 54class LoginResult:
 55    """A successful login carries the matched user.
 56
 57    Kept as a distinct type so ``if login(...):`` reads naturally
 58    without confusing ``User``-vs-``None`` at the call site."""
 59    user: User
 60
 61
 62class AuthManager:
 63    """Load users, verify credentials, mint session cookies.
 64
 65    One instance per server lifetime. Constructed by
 66    :func:`build_auth_manager` which handles the bootstrap +
 67    users-file discovery — direct instantiation is fine for tests
 68    that seed a specific set of users.
 69    """
 70
 71    def __init__(
 72        self,
 73        users: UsersFile,
 74        *,
 75        users_path: pathlib.Path,
 76        secret_key: str,
 77    ) -> None:
 78        self._users = users
 79        self._users_path = users_path
 80        self._serializer = URLSafeTimedSerializer(
 81            secret_key, salt="jmri-session-v1",
 82        )
 83
 84    # ---- user lookup + password flow ----------------------------------
 85
 86    def login_password(
 87        self, username: str, password: str,
 88    ) -> LoginResult | None:
 89        """Verify *password* for *username*. Returns the login
 90        result on success, ``None`` on any failure (unknown user,
 91        wrong password, malformed hash). Never leaks which of the
 92        two failed via return value — the log line records it for
 93        the operator's audit."""
 94        user = self._users.by_username(username)
 95        if user is None:
 96            _log.info("auth: login failed for unknown user %r", username)
 97            return None
 98        if not verify_password(password, user.password_hash):
 99            _log.info("auth: bad password for user %r", username)
100            return None
101        return LoginResult(user=user)
102
103    def login_token(self, token: str) -> LoginResult | None:
104        """Match a Bearer token against every user's personal
105        token. Returns the login result on success, ``None`` on
106        no match. Empty token → ``None`` without querying the
107        store."""
108        user = self._users.by_token(token)
109        if user is None:
110            return None
111        return LoginResult(user=user)
112
113    # ---- session cookies ----------------------------------------------
114
115    def mint_session_cookie(self, user: User) -> str:
116        """Encode *user* as a signed cookie value. The value round-
117        trips through :meth:`verify_session_cookie`; anything else
118        (tampered value, expired timestamp) returns ``None`` there."""
119        return self._serializer.dumps({"username": user.username})
120
121    def verify_session_cookie(self, cookie_value: str) -> User | None:
122        """Decode + validate. Returns the current :class:`User` for
123        the encoded username, or ``None`` for tampered / expired /
124        deleted-user cases.
125
126        Deleted-user handling matters: a user removed from
127        ``users.yaml`` should be locked out immediately, not on
128        session expiry. Re-checking the store on every verify is
129        cheap (dict lookup) and closes that hole."""
130        if not cookie_value:
131            return None
132        try:
133            payload = self._serializer.loads(
134                cookie_value, max_age=SESSION_MAX_AGE_SECONDS,
135            )
136        except SignatureExpired:
137            return None
138        except BadSignature:
139            return None
140        if not isinstance(payload, dict):
141            return None
142        username = payload.get("username")
143        if not isinstance(username, str):
144            return None
145        # Re-fetch so a role change (or deletion) after cookie
146        # mint takes effect immediately.
147        return self._users.by_username(username)
148
149    # ---- introspection -------------------------------------------------
150
151    @property
152    def users(self) -> list[User]:
153        """Read-only snapshot of the current user list. Mutations
154        go through :meth:`add_user` / :meth:`remove_user` so the
155        on-disk file stays in sync."""
156        return list(self._users.users)
157
158    def has_users(self) -> bool:
159        return bool(self._users.users)
160
161    def by_username(self, username: str) -> User | None:
162        return self._users.by_username(username)
163
164    # ---- mutations ------------------------------------------------------
165
166    def add_user(
167        self,
168        username: str,
169        password: str,
170        *,
171        role: Role = Role.VIEWER,
172        token: str | None = None,
173    ) -> User:
174        """Create a new user. Persists to disk atomically.
175
176        ``token`` defaults to a fresh random one — a client (mobile
177        throttle, etc.) can start using it immediately. Pass an
178        explicit empty string to create a login-only account.
179        """
180        if not username:
181            raise ValueError("username must be non-empty")
182        if self._users.by_username(username) is not None:
183            raise ValueError(f"user {username!r} already exists")
184        actual_token = (
185            token if token is not None else secrets.token_urlsafe(32)
186        )
187        user = User(
188            username=username,
189            password_hash=hash_password(password),
190            role=role,
191            token=actual_token,
192        )
193        self._users.users.append(user)
194        self._persist()
195        return user
196
197    def remove_user(self, username: str, *, current_username: str) -> None:
198        """Delete *username*. Two guards:
199
200        * Cannot delete yourself — refuses when
201          ``username == current_username`` so an admin can't
202          accidentally lock the whole install out.
203        * Cannot delete the last admin — refuses when *username*
204          is the sole entry with :data:`Role.ADMIN`. Otherwise a
205          botched cleanup would leave the install with no admin
206          user, which no future user-CRUD flow could recover from.
207        """
208        if username == current_username:
209            raise ValueError("cannot delete your own account")
210        target = self._users.by_username(username)
211        if target is None:
212            raise LookupError(f"no user {username!r}")
213        if target.role is Role.ADMIN:
214            admins = [u for u in self._users.users if u.role is Role.ADMIN]
215            if len(admins) <= 1:
216                raise ValueError(
217                    "cannot delete the last admin — promote someone else first",
218                )
219        self._users.users = [
220            u for u in self._users.users if u.username != username
221        ]
222        self._persist()
223
224    def set_password(self, username: str, new_password: str) -> None:
225        """Update *username*'s password hash. Existing session
226        cookies stay valid (they encode username only) — a
227        password change without a forced-logout is the JMRI
228        Java convention and matches "remember me" behaviour."""
229        user = self._users.by_username(username)
230        if user is None:
231            raise LookupError(f"no user {username!r}")
232        user.password_hash = hash_password(new_password)
233        self._persist()
234
235    def set_role(
236        self, username: str, new_role: Role, *, current_username: str,
237    ) -> None:
238        """Change *username*'s role. Same last-admin guard as
239        :meth:`remove_user` — demoting the only admin would
240        strand the install."""
241        if username == current_username and new_role is not Role.ADMIN:
242            raise ValueError("cannot demote your own account")
243        user = self._users.by_username(username)
244        if user is None:
245            raise LookupError(f"no user {username!r}")
246        if user.role is Role.ADMIN and new_role is not Role.ADMIN:
247            admins = [u for u in self._users.users if u.role is Role.ADMIN]
248            if len(admins) <= 1:
249                raise ValueError(
250                    "cannot demote the last admin — promote someone else first",
251                )
252        user.role = new_role
253        self._persist()
254
255    def regenerate_token(self, username: str) -> str:
256        """Mint a new token for *username*, replacing the old.
257        Returns the new value. Any mobile client using the old
258        token is immediately locked out — that's the point of
259        this endpoint (a leaked token gets cycled)."""
260        user = self._users.by_username(username)
261        if user is None:
262            raise LookupError(f"no user {username!r}")
263        user.token = secrets.token_urlsafe(32)
264        self._persist()
265        return user.token
266
267    def _persist(self) -> None:
268        """Write the current users file to disk. Called after
269        every mutation so an operator hand-editing users.yaml
270        alongside the running server sees a coherent file."""
271        save_users_file(self._users_path, self._users)
272
273
274# ---- bootstrap ----------------------------------------------------------
275
276
277def build_auth_manager(settings_root: pathlib.Path) -> AuthManager:
278    """Load ``<settings-root>/users.yaml`` — or mint a default
279    admin if missing — and return a fully-configured
280    :class:`AuthManager`.
281
282    Called once at server startup when
283    ``JMRI_AUTH_ENABLED=1``. With auth off, callers should not
284    invoke this; the server sets ``ctx.auth = None`` instead and
285    the middleware becomes a no-op.
286    """
287    users_path = settings_root / USERS_FILE_NAME
288    users = load_users_file(users_path)
289    if not users.users:
290        users = _bootstrap_default_admin(settings_root)
291        save_users_file(users_path, users)
292
293    secret = _resolve_secret(settings_root)
294    return AuthManager(users, users_path=users_path, secret_key=secret)
295
296
297def _resolve_secret(settings_root: pathlib.Path) -> str:
298    """Determine the session signing key. Precedence:
299
300    1. ``JMRI_AUTH_SECRET`` — explicit operator override. Always wins.
301       Rotating this value invalidates every outstanding session.
302    2. ``JMRI_AUTH_EPHEMERAL_SECRET=1`` — mint a fresh random secret
303       at every server boot. All sessions die on restart, which is
304       what you want in dev / when testing auth flows / on a shared
305       machine where you don't want the "convenience persistence"
306       across reboots.
307    3. Derived stable key from the settings root — the default.
308       Sessions survive restarts (matches "remember me" semantics).
309    """
310    explicit = os.environ.get(SECRET_ENV)
311    if explicit:
312        return explicit
313    if _get_bool_env(EPHEMERAL_SECRET_ENV):
314        _log.info(
315            "auth: %s=1 — sessions will invalidate on every restart",
316            EPHEMERAL_SECRET_ENV,
317        )
318        return secrets.token_hex(32)
319    return _derive_secret_from_root(settings_root)
320
321
322def _get_bool_env(name: str) -> bool:
323    """Same truthy set as :mod:`jmri_core.env` — kept local to avoid
324    a circular import when the auth module is used standalone."""
325    raw = os.environ.get(name, "")
326    return raw.lower() in ("1", "true", "yes", "on")
327
328
329def _bootstrap_default_admin(settings_root: pathlib.Path) -> UsersFile:
330    """Mint an ``admin`` user with random password + token; write
331    the plaintext password to ``<settings-root>/admin_password.txt``
332    (mode 0600) so the operator can retrieve it on first start.
333
334    On subsequent starts the users file already has entries, so
335    this path never runs — the ``admin_password.txt`` sticks
336    around as a record of the initial secret until the operator
337    deletes it or changes the admin password."""
338    password = secrets.token_urlsafe(24)
339    token = os.environ.get(ADMIN_TOKEN_ENV) or secrets.token_urlsafe(32)
340    admin = User(
341        username="admin",
342        password_hash=hash_password(password),
343        role=Role.ADMIN,
344        token=token,
345    )
346    _write_admin_password_file(settings_root, password, token)
347    _log.warning(
348        "auth bootstrap: no users.yaml found. Default admin user created; "
349        "password written to %s (mode 0600). Log in as 'admin' and change "
350        "the password via a future user CRUD flow, or hand-edit users.yaml.",
351        settings_root / ADMIN_PASSWORD_FILE,
352    )
353    return UsersFile(users=[admin])
354
355
356def _write_admin_password_file(
357    settings_root: pathlib.Path, password: str, token: str,
358) -> None:
359    """Persist the bootstrap admin credentials so the operator can
360    read them on first login. Includes the API token too — mobile
361    clients need it to connect without a browser round-trip."""
362    settings_root.mkdir(parents=True, exist_ok=True)
363    path = settings_root / ADMIN_PASSWORD_FILE
364    body = (
365        "# JMRI auth bootstrap credentials — first-run only.\n"
366        "# Delete this file after you've copied the values into a\n"
367        "# password manager. Regeneration requires deleting the users\n"
368        "# entry as well (there's no re-bootstrap flow yet).\n"
369        f"admin_password: {password}\n"
370        f"admin_token: {token}\n"
371    )
372    path.write_text(body, encoding="utf-8")
373    try:
374        os.chmod(path, 0o600)
375    except (OSError, PermissionError) as e:  # pragma: no cover
376        _log.debug("chmod on %s ignored: %s", path, e)
377
378
379def _derive_secret_from_root(settings_root: pathlib.Path) -> str:
380    """Derive a stable session-signing secret from the settings
381    root path + host hostname when ``JMRI_AUTH_SECRET`` isn't set.
382
383    Stable across restarts (so sessions survive), unique per
384    install (so a leaked cookie from one Pi doesn't unlock
385    another). For a production deployment with real users the
386    operator should set ``JMRI_AUTH_SECRET`` explicitly — this
387    fallback exists so single-Pi hobby installs don't need an
388    extra config step."""
389    import hashlib  # noqa: PLC0415
390    import socket  # noqa: PLC0415
391    seed = f"{settings_root.resolve()}::{socket.gethostname()}"
392    return hashlib.sha256(seed.encode("utf-8")).hexdigest()