make allows_jwt a real decorator. Tests included!
This commit is contained in:
@@ -227,8 +227,44 @@ class TestMail(unittest.TestCase):
|
|||||||
# self.assertEqual(form.xml(), rtn)
|
# self.assertEqual(form.xml(), rtn)
|
||||||
|
|
||||||
# TODO: class TestAuthJWT(unittest.TestCase):
|
# 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)
|
||||||
|
print self.current.request.vars
|
||||||
|
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'")
|
@unittest.skipIf(IS_IMAP, "TODO: Imap raises 'Connection refused'")
|
||||||
# class TestAuth(unittest.TestCase):
|
# class TestAuth(unittest.TestCase):
|
||||||
#
|
#
|
||||||
|
|||||||
+39
-8
@@ -1192,6 +1192,15 @@ class AuthJWT(object):
|
|||||||
def protected():
|
def protected():
|
||||||
return '%s$%s' % (request.now, auth.user_id)
|
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,
|
def __init__(self,
|
||||||
@@ -1411,13 +1420,15 @@ class AuthJWT(object):
|
|||||||
self.auth.user_groups = tokend['user_groups']
|
self.auth.user_groups = tokend['user_groups']
|
||||||
self.auth.hmac_key = tokend['hmac_key']
|
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
|
The method that extracts and validates the token, either
|
||||||
_token var
|
from the header or the _token var
|
||||||
|
|
||||||
"""
|
"""
|
||||||
request = current.request
|
token = None
|
||||||
token_in_header = request.env.http_authorization
|
token_in_header = request.env.http_authorization
|
||||||
|
logger.debug('%s' % token_in_header)
|
||||||
if token_in_header:
|
if token_in_header:
|
||||||
parts = token_in_header.split()
|
parts = token_in_header.split()
|
||||||
if parts[0].lower() != self.header_prefix.lower():
|
if parts[0].lower() != self.header_prefix.lower():
|
||||||
@@ -1429,11 +1440,31 @@ class AuthJWT(object):
|
|||||||
token = parts[1]
|
token = parts[1]
|
||||||
else:
|
else:
|
||||||
token = request.vars._token
|
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):
|
class Auth(object):
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user