from datetime import datetime, timedelta, timezone from typing import Optional from jose import JWTError, jwt from passlib.context import CryptContext from fastapi import Cookie, HTTPException, status, Depends from sqlalchemy.orm import Session from app.database import get_db, settings from app import models pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def hash_password(password: str) -> str: return pwd_context.hash(password) def verify_password(plain: str, hashed: str) -> bool: return pwd_context.verify(plain, hashed) def create_access_token(user_id: int) -> str: expire = datetime.now(timezone.utc) + timedelta(minutes=settings.jwt_expire_minutes) return jwt.encode( {"sub": str(user_id), "exp": expire}, settings.jwt_secret_key, algorithm=settings.jwt_algorithm, ) def get_current_user( access_token: Optional[str] = Cookie(default=None), db: Session = Depends(get_db), ) -> models.User: credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated", ) if not access_token: raise credentials_exception try: payload = jwt.decode(access_token, settings.jwt_secret_key, algorithms=[settings.jwt_algorithm]) user_id: str = payload.get("sub") if user_id is None: raise credentials_exception except JWTError: raise credentials_exception user = db.get(models.User, int(user_id)) if user is None or not user.is_active: raise credentials_exception return user