KB8PMY
/
ADN-DMR-Peer-Server
mirror of https://github.com/Amateur-Digital-Network/ADN-DMR-Peer-Server.git
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #69 from ce5rpy/fix/security-download-blocks-reactor

Browse Source 
fix: keep hotspots connected when the security server is unreachable
pull/71/head
ce5rpy 1 month ago committed by GitHub
parent c2bd1c89af 49114feee9
commit 7c2632eb3a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 428 additions and 83 deletions
Show Stats Download Patch File Download Diff File

4
src/adn_server/application/ports.py
Unescape View File

@ -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."""
...

46
src/adn_server/infrastructure/bootstrap/peer_server.py
Unescape View File

@ -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
)
)
d.addErrback(
lambda failure: logger.warning(
"(ALIAS) alias reload failed: %s", failure.getErrorMessage()
)
)
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)
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)")

169
src/adn_server/infrastructure/persistence/alias_loader.py
Unescape View File

@ -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():
"""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 {}
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:
if _blake2bsum(file_path) != expected_checksum:
logger.error("(ALIAS) ID ALIAS MAPPER: problem with blake2bsum of %s file. not updating.", name)
return {}
result = _load_verified(full)
loaded_from_primary = True
except Exception as e:
logger.error("(ALIAS) ID ALIAS MAPPER: problem with blake2bsum of %s file: %s", name, e)
return {}
return self._load_id_json(file_path)
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:
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 {}
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)
return {}
return self._load_server_tsv(path, file_name)
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."""

75
src/adn_server/infrastructure/security/password_download.py
Unescape View File

@ -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:
try:
def _resolve_hostname(hostname: str, timeout: int = DNS_TIMEOUT) -> str | None:
old_timeout = socket.getdefaulttimeout()
try:
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]) -> None:
"""Periodic password file download (every 5 min)."""
def periodic_download(self, config: dict[str, Any]) -> bool:
"""Periodic password file download. Blocking — call off the reactor thread.
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

20
src/adn_server/infrastructure/security/user_passwords_loader.py
Unescape View File

@ -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

58
tests/infrastructure/test_alias_reload_resilience.py
Unescape View File

@ -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"

102
tests/infrastructure/test_security_downloader.py
Unescape View File

@ -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"<html>502 Bad Gateway</html>"
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"}}'

27
tests/infrastructure/test_user_passwords_loader_resilience.py
Unescape View File

@ -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"}
Write Preview
Loading…
Cancel
Save

Powered by TurnKey Linux.