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

如何解决使用 Flask 请求上下文进行用户管理

我目前正在开发一个项目，我正在使用 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",)
    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",)

    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",)
        if unverified_header["alg"] == "HS256":
            raise AuthError(
                {
                    "code": "invalid_header",)
        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",)
            except Exception:
                raise AuthError(
                    {
                        "code": "invalid_header","description": "Unable to parse authentication" " token.",)

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

    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，但这似乎同样不安全，因为它是应用程序上下文。谢谢！