Merge pull request #91 from Amateur-Digital-Network/perf/user-passwords-load-once

perf(auth): load user passwords when they change, not every 10s

src/adn_server/infrastructure/security/password_crypto.py

@@ -66,3 +66,20 @@ def decrypt_password(encrypted_password: Optional[str], key_path: str = "config/
         return decrypted.decode("utf-8")
     except Exception:
         return encrypted_password
+
+
+def decrypt_passwords(
+    encrypted: dict[str, str], key_path: str = "config/encryption_key.secret"
+) -> dict[str, str]:
+    """Decrypt a whole table on one Fernet: the key is read once, not once per entry."""
+    fernet = get_fernet(key_path)
+    decrypted: dict[str, str] = {}
+    for radio_id, password in encrypted.items():
+        if not password:
+            decrypted[str(radio_id)] = ""
+            continue
+        try:
+            decrypted[str(radio_id)] = fernet.decrypt(password.encode("utf-8")).decode("utf-8")
+        except Exception:
+            decrypted[str(radio_id)] = password
+    return decrypted

src/adn_server/infrastructure/security/user_passwords_loader.py

@@ -28,32 +28,27 @@ from __future__ import annotations
 import json
 import logging
 import os
-import time
 from typing import Any
 
 logger = logging.getLogger(__name__)
 
-USER_PASSWORDS_RELOAD_INTERVAL = 10.0
-_last_load = 0.0
-
 
 class UserPasswordsLoader:
-    """Load and cache decrypted user passwords from GLOBAL.USERS_PASS (JSON with 'passwords' dict)."""
+    """Load and cache decrypted user passwords from GLOBAL.USERS_PASS (JSON with 'passwords' dict).
+
+    Loaded at startup and again whenever the security downloader replaces the file,
+    which is the only way it changes. Re-reading it on a timer decrypted the whole
+    table ~30 times between downloads for an identical result.
+    """
 
     def __init__(self, project_root: str) -> None:
         self._project_root = project_root
         self._passwords: dict[str, str] = {}
-        self._config: dict[str, Any] = {}
         self._config_dir = os.path.join(project_root, "config")
         self._key_path = os.path.join(self._config_dir, "encryption_key.secret")
 
     def load(self, config: dict[str, Any]) -> dict[str, str]:
         """Load user_passwords.json from data dir, decrypt each; return passwords dict. Legacy load_user_passwords."""
-        global _last_load
-        self._config = config
-        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,
@@ -68,7 +63,6 @@ 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)
         if not os.path.exists(path):
-            _last_load = now
             if previous:
                 logger.warning("(AUTH) user passwords file missing, keeping cached passwords")
                 return self._passwords
@@ -79,13 +73,9 @@ class UserPasswordsLoader:
                 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
+            from .password_crypto import decrypt_passwords
 
-            new_passwords: dict[str, str] = {}
-            for radio_id, pwd in encrypted.items():
-                new_passwords[str(radio_id)] = decrypt_password(pwd, self._key_path) or ""
-            self._passwords = new_passwords
+            self._passwords = decrypt_passwords(data.get("passwords", {}), self._key_path)
             logger.debug("(AUTH) Loaded %d individual passwords from %s", len(self._passwords), path)
         except (FileNotFoundError, json.JSONDecodeError, ValueError, Exception) as e:
             logger.warning("(AUTH) Could not load user passwords: %s", e)
@@ -94,13 +84,10 @@ class UserPasswordsLoader:
                 self._passwords = previous
             else:
                 self._passwords = {}
-        _last_load = now
         return self._passwords
 
     def get_user_password(self, radio_id: int) -> bytes | None:
         """Return password for radio_id (for login auth); 7-char prefix match like legacy. Legacy get_user_password."""
-        if time.time() - _last_load >= USER_PASSWORDS_RELOAD_INTERVAL and self._config:
-            self.load(self._config)
         radio_id_str = str(radio_id)
         if not radio_id_str.isdigit():
             return None