import logging from datetime import datetime, timedelta, timezone from typing import Annotated from uuid import uuid4 import jwt from fastapi import Depends, HTTPException, Request, Response, status from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from jb.config import settings from jb.dependencies import get_gr_api_manager from jb.managers.gr_api import GRApiManager from jb.models.auth import User bearer = HTTPBearer(auto_error=False) logger = logging.getLogger(__name__) SESSION_COOKIE_NAME = "jb_session" JWT_ISSUER = "jamesbillings67" JWT_AUDIENCE = "jamesbillings67" def get_authenticated_user( request: Request, credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer)], gr_api: Annotated[GRApiManager, Depends(get_gr_api_manager)], ) -> User: """FastAPI dependency for endpoints requiring a valid session.""" if settings.session_jwt_secret is None: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Account session signing key is not configured", ) token = request.cookies.get(SESSION_COOKIE_NAME) if credentials is not None and credentials.scheme.lower() == "bearer": token = credentials.credentials if token is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing session token", headers={"WWW-Authenticate": "Bearer"}, ) try: claims = jwt.decode( token, settings.session_jwt_secret.get_secret_value(), algorithms=["HS256"], issuer=JWT_ISSUER, audience=JWT_AUDIENCE, options={"require": ["sub", "iat", "exp", "jti", "iss", "aud", "type"]}, ) except jwt.PyJWTError: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or expired session token", headers={"WWW-Authenticate": "Bearer"}, ) if claims["type"] != "session" or not isinstance(claims["sub"], str): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or expired session token", headers={"WWW-Authenticate": "Bearer"}, ) product_user_id = claims.get("sub") user = gr_api.get_user(product_user_id=product_user_id) return user def create_session(product_user_id: str) -> str: now = datetime.now(timezone.utc) return jwt.encode( { "sub": product_user_id, "iat": now, "exp": now + timedelta(seconds=settings.session_token_ttl_seconds), "jti": uuid4().hex, "iss": JWT_ISSUER, "aud": JWT_AUDIENCE, "type": "session", }, settings.session_jwt_secret.get_secret_value(), algorithm="HS256", )