"""Authentication utilities and dependencies.""" import json import logging import os import secrets from datetime import datetime, timedelta, timezone from enum import Enum from typing import Annotated, Optional from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer from jose import JWTError, jwt from passlib.context import CryptContext from pydantic import BaseModel from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from spoolman.database import models from spoolman.database.database import get_db_session logger = logging.getLogger(__name__) # JWT Configuration SECRET_KEY = os.environ.get("SPOOLMAN_AUTH_SECRET_KEY", secrets.token_urlsafe(32)) ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = int(os.environ.get("SPOOLMAN_AUTH_TOKEN_EXPIRE_MINUTES", "1440")) # 24 hours default # Password hashing pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") # OAuth2 scheme - auto_error=False allows unauthenticated access when auth is disabled oauth2_scheme = OAuth2PasswordBearer(tokenUrl="api/v1/auth/login", auto_error=False) # Cached auth enabled state (loaded from database on startup) _auth_enabled_cache: bool = False def is_auth_enabled() -> bool: """Check if authentication is enabled (from cached database setting).""" return _auth_enabled_cache def set_auth_enabled_cache(enabled: bool) -> None: """Update the cached auth enabled state.""" global _auth_enabled_cache _auth_enabled_cache = enabled logger.info("Auth enabled cache updated: %s", enabled) async def load_auth_enabled_from_db(db: AsyncSession) -> bool: """Load auth_enabled setting from database and update cache.""" from spoolman.settings import SETTINGS setting = await db.get(models.Setting, "auth_enabled") if setting is not None: enabled = json.loads(setting.value) else: # Use default from settings definition enabled = json.loads(SETTINGS["auth_enabled"].default) set_auth_enabled_cache(enabled) return enabled class Role(str, Enum): """User roles with increasing permissions.""" VIEWER = "viewer" # Can only view OPERATOR = "operator" # Can view + record usage EDITOR = "editor" # Can view + use + create + edit ADMIN = "admin" # Full access # Role hierarchy - each role includes permissions of roles below it ROLE_HIERARCHY = { Role.VIEWER: 0, Role.OPERATOR: 1, Role.EDITOR: 2, Role.ADMIN: 3, } class Token(BaseModel): """JWT token response.""" access_token: str token_type: str = "bearer" class TokenData(BaseModel): """Data extracted from JWT token.""" username: str role: Role class UserResponse(BaseModel): """User data returned by API (no password).""" id: int username: str role: Role is_active: bool created_at: datetime @staticmethod def from_db(user: models.User) -> "UserResponse": """Convert database model to response model.""" return UserResponse( id=user.id, username=user.username, role=Role(user.role), is_active=user.is_active, created_at=user.created_at, ) def verify_password(plain_password: str, hashed_password: str) -> bool: """Verify a plain password against a hashed password.""" return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password: str) -> str: """Hash a password.""" return pwd_context.hash(password) def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str: """Create a JWT access token.""" to_encode = data.copy() expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)) to_encode.update({"exp": expire}) return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) async def get_user_by_username(db: AsyncSession, username: str) -> Optional[models.User]: """Get a user by username.""" result = await db.execute(select(models.User).where(models.User.username == username)) return result.scalar_one_or_none() async def authenticate_user(db: AsyncSession, username: str, password: str) -> Optional[models.User]: """Authenticate a user by username and password.""" user = await get_user_by_username(db, username) if not user or not verify_password(password, user.hashed_password): return None if not user.is_active: return None return user async def get_current_user( token: Annotated[Optional[str], Depends(oauth2_scheme)], db: Annotated[AsyncSession, Depends(get_db_session)], ) -> Optional[models.User]: """Get the current authenticated user from JWT token. Returns None if auth is disabled or no token provided. Raises HTTPException if token is invalid. """ if not is_auth_enabled(): return None if token is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated", headers={"WWW-Authenticate": "Bearer"}, ) try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) username: str = payload.get("sub") if username is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token", headers={"WWW-Authenticate": "Bearer"}, ) except JWTError: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token", headers={"WWW-Authenticate": "Bearer"}, ) user = await get_user_by_username(db, username) if user is None or not user.is_active: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found or inactive", headers={"WWW-Authenticate": "Bearer"}, ) return user def require_role(minimum_role: Role): """Dependency factory that requires a minimum role level. Usage: @router.post("/spool", dependencies=[Depends(require_role(Role.EDITOR))]) async def create_spool(...): ... """ async def role_checker( current_user: Annotated[Optional[models.User], Depends(get_current_user)], ) -> Optional[models.User]: # If auth is disabled, allow all operations if not is_auth_enabled(): return None if current_user is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated", ) user_role = Role(current_user.role) if ROLE_HIERARCHY[user_role] < ROLE_HIERARCHY[minimum_role]: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=f"Requires {minimum_role.value} role or higher", ) return current_user return role_checker # Common dependencies for different permission levels RequireViewer = Depends(require_role(Role.VIEWER)) RequireOperator = Depends(require_role(Role.OPERATOR)) RequireEditor = Depends(require_role(Role.EDITOR)) RequireAdmin = Depends(require_role(Role.ADMIN))