kallithea Changeset - 06398585de03

Changeset - 06398585de03

Parent rev.

Child rev.

[Not reviewed]

default

0 3 0

Søren Løvborg - 9 years ago 2016-11-10 20:38:40
sorenl@unity3d.com

auth: track API key used for authentication in AuthUser

This allows us to define only once how an API key is passed to the app.
We might e.g. allow API keys to be passed in an HTTP header; with this
change, we only need to update the code in one place.

Also change the code to verify up front that the API key resolved to a
valid and active user, so LoginRequired doesn't need to do that.

Also return plain 403 Forbidden for bad API keys instead of redirecting
to the login form, which makes more sense for non-interactive clients
(the typical users of API keys).

3 files changed with 22 insertions and 22 deletions:

kallithea/lib/auth.py

kallithea/lib/base.py

kallithea/tests/functional/test_login.py

0 comments (0 inline, 0 general)

kallithea/lib/auth.py

➞

Show inline comments

@@ @@ -92,970 +92,967 @@ class PasswordGenerator(object): @@
             if i < len(alphabet):
                 l.append(alphabet[i])
         return ''.join(l)
 def get_crypt_password(password):
     """
     Cryptographic function used for password hashing based on pybcrypt
     or Python's own OpenSSL wrapper on windows
     :param password: password to hash
     """
     if is_windows:
         return hashlib.sha256(password).hexdigest()
     elif is_unix:
         import bcrypt
         return bcrypt.hashpw(safe_str(password), bcrypt.gensalt(10))
     else:
         raise Exception('Unknown or unsupported platform %s' \
                         % __platform__)
 def check_password(password, hashed):
     """
     Checks matching password with it's hashed value, runs different
     implementation based on platform it runs on
     :param password: password
     :param hashed: password in hashed form
     """
     if is_windows:
         return hashlib.sha256(password).hexdigest() == hashed
     elif is_unix:
         import bcrypt
         return bcrypt.checkpw(safe_str(password), safe_str(hashed))
     else:
         raise Exception('Unknown or unsupported platform %s' \
                         % __platform__)
 def _cached_perms_data(user_id, user_is_admin, user_inherit_default_permissions,
                        explicit, algo):
     RK = 'repositories'
     GK = 'repositories_groups'
     UK = 'user_groups'
     GLOBAL = 'global'
     PERM_WEIGHTS = Permission.PERM_WEIGHTS
     permissions = {RK: {}, GK: {}, UK: {}, GLOBAL: set()}
     def _choose_perm(new_perm, cur_perm):
         new_perm_val = PERM_WEIGHTS[new_perm]
         cur_perm_val = PERM_WEIGHTS[cur_perm]
         if algo == 'higherwin':
             if new_perm_val > cur_perm_val:
                 return new_perm
             return cur_perm
         elif algo == 'lowerwin':
             if new_perm_val < cur_perm_val:
                 return new_perm
             return cur_perm
     #======================================================================
     # fetch default permissions
     #======================================================================
     default_user = User.get_by_username('default', cache=True)
     default_user_id = default_user.user_id
     default_repo_perms = Permission.get_default_perms(default_user_id)
     default_repo_groups_perms = Permission.get_default_group_perms(default_user_id)
     default_user_group_perms = Permission.get_default_user_group_perms(default_user_id)
     if user_is_admin:
         #==================================================================
         # admin users have all rights;
         # based on default permissions, just set everything to admin
         #==================================================================
         permissions[GLOBAL].add('hg.admin')
         permissions[GLOBAL].add('hg.create.write_on_repogroup.true')
         # repositories
         for perm in default_repo_perms:
             r_k = perm.UserRepoToPerm.repository.repo_name
             p = 'repository.admin'
             permissions[RK][r_k] = p
         # repository groups
         for perm in default_repo_groups_perms:
             rg_k = perm.UserRepoGroupToPerm.group.group_name
             p = 'group.admin'
             permissions[GK][rg_k] = p
         # user groups
         for perm in default_user_group_perms:
             u_k = perm.UserUserGroupToPerm.user_group.users_group_name
             p = 'usergroup.admin'
             permissions[UK][u_k] = p
         return permissions
     #==================================================================
     # SET DEFAULTS GLOBAL, REPOS, REPOSITORY GROUPS
     #==================================================================
     # default global permissions taken from the default user
     default_global_perms = UserToPerm.query() \
         .filter(UserToPerm.user_id == default_user_id) \
         .options(joinedload(UserToPerm.permission))
     for perm in default_global_perms:
         permissions[GLOBAL].add(perm.permission.permission_name)
     # defaults for repositories, taken from default user
     for perm in default_repo_perms:
         r_k = perm.UserRepoToPerm.repository.repo_name
         if perm.Repository.private and not (perm.Repository.owner_id == user_id):
             # disable defaults for private repos,
             p = 'repository.none'
         elif perm.Repository.owner_id == user_id:
             # set admin if owner
             p = 'repository.admin'
         else:
             p = perm.Permission.permission_name
         permissions[RK][r_k] = p
     # defaults for repository groups taken from default user permission
     # on given group
     for perm in default_repo_groups_perms:
         rg_k = perm.UserRepoGroupToPerm.group.group_name
         p = perm.Permission.permission_name
         permissions[GK][rg_k] = p
     # defaults for user groups taken from default user permission
     # on given user group
     for perm in default_user_group_perms:
         u_k = perm.UserUserGroupToPerm.user_group.users_group_name
         p = perm.Permission.permission_name
         permissions[UK][u_k] = p
     #======================================================================
     # !! OVERRIDE GLOBALS !! with user permissions if any found
     #======================================================================
     # those can be configured from groups or users explicitly
     _configurable = set([
         'hg.fork.none', 'hg.fork.repository',
         'hg.create.none', 'hg.create.repository',
         'hg.usergroup.create.false', 'hg.usergroup.create.true'
     ])
     # USER GROUPS comes first
     # user group global permissions
     user_perms_from_users_groups = Session().query(UserGroupToPerm) \
         .options(joinedload(UserGroupToPerm.permission)) \
         .join((UserGroupMember, UserGroupToPerm.users_group_id ==
                UserGroupMember.users_group_id)) \
         .filter(UserGroupMember.user_id == user_id) \
         .join((UserGroup, UserGroupMember.users_group_id ==
                UserGroup.users_group_id)) \
         .filter(UserGroup.users_group_active == True) \
         .order_by(UserGroupToPerm.users_group_id) \
         .all()
     # need to group here by groups since user can be in more than
     # one group
     _grouped = [[x, list(y)] for x, y in
                 itertools.groupby(user_perms_from_users_groups,
                                   lambda x:x.users_group)]
     for gr, perms in _grouped:
         # since user can be in multiple groups iterate over them and
         # select the lowest permissions first (more explicit)
         ##TODO: do this^^
         if not gr.inherit_default_permissions:
             # NEED TO IGNORE all configurable permissions and
             # replace them with explicitly set
             permissions[GLOBAL] = permissions[GLOBAL] \
                                             .difference(_configurable)
         for perm in perms:
             permissions[GLOBAL].add(perm.permission.permission_name)
     # user specific global permissions
     user_perms = Session().query(UserToPerm) \
             .options(joinedload(UserToPerm.permission)) \
             .filter(UserToPerm.user_id == user_id).all()
     if not user_inherit_default_permissions:
         # NEED TO IGNORE all configurable permissions and
         # replace them with explicitly set
         permissions[GLOBAL] = permissions[GLOBAL] \
                                         .difference(_configurable)
         for perm in user_perms:
             permissions[GLOBAL].add(perm.permission.permission_name)
     ## END GLOBAL PERMISSIONS
     #======================================================================
     # !! PERMISSIONS FOR REPOSITORIES !!
     #======================================================================
     #======================================================================
     # check if user is part of user groups for this repository and
     # fill in his permission from it. _choose_perm decides of which
     # permission should be selected based on selected method
     #======================================================================
     # user group for repositories permissions
     user_repo_perms_from_users_groups = \
      Session().query(UserGroupRepoToPerm, Permission, Repository,) \
         .join((Repository, UserGroupRepoToPerm.repository_id ==
                Repository.repo_id)) \
         .join((Permission, UserGroupRepoToPerm.permission_id ==
                Permission.permission_id)) \
         .join((UserGroup, UserGroupRepoToPerm.users_group_id ==
                UserGroup.users_group_id)) \
         .filter(UserGroup.users_group_active == True) \
         .join((UserGroupMember, UserGroupRepoToPerm.users_group_id ==
                UserGroupMember.users_group_id)) \
         .filter(UserGroupMember.user_id == user_id) \
         .all()
     multiple_counter = collections.defaultdict(int)
     for perm in user_repo_perms_from_users_groups:
         r_k = perm.UserGroupRepoToPerm.repository.repo_name
         multiple_counter[r_k] += 1
         p = perm.Permission.permission_name
         cur_perm = permissions[RK][r_k]
         if perm.Repository.owner_id == user_id:
             # set admin if owner
             p = 'repository.admin'
         else:
             if multiple_counter[r_k] > 1:
                 p = _choose_perm(p, cur_perm)
         permissions[RK][r_k] = p
     # user explicit permissions for repositories, overrides any specified
     # by the group permission
     user_repo_perms = Permission.get_default_perms(user_id)
     for perm in user_repo_perms:
         r_k = perm.UserRepoToPerm.repository.repo_name
         cur_perm = permissions[RK][r_k]
         # set admin if owner
         if perm.Repository.owner_id == user_id:
             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((UserGroup, UserGroupRepoGroupToPerm.users_group_id ==
             UserGroup.users_group_id)) \
      .filter(UserGroup.users_group_active == True) \
      .join((UserGroupMember, UserGroupRepoGroupToPerm.users_group_id
             == UserGroupMember.users_group_id)) \
      .filter(UserGroupMember.user_id == user_id) \
      .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(user_id)
     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 == user_id) \
      .join((UserGroup, UserGroupMember.users_group_id ==
             UserGroup.users_group_id), aliased=True, from_joinpoint=True) \
      .filter(UserGroup.users_group_active == True) \
      .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(user_id)
     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):
     """
     Represents a Kallithea user, including various authentication and
     authorization information. Typically used to store the current user,
     but is also used as a generic user information data structure in
     parts of the code, e.g. user management.
     Constructed from a database `User` object, a user ID or cookie dict,
     it looks up the user (if needed) and copies all attributes to itself,
     adding various non-persistent data. If lookup fails but anonymous
     access to Kallithea is enabled, the default user is loaded instead.
     `AuthUser` does not by itself authenticate users and the constructor
     sets the `is_authenticated` field to False. It's up to other parts
     of the code to check e.g. if a supplied password is correct, and if
     so, set `is_authenticated` to True.
     However, `AuthUser` does refuse to load a user that is not `active`.
     Note that Kallithea distinguishes between the default user (an actual
     user in the database with username "default") and "no user" (no actual
     User object, AuthUser filled with blank values and username "None").
     If the default user is active, that will always be used instead of
     "no user". On the other hand, if the default user is disabled (and
     there is no login information), we instead get "no user"; this should
     only happen on the login page (as all other requests are redirected).
     `is_default_user` specifically checks if the AuthUser is the user named
     "default". Use `is_anonymous` to check for both "default" and "no user".
     """
     def __init__(self, user_id=None, dbuser=None,
+    def __init__(self, user_id=None, dbuser=None, authenticating_api_key=None,
             is_external_auth=False):
         self.is_authenticated = False
         self.is_external_auth = is_external_auth
         self.authenticating_api_key = authenticating_api_key
         user_model = UserModel()
         self._default_user = User.get_default_user(cache=True)
         # These attributes will be overridden by fill_data, below, unless the
         # requested user cannot be found and the default anonymous user is
         # not enabled.
         self.user_id = None
         self.username = None
         self.api_key = None
         self.name = ''
         self.lastname = ''
         self.email = ''
         self.admin = False
         self.inherit_default_permissions = False
         # Look up database user, if necessary.
         if user_id is not None:
             log.debug('Auth User lookup by USER ID %s', user_id)
             dbuser = user_model.get(user_id)
         else:
             # Note: dbuser is allowed to be None.
             log.debug('Auth User lookup by database user %s', dbuser)
         is_user_loaded = self._fill_data(dbuser)
         # If user cannot be found, try falling back to anonymous.
         if not is_user_loaded:
             is_user_loaded =  self._fill_data(self._default_user)
         self.is_default_user = (self.user_id == self._default_user.user_id)
         self.is_anonymous = not is_user_loaded or self.is_default_user
         if not self.username:
             self.username = 'None'
         log.debug('Auth User is now %s', self)
     def _fill_data(self, dbuser):
         """
         Copies database fields from a `db.User` to this `AuthUser`. Does
         not copy `api_keys` and `permissions` attributes.
         Checks that `dbuser` is `active` (and not None) before copying;
         returns True on success.
         """
         if dbuser is not None and dbuser.active:
             log.debug('filling %s data', dbuser)
             for k, v in dbuser.get_dict().iteritems():
                 assert k not in ['api_keys', 'permissions']
                 setattr(self, k, v)
             return True
         return False
     @LazyProperty
     def permissions(self):
         return self.__get_perms(user=self, cache=False)
     @property
     def api_keys(self):
         return self._get_api_keys()
     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: `AuthUser` instance
         :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']
     @staticmethod
     def check_ip_allowed(user, ip_addr):
         """
         Check if the given IP address (a `str`) is allowed for the given
         user (an `AuthUser` or `db.User`).
         """
         allowed_ips = AuthUser.get_allowed_ips(user.user_id, cache=True,
             inherit_from_default=user.inherit_default_permissions)
         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] auth:%s')>" \
             % (self.user_id, self.username, (self.is_authenticated or self.is_default_user))
     def to_cookie(self):
         """ Serializes this login session to a cookie `dict`. """
         return {
             'user_id': self.user_id,
             'is_external_auth': self.is_external_auth,
+        }
     @staticmethod
     def from_cookie(cookie):
         """
         Deserializes an `AuthUser` from a cookie `dict`.
         """
         au = AuthUser(
             user_id=cookie.get('user_id'),
             is_external_auth=cookie.get('is_external_auth', False),
+        )
         au.is_authenticated = True
         return au
     @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 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 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):
     """Return an exception that must be raised. It will redirect to the login
     page which will redirect back to the current URL after authentication.
     The optional message will be shown in a flash message."""
     from kallithea.lib import helpers as h
     if message:
         h.flash(h.literal(message), category='warning')
     p = request.path_qs
     log.debug('Redirecting to login page, origin: %s', p)
     return HTTPFound(location=url('login_home', came_from=p))
 class LoginRequired(object):
     """Client must be logged in as a valid User (but the "default" user,
     if enabled, is considered valid), or we'll redirect to the login page.
     Also checks that IP address is allowed, and if using API key instead
     of regular cookie authentication, checks that API key access is allowed
     (based on `api_access` parameter and the API view whitelist).
     """
     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):
         controller = fargs[0]
         user = controller.authuser
         loc = "%s:%s" % (controller.__class__.__name__, func.__name__)
         log.debug('Checking access for user %s @ %s', user, loc)
         if not AuthUser.check_ip_allowed(user, controller.ip_addr):
             raise _redirect_to_login(_('IP %s not allowed') % controller.ip_addr)
         # check if we used an API key and it's a valid one
         api_key = request.GET.get('api_key')
         # Check if we used an API key to authenticate.
         api_key = user.authenticating_api_key
         if api_key is not None:
             # explicit controller is enabled or API is in our whitelist
             if self.api_access or allowed_api_access(loc, api_key=api_key):
                 if api_key in user.api_keys:
             # Check that controller is enabled for API key usage.
             if not self.api_access and not allowed_api_access(loc, api_key=api_key):
                 # controller does not allow API access
                 log.warning('API access to %s is not allowed', loc)
                 raise HTTPForbidden()
                     log.info('user %s authenticated with API key ****%s @ %s',
                              user, api_key[-4:], loc)
                     return func(*fargs, **fkwargs)
                 else:
                     log.warning('API key ****%s is NOT valid', api_key[-4:])
                     raise _redirect_to_login(_('Invalid API key'))
             else:
                 # controller does not allow API access
                 log.warning('API access to %s is not allowed', loc)
                 raise HTTPForbidden()
         # CSRF protection: Whenever a request has ambient authority (whether
         # through a session cookie or its origin IP address), it must include
         # the correct token, unless the HTTP method is GET or HEAD (and thus
         # guaranteed to be side effect free. In practice, the only situation
         # where we allow side effects without ambient authority is when the
         # authority comes from an API key; and that is handled above.
         if request.method not in ['GET', 'HEAD']:
             token = request.POST.get(secure_form.token_key)
             if not token or token != secure_form.authentication_token():
                 log.error('CSRF check failed')
                 raise HTTPForbidden()
         # regular user authentication
         if user.is_authenticated or user.is_default_user:
             log.info('user %s authenticated with regular auth @ %s', user, loc)
             return func(*fargs, **fkwargs)
         else:
             log.warning('user %s NOT authenticated with regular auth @ %s', user, loc)
             raise _redirect_to_login()
 class NotAnonymous(object):
     """Ensures that client is not logged in as the "default" user, and
     redirects to the login page otherwise. Must be used together with
     LoginRequired."""
     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)
         if self.user.is_default_user:
             raise _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)
             if self.user.is_default_user:
                 raise _redirect_to_login(_('You need to be signed in to view this page'))
             else:
                 raise HTTPForbidden()
     def check_permissions(self):
         """Dummy function for overriding"""
         raise Exception('You have to write this function in child class')
 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 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 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 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 __nonzero__(self):
         """ Defend against accidentally forgetting to call the object
             and instead evaluating it directly in a boolean context,
             which could have security implications.
         """
         raise AssertionError(self.__class__.__name__ + ' is not a bool and must be called!')
     def __call__(self, check_location='unspecified location'):
         user = request.user
         assert user
         assert isinstance(user, AuthUser), user
         cls_name = self.__class__.__name__
         check_scope = self._scope()
         log.debug('checking cls:%s %s usr:%s %s @ %s', cls_name,
                   self.required_perms, user, check_scope,
                   check_location)
         self.user_perms = user.permissions
         result = self.check_permissions()
         result_text = 'granted' if result else 'denied'
         log.debug('Permission to %s %s for user: %s @ %s',
             check_scope, result_text, user, check_location)
         return result
     def check_permissions(self):
         """Dummy function for overriding"""
         raise Exception('You have to write this function in child class')
     def _scope(self):
         return '(unknown scope)'
 class HasPermissionAny(PermsFunction):
     def check_permissions(self):
         if self.required_perms.intersection(self.user_perms.get('global')):
             return True
         return False
     def _scope(self):
         return 'global'
 class HasRepoPermissionAny(PermsFunction):
     def __call__(self, repo_name=None, check_location=''):
         self.repo_name = repo_name
         return super(HasRepoPermissionAny, self).__call__(check_location)
     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
     def _scope(self):
         return 'repo:%s' % self.repo_name
 class HasRepoGroupPermissionAny(PermsFunction):
     def __call__(self, group_name=None, check_location=''):
         self.group_name = group_name
         return super(HasRepoGroupPermissionAny, self).__call__(check_location)
     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
     def _scope(self):
         return 'repogroup:%s' % self.group_name
 class HasUserGroupPermissionAny(PermsFunction):
     def __call__(self, user_group_name=None, check_location=''):
         self.user_group_name = user_group_name
         return super(HasUserGroupPermissionAny, self).__call__(check_location)
     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
     def _scope(self):
         return 'usergroup:%s' % self.user_group_name
 #==============================================================================
 # 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
         return self.check_permissions()
     def check_permissions(self):
         log.debug('checking VCS protocol '
                   'permissions %s for user:%s repository:%s', self.user_perms,
                                                 self.username, self.repo_name)
         if self.required_perms.intersection(self.user_perms):
             log.debug('Permission to repo: %s granted for user: %s @ %s',
                       self.repo_name, self.username, 'PermissionMiddleware')
             return True
         log.debug('Permission to repo: %s denied for user: %s @ %s',
                   self.repo_name, self.username, 'PermissionMiddleware')
         return False
 def check_ip_access(source_ip, allowed_ips=None):
     """
     Checks if source_ip is a subnet of any of allowed_ips.
     :param source_ip:
     :param allowed_ips: list of allowed ips together with mask
     """
     from kallithea.lib import ipaddr
     log.debug('checking if ip:%s is subnet of %s', source_ip, allowed_ips)
     if isinstance(allowed_ips, (tuple, list, set)):
         for ip in allowed_ips:
             if ipaddr.IPAddress(source_ip) in ipaddr.IPNetwork(ip):
                 log.debug('IP %s is network %s',
                           ipaddr.IPAddress(source_ip), ipaddr.IPNetwork(ip))
                 return True
     return False

kallithea/lib/base.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.lib.base
 ~~~~~~~~~~~~~~~~~~
 The base Controller API
 Provides the BaseController class for subclassing. And usage in different
 controllers
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Oct 06, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import datetime
 import logging
 import time
 import traceback
 import webob.exc
 import paste.httpexceptions
 import paste.auth.basic
 import paste.httpheaders
 from webhelpers.pylonslib import secure_form
 from pylons import config, tmpl_context as c, request, session
 from pylons.controllers import WSGIController
 from pylons.templating import render_mako as render  # don't remove this import
 from pylons.i18n.translation import _
 from kallithea import __version__, BACKENDS
 from kallithea.config.routing import url
 from kallithea.lib.utils2 import str2bool, safe_unicode, AttributeDict, \
     safe_str, safe_int
 from kallithea.lib import auth_modules
 from kallithea.lib.auth import AuthUser, HasPermissionAnyMiddleware
 from kallithea.lib.utils import get_repo_slug
 from kallithea.lib.exceptions import UserCreationError
 from kallithea.lib.vcs.exceptions import RepositoryError, EmptyRepositoryError, ChangesetDoesNotExistError
 from kallithea.model import meta
 from kallithea.model.db import PullRequest, Repository, Ui, User, Setting
 from kallithea.model.notification import NotificationModel
 from kallithea.model.scm import ScmModel
 log = logging.getLogger(__name__)
 def _filter_proxy(ip):
     """
     HEADERS can have multiple ips inside the left-most being the original
     client, and each successive proxy that passed the request adding the IP
     address where it received the request from.
     :param ip:
     """
     if ',' in ip:
         _ips = ip.split(',')
         _first_ip = _ips[0].strip()
         log.debug('Got multiple IPs %s, using %s', ','.join(_ips), _first_ip)
         return _first_ip
     return ip
 def _get_ip_addr(environ):
     proxy_key = 'HTTP_X_REAL_IP'
     proxy_key2 = 'HTTP_X_FORWARDED_FOR'
     def_key = 'REMOTE_ADDR'
     ip = environ.get(proxy_key)
     if ip:
         return _filter_proxy(ip)
     ip = environ.get(proxy_key2)
     if ip:
         return _filter_proxy(ip)
     ip = environ.get(def_key, '0.0.0.0')
     return _filter_proxy(ip)
 def _get_access_path(environ):
     path = environ.get('PATH_INFO')
     org_req = environ.get('pylons.original_request')
     if org_req:
         path = org_req.environ.get('PATH_INFO')
     return path
 def log_in_user(user, remember, is_external_auth):
     """
     Log a `User` in and update session and cookies. If `remember` is True,
     the session cookie is set to expire in a year; otherwise, it expires at
     the end of the browser session.
     Returns populated `AuthUser` object.
     """
     user.update_lastlogin()
     meta.Session().commit()
     auth_user = AuthUser(dbuser=user,
                          is_external_auth=is_external_auth)
     # It should not be possible to explicitly log in as the default user.
     assert not auth_user.is_default_user
     auth_user.is_authenticated = True
     # Start new session to prevent session fixation attacks.
     session.invalidate()
     session['authuser'] = cookie = auth_user.to_cookie()
     # If they want to be remembered, update the cookie.
     # NOTE: Assumes that beaker defaults to browser session cookie.
     if remember:
         t = datetime.datetime.now() + datetime.timedelta(days=365)
         session._set_cookie_expires(t)
     session.save()
     log.info('user %s is now authenticated and stored in '
              'session, session attrs %s', user.username, cookie)
     # dumps session attrs back to cookie
     session._update_cookie_out()
     return auth_user
 class BasicAuth(paste.auth.basic.AuthBasicAuthenticator):
     def __init__(self, realm, authfunc, auth_http_code=None):
         self.realm = realm
         self.authfunc = authfunc
         self._rc_auth_http_code = auth_http_code
     def build_authentication(self):
         head = paste.httpheaders.WWW_AUTHENTICATE.tuples('Basic realm="%s"' % self.realm)
         if self._rc_auth_http_code and self._rc_auth_http_code == '403':
             # return 403 if alternative http return code is specified in
             # Kallithea config
             return paste.httpexceptions.HTTPForbidden(headers=head)
         return paste.httpexceptions.HTTPUnauthorized(headers=head)
     def authenticate(self, environ):
         authorization = paste.httpheaders.AUTHORIZATION(environ)
         if not authorization:
             return self.build_authentication()
         (authmeth, auth) = authorization.split(' ', 1)
         if 'basic' != authmeth.lower():
             return self.build_authentication()
         auth = auth.strip().decode('base64')
         _parts = auth.split(':', 1)
         if len(_parts) == 2:
             username, password = _parts
             if self.authfunc(username, password, environ) is not None:
                 return username
         return self.build_authentication()
     __call__ = authenticate
 class BaseVCSController(object):
     def __init__(self, application, config):
         self.application = application
         self.config = config
         # base path of repo locations
         self.basepath = self.config['base_path']
         # authenticate this VCS request using the authentication modules
         self.authenticate = BasicAuth('', auth_modules.authenticate,
                                       config.get('auth_ret_code'))
         self.ip_addr = '0.0.0.0'
     def _handle_request(self, environ, start_response):
         raise NotImplementedError()
     def _get_by_id(self, repo_name):
         """
         Gets a special pattern _<ID> from clone url and tries to replace it
         with a repository_name for support of _<ID> permanent URLs
         :param repo_name:
         """
         data = repo_name.split('/')
         if len(data) >= 2:
             from kallithea.lib.utils import get_repo_by_id
             by_id_match = get_repo_by_id(repo_name)
             if by_id_match:
                 data[1] = safe_str(by_id_match)
         return '/'.join(data)
     def _invalidate_cache(self, repo_name):
         """
         Sets cache for this repository for invalidation on next access
         :param repo_name: full repo name, also a cache key
         """
         ScmModel().mark_for_invalidation(repo_name)
     def _check_permission(self, action, user, repo_name, ip_addr=None):
         """
         Checks permissions using action (push/pull) user and repository
         name
         :param action: push or pull action
         :param user: `User` instance
         :param repo_name: repository name
         """
         # check IP
         ip_allowed = AuthUser.check_ip_allowed(user, ip_addr)
         if ip_allowed:
             log.info('Access for IP:%s allowed', ip_addr)
         else:
             return False
         if action == 'push':
             if not HasPermissionAnyMiddleware('repository.write',
                                               'repository.admin')(user,
                                                                   repo_name):
                 return False
         else:
             #any other action need at least read permission
             if not HasPermissionAnyMiddleware('repository.read',
                                               'repository.write',
                                               'repository.admin')(user,
                                                                   repo_name):
                 return False
         return True
     def _get_ip_addr(self, environ):
         return _get_ip_addr(environ)
     def _check_locking_state(self, environ, action, repo, user_id):
         """
         Checks locking on this repository, if locking is enabled and lock is
         present returns a tuple of make_lock, locked, locked_by.
         make_lock can have 3 states None (do nothing) True, make lock
         False release lock, This value is later propagated to hooks, which
         do the locking. Think about this as signals passed to hooks what to do.
         """
         locked = False  # defines that locked error should be thrown to user
         make_lock = None
         repo = Repository.get_by_repo_name(repo)
         user = User.get(user_id)
         # this is kind of hacky, but due to how mercurial handles client-server
         # server see all operation on changeset; bookmarks, phases and
         # obsolescence marker in different transaction, we don't want to check
         # locking on those
         obsolete_call = environ['QUERY_STRING'] in ['cmd=listkeys',]
         locked_by = repo.locked
         if repo and repo.enable_locking and not obsolete_call:
             if action == 'push':
                 #check if it's already locked !, if it is compare users
                 user_id, _date = repo.locked
                 if user.user_id == user_id:
                     log.debug('Got push from user %s, now unlocking', user)
                     # unlock if we have push from user who locked
                     make_lock = False
                 else:
                     # we're not the same user who locked, ban with 423 !
                     locked = True
             if action == 'pull':
                 if repo.locked[0] and repo.locked[1]:
                     locked = True
                 else:
                     log.debug('Setting lock on repo %s by %s', repo, user)
                     make_lock = True
         else:
             log.debug('Repository %s do not have locking enabled', repo)
         log.debug('FINAL locking values make_lock:%s,locked:%s,locked_by:%s',
                   make_lock, locked, locked_by)
         return make_lock, locked, locked_by
     def __call__(self, environ, start_response):
         start = time.time()
         try:
             return self._handle_request(environ, start_response)
         finally:
             log = logging.getLogger('kallithea.' + self.__class__.__name__)
             log.debug('Request time: %.3fs', time.time() - start)
             meta.Session.remove()
 class BaseController(WSGIController):
     def __before__(self):
         """
         __before__ is called before controller methods and after __call__
         """
         c.kallithea_version = __version__
         rc_config = Setting.get_app_settings()
         # Visual options
         c.visual = AttributeDict({})
         ## DB stored
         c.visual.show_public_icon = str2bool(rc_config.get('show_public_icon'))
         c.visual.show_private_icon = str2bool(rc_config.get('show_private_icon'))
         c.visual.stylify_metatags = str2bool(rc_config.get('stylify_metatags'))
         c.visual.page_size = safe_int(rc_config.get('dashboard_items', 100))
         c.visual.admin_grid_items = safe_int(rc_config.get('admin_grid_items', 100))
         c.visual.repository_fields = str2bool(rc_config.get('repository_fields'))
         c.visual.show_version = str2bool(rc_config.get('show_version'))
         c.visual.use_gravatar = str2bool(rc_config.get('use_gravatar'))
         c.visual.gravatar_url = rc_config.get('gravatar_url')
         c.ga_code = rc_config.get('ga_code')
         # TODO: replace undocumented backwards compatibility hack with db upgrade and rename ga_code
         if c.ga_code and '<' not in c.ga_code:
             c.ga_code = '''<script type="text/javascript">
                 var _gaq = _gaq || [];
                 _gaq.push(['_setAccount', '%s']);
                 _gaq.push(['_trackPageview']);
                 (function() {
                     var ga = document.createElement('script'); ga.type = 'text/javascript'; ga.async = true;
                     ga.src = ('https:' == document.location.protocol ? 'https://ssl' : 'http://www') + '.google-analytics.com/ga.js';
                     var s = document.getElementsByTagName('script')[0]; s.parentNode.insertBefore(ga, s);
                     })();
             </script>''' % c.ga_code
         c.site_name = rc_config.get('title')
         c.clone_uri_tmpl = rc_config.get('clone_uri_tmpl')
         ## INI stored
         c.visual.allow_repo_location_change = str2bool(config.get('allow_repo_location_change', True))
         c.visual.allow_custom_hooks_settings = str2bool(config.get('allow_custom_hooks_settings', True))
         c.instance_id = config.get('instance_id')
         c.issues_url = config.get('bugtracker', url('issues_url'))
         # END CONFIG VARS
         c.repo_name = get_repo_slug(request)  # can be empty
         c.backends = BACKENDS.keys()
         c.unread_notifications = NotificationModel() \
                         .get_unread_cnt_for_user(c.authuser.user_id)
         self.cut_off_limit = safe_int(config.get('cut_off_limit'))
         c.my_pr_count = PullRequest.query(reviewer_id=c.authuser.user_id, include_closed=False).count()
         self.sa = meta.Session
         self.scm_model = ScmModel(self.sa)
     @staticmethod
     def _determine_auth_user(api_key, session_authuser):
         """
         Create an `AuthUser` object given the API key (if any) and the
         value of the authuser session cookie.
         """
         # Authenticate by API key
         if api_key:
             # when using API_KEY we are sure user exists.
             return AuthUser(dbuser=User.get_by_api_key(api_key),
                             is_external_auth=True)
         if api_key is not None:
             au = AuthUser(dbuser=User.get_by_api_key(api_key),
                 authenticating_api_key=api_key, is_external_auth=True)
             if au.is_anonymous:
                 log.warning('API key ****%s is NOT valid', api_key[-4:])
                 raise webob.exc.HTTPForbidden(_('Invalid API key'))
             return au
         # Authenticate by session cookie
         # In ancient login sessions, 'authuser' may not be a dict.
         # In that case, the user will have to log in again.
         # v0.3 and earlier included an 'is_authenticated' key; if present,
         # this must be True.
         if isinstance(session_authuser, dict) and session_authuser.get('is_authenticated', True):
             try:
                 return AuthUser.from_cookie(session_authuser)
             except UserCreationError as e:
                 # container auth or other auth functions that create users on
                 # the fly can throw UserCreationError to signal issues with
                 # user creation. Explanation should be provided in the
                 # exception object.
                 from kallithea.lib import helpers as h
                 h.flash(e, 'error', logf=log.error)
         # Authenticate by auth_container plugin (if enabled)
         if any(
             auth_modules.importplugin(name).is_container_auth
             for name in Setting.get_auth_plugins()
         ):
             try:
                 user_info = auth_modules.authenticate('', '', request.environ)
             except UserCreationError as e:
                 from kallithea.lib import helpers as h
                 h.flash(e, 'error', logf=log.error)
             else:
                 if user_info is not None:
                     username = user_info['username']
                     user = User.get_by_username(username, case_insensitive=True)
                     return log_in_user(user, remember=False,
                                        is_external_auth=True)
         # User is anonymous
         return AuthUser()
     @staticmethod
     def _basic_security_checks():
         """Perform basic security/sanity checks before processing the request."""
         # Only allow the following HTTP request methods.
         if request.method not in ['GET', 'HEAD', 'POST']:
             raise webob.exc.HTTPMethodNotAllowed()
         # Also verify the _method override - no longer allowed.
         if request.params.get('_method') is None:
             pass # no override, no problem
         else:
             raise webob.exc.HTTPMethodNotAllowed()
         # Make sure CSRF token never appears in the URL. If so, invalidate it.
         if secure_form.token_key in request.GET:
             log.error('CSRF key leak detected')
             session.pop(secure_form.token_key, None)
             session.save()
             from kallithea.lib import helpers as h
             h.flash(_('CSRF token leak has been detected - all form tokens have been expired'),
                     category='error')
         # WebOb already ignores request payload parameters for anything other
         # than POST/PUT, but double-check since other Kallithea code relies on
         # this assumption.
         if request.method not in ['POST', 'PUT'] and request.POST:
             log.error('%r request with payload parameters; WebOb should have stopped this', request.method)
             raise webob.exc.HTTPBadRequest()
     def __call__(self, environ, start_response):
         """Invoke the Controller"""
         # WSGIController.__call__ dispatches to the Controller method
         # the request is routed to. This routing information is
         # available in environ['pylons.routes_dict']
         try:
             self.ip_addr = _get_ip_addr(environ)
             # make sure that we update permissions each time we call controller
             self._basic_security_checks()
             #set globals for auth user
             self.authuser = c.authuser = request.user = self._determine_auth_user(
                 request.GET.get('api_key'),
                 session.get('authuser'),
+            )
             log.info('IP: %s User: %s accessed %s',
                 self.ip_addr, self.authuser,
                 safe_unicode(_get_access_path(environ)),
+            )
             return WSGIController.__call__(self, environ, start_response)
         except webob.exc.HTTPException as e:
             return e(environ, start_response)
         finally:
             meta.Session.remove()
 class BaseRepoController(BaseController):
     """
     Base class for controllers responsible for loading all needed data for
     repository loaded items are
     c.db_repo_scm_instance: instance of scm repository
     c.db_repo: instance of db
     c.repository_followers: number of followers
     c.repository_forks: number of forks
     c.repository_following: weather the current user is following the current repo
     """
     def __before__(self):
         super(BaseRepoController, self).__before__()
         if c.repo_name:  # extracted from routes
             _dbr = Repository.get_by_repo_name(c.repo_name)
             if not _dbr:
                 return
             log.debug('Found repository in database %s with state `%s`',
                       safe_unicode(_dbr), safe_unicode(_dbr.repo_state))
             route = getattr(request.environ.get('routes.route'), 'name', '')
             # allow to delete repos that are somehow damages in filesystem
             if route in ['delete_repo']:
                 return
             if _dbr.repo_state in [Repository.STATE_PENDING]:
                 if route in ['repo_creating_home']:
                     return
                 check_url = url('repo_creating_home', repo_name=c.repo_name)
                 raise webob.exc.HTTPFound(location=check_url)
             dbr = c.db_repo = _dbr
             c.db_repo_scm_instance = c.db_repo.scm_instance
             if c.db_repo_scm_instance is None:
                 log.error('%s this repository is present in database but it '
                           'cannot be created as an scm instance', c.repo_name)
                 from kallithea.lib import helpers as h
                 h.flash(h.literal(_('Repository not found in the filesystem')),
                         category='error')
                 raise paste.httpexceptions.HTTPNotFound()
             # some globals counter for menu
             c.repository_followers = self.scm_model.get_followers(dbr)
             c.repository_forks = self.scm_model.get_forks(dbr)
             c.repository_pull_requests = self.scm_model.get_pull_requests(dbr)
             c.repository_following = self.scm_model.is_following_repo(
                                     c.repo_name, self.authuser.user_id)
     @staticmethod
     def _get_ref_rev(repo, ref_type, ref_name, returnempty=False):
         """
         Safe way to get changeset. If error occurs show error.
         """
         from kallithea.lib import helpers as h
         try:
             return repo.scm_instance.get_ref_revision(ref_type, ref_name)
         except EmptyRepositoryError as e:
             if returnempty:
                 return repo.scm_instance.EMPTY_CHANGESET
             h.flash(h.literal(_('There are no changesets yet')),
                     category='error')
             raise webob.exc.HTTPNotFound()
         except ChangesetDoesNotExistError as e:
             h.flash(h.literal(_('Changeset for %s %s not found in %s') %
                               (ref_type, ref_name, repo.repo_name)),
                     category='error')
             raise webob.exc.HTTPNotFound()
         except RepositoryError as e:
             log.error(traceback.format_exc())
             h.flash(safe_str(e), category='error')
             raise webob.exc.HTTPBadRequest()
 class WSGIResultCloseCallback(object):
     """Wrap a WSGI result and let close call close after calling the
     close method on the result.
     """
     def __init__(self, result, close):
         self._result = result
         self._close = close
     def __iter__(self):
         return iter(self._result)
     def close(self):
         if hasattr(self._result, 'close'):
             self._result.close()
         self._close()

