jmri_server.services

23class AudioService(_BeanService):
24    topic = "audio"
25
26    def manager(self) -> AudioManager:
27        return self.ctx.connection.audios
28
29    def project(self, bean: Audio) -> dict[str, Any]:
30        return _common_named_bean(bean) | {
31            "state": bean.state,
32            "kind": bean.kind,
33            "url": bean.url,
34            "gain": bean.gain,
35        }
36
37    async def apply(self, bean: Audio, data: dict[str, Any]) -> None:
38        if "kind" in data:
39            kind = data["kind"]
40            if kind not in ("source", "buffer", "listener"):
41                raise JsonServiceError(400, f"invalid audio kind: {kind!r}")
42            bean.kind = kind
43        if "url" in data:
44            bean.url = str(data["url"]) if data["url"] is not None else None
45        if "gain" in data:
46            try:
47                gain = float(data["gain"])
48            except (TypeError, ValueError):
49                raise JsonServiceError(400, "audio.gain must be numeric") from None
50            if not 0.0 <= gain <= 1.0:
51                raise JsonServiceError(400, "audio.gain must be in [0.0, 1.0]")
52            bean.gain = gain
53        # Explicit state override (rare — clients normally send
54        # ``command`` instead) but scripts sometimes set state directly.
55        if "state" in data:
56            state = int(data["state"])
57            if state not in (AUDIO_STATE_STOPPED, AUDIO_STATE_PLAYING):
58                raise JsonServiceError(
59                    400,
60                    f"audio.state must be {AUDIO_STATE_STOPPED} (stopped) or "
61                    f"{AUDIO_STATE_PLAYING} (playing)",
62                )
63            if state == AUDIO_STATE_PLAYING:
64                await bean.play()
65            else:
66                await bean.stop()
67        # JMRI Java's Play / Stop verbs. Only meaningful on sources; a
68        # buffer or listener silently no-ops the command rather than
69        # 400-ing to keep the API forgiving for polymorphic clients.
70        command = data.get("command")
71        if command == "Play":
72            if bean.kind == "source":
73                await bean.play()
74        elif command == "Stop":
75            if bean.kind == "source":
76                await bean.stop()
77        elif command is not None:
78            raise JsonServiceError(400, f"unknown audio command: {command!r}")

