From 876e1c3f49314e95c98215770ab67ea12f85e057 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rodrigo=20P=C3=A9rez?= Date: Mon, 22 Jun 2026 13:25:12 +0000 Subject: [PATCH] fix: reject malformed peer OPTIONS for monitor and downlink Validate TS1/TS2 static tokens as numeric TG ids so comma-separated garbage (e.g. TS2=730444,VOICE=0) is ignored in topology, downlink, and static bridge refresh. --- src/adn_server/application/report/payloads.py | 104 ++++++++++++++---- .../application/routing/subscription_table.py | 11 +- tests/application/test_report_payloads.py | 94 +++++++++++++++- 3 files changed, 175 insertions(+), 34 deletions(-) diff --git a/src/adn_server/application/report/payloads.py b/src/adn_server/application/report/payloads.py index 3fb38fb..3df48fc 100644 --- a/src/adn_server/application/report/payloads.py +++ b/src/adn_server/application/report/payloads.py @@ -60,6 +60,34 @@ def _peer_connected(peer: dict[str, Any]) -> bool: return peer.get("CONNECTION") == "YES" +def _parse_options_kv(options: Any) -> dict[str, str]: + """Parse RPTO OPTIONS string into upper-case keys (legacy normalisation).""" + if options is None: + return {} + if isinstance(options, bytes): + text = options.decode("utf-8", errors="replace") + else: + text = str(options) + text = text.rstrip("\x00").encode("ascii", "ignore").decode() + text = re.sub(r"['\"]", "", text).strip() + if not text: + return {} + parsed: dict[str, str] = {} + for part in text.split(";"): + part = part.strip() + if "=" not in part: + continue + key, value = part.split("=", 1) + parsed[key.strip().upper()] = value.strip() + for old, new in (("TS1", "TS1_STATIC"), ("TS2", "TS2_STATIC"), ("TIMER", "DEFAULT_UA_TIMER")): + if old in parsed and new not in parsed: + parsed[new] = parsed[old] + return parsed + + +_STATIC_TG_TOKEN_RE = re.compile(r"^\d+$") + + def static_tg_list(value: Any) -> list[str]: """Normalize legacy TS1_STATIC / TS2_STATIC (comma string or list) to string TG ids.""" if value is None: @@ -93,33 +121,61 @@ def normalize_static_tg_slot_lists( return dedupe_static_tg_list(ts1), dedupe_static_tg_list(ts2) -def _parse_options_kv(options: Any) -> dict[str, str]: - """Parse RPTO OPTIONS string into upper-case keys (legacy normalisation).""" - if options is None: - return {} - if isinstance(options, bytes): - text = options.decode("utf-8", errors="replace") - else: - text = str(options) - text = text.rstrip("\x00").encode("ascii", "ignore").decode() - text = re.sub(r"['\"]", "", text).strip() - if not text: - return {} - parsed: dict[str, str] = {} - for part in text.split(";"): - part = part.strip() - if "=" not in part: +def _static_slot_csv_valid(raw: str) -> bool: + """True when every comma-separated token in TS1/TS2 is a numeric talkgroup id.""" + val = raw.strip() + if not val: + return True + if re.search(r"[^\d,]", val): + return False + for token in val.split(","): + token = token.strip() + if not token: continue - key, value = part.split("=", 1) - parsed[key.strip().upper()] = value.strip() - for old, new in (("TS1", "TS1_STATIC"), ("TS2", "TS2_STATIC"), ("TIMER", "DEFAULT_UA_TIMER")): - if old in parsed and new not in parsed: - parsed[new] = parsed[old] - return parsed + if not _STATIC_TG_TOKEN_RE.match(token): + return False + return True + + +def peer_options_pass_only(options: Any) -> bool: + """True when OPTIONS is exclusively ``PASS=password`` (self-service RPTO).""" + parsed = _parse_options_kv(options) + return list(parsed.keys()) == ["PASS"] and bool(parsed["PASS"].strip()) + + +def peer_options_pass_valid(options: Any) -> bool: + """True when ``PASS=`` is absent or is the only key with a non-empty value.""" + parsed = _parse_options_kv(options) + if "PASS" not in parsed: + return True + return peer_options_pass_only(options) + + +def peer_options_static_valid(options: Any) -> bool: + """False when TS1/TS2 static lists contain non-numeric tokens (legacy bridge_master parity). + + ``PASS=`` must be sent alone (self-service); combined PASS+OPTIONS is rejected here. + """ + parsed = _parse_options_kv(options) + if not parsed: + return True + if "PASS" in parsed: + return False + for slot in (1, 2): + for i in range(1, 10): + key = f"TS{slot}_{i}" + if key in parsed and not _STATIC_TG_TOKEN_RE.match(parsed[key].strip()): + return False + for key in ("TS1_STATIC", "TS2_STATIC"): + if key in parsed and not _static_slot_csv_valid(parsed[key]): + return False + return True def parse_peer_options_fields(options: Any) -> dict[str, Any]: """Parse OPTIONS into static lists plus optional ``SINGLE`` / ``TIMER`` when present.""" + if not peer_options_static_valid(options): + return {} parsed = _parse_options_kv(options) if not parsed: return {} @@ -170,6 +226,8 @@ def resolve_peer_single_and_timer( def parse_peer_options_static(options: Any) -> tuple[list[str], list[str]]: """Parse hotspot RPTO OPTIONS (``TS1=…;TS2=…;``) into static TG id lists.""" + if not peer_options_static_valid(options): + return [], [] parsed = _parse_options_kv(options) if not parsed: return [], [] @@ -284,7 +342,7 @@ def _topology_peer_row( if text is not None: row[json_key] = text yaml_cfg = sys_cfg if isinstance(sys_cfg, dict) else {} - if "OPTIONS" in peer: + if "OPTIONS" in peer and peer_options_static_valid(peer.get("OPTIONS")): opt_text = _sanitized_peer_options_text(peer.get("OPTIONS")) if opt_text: row["options"] = opt_text diff --git a/src/adn_server/application/routing/subscription_table.py b/src/adn_server/application/routing/subscription_table.py index be42a8a..407e569 100644 --- a/src/adn_server/application/routing/subscription_table.py +++ b/src/adn_server/application/routing/subscription_table.py @@ -468,14 +468,9 @@ class SubscriptionTableMixin: def _options_static_lists_valid(self, opt_str: bytes | str) -> bool: """Legacy: malformed TS1/TS2 in OPTIONS aborts static bridge refresh.""" - parsed = self._parse_options_string(opt_str) - if not parsed: - return False - for key in ("TS1_STATIC", "TS2_STATIC"): - val = str(parsed.get(key) or "").strip() - if val and re.search(r"[^\d,]", val): - return False - return True + from adn_server.application.report.payloads import peer_options_static_valid + + return peer_options_static_valid(opt_str) def _should_apply_system_single_from_options(self, system_name: str) -> bool: """Whether peer OPTIONS may overwrite system ``SINGLE_MODE`` (legacy single-hotspot only). diff --git a/tests/application/test_report_payloads.py b/tests/application/test_report_payloads.py index 1dfa3e7..22dfdbf 100644 --- a/tests/application/test_report_payloads.py +++ b/tests/application/test_report_payloads.py @@ -37,7 +37,11 @@ from adn_server.application.report import ( ) from adn_server.application.routing.helpers import peer_should_receive_group_voice from adn_server.application.report.payloads import ( + parse_peer_options_fields, parse_peer_options_static, + peer_options_pass_only, + peer_options_pass_valid, + peer_options_static_valid, resolve_peer_single_and_timer, ) from adn_server.domain import bytes_3, bytes_4 @@ -59,6 +63,90 @@ def test_parse_peer_options_static_ts2(): assert ts2 == ["730444"] +def test_malformed_comma_before_voice_rejects_static() -> None: + """WPSD RPTO typo: comma instead of semicolon between TS2 and VOICE (730264101).""" + opts = b"TS2=730444,VOICE=0;TIMER=300;" + assert peer_options_static_valid(opts) is False + assert parse_peer_options_static(opts) == ([], []) + assert parse_peer_options_fields(opts) == {} + + +def test_build_topology_ignores_malformed_peer_options() -> None: + systems = { + "SYSTEM": { + "MODE": "MASTER", + "ENABLED": True, + "DEFAULT_UA_TIMER": 60, + "PEERS": { + bytes_4(730264101): { + "CONNECTION": "YES", + "OPTIONS": b"TS2=730444,VOICE=0;TIMER=300;", + }, + }, + }, + } + doc = build_topology(systems, seq=1) + peer = doc["systems"][0]["peers"][0] + assert "ts2_static" not in peer + assert "ts1_static" not in peer + assert "options" not in peer + assert peer["ua_timer_min"] == 60.0 + + +def test_malformed_options_not_eligible_for_group_voice_downlink() -> None: + peer = {"OPTIONS": b"TS2=730444,VOICE=0;TIMER=300;"} + assert not peer_should_receive_group_voice(peer, 2, 730444, connected_count=8) + + +def test_pass_only_options_valid_for_pass_not_static() -> None: + opts = b"PASS=secret123;" + assert peer_options_pass_only(opts) is True + assert peer_options_pass_valid(opts) is True + assert peer_options_static_valid(opts) is False + assert parse_peer_options_static(opts) == ([], []) + assert parse_peer_options_fields(opts) == {} + + +def test_pass_combined_with_static_options_invalid() -> None: + opts = b"PASS=secret123;TS2=730444;SINGLE=1;" + assert peer_options_pass_valid(opts) is False + assert peer_options_static_valid(opts) is False + assert parse_peer_options_static(opts) == ([], []) + + +def test_pass_with_malformed_static_rejects_all() -> None: + opts = b"PASS=secret123;TS2=730444,VOICE=0;TIMER=300;" + assert peer_options_pass_valid(opts) is False + assert peer_options_static_valid(opts) is False + assert parse_peer_options_static(opts) == ([], []) + + +def test_pass_empty_value_invalid() -> None: + assert peer_options_pass_valid(b"PASS=;") is False + assert peer_options_pass_valid(b"PASS=;TS2=730;") is False + + +def test_topology_pass_only_omits_options_and_static() -> None: + systems = { + "SYSTEM": { + "MODE": "MASTER", + "ENABLED": True, + "DEFAULT_UA_TIMER": 60, + "PEERS": { + bytes_4(730039101): { + "CONNECTION": "YES", + "OPTIONS": b"PASS=secret123;", + }, + }, + }, + } + doc = build_topology(systems, seq=1) + peer = doc["systems"][0]["peers"][0] + assert "options" not in peer + assert "ts1_static" not in peer + assert "ts2_static" not in peer + + def test_resolve_peer_single_and_timer_yaml_defaults() -> None: yaml_cfg = {"SINGLE_MODE": False, "DEFAULT_UA_TIMER": 60} single, timer = resolve_peer_single_and_timer({}, yaml_cfg) @@ -125,7 +213,7 @@ def test_build_topology_exports_peer_options_static() -> None: assert peer["options"] == "TS2=730444;TIMER=15;" -def test_build_topology_omits_pass_from_peer_options() -> None: +def test_build_topology_rejects_combined_pass_and_options() -> None: systems = { "SYSTEM": { "MODE": "MASTER", @@ -140,8 +228,8 @@ def test_build_topology_omits_pass_from_peer_options() -> None: } doc = build_topology(systems, seq=1) peer = doc["systems"][0]["peers"][0] - assert peer["options"] == "TS2=730;SINGLE=1;" - assert "PASS" not in peer["options"] + assert "options" not in peer + assert "ts2_static" not in peer def test_build_topology_exports_master_static_tgs() -> None: