Changeset - 3abfc5aa0dd7
Parent rev.
Child rev.
[Not reviewed]
default
1 1 0
Mads Kiilerich - 7 years ago 2018-12-29 01:48:02
mads@kiilerich.com
auth: drop support for different "algorithms" for computing permission

Drop partially implemented support for different permission "algorithms". Only
"higherwin" is used and makes sense - just hardcode that.
2 files changed with 6 insertions and 18 deletions:
kallithea/lib/auth.py
6
17
kallithea/tests/models/test_user_permissions_on_repos.py
1
0 comments (0 inline, 0 general)
kallithea/lib/auth.py
➞
Show inline comments
 
@@ -112,67 +112,62 @@ 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

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            return bcrypt.checkpw(safe_str(password), safe_str(hashed))

			




	
	
	
		
 
        except ValueError as e:

			




	
	
	
		
 
            # bcrypt will throw ValueError 'Invalid hashed_password salt' on all password errors

			




	
	
	
		
 
            log.error('error from bcrypt checking password: %s', e)

			




	
	
	
		
 
            return False

			




	
	
	
		
 
    else:

			




	
	
	
		
 
        raise Exception('Unknown or unsupported platform %s'

			




	
	
	
		
 
                        % __platform__)

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def _cached_perms_data(user_id, user_is_admin, user_inherit_default_permissions,

			




	
	
	
		
 
                       explicit, algo):

			




	
	
	
		
 
                       explicit):

			




	
	
	
		
 
    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

			




	
	
		
 
@@ -268,50 +263,49 @@ def _cached_perms_data(user_id, user_is_

			




	
	
	
		
 
                                            .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

			




	
	
	
		
 
    # fill in his permission from it.

			




	
	
	
		
 
    #======================================================================

			




	
	
	
		
 

			




	
	
	
		
 
    # 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]

			




	
	
	
		
 

			




	
	
		
 
@@ -322,50 +316,49 @@ def _cached_perms_data(user_id, user_is_

			




	
	
	
		
 
            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

			




	
	
	
		
 
    # fill in his permission from it.

			




	
	
	
		
 
    #======================================================================

			




	
	
	
		
 
    # 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

			




	
	
		
 
@@ -559,73 +552,69 @@ class AuthUser(object):

			




	
	
	
		
 
            'admin': ['group.admin'],

			




	
	
	
		
 
        }[level]

			




	
	
	
		
 
        actual_perm = self.permissions['repositories_groups'].get(repo_group_name)

			




	
	
	
		
 
        ok = actual_perm in required_perms

			




	
	
	
		
 
        log.debug('Checking if user %r can %r repo group %r (%s): %s (has %r)',

			




	
	
	
		
 
            self.username, level, repo_group_name, purpose, ok, actual_perm)

			




	
	
	
		
 
        return ok

			




	
	
	
		
 

			




	
	
	
		
 
    def has_user_group_permission_level(self, user_group_name, level, purpose=None):

			




	
	
	
		
 
        required_perms = {

			




	
	
	
		
 
            'read': ['usergroup.read', 'usergroup.write', 'usergroup.admin'],

			




	
	
	
		
 
            'write': ['usergroup.write', 'usergroup.admin'],

			




	
	
	
		
 
            'admin': ['usergroup.admin'],

			




	
	
	
		
 
        }[level]

			




	
	
	
		
 
        actual_perm = self.permissions['user_groups'].get(user_group_name)

			




	
	
	
		
 
        ok = actual_perm in required_perms

			




	
	
	
		
 
        log.debug('Checking if user %r can %r user group %r (%s): %s (has %r)',

			




	
	
	
		
 
            self.username, level, user_group_name, purpose, ok, actual_perm)

			




	
	
	
		
 
        return ok

			




	
	
	
		
 

			




	
	
	
		
 
    @property

			




	
	
	
		
 
    def api_keys(self):

			




	
	
	
		
 
        return self._get_api_keys()

			




	
	
	
		
 

			




	
	
	
		
 
    def __get_perms(self, user, explicit=True, algo='higherwin', cache=False):

			




	
	
	
		
 
    def __get_perms(self, user, explicit=True, 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

			




	
	
	
		
 
            compute the decision

			




	
	
	
		
 
        """

			




	
	
	
		
 
        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)

			




	
	
	
		
 
                       user_inherit_default_permissions, explicit)

			




	
	
	
		
 

			




	
	
	
		
 
    def _get_api_keys(self):

			




	
	
	
		
 
        api_keys = [self.api_key]

			




	
	
	
		
 
        for api_key in UserApiKeys.query() \

			




	
	
	
		
 
                .filter_by(user_id=self.user_id, is_expired=False):

			




	
	
	
		
 
            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):

			




	
	
	
		
 
        """

			




        

    


    
    

    

        

                

                    kallithea/tests/models/test_user_permissions_on_repos.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
		
 
deleted file

			




        

    



    


    



    



      

      





    
    0 comments (0 inline, 0 general)
    





    


  

  







  


    




    








    
        
    
    
        This site is powered by
            Kallithea 0.7.0,
        which is
        © 2010–2025 by various authors & licensed under GPLv3.