107class BeanTransferService(JsonService):
108    """Cross-profile bean transfer. Read-only ``list`` +
109    ``post`` with ``action=preview|apply``."""
110
111    topic = "beanTransfer"
112    # ``list`` is safe for any role. ``post`` writes across
113    # profiles — admin-only, since a compromised operator token
114    # shouldn't be able to leak or overwrite the production
115    # profile's beans from a staging session.
116    REQUIRED_ROLE = {
117        "list": Role.VIEWER,
118        "post": Role.ADMIN,
119    }
120
121    async def on_list(
122        self, data: dict[str, Any] | None,
123    ) -> dict[str, Any]:
124        registry = self._registry_or_none()
125        profiles_out: list[dict[str, Any]] = []
126        if registry is not None:
127            active = registry.active()
128            active_id = active.id if active else None
129            for ref in registry.profiles:
130                profile = _profile_from_ref(registry, ref)
131                store = _open_store(profile)
132                profiles_out.append({
133                    "id": ref.id,
134                    "name": ref.name,
135                    "path": str(profile.profile_dir),
136                    "isActive": ref.id == active_id,
137                    "counts": {
138                        kind: _count(store, kind)
139                        for kind in SUPPORTED_TYPES
140                    },
141                })
142        return {
143            "profiles": profiles_out,
144            "types": list(SUPPORTED_TYPES),
145        }
146
147    async def on_post(self, data: dict[str, Any] | None) -> dict[str, Any]:
148        data = data or {}
149        action = data.get("action")
150        if action == "preview":
151            return await self._preview(data)
152        if action == "apply":
153            return await self._apply(data)
154        raise JsonServiceError(
155            400, f"unknown beanTransfer action: {action!r}",
156        )
157
158    # ---- helpers -----------------------------------------------------
159
160    def _registry_or_none(self) -> ProfileRegistry | None:
161        registry = getattr(self.ctx, "profile_registry", None)
162        return registry if isinstance(registry, ProfileRegistry) else None
163
164    def _require_registry(self) -> ProfileRegistry:
165        registry = self._registry_or_none()
166        if registry is None:
167            raise JsonServiceError(
168                503,
169                "no profile registry loaded (no profiles.xml at settings root)",
170            )
171        return registry
172
173    def _resolve_pair(
174        self, data: dict[str, Any],
175    ) -> tuple[Profile, Profile, ProfileRef, ProfileRef, list[str]]:
176        registry = self._require_registry()
177        src_id = data.get("from")
178        dst_id = data.get("to")
179        if not isinstance(src_id, str) or not isinstance(dst_id, str):
180            raise JsonServiceError(
181                400, "beanTransfer requires 'from' and 'to' profile ids",
182            )
183        if src_id == dst_id:
184            raise JsonServiceError(
185                400, "beanTransfer 'from' and 'to' must differ",
186            )
187        src_ref = registry.by_id(src_id)
188        if src_ref is None:
189            raise JsonServiceError(
190                404, f"no source profile with id {src_id!r}",
191            )
192        dst_ref = registry.by_id(dst_id)
193        if dst_ref is None:
194            raise JsonServiceError(
195                404, f"no target profile with id {dst_id!r}",
196            )
197        raw_types = data.get("types")
198        if raw_types is None:
199            types = list(SUPPORTED_TYPES)
200        elif isinstance(raw_types, list):
201            types = [t for t in raw_types if isinstance(t, str)]
202        else:
203            raise JsonServiceError(
204                400, "'types' must be an array of bean-kind strings",
205            )
206        unknown = [t for t in types if t not in SUPPORTED_TYPES]
207        if unknown:
208            raise JsonServiceError(
209                400,
210                f"unsupported bean types: {unknown!r}. "
211                f"Supported: {list(SUPPORTED_TYPES)}",
212            )
213        if not types:
214            raise JsonServiceError(
215                400, "'types' must not be empty",
216            )
217        return (
218            _profile_from_ref(registry, src_ref),
219            _profile_from_ref(registry, dst_ref),
220            src_ref,
221            dst_ref,
222            types,
223        )
224
225    async def _preview(self, data: dict[str, Any]) -> dict[str, Any]:
226        src, dst, _sref, _dref, types = self._resolve_pair(data)
227        src_store = _open_store(src)
228        dst_store = _open_store(dst)
229        counts: dict[str, int] = {}
230        will_overwrite: dict[str, int] = {}
231        for kind in types:
232            src_names = _existing_names(src_store, kind)
233            counts[kind] = len(src_names)
234            will_overwrite[kind] = len(
235                src_names & _existing_names(dst_store, kind)
236            )
237        return {"counts": counts, "willOverwrite": will_overwrite}
238
239    async def _apply(self, data: dict[str, Any]) -> dict[str, Any]:
240        registry = self._require_registry()
241        src, dst, _sref, dst_ref, types = self._resolve_pair(data)
242        src_store = _open_store(src)
243        active = registry.active()
244        target_is_active = active is not None and dst_ref.id == active.id
245
246        # If the target is the active profile, write through the
247        # ctx.config_store the live server holds — otherwise a fresh
248        # instance we throw away, since no in-memory managers are
249        # attached to the target's beans.
250        if target_is_active and isinstance(self.ctx.config_store, ConfigStore):
251            dst_store = self.ctx.config_store
252        else:
253            dst_store = _open_store(dst)
254
255        copied: dict[str, int] = {}
256        for kind in types:
257            copied[kind] = _copy_kind(src_store, dst_store, kind)
258
259        # Re-populate live managers when we've just written into the
260        # active profile. Otherwise the caller would have to switch
261        # profile (or restart) to see the new beans in the running
262        # connection.
263        if target_is_active:
264            from jmri_server.app import _restore_beans_from_config  # noqa: PLC0415
265            await _restore_beans_from_config(
266                self.ctx.connection, dst_store,
267            )
268
269        return {"copied": copied, "targetIsActive": target_is_active}

