将对象从服务器拦截器传递给函数

Question

我创建了一个简单的服务器拦截器，它根据 JWT 令牌检索用户。

但现在我想让它对我服务的所有方法可用。

目前我正在使用装饰器。但我想避免装饰所有的方法。以防万一，只装饰那些不需要用户的。

有人可以给我一个线索吗？

这是我的代码：

class AuthInterceptor(grpc.ServerInterceptor):
"""Authorization Interceptor"""

def __init__(self, loader):
    self._loader = loader

def intercept_service(self, continuation, handler_call_details):
    # Authenticate if we not requesting token.
    if not handler_call_details.method.endswith('GetToken'):
        # My Authentication class.
        auth = EosJWTAuth()
        # Authenticate using the headers tokens to get the user.
        user = auth.authenticate(
            dict(handler_call_details.invocation_metadata))[0]

        # Do something here to pass the authenticated user to the functions.

    cont = continuation(handler_call_details)
    return cont

我希望我的方法可以像这样访问用户。

class UserService(BaseService, users_pb2_grpc.UserServicer):
"""User service."""

def get_age(self, request, context):
    """Get user's age"""
    user = context.get_user()
    # or user = context.user 
    # or user = self.user 
    # os user = request.get_user() 
    return pb.UserInfo(name=user.name, age=user.age)

Answer 1

这是 Web 服务器的常见需求，向处理程序添加装饰器以显式设置 authentication/authorization 的要求是个好主意。它有助于提高可读性，并降低整体复杂性。

但是，这里有一个解决方法可以解决您的问题。它使用 Python 元类来自动修饰每个服务程序方法。

import grpc
import functools
import six

def auth_decorator(func):
    @functools.wraps(func)
    def wrapper(request, context):
        if not func.__name__.endswith('GetToken'):
            auth = FooAuthClass()
            try:
                user = auth.authenticate(
                    dict(context.invocation_metadata)
                )[0]
                request.user = user
            except UserNotFound:
                context.abort(
                    grpc.StatusCode.UNAUTHENTICATED,
                    'Permission denied.',
                )
        return func(request, context)

    return wrapper

class AuthMeta:

    def __new__(self, class_name, bases, namespace):
        for key, value in list(namespace.items()):
            if callable(value):
                namespace[key] = auth_decorator(value)
        return type.__new__(self, class_name, bases, namespace)

class BusinessServer(FooServicer, six.with_metaclass(AuthMeta)):

    def LogicA(self, request, context):
        # request.user accessible
        ...

    def LogicGetToken(self, request, context):
        # request.user not accessible
        ...

将对象从服务器拦截器传递给函数

Pass objects from server interceptors to functions

python

grpc

grpc-python