Changeset - 3505a6be2988
[Not reviewed]
default
0 2 0
Søren Løvborg - 9 years ago 2017-02-23 20:26:27
sorenl@unity3d.com
repogroups: fix private repo recursion check

The purpose of this check is to ensure that we don't recursively assign
"default" user perms for a repo with the "private" flag set (because in
that case, the "default" user perms should always be "no access").

(The check, and this fix, is of course only applicable to Kallithea
instances that have anonymous access enabled to begin with.)

However, the check was only functional if the user was specified as a
username. This is apparently always the case when Kallithea is running,
but was not e.g. the case in the test suite, which consistently passed
a user ID instead of a username.

This commit ensures that the user is always resolved before the check is
made. There's no significant overhead to this, as the code immediately
calls RepoModel().grant_user_permission, which resolved the user anyway.
This change just moves the database lookup a bit earlier.

Fixing this revealed the matching test case to be broken, so it has been
fixed as well.

Down the road, we should eliminate Kallithea's bizarre practice of
passing around usernames and user IDs, in favor of passing actual User
objects. That'll get rid of mistakes like these, as well as repeated
needless database lookups.
2 files changed with 7 insertions and 3 deletions:
0 comments (0 inline, 0 general)
kallithea/model/repo_group.py
Show inline comments
 
@@ -153,99 +153,101 @@ class RepoGroupModel(BaseModel):
 

	
 
            if parent_group and copy_permissions:
 
                # copy permissions from parent
 
                user_perms = UserRepoGroupToPerm.query() \
 
                    .filter(UserRepoGroupToPerm.group == parent_group).all()
 

	
 
                group_perms = UserGroupRepoGroupToPerm.query() \
 
                    .filter(UserGroupRepoGroupToPerm.group == parent_group).all()
 

	
 
                for perm in user_perms:
 
                    # don't copy over the permission for user who is creating
 
                    # this group, if he is not super admin he get's admin
 
                    # permission set above
 
                    if perm.user != owner or owner.is_admin:
 
                        UserRepoGroupToPerm.create(perm.user, new_repo_group, perm.permission)
 

	
 
                for perm in group_perms:
 
                    UserGroupRepoGroupToPerm.create(perm.users_group, new_repo_group, perm.permission)
 
            else:
 
                perm_obj = self._create_default_perms(new_repo_group)
 
                self.sa.add(perm_obj)
 

	
 
            if not just_db:
 
                # we need to flush here, in order to check if database won't
 
                # throw any exceptions, create filesystem dirs at the very end
 
                self.sa.flush()
 
                self._create_group(new_repo_group.group_name)
 

	
 
            return new_repo_group
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise
 

	
 
    def _update_permissions(self, repo_group, perms_new=None,
 
                            perms_updates=None, recursive=None,
 
                            check_perms=True):
 
        from kallithea.model.repo import RepoModel
 
        from kallithea.lib.auth import HasUserGroupPermissionAny
 

	
 
        if not perms_new:
 
            perms_new = []
 
        if not perms_updates:
 
            perms_updates = []
 

	
 
        def _set_perm_user(obj, user, perm):
 
            if isinstance(obj, RepoGroup):
 
                self.grant_user_permission(repo_group=obj, user=user, perm=perm)
 
            elif isinstance(obj, Repository):
 
                user = User.guess_instance(user)
 

	
 
                # private repos will not allow to change the default permissions
 
                # using recursive mode
 
                if obj.private and user == User.DEFAULT_USER:
 
                if obj.private and user.username == User.DEFAULT_USER:
 
                    return
 

	
 
                # we set group permission but we have to switch to repo
 
                # permission
 
                perm = perm.replace('group.', 'repository.')
 
                RepoModel().grant_user_permission(
 
                    repo=obj, user=user, perm=perm
 
                )
 

	
 
        def _set_perm_group(obj, users_group, perm):
 
            if isinstance(obj, RepoGroup):
 
                self.grant_user_group_permission(repo_group=obj,
 
                                                  group_name=users_group,
 
                                                  perm=perm)
 
            elif isinstance(obj, Repository):
 
                # we set group permission but we have to switch to repo
 
                # permission
 
                perm = perm.replace('group.', 'repository.')
 
                RepoModel().grant_user_group_permission(
 
                    repo=obj, group_name=users_group, perm=perm
 
                )
 

	
 
        # start updates
 
        updates = []
 
        log.debug('Now updating permissions for %s in recursive mode:%s',
 
                  repo_group, recursive)
 

	
 
        for obj in repo_group.recursive_groups_and_repos():
 
            # iterated obj is an instance of a repos group or repository in
 
            # that group, recursive option can be: none, repos, groups, all
 
            if recursive == 'all':
 
                pass
 
            elif recursive == 'repos':
 
                # skip groups, other than this one
 
                if isinstance(obj, RepoGroup) and not obj == repo_group:
 
                    continue
 
            elif recursive == 'groups':
 
                # skip repos
 
                if isinstance(obj, Repository):
 
                    continue
 
            else:  # recursive == 'none': # DEFAULT don't apply to iterated objects
 
                obj = repo_group
 
                # also we do a break at the end of this loop.
 

	
 
            # update permissions
 
            for member, perm, member_type in perms_updates:
 
                ## set for user
 
                if member_type == 'user':
