Merge pull request #1326 from michele-comitini/jwt-decorator-fix
make allows_jwt a real decorator. Tests included!
This commit is contained in:
@@ -227,8 +227,43 @@ class TestMail(unittest.TestCase):
|
||||
# self.assertEqual(form.xml(), rtn)
|
||||
|
||||
# TODO: class TestAuthJWT(unittest.TestCase):
|
||||
class TestAuthJWT(unittest.TestCase):
|
||||
def setUp(self):
|
||||
from gluon.tools import AuthJWT
|
||||
|
||||
from gluon import current
|
||||
|
||||
self.request = Request(env={})
|
||||
self.request.application = 'a'
|
||||
self.request.controller = 'c'
|
||||
self.request.function = 'f'
|
||||
self.request.folder = 'applications/admin'
|
||||
self.current = current
|
||||
self.current.request = self.request
|
||||
|
||||
self.db = DAL(DEFAULT_URI, check_reserved=['all'])
|
||||
self.auth = Auth(self.db)
|
||||
self.auth.define_tables(username=True, signature=False)
|
||||
self.user_data = dict(username='jwtuser', password='jwtuser123')
|
||||
self.db.auth_user.insert(username=self.user_data['username'],
|
||||
password=str(
|
||||
self.db.auth_user.password.requires[0](
|
||||
self.user_data['password'])[0]))
|
||||
self.jwtauth = AuthJWT(self.auth, secret_key='secret', verify_expiration=True)
|
||||
|
||||
|
||||
def test_jwt_token_manager(self):
|
||||
self.request.vars.update(self.user_data)
|
||||
self.token = self.jwtauth.jwt_token_manager()
|
||||
|
||||
|
||||
def test_allows_jwt(self):
|
||||
request = self.request
|
||||
@self.jwtauth.allows_jwt()
|
||||
def optional_auth():
|
||||
assertEqual(self.user_data['username'], self.auth.user.username)
|
||||
|
||||
|
||||
@unittest.skipIf(IS_IMAP, "TODO: Imap raises 'Connection refused'")
|
||||
# class TestAuth(unittest.TestCase):
|
||||
#
|
||||
|
||||
+38
-8
@@ -1192,6 +1192,15 @@ class AuthJWT(object):
|
||||
def protected():
|
||||
return '%s$%s' % (request.now, auth.user_id)
|
||||
|
||||
To inject optional auth info into the action with JWT
|
||||
@myjwt.allows_jwt()
|
||||
def unprotected():
|
||||
if auth.user:
|
||||
return '%s$%s' % (request.now, auth.user_id)
|
||||
|
||||
return "No auth info!"
|
||||
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self,
|
||||
@@ -1411,12 +1420,13 @@ class AuthJWT(object):
|
||||
self.auth.user_groups = tokend['user_groups']
|
||||
self.auth.hmac_key = tokend['hmac_key']
|
||||
|
||||
def allows_jwt(self, otherwise=None):
|
||||
def get_jwt_token_from_request(self):
|
||||
"""
|
||||
The validator that checks for the header or the
|
||||
_token var
|
||||
The method that extracts and validates the token, either
|
||||
from the header or the _token var
|
||||
|
||||
"""
|
||||
request = current.request
|
||||
token = None
|
||||
token_in_header = request.env.http_authorization
|
||||
if token_in_header:
|
||||
parts = token_in_header.split()
|
||||
@@ -1429,11 +1439,31 @@ class AuthJWT(object):
|
||||
token = parts[1]
|
||||
else:
|
||||
token = request.vars._token
|
||||
if token and len(token) < self.max_header_length:
|
||||
tokend = self.load_token(token)
|
||||
self.inject_token(tokend)
|
||||
return self.auth.requires(True, otherwise=otherwise)
|
||||
|
||||
return token
|
||||
|
||||
def allows_jwt(self, otherwise=None):
|
||||
"""
|
||||
The decorator that takes care of injecting auth info in the decorated action.
|
||||
Works w/o resorting to session.
|
||||
"""
|
||||
def decorator(action):
|
||||
def f(*args, **kwargs):
|
||||
token = self.get_jwt_token_from_request()
|
||||
if token and len(token) < self.max_header_length:
|
||||
try:
|
||||
tokend = self.load_token(token)
|
||||
except ValueError:
|
||||
raise HTTP(400, 'Invalid JWT header, wrong token format')
|
||||
self.inject_token(tokend)
|
||||
return action(*args, **kwargs)
|
||||
|
||||
f.__doc__ = action.__doc__
|
||||
f.__name__ = action.__name__
|
||||
f.__dict__.update(action.__dict__)
|
||||
return f
|
||||
|
||||
return decorator
|
||||
|
||||
class Auth(object):
|
||||
|
||||
|
||||
Reference in New Issue
Block a user