@@ -184,210 +184,210 @@ class RepoGroupModel(object):
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 HasUserGroupPermissionLevel
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.is_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):
self.grant_user_group_permission(repo_group=obj,
group_name=users_group,
perm=perm)
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):
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':
# this updates also current one if found
_set_perm_user(obj, user=member, perm=perm)
## set for user group
else:
# check if we have permissions to alter this usergroup's access
if not check_perms or HasUserGroupPermissionLevel('read')(member):
_set_perm_group(obj, users_group=member, perm=perm)
# set new permissions
for member, perm, member_type in perms_new:
updates.append(obj)
# if it's not recursive call for all,repos,groups
# break the loop and don't proceed with other changes
if recursive not in ['all', 'repos', 'groups']:
break
return updates
def update(self, repo_group, kwargs):
def update(self, repo_group, repo_group_args):
try:
repo_group = RepoGroup.guess_instance(repo_group)
old_path = repo_group.full_path
# change properties
if 'group_description' in kwargs:
repo_group.group_description = kwargs['group_description']
if 'parent_group_id' in kwargs:
repo_group.parent_group_id = kwargs['parent_group_id']
if 'enable_locking' in kwargs:
repo_group.enable_locking = kwargs['enable_locking']
if 'group_description' in repo_group_args:
repo_group.group_description = repo_group_args['group_description']
if 'parent_group_id' in repo_group_args:
repo_group.parent_group_id = repo_group_args['parent_group_id']
if 'enable_locking' in repo_group_args:
repo_group.enable_locking = repo_group_args['enable_locking']
assert kwargs['parent_group_id'] != u'-1', kwargs # RepoGroupForm should have converted to None
repo_group.parent_group = RepoGroup.get(kwargs['parent_group_id'])
if 'group_name' in kwargs:
group_name = kwargs['group_name']
assert repo_group_args['parent_group_id'] != u'-1', repo_group_args # RepoGroupForm should have converted to None
repo_group.parent_group = RepoGroup.get(repo_group_args['parent_group_id'])
if 'group_name' in repo_group_args:
group_name = repo_group_args['group_name']
if kallithea.lib.utils2.repo_name_slug(group_name) != group_name:
raise Exception('invalid repo group name %s' % group_name)
repo_group.group_name = repo_group.get_new_name(group_name)
new_path = repo_group.full_path
Session().add(repo_group)
# iterate over all members of this groups and do fixes
# set locking if given
# if obj is a repoGroup also fix the name of the group according
# to the parent
# if obj is a Repo fix it's name
# this can be potentially heavy operation
# set the value from it's parent
obj.enable_locking = repo_group.enable_locking
new_name = obj.get_new_name(obj.name)
log.debug('Fixing group %s to new name %s' \
% (obj.group_name, new_name))
obj.group_name = new_name
# we need to get all repositories from this new group and
# rename them accordingly to new group path
new_name = obj.get_new_name(obj.just_name)
log.debug('Fixing repo %s to new name %s' \
% (obj.repo_name, new_name))
obj.repo_name = new_name
self._rename_group(old_path, new_path)
return repo_group
def delete(self, repo_group, force_delete=False):
Session().delete(repo_group)
self._delete_group(repo_group, force_delete)
log.error('Error removing repo_group %s', repo_group)
def add_permission(self, repo_group, obj, obj_type, perm, recursive):
perm = Permission.guess_instance(perm)
for el in repo_group.recursive_groups_and_repos():
if isinstance(el, RepoGroup) and not el == repo_group:
if isinstance(el, Repository):
el = repo_group
if isinstance(el, RepoGroup):
if obj_type == 'user':
RepoGroupModel().grant_user_permission(el, user=obj, perm=perm)
elif obj_type == 'user_group':
RepoGroupModel().grant_user_group_permission(el, group_name=obj, perm=perm)
raise Exception('undefined object type %s' % obj_type)
elif isinstance(el, Repository):
# for repos we need to hotfix the name of permission
_perm = perm.permission_name.replace('group.', 'repository.')
RepoModel().grant_user_permission(el, user=obj, perm=_perm)
RepoModel().grant_user_group_permission(el, group_name=obj, perm=_perm)
raise Exception('el should be instance of Repository or '
'RepositoryGroup got %s instead' % type(el))
def delete_permission(self, repo_group, obj, obj_type, recursive):
"""
Revokes permission for repo_group for given obj(user or users_group),
obj_type can be user or user group
Status change: