From a18122e7f4bfe3ee0b43c88b91edc01a1413ed2e Mon Sep 17 00:00:00 2001 From: yellowcooln <12516003+yellowcooln@users.noreply.github.com> Date: Tue, 22 Sep 2026 22:53:07 -0400 Subject: [PATCH] fix(sensors): secure configuration and support multiple modems --- docs/adding_sensors.md | 6 +- repeater/sensors/openhop_modem.py | 2 +- repeater/web/api_endpoints.py | 95 +++++++++-- repeater/web/http_server.py | 39 +---- repeater/web/openapi.yaml | 26 ++- tests/test_http_server_unit.py | 14 ++ tests/test_openhop_modem_sensor.py | 10 ++ tests/test_sensor_manager_api.py | 245 ++++++++++++++++++++++++++++ tests/test_sensor_root_http_auth.py | 67 ++++++++ 9 files changed, 457 insertions(+), 47 deletions(-) create mode 100644 tests/test_sensor_manager_api.py create mode 100644 tests/test_sensor_root_http_auth.py diff --git a/docs/adding_sensors.md b/docs/adding_sensors.md index 3758988..1bd6cf6 100644 --- a/docs/adding_sensors.md +++ b/docs/adding_sensors.md @@ -12,7 +12,7 @@ Sensors in openhop-repeater are self-contained modules that live in `repeater/se | `SensorRegistry` | `repeater/sensors/registry.py` | Maps type strings → sensor classes via `@SensorRegistry.register` | | `SensorManager` | `repeater/sensors/manager.py` | Reads config, imports sensor modules, polls sensors in background | -When `SensorManager` loads a sensor of type `"foo"`, it calls `importlib.import_module("repeater.sensors.foo")`. That import runs the `@SensorRegistry.register("foo")` decorator on your class, making it available. No changes to `__init__.py` or the manager are needed. +When `SensorManager` loads a configured sensor of type `"foo"`, it imports `repeater.sensors.foo`. The Sensor Manager API also imports installed sensor modules before enumerating `SensorRegistry`, so the UI's Add Sensor menu discovers new types without a hardcoded frontend list. Give the class a `_settings_schema` for its dynamic form; installing a module does not automatically create a sensor definition or detect physical hardware. Restart the backend after installing code, then add/configure the sensor in the UI. --- @@ -50,6 +50,9 @@ from .registry import SensorRegistry @SensorRegistry.register("") class MySensor(SensorBase): sensor_type = "" + _settings_schema = [ + {"key": "some_option", "type": "string", "label": "Some Option", "default": "default"}, + ] def __init__(self, name: str, config: Optional[Dict[str, Any]] = None, log=None): super().__init__(name=name, config=config, log=log) @@ -90,6 +93,7 @@ class MySensor(SensorBase): Key rules: - **`sensor_type`** class attribute must match the string passed to `@SensorRegistry.register`. +- Multiple definitions may use the same sensor type (for example two `openhop_modem` instances), but each must have a unique name: the poller caches readings by name. The `pymc_modem` alias remains loadable for old configurations but is not offered as a new type in Sensor Manager. - **`self.settings`** is the `settings:` block from the sensor's config entry (a plain dict). - **`ensure_python_modules`** handles missing dependencies gracefully. Pass a multi-line list of `(import_name, pip_package)` tuples. Returns `False` and logs a warning if any are missing and `auto_install_packages` is `false`; installs them via pip if `true`. Sensor-specific packages belong here — do **not** add them to `pyproject.toml`. - **`_read`** must return a flat `dict[str, Any]`. The base class wraps it in a standard envelope (`name`, `type`, `ok`, `timestamp`, `data`, optional `error`). diff --git a/repeater/sensors/openhop_modem.py b/repeater/sensors/openhop_modem.py index 8e4049d..09c36c4 100644 --- a/repeater/sensors/openhop_modem.py +++ b/repeater/sensors/openhop_modem.py @@ -36,7 +36,7 @@ class OpenHopModemSensor(SensorBase): _settings_schema = [ {"key": "host", "type": "string", "label": "Host", "default": "", "help": "Modem hostname or IP address"}, - {"key": "port", "type": "integer", "label": "Port", "default": 5055, "help": "Modem HTTP port"}, + {"key": "port", "type": "integer", "label": "Port", "default": 80, "help": "Modem HTTP API port (not the radio TCP port)"}, {"key": "scheme", "type": "string", "label": "Scheme", "default": "http", "help": "HTTP or HTTPS"}, {"key": "endpoint", "type": "string", "label": "Endpoint", "default": "/api/stats", "help": "Stats API path"}, {"key": "username", "type": "string", "label": "Username", "default": "admin", "help": "Basic auth username"}, diff --git a/repeater/web/api_endpoints.py b/repeater/web/api_endpoints.py index db63ace..c1adf83 100644 --- a/repeater/web/api_endpoints.py +++ b/repeater/web/api_endpoints.py @@ -8924,17 +8924,21 @@ class APIEndpoints: def sensors_types(self): """Return list of available sensor types with their settings schemas.""" try: + import importlib + import pkgutil + + import repeater.sensors as sensor_package from repeater.sensors import SensorRegistry - from repeater.sensors import bme280 as _bme280 # noqa: F401 - from repeater.sensors import ens210 as _ens210 # noqa: F401 - from repeater.sensors import hardware_stats as _hw # noqa: F401 - from repeater.sensors import ina219 as _ina219 # noqa: F401 - from repeater.sensors import lafvin_ups_3s as _lafvin # noqa: F401 - from repeater.sensors import openhop_modem as _modem # noqa: F401 - from repeater.sensors import pymc_modem as _pymc # noqa: F401 - from repeater.sensors import shtc3 as _shtc3 # noqa: F401 - from repeater.sensors import waveshare_ups_d as _upsd # noqa: F401 - from repeater.sensors import waveshare_ups_e as _upse # noqa: F401 + + # Import installed modules so their registry decorators run. Sensor + # modules should defer optional hardware imports until read time. + for module in pkgutil.iter_modules(sensor_package.__path__): + if module.name.startswith("_") or module.name in {"base", "manager", "registry"}: + continue + try: + importlib.import_module(f"repeater.sensors.{module.name}") + except ImportError as exc: + logger.warning("Skipping unavailable sensor module %s: %s", module.name, exc) type_descriptions = { "bme280": "Temperature, humidity, and barometric pressure", @@ -8950,6 +8954,10 @@ class APIEndpoints: types = [] for sensor_type in SensorRegistry.available_types(): + # Compatibility aliases remain loadable for existing definitions, + # but are not choices for newly configured sensors. + if sensor_type == "pymc_modem": + continue entry = {"type": sensor_type} if sensor_type in type_descriptions: entry["name"] = sensor_type.replace("_", " ").title() @@ -8983,15 +8991,27 @@ class APIEndpoints: if not isinstance(section, dict): section = {} + from copy import deepcopy + definitions = section.get("definitions", []) if not isinstance(definitions, list): definitions = [] + public_definitions = deepcopy(definitions) + for definition in public_definitions: + if isinstance(definition, dict): + # A stable client-side identity survives a rename in the same + # edit session, even after another sensor of this type is added. + definition["_original_name"] = definition.get("name") + if isinstance(definition, dict) and isinstance(definition.get("settings"), dict): + for key, value in definition["settings"].items(): + if key.lower() == "password" and value: + definition["settings"][key] = "*****" return self._success({ "enabled": bool(section.get("enabled", False)), "poll_interval_seconds": float(section.get("poll_interval_seconds", 30.0)), "auto_install_packages": bool(section.get("auto_install_packages", False)), - "definitions": definitions, + "definitions": public_definitions, }) except Exception as e: logger.error(f"Error reading sensor config: {e}", exc_info=True) @@ -9040,18 +9060,67 @@ class APIEndpoints: "auto_install_packages": bool(body.get("auto_install_packages", False)), } - definitions = body.get("definitions", []) + from copy import deepcopy + + definitions = deepcopy(body.get("definitions", [])) if not isinstance(definitions, list): return self._error("definitions must be an array") + old_section = config_yaml.get("sensors", {}) + old_definitions = old_section.get("definitions", []) if isinstance(old_section, dict) else [] + if not isinstance(old_definitions, list): + old_definitions = [] - # Validate each definition + # Names key the poller's reading/interval caches; types may repeat, + # but names must remain unique across all configured sensors. + names = set() + origins = set() for i, defn in enumerate(definitions): if not isinstance(defn, dict): return self._error(f"definitions[{i}] must be an object") + original_name = defn.pop("_original_name", None) + if original_name is not None: + origin = (str(defn.get("type")), original_name) + if not isinstance(original_name, str) or origin in origins: + return self._error(f"definitions[{i}] has an invalid sensor origin") + origins.add(origin) if "type" not in defn: return self._error(f"definitions[{i}] missing required 'type' field") if "name" not in defn: return self._error(f"definitions[{i}] missing required 'name' field") + name = defn["name"] + if not isinstance(name, str) or not name.strip(): + return self._error(f"definitions[{i}] name must be a nonempty string") + if name in names: + return self._error(f"Duplicate sensor name: {name}") + names.add(name) + # Preserve masked credentials by stable sensor identity, not list position. + settings = defn.get("settings") + if isinstance(settings, dict) and settings.get("password") == "*****": + lookup_name = original_name if original_name is not None else defn["name"] + matches = [ + old for old in old_definitions + if isinstance(old, dict) + and old.get("name") == lookup_name + and old.get("type") == defn["type"] + and isinstance(old.get("settings"), dict) + ] + if not matches and original_name is None: + # Legacy clients without an origin can rename only when + # this type is unique on both sides. + old_same_type = [ + old for old in old_definitions + if isinstance(old, dict) and old.get("type") == defn["type"] + and isinstance(old.get("settings"), dict) + ] + new_same_type = [ + new for new in definitions + if isinstance(new, dict) and new.get("type") == defn["type"] + ] + if len(old_same_type) == len(new_same_type) == 1: + matches = old_same_type + if len(matches) != 1 or not matches[0]["settings"].get("password"): + return self._error(f"definitions[{i}] cannot preserve password") + settings["password"] = matches[0]["settings"]["password"] sensors_section["definitions"] = definitions config_yaml["sensors"] = sensors_section diff --git a/repeater/web/http_server.py b/repeater/web/http_server.py index f4b9b56..d88711c 100644 --- a/repeater/web/http_server.py +++ b/repeater/web/http_server.py @@ -491,44 +491,23 @@ class StatsApp: logger.error(f"Error serving index.html: {e}") raise cherrypy.HTTPError(500, "Internal server error") - # ============================================================================ - # Sensor Manager endpoint delegation (APIEndpoints methods → StatsApp) - # ============================================================================ - - @cherrypy.expose - @cherrypy.tools.json_out() - def sensors_types(self): - """Delegate to APIEndpoints.sensors_types().""" - return self.api.sensors_types() - - @cherrypy.expose - @cherrypy.tools.json_out() - def sensors_config(self): - """Delegate to APIEndpoints.sensors_config().""" - return self.api.sensors_config() - - @cherrypy.expose - @cherrypy.tools.json_out() - @cherrypy.tools.json_in() - def sensors_config_update(self): - """Delegate to APIEndpoints.sensors_config_update().""" - return self.api.sensors_config_update() - - @cherrypy.expose - @cherrypy.tools.json_out() - def sensors_read(self): - """Delegate to APIEndpoints.sensors_read().""" - return self.api.sensors_read() - @cherrypy.expose def default(self, *args, **kwargs): """Handle client-side routing - serve index.html for all non-API routes.""" + # Sensor APIs live only under /api, where require_auth is enabled. + # Do not let obsolete root API URLs fall through to the SPA either. + if args and args[0] in { + "sensors_types", + "sensors_config", + "sensors_config_update", + "sensors_read", + }: + raise cherrypy.NotFound() self._resolve_html_dir() # Handle OPTIONS requests for any path if cherrypy.request.method == "OPTIONS": return "" - # Application UI plugins: /plugins/{id}/... if args and args[0] == "plugins": if len(args) < 2: diff --git a/repeater/web/openapi.yaml b/repeater/web/openapi.yaml index 02df24e..4a1da4e 100644 --- a/repeater/web/openapi.yaml +++ b/repeater/web/openapi.yaml @@ -910,7 +910,9 @@ paths: get: tags: [Sensors] summary: List available sensor types - description: Returns all registered sensor types with their settings schemas. + description: | + Discover installed sensor modules and return registered types with their settings schemas. + Compatibility aliases such as pymc_modem remain loadable from existing definitions but are omitted from add-new choices. responses: '200': description: Sensor types list @@ -956,7 +958,11 @@ paths: get: tags: [Sensors] summary: Get current sensor configuration - description: Returns the current sensor definitions from config.yaml. + description: | + Returns current sensor definitions. Nonempty settings.password values are + masked as *****; stored credentials are never included in this response. + Each definition also includes _original_name so the client can preserve + its password safely when renaming it while adding another sensor of the same type. responses: '200': description: Sensor configuration @@ -983,6 +989,9 @@ paths: properties: name: type: string + _original_name: + type: string + description: The stored name used for masked-password round trips; not persisted on update. type: type: string enabled: @@ -998,6 +1007,13 @@ paths: description: | Update sensor definitions and persist to config.yaml. Requires POST body with enabled, poll_interval_seconds, auto_install_packages, and definitions array. + Multiple definitions may use the same type (for example two openhop_modem sensors), but names must be unique. + Existing pymc_modem definitions remain supported for compatibility. + For an existing sensor, password value ***** keeps its stored password; + an empty string clears it, and any other value replaces it. Pass the _original_name + returned by GET when renaming a sensor so password preservation stays tied to its + original identity, even when another sensor of the same type is added. The field + is not persisted; omit it for new definitions. Always returns restart_required=true. security: - BearerAuth: [] @@ -1025,6 +1041,9 @@ paths: properties: name: type: string + _original_name: + type: string + description: Optional origin from GET for an existing definition; not persisted. type: type: string enabled: @@ -1098,6 +1117,9 @@ paths: properties: name: type: string + _original_name: + type: string + description: Optional origin from GET for an existing definition; not persisted. type: type: string enabled: diff --git a/tests/test_http_server_unit.py b/tests/test_http_server_unit.py index 7d6523a..32263fe 100644 --- a/tests/test_http_server_unit.py +++ b/tests/test_http_server_unit.py @@ -132,6 +132,20 @@ def test_stats_app_index_and_default_routing(monkeypatch, tmp_path): assert app.default("route") == "ok" +def test_root_sensor_api_aliases_are_not_exposed(monkeypatch, tmp_path): + (tmp_path / "index.html").write_text("ok", encoding="utf-8") + fake_api = SimpleNamespace(config_manager=object(), docs=lambda: "d") + monkeypatch.setattr(hs, "APIEndpoints", lambda *args, **kwargs: fake_api) + app = hs.StatsApp(config={"web": {"web_path": str(tmp_path)}}) + monkeypatch.setattr(cherrypy, "request", SimpleNamespace(method="GET"), raising=False) + + for name in ("sensors_config", "sensors_config_update", "sensors_types", "sensors_read"): + assert not hasattr(app, name) + with pytest.raises(cherrypy.NotFound): + app.default(name) + assert app.default("sensors") == "ok" + + def test_stats_app_exposes_compiled_ui_favicon(monkeypatch, tmp_path): favicon = b"compiled-ui-favicon" (tmp_path / "favicon.ico").write_bytes(favicon) diff --git a/tests/test_openhop_modem_sensor.py b/tests/test_openhop_modem_sensor.py index be5d7f4..ff0f44c 100644 --- a/tests/test_openhop_modem_sensor.py +++ b/tests/test_openhop_modem_sensor.py @@ -129,6 +129,16 @@ def test_openhop_modem_sensor_exposes_full_flat_diagnostics(monkeypatch): assert data["solar_charge_rate_percent_per_hour"] == 4.5 +def test_modem_sensor_defaults_to_http_port_not_radio_tcp_port(): + defaults = {field["key"]: field["default"] for field in OpenHopModemSensor._settings_schema} + assert defaults["port"] == 80 + defaults["host"] = "modem.local" + sensor = OpenHopModemSensor("modem", {"settings": defaults}) + assert sensor.url == "http://modem.local:80/api/stats" + explicit = OpenHopModemSensor("other", {"settings": {"host": "modem.local", "port": 8080}}) + assert explicit.url == "http://modem.local:8080/api/stats" + + def test_openhop_modem_sensor_accepts_legacy_token_flag(): sensor = OpenHopModemSensor("modem", {"settings": {"base_url": "http://openhop-modem.local"}}) diff --git a/tests/test_sensor_manager_api.py b/tests/test_sensor_manager_api.py new file mode 100644 index 0000000..f597b6e --- /dev/null +++ b/tests/test_sensor_manager_api.py @@ -0,0 +1,245 @@ +"""Sensor manager discovery and secret round-trip contracts.""" + +import copy + +import cherrypy +import yaml + +from repeater.sensors import SensorRegistry +from repeater.web.api_endpoints import APIEndpoints + + +def _api(tmp_path): + config = { + "sensors": { + "enabled": True, + "definitions": [ + { + "name": "modem", + "type": "openhop_modem", + "enabled": True, + "settings": {"host": "modem.local", "password": "private-value"}, + }, + {"name": "other", "type": "bme280", "settings": {"bus_number": 1}}, + ], + }, + } + path = tmp_path / "config.yaml" + path.write_text(yaml.safe_dump(config)) + return APIEndpoints(config=copy.deepcopy(config), config_path=str(path)), path + + +def test_sensor_config_masks_password_without_mutating_runtime(tmp_path): + api, _ = _api(tmp_path) + result = api.sensors_config() + assert result["success"] is True + definitions = result["data"]["definitions"] + assert definitions[0]["settings"]["password"] == "*****" + assert definitions[1]["settings"]["bus_number"] == 1 + assert api.config["sensors"]["definitions"][0]["settings"]["password"] == "private-value" + + +def test_masked_password_round_trip_preserves_secret_and_other_edits(tmp_path): + api, path = _api(tmp_path) + body = api.sensors_config()["data"] + body["definitions"][0]["settings"]["host"] = "new-modem.local" + cherrypy.request.method = "POST" + cherrypy.request.json = body + assert api.sensors_config_update()["success"] is True + saved = yaml.safe_load(path.read_text())["sensors"]["definitions"][0]["settings"] + assert saved["password"] == "private-value" + assert saved["host"] == "new-modem.local" + assert "private-value" not in str(body) + + +def test_renaming_unique_sensor_keeps_its_masked_password(tmp_path): + api, path = _api(tmp_path) + body = api.sensors_config()["data"] + body["definitions"][0]["name"] = "renamed-modem" + cherrypy.request.method = "POST" + cherrypy.request.json = body + assert api.sensors_config_update()["success"] is True + saved = yaml.safe_load(path.read_text())["sensors"]["definitions"][0] + assert saved["name"] == "renamed-modem" + assert saved["settings"]["password"] == "private-value" + + +def test_rename_existing_modem_while_adding_second_preserves_correct_password(tmp_path): + api, path = _api(tmp_path) + api.config["sensors"]["definitions"].reverse() + path.write_text(yaml.safe_dump(api.config)) + body = api.sensors_config()["data"] + body["definitions"][1]["name"] = "renamed-modem" + body["definitions"].append( + { + "name": "new-modem", + "type": "openhop_modem", + "enabled": True, + "settings": {"host": "second.local", "password": "second-private-value"}, + } + ) + cherrypy.request.method = "POST" + cherrypy.request.json = body + result = api.sensors_config_update() + assert result["success"] is True, result.get("error") + saved = yaml.safe_load(path.read_text())["sensors"]["definitions"] + assert saved[1]["settings"]["password"] == "private-value" + assert saved[2]["settings"]["password"] == "second-private-value" + assert all("_original_name" not in d for d in saved) + + +def test_swapping_modem_names_does_not_swap_their_passwords(tmp_path): + api, path = _api(tmp_path) + first = api.config["sensors"]["definitions"][0] + api.config["sensors"]["definitions"] = [ + first, + { + "name": "second", + "type": "openhop_modem", + "enabled": True, + "settings": {"host": "second.local", "password": "second-secret"}, + }, + ] + path.write_text(yaml.safe_dump(api.config)) + body = api.sensors_config()["data"] + body["definitions"][0]["name"] = "second" + body["definitions"][1]["name"] = "modem" + cherrypy.request.method = "POST" + cherrypy.request.json = body + assert api.sensors_config_update()["success"] is True + saved = yaml.safe_load(path.read_text())["sensors"]["definitions"] + assert [d["settings"]["password"] for d in saved] == ["private-value", "second-secret"] + + +def test_duplicate_origin_cannot_copy_another_modems_masked_password(tmp_path): + api, path = _api(tmp_path) + original = path.read_text() + body = api.sensors_config()["data"] + duplicate = copy.deepcopy(body["definitions"][0]) + duplicate["name"] = "new-modem" + duplicate["settings"]["host"] = "new.local" + body["definitions"].append(duplicate) + cherrypy.request.method = "POST" + cherrypy.request.json = body + assert api.sensors_config_update()["success"] is False + assert path.read_text() == original + + +def test_new_password_and_explicit_clear(tmp_path): + api, path = _api(tmp_path) + cherrypy.request.method = "POST" + body = api.sensors_config()["data"] + body["definitions"][0]["settings"]["password"] = "replacement" + cherrypy.request.json = body + assert api.sensors_config_update()["success"] is True + assert ( + yaml.safe_load(path.read_text())["sensors"]["definitions"][0]["settings"]["password"] + == "replacement" + ) + body["definitions"][0]["settings"]["password"] = "" + cherrypy.request.json = body + assert api.sensors_config_update()["success"] is True + assert ( + yaml.safe_load(path.read_text())["sensors"]["definitions"][0]["settings"]["password"] == "" + ) + + +def test_masked_password_on_new_sensor_is_rejected_without_writing(tmp_path): + api, path = _api(tmp_path) + before = path.read_text() + body = api.sensors_config()["data"] + body["definitions"].append( + {"name": "new-modem", "type": "openhop_modem", "settings": {"password": "*****"}} + ) + cherrypy.request.method = "POST" + cherrypy.request.json = body + assert api.sensors_config_update()["success"] is False + assert path.read_text() == before + + +def test_duplicate_sensor_names_are_rejected_without_writing(tmp_path): + api, path = _api(tmp_path) + original = path.read_text() + body = api.sensors_config()["data"] + body["definitions"].append( + {"name": "modem", "type": "openhop_modem", "settings": {"host": "other.local"}} + ) + cherrypy.request.method = "POST" + cherrypy.request.json = body + result = api.sensors_config_update() + assert result["success"] is False + assert path.read_text() == original + + +def test_two_modems_and_legacy_alias_remain_loadable(tmp_path): + from repeater.sensors import SensorManager + + api, path = _api(tmp_path) + section = api.config["sensors"] + section["definitions"] = [ + { + "name": "north", + "type": "openhop_modem", + "enabled": True, + "settings": {"host": "north.local", "password": "north-secret"}, + }, + { + "name": "south", + "type": "openhop_modem", + "enabled": True, + "settings": {"host": "south.local", "password": "south-secret"}, + }, + { + "name": "legacy", + "type": "pymc_modem", + "enabled": True, + "settings": {"host": "legacy.local", "password": "legacy-secret"}, + }, + ] + path.write_text(yaml.safe_dump(api.config)) + masked = api.sensors_config()["data"] + assert [d["settings"]["password"] for d in masked["definitions"]] == ["*****"] * 3 + masked["definitions"].reverse() + cherrypy.request.method = "POST" + cherrypy.request.json = masked + assert api.sensors_config_update()["success"] is True + saved = yaml.safe_load(path.read_text())["sensors"]["definitions"] + assert [(d["name"], d["type"]) for d in saved] == [ + ("legacy", "pymc_modem"), + ("south", "openhop_modem"), + ("north", "openhop_modem"), + ] + assert [d["settings"]["password"] for d in saved] == [ + "legacy-secret", + "south-secret", + "north-secret", + ] + manager = SensorManager(api.config) + assert [sensor.name for sensor in manager.sensors] == ["legacy", "south", "north"] + assert manager.get_summary()["loaded"] == 3 + + +def test_types_endpoint_discovers_new_installed_sensor_module(tmp_path, monkeypatch): + from repeater import sensors + + folder = tmp_path / "extra-sensors" + folder.mkdir() + (folder / "future_sensor.py").write_text( + "from repeater.sensors.registry import SensorRegistry\n" + "from repeater.sensors.base import SensorBase\n" + "@SensorRegistry.register('future_sensor')\n" + "class FutureSensor(SensorBase):\n" + " sensor_type = 'future_sensor'\n" + " _settings_schema = [{'key': 'pin', 'type': 'integer', 'label': 'Pin', 'default': 1}]\n" + " def _read(self): return {'value': 1}\n" + ) + monkeypatch.setattr(sensors, "__path__", list(sensors.__path__) + [str(folder)]) + try: + api, _ = _api(tmp_path) + result = api.sensors_types() + assert result["success"] is True + assert "pymc_modem" not in {entry["type"] for entry in result["data"]["types"]} + future = next(t for t in result["data"]["types"] if t["type"] == "future_sensor") + assert future["settings"][0]["key"] == "pin" + finally: + SensorRegistry._factories.pop("future_sensor", None) diff --git a/tests/test_sensor_root_http_auth.py b/tests/test_sensor_root_http_auth.py new file mode 100644 index 0000000..4dda6ef --- /dev/null +++ b/tests/test_sensor_root_http_auth.py @@ -0,0 +1,67 @@ +"""Actual CherryPy routing: the legacy root sensor URLs must not bypass /api auth.""" + +import json +import subprocess +import sys + + +def test_sensor_endpoints_only_exist_under_authenticated_api(tmp_path): + script = r""" +import json, pathlib, socket, sys, urllib.request, urllib.error +import yaml +from repeater.web import http_server as hs + +root = pathlib.Path(sys.argv[1]) +(root / "index.html").write_text("UI") +config = {"repeater": {"security": {"jwt_secret": "test-secret-only"}}, + "storage": {"storage_dir": str(root / "storage")}, + "web": {"web_path": str(root)}, + "sensors": {"enabled": True, "definitions": [{"name": "modem", "type": "openhop_modem", + "settings": {"password": "synthetic-secret"}}]}} +path = root / "config.yaml" +path.write_text(yaml.safe_dump(config)) +with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] +hs.WEBSOCKET_AVAILABLE = False +server = hs.HTTPStatsServer(host="127.0.0.1", port=port, config=config, config_path=str(path)) +base = f"http://127.0.0.1:{port}" +def status(route, data=None, headers=None): + req = urllib.request.Request(base + route, data=data, headers=headers or {}) + try: + with urllib.request.urlopen(req, timeout=5) as response: + return response.status, response.read() + except urllib.error.HTTPError as exc: + return exc.code, exc.read() +try: + server.start() + before = path.read_bytes() + roots = [status("/" + name)[0] for name in + ("sensors_config", "sensors_types", "sensors_read", "sensors_config_update")] + roots.append(status("/sensors_config_update", data=b'{"enabled":false}', + headers={"Content-Type":"application/json"})[0]) + unauth = status("/api/sensors_config")[0] + token = server.jwt_handler.create_jwt("admin", "test-client") + authorized, payload = status("/api/sensors_config", headers={"Authorization": "Bearer " + token}) + parsed = json.loads(payload) if authorized == 200 else {} + password = parsed.get("data", {}).get("definitions", [{}])[0].get("settings", {}).get("password") + print(json.dumps({"roots":roots, "unauth":unauth, "authorized":authorized, + "masked":password == "*****", "disk_untouched":path.read_bytes() == before})) +finally: + server.stop() +""" + result = subprocess.run( + [sys.executable, "-c", script, str(tmp_path)], + capture_output=True, + text=True, + timeout=30, + check=True, + ) + facts = json.loads(result.stdout.strip().splitlines()[-1]) + assert facts == { + "roots": [404, 404, 404, 404, 404], + "unauth": 401, + "authorized": 200, + "masked": True, + "disk_untouched": True, + }