Changeset - 43481c3d70ca
[Not reviewed]
beta
0 4 0
Marcin Kuzminski - 14 years ago 2012-03-14 17:51:00
marcin@python-works.com
#399 added inheritance of permissions for users group on repos groups
3 files changed with 94 insertions and 11 deletions:
0 comments (0 inline, 0 general)
docs/changelog.rst
Show inline comments
 
@@ -11,24 +11,25 @@ Changelog
 
:status: in-progress
 
:branch: beta
 

	
 
news
 
++++
 

	
 
- Whoosh logging is now controlled by the .ini files logging setup
 
- added clone-url into edit form on /settings page
 
- added help text into repo add/edit forms
 
- created rcextensions module with additional mappings (ref #322) and
 
  post push/pull/create repo hooks callbacks
 
- implemented #377 Users view for his own permissions on account page
 
- #399 added inheritance of permissions for users group on repos groups
 

	
 
fixes
 
+++++
 

	
 
- fixed #390 cache invalidation problems on repos inside group
 
- fixed #385 clone by ID url was loosing proxy prefix in URL
 
- fixed some unicode problems with waitress
 
- fixed issue with escaping < and > in changeset commits
 
- fixed error occurring during recursive group creation in API 
 
  create_repo function
 
- fixed #393 py2.5 fixes for routes url generator
 
- fixed #397 Private repository groups shows up before login
rhodecode/model/user.py
Show inline comments
 
@@ -26,25 +26,26 @@
 
import logging
 
import traceback
 

	
 
from pylons import url
 
from pylons.i18n.translation import _
 

	
 
from rhodecode.lib.utils2 import safe_unicode, generate_api_key
 
from rhodecode.lib.caching_query import FromCache
 

	
 
from rhodecode.model import BaseModel
 
from rhodecode.model.db import User, UserRepoToPerm, Repository, Permission, \
 
    UserToPerm, UsersGroupRepoToPerm, UsersGroupToPerm, UsersGroupMember, \
 
    Notification, RepoGroup, UserRepoGroupToPerm, UsersGroup
 
    Notification, RepoGroup, UserRepoGroupToPerm, UsersGroup,\
 
    UsersGroupRepoGroupToPerm
 
from rhodecode.lib.exceptions import DefaultUserException, \
 
    UserOwnsReposException
 

	
 
from sqlalchemy.exc import DatabaseError
 

	
 
from sqlalchemy.orm import joinedload
 

	
 
log = logging.getLogger(__name__)
 

	
 

	
 
PERM_WEIGHTS = {
 
    'repository.none': 0,
 
@@ -401,124 +402,150 @@ class UserModel(BaseModel):
 
            #==================================================================
 
            # set default permissions first for repositories and groups
 
            #==================================================================
 
            uid = user.user_id
 

	
 
            # default global permissions
 
            default_global_perms = self.sa.query(UserToPerm)\
 
                .filter(UserToPerm.user_id == default_user_id)
 

	
 
            for perm in default_global_perms:
 
                user.permissions[GLOBAL].add(perm.permission.permission_name)
 

	
 
            # default for repositories
 
            # defaults for repositories, taken from default user
 
            for perm in default_repo_perms:
 
                r_k = perm.UserRepoToPerm.repository.repo_name
 
                if perm.Repository.private and not (perm.Repository.user_id == uid):
 
                    # disable defaults for private repos,
 
                    p = 'repository.none'
 
                elif perm.Repository.user_id == uid:
 
                    # set admin if owner
 
                    p = 'repository.admin'
 
                else:
 
                    p = perm.Permission.permission_name
 

	
 
                user.permissions[RK][r_k] = p
 

	
 
            # default for repositories groups
 
            # defaults for repositories groups taken from default user permission
 
            # on given group
 
            for perm in default_repo_groups_perms:
 
                rg_k = perm.UserRepoGroupToPerm.group.group_name
 
                p = perm.Permission.permission_name
 
                user.permissions[GK][rg_k] = p
 

	
 
            #==================================================================
 
            # overwrite default with user permissions if any
 
            # overwrite defaults with user permissions if any found
 
            #==================================================================
 

	
 
            # user global
 
            # user global permissions
 
            user_perms = self.sa.query(UserToPerm)\
 
                    .options(joinedload(UserToPerm.permission))\
 
                    .filter(UserToPerm.user_id == uid).all()
 

	
 
            for perm in user_perms:
 
                user.permissions[GLOBAL].add(perm.permission.permission_name)
 

	
 
            # user repositories
 
            # user explicit permissions for repositories
 
            user_repo_perms = \
 
             self.sa.query(UserRepoToPerm, Permission, Repository)\
 
             .join((Repository, UserRepoToPerm.repository_id == Repository.repo_id))\
 
             .join((Permission, UserRepoToPerm.permission_id == Permission.permission_id))\
 
             .filter(UserRepoToPerm.user_id == uid)\
 
             .all()
 

	
 
            for perm in user_repo_perms:
 
                # set admin if owner
 
                r_k = perm.UserRepoToPerm.repository.repo_name
 
                if perm.Repository.user_id == uid:
 
                    p = 'repository.admin'
 
                else:
 
                    p = perm.Permission.permission_name
 
                user.permissions[RK][r_k] = p
 

	
 
            #==================================================================
 
            # check if user is part of groups for this repository and fill in
 
            # (or replace with higher) permissions
 
            # check if user is part of user groups for this repository and
 
            # fill in (or replace with higher) permissions
 
            #==================================================================
 

	
 
            # users group global
 
            user_perms_from_users_groups = self.sa.query(UsersGroupToPerm)\
 
                .options(joinedload(UsersGroupToPerm.permission))\
 
                .join((UsersGroupMember, UsersGroupToPerm.users_group_id ==
 
                       UsersGroupMember.users_group_id))\
 
                .filter(UsersGroupMember.user_id == uid).all()
 

	
 
            for perm in user_perms_from_users_groups:
 
                user.permissions[GLOBAL].add(perm.permission.permission_name)
 

	
 
            # users group repositories
 
            # users group for repositories permissions
 
            user_repo_perms_from_users_groups = \
 
             self.sa.query(UsersGroupRepoToPerm, Permission, Repository,)\
 
             .join((Repository, UsersGroupRepoToPerm.repository_id == Repository.repo_id))\
 
             .join((Permission, UsersGroupRepoToPerm.permission_id == Permission.permission_id))\
 
             .join((UsersGroupMember, UsersGroupRepoToPerm.users_group_id == UsersGroupMember.users_group_id))\
 
             .filter(UsersGroupMember.user_id == uid)\
 
             .all()
 

	
 
            for perm in user_repo_perms_from_users_groups:
 
                r_k = perm.UsersGroupRepoToPerm.repository.repo_name
 
                p = perm.Permission.permission_name
 
                cur_perm = user.permissions[RK][r_k]
 
                # overwrite permission only if it's greater than permission
 
                # given from other sources
 
                if PERM_WEIGHTS[p] > PERM_WEIGHTS[cur_perm]:
 
                    user.permissions[RK][r_k] = p
 

	
 
            #==================================================================
 
            # get access for this user for repos group and override defaults
 
            #==================================================================
 

	
 
            # user repositories groups
 
            # user explicit permissions for repository
 
            user_repo_groups_perms = \
 
             self.sa.query(UserRepoGroupToPerm, Permission, RepoGroup)\
 
             .join((RepoGroup, UserRepoGroupToPerm.group_id == RepoGroup.group_id))\
 
             .join((Permission, UserRepoGroupToPerm.permission_id == Permission.permission_id))\
 
             .filter(UserRepoGroupToPerm.user_id == uid)\
 
             .all()
 

	
 
            for perm in user_repo_groups_perms:
 
                rg_k = perm.UserRepoGroupToPerm.group.group_name
 
                p = perm.Permission.permission_name
 
                cur_perm = user.permissions[GK][rg_k]
 
                if PERM_WEIGHTS[p] > PERM_WEIGHTS[cur_perm]:
 
                    user.permissions[GK][rg_k] = p
 

	
 
            #==================================================================
 
            # check if user is part of user groups for this repo group and
 
            # fill in (or replace with higher) permissions
 
            #==================================================================
 

	
 
            # users group for repositories permissions
 
            user_repo_group_perms_from_users_groups = \
 
             self.sa.query(UsersGroupRepoGroupToPerm, Permission, RepoGroup)\
 
             .join((RepoGroup, UsersGroupRepoGroupToPerm.group_id == RepoGroup.group_id))\
 
             .join((Permission, UsersGroupRepoGroupToPerm.permission_id == Permission.permission_id))\
 
             .join((UsersGroupMember, UsersGroupRepoGroupToPerm.users_group_id == UsersGroupMember.users_group_id))\
 
             .filter(UsersGroupMember.user_id == uid)\
 
             .all()
 
            
 
            for perm in user_repo_group_perms_from_users_groups:
 
                g_k = perm.UsersGroupRepoGroupToPerm.group.group_name
 
                print perm, g_k
 
                p = perm.Permission.permission_name
 
                cur_perm = user.permissions[GK][g_k]
 
                # overwrite permission only if it's greater than permission
 
                # given from other sources
 
                if PERM_WEIGHTS[p] > PERM_WEIGHTS[cur_perm]:
 
                    user.permissions[GK][g_k] = p
 

	
 
        return user
 

	
 
    def has_perm(self, user, perm):
 
        if not isinstance(perm, Permission):
 
            raise Exception('perm needs to be an instance of Permission class '
 
                            'got %s instead' % type(perm))
 

	
 
        user = self.__get_user(user)
 

	
 
        return UserToPerm.query().filter(UserToPerm.user == user)\
 
            .filter(UserToPerm.permission == perm).scalar() is not None
 

	
rhodecode/tests/test_models.py
Show inline comments
 
import os
 
import unittest
 
from rhodecode.tests import *
 

	
 
from rhodecode.model.repos_group import ReposGroupModel
 
from rhodecode.model.repo import RepoModel
 
from rhodecode.model.db import RepoGroup, User, Notification, UserNotification, \
 
    UsersGroup, UsersGroupMember, Permission
 
    UsersGroup, UsersGroupMember, Permission, UsersGroupRepoGroupToPerm
 
from sqlalchemy.exc import IntegrityError
 
from rhodecode.model.user import UserModel
 

	
 
from rhodecode.model.meta import Session
 
from rhodecode.model.notification import NotificationModel
 
from rhodecode.model.users_group import UsersGroupModel
 
from rhodecode.lib.auth import AuthUser
 

	
 

	
 
def _make_group(path, desc='desc', parent_id=None,
 
                 skip_if_exists=False):
 

	
 
@@ -599,24 +599,25 @@ class TestPermissions(unittest.TestCase)
 
        a1_auth = AuthUser(user_id=self.anon.user_id)
 
        self.assertEqual(a1_auth.permissions['repositories_groups'],
 
                 {u'group1': u'group.read', u'group2': u'group.read'})
 

	
 
        #Change perms to none for both groups
 
        ReposGroupModel().grant_user_permission(repos_group=self.g1,
 
                                                user=self.anon,
 
                                                perm='group.none')
 
        ReposGroupModel().grant_user_permission(repos_group=self.g2,
 
                                                user=self.anon,
 
                                                perm='group.none')
 

	
 

	
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        self.assertEqual(u1_auth.permissions['repositories_groups'],
 
                 {u'group1': u'group.none', u'group2': u'group.none'})
 

	
 
        a1_auth = AuthUser(user_id=self.anon.user_id)
 
        self.assertEqual(a1_auth.permissions['repositories_groups'],
 
                 {u'group1': u'group.none', u'group2': u'group.none'})
 

	
 
        # add repo to group
 
        form_data = {
 
            'repo_name':HG_REPO,
 
            'repo_name_full':os.path.join(self.g1.group_name,HG_REPO),
 
@@ -649,12 +650,66 @@ class TestPermissions(unittest.TestCase)
 
        #u1 and anon should have not change perms while u2 should !
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        self.assertEqual(u1_auth.permissions['repositories_groups'],
 
                 {u'group1': u'group.none', u'group2': u'group.none'})
 

	
 
        u2_auth = AuthUser(user_id=self.u2.user_id)
 
        self.assertEqual(u2_auth.permissions['repositories_groups'],
 
                 {u'group1': u'group.read', u'group2': u'group.read'})
 

	
 
        a1_auth = AuthUser(user_id=self.anon.user_id)
 
        self.assertEqual(a1_auth.permissions['repositories_groups'],
 
                 {u'group1': u'group.none', u'group2': u'group.none'})
 

	
 
    def test_repo_group_user_as_user_group_member(self):
 
        # create Group1
 
        self.g1 = _make_group('group1', skip_if_exists=True)
 
        Session.commit()
 
        a1_auth = AuthUser(user_id=self.anon.user_id)
 

	
 
        self.assertEqual(a1_auth.permissions['repositories_groups'],
 
                         {u'group1': u'group.read'})
 

	
 
        # set default permission to none
 
        ReposGroupModel().grant_user_permission(repos_group=self.g1,
 
                                                user=self.anon,
 
                                                perm='group.none')
 
        # make group
 
        self.ug1 = UsersGroupModel().create('G1')
 
        # add user to group
 
        UsersGroupModel().add_user_to_group(self.ug1, self.u1)
 
        Session.commit()
 

	
 
        # check if user is in the group
 
        membrs = [x.user_id for x in UsersGroupModel().get(self.ug1.users_group_id).members]
 
        self.assertEqual(membrs, [self.u1.user_id])
 
        # add some user to that group
 

	
 
        # check his permissions
 
        a1_auth = AuthUser(user_id=self.anon.user_id)
 
        self.assertEqual(a1_auth.permissions['repositories_groups'],
 
                         {u'group1': u'group.none'})
 

	
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        self.assertEqual(u1_auth.permissions['repositories_groups'],
 
                         {u'group1': u'group.none'})
 

	
 
        # grant ug1 read permissions for
 
        ReposGroupModel().grant_users_group_permission(repos_group=self.g1,
 
                                                       group_name=self.ug1,
 
                                                       perm='group.read')
 
        Session.commit()
 
        # check if the
 
        obj = Session.query(UsersGroupRepoGroupToPerm)\
 
            .filter(UsersGroupRepoGroupToPerm.group == self.g1)\
 
            .filter(UsersGroupRepoGroupToPerm.users_group == self.ug1)\
 
            .scalar()
 
        self.assertEqual(obj.permission.permission_name, 'group.read')
 

	
 
        a1_auth = AuthUser(user_id=self.anon.user_id)
 

	
 
        self.assertEqual(a1_auth.permissions['repositories_groups'],
 
                         {u'group1': u'group.none'})
 

	
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        self.assertEqual(u1_auth.permissions['repositories_groups'],
 
                         {u'group1': u'group.read'})
0 comments (0 inline, 0 general)