# Copyright (C) 2011-2024 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 . # # 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