392class BlockService(_BeanService):
393    topic = "block"
394
395    def manager(self) -> Manager[Block]:
396        return self.ctx.connection.blocks
397
398    def project(self, bean: Block) -> dict[str, Any]:
399        v = bean.value
400        return _common_named_bean(bean) | {
401            "state": to_wire(int(bean.state)),
402            "value": v if v is None or isinstance(v, str) else str(v),
403            "sensor": bean.sensor.system_name if bean.sensor is not None else None,
404            "reporter": None,
405            "speed": "",
406            "curvature": 0,
407            "direction": 0,
408            "length": 0.0,
409            "permissive": False,
410            "speedLimit": -1.0,
411            "denied": [],
412        }
413
414    async def apply(self, bean: Block, data: dict[str, Any]) -> None:
415        if "value" in data:
416            await bean.set_value(data["value"])
417        if "sensor" in data:
418            sensor_name = data["sensor"]
419            sensor = (
420                self.ctx.connection.sensors.get(sensor_name) if sensor_name else None
421            )
422            bean.attach_sensor(sensor)

 66class ConfigProfileService(JsonService):
 67    topic = "configProfile"
 68    # Profile CRUD is admin-only — a compromised operator token
 69    # shouldn't be able to create/delete profiles or their
 70    # backing data.
 71    REQUIRED_ROLE = {
 72        "list": Role.VIEWER,
 73        "get": Role.VIEWER,
 74        "put": Role.ADMIN,
 75        "delete": Role.ADMIN,
 76        "post": Role.ADMIN,
 77    }
 78
 79    async def on_list(
 80        self, data: dict[str, Any] | None,
 81    ) -> list[dict[str, Any]]:
 82        registry = self._registry_or_none()
 83        if registry is None:
 84            return []
 85        return [_project(ref, registry) for ref in registry.profiles]
 86
 87    async def on_put(self, data: dict[str, Any] | None) -> dict[str, Any]:
 88        data = data or {}
 89        name = data.get("name")
 90        if not isinstance(name, str) or not name:
 91            raise JsonServiceError(400, "configProfile.put requires 'name'")
 92        if not _SAFE_NAME_RE.match(name):
 93            raise JsonServiceError(
 94                400,
 95                "profile name must match [A-Za-z0-9._-]+ "
 96                "(filesystem-safe, no colon or path separators)",
 97            )
 98        registry = self._require_registry()
 99        # Reject duplicates by name — same-name entries would collide
