rasa/core/server.py
def requires_auth(app: Sanic,
                  token: Optional[Text] = None
                  ) -> Callable[[Any], Any]:
    """Wraps a request handler with token authentication."""

    def decorator(f: Callable[[Any, Any, Any], Any]
                  ) -> Callable[[Any, Any], Any]:
        def sender_id_from_args(args: Any,
                                kwargs: Any) -> Optional[Text]:
            argnames = utils.arguments_of(f)
            try:
                sender_id_arg_idx = argnames.index("sender_id")
                if "sender_id" in kwargs:  # try to fetch from kwargs first
                    return kwargs["sender_id"]
                if sender_id_arg_idx < len(args):
                    return args[sender_id_arg_idx]
                return None
            except ValueError:
                return None

        def sufficient_scope(request,
                             *args: Any,
                             **kwargs: Any) -> Optional[bool]:
            jwt_data = request.app.auth.extract_payload(request)
            user = jwt_data.get("user", {})

            username = user.get("username", None)
            role = user.get("role", None)

            if role == "admin":
                return True
            elif role == "user":
                sender_id = sender_id_from_args(args, kwargs)
                return sender_id is not None and username == sender_id
            else:
                return False

        @wraps(f)
        async def decorated(request: Request,
                            *args: Any,
                            **kwargs: Any) -> Any:

            provided = utils.default_arg(request, 'token', None)
            # noinspection PyProtectedMember
            if token is not None and provided == token:
                result = f(request, *args, **kwargs)
                if isawaitable(result):
                    result = await result
                return result
            elif (app.config.get('USE_JWT') and
                  request.app.auth.is_authenticated(request)):
                if sufficient_scope(request, *args, **kwargs):
                    result = f(request, *args, **kwargs)
                    if isawaitable(result):
                        result = await result
                    return result
                raise ErrorResponse(
                    403, "NotAuthorized",
                    "User has insufficient permissions.",
                    help_url=_docs(
                        "/server.html#security-considerations"))
            elif token is None and app.config.get('USE_JWT') is None:
                # authentication is disabled
                result = f(request, *args, **kwargs)
                if isawaitable(result):
                    result = await result
                return result
            raise ErrorResponse(
                401, "NotAuthenticated", "User is not authenticated.",
                help_url=_docs("/server.html#security-considerations"))

        return decorated

    return decorator
Wraps a request handler with token authentication.
Wraps a request handler with token authentication.
Wraps a request handler with token authentication .
def requires_auth(app: Sanic,
                  token: Optional[Text] = None
                  ) -> Callable[[Any], Any]:
    def decorator(f: Callable[[Any, Any, Any], Any]
                  ) -> Callable[[Any, Any], Any]:
        def sender_id_from_args(args: Any,
                                kwargs: Any) -> Optional[Text]:
            argnames = utils.arguments_of(f)
            try:
                sender_id_arg_idx = argnames.index("sender_id")
                if "sender_id" in kwargs:  
                    return kwargs["sender_id"]
                if sender_id_arg_idx < len(args):
                    return args[sender_id_arg_idx]
                return None
            except ValueError:
                return None
        def sufficient_scope(request,
                             *args: Any,
                             **kwargs: Any) -> Optional[bool]:
            jwt_data = request.app.auth.extract_payload(request)
            user = jwt_data.get("user", {})
            username = user.get("username", None)
            role = user.get("role", None)
            if role == "admin":
                return True
            elif role == "user":
                sender_id = sender_id_from_args(args, kwargs)
                return sender_id is not None and username == sender_id
            else:
                return False
        @wraps(f)
        async def decorated(request: Request,
                            *args: Any,
                            **kwargs: Any) -> Any:
            provided = utils.default_arg(request, 'token', None)
            if token is not None and provided == token:
                result = f(request, *args, **kwargs)
                if isawaitable(result):
                    result = await result
                return result
            elif (app.config.get('USE_JWT') and
                  request.app.auth.is_authenticated(request)):
                if sufficient_scope(request, *args, **kwargs):
                    result = f(request, *args, **kwargs)
                    if isawaitable(result):
                        result = await result
                    return result
                raise ErrorResponse(
                    403, "NotAuthorized",
                    "User has insufficient permissions.",
                    help_url=_docs(
                        "/server.html#security-considerations"))
            elif token is None and app.config.get('USE_JWT') is None:
                result = f(request, *args, **kwargs)
                if isawaitable(result):
                    result = await result
                return result
            raise ErrorResponse(
                401, "NotAuthenticated", "User is not authenticated.",
                help_url=_docs("/server.html#security-considerations"))
        return decorated
    return decorator
def requires_auth ( app : Sanic , token : Optional [ Text ] = None ) -> Callable [ [ Any ] , Any ] : def decorator ( f : Callable [ [ Any , Any , Any ] , Any ] ) -> Callable [ [ Any , Any ] , Any ] : def sender_id_from_args ( args : Any , kwargs : Any ) -> Optional [ Text ] : argnames = utils . arguments_of ( f ) try : sender_id_arg_idx = argnames . index ( "sender_id" ) if "sender_id" in kwargs : return kwargs [ "sender_id" ] if sender_id_arg_idx < len ( args ) : return args [ sender_id_arg_idx ] return None except ValueError : return None def sufficient_scope ( request , * args : Any , ** kwargs : Any ) -> Optional [ bool ] : jwt_data = request . app . auth . extract_payload ( request ) user = jwt_data . get ( "user" , { } ) username = user . get ( "username" , None ) role = user . get ( "role" , None ) if role == "admin" : return True elif role == "user" : sender_id = sender_id_from_args ( args , kwargs ) return sender_id is not None and username == sender_id else : return False @ wraps ( f ) async def decorated ( request : Request , * args : Any , ** kwargs : Any ) -> Any : provided = utils . default_arg ( request , 'token' , None ) if token is not None and provided == token : result = f ( request , * args , ** kwargs ) if isawaitable ( result ) : result = await result return result elif ( app . config . get ( 'USE_JWT' ) and request . app . auth . is_authenticated ( request ) ) : if sufficient_scope ( request , * args , ** kwargs ) : result = f ( request , * args , ** kwargs ) if isawaitable ( result ) : result = await result return result raise ErrorResponse ( 403 , "NotAuthorized" , "User has insufficient permissions." , help_url = _docs ( "/server.html#security-considerations" ) ) elif token is None and app . config . get ( 'USE_JWT' ) is None : result = f ( request , * args , ** kwargs ) if isawaitable ( result ) : result = await result return result raise ErrorResponse ( 401 , "NotAuthenticated" , "User is not authenticated." , help_url = _docs ( "/server.html#security-considerations" ) ) return decorated return decorator
Wraps a request handler with token authentication.
Wraps a request handler with token authentication .
"""Wraps a request handler with token authentication.""" # try to fetch from kwargs first # noinspection PyProtectedMember # authentication is disabled
app
Sanic
token
Optional[Text]