fix(sensors): secure configuration and support multiple modems

This commit is contained in:
yellowcooln committed 2026-09-23 11:27:39 -04:00
1 parent 4958cbf8e9
commit a18122e7f4
9 files changed
+457 -47

No files matched your search

+5 -1
View File
@@ -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("<type>")
class MySensor(SensorBase):
sensor_type = "<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`).
+1 -1
View File
@@ -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"},
+82 -13
View File
@@ -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
+9 -30
View File
@@ -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:
+24 -2
View File
@@ -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:
+14
View File
@@ -132,6 +132,20 @@ def test_stats_app_index_and_default_routing(monkeypatch, tmp_path):
assert app.default("route") == "<html>ok</html>"
def test_root_sensor_api_aliases_are_not_exposed(monkeypatch, tmp_path):
(tmp_path / "index.html").write_text("<html>ok</html>", 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") == "<html>ok</html>"
def test_stats_app_exposes_compiled_ui_favicon(monkeypatch, tmp_path):
favicon = b"compiled-ui-favicon"
(tmp_path / "favicon.ico").write_bytes(favicon)
+10
View File
@@ -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"}})
+245
View File
@@ -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)
+67
View File
@@ -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("<html>UI</html>")
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,
}