Merge dev to main: Security improvements and comprehensive test infrastructure #39
@@ -100,7 +100,8 @@ async def get_current_user_ws(token: str):
|
||||
return user_from_access_token(token, include_auth_header=False)
|
||||
|
||||
# TOTP Persistence
|
||||
AUTH_FILE = config.VOLUME_ROOT + "/docker/nas-dashboard/auth.json"
|
||||
def get_auth_file():
|
||||
return config.VOLUME_ROOT + "/docker/nas-dashboard/auth.json"
|
||||
import json, os
|
||||
import base64
|
||||
import hashlib
|
||||
@@ -127,7 +128,7 @@ def _update_auth_data(mutator):
|
||||
with _auth_lock:
|
||||
data = _load_auth_data()
|
||||
result = mutator(data)
|
||||
_atomic_json_write(AUTH_FILE, data)
|
||||
_atomic_json_write(get_auth_file(), data)
|
||||
return result
|
||||
|
||||
def _get_fernet():
|
||||
@@ -162,15 +163,16 @@ def _decrypt(ciphertext: str) -> str:
|
||||
def _load_auth_data():
|
||||
with _auth_lock:
|
||||
try:
|
||||
if os.path.exists(AUTH_FILE):
|
||||
with open(AUTH_FILE, "r") as f:
|
||||
auth_path = get_auth_file()
|
||||
if os.path.exists(auth_path):
|
||||
with open(auth_path, "r") as f:
|
||||
return json.load(f)
|
||||
except Exception:
|
||||
pass
|
||||
return {}
|
||||
|
||||
def _save_auth_data(data):
|
||||
_atomic_json_write(AUTH_FILE, data)
|
||||
_atomic_json_write(get_auth_file(), data)
|
||||
|
||||
def load_totp_secret():
|
||||
encrypted = _load_auth_data().get("totp_secret", "")
|
||||
|
||||
@@ -148,6 +148,8 @@ def test_app(mock_config, temp_auth_file, temp_rbac_file, mock_docker_client):
|
||||
# Patch Docker client before importing main
|
||||
with patch("docker.DockerClient", return_value=mock_docker_client):
|
||||
from main import app
|
||||
# Disable rate limiting during tests to avoid 429 errors
|
||||
app.state.limiter.enabled = False
|
||||
client = TestClient(app)
|
||||
yield client
|
||||
|
||||
|
||||
Reference in New Issue
Block a user