diff --git a/src/adn_server/application/ports.py b/src/adn_server/application/ports.py index 4cd7451..f4e8d7c 100644 --- a/src/adn_server/application/ports.py +++ b/src/adn_server/application/ports.py @@ -215,8 +215,8 @@ class SecurityDownloader(ABC): ... @abstractmethod - def periodic_download(self, config: dict[str, Any]) -> None: - """Periodic password/encryption download.""" + def periodic_download(self, config: dict[str, Any]) -> bool: + """Periodic password/encryption download. Returns True when passwords file was updated.""" ... diff --git a/src/adn_server/infrastructure/bootstrap/peer_server.py b/src/adn_server/infrastructure/bootstrap/peer_server.py index 60b5625..ce9e166 100644 --- a/src/adn_server/infrastructure/bootstrap/peer_server.py +++ b/src/adn_server/infrastructure/bootstrap/peer_server.py @@ -477,19 +477,19 @@ def run_peer_server( def alias_reload_loop(): logger.debug("(ALIAS) starting alias thread") - try: - peer_ids, subscriber_ids, talkgroup_ids, local_subscriber_ids, server_ids, checksums = ( - alias_loader.load_aliases(config) + # Downloads run in the thread pool (blocking HTTP would stall hotspot pings); + # the config swap is applied back on the reactor thread. + d = threads.deferToThread(alias_loader.load_aliases, config) + d.addCallback( + lambda loaded: DefaultAliasLoader.merge_reload_into_config( + config, alias_loader, *loaded ) - config["_SUB_IDS"] = subscriber_ids - config["_SUB_PROFILES"] = alias_loader.load_subscriber_profiles(config) - config["_PEER_IDS"] = peer_ids - config["_TG_IDS"] = talkgroup_ids - config["_LOCAL_SUBSCRIBER_IDS"] = local_subscriber_ids - config["_SERVER_IDS"] = server_ids - config["CHECKSUMS"] = checksums - except Exception as e: - logger.warning("(ALIAS) alias reload failed: %s", e) + ) + d.addErrback( + lambda failure: logger.warning( + "(ALIAS) alias reload failed: %s", failure.getErrorMessage() + ) + ) task.LoopingCall(alias_reload_loop).start(alias_interval).addErrback(_looping_errback, logger) @@ -733,8 +733,28 @@ def run_peer_server( task.LoopingCall(voice_reload_loop).start(15).addErrback(_looping_errback, logger) logger.info("(VOICE-RELOAD) config file watch active (every 15 seconds)") + # The download is blocking HTTP: run it in the thread pool. On the reactor thread a + # slow/dead security server stalls RPTPING handling past PING_TIME * MAX_MISSED and + # every hotspot gets timed out and reconnects. + security_download_busy = {"value": False} + + def _security_download_done(updated): + security_download_busy["value"] = False + if updated: + user_passwords_loader.load(config) + + def _security_download_error(failure): + security_download_busy["value"] = False + logger.error("(SECURITY) Periodic download failed: %s", failure.getErrorMessage()) + def security_loop(): - security.periodic_download(config) + if security_download_busy["value"]: + logger.warning("(SECURITY) Previous download still running, skipping this cycle") + return + security_download_busy["value"] = True + threads.deferToThread(security.periodic_download, config).addCallbacks( + _security_download_done, _security_download_error + ) task.LoopingCall(security_loop).start(300).addErrback(_looping_errback, logger) logger.info("(SECURITY) Periodic password download task started (every 5 minutes)") diff --git a/src/adn_server/infrastructure/persistence/alias_loader.py b/src/adn_server/infrastructure/persistence/alias_loader.py index 85be503..a04683b 100644 --- a/src/adn_server/infrastructure/persistence/alias_loader.py +++ b/src/adn_server/infrastructure/persistence/alias_loader.py @@ -29,6 +29,7 @@ import csv import hashlib import json import logging +import shutil import ssl import time from pathlib import Path @@ -117,17 +118,64 @@ class DefaultAliasLoader(AliasLoader): sub_file = aliases.get("SUBSCRIBER_FILE", "subscriber_ids.json") tgid_file = aliases.get("TGID_FILE", "talkgroup_ids.json") server_file = aliases.get("SERVER_ID_FILE", "server_ids.tsv") - peer_ids = self._load_id_json_verified(path / peer_file, checksums.get("peer_ids"), "peer_ids") - subscriber_ids = self._load_id_json_verified(path / sub_file, checksums.get("subscriber_ids"), "subscriber_ids") - talkgroup_ids = self._load_id_json_verified(path / tgid_file, checksums.get("talkgroup_ids"), "talkgroup_ids") + peer_ids = self._load_id_dict_with_backup( + path, peer_file, checksums.get("peer_ids"), "peer_ids", + ) + subscriber_ids = self._load_id_dict_with_backup( + path, sub_file, checksums.get("subscriber_ids"), "subscriber_ids", + ) + talkgroup_ids = self._load_id_dict_with_backup( + path, tgid_file, checksums.get("talkgroup_ids"), "talkgroup_ids", + ) local_subscriber_ids = self._load_id_json( path / aliases.get("LOCAL_SUBSCRIBER_FILE", "subscriber_ids.json") ) - server_ids = self._load_server_tsv_verified(path, server_file, checksums.get("server_ids")) - if server_ids: - logger.info("(ALIAS) ID ALIAS MAPPER: server_ids dictionary is available") + server_ids = self._load_server_tsv_with_backup( + path, server_file, checksums.get("server_ids"), + ) return (peer_ids, subscriber_ids, talkgroup_ids, local_subscriber_ids, server_ids, checksums) + @staticmethod + def merge_reload_into_config( + config: dict[str, Any], + alias_loader: AliasLoader, + peer_ids: dict[int, str], + subscriber_ids: dict[int, str], + talkgroup_ids: dict[int, str], + local_subscriber_ids: dict[int, str], + server_ids: dict[str, str], + checksums: dict[str, str], + ) -> None: + """Apply alias reload without wiping in-memory tables on partial download failure.""" + def _keep(key: str, new_val: dict, label: str) -> None: + if new_val: + config[key] = new_val + elif config.get(key): + logger.warning( + "(ALIAS) reload kept previous %s (%d entries)", + label, + len(config[key]), + ) + + _keep("_PEER_IDS", peer_ids, "peer_ids") + if subscriber_ids: + sub = dict(subscriber_ids) + sub[900999] = "D-APRS" + sub[4294967295] = "SC" + config["_SUB_IDS"] = sub + if isinstance(alias_loader, DefaultAliasLoader): + config["_SUB_PROFILES"] = alias_loader.load_subscriber_profiles(config) + elif config.get("_SUB_IDS"): + logger.warning( + "(ALIAS) reload kept previous subscriber_ids (%d entries)", + len(config["_SUB_IDS"]), + ) + _keep("_TG_IDS", talkgroup_ids, "talkgroup_ids") + _keep("_LOCAL_SUBSCRIBER_IDS", local_subscriber_ids, "local_subscriber_ids") + _keep("_SERVER_IDS", server_ids, "server_ids") + if checksums: + config["CHECKSUMS"] = checksums + def _load_checksums(self, path: Path, file_name: str | None) -> dict[str, str]: """Load checksum JSON (legacy load_json of CHECKSUM_FILE). Keys e.g. peer_ids, subscriber_ids, talkgroup_ids, server_ids.""" if not file_name: @@ -162,21 +210,60 @@ class DefaultAliasLoader(AliasLoader): logger.warning("(ALIAS) ID ALIAS MAPPER: %s could not be read: %s", file_name, err) return {} - def _load_id_json_verified( - self, file_path: Path, expected_checksum: str | None, name: str + def _load_id_dict_with_backup( + self, + path: Path, + file_name: str, + expected_checksum: str | None, + name: str, ) -> dict[int, str]: - """Load ID JSON; if expected_checksum given, verify blake2b first (legacy).""" - if not file_path.is_file(): - return {} - if expected_checksum: - try: - if _blake2bsum(file_path) != expected_checksum: - logger.error("(ALIAS) ID ALIAS MAPPER: problem with blake2bsum of %s file. not updating.", name) - return {} - except Exception as e: - logger.error("(ALIAS) ID ALIAS MAPPER: problem with blake2bsum of %s file: %s", name, e) + """Legacy mk_aliases peer/subscriber/tgid load with .bak fallback.""" + full = path / file_name + bak = path / f"{file_name}.bak" + result: dict[int, str] = {} + loaded_from_primary = False + + def _load_verified(target: Path) -> dict[int, str]: + if not target.is_file(): return {} - return self._load_id_json(file_path) + if expected_checksum: + if _blake2bsum(target) != expected_checksum: + raise ValueError("bad checksum") + loaded = self._load_id_json(target) + if not loaded: + raise ValueError("empty or invalid dictionary data") + return loaded + + try: + result = _load_verified(full) + loaded_from_primary = True + except Exception as e: + logger.error( + "(ALIAS) ID ALIAS MAPPER: problem with blake2bsum of %s file. not updating.: %s", + name, + e, + ) + if bak.is_file(): + try: + result = self._load_id_json(bak) + except Exception as f: + logger.error( + "(ALIAS) ID ALIAS MAPPER: Tried backup %s file, but couldn't load that either: %s", + name, + f, + ) + if result: + logger.info("(ALIAS) ID ALIAS MAPPER: %s dictionary is available", name) + if loaded_from_primary and full.is_file(): + try: + shutil.copy(full, bak) + except OSError as g: + logger.info( + "(ALIAS) ID ALIAS MAPPER: couldn't make backup copy of %s file %s", + name, + g, + ) + return result def _load_id_json(self, file_path: Path) -> dict[int, str]: """Load JSON with 'id' -> 'callsign' structure; return {int(id): callsign}.""" @@ -202,22 +289,50 @@ class DefaultAliasLoader(AliasLoader): pass return out - def _load_server_tsv_verified( - self, path: Path, file_name: str, expected_checksum: str | None + def _load_server_tsv_with_backup( + self, + path: Path, + file_name: str, + expected_checksum: str | None, ) -> dict[str, str]: - """Load server_ids TSV; if expected_checksum given, verify blake2b first (legacy).""" + """Legacy mk_aliases server_ids load with .bak fallback.""" full = path / file_name - if not full.is_file(): - return {} - if expected_checksum: - try: + bak = path / f"{file_name}.bak" + result: dict[str, str] = {} + loaded_from_primary = False + + try: + if expected_checksum and full.is_file(): if _blake2bsum(full) != expected_checksum: - logger.error("(ALIAS) ID ALIAS MAPPER: problem with blake2bsum of server_ids file: not updating.") - return {} - except Exception as e: - logger.error("(ALIAS) ID ALIAS MAPPER: problem with blake2bsum of server_ids file: %s", e) - return {} - return self._load_server_tsv(path, file_name) + raise ValueError("bad checksum") + result = self._load_server_tsv(path, file_name) + if full.is_file() and not result: + raise ValueError("empty server_ids") + loaded_from_primary = bool(result) + except Exception as e: + logger.error( + "(ALIAS) ID ALIAS MAPPER: problem with blake2bsum of server_ids file: %s", + e, + ) + if bak.is_file(): + try: + result = self._load_server_tsv(path, f"{file_name}.bak") + except Exception as f: + logger.error( + "(ALIAS) ID ALIAS MAPPER: Tried backup server_ids file, but couldn't load that either: %s", + f, + ) + if result: + logger.info("(ALIAS) ID ALIAS MAPPER: server_ids dictionary is available") + if loaded_from_primary and full.is_file(): + try: + shutil.copy(full, bak) + except OSError as g: + logger.info( + "(ALIAS) ID ALIAS MAPPER: couldn't make backup copy of server_ids file %s", + g, + ) + return result def load_subscriber_profiles(self, config: dict[str, Any]) -> dict[int, dict[str, str]]: """Load {id: {callsign, fname, surname, talker_alias?}} from subscriber JSON files.""" diff --git a/src/adn_server/infrastructure/security/password_download.py b/src/adn_server/infrastructure/security/password_download.py index d4ac36c..47dceb0 100644 --- a/src/adn_server/infrastructure/security/password_download.py +++ b/src/adn_server/infrastructure/security/password_download.py @@ -30,7 +30,7 @@ import os import shutil import socket import tempfile -import time +import json from typing import Any from urllib.error import HTTPError, URLError from urllib.parse import quote @@ -40,19 +40,20 @@ from ...application.ports import SecurityDownloader logger = logging.getLogger(__name__) -DOWNLOAD_INTERVAL_PASSWORDS = 300 +# Bounded so a dead security server cannot stall a caller longer than +# PING_TIME * MAX_MISSED (30s with default config) and time out every hotspot. +DNS_TIMEOUT = 5 +DOWNLOAD_TIMEOUT = 10 -_last_passwords_download = 0.0 _last_passwords_size = 0 _last_passwords_content: bytes | None = None -def _resolve_hostname(hostname: str, timeout: int = 10) -> str | None: +def _resolve_hostname(hostname: str, timeout: int = DNS_TIMEOUT) -> str | None: + old_timeout = socket.getdefaulttimeout() try: - old_timeout = socket.getdefaulttimeout() socket.setdefaulttimeout(timeout) ip = socket.gethostbyname(hostname) - socket.setdefaulttimeout(old_timeout) logger.debug("(SECURITY) Resolved %s to %s", hostname, ip) return ip except socket.gaierror as e: @@ -61,6 +62,9 @@ def _resolve_hostname(hostname: str, timeout: int = 10) -> str | None: except Exception as e: logger.error("(SECURITY) Unexpected error resolving %s: %s", hostname, e) return None + finally: + # Never leak the process-wide default onto unrelated sockets (MySQL, aliases). + socket.setdefaulttimeout(old_timeout) def _build_download_url(config: dict[str, Any], filename: str) -> tuple[str | None, str | None]: @@ -82,14 +86,19 @@ def _build_download_url(config: dict[str, Any], filename: str) -> tuple[str | No return url, url_security +def _log_target(url: str) -> str: + """Host and path only: the query string carries PASS_SECURITY and must never be logged.""" + return url.split("?", 1)[0] + + def _download_file_safely( - url: str, dest_path: str, timeout: int = 60 + url: str, dest_path: str, timeout: int = DOWNLOAD_TIMEOUT ) -> bool: try: fd, temp_path = tempfile.mkstemp() os.close(fd) try: - logger.debug("(SECURITY) Attempting download from: %s", url) + logger.debug("(SECURITY) Attempting download from: %s", _log_target(url)) req = Request(url) req.add_header("User-Agent", "ADN-Systems-DMR/1.0") with urlopen(req, timeout=timeout) as response: @@ -136,13 +145,14 @@ def _download_encryption_key(config: dict[str, Any], config_dir: str) -> bool: return _download_file_safely(url, dest_path) -def _download_user_passwords( - config: dict[str, Any], data_dir: str, force: bool = False -) -> bool: - global _last_passwords_download, _last_passwords_size, _last_passwords_content - now = time.time() - if not force and (now - _last_passwords_download) < DOWNLOAD_INTERVAL_PASSWORDS: - return False +def _download_user_passwords(config: dict[str, Any], data_dir: str) -> bool: + """Fetch user passwords; keep the file on disk untouched unless a valid payload arrives. + + Pacing is the caller's job (the 300s LoopingCall). An extra interval guard here + would reject ticks that arrive a few ms early and silently stretch the real + retry period to a multiple of the loop interval. + """ + global _last_passwords_size, _last_passwords_content g = config.get("GLOBAL", {}) users_pass = (g.get("USERS_PASS") or "user_passwords.json").strip() dest_path = os.path.join(data_dir, users_pass) @@ -151,19 +161,24 @@ def _download_user_passwords( logger.debug("(SECURITY) Security server not configured, skipping passwords download") return False try: - logger.debug("(SECURITY) Downloading passwords from: %s", url) + logger.debug("(SECURITY) Downloading passwords from: %s", _log_target(url)) req = Request(url) req.add_header("User-Agent", "ADN-Systems-DMR/1.0") - with urlopen(req, timeout=60) as response: + with urlopen(req, timeout=DOWNLOAD_TIMEOUT) as response: new_content = response.read() new_size = len(new_content) if new_size == 0: logger.warning("(SECURITY) Downloaded passwords file is empty, keeping existing") - _last_passwords_download = now + return False + try: + parsed = json.loads(new_content) + if not isinstance(parsed, dict) or not isinstance(parsed.get("passwords"), dict): + raise ValueError("invalid passwords JSON shape") + except (json.JSONDecodeError, ValueError) as e: + logger.warning("(SECURITY) Downloaded passwords invalid, keeping existing: %s", e) return False if _last_passwords_content is not None and new_content == _last_passwords_content: logger.debug("(SECURITY) Passwords file unchanged, no update needed") - _last_passwords_download = now return False fd, temp_path = tempfile.mkstemp() os.close(fd) @@ -173,24 +188,19 @@ def _download_user_passwords( shutil.move(temp_path, dest_path) _last_passwords_content = new_content _last_passwords_size = new_size - _last_passwords_download = now logger.info("(SECURITY) Successfully updated passwords file: %s (%d bytes)", dest_path, new_size) return True except HTTPError as e: logger.error("(SECURITY) HTTP error downloading passwords: %s (Code: %d)", e, e.code) - _last_passwords_download = now return False except URLError as e: logger.error("(SECURITY) URL error downloading passwords: %s", e.reason) - _last_passwords_download = now return False except socket.timeout: logger.error("(SECURITY) Timeout downloading passwords") - _last_passwords_download = now return False except Exception as e: logger.error("(SECURITY) Unexpected error downloading passwords: %s", e) - _last_passwords_download = now return False @@ -208,7 +218,7 @@ class DefaultSecurityDownloader(SecurityDownloader): return os.path.join(self._project_root, path) def init_downloads(self, config: dict[str, Any]) -> None: - """One-time init: resolve hostname, download encryption key and passwords (force).""" + """One-time init: resolve hostname, download encryption key and passwords.""" url_security = (config.get("GLOBAL", {}).get("URL_SECURITY") or "").strip() if not url_security: logger.info("(SECURITY) Central security server not configured") @@ -229,12 +239,15 @@ class DefaultSecurityDownloader(SecurityDownloader): config_dir = self._config_dir(config) data_dir = self._data_dir(config) _download_encryption_key(config, config_dir) - _download_user_passwords(config, data_dir, force=True) + _download_user_passwords(config, data_dir) + + def periodic_download(self, config: dict[str, Any]) -> bool: + """Periodic password file download. Blocking — call off the reactor thread. - def periodic_download(self, config: dict[str, Any]) -> None: - """Periodic password file download (every 5 min).""" + Returns True only when the file on disk was replaced with a new valid payload. + """ data_dir = self._data_dir(config) - _download_user_passwords(config, data_dir) + return _download_user_passwords(config, data_dir) class StubSecurityDownloader(SecurityDownloader): @@ -243,5 +256,5 @@ class StubSecurityDownloader(SecurityDownloader): def init_downloads(self, config: dict[str, Any]) -> None: pass - def periodic_download(self, config: dict[str, Any]) -> None: - pass + def periodic_download(self, config: dict[str, Any]) -> bool: + return False diff --git a/src/adn_server/infrastructure/security/user_passwords_loader.py b/src/adn_server/infrastructure/security/user_passwords_loader.py index c90df70..cec2e52 100644 --- a/src/adn_server/infrastructure/security/user_passwords_loader.py +++ b/src/adn_server/infrastructure/security/user_passwords_loader.py @@ -54,6 +54,7 @@ class UserPasswordsLoader: now = time.time() if now - _last_load < USER_PASSWORDS_RELOAD_INTERVAL and self._passwords: return self._passwords + previous = dict(self._passwords) data_dir = os.path.join( self._project_root, (config.get("ALIASES", {}).get("PATH") or "data").rstrip("/"), @@ -66,20 +67,33 @@ class UserPasswordsLoader: ) hash_encrypt = (config.get("GLOBAL", {}).get("HASH_ENCRYPT") or "encryption_key.secret").strip() self._key_path = os.path.join(key_path, hash_encrypt) - self._passwords = {} if not os.path.exists(path): _last_load = now + if previous: + logger.warning("(AUTH) user passwords file missing, keeping cached passwords") + return self._passwords + self._passwords = {} return self._passwords try: with open(path, "r", encoding="utf-8") as f: data = json.load(f) + if not isinstance(data, dict) or not isinstance(data.get("passwords"), dict): + raise ValueError("invalid user_passwords.json shape") encrypted = data.get("passwords", {}) from .password_crypto import decrypt_password + + new_passwords: dict[str, str] = {} for radio_id, pwd in encrypted.items(): - self._passwords[str(radio_id)] = decrypt_password(pwd, self._key_path) or "" + new_passwords[str(radio_id)] = decrypt_password(pwd, self._key_path) or "" + self._passwords = new_passwords logger.debug("(AUTH) Loaded %d individual passwords from %s", len(self._passwords), path) - except (FileNotFoundError, json.JSONDecodeError, Exception) as e: + except (FileNotFoundError, json.JSONDecodeError, ValueError, Exception) as e: logger.warning("(AUTH) Could not load user passwords: %s", e) + if previous: + logger.warning("(AUTH) keeping previous cached passwords (%d entries)", len(previous)) + self._passwords = previous + else: + self._passwords = {} _last_load = now return self._passwords diff --git a/src/adn_server/infrastructure/twisted_adapters/udp_hbp.py b/src/adn_server/infrastructure/twisted_adapters/udp_hbp.py index ca181aa..1675633 100644 --- a/src/adn_server/infrastructure/twisted_adapters/udp_hbp.py +++ b/src/adn_server/infrastructure/twisted_adapters/udp_hbp.py @@ -436,12 +436,22 @@ class HBPProtocol(DatagramProtocol): to blast the call to peers it can't possibly be for. Returns a 1-tuple with the exact peer_id when it's still connected here. + TG/ID 4000 is the dynamic-TG reset control code (see + ``_handle_tg4000_packet``/``routing_use_cases.dmrd_received``), never + a real subscriber -- it can never appear in _SUB_MAP, so it would + otherwise always fall into the "unknown destination" broadcast + fallback above and get blasted to every peer. That fallback runs + before dmrd_received's own dst_id == 4000 guard ever sees the + packet, so it must be special-cased here too. + The report_slot / "SYSTEM-N" monitor display name is NOT used for this — it's cosmetic and can be reassigned to a different peer across refreshes (self-service peers without a stable report_slot fall back to a sorted-by-id allocation recomputed each time). The raw peer_id is what's actually stable. """ + if int_id(dst_id) == 4000: + return () sub_map = self._CONFIG.get("_SUB_MAP") if not sub_map: return None diff --git a/tests/infrastructure/test_alias_reload_resilience.py b/tests/infrastructure/test_alias_reload_resilience.py new file mode 100644 index 0000000..221cddf --- /dev/null +++ b/tests/infrastructure/test_alias_reload_resilience.py @@ -0,0 +1,58 @@ +# ADN DMR Peer Server - alias reload resilience when selfcare download fails + +from __future__ import annotations + +import json +from pathlib import Path + +from adn_server.infrastructure.persistence.alias_loader import DefaultAliasLoader, try_download + + +def _write_subscriber_file(path: Path, file_name: str, rid: int, callsign: str) -> None: + data = {"subscribers": [{"id": rid, "callsign": callsign}]} + (path / file_name).write_text(json.dumps(data), encoding="utf-8") + + +def test_try_download_failure_does_not_erase_existing_file(tmp_path: Path) -> None: + file_name = "subscriber_ids.json" + full = tmp_path / file_name + full.write_bytes(b'{"subscribers":[{"id":7300391,"callsign":"CE5RPY"}]}') + # stale_sec=0 forces download attempt; bad URL simulates selfcare down + result = try_download(tmp_path, file_name, "http://127.0.0.1:1/nope.json", stale_sec=0) + assert "could not be downloaded" in result or "IOError" in result + assert full.read_bytes().startswith(b"{") + + +def test_merge_reload_keeps_previous_sub_ids_on_empty_reload() -> None: + loader = DefaultAliasLoader() + config = { + "_SUB_IDS": {7300391: "CE5RPY"}, + "_PEER_IDS": {730039101: "CE5RPY"}, + "ALIASES": {"PATH": "."}, + } + DefaultAliasLoader.merge_reload_into_config( + config, + loader, + {}, + {}, + {}, + {}, + {}, + {}, + ) + assert config["_SUB_IDS"] == {7300391: "CE5RPY"} + assert config["_PEER_IDS"] == {730039101: "CE5RPY"} + + +def test_load_id_dict_with_backup_uses_bak_on_checksum_mismatch(tmp_path: Path) -> None: + loader = DefaultAliasLoader() + file_name = "subscriber_ids.json" + _write_subscriber_file(tmp_path, file_name, 1111111, "BAD") + _write_subscriber_file(tmp_path, f"{file_name}.bak", 7300391, "GOOD") + loaded = loader._load_id_dict_with_backup( + tmp_path, + file_name, + "deadbeef", + "subscriber_ids", + ) + assert loaded.get(7300391) == "GOOD" diff --git a/tests/infrastructure/test_hbp_private_call_targeting.py b/tests/infrastructure/test_hbp_private_call_targeting.py index 3f9f951..33d493e 100644 --- a/tests/infrastructure/test_hbp_private_call_targeting.py +++ b/tests/infrastructure/test_hbp_private_call_targeting.py @@ -31,6 +31,8 @@ parity) -- see test_hbp_repeat_private_call.py.""" from __future__ import annotations +import dataclasses + from tests.harness.deterministic import DeterministicScenario, PacketSpec from tests.support.hbp_repeat_stack import build_hbp_repeat_stack @@ -110,6 +112,24 @@ def test_private_call_not_delivered_locally_when_known_on_different_system() -> assert not stack.transport.for_addr(_ADDR_OTHER), "destination known elsewhere must not repeat locally" +def test_private_call_to_4000_not_broadcast() -> None: + """TG/ID 4000 is the dynamic-TG reset control code, never a real subscriber -- + it can never appear in _SUB_MAP, so without a special case it always fell into + the "unknown destination" broadcast fallback and reached every connected peer + instead of being stopped at the server (dmrd_received's own dst_id == 4000 + guard runs too late: _pvt_repeat_targets/the REPEAT loop already ran).""" + stack = build_hbp_repeat_stack(talker_alias=True) + stack.register_peer(_PEER_TX, _ADDR_TX, options="TS2=7304;") + stack.register_peer(_PEER_RX, _ADDR_RX, options="TS2=7304;") + stack.register_peer(_PEER_OTHER, _ADDR_OTHER, options="TS2=7304;") + + base = dataclasses.replace(_private_spec(), dst_id=4000) + _fire_private_call(stack, base) + + assert not stack.transport.for_addr(_ADDR_RX), "TG 4000 must not be broadcast to any peer" + assert not stack.transport.for_addr(_ADDR_OTHER), "TG 4000 must not be broadcast to any peer" + + def test_sub_map_entries_for_peer_purged_on_reconnect() -> None: """Once a hotspot finishes (re)logging in, any SUB_MAP entry pointing at it is stale -- the radio may have moved to a different hotspot while this one was diff --git a/tests/infrastructure/test_security_downloader.py b/tests/infrastructure/test_security_downloader.py index 7944863..56c2730 100644 --- a/tests/infrastructure/test_security_downloader.py +++ b/tests/infrastructure/test_security_downloader.py @@ -18,14 +18,21 @@ # Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA ############################################################################### -"""SecurityDownloader stub and password_crypto round-trip.""" +"""SecurityDownloader stub, password_crypto round-trip and offline-server resilience.""" from __future__ import annotations +import socket +from urllib.error import URLError + import pytest +from adn_server.infrastructure.security import password_download from adn_server.infrastructure.security.password_crypto import decrypt_password -from adn_server.infrastructure.security.password_download import StubSecurityDownloader +from adn_server.infrastructure.security.password_download import ( + DefaultSecurityDownloader, + StubSecurityDownloader, +) pytest.importorskip("cryptography") from cryptography.fernet import Fernet # noqa: E402 @@ -45,3 +52,94 @@ def test_stub_security_downloader_is_noop() -> None: config = {"GLOBAL": {"URL_SECURITY": "", "PORT_SECURITY": "", "PASS_SECURITY": ""}} stub.init_downloads(config) stub.periodic_download(config) + + +def test_resolve_hostname_restores_default_socket_timeout() -> None: + """A failed lookup must not leave its timeout on every other socket in the process.""" + before = socket.getdefaulttimeout() + assert password_download._resolve_hostname("no-such-host.invalid") is None + assert socket.getdefaulttimeout() == before + + +def _offline_config(tmp_path) -> dict: + return { + "GLOBAL": { + "URL_SECURITY": "127.0.0.1", + "PORT_SECURITY": "1", + "PASS_SECURITY": "s3cr3t", + "USERS_PASS": "user_passwords.json", + "CONFIG_PATH": "config", + }, + "ALIASES": {"PATH": "data"}, + } + + +def test_offline_server_keeps_existing_passwords_and_retries_every_call( + tmp_path, monkeypatch +) -> None: + data_dir = tmp_path / "data" + data_dir.mkdir() + pw_file = data_dir / "user_passwords.json" + pw_file.write_text('{"passwords": {"7300391": "enc"}}', encoding="utf-8") + + attempts = [] + + def _refused(*args, **kwargs): + attempts.append(1) + raise URLError("Connection refused") + + monkeypatch.setattr(password_download, "urlopen", _refused) + downloader = DefaultSecurityDownloader(str(tmp_path)) + config = _offline_config(tmp_path) + + assert downloader.periodic_download(config) is False + assert downloader.periodic_download(config) is False + # No internal interval gate: every scheduled cycle really attempts the download. + assert len(attempts) == 2 + assert pw_file.read_text(encoding="utf-8") == '{"passwords": {"7300391": "enc"}}' + + +def test_pass_security_is_never_logged(tmp_path, monkeypatch, caplog) -> None: + """LOG_LEVEL DEBUG in production must not write PASS_SECURITY into adn-server.log.""" + (tmp_path / "data").mkdir() + (tmp_path / "config").mkdir() + + def _refused(*args, **kwargs): + raise URLError("Connection refused") + + monkeypatch.setattr(password_download, "urlopen", _refused) + downloader = DefaultSecurityDownloader(str(tmp_path)) + config = _offline_config(tmp_path) + + with caplog.at_level("DEBUG"): + downloader.init_downloads(config) + downloader.periodic_download(config) + + assert caplog.text, "expected the downloader to log something at DEBUG" + assert config["GLOBAL"]["PASS_SECURITY"] not in caplog.text + assert "pass=" not in caplog.text + # The host is still there so operators can tell which server failed. + assert "127.0.0.1" in caplog.text + + +def test_invalid_payload_does_not_overwrite_existing_file(tmp_path, monkeypatch) -> None: + data_dir = tmp_path / "data" + data_dir.mkdir() + pw_file = data_dir / "user_passwords.json" + pw_file.write_text('{"passwords": {"7300391": "enc"}}', encoding="utf-8") + + class _Response: + def read(self): + return b"502 Bad Gateway" + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + monkeypatch.setattr(password_download, "urlopen", lambda *a, **k: _Response()) + downloader = DefaultSecurityDownloader(str(tmp_path)) + + assert downloader.periodic_download(_offline_config(tmp_path)) is False + assert pw_file.read_text(encoding="utf-8") == '{"passwords": {"7300391": "enc"}}' diff --git a/tests/infrastructure/test_user_passwords_loader_resilience.py b/tests/infrastructure/test_user_passwords_loader_resilience.py new file mode 100644 index 0000000..d36de59 --- /dev/null +++ b/tests/infrastructure/test_user_passwords_loader_resilience.py @@ -0,0 +1,27 @@ +# ADN DMR Peer Server - user passwords loader keeps cache on bad reload + +from __future__ import annotations + +import json +from pathlib import Path + +from adn_server.infrastructure.security.user_passwords_loader import UserPasswordsLoader + + +def test_load_keeps_cached_passwords_when_file_becomes_invalid(tmp_path: Path) -> None: + data_dir = tmp_path / "data" + data_dir.mkdir() + pw_file = data_dir / "user_passwords.json" + pw_file.write_text( + json.dumps({"passwords": {"7300391": "enc"}}), + encoding="utf-8", + ) + loader = UserPasswordsLoader(str(tmp_path)) + config = { + "ALIASES": {"PATH": "data/"}, + "GLOBAL": {"USERS_PASS": "user_passwords.json", "CONFIG_PATH": "config"}, + } + loader._passwords = {"7300391": "secret"} + pw_file.write_text("not json", encoding="utf-8") + result = loader.load(config) + assert result == {"7300391": "secret"}