import os from datetime import datetime, timedelta from typing import Optional import jwt import bcrypt from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer from sqlalchemy.orm import Session from app.db.database import get_db from app.models import models # Password hashing setup # We use bcrypt directly to bypass a known incompatibility between passlib and bcrypt >= 4.0.0 # JWT Setup SECRET_KEY = os.environ.get("SECRET_KEY", "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7") ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 * 7 # 7 days oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login") def verify_password(plain_password: str, hashed_password: str) -> bool: try: return bcrypt.checkpw(plain_password.encode('utf-8'), hashed_password.encode('utf-8')) except Exception: return False def get_password_hash(password: str) -> str: salt = bcrypt.gensalt() return bcrypt.hashpw(password.encode('utf-8'), salt).decode('utf-8') def create_access_token(data: dict, expires_delta: Optional[timedelta] = None): to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=15) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=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, SECRET_KEY, algorithms=[ALGORITHM]) email: str = payload.get("sub") if email is None: raise credentials_exception except jwt.PyJWTError: raise credentials_exception user = db.query(models.User).filter(models.User.email == email).first() if user is None: raise credentials_exception return user