jmri_server.auth

 63class AuthManager:
 64    """Load users, verify credentials, mint session cookies.
 65
 66    One instance per server lifetime. Constructed by
 67    :func:`build_auth_manager` which handles the bootstrap +
 68    users-file discovery — direct instantiation is fine for tests
 69    that seed a specific set of users.
 70    """
 71
 72    def __init__(
 73        self,
 74        users: UsersFile,
 75        *,
 76        users_path: pathlib.Path,
 77        secret_key: str,
 78    ) -> None:
 79        self._users = users
 80        self._users_path = users_path
 81        self._serializer = URLSafeTimedSerializer(
 82            secret_key, salt="jmri-session-v1",
 83        )
 84
 85    # ---- user lookup + password flow ----------------------------------
 86
 87    def login_password(
 88        self, username: str, password: str,
 89    ) -> LoginResult | None:
 90        """Verify *password* for *username*. Returns the login
 91        result on success, ``None`` on any failure (unknown user,
 92        wrong password, malformed hash). Never leaks which of the
 93        two failed via return value — the log line records it for
 94        the operator's audit."""
 95        user = self._users.by_username(username)
 96        if user is None:
 97            _log.info("auth: login failed for unknown user %r", username)
 98            return None
 99        if not verify_password(password, user.password_hash):
100            _log.info("auth: bad password for user %r", username)
101            return None
102        return LoginResult(user=user)
103
104    def login_token(self, token: str) -> LoginResult | None:
105        """Match a Bearer token against every user's personal
106        token. Returns the login result on success, ``None`` on
107        no match. Empty token → ``None`` without querying the
108        store."""
109        user = self._users.by_token(token)
110        if user is None:
111            return None
112        return LoginResult(user=user)
113
114    # ---- session cookies ----------------------------------------------
115
116    def mint_session_cookie(self, user: User) -> str:
117        """Encode *user* as a signed cookie value. The value round-
118        trips through :meth:`verify_session_cookie`; anything else
119        (tampered value, expired timestamp) returns ``None`` there."""
120        return self._serializer.dumps({"username": user.username})
121
122    def verify_session_cookie(self, cookie_value: str) -> User | None:
123        """Decode + validate. Returns the current :class:`User` for
124        the encoded username, or ``None`` for tampered / expired /
125        deleted-user cases.
126
127        Deleted-user handling matters: a user removed from
128        ``users.yaml`` should be locked out immediately, not on
129        session expiry. Re-checking the store on every verify is
130        cheap (dict lookup) and closes that hole."""
131        if not cookie_value:
132            return None
133        try:
134            payload = self._serializer.loads(
135                cookie_value, max_age=SESSION_MAX_AGE_SECONDS,
136            )
137        except SignatureExpired:
138            return None
139        except BadSignature:
140            return None
141        if not isinstance(payload, dict):
142            return None
143        username = payload.get("username")
144        if not isinstance(username, str):
145            return None
146        # Re-fetch so a role change (or deletion) after cookie
147        # mint takes effect immediately.
148        return self._users.by_username(username)
149
150    # ---- introspection -------------------------------------------------
151
152    @property
153    def users(self) -> list[User]:
154        """Read-only snapshot of the current user list. Mutations
155        go through :meth:`add_user` / :meth:`remove_user` so the
156        on-disk file stays in sync."""
157        return list(self._users.users)
158
159    def has_users(self) -> bool:
160        return bool(self._users.users)
161
162    def by_username(self, username: str) -> User | None:
163        return self._users.by_username(username)
164
165    # ---- mutations ------------------------------------------------------
166
167    def add_user(
168        self,
169        username: str,
170        password: str,
171        *,
172        role: Role = Role.VIEWER,
173        token: str | None = None,
174    ) -> User:
175        """Create a new user. Persists to disk atomically.
176
177        ``token`` defaults to a fresh random one — a client (mobile
178        throttle, etc.) can start using it immediately. Pass an
179        explicit empty string to create a login-only account.
180        """
181        if not username:
182            raise ValueError("username must be non-empty")
183        if self._users.by_username(username) is not None:
184            raise ValueError(f"user {username!r} already exists")
185        actual_token = (
186            token if token is not None else secrets.token_urlsafe(32)
187        )
188        user = User(
189            username=username,
190            password_hash=hash_password(password),
191            role=role,
192            token=actual_token,
193        )
194        self._users.users.append(user)
195        self._persist()
196        return user
197
198    def remove_user(self, username: str, *, current_username: str) -> None:
199        """Delete *username*. Two guards:
200
201        * Cannot delete yourself — refuses when
202          ``username == current_username`` so an admin can't
203          accidentally lock the whole install out.
204        * Cannot delete the last admin — refuses when *username*
205          is the sole entry with :data:`Role.ADMIN`. Otherwise a
206          botched cleanup would leave the install with no admin
207          user, which no future user-CRUD flow could recover from.
208        """
209        if username == current_username:
210            raise ValueError("cannot delete your own account")
211        target = self._users.by_username(username)
212        if target is None:
213            raise LookupError(f"no user {username!r}")
214        if target.role is Role.ADMIN:
215            admins = [u for u in self._users.users if u.role is Role.ADMIN]
216            if len(admins) <= 1:
217                raise ValueError(
218                    "cannot delete the last admin — promote someone else first",
219                )
220        self._users.users = [
221            u for u in self._users.users if u.username != username
222        ]
223        self._persist()
224
225    def set_password(self, username: str, new_password: str) -> None:
226        """Update *username*'s password hash. Existing session
227        cookies stay valid (they encode username only) — a
228        password change without a forced-logout is the JMRI
229        Java convention and matches "remember me" behaviour."""
230        user = self._users.by_username(username)
231        if user is None:
232            raise LookupError(f"no user {username!r}")
233        user.password_hash = hash_password(new_password)
234        self._persist()
235
236    def set_role(
237        self, username: str, new_role: Role, *, current_username: str,
238    ) -> None:
239        """Change *username*'s role. Same last-admin guard as
240        :meth:`remove_user` — demoting the only admin would
241        strand the install."""
242        if username == current_username and new_role is not Role.ADMIN:
243            raise ValueError("cannot demote your own account")
244        user = self._users.by_username(username)
245        if user is None:
246            raise LookupError(f"no user {username!r}")
247        if user.role is Role.ADMIN and new_role is not Role.ADMIN:
248            admins = [u for u in self._users.users if u.role is Role.ADMIN]
249            if len(admins) <= 1:
250                raise ValueError(
251                    "cannot demote the last admin — promote someone else first",
252                )
253        user.role = new_role
254        self._persist()
255
256    def regenerate_token(self, username: str) -> str:
257        """Mint a new token for *username*, replacing the old.
258        Returns the new value. Any mobile client using the old
259        token is immediately locked out — that's the point of
260        this endpoint (a leaked token gets cycled)."""
261        user = self._users.by_username(username)
262        if user is None:
263            raise LookupError(f"no user {username!r}")
264        user.token = secrets.token_urlsafe(32)
265        self._persist()
266        return user.token
267
268    def _persist(self) -> None:
269        """Write the current users file to disk. Called after
270        every mutation so an operator hand-editing users.yaml
271        alongside the running server sees a coherent file."""
272        save_users_file(self._users_path, self._users)

