from datetime import datetime, timedelta, timezone from typing import Optional from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer from jose import JWTError, jwt from passlib.context import CryptContext from sqlalchemy.orm import Session from . import crud, models, schemas from .database import get_db from .settings import settings # Password Hashing pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") # OAuth2 Scheme oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token") 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() if expires_delta: expire = datetime.now(timezone.utc) + expires_delta else: expire = datetime.now(timezone.utc) + timedelta(minutes=15) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, settings.secret_key, algorithm=settings.algorithm) return encoded_jwt def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)): credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) try: payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm]) username: str = payload.get("sub") if username is None: raise credentials_exception token_data = schemas.TokenData(username=username) except JWTError: raise credentials_exception user = crud.get_user_by_username(db, username=token_data.username) if user is None: raise credentials_exception return user def get_current_active_teacher(current_user: models.User = Depends(get_current_user)): if current_user.role != models.UserRole.teacher: raise HTTPException(status_code=403, detail="Not authorized") return current_user