""" Authentication utilities for JWT tokens and password hashing """ import os from datetime import datetime, timedelta from typing import Optional, Dict, Any from jose import JWTError, jwt from passlib.context import CryptContext from fastapi import HTTPException, status, Depends from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from sqlalchemy.orm import Session from models import get_db, User from models.utils import get_user_by_email # Password hashing pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") # JWT settings SECRET_KEY = os.getenv("JWT_SECRET_KEY", "your-super-secret-jwt-key-change-this-in-production") ALGORITHM = os.getenv("JWT_ALGORITHM", "HS256") ACCESS_TOKEN_EXPIRE_MINUTES = int(os.getenv("JWT_ACCESS_TOKEN_EXPIRE_MINUTES", "30")) # HTTP Bearer token scheme security = HTTPBearer() def verify_password(plain_password: str, hashed_password: str) -> bool: """Verify a password against its hash""" 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[str, Any], expires_delta: Optional[timedelta] = None) -> str: """Create a JWT access token""" to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt def verify_token(token: str) -> Optional[Dict[str, Any]]: """Verify and decode a JWT token""" try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) return payload except JWTError: return None def authenticate_user(db: Session, email: str, password: str) -> Optional[User]: """Authenticate a user with email and password""" user = get_user_by_email(db, email) if not user: return None if not verify_password(password, user.password_hash): return None return user def get_current_user( credentials: HTTPAuthorizationCredentials = Depends(security), db: Session = Depends(get_db) ) -> User: """Get the current authenticated user from JWT token""" credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) try: payload = verify_token(credentials.credentials) if payload is None: raise credentials_exception email: str = payload.get("sub") if email is None: raise credentials_exception except JWTError: raise credentials_exception user = get_user_by_email(db, email=email) if user is None: raise credentials_exception # Update last login user.last_login = datetime.utcnow() db.commit() return user def get_current_active_user(current_user: User = Depends(get_current_user)) -> User: """Get the current active user""" # status is stored as a string, not an enum status_value = current_user.status.upper() if isinstance(current_user.status, str) else current_user.status if status_value != "ACTIVE": raise HTTPException(status_code=400, detail="Inactive user") return current_user def require_role(required_roles: list): """Decorator to require specific user roles""" def role_checker(current_user: User = Depends(get_current_active_user)) -> User: # role is stored as a string, not an enum role_value = current_user.role.upper() if isinstance(current_user.role, str) else current_user.role if role_value not in [r.upper() for r in required_roles]: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions" ) return current_user return role_checker # Common role dependencies require_admin = require_role(["admin"]) require_recruiter = require_role(["admin", "recruiter", "hr_manager"]) require_interviewer = require_role(["admin", "recruiter", "hr_manager", "interviewer"])