kallithea/tests/models/test_user_permissions_on_repo_groups.py
Show inline comments
 
import functools
 

	
 
from kallithea.model.repo_group import RepoGroupModel
 
from kallithea.model.db import RepoGroup, User
 
from kallithea.model.db import RepoGroup, Repository, User
 

	
 
from kallithea.model.meta import Session
 
from kallithea.tests.models.common import _create_project_tree, check_tree_perms, \
 
    _get_perms, _check_expected_count, expected_count, _destroy_project_tree
 

	
 

	
 
test_u1_id = None
 
_get_repo_perms = None
 
_get_group_perms = None
 

	
 

	
 
def permissions_setup_func(group_name=u'g0', perm='group.read', recursive='all',
 
                           user_id=None):
 
    """
 
    Resets all permissions to perm attribute
 
    """
 
    if not user_id:
 
        user_id = test_u1_id
 
        permissions_setup_func(group_name, perm, recursive,
 
                               user_id=User.get_default_user().user_id)
 

	
 
    repo_group = RepoGroup.get_by_group_name(group_name=group_name)
 
    if not repo_group:
 
        raise Exception('Cannot get group %s' % group_name)
 

	
 
    # Start with a baseline that current group can read recursive
 
    perms_updates = [[user_id, 'group.read', 'user']]
 
    RepoGroupModel()._update_permissions(repo_group,
 
                                         perms_updates=perms_updates,
 
                                         recursive='all', check_perms=False)
 

	
 
    perms_updates = [[user_id, perm, 'user']]
 
    RepoGroupModel()._update_permissions(repo_group,
 
                                         perms_updates=perms_updates,
 
                                         recursive=recursive, check_perms=False)
 
    Session().commit()
 

	
 

	
 
def setup_module():
 
    global test_u1_id, _get_repo_perms, _get_group_perms
 
    test_u1 = _create_project_tree()
 
    Session().commit()
 
    test_u1_id = test_u1.user_id
 
    _get_repo_perms = functools.partial(_get_perms, key='repositories',
 
                                        test_u1_id=test_u1_id)
 
    _get_group_perms = functools.partial(_get_perms, key='repositories_groups',
 
                                         test_u1_id=test_u1_id)
 

	
 
@@ -88,97 +88,99 @@ def test_user_permissions_on_group_witho
 

	
 
    items = [x for x in _get_group_perms(group, recursive)]
 
    expected = 1
 
    assert len(items) == expected, ' %s != %s' % (len(items), expected)
 
    for name, perm in items:
 
        check_tree_perms(name, perm, group, 'group.write')
 

	
 

	
 
