Flask-restful API授权。在装饰器中访问 current_identity
Flask-restful API Authorization. Access current_identity inside decorator
我使用 flask-restful 创建我的 API。我已经使用 flask-jwt
启用基于 JWT
的身份验证。现在我需要做授权。
我试过放置我的授权装饰器。
test.py (/test api)
from flask_restful import Resource
from flask_jwt import jwt_required
from authorization_helper import authorized_api_user_type
class Test(Resource):
decorators = [jwt_required(), authorized_api_user_type()]
def get(self):
return 'GET OK'
def post(self):
return 'POST OK'
基本上为了处理基本授权,我需要访问 current_identity
并检查它的类型。然后根据它的类型,我将决定用户是否有权访问 api / resources.
但是 current_identity
在该装饰器中似乎是 empty
。因此,为了间接获取它,我必须查看 jwt_handler
的代码并执行那里完成的操作。
authorization_helper.py
from functools import wraps
from flask_jwt import _jwt, JWTError
import jwt
from models import Teacher, Student
def authorized_api_user_type(realm=None, user_type='teacher'):
def wrapper(fn):
@wraps(fn)
def decorator(*args, **kwargs):
token = _jwt.request_callback()
if token is None:
raise JWTError('Authorization Required', 'Request does not contain an access token',
headers={'WWW-Authenticate': 'JWT realm="%s"' % realm})
try:
payload = _jwt.jwt_decode_callback(token)
except jwt.InvalidTokenError as e:
raise JWTError('Invalid token', str(e))
identity = _jwt.identity_callback(payload)
if user_type == 'student' and isinstance(identity, Student):
return fn(*args, **kwargs)
elif user_type == 'teacher' and isinstance(identity, Teacher):
return fn(*args, **kwargs)
# NOTE - By default JWTError throws 401. We needed 404. Hence status_code=404
raise JWTError('Unauthorized',
'You are unauthorized to request the api or access the resource',
status_code=404)
return decorator
return wrapper
为什么我不能在我的 authorized_api_user_type
装饰器中访问 current_identity
?在 flask-restful 中进行授权的正确方法是什么?
我当前的解决方案如下:
@app.before_request
def detect_something():
header = request.headers.get('Authorization')
if header:
_, token = header.split()
request.identity = identity(jwt.decode(token,
app.config['SECRET_KEY']))
之后我们可以通过request.identity
在装饰器中访问身份。我从代码中删除了所有地方的 current_identity
。这仍然是混乱的方式。
这里是 Flask-JWT
和 Flask-Restful
的快速入门组合。
from flask import Flask
from flask_restful import Resource, Api, abort
from functools import wraps
app = Flask(__name__)
api = Api(app)
from flask_jwt import JWT, jwt_required, current_identity
from werkzeug.security import safe_str_cmp
class User(object):
def __init__(self, id, username, password):
self.id = id
self.username = username
self.password = password
def __str__(self):
return "User(id='%s')" % self.id
users = [
User(1, 'user1', 'abcxyz'),
User(2, 'user2', 'abcxyz'),
]
username_table = {u.username: u for u in users}
userid_table = {u.id: u for u in users}
def authenticate(username, password):
user = username_table.get(username, None)
if user and safe_str_cmp(user.password.encode('utf-8'), password.encode('utf-8')):
return user
def identity(payload):
user_id = payload['identity']
return userid_table.get(user_id, None)
app.config['SECRET_KEY'] = 'super-secret'
jwt = JWT(app, authenticate, identity)
def checkuser(func):
@wraps(func)
def wrapper(*args, **kwargs):
if current_identity.username == 'user1':
return func(*args, **kwargs)
return abort(401)
return wrapper
class HelloWorld(Resource):
decorators = [checkuser, jwt_required()]
def get(self):
return {'hello': current_identity.username}
api.add_resource(HelloWorld, '/')
if __name__ == '__main__':
app.run(debug=True)
POST
{
"username": "user1",
"password": "abcxyz"
}
到localhost:5000/auth
并得到access_token
的回应。
然后 GET localhost:5000/
和 header
Authorization: JWT `the access_token value above`
你会得到
{
"hello": "user1"
}
如果您尝试使用 user2 的 JWT 令牌访问 localhost:5000/
,您将获得 401
.
装饰器是这样包装的:
for decorator in self.decorators:
resource_func = decorator(resource_func)
https://github.com/flask-restful/flask-restful/blob/master/flask_restful/init.py#L445
所以装饰器数组中后面的那个会更早到达 运行。
更多参考:
https://github.com/rchampa/timetable/blob/master/restful/users.py
使用这个:
from flask_jwt import current_identity
@jwt_required()
def get(self):
return {'current_identity': current_identity.json()}
我使用 flask-restful 创建我的 API。我已经使用 flask-jwt
启用基于 JWT
的身份验证。现在我需要做授权。
我试过放置我的授权装饰器。
test.py (/test api)
from flask_restful import Resource
from flask_jwt import jwt_required
from authorization_helper import authorized_api_user_type
class Test(Resource):
decorators = [jwt_required(), authorized_api_user_type()]
def get(self):
return 'GET OK'
def post(self):
return 'POST OK'
基本上为了处理基本授权,我需要访问 current_identity
并检查它的类型。然后根据它的类型,我将决定用户是否有权访问 api / resources.
但是 current_identity
在该装饰器中似乎是 empty
。因此,为了间接获取它,我必须查看 jwt_handler
的代码并执行那里完成的操作。
authorization_helper.py
from functools import wraps
from flask_jwt import _jwt, JWTError
import jwt
from models import Teacher, Student
def authorized_api_user_type(realm=None, user_type='teacher'):
def wrapper(fn):
@wraps(fn)
def decorator(*args, **kwargs):
token = _jwt.request_callback()
if token is None:
raise JWTError('Authorization Required', 'Request does not contain an access token',
headers={'WWW-Authenticate': 'JWT realm="%s"' % realm})
try:
payload = _jwt.jwt_decode_callback(token)
except jwt.InvalidTokenError as e:
raise JWTError('Invalid token', str(e))
identity = _jwt.identity_callback(payload)
if user_type == 'student' and isinstance(identity, Student):
return fn(*args, **kwargs)
elif user_type == 'teacher' and isinstance(identity, Teacher):
return fn(*args, **kwargs)
# NOTE - By default JWTError throws 401. We needed 404. Hence status_code=404
raise JWTError('Unauthorized',
'You are unauthorized to request the api or access the resource',
status_code=404)
return decorator
return wrapper
为什么我不能在我的 authorized_api_user_type
装饰器中访问 current_identity
?在 flask-restful 中进行授权的正确方法是什么?
我当前的解决方案如下:
@app.before_request
def detect_something():
header = request.headers.get('Authorization')
if header:
_, token = header.split()
request.identity = identity(jwt.decode(token,
app.config['SECRET_KEY']))
之后我们可以通过request.identity
在装饰器中访问身份。我从代码中删除了所有地方的 current_identity
。这仍然是混乱的方式。
这里是 Flask-JWT
和 Flask-Restful
的快速入门组合。
from flask import Flask
from flask_restful import Resource, Api, abort
from functools import wraps
app = Flask(__name__)
api = Api(app)
from flask_jwt import JWT, jwt_required, current_identity
from werkzeug.security import safe_str_cmp
class User(object):
def __init__(self, id, username, password):
self.id = id
self.username = username
self.password = password
def __str__(self):
return "User(id='%s')" % self.id
users = [
User(1, 'user1', 'abcxyz'),
User(2, 'user2', 'abcxyz'),
]
username_table = {u.username: u for u in users}
userid_table = {u.id: u for u in users}
def authenticate(username, password):
user = username_table.get(username, None)
if user and safe_str_cmp(user.password.encode('utf-8'), password.encode('utf-8')):
return user
def identity(payload):
user_id = payload['identity']
return userid_table.get(user_id, None)
app.config['SECRET_KEY'] = 'super-secret'
jwt = JWT(app, authenticate, identity)
def checkuser(func):
@wraps(func)
def wrapper(*args, **kwargs):
if current_identity.username == 'user1':
return func(*args, **kwargs)
return abort(401)
return wrapper
class HelloWorld(Resource):
decorators = [checkuser, jwt_required()]
def get(self):
return {'hello': current_identity.username}
api.add_resource(HelloWorld, '/')
if __name__ == '__main__':
app.run(debug=True)
POST
{
"username": "user1",
"password": "abcxyz"
}
到localhost:5000/auth
并得到access_token
的回应。
然后 GET localhost:5000/
和 header
Authorization: JWT `the access_token value above`
你会得到
{
"hello": "user1"
}
如果您尝试使用 user2 的 JWT 令牌访问 localhost:5000/
,您将获得 401
.
装饰器是这样包装的:
for decorator in self.decorators:
resource_func = decorator(resource_func)
https://github.com/flask-restful/flask-restful/blob/master/flask_restful/init.py#L445
所以装饰器数组中后面的那个会更早到达 运行。
更多参考:
https://github.com/rchampa/timetable/blob/master/restful/users.py
使用这个:
from flask_jwt import current_identity
@jwt_required()
def get(self):
return {'current_identity': current_identity.json()}