jmri_server.services.config_user

  1"""``ConfigUserService`` — user CRUD + self-service password/token.
  2
  3Wire shape (rewrite-specific extension — JMRI Java doesn't have
  4this topic since its Swing UI edits users through a native dialog)::
  5
  6    method=list                            → array of {username, role, has_token, self}
  7    method=get, data={"username": "..."}    → one user projection
  8    method=put,  data={"username": "...",
  9                       "password": "...",
 10                       "role": "viewer|operator|admin"}
 11        → creates or updates. On existing user, updates only the
 12          fields you passed. Admin-only.
 13
 14    method=delete, data={"username": "..."} → remove user. Admin only.
 15                                             Refuses self-delete + last-admin.
 16
 17    method=post, data={"action": "change_password",
 18                       "username": "...",
 19                       "new_password": "..."}
 20        → change password. Self only for non-admins; admins can
 21          change anyone's.
 22
 23    method=post, data={"action": "regenerate_token",
 24                       "username": "..."}
 25        → mint new API token, returns {"token": "..."}. Self only
 26          for non-admins; admins can regenerate anyone's.
 27
 28**Never returns the password_hash** in list/get/put — that would
 29leak the bcrypt salt across the wire even if hashed, and there's
 30no legitimate client-side use for it. Callers change passwords
 31via the ``change_password`` action, not by reading the current
 32hash and writing a modified one.
 33"""
 34from __future__ import annotations
 35
 36import logging
 37from typing import Any
 38
 39from jmri_server.auth.models import Role, User
 40from jmri_server.service import JsonService, JsonServiceError
 41
 42_log = logging.getLogger(__name__)
 43
 44
 45def _project(user: User, *, self_flag: bool = False) -> dict[str, Any]:
 46    """Wire projection of a User. Excludes password_hash and the
 47    token itself; only reports whether one exists. The token is
 48    returned only on regenerate."""
 49    return {
 50        "username": user.username,
 51        "role": user.role.value,
 52        "has_token": bool(user.token),
 53        "self": self_flag,
 54    }
 55
 56
 57class ConfigUserService(JsonService):
 58    topic = "configUser"
 59
 60    # User CRUD is admin-only for read + mutate. The self-service
 61    # POST actions do their own finer-grained checks (a viewer can
 62    # change their own password, not anyone else's) so ``post``
 63    # is gated at operator — the action handler enforces the rest.
 64    REQUIRED_ROLE = {
 65        "list": Role.ADMIN,
 66        "get": Role.ADMIN,
 67        "put": Role.ADMIN,
 68        "delete": Role.ADMIN,
 69        "post": Role.VIEWER,   # self-service always available
 70    }
 71
 72    # ---- helpers -------------------------------------------------------
 73
 74    def _auth(self):
 75        auth = getattr(self.ctx, "auth", None)
 76        if auth is None:
 77            raise JsonServiceError(
 78                503,
 79                "auth is disabled on this server (JMRI_AUTH_ENABLED not set)",
 80            )
 81        return auth
 82
 83    def _current_username(self) -> str | None:
 84        return getattr(self.ctx, "_current_username", None)
 85
 86    def _current_role(self) -> Role:
 87        """The current role is passed through dispatch to the JsonService
 88        signature, not to instances — but we can infer from the
 89        current user. When auth is off the middleware sets role=admin;
 90        we mirror that here by treating "no user" as admin."""
 91        current = self._current_username()
 92        if current is None:
 93            return Role.ADMIN
 94        user = self._auth().by_username(current)
 95        return user.role if user else Role.VIEWER
 96
 97    # ---- CRUD ----------------------------------------------------------
 98
 99    async def on_list(
100        self, data: dict[str, Any] | None,
101    ) -> list[dict[str, Any]]:
102        auth = self._auth()
103        current = self._current_username()
104        return [
105            _project(u, self_flag=(u.username == current))
106            for u in auth.users
107        ]
108
109    async def on_get(
110        self, data: dict[str, Any] | None,
111    ) -> dict[str, Any]:
112        data = data or {}
113        username = data.get("username")
114        if not isinstance(username, str) or not username:
115            raise JsonServiceError(400, "configUser.get requires 'username'")
116        auth = self._auth()
117        user = auth.by_username(username)
118        if user is None:
119            raise JsonServiceError(404, f"no user {username!r}")
120        return _project(
121            user, self_flag=(user.username == self._current_username()),
122        )
123
124    async def on_put(
125        self, data: dict[str, Any] | None,
126    ) -> dict[str, Any]:
127        """Create or update.
128
129        For a new user: ``password`` is required, ``role`` defaults
130        to viewer. For an existing user: only the fields supplied
131        are updated (role and/or password). Changing your own role
132        would trigger the self-demotion guard — see ``set_role``.
133        """
134        data = data or {}
135        username = data.get("username")
136        password = data.get("password")
137        role_raw = data.get("role")
138        if not isinstance(username, str) or not username:
139            raise JsonServiceError(400, "configUser.put requires 'username'")
140        role: Role | None = None
141        if role_raw is not None:
142            try:
143                role = Role(role_raw)
144            except ValueError:
145                raise JsonServiceError(
146                    400,
147                    f"invalid role {role_raw!r}; must be one of "
148                    f"{[r.value for r in Role]}",
149                ) from None
150
151        auth = self._auth()
152        current = self._current_username() or ""
153        existing = auth.by_username(username)
154        if existing is None:
155            if not isinstance(password, str) or not password:
156                raise JsonServiceError(
157                    400, "creating a user requires 'password'",
158                )
159            user = auth.add_user(
160                username, password,
161                role=role if role is not None else Role.VIEWER,
162            )
163            return _project(user, self_flag=(username == current))
164
165        # Existing user — update only what's supplied.
166        if isinstance(password, str) and password:
167            auth.set_password(username, password)
168        if role is not None and role is not existing.role:
169            try:
170                auth.set_role(
171                    username, role, current_username=current,
172                )
173            except ValueError as e:
174                raise JsonServiceError(409, str(e)) from None
175        return _project(
176            auth.by_username(username),
177            self_flag=(username == current),
178        )
179
180    async def on_delete(
181        self, data: dict[str, Any] | None,
182    ) -> dict[str, Any]:
183        data = data or {}
184        username = data.get("username")
185        if not isinstance(username, str) or not username:
186            raise JsonServiceError(400, "configUser.delete requires 'username'")
187        auth = self._auth()
188        current = self._current_username() or ""
189        try:
190            auth.remove_user(username, current_username=current)
191        except LookupError as e:
192            raise JsonServiceError(404, str(e)) from None
193        except ValueError as e:
194            raise JsonServiceError(409, str(e)) from None
195        return {"username": username, "deleted": True}
196
197    # ---- actions -------------------------------------------------------
198
199    async def on_post(
200        self, data: dict[str, Any] | None,
201    ) -> dict[str, Any]:
202        data = data or {}
203        action = data.get("action")
204        if action == "change_password":
205            return await self._change_password(data)
206        if action == "regenerate_token":
207            return await self._regenerate_token(data)
208        raise JsonServiceError(
209            400, f"unknown configUser action: {action!r}",
210        )
211
212    async def _change_password(
213        self, data: dict[str, Any],
214    ) -> dict[str, Any]:
215        username = data.get("username")
216        new_password = data.get("new_password")
217        if not isinstance(username, str) or not username:
218            raise JsonServiceError(400, "change_password requires 'username'")
219        if not isinstance(new_password, str) or not new_password:
220            raise JsonServiceError(400, "change_password requires 'new_password'")
221        current = self._current_username() or ""
222        if username != current and self._current_role() is not Role.ADMIN:
223            raise JsonServiceError(
224                403,
225                "only admins can change other users' passwords",
226            )
227        auth = self._auth()
228        try:
229            auth.set_password(username, new_password)
230        except LookupError as e:
231            raise JsonServiceError(404, str(e)) from None
232        return {"username": username, "changed": True}
233
234    async def _regenerate_token(
235        self, data: dict[str, Any],
236    ) -> dict[str, Any]:
237        username = data.get("username")
238        if not isinstance(username, str) or not username:
239            raise JsonServiceError(
240                400, "regenerate_token requires 'username'",
241            )
242        current = self._current_username() or ""
243        if username != current and self._current_role() is not Role.ADMIN:
244            raise JsonServiceError(
245                403,
246                "only admins can regenerate other users' tokens",
247            )
248        auth = self._auth()
249        try:
250            token = auth.regenerate_token(username)
251        except LookupError as e:
252            raise JsonServiceError(404, str(e)) from None
253        return {"username": username, "token": token}