fix: harden dashboard auth and terminal flows

Tighten terminal websocket auth and proxy trust handling while making file-backed auth/RBAC writes atomic to reduce high-impact security and persistence risks.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

dashboard/backend/auth.py

@@ -1,6 +1,8 @@
 from datetime import datetime, timedelta, timezone
 from typing import Optional
 import time
+import threading
+import tempfile
 import jwt
 from passlib.context import CryptContext
 import pyotp
@@ -78,6 +80,30 @@ import base64
 import hashlib
 from cryptography.fernet import Fernet
 
+_auth_lock = threading.RLock()
+
+
+def _atomic_json_write(path: str, data: dict):
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    fd, temp_path = tempfile.mkstemp(dir=os.path.dirname(path), prefix=".tmp-auth-", suffix=".json")
+    try:
+        with os.fdopen(fd, "w") as f:
+            json.dump(data, f)
+            f.flush()
+            os.fsync(f.fileno())
+        os.replace(temp_path, path)
+    finally:
+        if os.path.exists(temp_path):
+            os.unlink(temp_path)
+
+
+def _update_auth_data(mutator):
+    with _auth_lock:
+        data = _load_auth_data()
+        result = mutator(data)
+        _atomic_json_write(AUTH_FILE, data)
+        return result
+
 def _get_fernet():
     from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
     from cryptography.hazmat.primitives import hashes
@@ -108,18 +134,17 @@ def _decrypt(ciphertext: str) -> str:
         return ciphertext  # fallback for unencrypted legacy values
 
 def _load_auth_data():
-    try:
-        if os.path.exists(AUTH_FILE):
-            with open(AUTH_FILE, "r") as f:
-                return json.load(f)
-    except Exception:
-        pass
-    return {}
+    with _auth_lock:
+        try:
+            if os.path.exists(AUTH_FILE):
+                with open(AUTH_FILE, "r") as f:
+                    return json.load(f)
+        except Exception:
+            pass
+        return {}
 
 def _save_auth_data(data):
-    os.makedirs(os.path.dirname(AUTH_FILE), exist_ok=True)
-    with open(AUTH_FILE, "w") as f:
-        json.dump(data, f)
+    _atomic_json_write(AUTH_FILE, data)
 
 def load_totp_secret():
     encrypted = _load_auth_data().get("totp_secret", "")
@@ -128,17 +153,17 @@ def load_totp_secret():
     return _decrypt(encrypted)
 
 def save_totp_secret(secret: str):
-    data = _load_auth_data()
-    data["totp_secret"] = _encrypt(secret) if secret else ""
-    _save_auth_data(data)
+    def mutator(data):
+        data["totp_secret"] = _encrypt(secret) if secret else ""
+    _update_auth_data(mutator)
 
 def load_password_hash():
     return _load_auth_data().get("password_hash", "") or config.ADMIN_PASSWORD_HASH
 
 def save_password_hash(hashed: str):
-    data = _load_auth_data()
-    data["password_hash"] = hashed
-    _save_auth_data(data)
+    def mutator(data):
+        data["password_hash"] = hashed
+    _update_auth_data(mutator)
         
 def verify_totp(token: str, secret: str = None):
     if secret is None:
@@ -148,41 +173,42 @@ def verify_totp(token: str, secret: str = None):
     totp = pyotp.TOTP(secret)
     if not totp.verify(token):
         return False
-    # Replay protection: reject reuse within same 30s window
-    current_ts = int(time.time()) // 30
-    data = _load_auth_data()
-    if data.get("last_totp_ts") == current_ts:
-        return False
-    data["last_totp_ts"] = current_ts
-    _save_auth_data(data)
-    return True
+
+    def mutator(data):
+        current_ts = int(time.time()) // 30
+        if data.get("last_totp_ts") == current_ts:
+            return False
+        data["last_totp_ts"] = current_ts
+        return True
+
+    return _update_auth_data(mutator)
 
 # Passkey credentials
 def load_passkey_credentials():
     return _load_auth_data().get("passkey_credentials", [])
 
 def save_passkey_credential(cred):
-    data = _load_auth_data()
-    creds = data.get("passkey_credentials", [])
-    creds.append(cred)
-    data["passkey_credentials"] = creds
-    _save_auth_data(data)
+    def mutator(data):
+        creds = data.get("passkey_credentials", [])
+        creds.append(cred)
+        data["passkey_credentials"] = creds
+    _update_auth_data(mutator)
 
 def clear_passkey_credentials():
-    data = _load_auth_data()
-    data["passkey_credentials"] = []
-    _save_auth_data(data)
+    def mutator(data):
+        data["passkey_credentials"] = []
+    _update_auth_data(mutator)
 
 # Token version for refresh token rotation
 def _load_token_version() -> int:
     return _load_auth_data().get("token_version", 0)
 
 def increment_token_version() -> int:
-    data = _load_auth_data()
-    v = data.get("token_version", 0) + 1
-    data["token_version"] = v
-    _save_auth_data(data)
-    return v
+    def mutator(data):
+        v = data.get("token_version", 0) + 1
+        data["token_version"] = v
+        return v
+    return _update_auth_data(mutator)
 
 def verify_token_version(tv: int) -> bool:
     return tv == _load_token_version()