nas-tools/dashboard/backend/auth.py

from datetime import datetime, timedelta
from typing import Optional
import jwt
from passlib.context import CryptContext
import pyotp
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
import config

# Password handling
pwd_context = CryptContext(schemes=["pbkdf2_sha256"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="api/auth/login")

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    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(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    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
    except jwt.PyJWTError:
        raise credentials_exception
    return username

async def get_current_user_ws(token: str):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
    )
    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
    except jwt.PyJWTError:
        raise credentials_exception
    return username

# 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:
        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)

def load_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"] = _encrypt(secret) if secret else ""
    _save_auth_data(data)

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 verify_totp(token: str, secret: str = None):
    if secret is None:
        secret = load_totp_secret()
    if not secret:
        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)