import os import boto3 import requests import jwt from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2AuthorizationCodeBearer from fastapi.responses import RedirectResponse REGION = "us-east-2" USER_POOL_ID = os.getenv("COGNITO_USER_POOL_ID") CLIENT_ID = os.getenv("COGNITO_CLIENT_ID") DOMAIN = f"https://iptv-updater.auth.{REGION}.amazoncognito.com" REDIRECT_URI = f"http://localhost:8000/auth/callback" oauth2_scheme = OAuth2AuthorizationCodeBearer( authorizationUrl=f"{DOMAIN}/oauth2/authorize", tokenUrl=f"{DOMAIN}/oauth2/token" ) def exchange_code_for_token(code: str): token_url = f"{DOMAIN}/oauth2/token" data = { 'grant_type': 'authorization_code', 'client_id': CLIENT_ID, 'code': code, 'redirect_uri': REDIRECT_URI } response = requests.post(token_url, data=data) if response.status_code == 200: return response.json() print(f"Token exchange failed: {response.text}") # Add logging raise HTTPException(status_code=400, detail="Failed to exchange code for token") async def get_current_user(token: str = Depends(oauth2_scheme)): if not token: return RedirectResponse( f"{DOMAIN}/login?client_id={CLIENT_ID}" f"&response_type=code" f"&scope=openid+email+profile" # Added more scopes f"&redirect_uri={REDIRECT_URI}" ) try: # Decode JWT token instead of using get_user decoded = jwt.decode( token, options={"verify_signature": False} # We trust tokens from Cognito ) return { "Username": decoded.get("email") or decoded.get("sub"), "UserAttributes": [ {"Name": k, "Value": v} for k, v in decoded.items() ] } except Exception as e: print(f"Token verification failed: {str(e)}") # Add logging raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid authentication credentials", headers={"WWW-Authenticate": "Bearer"}, )