import os from datetime import datetime, timedelta, timezone from typing import Optional import bcrypt import jwt from jwt.exceptions import PyJWTError from fastapi import Depends, HTTPException, status from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from sqlalchemy.orm import Session import database import models SECRET = os.getenv("GYM_JWT_SECRET", "dev-secret-change-me") ALGORITHM = "HS256" EXPIRE_DAYS = int(os.getenv("GYM_JWT_EXPIRE_DAYS", "30")) bearer_scheme = HTTPBearer(auto_error=False) def hash_password(password: str) -> str: return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8") def verify_password(password: str, hashed: str) -> bool: try: return bcrypt.checkpw(password.encode("utf-8"), hashed.encode("utf-8")) except ValueError: return False def create_access_token(user: models.User) -> str: now = datetime.now(timezone.utc) payload = { "sub": str(user.id), "username": user.username, "iat": int(now.timestamp()), "exp": int((now + timedelta(days=EXPIRE_DAYS)).timestamp()), } return jwt.encode(payload, SECRET, algorithm=ALGORITHM) def get_current_user( credentials: Optional[HTTPAuthorizationCredentials] = Depends(bearer_scheme), db: Session = Depends(database.get_db), ) -> models.User: if credentials is None or not credentials.credentials: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated", ) try: payload = jwt.decode(credentials.credentials, SECRET, algorithms=[ALGORITHM]) user_id = int(payload["sub"]) except (PyJWTError, KeyError, ValueError): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token", ) user = db.query(models.User).filter(models.User.id == user_id).first() if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found", ) return user