Add security improvements: refresh tokens, passkeys, CSP, audit logging, TOTP encryption
Deploy Dashboard / deploy (push) Successful in 1m26s

- Shorten access token to 30min with 7-day refresh token flow
- Add WebAuthn passkey registration and login with TOTP fallback
- Add Content-Security-Policy header
- Add audit logging middleware to /volume1/docker/nas-dashboard/audit.log
- Block /volume1/docker/ in files endpoint
- Encrypt TOTP secret at rest with Fernet (derived from SECRET_KEY)
- New deps: py_webauthn, cryptography
This commit is contained in:
Gan, Jimmy
2026-02-22 10:54:23 +08:00
parent 8392b49bb7
commit 9683cd3165
10 changed files with 484 additions and 46 deletions
+56 -15
View File
@@ -19,13 +19,15 @@ def get_password_hash(password):
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, config.SECRET_KEY, algorithm=config.ALGORITHM)
return encoded_jwt
expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
to_encode.update({"exp": expire, "type": "access"})
return jwt.encode(to_encode, config.SECRET_KEY, algorithm=config.ALGORITHM)
def create_refresh_token(data: dict):
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(minutes=config.REFRESH_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire, "type": "refresh"})
return jwt.encode(to_encode, config.SECRET_KEY, algorithm=config.ALGORITHM)
async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
@@ -35,13 +37,11 @@ async def get_current_user(token: str = Depends(oauth2_scheme)):
)
try:
payload = jwt.decode(token, config.SECRET_KEY, algorithms=[config.ALGORITHM])
username: str = payload.get("sub")
if username is None:
if payload.get("type") != "access":
raise credentials_exception
username: str = payload.get("sub")
if username is None or username != config.ADMIN_USER:
raise credentials_exception
# In a real app, we'd check DB here.
# For this single-user system, we just check if it matches config.
if username != config.ADMIN_USER:
raise credentials_exception
except jwt.PyJWTError:
raise credentials_exception
return username
@@ -53,6 +53,8 @@ async def get_current_user_ws(token: str):
)
try:
payload = jwt.decode(token, config.SECRET_KEY, algorithms=[config.ALGORITHM])
if payload.get("type") != "access":
raise credentials_exception
username: str = payload.get("sub")
if username is None or username != config.ADMIN_USER:
raise credentials_exception
@@ -63,6 +65,26 @@ async def get_current_user_ws(token: str):
# TOTP Persistence
AUTH_FILE = config.VOLUME_ROOT + "/docker/nas-dashboard/auth.json"
import json, os
import base64
import hashlib
from cryptography.fernet import Fernet
def _get_fernet():
key = base64.urlsafe_b64encode(hashlib.sha256(config.SECRET_KEY.encode()).digest())
return Fernet(key)
def _encrypt(plaintext: str) -> str:
if not plaintext:
return ""
return _get_fernet().encrypt(plaintext.encode()).decode()
def _decrypt(ciphertext: str) -> str:
if not ciphertext:
return ""
try:
return _get_fernet().decrypt(ciphertext.encode()).decode()
except Exception:
return ciphertext # fallback for unencrypted legacy values
def _load_auth_data():
try:
@@ -79,11 +101,14 @@ def _save_auth_data(data):
json.dump(data, f)
def load_totp_secret():
return _load_auth_data().get("totp_secret", "") or config.TOTP_SECRET
encrypted = _load_auth_data().get("totp_secret", "")
if not encrypted:
return config.TOTP_SECRET
return _decrypt(encrypted)
def save_totp_secret(secret: str):
data = _load_auth_data()
data["totp_secret"] = secret
data["totp_secret"] = _encrypt(secret) if secret else ""
_save_auth_data(data)
def load_password_hash():
@@ -101,3 +126,19 @@ def verify_totp(token: str, secret: str = None):
return True # 2FA not enabled
totp = pyotp.TOTP(secret)
return totp.verify(token)
# 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 clear_passkey_credentials():
data = _load_auth_data()
data["passkey_credentials"] = []
_save_auth_data(data)