100        # on disk since we store per-name subdirectories.
101        for existing in registry.profiles:
102            if existing.name == name:
103                raise JsonServiceError(
104                    409, f"profile with name {name!r} already exists",
105                )
106        # Auto-generate an id if the caller didn't specify one. JMRI's
107        # own convention is ``<name>.<8-hex>``; matching that keeps
108        # profiles.xml recognizable by a JMRI Java install if the user
109        # ever imports our settings back into it.
110        raw_id = data.get("id")
111        if isinstance(raw_id, str) and raw_id:
112            if registry.by_id(raw_id) is not None:
113                raise JsonServiceError(
114                    409, f"profile id {raw_id!r} already exists",
115                )
116            profile_id = raw_id
117        else:
118            profile_id = f"{name}.{uuid.uuid4().hex[:8]}"
119        ref = ProfileRef(
120            id=profile_id,
121            path=f"settings:{name}/",
122            name=name,
123        )
124        registry.profiles.append(ref)
125        # Persist profiles.xml before touching the filesystem so a
126        # save-failure doesn't leave orphan directories. Create the
127        # profile's dir + an empty PanelPro.xml afterwards.
128        save_profile_registry(registry)
129        profile = ref.to_profile(registry.settings_root)
130        profile.ensure()
131        _write_empty_config(profile)
132        return _project(ref, registry)
133
134    async def on_delete(self, data: dict[str, Any] | None) -> dict[str, Any]:
135        data = data or {}
136        profile_id = data.get("id")
137        if not isinstance(profile_id, str) or not profile_id:
138            raise JsonServiceError(400, "configProfile.delete requires 'id'")
139        registry = self._require_registry()
140        ref = registry.by_id(profile_id)
141        if ref is None:
142            raise JsonServiceError(
143                404, f"no profile with id {profile_id!r}",
144            )
145        active = registry.active()
146        if active is not None and active.id == profile_id:
147            raise JsonServiceError(
148                409,
149                "can't delete the active profile; switch to another profile first",
150            )
151        registry.profiles = [p for p in registry.profiles if p.id != profile_id]
152        save_profile_registry(registry)
153        return {"id": profile_id, "deleted": True}
154
155    # ---- helpers ----------------------------------------------------
156
157    def _registry_or_none(self) -> ProfileRegistry | None:
158        registry = getattr(self.ctx, "profile_registry", None)
159        return registry if isinstance(registry, ProfileRegistry) else None
160
161    def _require_registry(self) -> ProfileRegistry:
162        registry = self._registry_or_none()
163        if registry is None:
164            raise JsonServiceError(
165                503,
166                "no profile registry loaded (no profiles.xml at settings root)",
167            )
168        return registry

 58class ConfigUserService(JsonService):
 59    topic = "configUser"
 60
 61    # User CRUD is admin-only for read + mutate. The self-service
 62    # POST actions do their own finer-grained checks (a viewer can
 63    # change their own password, not anyone else's) so ``post``
 64    # is gated at operator — the action handler enforces the rest.
 65    REQUIRED_ROLE = {
 66        "list": Role.ADMIN,
 67        "get": Role.ADMIN,
 68        "put": Role.ADMIN,
 69        "delete": Role.ADMIN,
 70        "post": Role.VIEWER,   # self-service always available
 71    }
 72
 73    # ---- helpers -------------------------------------------------------
 74
 75    def _auth(self):
 76        auth = getattr(self.ctx, "auth", None)
 77        if auth is None:
 78            raise JsonServiceError(
 79                503,
 80                "auth is disabled on this server (JMRI_AUTH_ENABLED not set)",
 81            )
 82        return auth
 83
 84    def _current_username(self) -> str | None:
 85        return getattr(self.ctx, "_current_username", None)
 86
 87    def _current_role(self) -> Role:
 88        """The current role is passed through dispatch to the JsonService
 89        signature, not to instances — but we can infer from the
 90        current user. When auth is off the middleware sets role=admin;
 91        we mirror that here by treating "no user" as admin."""
 92        current = self._current_username()
 93        if current is None:
 94            return Role.ADMIN
 95        user = self._auth().by_username(current)
 96        return user.role if user else Role.VIEWER
 97
 98    # ---- CRUD ----------------------------------------------------------
 99
