##// END OF EJS Templates
fix(deps): bring back pyramid-mako removed in debugtoolbar
fix(deps): bring back pyramid-mako removed in debugtoolbar

File last commit:

r5095:aa627a5f default
r5517:6d578f95 default
Show More
user_group.py
752 lines | 28.9 KiB | text/x-python | PythonLexer
# Copyright (C) 2011-2023 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# 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 Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
import logging
import traceback
from rhodecode.lib.utils2 import safe_str
from rhodecode.lib.exceptions import (
UserGroupAssignedException, RepoGroupAssignmentError)
from rhodecode.lib.utils2 import (
get_current_rhodecode_user, action_logger_generic)
from rhodecode.model import BaseModel
from rhodecode.model.scm import UserGroupList
from rhodecode.model.db import (
joinedload, true, func, User, UserGroupMember, UserGroup,
UserGroupRepoToPerm, Permission, UserGroupToPerm, UserUserGroupToPerm,
UserGroupUserGroupToPerm, UserGroupRepoGroupToPerm)
log = logging.getLogger(__name__)
class UserGroupModel(BaseModel):
cls = UserGroup
def _get_user_group(self, user_group):
return self._get_instance(UserGroup, user_group,
callback=UserGroup.get_by_group_name)
def _create_default_perms(self, user_group):
# create default permission
default_perm = 'usergroup.read'
def_user = User.get_default_user()
for p in def_user.user_perms:
if p.permission.permission_name.startswith('usergroup.'):
default_perm = p.permission.permission_name
break
user_group_to_perm = UserUserGroupToPerm()
user_group_to_perm.permission = Permission.get_by_key(default_perm)
user_group_to_perm.user_group = user_group
user_group_to_perm.user = def_user
return user_group_to_perm
def update_permissions(
self, user_group, perm_additions=None, perm_updates=None,
perm_deletions=None, check_perms=True, cur_user=None):
from rhodecode.lib.auth import HasUserGroupPermissionAny
if not perm_additions:
perm_additions = []
if not perm_updates:
perm_updates = []
if not perm_deletions:
perm_deletions = []
req_perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin')
changes = {
'added': [],
'updated': [],
'deleted': []
}
change_obj = user_group.get_api_data()
# update permissions
for member_id, perm, member_type in perm_updates:
member_id = int(member_id)
if member_type == 'user':
member_name = User.get(member_id).username
# this updates existing one
self.grant_user_permission(
user_group=user_group, user=member_id, perm=perm
)
elif member_type == 'user_group':
# check if we have permissions to alter this usergroup
member_name = UserGroup.get(member_id).users_group_name
if not check_perms or HasUserGroupPermissionAny(
*req_perms)(member_name, user=cur_user):
self.grant_user_group_permission(
target_user_group=user_group, user_group=member_id, perm=perm)
else:
raise ValueError("member_type must be 'user' or 'user_group' "
"got {} instead".format(member_type))
changes['updated'].append({
'change_obj': change_obj,
'type': member_type, 'id': member_id,
'name': member_name, 'new_perm': perm})
# set new permissions
for member_id, perm, member_type in perm_additions:
member_id = int(member_id)
if member_type == 'user':
member_name = User.get(member_id).username
self.grant_user_permission(
user_group=user_group, user=member_id, perm=perm)
elif member_type == 'user_group':
# check if we have permissions to alter this usergroup
member_name = UserGroup.get(member_id).users_group_name
if not check_perms or HasUserGroupPermissionAny(
*req_perms)(member_name, user=cur_user):
self.grant_user_group_permission(
target_user_group=user_group, user_group=member_id, perm=perm)
else:
raise ValueError("member_type must be 'user' or 'user_group' "
"got {} instead".format(member_type))
changes['added'].append({
'change_obj': change_obj,
'type': member_type, 'id': member_id,
'name': member_name, 'new_perm': perm})
# delete permissions
for member_id, perm, member_type in perm_deletions:
member_id = int(member_id)
if member_type == 'user':
member_name = User.get(member_id).username
self.revoke_user_permission(user_group=user_group, user=member_id)
elif member_type == 'user_group':
# check if we have permissions to alter this usergroup
member_name = UserGroup.get(member_id).users_group_name
if not check_perms or HasUserGroupPermissionAny(
*req_perms)(member_name, user=cur_user):
self.revoke_user_group_permission(
target_user_group=user_group, user_group=member_id)
else:
raise ValueError("member_type must be 'user' or 'user_group' "
"got {} instead".format(member_type))
changes['deleted'].append({
'change_obj': change_obj,
'type': member_type, 'id': member_id,
'name': member_name, 'new_perm': perm})
return changes
def get(self, user_group_id, cache=False):
return UserGroup.get(user_group_id)
def get_group(self, user_group):
return self._get_user_group(user_group)
def get_by_name(self, name, cache=False, case_insensitive=False):
return UserGroup.get_by_group_name(name, cache, case_insensitive)
def create(self, name, description, owner, active=True, group_data=None):
try:
new_user_group = UserGroup()
new_user_group.user = self._get_user(owner)
new_user_group.users_group_name = name
new_user_group.user_group_description = description
new_user_group.users_group_active = active
if group_data:
new_user_group.group_data = group_data
self.sa.add(new_user_group)
perm_obj = self._create_default_perms(new_user_group)
self.sa.add(perm_obj)
self.grant_user_permission(user_group=new_user_group,
user=owner, perm='usergroup.admin')
return new_user_group
except Exception:
log.error(traceback.format_exc())
raise
def _get_memberships_for_user_ids(self, user_group, user_id_list):
members = []
for user_id in user_id_list:
member = self._get_membership(user_group.users_group_id, user_id)
members.append(member)
return members
def _get_added_and_removed_user_ids(self, user_group, user_id_list):
current_members = user_group.members or []
current_members_ids = [m.user.user_id for m in current_members]
added_members = [
user_id for user_id in user_id_list
if user_id not in current_members_ids]
if user_id_list == []:
# all members were deleted
deleted_members = current_members_ids
else:
deleted_members = [
user_id for user_id in current_members_ids
if user_id not in user_id_list]
return added_members, deleted_members
def _set_users_as_members(self, user_group, user_ids):
user_group.members = []
self.sa.flush()
members = self._get_memberships_for_user_ids(
user_group, user_ids)
user_group.members = members
self.sa.add(user_group)
def _update_members_from_user_ids(self, user_group, user_ids):
added, removed = self._get_added_and_removed_user_ids(
user_group, user_ids)
self._set_users_as_members(user_group, user_ids)
self._log_user_changes('added to', user_group, added)
self._log_user_changes('removed from', user_group, removed)
return added, removed
def _clean_members_data(self, members_data):
if not members_data:
members_data = []
members = []
for user in members_data:
uid = int(user['member_user_id'])
if uid not in members and user['type'] in ['new', 'existing']:
members.append(uid)
return members
def update(self, user_group, form_data, group_data=None):
user_group = self._get_user_group(user_group)
if 'users_group_name' in form_data:
user_group.users_group_name = form_data['users_group_name']
if 'users_group_active' in form_data:
user_group.users_group_active = form_data['users_group_active']
if 'user_group_description' in form_data:
user_group.user_group_description = form_data[
'user_group_description']
# handle owner change
if 'user' in form_data:
owner = form_data['user']
if isinstance(owner, str):
owner = User.get_by_username(form_data['user'])
if not isinstance(owner, User):
raise ValueError(
'invalid owner for user group: %s' % form_data['user'])
user_group.user = owner
added_user_ids = []
removed_user_ids = []
if 'users_group_members' in form_data:
members_id_list = self._clean_members_data(
form_data['users_group_members'])
added_user_ids, removed_user_ids = \
self._update_members_from_user_ids(user_group, members_id_list)
if group_data:
new_group_data = {}
new_group_data.update(group_data)
user_group.group_data = new_group_data
self.sa.add(user_group)
return user_group, added_user_ids, removed_user_ids
def delete(self, user_group, force=False):
"""
Deletes repository group, unless force flag is used
raises exception if there are members in that group, else deletes
group and users
:param user_group:
:param force:
"""
user_group = self._get_user_group(user_group)
if not user_group:
return
try:
# check if this group is not assigned to repo
assigned_to_repo = [x.repository for x in UserGroupRepoToPerm.query()\
.filter(UserGroupRepoToPerm.users_group == user_group).all()]
# check if this group is not assigned to repo
assigned_to_repo_group = [x.group for x in UserGroupRepoGroupToPerm.query()\
.filter(UserGroupRepoGroupToPerm.users_group == user_group).all()]
if (assigned_to_repo or assigned_to_repo_group) and not force:
assigned = ','.join(map(safe_str,
assigned_to_repo+assigned_to_repo_group))
raise UserGroupAssignedException(
'UserGroup assigned to {}'.format(assigned))
self.sa.delete(user_group)
except Exception:
log.error(traceback.format_exc())
raise
def _log_user_changes(self, action, user_group, user_or_users):
users = user_or_users
if not isinstance(users, (list, tuple)):
users = [users]
group_name = user_group.users_group_name
for user_or_user_id in users:
user = self._get_user(user_or_user_id)
log_text = 'User {user} {action} {group}'.format(
action=action, user=user.username, group=group_name)
action_logger_generic(log_text)
def _find_user_in_group(self, user, user_group):
user_group_member = None
for m in user_group.members:
if m.user_id == user.user_id:
# Found this user's membership row
user_group_member = m
break
return user_group_member
def _get_membership(self, user_group_id, user_id):
user_group_member = UserGroupMember(user_group_id, user_id)
return user_group_member
def add_user_to_group(self, user_group, user):
user_group = self._get_user_group(user_group)
user = self._get_user(user)
user_member = self._find_user_in_group(user, user_group)
if user_member:
# user already in the group, skip
return True
member = self._get_membership(
user_group.users_group_id, user.user_id)
user_group.members.append(member)
try:
self.sa.add(member)
except Exception:
# what could go wrong here?
log.error(traceback.format_exc())
raise
self._log_user_changes('added to', user_group, user)
return member
def remove_user_from_group(self, user_group, user):
user_group = self._get_user_group(user_group)
user = self._get_user(user)
user_group_member = self._find_user_in_group(user, user_group)
if not user_group_member:
# User isn't in that group
return False
try:
self.sa.delete(user_group_member)
except Exception:
log.error(traceback.format_exc())
raise
self._log_user_changes('removed from', user_group, user)
return True
def has_perm(self, user_group, perm):
user_group = self._get_user_group(user_group)
perm = self._get_perm(perm)
return UserGroupToPerm.query()\
.filter(UserGroupToPerm.users_group == user_group)\
.filter(UserGroupToPerm.permission == perm).scalar() is not None
def grant_perm(self, user_group, perm):
user_group = self._get_user_group(user_group)
perm = self._get_perm(perm)
# if this permission is already granted skip it
_perm = UserGroupToPerm.query()\
.filter(UserGroupToPerm.users_group == user_group)\
.filter(UserGroupToPerm.permission == perm)\
.scalar()
if _perm:
return
new = UserGroupToPerm()
new.users_group = user_group
new.permission = perm
self.sa.add(new)
return new
def revoke_perm(self, user_group, perm):
user_group = self._get_user_group(user_group)
perm = self._get_perm(perm)
obj = UserGroupToPerm.query()\
.filter(UserGroupToPerm.users_group == user_group)\
.filter(UserGroupToPerm.permission == perm).scalar()
if obj:
self.sa.delete(obj)
def grant_user_permission(self, user_group, user, perm):
"""
Grant permission for user on given user group, or update
existing one if found
:param user_group: Instance of UserGroup, users_group_id,
or users_group_name
:param user: Instance of User, user_id or username
:param perm: Instance of Permission, or permission_name
"""
changes = {
'added': [],
'updated': [],
'deleted': []
}
user_group = self._get_user_group(user_group)
user = self._get_user(user)
permission = self._get_perm(perm)
perm_name = permission.permission_name
member_id = user.user_id
member_name = user.username
# check if we have that permission already
obj = self.sa.query(UserUserGroupToPerm)\
.filter(UserUserGroupToPerm.user == user)\
.filter(UserUserGroupToPerm.user_group == user_group)\
.scalar()
if obj is None:
# create new !
obj = UserUserGroupToPerm()
obj.user_group = user_group
obj.user = user
obj.permission = permission
self.sa.add(obj)
log.debug('Granted perm %s to %s on %s', perm, user, user_group)
action_logger_generic(
'granted permission: {} to user: {} on usergroup: {}'.format(
perm, user, user_group), namespace='security.usergroup')
changes['added'].append({
'change_obj': user_group.get_api_data(),
'type': 'user', 'id': member_id,
'name': member_name, 'new_perm': perm_name})
return changes
def revoke_user_permission(self, user_group, user):
"""
Revoke permission for user on given user group
:param user_group: Instance of UserGroup, users_group_id,
or users_group name
:param user: Instance of User, user_id or username
"""
changes = {
'added': [],
'updated': [],
'deleted': []
}
user_group = self._get_user_group(user_group)
user = self._get_user(user)
perm_name = 'usergroup.none'
member_id = user.user_id
member_name = user.username
obj = self.sa.query(UserUserGroupToPerm)\
.filter(UserUserGroupToPerm.user == user)\
.filter(UserUserGroupToPerm.user_group == user_group)\
.scalar()
if obj:
self.sa.delete(obj)
log.debug('Revoked perm on %s on %s', user_group, user)
action_logger_generic(
'revoked permission from user: {} on usergroup: {}'.format(
user, user_group), namespace='security.usergroup')
changes['deleted'].append({
'change_obj': user_group.get_api_data(),
'type': 'user', 'id': member_id,
'name': member_name, 'new_perm': perm_name})
return changes
def grant_user_group_permission(self, target_user_group, user_group, perm):
"""
Grant user group permission for given target_user_group
:param target_user_group:
:param user_group:
:param perm:
"""
changes = {
'added': [],
'updated': [],
'deleted': []
}
target_user_group = self._get_user_group(target_user_group)
user_group = self._get_user_group(user_group)
permission = self._get_perm(perm)
perm_name = permission.permission_name
member_id = user_group.users_group_id
member_name = user_group.users_group_name
# forbid assigning same user group to itself
if target_user_group == user_group:
raise RepoGroupAssignmentError('target repo:%s cannot be '
'assigned to itself' % target_user_group)
# check if we have that permission already
obj = self.sa.query(UserGroupUserGroupToPerm)\
.filter(UserGroupUserGroupToPerm.target_user_group == target_user_group)\
.filter(UserGroupUserGroupToPerm.user_group == user_group)\
.scalar()
if obj is None:
# create new !
obj = UserGroupUserGroupToPerm()
obj.user_group = user_group
obj.target_user_group = target_user_group
obj.permission = permission
self.sa.add(obj)
log.debug(
'Granted perm %s to %s on %s', perm, target_user_group, user_group)
action_logger_generic(
'granted permission: {} to usergroup: {} on usergroup: {}'.format(
perm, user_group, target_user_group),
namespace='security.usergroup')
changes['added'].append({
'change_obj': target_user_group.get_api_data(),
'type': 'user_group', 'id': member_id,
'name': member_name, 'new_perm': perm_name})
return changes
def revoke_user_group_permission(self, target_user_group, user_group):
"""
Revoke user group permission for given target_user_group
:param target_user_group:
:param user_group:
"""
changes = {
'added': [],
'updated': [],
'deleted': []
}
target_user_group = self._get_user_group(target_user_group)
user_group = self._get_user_group(user_group)
perm_name = 'usergroup.none'
member_id = user_group.users_group_id
member_name = user_group.users_group_name
obj = self.sa.query(UserGroupUserGroupToPerm)\
.filter(UserGroupUserGroupToPerm.target_user_group == target_user_group)\
.filter(UserGroupUserGroupToPerm.user_group == user_group)\
.scalar()
if obj:
self.sa.delete(obj)
log.debug(
'Revoked perm on %s on %s', target_user_group, user_group)
action_logger_generic(
'revoked permission from usergroup: {} on usergroup: {}'.format(
user_group, target_user_group),
namespace='security.repogroup')
changes['deleted'].append({
'change_obj': target_user_group.get_api_data(),
'type': 'user_group', 'id': member_id,
'name': member_name, 'new_perm': perm_name})
return changes
def get_perms_summary(self, user_group_id):
permissions = {
'repositories': {},
'repositories_groups': {},
}
ugroup_repo_perms = UserGroupRepoToPerm.query()\
.options(joinedload(UserGroupRepoToPerm.permission))\
.options(joinedload(UserGroupRepoToPerm.repository))\
.filter(UserGroupRepoToPerm.users_group_id == user_group_id)\
.all()
for gr in ugroup_repo_perms:
permissions['repositories'][gr.repository.repo_name] \
= gr.permission.permission_name
ugroup_group_perms = UserGroupRepoGroupToPerm.query()\
.options(joinedload(UserGroupRepoGroupToPerm.permission))\
.options(joinedload(UserGroupRepoGroupToPerm.group))\
.filter(UserGroupRepoGroupToPerm.users_group_id == user_group_id)\
.all()
for gr in ugroup_group_perms:
permissions['repositories_groups'][gr.group.group_name] \
= gr.permission.permission_name
return permissions
def enforce_groups(self, user, groups, extern_type=None):
user = self._get_user(user)
current_groups = user.group_member
# find the external created groups, i.e automatically created
log.debug('Enforcing user group set `%s` on user %s', groups, user)
# calculate from what groups user should be removed
# external_groups that are not in groups
for gr in [x.users_group for x in current_groups]:
managed = gr.group_data.get('extern_type')
if managed:
if gr.users_group_name not in groups:
log.debug('Removing user %s from user group %s. '
'Group sync managed by: %s', user, gr, managed)
self.remove_user_from_group(gr, user)
else:
log.debug('Skipping removal from group %s since it is '
'not set to be automatically synchronized', gr)
# now we calculate in which groups user should be == groups params
owner = User.get_first_super_admin().username
for gr in set(groups):
existing_group = UserGroup.get_by_group_name(gr)
if not existing_group:
desc = 'Automatically created from plugin:%s' % extern_type
# we use first admin account to set the owner of the group
existing_group = UserGroupModel().create(
gr, desc, owner, group_data={'extern_type': extern_type})
# we can only add users to groups which have set sync flag via
# extern_type attribute.
# This is either set and created via plugins, or manually
managed = existing_group.group_data.get('extern_type')
if managed:
log.debug('Adding user %s to user group %s', user, gr)
UserGroupModel().add_user_to_group(existing_group, user)
else:
log.debug('Skipping addition to group %s since it is '
'not set to be automatically synchronized', gr)
def change_groups(self, user, groups):
"""
This method changes user group assignment
:param user: User
:param groups: array of UserGroupModel
"""
user = self._get_user(user)
log.debug('Changing user(%s) assignment to groups(%s)', user, groups)
current_groups = user.group_member
current_groups = [x.users_group for x in current_groups]
# calculate from what groups user should be removed/add
groups = set(groups)
current_groups = set(current_groups)
groups_to_remove = current_groups - groups
groups_to_add = groups - current_groups
removed_from_groups = []
added_to_groups = []
for gr in groups_to_remove:
log.debug('Removing user %s from user group %s',
user.username, gr.users_group_name)
removed_from_groups.append(gr.users_group_id)
self.remove_user_from_group(gr.users_group_name, user.username)
for gr in groups_to_add:
log.debug('Adding user %s to user group %s',
user.username, gr.users_group_name)
added_to_groups.append(gr.users_group_id)
UserGroupModel().add_user_to_group(
gr.users_group_name, user.username)
return added_to_groups, removed_from_groups
def _serialize_user_group(self, user_group):
import rhodecode.lib.helpers as h
return {
'id': user_group.users_group_id,
# TODO: marcink figure out a way to generate the url for the
# icon
'icon_link': '',
'value_display': 'Group: %s (%d members)' % (
user_group.users_group_name, len(user_group.members),),
'value': user_group.users_group_name,
'description': user_group.user_group_description,
'owner': user_group.user.username,
'owner_icon': h.gravatar_url(user_group.user.email, 30),
'value_display_owner': h.person(user_group.user.email),
'value_type': 'user_group',
'active': user_group.users_group_active,
}
def get_user_groups(self, name_contains=None, limit=20, only_active=True,
expand_groups=False):
query = self.sa.query(UserGroup)
if only_active:
query = query.filter(UserGroup.users_group_active == true())
if name_contains:
ilike_expression = f'%{safe_str(name_contains)}%'
query = query.filter(
UserGroup.users_group_name.ilike(ilike_expression))\
.order_by(func.length(UserGroup.users_group_name))\
.order_by(UserGroup.users_group_name)
query = query.limit(limit)
user_groups = query.all()
perm_set = ['usergroup.read', 'usergroup.write', 'usergroup.admin']
user_groups = UserGroupList(user_groups, perm_set=perm_set)
# store same serialize method to extract data from User
from rhodecode.model.user import UserModel
serialize_user = UserModel()._serialize_user
_groups = []
for group in user_groups:
entry = self._serialize_user_group(group)
if expand_groups:
expanded_members = []
for member in group.members:
expanded_members.append(serialize_user(member.user))
entry['members'] = expanded_members
_groups.append(entry)
return _groups
@staticmethod
def get_user_groups_as_dict(user_group):
import rhodecode.lib.helpers as h
data = {
'users_group_id': user_group.users_group_id,
'group_name': h.link_to_group(user_group.users_group_name),
'group_description': user_group.user_group_description,
'active': user_group.users_group_active,
"owner": user_group.user.username,
'owner_icon': h.gravatar_url(user_group.user.email, 30),
"owner_data": {
'owner': user_group.user.username,
'owner_icon': h.gravatar_url(user_group.user.email, 30)}
}
return data