36 lines
1.2 KiB
Python
36 lines
1.2 KiB
Python
from fastapi import Depends, HTTPException, status
|
|
from fastapi.security import OAuth2AuthorizationCodeBearer
|
|
from fastapi.responses import RedirectResponse
|
|
from typing import Optional
|
|
import os
|
|
import boto3
|
|
|
|
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"
|
|
|
|
oauth2_scheme = OAuth2AuthorizationCodeBearer(
|
|
authorizationUrl=f"{DOMAIN}/oauth2/authorize",
|
|
tokenUrl=f"{DOMAIN}/oauth2/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"
|
|
f"&redirect_uri=http://localhost:8000/auth/callback"
|
|
)
|
|
|
|
try:
|
|
cognito = boto3.client('cognito-idp', region_name=REGION)
|
|
response = cognito.get_user(AccessToken=token)
|
|
return response
|
|
except Exception as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid authentication credentials",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
) |