Method dispatch

src/core/http/middleware/auth_bearer.py:37–71

(self, request: Request, call_next: RequestResponseEndpoint)

        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
    def _is_excluded_path(path: str) -> bool: