syedkhizarrayaz/BM-AI-Analysis-And-Alert-Prioritization-Agent

1# imports2from fastapi import Depends, HTTPException, status3from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm4from passlib.context import CryptContext5from pydantic import BaseModel6from typing import Optional7import jwt8from datetime import datetime, timedelta9from config import SECRET_KEY, USERNAME, PASSWORD10import logging11 12 13# **Setup logging** (add this section to configure the logging)14logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[15    logging.FileHandler("authapi.log"),16    logging.StreamHandler()17])18 19logger = logging.getLogger(__name__)20 21 22logger.info(f"SECRET_KEY: {SECRET_KEY}, USERNAME: {USERNAME}, PASSWORD: {'***' if PASSWORD else None}")23 24# Password and JWT settings25SECRET_KEY = SECRET_KEY26ALGORITHM = "HS256"27ACCESS_TOKEN_EXPIRE_MINUTES = 3000000028 29# Password hashing30pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")31oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")32 33# Dummy database for users34# Only create user entry if credentials are provided35users_db = {}36if USERNAME and PASSWORD:37    try:38        users_db[USERNAME] = {39            "username": USERNAME,40            "full_name": "Admin User",41            "email": "admin@example.com",42            "hashed_password": pwd_context.hash(PASSWORD),43            "disabled": False,44        }45        logger.info(f"User '{USERNAME}' added to database")46    except Exception as e:47        logger.error(f"Failed to create user entry: {str(e)}")48        raise49else:50    logger.warning("USERNAME or PASSWORD not set. Authentication will not work. Please set environment variables.")51 52# token class53class Token(BaseModel):54    access_token: str55    token_type: str56 57# token data class58class TokenData(BaseModel):59    username: Optional[str] = None60 61# user class62class User(BaseModel):63    username: str64    full_name: Optional[str] = None65    email: Optional[str] = None66    disabled: Optional[bool] = None67 68# hash user password class69class UserInDB(User):70    hashed_password: str71 72# verify user password function73def verify_password(plain_password, hashed_password):74    print("plain pass: ", plain_password, "hash pass: ", hashed_password)75    result = pwd_context.verify(plain_password, hashed_password)76    logger.info(f"Password verification result: {result}")77    return result78 79# get the hashed password80def get_password_hash(password):81    hashed_password = pwd_context.hash(password)82    logger.info("Password hashed successfully")83    return hashed_password84 85# get the user from database86def get_user(db, username: str):87    user = db.get(username)88    if user:89        logger.info(f"User {username} found in the database.")90        return UserInDB(**user)91    else:92        logger.warning(f"User {username} not found in the database.")93        return None94 95# authenticate incoming user96def authenticate_user(username: str, password: str):97    user = get_user(users_db, username)98    if not user or not verify_password(password, user.hashed_password):99        logger.warning(f"Authentication failed for user: {username}")100        return False101    logger.info(f"User {username} authenticated successfully.")102    return user103 104# create access token105def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):106    to_encode = data.copy()107    expire = datetime.utcnow() + (expires_delta if expires_delta else timedelta(minutes=15))108    to_encode.update({"exp": expire})109    logger.info(f"Access token created with expiration at {expire}.")110    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)111 112# get the current user login credentials113async def get_current_user(token: str = Depends(oauth2_scheme)):114    credentials_exception = HTTPException(115        status_code=status.HTTP_401_UNAUTHORIZED,116        detail="Could not validate credentials",117        headers={"WWW-Authenticate": "Bearer"},118    )119    try:120        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])121        username: str = payload.get("sub")122        if username is None:123            logger.warning("Token does not contain a valid username.")124            raise credentials_exception125        token_data = TokenData(username=username)126        logger.info(f"Token validated for username: {username}")127    except jwt.PyJWTError as e:128        logger.error(f"JWT decoding failed: {str(e)}")129        raise credentials_exception130    user = get_user(users_db, username=token_data.username)131    if user is None:132        logger.warning("User not found during token validation.")133        raise credentials_exception134    return user135 136# get the active current user137async def get_current_active_user(current_user: User = Depends(get_current_user)):138    if current_user.disabled:139        logger.warning(f"Attempt to use inactive user account: {current_user.username}")140        raise HTTPException(status_code=400, detail="Inactive user")141    logger.info(f"User {current_user.username} is active.")142    return current_user143