users_group.py
295 lines
| 10.3 KiB
| text/x-python
|
PythonLexer
|
r1586 | # -*- coding: utf-8 -*- | ||
""" | ||||
rhodecode.model.users_group | ||||
~~~~~~~~~~~~~~~~~~~~~~~~~~~ | ||||
|
r3410 | user group model for RhodeCode | ||
|
r1586 | |||
:created_on: Oct 1, 2011 | ||||
:author: nvinot | ||||
:copyright: (C) 2011-2011 Nicolas Vinot <aeris@imirhil.fr> | ||||
r1824 | :copyright: (C) 2010-2012 Marcin Kuzminski <marcin@python-works.com> | |||
|
r1586 | :license: GPLv3, see COPYING for more details. | ||
""" | ||||
# This program is free software: you can redistribute it and/or modify | ||||
# it under the terms of the GNU General Public License as published by | ||||
# the Free Software Foundation, either version 3 of the License, or | ||||
# (at your option) any later version. | ||||
# | ||||
# 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 General Public License | ||||
# along with this program. If not, see <http://www.gnu.org/licenses/>. | ||||
import logging | ||||
import traceback | ||||
r1692 | from rhodecode.model import BaseModel | |||
|
r3417 | from rhodecode.model.db import UserGroupMember, UserGroup,\ | ||
r3714 | UserGroupRepoToPerm, Permission, UserGroupToPerm, User, UserUserGroupToPerm | |||
|
r3417 | from rhodecode.lib.exceptions import UserGroupsAssignedException | ||
|
r1586 | |||
|
r1593 | log = logging.getLogger(__name__) | ||
|
r1586 | |||
r1716 | ||||
|
r3417 | class UserGroupModel(BaseModel): | ||
|
r1586 | |||
|
r3417 | cls = UserGroup | ||
r2524 | ||||
r3714 | def _get_user_group(self, users_group): | |||
|
r3417 | return self._get_instance(UserGroup, users_group, | ||
callback=UserGroup.get_by_group_name) | ||||
r1982 | ||||
r3714 | def _create_default_perms(self, user_group): | |||
# create default permission | ||||
default_perm = 'usergroup.read' | ||||
def_user = User.get_by_username('default') | ||||
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_id = def_user.user_id | ||||
return user_group_to_perm | ||||
def _update_permissions(self, user_group, perms_new=None, | ||||
perms_updates=None): | ||||
if not perms_new: | ||||
perms_new = [] | ||||
if not perms_updates: | ||||
perms_updates = [] | ||||
# update permissions | ||||
for member, perm, member_type in perms_updates: | ||||
if member_type == 'user': | ||||
# this updates existing one | ||||
self.grant_user_permission( | ||||
user_group=user_group, user=member, perm=perm | ||||
) | ||||
else: | ||||
self.grant_users_group_permission( | ||||
user_group=user_group, group_name=member, perm=perm | ||||
) | ||||
# set new permissions | ||||
for member, perm, member_type in perms_new: | ||||
if member_type == 'user': | ||||
self.grant_user_permission( | ||||
user_group=user_group, user=member, perm=perm | ||||
) | ||||
else: | ||||
self.grant_users_group_permission( | ||||
user_group=user_group, group_name=member, perm=perm | ||||
) | ||||
r1716 | def get(self, users_group_id, cache=False): | |||
|
r3417 | return UserGroup.get(users_group_id) | ||
|
r1586 | |||
r2522 | def get_group(self, users_group): | |||
r3714 | return self._get_user_group(users_group) | |||
r2522 | ||||
r1716 | def get_by_name(self, name, cache=False, case_insensitive=False): | |||
|
r3417 | return UserGroup.get_by_group_name(name, cache, case_insensitive) | ||
|
r1586 | |||
r3714 | def create(self, name, owner, active=True): | |||
r1749 | try: | |||
r3714 | new_user_group = UserGroup() | |||
new_user_group.user = self._get_user(owner) | ||||
new_user_group.users_group_name = name | ||||
new_user_group.users_group_active = active | ||||
self.sa.add(new_user_group) | ||||
perm_obj = self._create_default_perms(new_user_group) | ||||
self.sa.add(perm_obj) | ||||
return new_user_group | ||||
r3631 | except Exception: | |||
r1749 | log.error(traceback.format_exc()) | |||
raise | ||||
def update(self, users_group, form_data): | ||||
try: | ||||
r3714 | users_group = self._get_user_group(users_group) | |||
r1749 | ||||
for k, v in form_data.items(): | ||||
if k == 'users_group_members': | ||||
users_group.members = [] | ||||
self.sa.flush() | ||||
members_list = [] | ||||
if v: | ||||
v = [v] if isinstance(v, basestring) else v | ||||
for u_id in set(v): | ||||
|
r3417 | member = UserGroupMember(users_group.users_group_id, u_id) | ||
r1749 | members_list.append(member) | |||
setattr(users_group, 'members', members_list) | ||||
setattr(users_group, k, v) | ||||
self.sa.add(users_group) | ||||
r3631 | except Exception: | |||
r1749 | log.error(traceback.format_exc()) | |||
raise | ||||
r1713 | ||||
r1982 | def delete(self, users_group, force=False): | |||
""" | ||||
|
r3653 | Deletes repository group, unless force flag is used | ||
r1982 | raises exception if there are members in that group, else deletes | |||
group and users | ||||
:param users_group: | ||||
:param force: | ||||
""" | ||||
r1749 | try: | |||
r3714 | users_group = self._get_user_group(users_group) | |||
r1818 | ||||
r1749 | # check if this group is not assigned to repo | |||
|
r3417 | assigned_groups = UserGroupRepoToPerm.query()\ | ||
.filter(UserGroupRepoToPerm.users_group == users_group).all() | ||||
r1749 | ||||
|
r3625 | if assigned_groups and not force: | ||
|
r3417 | raise UserGroupsAssignedException('RepoGroup assigned to %s' % | ||
r1749 | assigned_groups) | |||
r1818 | ||||
r1749 | self.sa.delete(users_group) | |||
r3631 | except Exception: | |||
r1749 | log.error(traceback.format_exc()) | |||
raise | ||||
r1713 | ||||
|
r1593 | def add_user_to_group(self, users_group, user): | ||
r3714 | users_group = self._get_user_group(users_group) | |||
r2432 | user = self._get_user(user) | |||
r1989 | ||||
|
r1589 | for m in users_group.members: | ||
u = m.user | ||||
if u.user_id == user.user_id: | ||||
r1989 | return True | |||
|
r1589 | |||
|
r1586 | try: | ||
|
r3417 | users_group_member = UserGroupMember() | ||
|
r1586 | users_group_member.user = user | ||
users_group_member.users_group = users_group | ||||
|
r1593 | users_group.members.append(users_group_member) | ||
user.group_member.append(users_group_member) | ||||
|
r1587 | |||
|
r1593 | self.sa.add(users_group_member) | ||
|
r1586 | return users_group_member | ||
r3631 | except Exception: | |||
|
r1593 | log.error(traceback.format_exc()) | ||
|
r1587 | raise | ||
r1749 | ||||
r1989 | def remove_user_from_group(self, users_group, user): | |||
r3714 | users_group = self._get_user_group(users_group) | |||
r2432 | user = self._get_user(user) | |||
r1989 | ||||
users_group_member = None | ||||
for m in users_group.members: | ||||
if m.user.user_id == user.user_id: | ||||
# Found this user's membership row | ||||
users_group_member = m | ||||
break | ||||
if users_group_member: | ||||
try: | ||||
self.sa.delete(users_group_member) | ||||
return True | ||||
r3631 | except Exception: | |||
r1989 | log.error(traceback.format_exc()) | |||
raise | ||||
else: | ||||
# User isn't in that group | ||||
return False | ||||
r1749 | def has_perm(self, users_group, perm): | |||
r3714 | users_group = self._get_user_group(users_group) | |||
r2432 | perm = self._get_perm(perm) | |||
r1749 | ||||
|
r3417 | return UserGroupToPerm.query()\ | ||
.filter(UserGroupToPerm.users_group == users_group)\ | ||||
.filter(UserGroupToPerm.permission == perm).scalar() is not None | ||||
r1749 | ||||
def grant_perm(self, users_group, perm): | ||||
r3714 | users_group = self._get_user_group(users_group) | |||
r2709 | perm = self._get_perm(perm) | |||
r1749 | ||||
r2078 | # if this permission is already granted skip it | |||
|
r3417 | _perm = UserGroupToPerm.query()\ | ||
.filter(UserGroupToPerm.users_group == users_group)\ | ||||
.filter(UserGroupToPerm.permission == perm)\ | ||||
r2078 | .scalar() | |||
if _perm: | ||||
return | ||||
|
r3417 | new = UserGroupToPerm() | ||
r1749 | new.users_group = users_group | |||
new.permission = perm | ||||
self.sa.add(new) | ||||
def revoke_perm(self, users_group, perm): | ||||
r3714 | users_group = self._get_user_group(users_group) | |||
r2432 | perm = self._get_perm(perm) | |||
r1818 | ||||
|
r3417 | obj = UserGroupToPerm.query()\ | ||
.filter(UserGroupToPerm.users_group == users_group)\ | ||||
.filter(UserGroupToPerm.permission == perm).scalar() | ||||
r1932 | if obj: | |||
self.sa.delete(obj) | ||||
r3714 | ||||
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 | ||||
""" | ||||
user_group = self._get_user_group(user_group) | ||||
user = self._get_user(user) | ||||
permission = self._get_perm(perm) | ||||
# 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)) | ||||
def revoke_user_permission(self, user_group, user): | ||||
""" | ||||
Revoke permission for user on given repository group | ||||
:param user_group: Instance of ReposGroup, repositories_group_id, | ||||
or repositories_group name | ||||
:param user: Instance of User, user_id or username | ||||
""" | ||||
user_group = self._get_user_group(user_group) | ||||
user = self._get_user(user) | ||||
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)) | ||||
def grant_users_group_permission(self, user_group, group_name, perm): | ||||
raise NotImplementedError() | ||||
def revoke_users_group_permission(self, user_group, group_name): | ||||
raise NotImplementedError() | ||||