Multimedika/Bot_Development

1import os2from datetime import timedelta, datetime, timezone3from typing import Annotated4 5from dotenv import load_dotenv6from fastapi import APIRouter, Depends7from fastapi.responses import JSONResponse8from fastapi.security import OAuth2PasswordBearer9from jose import jwt, JWTError10from passlib.context import CryptContext11from sqlalchemy.orm import Session12from starlette import status13 14from db.models import User15from db.database import get_db16 17 18load_dotenv()19 20oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")21 22# Custom OAuth2 request form to accept email, username, password, and role_id23router = APIRouter(prefix="/auth", tags=["auth"])24 25SECRET_KEY = os.getenv("SECRET_KEY")26ALGORITHM = "HS256"27 28bcrypt_context = CryptContext(schemes=["bcrypt"], deprecated="auto")29 30# Database dependency31db_dependency = Annotated[Session, Depends(get_db)]32 33 34def authenticate_user(email: str, password: str, db):35    user = db.query(User).filter(User.email == email).first()36    if not user:37        return False38 39    if not bcrypt_context.verify(password, user.hashed_password):40        return False41    return user42 43 44def create_access_token(45    username: str, name: str, user_id: int, role_id: int, expires_delta: timedelta, email: str46):47    encode = {"sub": username, "name":name, "id": user_id, "role_id": role_id, "email": email}48    expires = datetime.now(timezone.utc) + expires_delta49    encode.update({"exp": expires})50    return jwt.encode(encode, SECRET_KEY, algorithm=ALGORITHM)51 52 53async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):54    try:55        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])56        username: str = payload.get("sub")57        name: str = payload.get("name")58        user_id: int = payload.get("id")59        role_id: int = payload.get("role_id")60        email: str = payload.get("email")61 62        if username is None or user_id is None:63            return JSONResponse(64                status_code=status.HTTP_401_UNAUTHORIZED,65                content="Could not validate user.",66            )67 68        return {"username": username, "name" : name, "id": user_id, "role_id": role_id, "email": email}69 70    except JWTError:71        return JSONResponse(72            status_code=status.HTTP_401_UNAUTHORIZED, content="Could not validate user."73        )74 75user_dependency = Annotated[dict, Depends(get_current_user)]76def check_user_authentication(user: user_dependency):77    """Helper function to check if the user is authenticated."""78    if user is None:79        return JSONResponse(status_code=401, content="Authentication Failed")80    return None81 82def check_admin_authentication(user: user_dependency):83    """Helper function to check if the user is authenticated."""84    if user is None or user.get("role_id") != 1:85        return JSONResponse(status_code=401, content="Authentication Admin Failed")86    return None