fix: secure modem migration and backup paths

This commit is contained in:
yellowcooln committed 2026-08-24 09:00:50 -04:00
1 parent ada8ba98fb
commit f280bbdc5c
7 files changed
+142 -18

No files matched your search

+4
View File
@@ -29,6 +29,10 @@ services:
volumes:
- ${OPENHOP_CONFIG_VOLUME:-${PYMC_CONFIG_VOLUME:-openhop-repeater-config}}:/etc/openhop_repeater
- ${OPENHOP_DATA_VOLUME:-${PYMC_DATA_VOLUME:-openhop-repeater-data}}:/var/lib/openhop_repeater
# Optional host-device discovery for the setup UI. This read-only mount
# only exposes names under /host/dev; the selected serial device must
# still be granted under `devices:` above before the container can open it.
# - /dev:/host/dev:ro
volumes:
openhop-repeater-config:
+2 -1
View File
@@ -33,7 +33,7 @@ from openhop_core.protocol.packet_utils import (
from repeater.airtime import AirtimeManager
from repeater.data_acquisition import StorageCollector
from repeater.modem_config import normalize_modem_config
from repeater.modem_config import normalize_modem_config, redact_modem_tokens_in_place
from repeater.neighbour_links import NeighbourLinkTracker
from repeater.policy_engine import PolicyDecision, PolicyEngine
@@ -1661,6 +1661,7 @@ class RepeaterHandler(BaseHandler):
def get_stats(self) -> dict:
runtime_config = normalize_modem_config(self.config, warn=False)
redact_modem_tokens_in_place(runtime_config)
uptime_seconds = time.time() - self.start_time
+26 -4
View File
@@ -57,16 +57,38 @@ def _normalize_sensor_and_gps_aliases(config: dict[str, Any]) -> None:
sensor_type = definition.get("type")
if isinstance(sensor_type, str):
normalized_type = sensor_type.strip().lower()
definition["type"] = LEGACY_MODEM_SENSOR_TYPES.get(
normalized_type, normalized_type
)
canonical_type = LEGACY_MODEM_SENSOR_TYPES.get(normalized_type)
if canonical_type is not None:
definition["type"] = canonical_type
gps = config.get("gps")
if isinstance(gps, dict):
source = gps.get("source")
if isinstance(source, str):
normalized_source = source.strip().lower()
gps["source"] = LEGACY_MODEM_GPS_SOURCES.get(normalized_source, normalized_source)
canonical_source = LEGACY_MODEM_GPS_SOURCES.get(normalized_source)
if canonical_source is not None:
gps["source"] = canonical_source
def redact_modem_tokens_in_place(config: dict, *, replacement: Any = None) -> dict:
"""Remove or replace modem TCP tokens in top-level and multi-radio sections."""
if not isinstance(config, dict):
return config
sections = [config.get("modem_tcp")]
radios = config.get("radios")
if isinstance(radios, list):
sections.extend(entry.get("modem_tcp") for entry in radios if isinstance(entry, dict))
for section in sections:
if not isinstance(section, dict) or "token" not in section:
continue
if replacement is None:
section.pop("token", None)
else:
section["token"] = replacement
return config
def normalize_modem_config_in_place(config: dict, *, warn: bool = True) -> dict:
+12 -11
View File
@@ -27,7 +27,11 @@ from repeater.companion.utils import (
)
from repeater.config import resolve_storage_dir
from repeater.handler_helpers.acl import role_name as acl_role_name
from repeater.modem_config import LEGACY_MODEM_RADIO_TYPES, normalize_modem_config
from repeater.modem_config import (
LEGACY_MODEM_RADIO_TYPES,
normalize_modem_config,
redact_modem_tokens_in_place,
)
from repeater.policy_engine import PolicyEngine
from repeater.service_utils import get_buildroot_image_info
from repeater.utils_packet import create_scoped_advert_packet
@@ -1453,7 +1457,7 @@ class APIEndpoints:
dedup = {}
for item in devices:
dev = item.get("device")
if not dev:
if not dev or (os.path.islink(dev) and not os.path.exists(dev)):
continue
try:
target = os.path.realpath(dev)
@@ -1788,6 +1792,7 @@ class APIEndpoints:
try:
stats = self.stats_getter() if self.stats_getter else {}
runtime_config = normalize_modem_config(self.config, warn=False)
redact_modem_tokens_in_place(runtime_config)
# Include active radio configuration in stats so UI can hydrate
# directly from this endpoint without additional config fetches.
stats["radio_type"] = runtime_config.get("radio_type")
@@ -7829,15 +7834,7 @@ class APIEndpoints:
if "identity_key" in rep:
del rep["identity_key"]
tcp_cfg = exported.get("modem_tcp")
if isinstance(tcp_cfg, dict) and "token" in tcp_cfg:
tcp_cfg["token"] = REDACTED_CONFIG_VALUE
for radio_entry in exported.get("radios") or []:
if not isinstance(radio_entry, dict):
continue
nested_tcp_cfg = radio_entry.get("modem_tcp")
if isinstance(nested_tcp_cfg, dict) and "token" in nested_tcp_cfg:
nested_tcp_cfg["token"] = REDACTED_CONFIG_VALUE
redact_modem_tokens_in_place(exported, replacement=REDACTED_CONFIG_VALUE)
# Redact identity keys in companion / room_server configs
for section in ("room_servers", "companions"):
@@ -7991,6 +7988,8 @@ class APIEndpoints:
"kiss",
"modem_usb",
"modem_tcp",
"sensors",
"gps",
"mqtt_brokers",
"mqtt",
"identities",
@@ -8055,6 +8054,8 @@ class APIEndpoints:
"kiss",
"modem_usb",
"modem_tcp",
"sensors",
"gps",
"radio_type",
"mqtt_brokers",
"letsmesh",
+38 -1
View File
@@ -617,16 +617,20 @@ def test_serial_ports_tolerates_broken_host_alias(cherrypy_ctx):
api = _make_api()
broken = "/host/dev/serial/by-id/broken"
realpath = MagicMock(return_value="/host/dev/missing-target")
with (
patch("serial.tools.list_ports.comports", return_value=[]),
patch("os.path.isdir", return_value=True),
patch("glob.glob", side_effect=lambda pattern: [broken] if "by-id" in pattern else []),
patch("os.path.realpath", side_effect=OSError("inaccessible")),
patch("os.path.islink", side_effect=lambda path: path == broken),
patch("os.path.exists", return_value=False),
patch("os.path.realpath", realpath),
):
result = api.serial_ports()
assert result["success"] is True
assert result["data"] == []
realpath.assert_not_called()
def test_config_export_redacts_secrets_and_identity_keys(cherrypy_ctx):
@@ -689,6 +693,29 @@ def test_config_export_full_backup_includes_hex_keys(cherrypy_ctx):
assert result["data"]["meta"]["includes_secrets"] is True
def test_stats_omits_top_level_and_multi_radio_modem_tokens(cherrypy_ctx):
del cherrypy_ctx
config = {
"modem_tcp": {"host": "root.local", "token": "root-secret"},
"radios": [
{
"id": "tcp",
"radio_type": "modem_tcp",
"modem_tcp": {"host": "nested.local", "token": "nested-secret"},
}
],
}
api = _make_api(config)
api.stats_getter = lambda: {}
result = api.stats()
assert result["modem_tcp"] == {"host": "root.local"}
assert result["radios"][0]["modem_tcp"] == {"host": "nested.local"}
assert config["modem_tcp"]["token"] == "root-secret"
assert config["radios"][0]["modem_tcp"]["token"] == "nested-secret"
def test_config_export_canonicalizes_modem_keys_and_redacts_token(cherrypy_ctx):
request, _ = cherrypy_ctx
request.method = "GET"
@@ -782,6 +809,12 @@ def test_config_import_updates_sections_and_preserves_redacted(cherrypy_ctx):
]
},
"radio": {"frequency": 915000000},
"sensors": {
"definitions": [
{"name": "modem", "type": "pymc_modem", "settings": {"host": "m.local"}}
]
},
"gps": {"source": "pymc_modem", "host": "m.local"},
"radio_type": "pymc_usb",
"pymc_tcp": {
"host": "restored.local",
@@ -814,6 +847,8 @@ def test_config_import_updates_sections_and_preserves_redacted(cherrypy_ctx):
"repeater",
"identities",
"radio",
"sensors",
"gps",
"radios",
"radio_type",
"modem_tcp",
@@ -836,6 +871,8 @@ def test_config_import_updates_sections_and_preserves_redacted(cherrypy_ctx):
"port": 6001,
"token": "keep-nested-token",
}
assert api.config["sensors"]["definitions"][0]["type"] == "openhop_modem"
assert api.config["gps"] == {"source": "modem_http", "host": "m.local"}
def test_config_import_preserves_omitted_modem_tokens_but_accepts_replacements(cherrypy_ctx):
+18
View File
@@ -1506,6 +1506,24 @@ class TestStatistics:
stats = handler.get_stats()
assert stats["local_hash"] == f"0x{LOCAL_HASH:02x}"
def test_get_stats_redacts_top_level_and_multi_radio_modem_tokens(self, handler):
handler.config["modem_tcp"] = {"host": "root.local", "token": "root-secret"}
handler.config["radios"] = [
{
"id": "tcp",
"radio_type": "modem_tcp",
"modem_tcp": {"host": "nested.local", "token": "nested-secret"},
}
]
with patch.object(handler, "storage", None):
stats = handler.get_stats()
assert stats["config"]["modem_tcp"] == {"host": "root.local"}
assert stats["config"]["radios"][0]["modem_tcp"] == {"host": "nested.local"}
assert handler.config["modem_tcp"]["token"] == "root-secret"
assert handler.config["radios"][0]["modem_tcp"]["token"] == "nested-secret"
@pytest.mark.asyncio
class TestNeighbourLinkObservation:
+42 -1
View File
@@ -3,7 +3,7 @@ import logging
import yaml
from repeater.modem_config import normalize_modem_config
from repeater.modem_config import normalize_modem_config, redact_modem_tokens_in_place
def test_canonical_modem_config_is_unchanged_and_input_is_not_mutated():
@@ -170,3 +170,44 @@ def test_sensor_and_gps_aliases_migrate_to_canonical_names():
assert normalized["sensors"]["definitions"][0]["type"] == "openhop_modem"
assert normalized["sensors"]["definitions"][1]["type"] == "hardware_stats"
assert normalized["gps"]["source"] == "modem_http"
def test_unrelated_sensor_and_gps_names_keep_original_spelling():
normalized = normalize_modem_config(
{
"sensors": {"definitions": [{"name": "custom", "type": "Vendor_Custom"}]},
"gps": {"source": "Vendor_Source"},
}
)
assert normalized["sensors"]["definitions"][0]["type"] == "Vendor_Custom"
assert normalized["gps"]["source"] == "Vendor_Source"
def test_redact_modem_tokens_handles_top_level_and_multi_radio_without_mutating_others():
config = {
"modem_tcp": {"host": "root.local", "token": "root-secret"},
"radios": [
{"id": "tcp", "modem_tcp": {"host": "nested.local", "token": "nested-secret"}},
{"id": "usb", "modem_usb": {"port": "/dev/ttyACM0"}},
"malformed",
],
}
redact_modem_tokens_in_place(config)
assert config["modem_tcp"] == {"host": "root.local"}
assert config["radios"][0]["modem_tcp"] == {"host": "nested.local"}
assert config["radios"][1]["modem_usb"] == {"port": "/dev/ttyACM0"}
def test_redact_modem_tokens_can_replace_tokens_with_export_sentinel():
config = {
"modem_tcp": {"token": "root-secret"},
"radios": [{"modem_tcp": {"token": "nested-secret"}}],
}
redact_modem_tokens_in_place(config, replacement="*** REDACTED ***")
assert config["modem_tcp"]["token"] == "*** REDACTED ***"
assert config["radios"][0]["modem_tcp"]["token"] == "*** REDACTED ***"