Viên ngọc phương Đông

ruạṛ

from starlette.middleware.base import BaseHTTPMiddleware
from typing import List
from fastapi import Request, Response
from app.auth_framework.AuthStrategy import AuthStrategy

ALLOWED_PATHS = ["/docs", "/openapi.json", "/redoc", "/bart/healthcheck"]

class AuthMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, auth_strategies: List[AuthStrategy]):
        """
        auth_strategies: List of authentication strategies to check in order.
        """
        super().__init__(app)
        self.auth_strategies = auth_strategies

async def dispatch(self, request: Request, call_next):
        # path = request.url.path
        path = request.scope.get("path")
        print("path:",path)
        if path in ALLOWED_PATHS:
            return await call_next(request)

for strategy in self.auth_strategies:
            auth_response = await strategy.authenticate(request)
            if auth_response is None:  # Authenticated successfully
                return await call_next(request)
            elif auth_response.status_code == 401:  # Unauthorized
                return auth_response  # Stop checking and return unauthorized

return Response("Forbidden", status_code=403)  # No strategy matched

Register