Class AuthBearer

src/core/http/middleware/auth_bearer.py:16–78

class AuthBearer(BaseHTTPMiddleware):
    EXCLUDED_PATHS = [
        "/docs",
        "/redoc",
        "/openapi.json",
        "/auth/login",
        "/auth/refresh",
        "/auth/signup",
        "/auth/re-send-confirm-email",
        "/auth/confirm-email",
        "/health",
        "/ws",
    ]

    def __init__(self, app: ASGIApp, user_service: UserService, hash_service: HashService) -> None:
        super().__init__(app=app)
        self.hash_service = hash_service
        self.user_service = user_service
        self.bearer_scheme = HTTPBearer(auto_error=False)
        self.logger = logging.getLogger(__name__)

    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
        request.state.is_authenticated = False
        if request.method == "OPTIONS" or AuthBearer._is_excluded_path(request.url.path):
            return await call_next(request)

        authorization = request.headers.get("Authorization")
        if not authorization:
            raise UnauthorizedException(error_no=ErrorNo.AUTHORIZATION_HEADER_NOT_FOUND, message="Unauthorized!")

        try:
            scheme, token = authorization.split(" ", 1)
            if scheme.lower() != "bearer":
                raise UnauthorizedException(error_no=ErrorNo.AUTHORIZATION_BEARER_FORMAT_ERROR, message="Unauthorized!")
        except ValueError as e:
            self.logger.error(str(e), exc_info=e)
            raise UnauthorizedException(error_no=ErrorNo.AUTHORIZATION_BEARER_INVALID, message="Unauthorized!")

        payload = self.hash_service.verify_token(token)
        if not payload:
            raise UnauthorizedException(
                error_no=ErrorNo.AUTHORIZATION_BEARER_TOKEN_INVALID_OR_EXPIRED, message="Unauthorized!"
            )

        user = await self.user_service.get_by_id(int(payload.subject))
        if not user:
            raise UnauthorizedException(error_no=ErrorNo.AUTHORIZATION_USER_NOT_FOUND, message="Unauthorized!")
        if user.status != UserStatus.ACTIVE:
            raise UnauthorizedException(error_no=ErrorNo.AUTHORIZATION_USER_NOT_ACTIVE, message="Unauthorized!")
        if user.session != payload.session:
            raise UnauthorizedException(error_no=ErrorNo.AUTHORIZATION_USER_SESSION_INVALID, message="Unauthorized!")

        request.state.user = user
        request.state.is_authenticated = True

        return await call_next(request)

    @staticmethod