使用 Flask 请求上下文进行用户管理

Question

我目前正在开发一个项目，我正在使用 Flask、Flask-RESTful 和 Auth0 构建 API 以进行用户身份验证。目前，我正在努力将示例装饰器中的 Auth0 子 ID 传递给 Flask-RESTful 方法。这是下面的代码。

Auth0 Docs提供的Auth装饰器：

def get_token_auth_header():
    """Obtains the access token from the Authorization Header"""
    auth = request.headers.get("Authorization", None)
    if not auth:
        raise AuthError(
            {
                "code": "authorization_header_missing",
                "description": "Authorization header is expected",
            },
            401,
        )

    parts = auth.split()

    if parts[0].lower() != "bearer":
        raise AuthError(
            {
                "code": "invalid_header",
                "description": "Authorization header must start with" " Bearer",
            },
            401,
        )
    elif len(parts) == 1:
        raise AuthError(
            {"code": "invalid_header", "description": "Token not found"}, 401
        )
    elif len(parts) > 2:
        raise AuthError(
            {
                "code": "invalid_header",
                "description": "Authorization header must be" " Bearer token",
            },
            401,
        )

    token = parts[1]
    return token

def requires_auth(f):
    """Determines if the access token is valid"""

    @wraps(f)
    def decorated(*args, **kwargs):
        token = get_token_auth_header()
        jsonurl = urlopen("https://" + AUTH0_DOMAIN + "/.well-known/jwks.json")
        jwks = json.loads(jsonurl.read())
        try:
            unverified_header = jwt.get_unverified_header(token)
        except jwt.JWTError:
            raise AuthError(
                {
                    "code": "invalid_header",
                    "description": "Invalid header. "
                    "Use an RS256 signed JWT Access Token",
                },
                401,
            )
        if unverified_header["alg"] == "HS256":
            raise AuthError(
                {
                    "code": "invalid_header",
                    "description": "Invalid header. "
                    "Use an RS256 signed JWT Access Token",
                },
                401,
            )
        rsa_key = {}
        for key in jwks["keys"]:
            if key["kid"] == unverified_header["kid"]:
                rsa_key = {
                    "kty": key["kty"],
                    "kid": key["kid"],
                    "use": key["use"],
                    "n": key["n"],
                    "e": key["e"],
                }
        if rsa_key:
            try:
                payload = jwt.decode(
                    token,
                    rsa_key,
                    algorithms=ALGORITHMS,
                    audience=API_IDENTIFIER,
                    issuer="https://" + AUTH0_DOMAIN + "/",
                )
            except jwt.ExpiredSignatureError:
                raise AuthError(
                    {"code": "token_expired", "description": "token is expired"}, 401
                )
            except jwt.JWTClaimsError as jce:
                raise AuthError(
                    {
                        "code": "invalid_claims",
                        "description": "incorrect claims,"
                        " please check the audience and issuer",
                    },
                    401,
                )
            except Exception:
                raise AuthError(
                    {
                        "code": "invalid_header",
                        "description": "Unable to parse authentication" " token.",
                    },
                    401,
                )

            _request_ctx_stack.top.current_user = payload
            print(payload)
            return f(*args, **kwargs)
        raise AuthError(
            {"code": "invalid_header", "description": "Unable to find appropriate key"},
            401,
        )

    return decorated

使用示例：

class HealthCheck(Resource):

    @requires_auth
    @cross_origin(headers=["Content-Type", "Authorization", "Access-Control-Allow-Origin", "http://localhost:3000"])
    def post(self):
        print(f'Req: {_request_ctx_stack.top.current_user}')
        return jsonify(message=_request_ctx_stack.top.current_user)

具体代码：

_request_ctx_stack.top.current_user = payload

目前，requires_auth 包装器会将 Auth0 响应负载存储在 _request_ctx_stack.top.current_user 中。是否有处理此问题的最佳实践？我的资源中的上述实现是获取有效负载的最佳方式吗？有些东西告诉我有更好的方法。我尝试将其放入 kwargs 以在标题中访问，但 flask-restful 不喜欢这样。我也玩过flask.g，但这似乎同样不安全，因为它是应用程序上下文。谢谢！

Answer 1

前导下划线确实看起来很奇怪，但这似乎是处理这个问题的正确方法。如果您想稍微隐藏一下复杂性，可以使用LocalProxy，如下所示：

auth.py

from flask import _request_ctx_stack
from werkzeug.local import LocalProxy

current_user = LocalProxy(lambda: _request_ctx_stack.top.current_user)

# ...
def requires_auth(f):
# ...
                _request_ctx_stack.top.current_user = payload

routes.py

from .auth import current_user

class HealthCheck(Resource):

    @requires_auth
    @cross_origin(headers=["Content-Type", "Authorization", "Access-Control-Allow-Origin", "http://localhost:3000"])
    def post(self):
        print(f'Req: {current_user}')
        return jsonify(message=current_user)