from datetime import datetime, timedelta from typing import Optional, Any from jose import JWTError, jwt from passlib.context import CryptContext from fastapi import HTTPException, status, Depends from fastapi.security import OAuth2PasswordBearer from app.core.config import settings pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_V1_STR}/auth/login") def verify_password(plain: str, hashed: str) -> bool: return pwd_context.verify(plain, hashed) def hash_password(password: str) -> str: return pwd_context.hash(password) def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str: to_encode = data.copy() expire = datetime.utcnow() + (expires_delta or timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)) to_encode.update({"exp": expire}) return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM) def decode_token(token: str) -> dict: try: return jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]) except JWTError: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) async def get_current_user_payload(token: str = Depends(oauth2_scheme)) -> dict: return decode_token(token) def require_roles(*roles: str): """Dependency factory — require one of the specified roles.""" async def role_checker(payload: dict = Depends(get_current_user_payload)): if payload.get("role") not in roles: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Insufficient permissions" ) return payload return role_checker require_executive_or_above = require_roles("executive", "it_admin", "trustos_admin") require_it_or_above = require_roles("it_admin", "trustos_admin") require_admin = require_roles("trustos_admin")