def test_user_permissions_on_group_with_recursive_mode():
 

	
 
    # set permission to g0 recursive mode, all children including
 
    # other repos and groups should have this permission now set !
 
    recursive = 'all'
 
    group = u'g0'
 
    permissions_setup_func(group, 'group.write', recursive=recursive)
 

	
 
    repo_items = [x for x in _get_repo_perms(group, recursive)]
 
    items = [x for x in _get_group_perms(group, recursive)]
 
    _check_expected_count(items, repo_items, expected_count(group, True))
 

	
 
    for name, perm in repo_items:
 
        check_tree_perms(name, perm, group, 'repository.write')
 

	
 
    for name, perm in items:
 
        check_tree_perms(name, perm, group, 'group.write')
 

	
 

	
 
def test_user_permissions_on_group_with_recursive_mode_for_default_user():
 

	
 
    # set permission to g0 recursive mode, all children including
 
    # other repos and groups should have this permission now set !
 
    recursive = 'all'
 
    group = u'g0'
 
    default_user_id = User.get_default_user().user_id
 
    permissions_setup_func(group, 'group.write', recursive=recursive,
 
                           user_id=default_user_id)
 

	
 
    # change default to get perms for default user
 
    _get_repo_perms = functools.partial(_get_perms, key='repositories',
 
                                        test_u1_id=default_user_id)
 
    _get_group_perms = functools.partial(_get_perms, key='repositories_groups',
 
                                         test_u1_id=default_user_id)
 

	
 
    repo_items = [x for x in _get_repo_perms(group, recursive)]
 
    items = [x for x in _get_group_perms(group, recursive)]
 
    _check_expected_count(items, repo_items, expected_count(group, True))
 

	
 
    for name, perm in repo_items:
 
        check_tree_perms(name, perm, group, 'repository.write')
 
        # default user permissions do not "recurse into" private repos
 
        is_private = Repository.get_by_repo_name(name).private
 
        check_tree_perms(name, perm, group, 'repository.none' if is_private else 'repository.write')
 

	
 
    for name, perm in items:
 
        check_tree_perms(name, perm, group, 'group.write')
 

	
 

	
 
def test_user_permissions_on_group_with_recursive_mode_inner_group():
 
    ## set permission to g0_3 group to none
 
    recursive = 'all'
 
    group = u'g0/g0_3'
 
    permissions_setup_func(group, 'group.none', recursive=recursive)
 

	
 
    repo_items = [x for x in _get_repo_perms(group, recursive)]
 
    items = [x for x in _get_group_perms(group, recursive)]
 
    _check_expected_count(items, repo_items, expected_count(group, True))
 

	
 
    for name, perm in repo_items:
 
        check_tree_perms(name, perm, group, 'repository.none')
 

	
 
    for name, perm in items:
 
        check_tree_perms(name, perm, group, 'group.none')
 

	
 

	
 
def test_user_permissions_on_group_with_recursive_mode_deepest():
 
    ## set permission to g0_3 group to none
 
    recursive = 'all'
 
    group = u'g0/g0_1/g0_1_1'
 
    permissions_setup_func(group, 'group.write', recursive=recursive)
 

	
 
    repo_items = [x for x in _get_repo_perms(group, recursive)]
 
    items = [x for x in _get_group_perms(group, recursive)]
 
    _check_expected_count(items, repo_items, expected_count(group, True))
 

	
 
    for name, perm in repo_items:
 
        check_tree_perms(name, perm, group, 'repository.write')
 

	
 
    for name, perm in items:
 
        check_tree_perms(name, perm, group, 'group.write')
 

	
 

	
 
def test_user_permissions_on_group_with_recursive_mode_only_with_repos():
 
    ## set permission to g0_3 group to none
 
    recursive = 'all'
 
    group = u'g0/g0_2'
 
    permissions_setup_func(group, 'group.admin', recursive=recursive)
 

	
 
    repo_items = [x for x in _get_repo_perms(group, recursive)]
 
    items = [x for x in _get_group_perms(group, recursive)]
 
    _check_expected_count(items, repo_items, expected_count(group, True))
0 comments (0 inline, 0 general)