kallithea/tests/functional/test_login.py

➞

Show inline comments

@@ @@ -84,416 +84,416 @@ class TestLoginController(TestController @@
             assert not re.search(r';\s+(Max-Age|Expires)=', cookie, re.IGNORECASE), 'Cookie %r has expiration date, but should be a session cookie' % cookie
     def test_login_remember(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_REGULAR_LOGIN,
                                   'password': TEST_USER_REGULAR_PASS,
                                   'remember': True})
         assert 'Set-Cookie' in response.headers
         for cookie in response.headers.getall('Set-Cookie'):
             assert re.search(r';\s+(Max-Age|Expires)=', cookie, re.IGNORECASE), 'Cookie %r should have expiration date, but is a session cookie' % cookie
     def test_logout(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_REGULAR_LOGIN,
                                   'password': TEST_USER_REGULAR_PASS})
         # Verify that a login session has been established.
         response = self.app.get(url(controller='login', action='index'))
         response = response.follow()
         assert 'authuser' in response.session
         response.click('Log Out')
         # Verify that the login session has been terminated.
         response = self.app.get(url(controller='login', action='index'))
         assert 'authuser' not in response.session
     @parametrize('url_came_from', [
           ('data:text/html,<script>window.alert("xss")</script>',),
           ('mailto:test@example.com',),
           ('file:///etc/passwd',),
           ('ftp://ftp.example.com',),
           ('http://other.example.com/bl%C3%A5b%C3%A6rgr%C3%B8d',),
           ('//evil.example.com/',),
           ('/\r\nX-Header-Injection: boo',),
           ('/invälid_url_bytes',),
           ('non-absolute-path',),
     ])
     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_USER_ADMIN_LOGIN,
                                   'password': TEST_USER_ADMIN_PASS},
                                  status=400)
     def test_login_short_password(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_ADMIN_LOGIN,
                                   'password': 'as'})
         assert 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 username or password')
     # verify that get arguments are correctly passed along login redirection
     @parametrize('args,args_encoded', [
         ({'foo':'one', 'bar':'two'}, (('foo', 'one'), ('bar', 'two'))),
         ({'blue': u'blå'.encode('utf-8'), 'green':u'grøn'},
              (('blue', u'blå'.encode('utf-8')), ('green', u'grøn'.encode('utf-8')))),
     ])
     def test_redirection_to_login_form_preserves_get_args(self, args, args_encoded):
         with fixture.anon_access(False):
             response = self.app.get(url(controller='summary', action='index',
                                         repo_name=HG_REPO,
                                         **args))
             assert response.status == '302 Found'
             came_from = urlparse.parse_qs(urlparse.urlparse(response.location).query)['came_from'][0]
             came_from_qs = urlparse.parse_qsl(urlparse.urlparse(came_from).query)
             for encoded in args_encoded:
                 assert encoded in came_from_qs
     @parametrize('args,args_encoded', [
         ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),
         ({'blue': u'blå', 'green':u'grøn'},
              ('blue=bl%C3%A5', 'green=gr%C3%B8n')),
     ])
     def test_login_form_preserves_get_args(self, args, args_encoded):
         response = self.app.get(url(controller='login', action='index',
                                     came_from=url('/_admin/users', **args)))
         came_from = urlparse.parse_qs(urlparse.urlparse(response.form.action).query)['came_from'][0]
         for encoded in args_encoded:
             assert encoded in came_from
     @parametrize('args,args_encoded', [
         ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),
         ({'blue': u'blå', 'green':u'grøn'},
              ('blue=bl%C3%A5', 'green=gr%C3%B8n')),
     ])
     def test_redirection_after_successful_login_preserves_get_args(self, args, args_encoded):
         response = self.app.post(url(controller='login', action='index',
                                      came_from = url('/_admin/users', **args)),
                                  {'username': TEST_USER_ADMIN_LOGIN,
                                   'password': TEST_USER_ADMIN_PASS})
         assert response.status == '302 Found'
         for encoded in args_encoded:
             assert encoded in response.location
     @parametrize('args,args_encoded', [
         ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),
         ({'blue': u'blå', 'green':u'grøn'},
              ('blue=bl%C3%A5', 'green=gr%C3%B8n')),
     ])
     def test_login_form_after_incorrect_login_preserves_get_args(self, args, args_encoded):
         response = self.app.post(url(controller='login', action='index',
                                      came_from=url('/_admin/users', **args)),
                                  {'username': 'error',
                                   'password': 'test12'})
         response.mustcontain('Invalid username or password')
         came_from = urlparse.parse_qs(urlparse.urlparse(response.form.action).query)['came_from'][0]
         for encoded in args_encoded:
             assert encoded in came_from
     #==========================================================================
     # 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_USER_ADMIN_LOGIN
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': uname,
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': 'goodmail@example.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_USER_ADMIN_EMAIL,
                                              '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_USER_ADMIN_EMAIL.title(),
                                              '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'})
         assert 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 an '
                 'alphanumeric character')
     def test_register_err_case_sensitive(self):
         usr = TEST_USER_ADMIN_LOGIN.title()
         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('password', 'password_confirmation')._messages['password_mismatch']
         response.mustcontain(msg)
     def test_register_ok(self):
         username = 'test_regular4'
         password = 'qweqwe'
         email = 'user4@example.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 overridden
         assert response.status == '302 Found'
         self.checkSessionFlash(response, 'You have successfully registered with Kallithea')
         ret = Session().query(User).filter(User.username == 'test_regular4').one()
         assert ret.username == username
         assert check_password(password, ret.password) == True
         assert ret.email == email
         assert ret.name == name
         assert ret.lastname == lastname
         assert ret.api_key != None
         assert ret.admin == False
     #==========================================================================
     # PASSWORD RESET
     #==========================================================================
     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, }
