kallithea Changeset - e04106e46d6f

Changeset - e04106e46d6f

Parent rev.

Child rev.

[Not reviewed]

default

0 2 0

Thomas De Schampheleire - 11 years ago 2015-03-30 21:27:02
thomas.de.schampheleire@gmail.com

auth: return early in LoginRequired on API key validation

Simplify the logic in the LoginRequired decorator when Kallithea is accessed
using an API key. Either:
- the key is valid and API access is allowed for the accessed method
(continue), or
- the key is invalid (redirect to login page), or
- the accessed method does not allow API access (403 Forbidden)

In none of these cases does it make sense to continue checking for user
authentication, so return early.

2 files changed with 17 insertions and 19 deletions:

kallithea/lib/auth.py

kallithea/tests/functional/test_login.py

0 comments (0 inline, 0 general)

kallithea/lib/auth.py

➞

Show inline comments

@@ @@ -707,133 +707,131 @@ def set_available_permissions(config): @@
     :param config: current pylons config instance
     """
     log.info('getting information about all available permissions')
     try:
         sa = meta.Session
         all_perms = sa.query(Permission).all()
         config['available_permissions'] = [x.permission_name for x in all_perms]
     finally:
         meta.Session.remove()
 #==============================================================================
 # CHECK DECORATORS
 #==============================================================================
 def redirect_to_login(message=None):
     from kallithea.lib import helpers as h
     p = url.current()
     h.flash(h.literal(message), category='warning')
     log.debug('Redirecting to login page, origin: %s' % p)
     return redirect(url('login_home', came_from=p))
 class LoginRequired(object):
     """
     Must be logged in to execute this function else
     redirect to login page
     :param api_access: if enabled this checks only for valid auth token
         and grants access based on valid token
     """
     def __init__(self, api_access=False):
         self.api_access = api_access
     def __call__(self, func):
         return decorator(self.__wrapper, func)
     def __wrapper(self, func, *fargs, **fkwargs):
         cls = fargs[0]
         user = cls.authuser
         loc = "%s:%s" % (cls.__class__.__name__, func.__name__)
         log.debug('Checking access for user %s @ %s' % (user, loc))
         # check if our IP is allowed
         if not user.ip_allowed:
             return redirect_to_login(_('IP %s not allowed' % (user.ip_addr)))
         # check if we used an APIKEY and it's a valid one
         # defined whitelist of controllers which API access will be enabled
         _api_key = request.GET.get('api_key', '')
         api_access_valid = allowed_api_access(loc, api_key=_api_key)
         # check if we used an API key and it's a valid one
         api_key = request.GET.get('api_key')
         if api_key is not None:
         # explicit controller is enabled or API is in our whitelist
         if self.api_access or api_access_valid:
             log.debug('Checking API KEY access for %s' % cls)
             if _api_key and _api_key in user.api_keys:
                 api_access_valid = True
                 log.debug('API KEY ****%s is VALID' % _api_key[-4:])
             if self.api_access or allowed_api_access(loc, api_key=api_key):
                 if api_key in user.api_keys:
                     log.info('user %s authenticated with API key ****%s @ %s'
                              % (user, api_key[-4:], loc))
                     return func(*fargs, **fkwargs)
             else:
                 api_access_valid = False
                 if not _api_key:
                     log.debug("API KEY *NOT* present in request")
                     log.warning('API key ****%s is NOT valid' % api_key[-4:])
                     return redirect_to_login(_('Invalid API key'))
                 else:
                     log.warning("API KEY ****%s *NOT* valid" % _api_key[-4:])
                 # controller does not allow API access
                 log.warning('API access to %s is not allowed' % loc)
                 return abort(403)
         # CSRF protection - POSTs with session auth must contain correct token
-        if request.POST and user.is_authenticated and not api_access_valid:
         if request.POST and user.is_authenticated:
             token = request.POST.get(secure_form.token_key)
             if not token or token != secure_form.authentication_token():
                 log.error('CSRF check failed')
                 return abort(403)
         log.debug('Checking if %s is authenticated @ %s' % (user.username, loc))
         reason = 'RegularAuth' if user.is_authenticated else 'APIAuth'
-        if user.is_authenticated or api_access_valid:
         if user.is_authenticated:
             log.info('user %s authenticating with:%s IS authenticated on func %s '
                      % (user, reason, loc)
+            )
             return func(*fargs, **fkwargs)
         else:
             log.warning('user %s authenticating with:%s NOT authenticated on func: %s: '
                      'API_ACCESS:%s'
                      % (user, reason, loc, api_access_valid)
                      % (user, reason, loc)
+            )
             return redirect_to_login()
 class NotAnonymous(object):
     """
     Must be logged in to execute this function else
     redirect to login page"""
     def __call__(self, func):
         return decorator(self.__wrapper, func)
     def __wrapper(self, func, *fargs, **fkwargs):
         cls = fargs[0]
         self.user = cls.authuser
         log.debug('Checking if user is not anonymous @%s' % cls)
         anonymous = self.user.username == User.DEFAULT_USER
         if anonymous:
             return redirect_to_login(_('You need to be a registered user to '
                     'perform this action'))
         else:
             return func(*fargs, **fkwargs)
 class PermsDecorator(object):
     """Base class for controller decorators"""
     def __init__(self, *required_perms):
         self.required_perms = set(required_perms)
         self.user_perms = None
     def __call__(self, func):
         return decorator(self.__wrapper, func)
     def __wrapper(self, func, *fargs, **fkwargs):
         cls = fargs[0]
         self.user = cls.authuser
         self.user_perms = self.user.permissions
         log.debug('checking %s permissions %s for %s %s',
            self.__class__.__name__, self.required_perms, cls, self.user)
         if self.check_permissions():
             log.debug('Permission granted for %s %s' % (cls, self.user))
             return func(*fargs, **fkwargs)
         else:

kallithea/tests/functional/test_login.py

➞

Show inline comments

@@ @@ -274,97 +274,97 @@ class TestLoginController(TestController @@
         # BAD KEY
         key = "bad"
         response = self.app.get(url(controller='login',
                                     action='password_reset_confirmation',
                                     key=key))
         self.assertEqual(response.status, '302 Found')
         self.assertTrue(response.location.endswith(url('reset_password')))
         # GOOD KEY
         key = User.get_by_username(username).api_key
         response = self.app.get(url(controller='login',
                                     action='password_reset_confirmation',
                                     key=key))
         self.assertEqual(response.status, '302 Found')
         self.assertTrue(response.location.endswith(url('login_home')))
         self.checkSessionFlash(response,
                                ('Your password reset was successful, '
                                 'new password has been sent to your email'))
         response = response.follow()
     def _get_api_whitelist(self, values=None):
         config = {'api_access_controllers_whitelist': values or []}
         return config
     @parameterized.expand([
         ('none', None),
         ('empty_string', ''),
         ('fake_number', '123456'),
         ('proper_api_key', None)
     ])
     def test_access_not_whitelisted_page_via_api_key(self, test_name, api_key):
         whitelist = self._get_api_whitelist([])
         with mock.patch('kallithea.CONFIG', whitelist):
             self.assertEqual([],
                              whitelist['api_access_controllers_whitelist'])
             if test_name == 'proper_api_key':
                 #use builtin if api_key is None
                 api_key = User.get_first_admin().api_key
             with fixture.anon_access(False):
                 self.app.get(url(controller='changeset',
                                  action='changeset_raw',
                                  repo_name=HG_REPO, revision='tip', api_key=api_key),
-                             status=302)
+                             status=403)
     @parameterized.expand([
         ('none', None, 302),
         ('empty_string', '', 302),
         ('fake_number', '123456', 302),
         ('proper_api_key', None, 200)
     ])
     def test_access_whitelisted_page_via_api_key(self, test_name, api_key, code):
         whitelist = self._get_api_whitelist(['ChangesetController:changeset_raw'])
         with mock.patch('kallithea.CONFIG', whitelist):
             self.assertEqual(['ChangesetController:changeset_raw'],
                              whitelist['api_access_controllers_whitelist'])
             if test_name == 'proper_api_key':
                 api_key = User.get_first_admin().api_key
             with fixture.anon_access(False):
                 self.app.get(url(controller='changeset',
                                  action='changeset_raw',
                                  repo_name=HG_REPO, revision='tip', api_key=api_key),
                              status=code)
     def test_access_page_via_extra_api_key(self):
         whitelist = self._get_api_whitelist(['ChangesetController:changeset_raw'])
         with mock.patch('kallithea.CONFIG', whitelist):
             self.assertEqual(['ChangesetController:changeset_raw'],
                              whitelist['api_access_controllers_whitelist'])
             new_api_key = ApiKeyModel().create(TEST_USER_ADMIN_LOGIN, u'test')
             Session().commit()
             with fixture.anon_access(False):
                 self.app.get(url(controller='changeset',
                                  action='changeset_raw',
                                  repo_name=HG_REPO, revision='tip', api_key=new_api_key.api_key),
                              status=200)
     def test_access_page_via_expired_api_key(self):
         whitelist = self._get_api_whitelist(['ChangesetController:changeset_raw'])
         with mock.patch('kallithea.CONFIG', whitelist):
             self.assertEqual(['ChangesetController:changeset_raw'],
                              whitelist['api_access_controllers_whitelist'])
             new_api_key = ApiKeyModel().create(TEST_USER_ADMIN_LOGIN, u'test')
             Session().commit()
             #patch the api key and make it expired
             new_api_key.expires = 0
             Session().add(new_api_key)
             Session().commit()
             with fixture.anon_access(False):

0 comments (0 inline, 0 general)