54@dataclass(frozen=True, slots=True)
55class LoginResult:
56    """A successful login carries the matched user.
57
58    Kept as a distinct type so ``if login(...):`` reads naturally
59    without confusing ``User``-vs-``None`` at the call site."""
60    user: User

21class Role(StrEnum):
22    """Access role for one user.
23
24    Ordered by privilege (viewer < operator < admin). The
25    comparison operators are defined via :meth:`_priority` so
26    ``request.state.role >= Role.OPERATOR`` reads naturally on the
27    dispatch path.
28    """
29
30    VIEWER = "viewer"
31    OPERATOR = "operator"
32    ADMIN = "admin"
33
34    def _priority(self) -> int:
35        return {"viewer": 0, "operator": 1, "admin": 2}[self.value]
36
37    def __ge__(self, other: object) -> bool:  # type: ignore[override]
38        if not isinstance(other, Role):
39            return NotImplemented
40        return self._priority() >= other._priority()
41
42    def __gt__(self, other: object) -> bool:  # type: ignore[override]
43        if not isinstance(other, Role):
44            return NotImplemented
45        return self._priority() > other._priority()
46
47    def __le__(self, other: object) -> bool:  # type: ignore[override]
48        if not isinstance(other, Role):
49            return NotImplemented
50        return self._priority() <= other._priority()
51
52    def __lt__(self, other: object) -> bool:  # type: ignore[override]
53        if not isinstance(other, Role):
54            return NotImplemented
55        return self._priority() < other._priority()

58class User(BaseModel):
59    """One user record — matches one entry in ``users.yaml``.
60
61    ``password_hash`` is bcrypt-encoded; verify via
62    :func:`~jmri_server.auth.passwords.verify_password`. ``token``
63    is the long-lived personal API token — regenerated on demand,
64    never derived from the password so a password change doesn't
65    invalidate mobile clients.
66
67    ``token`` may be empty for a user that only logs in via the
68    web UI. The bootstrap admin always gets one so JMRI mobile
69    clients can be connected without a browser round-trip.
70    """
71
72    model_config = ConfigDict(frozen=False, extra="ignore")
73
74    username: str
75    password_hash: str
76    role: Role = Role.VIEWER
77    token: str = ""