+        )
         response.mustcontain('An email address must contain a single @')
     def test_forgot_password(self):
         response = self.app.get(url(controller='login',
                                     action='password_reset'))
         assert response.status == '200 OK'
         username = 'test_password_reset_1'
         password = 'qweqwe'
         email = 'username@example.com'
         name = u'passwd'
         lastname = u'reset'
         timestamp = int(time.time())
         new = User()
         new.username = username
         new.password = password
         new.email = email
         new.name = name
         new.lastname = lastname
         new.api_key = generate_api_key()
         Session().add(new)
         Session().commit()
         response = self.app.post(url(controller='login',
                                      action='password_reset'),
                                  {'email': email, })
         self.checkSessionFlash(response, 'A password reset confirmation code has been sent')
         response = response.follow()
         # BAD TOKEN
         token = "bad"
         response = self.app.post(url(controller='login',
                                      action='password_reset_confirmation'),
                                  {'email': email,
                                   'timestamp': timestamp,
                                   'password': "p@ssw0rd",
                                   'password_confirm': "p@ssw0rd",
                                   'token': token,
                                  })
         assert response.status == '200 OK'
         response.mustcontain('Invalid password reset token')
         # GOOD TOKEN
         # TODO: The token should ideally be taken from the mail sent
         # above, instead of being recalculated.
         token = UserModel().get_reset_password_token(
             User.get_by_username(username), timestamp, self.authentication_token())
         response = self.app.get(url(controller='login',
                                     action='password_reset_confirmation',
                                     email=email,
                                     timestamp=timestamp,
                                     token=token))
         assert response.status == '200 OK'
         response.mustcontain("You are about to set a new password for the email address %s" % email)
         response = self.app.post(url(controller='login',
                                      action='password_reset_confirmation'),
                                  {'email': email,
                                   'timestamp': timestamp,
                                   'password': "p@ssw0rd",
                                   'password_confirm': "p@ssw0rd",
                                   'token': token,
                                  })
         assert response.status == '302 Found'
         self.checkSessionFlash(response, 'Successfully updated password')
         response = response.follow()
     #==========================================================================
     # API
     #==========================================================================
     def _get_api_whitelist(self, values=None):
         config = {'api_access_controllers_whitelist': values or []}
         return config
     def _api_key_test(self, api_key, status):
         """Verifies HTTP status code for accessing an auth-requiring page,
         using the given api_key URL parameter. If api_key is None, no api_key
         parameter is passed at all. If api_key is True, a real, working API key
         is used.
         """
         with fixture.anon_access(False):
             if api_key is None:
                 params = {}
             else:
                 if api_key is True:
                     api_key = User.get_first_admin().api_key
                 params = {'api_key': api_key}
             self.app.get(url(controller='changeset', action='changeset_raw',
                              repo_name=HG_REPO, revision='tip', **params),
                          status=status)
     @parametrize('test_name,api_key,code', [
         ('none', None, 302),
         ('empty_string', '', 403),
         ('fake_number', '123456', 403),
         ('proper_api_key', True, 403)
     ])
     def test_access_not_whitelisted_page_via_api_key(self, test_name, api_key, code):
         whitelist = self._get_api_whitelist([])
         with mock.patch('kallithea.CONFIG', whitelist):
             assert [] == whitelist['api_access_controllers_whitelist']
             self._api_key_test(api_key, code)
     @parametrize('test_name,api_key,code', [
         ('none', None, 302),
         ('empty_string', '', 302),
         ('fake_number', '123456', 302),
         ('fake_not_alnum', 'a-z', 302),
         ('fake_api_key', '0123456789abcdef0123456789ABCDEF01234567', 302),
         ('empty_string', '', 403),
         ('fake_number', '123456', 403),
         ('fake_not_alnum', 'a-z', 403),
         ('fake_api_key', '0123456789abcdef0123456789ABCDEF01234567', 403),
         ('proper_api_key', True, 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):
             assert ['ChangesetController:changeset_raw'] == whitelist['api_access_controllers_whitelist']
             self._api_key_test(api_key, code)
     def test_access_page_via_extra_api_key(self):
         whitelist = self._get_api_whitelist(['ChangesetController:changeset_raw'])
         with mock.patch('kallithea.CONFIG', whitelist):
             assert ['ChangesetController:changeset_raw'] == whitelist['api_access_controllers_whitelist']
             new_api_key = ApiKeyModel().create(TEST_USER_ADMIN_LOGIN, u'test')
             Session().commit()
             self._api_key_test(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):
             assert ['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().commit()
-            self._api_key_test(new_api_key.api_key, status=302)
+            self._api_key_test(new_api_key.api_key, status=403)

0 comments (0 inline, 0 general)