From 49114feee954e864d99d85d95fbbb3e8d89a6938 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rodrigo=20P=C3=A9rez?= Date: Wed, 19 Aug 2026 09:51:06 -0400 Subject: [PATCH] fix: keep hotspots connected when the security server is unreachable The periodic security download ran blocking HTTP on the reactor thread, so a dead selfcare server stalled RPTPING handling well past PING_TIME * MAX_MISSED and every peer was timed out and forced to reconnect. Move the security and alias downloads to the thread pool, bound the DNS and HTTP timeouts below that budget, and skip a cycle when the previous one is still in flight. Also drop the redundant interval guard that silently stretched the real retry period to a multiple of the loop interval, restore the process-wide socket timeout after a failed DNS lookup, keep existing files and cached passwords when a download returns an empty or invalid payload, and remove the inherited 50Mb cap on subscriber_ids that production was already close to tripping. PASS_SECURITY is no longer written to the log. --- src/adn_server/application/ports.py | 4 +- .../infrastructure/bootstrap/peer_server.py | 46 +++-- .../persistence/alias_loader.py | 179 ++++++++++++++---- .../security/password_download.py | 75 +++++--- .../security/user_passwords_loader.py | 20 +- .../test_alias_reload_resilience.py | 58 ++++++ .../test_security_downloader.py | 102 +++++++++- .../test_user_passwords_loader_resilience.py | 27 +++ 8 files changed, 428 insertions(+), 83 deletions(-) create mode 100644 tests/infrastructure/test_alias_reload_resilience.py create mode 100644 tests/infrastructure/test_user_passwords_loader_resilience.py 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/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_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"}