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

@@ @@ -371,805 +371,803 @@ def _cached_perms_data(user_id, user_is_ @@
             p = 'repository.admin'
         else:
             p = perm.Permission.permission_name
             if not explicit:
                 p = _choose_perm(p, cur_perm)
         permissions[RK][r_k] = p
     #======================================================================
     # !! PERMISSIONS FOR REPOSITORY GROUPS !!
     #======================================================================
     #======================================================================
     # check if user is part of user groups for this repository groups and
     # fill in his permission from it. _choose_perm decides of which
     # permission should be selected based on selected method
     #======================================================================
     # user group for repo groups permissions
     user_repo_group_perms_from_users_groups = \
      Session().query(UserGroupRepoGroupToPerm, Permission, RepoGroup)\
      .join((RepoGroup, UserGroupRepoGroupToPerm.group_id == RepoGroup.group_id))\
      .join((Permission, UserGroupRepoGroupToPerm.permission_id
             == Permission.permission_id))\
      .join((UserGroupMember, UserGroupRepoGroupToPerm.users_group_id
             == UserGroupMember.users_group_id))\
      .filter(UserGroupMember.user_id == uid)\
      .all()
     multiple_counter = collections.defaultdict(int)
     for perm in user_repo_group_perms_from_users_groups:
         g_k = perm.UserGroupRepoGroupToPerm.group.group_name
         multiple_counter[g_k] += 1
         p = perm.Permission.permission_name
         cur_perm = permissions[GK][g_k]
         if multiple_counter[g_k] > 1:
             p = _choose_perm(p, cur_perm)
         permissions[GK][g_k] = p
     # user explicit permissions for repository groups
     user_repo_groups_perms = Permission.get_default_group_perms(uid)
     for perm in user_repo_groups_perms:
         rg_k = perm.UserRepoGroupToPerm.group.group_name
         p = perm.Permission.permission_name
         cur_perm = permissions[GK][rg_k]
         if not explicit:
             p = _choose_perm(p, cur_perm)
         permissions[GK][rg_k] = p
     #======================================================================
     # !! PERMISSIONS FOR USER GROUPS !!
     #======================================================================
     # user group for user group permissions
     user_group_user_groups_perms = \
      Session().query(UserGroupUserGroupToPerm, Permission, UserGroup)\
      .join((UserGroup, UserGroupUserGroupToPerm.target_user_group_id
             == UserGroup.users_group_id))\
      .join((Permission, UserGroupUserGroupToPerm.permission_id
             == Permission.permission_id))\
      .join((UserGroupMember, UserGroupUserGroupToPerm.user_group_id
             == UserGroupMember.users_group_id))\
      .filter(UserGroupMember.user_id == uid)\
      .all()
     multiple_counter = collections.defaultdict(int)
     for perm in user_group_user_groups_perms:
         g_k = perm.UserGroupUserGroupToPerm.target_user_group.users_group_name
         multiple_counter[g_k] += 1
         p = perm.Permission.permission_name
         cur_perm = permissions[UK][g_k]
         if multiple_counter[g_k] > 1:
             p = _choose_perm(p, cur_perm)
         permissions[UK][g_k] = p
     #user explicit permission for user groups
     user_user_groups_perms = Permission.get_default_user_group_perms(uid)
     for perm in user_user_groups_perms:
         u_k = perm.UserUserGroupToPerm.user_group.users_group_name
         p = perm.Permission.permission_name
         cur_perm = permissions[UK][u_k]
         if not explicit:
             p = _choose_perm(p, cur_perm)
         permissions[UK][u_k] = p
     return permissions
 def allowed_api_access(controller_name, whitelist=None, api_key=None):
     """
     Check if given controller_name is in whitelist API access
     """
     if not whitelist:
         from kallithea import CONFIG
         whitelist = aslist(CONFIG.get('api_access_controllers_whitelist'),
                            sep=',')
         log.debug('whitelist of API access is: %s' % (whitelist))
     api_access_valid = controller_name in whitelist
     if api_access_valid:
         log.debug('controller:%s is in API whitelist' % (controller_name))
     else:
         msg = 'controller: %s is *NOT* in API whitelist' % (controller_name)
         if api_key:
             #if we use API key and don't have access it's a warning
             log.warning(msg)
         else:
             log.debug(msg)
     return api_access_valid
 class AuthUser(object):
     """
     A simple object that handles all attributes of user in Kallithea
     It does lookup based on API key,given user, or user present in session
     Then it fills all required information for such user. It also checks if
     anonymous access is enabled and if so, it returns default user as logged in
     """
     def __init__(self, user_id=None, api_key=None, username=None, ip_addr=None):
         self.user_id = user_id
         self._api_key = api_key
         self.api_key = None
         self.username = username
         self.ip_addr = ip_addr
         self.name = ''
         self.lastname = ''
         self.email = ''
         self.is_authenticated = False
         self.admin = False
         self.inherit_default_permissions = False
         self.propagate_data()
         self._instance = None
     @LazyProperty
     def permissions(self):
         return self.get_perms(user=self, cache=False)
     @property
     def api_keys(self):
         return self.get_api_keys()
     def propagate_data(self):
         user_model = UserModel()
         self.anonymous_user = User.get_default_user(cache=True)
         is_user_loaded = False
         # lookup by userid
         if self.user_id is not None and self.user_id != self.anonymous_user.user_id:
             log.debug('Auth User lookup by USER ID %s' % self.user_id)
             is_user_loaded = user_model.fill_data(self, user_id=self.user_id)
         # try go get user by api key
         elif self._api_key and self._api_key != self.anonymous_user.api_key:
             log.debug('Auth User lookup by API KEY %s' % self._api_key)
             is_user_loaded = user_model.fill_data(self, api_key=self._api_key)
         # lookup by username
         elif self.username:
             log.debug('Auth User lookup by USER NAME %s' % self.username)
             is_user_loaded = user_model.fill_data(self, username=self.username)
         else:
             log.debug('No data in %s that could been used to log in' % self)
         if not is_user_loaded:
             # if we cannot authenticate user try anonymous
             if self.anonymous_user.active:
                 user_model.fill_data(self, user_id=self.anonymous_user.user_id)
                 # then we set this user is logged in
                 self.is_authenticated = True
             else:
                 self.user_id = None
                 self.username = None
                 self.is_authenticated = False
         if not self.username:
             self.username = 'None'
         log.debug('Auth User is now %s' % self)
     def get_perms(self, user, explicit=True, algo='higherwin', cache=False):
         """
         Fills user permission attribute with permissions taken from database
         works for permissions given for repositories, and for permissions that
         are granted to groups
         :param user: instance of User object from database
         :param explicit: In case there are permissions both for user and a group
             that user is part of, explicit flag will define if user will
             explicitly override permissions from group, if it's False it will
             make decision based on the algo
         :param algo: algorithm to decide what permission should be choose if
             it's multiple defined, eg user in two different groups. It also
             decides if explicit flag is turned off how to specify the permission
             for case when user is in a group + have defined separate permission
         """
         user_id = user.user_id
         user_is_admin = user.is_admin
         user_inherit_default_permissions = user.inherit_default_permissions
         log.debug('Getting PERMISSION tree')
         compute = conditional_cache('short_term', 'cache_desc',
                                     condition=cache, func=_cached_perms_data)
         return compute(user_id, user_is_admin,
                        user_inherit_default_permissions, explicit, algo)
     def get_api_keys(self):
         api_keys = [self.api_key]
         for api_key in UserApiKeys.query()\
                 .filter(UserApiKeys.user_id == self.user_id)\
                 .filter(or_(UserApiKeys.expires == -1,
                             UserApiKeys.expires >= time.time())).all():
             api_keys.append(api_key.api_key)
         return api_keys
     @property
     def is_admin(self):
         return self.admin
     @property
     def repositories_admin(self):
         """
         Returns list of repositories you're an admin of
         """
         return [x[0] for x in self.permissions['repositories'].iteritems()
                 if x[1] == 'repository.admin']
     @property
     def repository_groups_admin(self):
         """
         Returns list of repository groups you're an admin of
         """
         return [x[0] for x in self.permissions['repositories_groups'].iteritems()
                 if x[1] == 'group.admin']
     @property
     def user_groups_admin(self):
         """
         Returns list of user groups you're an admin of
         """
         return [x[0] for x in self.permissions['user_groups'].iteritems()
                 if x[1] == 'usergroup.admin']
     @property
     def ip_allowed(self):
         """
         Checks if ip_addr used in constructor is allowed from defined list of
         allowed ip_addresses for user
         :returns: boolean, True if ip is in allowed ip range
         """
         # check IP
         inherit = self.inherit_default_permissions
         return AuthUser.check_ip_allowed(self.user_id, self.ip_addr,
                                          inherit_from_default=inherit)
     @classmethod
     def check_ip_allowed(cls, user_id, ip_addr, inherit_from_default):
         allowed_ips = AuthUser.get_allowed_ips(user_id, cache=True,
                         inherit_from_default=inherit_from_default)
         if check_ip_access(source_ip=ip_addr, allowed_ips=allowed_ips):
             log.debug('IP:%s is in range of %s' % (ip_addr, allowed_ips))
             return True
         else:
             log.info('Access for IP:%s forbidden, '
                      'not in %s' % (ip_addr, allowed_ips))
             return False
     def __repr__(self):
         return "<AuthUser('id:%s[%s] ip:%s auth:%s')>"\
             % (self.user_id, self.username, self.ip_addr, self.is_authenticated)
     def set_authenticated(self, authenticated=True):
         if self.user_id != self.anonymous_user.user_id:
             self.is_authenticated = authenticated
     def get_cookie_store(self):
         return {'username': self.username,
                 'user_id': self.user_id,
                 'is_authenticated': self.is_authenticated}
     @classmethod
     def from_cookie_store(cls, cookie_store):
         """
         Creates AuthUser from a cookie store
         :param cls:
         :param cookie_store:
         """
         user_id = cookie_store.get('user_id')
         username = cookie_store.get('username')
         api_key = cookie_store.get('api_key')
         return AuthUser(user_id, api_key, username)
     @classmethod
     def get_allowed_ips(cls, user_id, cache=False, inherit_from_default=False):
         _set = set()
         if inherit_from_default:
             default_ips = UserIpMap.query().filter(UserIpMap.user ==
                                             User.get_default_user(cache=True))
             if cache:
                 default_ips = default_ips.options(FromCache("sql_cache_short",
                                                   "get_user_ips_default"))
             # populate from default user
             for ip in default_ips:
                 try:
                     _set.add(ip.ip_addr)
                 except ObjectDeletedError:
                     # since we use heavy caching sometimes it happens that we get
                     # deleted objects here, we just skip them
                     pass
         user_ips = UserIpMap.query().filter(UserIpMap.user_id == user_id)
         if cache:
             user_ips = user_ips.options(FromCache("sql_cache_short",
                                                   "get_user_ips_%s" % user_id))
         for ip in user_ips:
             try:
                 _set.add(ip.ip_addr)
             except ObjectDeletedError:
                 # since we use heavy caching sometimes it happens that we get
                 # deleted objects here, we just skip them
                 pass
         return _set or set(['0.0.0.0/0', '::/0'])
 def set_available_permissions(config):
     """
     This function will propagate pylons globals with all available defined
     permission given in db. We don't want to check each time from db for new
     permissions since adding a new permission also requires application restart
     ie. to decorate new views with the newly created permission
     :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:
             log.debug('Permission denied for %s %s' % (cls, self.user))
             anonymous = self.user.username == User.DEFAULT_USER
             if anonymous:
                 return redirect_to_login(_('You need to be signed in to view this page'))
             else:
                 # redirect with forbidden ret code
                 return abort(403)
     def check_permissions(self):
         """Dummy function for overriding"""
         raise Exception('You have to write this function in child class')
 class HasPermissionAllDecorator(PermsDecorator):
     """
     Checks for access permission for all given predicates. All of them
     have to be meet in order to fulfill the request
     """
     def check_permissions(self):
         if self.required_perms.issubset(self.user_perms.get('global')):
             return True
         return False
 class HasPermissionAnyDecorator(PermsDecorator):
     """
     Checks for access permission for any of given predicates. In order to
     fulfill the request any of predicates must be meet
     """
     def check_permissions(self):
         if self.required_perms.intersection(self.user_perms.get('global')):
             return True
         return False
 class HasRepoPermissionAllDecorator(PermsDecorator):
     """
     Checks for access permission for all given predicates for specific
     repository. All of them have to be meet in order to fulfill the request
     """
     def check_permissions(self):
         repo_name = get_repo_slug(request)
         try:
             user_perms = set([self.user_perms['repositories'][repo_name]])
         except KeyError:
             return False
         if self.required_perms.issubset(user_perms):
             return True
         return False
 class HasRepoPermissionAnyDecorator(PermsDecorator):
     """
     Checks for access permission for any of given predicates for specific
     repository. In order to fulfill the request any of predicates must be meet
     """
     def check_permissions(self):
         repo_name = get_repo_slug(request)
         try:
             user_perms = set([self.user_perms['repositories'][repo_name]])
         except KeyError:
             return False
         if self.required_perms.intersection(user_perms):
             return True
         return False
 class HasRepoGroupPermissionAllDecorator(PermsDecorator):
     """
     Checks for access permission for all given predicates for specific
     repository group. All of them have to be meet in order to fulfill the request
     """
     def check_permissions(self):
         group_name = get_repo_group_slug(request)
         try:
             user_perms = set([self.user_perms['repositories_groups'][group_name]])
         except KeyError:
             return False
         if self.required_perms.issubset(user_perms):
             return True
         return False
 class HasRepoGroupPermissionAnyDecorator(PermsDecorator):
     """
     Checks for access permission for any of given predicates for specific
     repository group. In order to fulfill the request any of predicates must be meet
     """
     def check_permissions(self):
         group_name = get_repo_group_slug(request)
         try:
             user_perms = set([self.user_perms['repositories_groups'][group_name]])
         except KeyError:
             return False
         if self.required_perms.intersection(user_perms):
             return True
         return False
 class HasUserGroupPermissionAllDecorator(PermsDecorator):
     """
     Checks for access permission for all given predicates for specific
     user group. All of them have to be meet in order to fulfill the request
     """
     def check_permissions(self):
         group_name = get_user_group_slug(request)
         try:
             user_perms = set([self.user_perms['user_groups'][group_name]])
         except KeyError:
             return False
         if self.required_perms.issubset(user_perms):
             return True
         return False
 class HasUserGroupPermissionAnyDecorator(PermsDecorator):
     """
     Checks for access permission for any of given predicates for specific
     user group. In order to fulfill the request any of predicates must be meet
     """
     def check_permissions(self):
         group_name = get_user_group_slug(request)
         try:
             user_perms = set([self.user_perms['user_groups'][group_name]])
         except KeyError:
             return False
         if self.required_perms.intersection(user_perms):
             return True
         return False
 #==============================================================================
 # CHECK FUNCTIONS
 #==============================================================================
 class PermsFunction(object):
     """Base function for other check functions"""
     def __init__(self, *perms):
         self.required_perms = set(perms)
         self.user_perms = None
         self.repo_name = None
         self.group_name = None
     def __call__(self, check_location='', user=None):
         if not user:
             #TODO: remove this someday,put as user as attribute here
             user = request.user
         # init auth user if not already given
         if not isinstance(user, AuthUser):
             user = AuthUser(user.user_id)
         cls_name = self.__class__.__name__
         check_scope = {
             'HasPermissionAll': '',
             'HasPermissionAny': '',
             'HasRepoPermissionAll': 'repo:%s' % self.repo_name,
             'HasRepoPermissionAny': 'repo:%s' % self.repo_name,
             'HasRepoGroupPermissionAll': 'group:%s' % self.group_name,
             'HasRepoGroupPermissionAny': 'group:%s' % self.group_name,
         }.get(cls_name, '?')
         log.debug('checking cls:%s %s usr:%s %s @ %s', cls_name,
                   self.required_perms, user, check_scope,
                   check_location or 'unspecified location')
         if not user:
             log.debug('Empty request user')
             return False
         self.user_perms = user.permissions
         if self.check_permissions():
             log.debug('Permission to %s granted for user: %s @ %s'
                       % (check_scope, user,
                          check_location or 'unspecified location'))
             return True
         else:
             log.debug('Permission to %s denied for user: %s @ %s'
                       % (check_scope, user,
                          check_location or 'unspecified location'))
             return False
     def check_permissions(self):
         """Dummy function for overriding"""
         raise Exception('You have to write this function in child class')
 class HasPermissionAll(PermsFunction):
     def check_permissions(self):
         if self.required_perms.issubset(self.user_perms.get('global')):
             return True
         return False
 class HasPermissionAny(PermsFunction):
     def check_permissions(self):
         if self.required_perms.intersection(self.user_perms.get('global')):
             return True
         return False
 class HasRepoPermissionAll(PermsFunction):
     def __call__(self, repo_name=None, check_location='', user=None):
         self.repo_name = repo_name
         return super(HasRepoPermissionAll, self).__call__(check_location, user)
     def check_permissions(self):
         if not self.repo_name:
             self.repo_name = get_repo_slug(request)
         try:
             self._user_perms = set(
                 [self.user_perms['repositories'][self.repo_name]]
+            )
         except KeyError:
             return False
         if self.required_perms.issubset(self._user_perms):
             return True
         return False
 class HasRepoPermissionAny(PermsFunction):
     def __call__(self, repo_name=None, check_location='', user=None):
         self.repo_name = repo_name
         return super(HasRepoPermissionAny, self).__call__(check_location, user)
     def check_permissions(self):
         if not self.repo_name:
             self.repo_name = get_repo_slug(request)
         try:
             self._user_perms = set(
                 [self.user_perms['repositories'][self.repo_name]]
+            )
         except KeyError:
             return False
         if self.required_perms.intersection(self._user_perms):
             return True
         return False
 class HasRepoGroupPermissionAny(PermsFunction):
     def __call__(self, group_name=None, check_location='', user=None):
         self.group_name = group_name
         return super(HasRepoGroupPermissionAny, self).__call__(check_location, user)
     def check_permissions(self):
         try:
             self._user_perms = set(
                 [self.user_perms['repositories_groups'][self.group_name]]
+            )
         except KeyError:
             return False
         if self.required_perms.intersection(self._user_perms):
             return True
         return False
 class HasRepoGroupPermissionAll(PermsFunction):
     def __call__(self, group_name=None, check_location='', user=None):
         self.group_name = group_name
         return super(HasRepoGroupPermissionAll, self).__call__(check_location, user)
     def check_permissions(self):
         try:
             self._user_perms = set(
                 [self.user_perms['repositories_groups'][self.group_name]]
+            )
         except KeyError:
             return False
         if self.required_perms.issubset(self._user_perms):
             return True
         return False
 class HasUserGroupPermissionAny(PermsFunction):
     def __call__(self, user_group_name=None, check_location='', user=None):
         self.user_group_name = user_group_name
         return super(HasUserGroupPermissionAny, self).__call__(check_location, user)
     def check_permissions(self):
         try:
             self._user_perms = set(
                 [self.user_perms['user_groups'][self.user_group_name]]
+            )
         except KeyError:
             return False
         if self.required_perms.intersection(self._user_perms):
             return True
         return False
 class HasUserGroupPermissionAll(PermsFunction):
     def __call__(self, user_group_name=None, check_location='', user=None):
         self.user_group_name = user_group_name
         return super(HasUserGroupPermissionAll, self).__call__(check_location, user)
     def check_permissions(self):
         try:
             self._user_perms = set(
                 [self.user_perms['user_groups'][self.user_group_name]]
+            )
         except KeyError:
             return False
         if self.required_perms.issubset(self._user_perms):
             return True
         return False
 #==============================================================================
 # SPECIAL VERSION TO HANDLE MIDDLEWARE AUTH
 #==============================================================================
 class HasPermissionAnyMiddleware(object):
     def __init__(self, *perms):
         self.required_perms = set(perms)
     def __call__(self, user, repo_name):
         # repo_name MUST be unicode, since we handle keys in permission
         # dict by unicode
         repo_name = safe_unicode(repo_name)
         usr = AuthUser(user.user_id)
         self.user_perms = set([usr.permissions['repositories'][repo_name]])
         self.username = user.username
         self.repo_name = repo_name

kallithea/tests/functional/test_login.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 from __future__ import with_statement
 import mock
 from kallithea.tests import *
 from kallithea.tests.fixture import Fixture
 from kallithea.lib.utils2 import generate_api_key
 from kallithea.lib.auth import check_password
 from kallithea.lib import helpers as h
 from kallithea.model.api_key import ApiKeyModel
 from kallithea.model import validators
 from kallithea.model.db import User, Notification
 from kallithea.model.meta import Session
 fixture = Fixture()
 class TestLoginController(TestController):
     def tearDown(self):
         for n in Notification.query().all():
             Session().delete(n)
         Session().commit()
         self.assertEqual(Notification.query().all(), [])
     def test_index(self):
         response = self.app.get(url(controller='login', action='index'))
         self.assertEqual(response.status, '200 OK')
         # Test response...
     def test_login_admin_ok(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': 'test_admin',
                                   'password': 'test12'})
         self.assertEqual(response.status, '302 Found')
         self.assertEqual(response.session['authuser'].get('username'),
                          'test_admin')
         response = response.follow()
         response.mustcontain('/%s' % HG_REPO)
     def test_login_regular_ok(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': 'test_regular',
                                   'password': 'test12'})
         self.assertEqual(response.status, '302 Found')
         self.assertEqual(response.session['authuser'].get('username'),
                          'test_regular')
         response = response.follow()
         response.mustcontain('/%s' % HG_REPO)
     def test_login_ok_came_from(self):
         test_came_from = '/_admin/users'
         response = self.app.post(url(controller='login', action='index',
                                      came_from=test_came_from),
                                  {'username': 'test_admin',
                                   'password': 'test12'})
         self.assertEqual(response.status, '302 Found')
         response = response.follow()
         self.assertEqual(response.status, '200 OK')
         response.mustcontain('Users Administration')
     @parameterized.expand([
           ('data:text/html,<script>window.alert("xss")</script>',),
           ('mailto:test@example.com',),
           ('file:///etc/passwd',),
           ('ftp://some.ftp.server',),
           ('http://other.domain',),
     ])
     def test_login_bad_came_froms(self, url_came_from):
         response = self.app.post(url(controller='login', action='index',
                                      came_from=url_came_from),
                                  {'username': 'test_admin',
                                   'password': 'test12'})
         self.assertEqual(response.status, '302 Found')
         self.assertEqual(response._environ['paste.testing_variables']
                          ['tmpl_context'].came_from, '/')
         response = response.follow()
         self.assertEqual(response.status, '200 OK')
     def test_login_short_password(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': 'test_admin',
                                   'password': 'as'})
         self.assertEqual(response.status, '200 OK')
         response.mustcontain('Enter 3 characters or more')
     def test_login_wrong_username_password(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': 'error',
                                   'password': 'test12'})
         response.mustcontain('invalid user name')
         response.mustcontain('invalid password')
     #==========================================================================
     # REGISTRATIONS
     #==========================================================================
     def test_register(self):
         response = self.app.get(url(controller='login', action='register'))
         response.mustcontain('Sign Up')
     def test_register_err_same_username(self):
         uname = 'test_admin'
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': uname,
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': 'goodmail@domain.com',
                                              'firstname': 'test',
                                              'lastname': 'test'})
         msg = validators.ValidUsername()._messages['username_exists']
         msg = h.html_escape(msg % {'username': uname})
         response.mustcontain(msg)
     def test_register_err_same_email(self):
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': 'test_admin_0',
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': 'test_admin@mail.com',
                                              'firstname': 'test',
                                              'lastname': 'test'})
         msg = validators.UniqSystemEmail()()._messages['email_taken']
         response.mustcontain(msg)
     def test_register_err_same_email_case_sensitive(self):
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': 'test_admin_1',
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': 'TesT_Admin@mail.COM',
                                              'firstname': 'test',
                                              'lastname': 'test'})
         msg = validators.UniqSystemEmail()()._messages['email_taken']
         response.mustcontain(msg)
     def test_register_err_wrong_data(self):
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': 'xs',
                                              'password': 'test',
                                              'password_confirmation': 'test',
                                              'email': 'goodmailm',
                                              'firstname': 'test',
                                              'lastname': 'test'})
         self.assertEqual(response.status, '200 OK')
         response.mustcontain('An email address must contain a single @')
         response.mustcontain('Enter a value 6 characters long or more')
     def test_register_err_username(self):
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': 'error user',
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': 'goodmailm',
                                              'firstname': 'test',
                                              'lastname': 'test'})
         response.mustcontain('An email address must contain a single @')
         response.mustcontain('Username may only contain '
                 'alphanumeric characters underscores, '
                 'periods or dashes and must begin with '
                 'alphanumeric character')
     def test_register_err_case_sensitive(self):
         usr = 'Test_Admin'
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': usr,
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': 'goodmailm',
                                              'firstname': 'test',
                                              'lastname': 'test'})
         response.mustcontain('An email address must contain a single @')
         msg = validators.ValidUsername()._messages['username_exists']
         msg = h.html_escape(msg % {'username': usr})
         response.mustcontain(msg)
     def test_register_special_chars(self):
         response = self.app.post(url(controller='login', action='register'),
                                         {'username': 'xxxaxn',
                                          'password': 'ąćźżąśśśś',
                                          'password_confirmation': 'ąćźżąśśśś',
                                          'email': 'goodmailm@test.plx',
                                          'firstname': 'test',
                                          'lastname': 'test'})
         msg = validators.ValidPassword()._messages['invalid_password']
         response.mustcontain(msg)
     def test_register_password_mismatch(self):
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': 'xs',
                                              'password': '123qwe',
                                              'password_confirmation': 'qwe123',
                                              'email': 'goodmailm@test.plxa',
                                              'firstname': 'test',
                                              'lastname': 'test'})
         msg = validators.ValidPasswordsMatch()._messages['password_mismatch']
         response.mustcontain(msg)
     def test_register_ok(self):
         username = 'test_regular4'
         password = 'qweqwe'
         email = 'username@test.com'
         name = 'testname'
         lastname = 'testlastname'
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': username,
                                              'password': password,
                                              'password_confirmation': password,
                                              'email': email,
                                              'firstname': name,
                                              'lastname': lastname,
                                              'admin': True})  # This should be overriden
         self.assertEqual(response.status, '302 Found')
         self.checkSessionFlash(response, 'You have successfully registered into Kallithea')
         ret = Session().query(User).filter(User.username == 'test_regular4').one()
         self.assertEqual(ret.username, username)
         self.assertEqual(check_password(password, ret.password), True)
         self.assertEqual(ret.email, email)
         self.assertEqual(ret.name, name)
         self.assertEqual(ret.lastname, lastname)
         self.assertNotEqual(ret.api_key, None)
         self.assertEqual(ret.admin, False)
     def test_forgot_password_wrong_mail(self):
         bad_email = 'username@wrongmail.org'
         response = self.app.post(
                         url(controller='login', action='password_reset'),
                             {'email': bad_email, }
+        )
         msg = validators.ValidSystemEmail()._messages['non_existing_email']
         msg = h.html_escape(msg % {'email': bad_email})
         response.mustcontain()
     def test_forgot_password(self):
         response = self.app.get(url(controller='login',
                                     action='password_reset'))
         self.assertEqual(response.status, '200 OK')
         username = 'test_password_reset_1'
         password = 'qweqwe'
         email = 'username@python-works.com'
         name = 'passwd'
         lastname = 'reset'
         new = User()
         new.username = username
         new.password = password
         new.email = email
         new.name = name
         new.lastname = lastname
         new.api_key = generate_api_key(username)
         Session().add(new)
         Session().commit()
         response = self.app.post(url(controller='login',
                                      action='password_reset'),
                                  {'email': email, })
         self.checkSessionFlash(response, 'Your password reset link was sent')
         response = response.follow()
         # 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):
                 self.app.get(url(controller='changeset',
                                  action='changeset_raw',
                                  repo_name=HG_REPO, revision='tip',
                                  api_key=new_api_key.api_key),
                              status=302)

0 comments (0 inline, 0 general)