89 lines
3.2 KiB
Python
89 lines
3.2 KiB
Python
import hashlib
|
|
import hmac
|
|
import os
|
|
import secrets
|
|
import time
|
|
from typing import Optional, Tuple
|
|
from fastapi import Request, HTTPException, Depends, status
|
|
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
|
from backend.config import get_setting, update_setting, load_settings
|
|
|
|
security = HTTPBearer(auto_error=False)
|
|
|
|
def hash_password(password: str, salt: Optional[str] = None) -> Tuple[str, str]:
|
|
if not salt:
|
|
salt = secrets.token_hex(16)
|
|
pwd_bytes = password.encode('utf-8')
|
|
salt_bytes = salt.encode('utf-8')
|
|
hash_bytes = hashlib.pbkdf2_hmac('sha256', pwd_bytes, salt_bytes, 100000)
|
|
return hash_bytes.hex(), salt
|
|
|
|
def verify_password(password: str, password_hash: str, salt: str) -> bool:
|
|
new_hash, _ = hash_password(password, salt)
|
|
return hmac.compare_digest(new_hash, password_hash)
|
|
|
|
def create_session_token(username: str = "admin", expires_in_seconds: int = 86400 * 7) -> str:
|
|
secret = get_setting("jwt_secret", "fallback-secret-key-32chars-min-ok")
|
|
expiry = int(time.time()) + expires_in_seconds
|
|
payload = f"{username}:{expiry}"
|
|
signature = hmac.new(secret.encode('utf-8'), payload.encode('utf-8'), hashlib.sha256).hexdigest()
|
|
return f"{payload}:{signature}"
|
|
|
|
def verify_session_token(token: str) -> bool:
|
|
try:
|
|
secret = get_setting("jwt_secret", "fallback-secret-key-32chars-min-ok")
|
|
parts = token.split(":")
|
|
if len(parts) != 3:
|
|
return False
|
|
username, expiry_str, signature = parts
|
|
expiry = int(expiry_str)
|
|
if time.time() > expiry:
|
|
return False
|
|
expected_payload = f"{username}:{expiry_str}"
|
|
expected_sig = hmac.new(secret.encode('utf-8'), expected_payload.encode('utf-8'), hashlib.sha256).hexdigest()
|
|
return hmac.compare_digest(signature, expected_sig)
|
|
except Exception:
|
|
return False
|
|
|
|
def is_setup_required() -> bool:
|
|
settings = load_settings()
|
|
return not bool(settings.get("admin_password_hash"))
|
|
|
|
async def require_auth(
|
|
request: Request,
|
|
credentials: Optional[HTTPAuthorizationCredentials] = Depends(security)
|
|
) -> bool:
|
|
settings = load_settings()
|
|
if not settings.get("auth_enabled", True):
|
|
return True
|
|
|
|
# If setup is required, permit setup endpoints but block others
|
|
if is_setup_required():
|
|
path = request.url.path
|
|
if path in ["/api/auth/status", "/api/auth/setup", "/api/auth/login"]:
|
|
return True
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Initial setup required. Please configure admin password."
|
|
)
|
|
|
|
token = None
|
|
# 1. Check Bearer Authorization header
|
|
if credentials:
|
|
token = credentials.credentials
|
|
|
|
# 2. Check Cookie
|
|
if not token and "session_token" in request.cookies:
|
|
token = request.cookies.get("session_token")
|
|
|
|
# 3. Check Query Parameter (for WebSockets / EventSource / direct download links)
|
|
if not token and "token" in request.query_params:
|
|
token = request.query_params.get("token")
|
|
|
|
if not token or not verify_session_token(token):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Unauthorized. Please log in."
|
|
)
|
|
return True
|