37class UsersFile(BaseModel):
38    """Top-level ``users.yaml`` shape."""
39
40    users: list[User] = Field(default_factory=list)
41
42    def by_username(self, username: str) -> User | None:
43        """Case-sensitive lookup. Returns ``None`` for unknown user.
44
45        JMRI usernames are treated as opaque identifiers; we do not
46        lowercase for lookup because doing so silently hides the
47        real difference between ``Admin`` and ``admin`` and
48        confuses log-line audits."""
49        for u in self.users:
50            if u.username == username:
51                return u
52        return None
53
54    def by_token(self, token: str) -> User | None:
55        """Constant-time-ish lookup by API token.
56
57        Bounded by user count and skips empty tokens (a user with
58        no token gets no matches). ``hmac.compare_digest`` guards
59        against timing side channels on the compare — matters when
60        the caller can flood the endpoint with guesses."""
61        if not token:
62            return None
63        import hmac  # noqa: PLC0415
64        for u in self.users:
65            if u.token and hmac.compare_digest(u.token, token):
66                return u
67        return None

278def build_auth_manager(settings_root: pathlib.Path) -> AuthManager:
279    """Load ``<settings-root>/users.yaml`` — or mint a default
280    admin if missing — and return a fully-configured
281    :class:`AuthManager`.
282
283    Called once at server startup when
284    ``JMRI_AUTH_ENABLED=1``. With auth off, callers should not
285    invoke this; the server sets ``ctx.auth = None`` instead and
286    the middleware becomes a no-op.
287    """
288    users_path = settings_root / USERS_FILE_NAME
289    users = load_users_file(users_path)
290    if not users.users:
291        users = _bootstrap_default_admin(settings_root)
292        save_users_file(users_path, users)
293
294    secret = _resolve_secret(settings_root)
295    return AuthManager(users, users_path=users_path, secret_key=secret)

23def hash_password(password: str) -> str:
24    """Return a bcrypt hash of *password* suitable for storage.
25
26    The returned string embeds the salt + work factor + hash, so
27    :func:`verify_password` doesn't need any external state to
28    validate. Idempotent-safe: repeated calls with the same
29    password produce different hashes (different salts) — verify
30    handles both.
31    """
32    if not password:
33        raise ValueError("password must be non-empty")
34    salted = bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt())
35    return salted.decode("utf-8")

70def load_users_file(path: str | pathlib.Path) -> UsersFile:
71    """Load *path* into a :class:`UsersFile`. Missing → empty file.
72
73    Malformed YAML raises ``yaml.YAMLError``; the caller decides
74    whether to keep starting (empty users file → bootstrap admin)
75    or crash out (edited file corrupt → refuse to start with auth
76    on, since silently ignoring would be worse than a loud stop).
77    """
78    p = pathlib.Path(path)
79    if not p.is_file():
80        return UsersFile(users=[])
81    raw: Any = yaml.safe_load(p.read_text(encoding="utf-8"))
82    if raw is None:
83        return UsersFile(users=[])
84    if not isinstance(raw, dict):
85        raise ValueError(
86            f"{path}: expected a mapping at the top level, got {type(raw).__name__}"
87        )
88    return UsersFile.model_validate(raw)

 91def save_users_file(path: str | pathlib.Path, file: UsersFile) -> None:
 92    """Atomically write *file* to *path* with mode 0600 on POSIX.
 93
 94    Uses temp file + rename so a crashed save can't leave the
 95    file half-written. On Windows, the ``chmod`` is a no-op (the
 96    OS ACL isn't representable through the same mode bits); the
 97    settings root should be user-only anyway.
 98    """
 99    p = pathlib.Path(path)
100    p.parent.mkdir(parents=True, exist_ok=True)
101    # ``mode="json"`` converts the ``Role`` enum to its string value —
102    # pyyaml's safe_dump can't represent StrEnum instances directly,
103    # and even if it could, we want the on-disk shape to be plain
104    # strings so a human editor doesn't see ``!!python/object:``
105    # references.
106    payload = {"users": [u.model_dump(mode="json") for u in file.users]}
107    fd, tmp = tempfile.mkstemp(
108        prefix=".users.yaml.", dir=str(p.parent),
109    )
110    try:
111        with os.fdopen(fd, "w", encoding="utf-8") as f:
112            yaml.safe_dump(payload, f, sort_keys=False)
113        # Best-effort mode 0600 — matters on multi-user hosts, no-op
114        # on Windows where the ACL model differs.
115        try:
116            os.chmod(tmp, 0o600)
117        except (OSError, PermissionError) as e:  # pragma: no cover
118            _log.debug("chmod on %s ignored: %s", tmp, e)
119        os.replace(tmp, p)
120    except Exception:
121        try:
122            os.unlink(tmp)
123        except OSError:
124            pass
125        raise

38def verify_password(password: str, password_hash: str) -> bool:
39    """Constant-time check of *password* against *password_hash*.
40
41    Returns ``False`` for an empty ``password_hash`` — a user with
42    no hash on file has no valid password, and short-circuiting to
43    ``False`` avoids leaking that state via bcrypt error messages.
44    """
45    if not password_hash:
46        return False
47    try:
48        return bcrypt.checkpw(
49            password.encode("utf-8"),
50            password_hash.encode("utf-8"),
51        )
52    except (ValueError, TypeError):
53        # Malformed hash — treat as no match rather than crashing.
54        # Real users see this after a hand-edit of users.yaml gone
55        # wrong; the auth log will show why they can't log in.
56        return False