fix: reject malformed peer OPTIONS for monitor and downlink

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

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).

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: