I use flask-restful to create my APIs. I have used flask-jwt for enabling authentication based on JWT. Now I need to do authorization.
I have tried putting my authorization decorator.
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'
Basically to handle the basic authorization, I need to access current_identity and check it's type. Then based on it's type I am gonna decide whether the user is authorized to access the api / resources.
But current_identity appears to be empty in that decorator. So to get it indirectly, I had to see the code of jwt_handler and do the things done there.
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
Why can't I just access current_identity in my authorized_api_user_type decorator? What is the RIGHT way of doing authorization in flask-restful?
Here is the combination of quickstarts of both Flask-JWT and 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"
}
To localhost:5000/auth and get the access_token in response.
Then GET localhost:5000/ with header
Authorization: JWT `the access_token value above`
You would get
{
"hello": "user1"
}
if you try to access localhost:5000/ with the JWT token of user2, you would get 401.
The decorators are wrapped in this way:
for decorator in self.decorators:
resource_func = decorator(resource_func)
https://github.com/flask-restful/flask-restful/blob/master/flask_restful/init.py#L445
So the later one in the decorators array gets to run earlier.
For more reference:
https://github.com/rchampa/timetable/blob/master/restful/users.py