import functools from typing import List from urllib.parse import urlparse, urlunparse import jwt from environs import Env from fastapi import Request, Response from fastapi.responses import JSONResponse from starlette.middleware.base import BaseHTTPMiddleware from .plugins import get_auth_plugin_from_env class AuthMiddleware(BaseHTTPMiddleware): def __init__( self, app, env: Env, login_path: str, white_list: List[str] = [], ): super().__init__(app) self.login_path = login_path self.white_list = white_list self.auth_plugin = get_auth_plugin_from_env(env) async def dispatch(self, request: Request, call_next): if request.url.path == self.login_path: return self.auth_plugin.login(request) if request.url.path in self.white_list: return await call_next(request) user = self.auth_plugin.get_user_identity(request) if user: request.state.user = user return await call_next(request) auth_url = self.auth_plugin.build_authorize_url(request) if auth_url: return JSONResponse({"auth_url": auth_url}, status_code=401) else: return JSONResponse( { "status": "error", "message": "please provide valid x-langlink-access-key, x-langlink-access-secret and x-langlink-user headers", }, status_code=403, )