100    async def on_list(
101        self, data: dict[str, Any] | None,
102    ) -> list[dict[str, Any]]:
103        auth = self._auth()
104        current = self._current_username()
105        return [
106            _project(u, self_flag=(u.username == current))
107            for u in auth.users
108        ]
109
110    async def on_get(
111        self, data: dict[str, Any] | None,
112    ) -> dict[str, Any]:
113        data = data or {}
114        username = data.get("username")
115        if not isinstance(username, str) or not username:
116            raise JsonServiceError(400, "configUser.get requires 'username'")
117        auth = self._auth()
118        user = auth.by_username(username)
119        if user is None:
120            raise JsonServiceError(404, f"no user {username!r}")
121        return _project(
122            user, self_flag=(user.username == self._current_username()),
123        )
124
125    async def on_put(
126        self, data: dict[str, Any] | None,
127    ) -> dict[str, Any]:
128        """Create or update.
129
130        For a new user: ``password`` is required, ``role`` defaults
131        to viewer. For an existing user: only the fields supplied
132        are updated (role and/or password). Changing your own role
133        would trigger the self-demotion guard — see ``set_role``.
134        """
135        data = data or {}
136        username = data.get("username")
137        password = data.get("password")
138        role_raw = data.get("role")
139        if not isinstance(username, str) or not username:
140            raise JsonServiceError(400, "configUser.put requires 'username'")
141        role: Role | None = None
142        if role_raw is not None:
143            try:
144                role = Role(role_raw)
145            except ValueError:
146                raise JsonServiceError(
147                    400,
148                    f"invalid role {role_raw!r}; must be one of "
149                    f"{[r.value for r in Role]}",
150                ) from None
151
152        auth = self._auth()
153        current = self._current_username() or ""
154        existing = auth.by_username(username)
155        if existing is None:
156            if not isinstance(password, str) or not password:
157                raise JsonServiceError(
158                    400, "creating a user requires 'password'",
159                )
160            user = auth.add_user(
161                username, password,
162                role=role if role is not None else Role.VIEWER,
163            )
164            return _project(user, self_flag=(username == current))
165
166        # Existing user — update only what's supplied.
167        if isinstance(password, str) and password:
168            auth.set_password(username, password)
169        if role is not None and role is not existing.role:
170            try:
171                auth.set_role(
172                    username, role, current_username=current,
173                )
174            except ValueError as e:
175                raise JsonServiceError(409, str(e)) from None
176        return _project(
177            auth.by_username(username),
178            self_flag=(username == current),
179        )
180
181    async def on_delete(
182        self, data: dict[str, Any] | None,
183    ) -> dict[str, Any]:
184        data = data or {}
185        username = data.get("username")
186        if not isinstance(username, str) or not username:
187            raise JsonServiceError(400, "configUser.delete requires 'username'")
188        auth = self._auth()
189        current = self._current_username() or ""
190        try:
191            auth.remove_user(username, current_username=current)
192        except LookupError as e:
193            raise JsonServiceError(404, str(e)) from None
194        except ValueError as e:
195            raise JsonServiceError(409, str(e)) from None
196        return {"username": username, "deleted": True}
197
198    # ---- actions -------------------------------------------------------
199
200    async def on_post(
201        self, data: dict[str, Any] | None,
202    ) -> dict[str, Any]:
203        data = data or {}
204        action = data.get("action")
205        if action == "change_password":
206            return await self._change_password(data)
207        if action == "regenerate_token":
208            return await self._regenerate_token(data)
209        raise JsonServiceError(
210            400, f"unknown configUser action: {action!r}",
211        )
212
213    async def _change_password(
214        self, data: dict[str, Any],
215    ) -> dict[str, Any]:
216        username = data.get("username")
217        new_password = data.get("new_password")
218        if not isinstance(username, str) or not username:
219            raise JsonServiceError(400, "change_password requires 'username'")
220        if not isinstance(new_password, str) or not new_password:
221            raise JsonServiceError(400, "change_password requires 'new_password'")
222        current = self._current_username() or ""
223        if username != current and self._current_role() is not Role.ADMIN:
224            raise JsonServiceError(
225                403,
226                "only admins can change other users' passwords",
227            )
228        auth = self._auth()
229        try:
230            auth.set_password(username, new_password)
231        except LookupError as e:
232            raise JsonServiceError(404, str(e)) from None
233        return {"username": username, "changed": True}
234
235    async def _regenerate_token(
236        self, data: dict[str, Any],
237    ) -> dict[str, Any]:
238        username = data.get("username")
239        if not isinstance(username, str) or not username:
240            raise JsonServiceError(
241                400, "regenerate_token requires 'username'",
242            )
243        current = self._current_username() or ""
244        if username != current and self._current_role() is not Role.ADMIN:
245            raise JsonServiceError(
246                403,
247                "only admins can regenerate other users' tokens",
248            )
249        auth = self._auth()
250        try:
251            token = auth.regenerate_token(username)
252        except LookupError as e:
253            raise JsonServiceError(404, str(e)) from None
254        return {"username": username, "token": token}