python - 使用 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",
},
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.g,但这似乎同样不安全,因为它是应用程序上下文。谢谢!
解决方案
前导下划线看起来确实很奇怪,但这似乎是处理这个问题的正确方法。如果您想稍微隐藏复杂性,可以使用LocalProxy,如下所示:
授权文件
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
路线.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)
推荐阅读
- mongodb - $lookup 在嵌套数组中加入
- javascript - 如何在没有 jquery 的情况下使用仅使用 javascript 的 EventListener 切换子菜单?
- ios - 如何像 truecaller 一样在 iPhone 中获取通话记录
- wordpress - 访问 Ajax Enqeued 文件以传递给 URL 以进行 Ajax 调用的问题
- java - Maven 打包几个阴影罐子
- c# - 错误(对象不能从 DBNull 转换为其他类型
- python - Python - 如何从单个元组列表中动态创建单独的元组列表
- c++ - 尝试将 arduino 连接到使用 sgcWebSockets 制作的 WebSocket 服务器时出现异常
- arrays - 如何获取从服务器写入的数组中 JSON 对象的长度?
- java - 如何将此 Java 接口和继承结构转换为 Golang?