Changeset - 1ecd6c0e2787
[Not reviewed]
default
0 6 0
Mads Kiilerich (mads) - 6 years ago 2020-08-18 16:40:19
mads@kiilerich.com
auth: refactor permissions

Avoid using complex vague typing in dict-of-dicts.
6 files changed with 108 insertions and 113 deletions:
0 comments (0 inline, 0 general)
kallithea/controllers/admin/my_account.py
Show inline comments
 
@@ -91,7 +91,7 @@ class MyAccountController(BaseController
 
        self.__load_data()
 
        c.perm_user = AuthUser(user_id=request.authuser.user_id)
 
        managed_fields = auth_modules.get_managed_fields(c.user)
 
        def_user_perms = AuthUser(dbuser=User.get_default_user()).permissions['global']
 
        def_user_perms = AuthUser(dbuser=User.get_default_user()).global_permissions
 
        if 'hg.register.none' in def_user_perms:
 
            managed_fields.extend(['username', 'firstname', 'lastname', 'email'])
 

	
kallithea/controllers/login.py
Show inline comments
 
@@ -118,7 +118,7 @@ class LoginController(BaseController):
 
    @HasPermissionAnyDecorator('hg.admin', 'hg.register.auto_activate',
 
                               'hg.register.manual_activate')
 
    def register(self):
 
        def_user_perms = AuthUser(dbuser=User.get_default_user()).permissions['global']
 
        def_user_perms = AuthUser(dbuser=User.get_default_user()).global_permissions
 
        c.auto_active = 'hg.register.auto_activate' in def_user_perms
 

	
 
        settings = Setting.get_app_settings()
kallithea/lib/auth.py
Show inline comments
 
@@ -43,7 +43,6 @@ import kallithea
 
from kallithea.config.routing import url
 
from kallithea.lib.utils import get_repo_group_slug, get_repo_slug, get_user_group_slug
 
from kallithea.lib.utils2 import ascii_bytes, ascii_str, safe_bytes
 
from kallithea.lib.vcs.utils.lazy import LazyProperty
 
from kallithea.model.db import (Permission, UserApiKeys, UserGroup, UserGroupMember, UserGroupRepoGroupToPerm, UserGroupRepoToPerm, UserGroupToPerm,
 
                                UserGroupUserGroupToPerm, UserIpMap, UserToPerm)
 
from kallithea.model.meta import Session
 
@@ -117,24 +116,24 @@ def check_password(password, hashed):
 
    return False
 

	
 

	
 
def _cached_perms_data(user_id, user_is_admin):
 
    RK = 'repositories'
 
    GK = 'repositories_groups'
 
    UK = 'user_groups'
 
    GLOBAL = 'global'
 
def get_user_permissions(user_id, user_is_admin):
 
    PERM_WEIGHTS = Permission.PERM_WEIGHTS
 
    permissions = {RK: {}, GK: {}, UK: {}, GLOBAL: set()}
 
    repository_permissions = {}
 
    repository_group_permissions = {}
 
    user_group_permissions = {}
 
    global_permissions = set()
 

	
 
    def bump_permission(kind, key, new_perm):
 
        """Add a new permission for kind and key.
 

	
 
    def bump_permission(permissions, key, new_perm):
 
        """Add a new permission for key to permissions.
 
        Assuming the permissions are comparable, set the new permission if it
 
        has higher weight, else drop it and keep the old permission.
 
        """
 
        cur_perm = permissions[kind][key]
 
        cur_perm = permissions[key]
 
        new_perm_val = PERM_WEIGHTS[new_perm]
 
        cur_perm_val = PERM_WEIGHTS[cur_perm]
 
        if new_perm_val > cur_perm_val:
 
            permissions[kind][key] = new_perm
 
            permissions[key] = new_perm
 

	
 
    #======================================================================
 
    # fetch default permissions
 
@@ -148,26 +147,26 @@ def _cached_perms_data(user_id, user_is_
 
        # admin users have all rights;
 
        # based on default permissions, just set everything to admin
 
        #==================================================================
 
        permissions[GLOBAL].add('hg.admin')
 
        global_permissions.add('hg.admin')
 

	
 
        # repositories
 
        for perm in default_repo_perms:
 
            r_k = perm.repository.repo_name
 
            p = 'repository.admin'
 
            permissions[RK][r_k] = p
 
            repository_permissions[r_k] = p
 

	
 
        # repository groups
 
        for perm in default_repo_groups_perms:
 
            rg_k = perm.group.group_name
 
            p = 'group.admin'
 
            permissions[GK][rg_k] = p
 
            repository_group_permissions[rg_k] = p
 

	
 
        # user groups
 
        for perm in default_user_group_perms:
 
            u_k = perm.user_group.users_group_name
 
            p = 'usergroup.admin'
 
            permissions[UK][u_k] = p
 
        return permissions
 
            user_group_permissions[u_k] = p
 
        return (repository_permissions, repository_group_permissions, user_group_permissions, global_permissions)
 

	
 
    #==================================================================
 
    # SET DEFAULTS GLOBAL, REPOS, REPOSITORY GROUPS
 
@@ -179,7 +178,7 @@ def _cached_perms_data(user_id, user_is_
 
        .options(joinedload(UserToPerm.permission))
 

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

	
 
    # defaults for repositories, taken from default user
 
    for perm in default_repo_perms:
 
@@ -190,21 +189,21 @@ def _cached_perms_data(user_id, user_is_
 
            p = 'repository.none'
 
        else:
 
            p = perm.permission.permission_name
 
        permissions[RK][r_k] = p
 
        repository_permissions[r_k] = p
 

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

	
 
    # defaults for user groups taken from default user permission
 
    # on given user group
 
    for perm in default_user_group_perms:
 
        u_k = perm.user_group.users_group_name
 
        p = perm.permission.permission_name
 
        permissions[UK][u_k] = p
 
        user_group_permissions[u_k] = p
 

	
 
    #======================================================================
 
    # !! Augment GLOBALS with user permissions if any found !!
 
@@ -229,7 +228,7 @@ def _cached_perms_data(user_id, user_is_
 
                                  lambda x:x.users_group)]
 
    for gr, perms in _grouped:
 
        for perm in perms:
 
            permissions[GLOBAL].add(perm.permission.permission_name)
 
            global_permissions.add(perm.permission.permission_name)
 

	
 
    # user specific global permissions
 
    user_perms = Session().query(UserToPerm) \
 
@@ -237,14 +236,14 @@ def _cached_perms_data(user_id, user_is_
 
            .filter(UserToPerm.user_id == user_id).all()
 

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

	
 
    # for each kind of global permissions, only keep the one with heighest weight
 
    kind_max_perm = {}
 
    for perm in sorted(permissions[GLOBAL], key=lambda n: PERM_WEIGHTS.get(n, -1)):
 
    for perm in sorted(global_permissions, key=lambda n: PERM_WEIGHTS.get(n, -1)):
 
        kind = perm.rsplit('.', 1)[0]
 
        kind_max_perm[kind] = perm
 
    permissions[GLOBAL] = set(kind_max_perm.values())
 
    global_permissions = set(kind_max_perm.values())
 
    ## END GLOBAL PERMISSIONS
 

	
 
    #======================================================================
 
@@ -269,14 +268,14 @@ def _cached_perms_data(user_id, user_is_
 
        .all()
 

	
 
    for perm in user_repo_perms_from_users_groups:
 
        bump_permission(RK,
 
        bump_permission(repository_permissions,
 
            perm.repository.repo_name,
 
            perm.permission.permission_name)
 

	
 
    # user permissions for repositories
 
    user_repo_perms = Permission.get_default_perms(user_id)
 
    for perm in user_repo_perms:
 
        bump_permission(RK,
 
        bump_permission(repository_permissions,
 
            perm.repository.repo_name,
 
            perm.permission.permission_name)
 

	
 
@@ -300,14 +299,14 @@ def _cached_perms_data(user_id, user_is_
 
     .all()
 

	
 
    for perm in user_repo_group_perms_from_users_groups:
 
        bump_permission(GK,
 
        bump_permission(repository_group_permissions,
 
            perm.group.group_name,
 
            perm.permission.permission_name)
 

	
 
    # user explicit permissions for repository groups
 
    user_repo_groups_perms = Permission.get_default_group_perms(user_id)
 
    for perm in user_repo_groups_perms:
 
        bump_permission(GK,
 
        bump_permission(repository_group_permissions,
 
            perm.group.group_name,
 
            perm.permission.permission_name)
 

	
 
@@ -329,18 +328,18 @@ def _cached_perms_data(user_id, user_is_
 
     .all()
 

	
 
    for perm in user_group_user_groups_perms:
 
        bump_permission(UK,
 
        bump_permission(user_group_permissions,
 
            perm.target_user_group.users_group_name,
 
            perm.permission.permission_name)
 

	
 
    # user explicit permission for user groups
 
    user_user_groups_perms = Permission.get_default_user_group_perms(user_id)
 
    for perm in user_user_groups_perms:
 
        bump_permission(UK,
 
        bump_permission(user_group_permissions,
 
            perm.user_group.users_group_name,
 
            perm.permission.permission_name)
 

	
 
    return permissions
 
    return (repository_permissions, repository_group_permissions, user_group_permissions, global_permissions)
 

	
 

	
 
class AuthUser(object):
 
@@ -428,17 +427,15 @@ class AuthUser(object):
 
            self.is_default_user = dbuser.is_default_user
 
        log.debug('Auth User is now %s', self)
 

	
 
    @LazyProperty
 
    def permissions(self):
 
        """
 
        Fills user permission attribute with permissions taken from database
 
        works for permissions given for repositories, and for permissions that
 
        are granted to groups
 

	
 
        :param user: `AuthUser` instance
 
        """
 
        log.debug('Getting PERMISSION tree for %s', self)
 
        return _cached_perms_data(self.user_id, self.is_admin)
 
        (self.repository_permissions, self.repository_group_permissions, self.user_group_permissions, self.global_permissions,
 
        )= get_user_permissions(self.user_id, self.is_admin)
 
        self.permissions = {
 
            'global': self.global_permissions,
 
            'repositories': self.repository_permissions,
 
            'repositories_groups': self.repository_group_permissions,
 
            'user_groups': self.user_group_permissions,
 
        } # backwards compatibility
 

	
 
    def has_repository_permission_level(self, repo_name, level, purpose=None):
 
        required_perms = {
 
@@ -446,7 +443,7 @@ class AuthUser(object):
 
            'write': ['repository.write', 'repository.admin'],
 
            'admin': ['repository.admin'],
 
        }[level]
 
        actual_perm = self.permissions['repositories'].get(repo_name)
 
        actual_perm = self.repository_permissions.get(repo_name)
 
        ok = actual_perm in required_perms
 
        log.debug('Checking if user %r can %r repo %r (%s): %s (has %r)',
 
            self.username, level, repo_name, purpose, ok, actual_perm)
 
@@ -458,7 +455,7 @@ class AuthUser(object):
 
            'write': ['group.write', 'group.admin'],
 
            'admin': ['group.admin'],
 
        }[level]
 
        actual_perm = self.permissions['repositories_groups'].get(repo_group_name)
 
        actual_perm = self.repository_group_permissions.get(repo_group_name)
 
        ok = actual_perm in required_perms
 
        log.debug('Checking if user %r can %r repo group %r (%s): %s (has %r)',
 
            self.username, level, repo_group_name, purpose, ok, actual_perm)
 
@@ -470,7 +467,7 @@ class AuthUser(object):
 
            'write': ['usergroup.write', 'usergroup.admin'],
 
            'admin': ['usergroup.admin'],
 
        }[level]
 
        actual_perm = self.permissions['user_groups'].get(user_group_name)
 
        actual_perm = self.user_group_permissions.get(user_group_name)
 
        ok = actual_perm in required_perms
 
        log.debug('Checking if user %r can %r user group %r (%s): %s (has %r)',
 
            self.username, level, user_group_name, purpose, ok, actual_perm)
 
@@ -497,7 +494,7 @@ class AuthUser(object):
 
        """
 
        Returns list of repositories you're an admin of
 
        """
 
        return [x[0] for x in self.permissions['repositories'].items()
 
        return [x[0] for x in self.repository_permissions.items()
 
                if x[1] == 'repository.admin']
 

	
 
    @property
 
@@ -505,7 +502,7 @@ class AuthUser(object):
 
        """
 
        Returns list of repository groups you're an admin of
 
        """
 
        return [x[0] for x in self.permissions['repositories_groups'].items()
 
        return [x[0] for x in self.repository_group_permissions.items()
 
                if x[1] == 'group.admin']
 

	
 
    @property
 
@@ -513,7 +510,7 @@ class AuthUser(object):
 
        """
 
        Returns list of user groups you're an admin of
 
        """
 
        return [x[0] for x in self.permissions['user_groups'].items()
 
        return [x[0] for x in self.user_group_permissions.items()
 
                if x[1] == 'usergroup.admin']
 

	
 
    def __repr__(self):
 
@@ -672,8 +669,7 @@ class HasPermissionAnyDecorator(_PermsDe
 
    """
 

	
 
    def check_permissions(self, user):
 
        global_permissions = user.permissions['global'] # usually very short
 
        return any(p in global_permissions for p in self.required_perms)
 
        return any(p in user.global_permissions for p in self.required_perms)
 

	
 

	
 
class _PermDecorator(_PermsDecorator):
 
@@ -739,8 +735,7 @@ class _PermsFunction(object):
 
class HasPermissionAny(_PermsFunction):
 

	
 
    def __call__(self, purpose=None):
 
        global_permissions = request.authuser.permissions['global'] # usually very short
 
        ok = any(p in global_permissions for p in self.required_perms)
 
        ok = any(p in request.authuser.global_permissions for p in self.required_perms)
 

	
 
        log.debug('Check %s for global %s (%s): %s',
 
            request.authuser.username, self.required_perms, purpose, ok)
 
@@ -783,7 +778,7 @@ class HasPermissionAnyMiddleware(object)
 

	
 
    def __call__(self, authuser, repo_name, purpose=None):
 
        try:
 
            ok = authuser.permissions['repositories'][repo_name] in self.required_perms
 
            ok = authuser.repository_permissions[repo_name] in self.required_perms
 
        except KeyError:
 
            ok = False
 

	
kallithea/lib/auth_modules/__init__.py
Show inline comments
 
@@ -240,7 +240,7 @@ class KallitheaExternalAuthPlugin(Kallit
 
            userobj, username, passwd, settings, **kwargs)
 
        if user_data is not None:
 
            if userobj is None: # external authentication of unknown user that will be created soon
 
                def_user_perms = AuthUser(dbuser=User.get_default_user()).permissions['global']
 
                def_user_perms = AuthUser(dbuser=User.get_default_user()).global_permissions
 
                active = 'hg.extern_activate.auto' in def_user_perms
 
            else:
 
                active = userobj.active
kallithea/model/repo.py
Show inline comments
 
@@ -102,7 +102,7 @@ class RepoModel(object):
 
        from kallithea.lib.auth import AuthUser
 
        auth_user = AuthUser(dbuser=User.guess_instance(user))
 
        repos = [repo_name
 
            for repo_name, perm in auth_user.permissions['repositories'].items()
 
            for repo_name, perm in auth_user.repository_permissions.items()
 
            if perm in ['repository.read', 'repository.write', 'repository.admin']
 
            ]
 
        return Repository.query().filter(Repository.repo_name.in_(repos))
kallithea/tests/models/test_permissions.py
Show inline comments
 
@@ -68,18 +68,18 @@ class TestPermissions(base.TestControlle
 

	
 
    def test_default_perms_set(self):
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        assert u1_auth.permissions['repositories'][base.HG_REPO] == 'repository.read'
 
        assert u1_auth.repository_permissions[base.HG_REPO] == 'repository.read'
 
        new_perm = 'repository.write'
 
        RepoModel().grant_user_permission(repo=base.HG_REPO, user=self.u1,
 
                                          perm=new_perm)
 
        Session().commit()
 

	
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        assert u1_auth.permissions['repositories'][base.HG_REPO] == new_perm
 
        assert u1_auth.repository_permissions[base.HG_REPO] == new_perm
 

	
 
    def test_default_admin_perms_set(self):
 
        a1_auth = AuthUser(user_id=self.a1.user_id)
 
        assert a1_auth.permissions['repositories'][base.HG_REPO] == 'repository.admin'
 
        assert a1_auth.repository_permissions[base.HG_REPO] == 'repository.admin'
 
        new_perm = 'repository.write'
 
        RepoModel().grant_user_permission(repo=base.HG_REPO, user=self.a1,
 
                                          perm=new_perm)
 
@@ -87,24 +87,24 @@ class TestPermissions(base.TestControlle
 
        # cannot really downgrade admins permissions !? they still gets set as
 
        # admin !
 
        u1_auth = AuthUser(user_id=self.a1.user_id)
 
        assert u1_auth.permissions['repositories'][base.HG_REPO] == 'repository.admin'
 
        assert u1_auth.repository_permissions[base.HG_REPO] == 'repository.admin'
 

	
 
    def test_default_group_perms(self):
 
        self.g1 = fixture.create_repo_group('test1', skip_if_exists=True)
 
        self.g2 = fixture.create_repo_group('test2', skip_if_exists=True)
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        assert u1_auth.permissions['repositories'][base.HG_REPO] == 'repository.read'
 
        assert u1_auth.permissions['repositories_groups'].get('test1') == 'group.read'
 
        assert u1_auth.permissions['repositories_groups'].get('test2') == 'group.read'
 
        assert u1_auth.permissions['global'] == set(Permission.DEFAULT_USER_PERMISSIONS)
 
        assert u1_auth.repository_permissions[base.HG_REPO] == 'repository.read'
 
        assert u1_auth.repository_group_permissions.get('test1') == 'group.read'
 
        assert u1_auth.repository_group_permissions.get('test2') == 'group.read'
 
        assert u1_auth.global_permissions == set(Permission.DEFAULT_USER_PERMISSIONS)
 

	
 
    def test_default_admin_group_perms(self):
 
        self.g1 = fixture.create_repo_group('test1', skip_if_exists=True)
 
        self.g2 = fixture.create_repo_group('test2', skip_if_exists=True)
 
        a1_auth = AuthUser(user_id=self.a1.user_id)
 
        assert a1_auth.permissions['repositories'][base.HG_REPO] == 'repository.admin'
 
        assert a1_auth.permissions['repositories_groups'].get('test1') == 'group.admin'
 
        assert a1_auth.permissions['repositories_groups'].get('test2') == 'group.admin'
 
        assert a1_auth.repository_permissions[base.HG_REPO] == 'repository.admin'
 
        assert a1_auth.repository_group_permissions.get('test1') == 'group.admin'
 
        assert a1_auth.repository_group_permissions.get('test2') == 'group.admin'
 

	
 
    def test_propagated_permission_from_users_group_by_explicit_perms_exist(self):
 
        # make group
 
@@ -115,7 +115,7 @@ class TestPermissions(base.TestControlle
 
        RepoModel().grant_user_permission(repo=base.HG_REPO, user=self.u1, perm='repository.none')
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        assert u1_auth.permissions['repositories'][base.HG_REPO] == 'repository.read' # inherit from default user
 
        assert u1_auth.repository_permissions[base.HG_REPO] == 'repository.read' # inherit from default user
 

	
 
        # grant perm for group this should override permission from user
 
        RepoModel().grant_user_group_permission(repo=base.HG_REPO,
 
@@ -124,7 +124,7 @@ class TestPermissions(base.TestControlle
 

	
 
        # verify that user group permissions win
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        assert u1_auth.permissions['repositories'][base.HG_REPO] == 'repository.write'
 
        assert u1_auth.repository_permissions[base.HG_REPO] == 'repository.write'
 

	
 
    def test_propagated_permission_from_users_group(self):
 
        # make group
 
@@ -138,7 +138,7 @@ class TestPermissions(base.TestControlle
 
                                                 perm=new_perm_gr)
 
        # check perms
 
        u3_auth = AuthUser(user_id=self.u3.user_id)
 
        assert u3_auth.permissions['repositories'][base.HG_REPO] == new_perm_gr
 
        assert u3_auth.repository_permissions[base.HG_REPO] == new_perm_gr
 

	
 
    def test_propagated_permission_from_users_group_lower_weight(self):
 
        # make group
 
@@ -152,7 +152,7 @@ class TestPermissions(base.TestControlle
 
                                          perm=new_perm_h)
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        assert u1_auth.permissions['repositories'][base.HG_REPO] == new_perm_h
 
        assert u1_auth.repository_permissions[base.HG_REPO] == new_perm_h
 

	
 
        # grant perm for group this should NOT override permission from user
 
        # since it's lower than granted
 
@@ -162,19 +162,19 @@ class TestPermissions(base.TestControlle
 
                                                 perm=new_perm_l)
 
        # check perms
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        assert u1_auth.permissions['repositories'][base.HG_REPO] == new_perm_h
 
        assert u1_auth.repository_permissions[base.HG_REPO] == new_perm_h
 

	
 
    def test_repo_in_group_permissions(self):
 
        self.g1 = fixture.create_repo_group('group1', skip_if_exists=True)
 
        self.g2 = fixture.create_repo_group('group2', skip_if_exists=True)
 
        # both perms should be read !
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        assert u1_auth.permissions['repositories_groups'].get('group1') == 'group.read'
 
        assert u1_auth.permissions['repositories_groups'].get('group2') == 'group.read'
 
        assert u1_auth.repository_group_permissions.get('group1') == 'group.read'
 
        assert u1_auth.repository_group_permissions.get('group2') == 'group.read'
 

	
 
        a1_auth = AuthUser(user_id=self.anon.user_id)
 
        assert a1_auth.permissions['repositories_groups'].get('group1') == 'group.read'
 
        assert a1_auth.permissions['repositories_groups'].get('group2') == 'group.read'
 
        assert a1_auth.repository_group_permissions.get('group1') == 'group.read'
 
        assert a1_auth.repository_group_permissions.get('group2') == 'group.read'
 

	
 
        # Change perms to none for both groups
 
        RepoGroupModel().grant_user_permission(repo_group=self.g1,
 
@@ -185,12 +185,12 @@ class TestPermissions(base.TestControlle
 
                                               perm='group.none')
 

	
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        assert u1_auth.permissions['repositories_groups'].get('group1') == 'group.none'
 
        assert u1_auth.permissions['repositories_groups'].get('group2') == 'group.none'
 
        assert u1_auth.repository_group_permissions.get('group1') == 'group.none'
 
        assert u1_auth.repository_group_permissions.get('group2') == 'group.none'
 

	
 
        a1_auth = AuthUser(user_id=self.anon.user_id)
 
        assert a1_auth.permissions['repositories_groups'].get('group1') == 'group.none'
 
        assert a1_auth.permissions['repositories_groups'].get('group2') == 'group.none'
 
        assert a1_auth.repository_group_permissions.get('group1') == 'group.none'
 
        assert a1_auth.repository_group_permissions.get('group2') == 'group.none'
 

	
 
        # add repo to group
 
        name = db.URL_SEP.join([self.g1.group_name, 'test_perm'])
 
@@ -200,12 +200,12 @@ class TestPermissions(base.TestControlle
 
                                             cur_user=self.u1,)
 

	
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        assert u1_auth.permissions['repositories_groups'].get('group1') == 'group.none'
 
        assert u1_auth.permissions['repositories_groups'].get('group2') == 'group.none'
 
        assert u1_auth.repository_group_permissions.get('group1') == 'group.none'
 
        assert u1_auth.repository_group_permissions.get('group2') == 'group.none'
 

	
 
        a1_auth = AuthUser(user_id=self.anon.user_id)
 
        assert a1_auth.permissions['repositories_groups'].get('group1') == 'group.none'
 
        assert a1_auth.permissions['repositories_groups'].get('group2') == 'group.none'
 
        assert a1_auth.repository_group_permissions.get('group1') == 'group.none'
 
        assert a1_auth.repository_group_permissions.get('group2') == 'group.none'
 

	
 
        # grant permission for u2 !
 
        RepoGroupModel().grant_user_permission(repo_group=self.g1, user=self.u2,
 
@@ -216,23 +216,23 @@ class TestPermissions(base.TestControlle
 
        assert self.u1 != self.u2
 
        # u1 and anon should have not change perms while u2 should !
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        assert u1_auth.permissions['repositories_groups'].get('group1') == 'group.none'
 
        assert u1_auth.permissions['repositories_groups'].get('group2') == 'group.none'
 
        assert u1_auth.repository_group_permissions.get('group1') == 'group.none'
 
        assert u1_auth.repository_group_permissions.get('group2') == 'group.none'
 

	
 
        u2_auth = AuthUser(user_id=self.u2.user_id)
 
        assert u2_auth.permissions['repositories_groups'].get('group1') == 'group.read'
 
        assert u2_auth.permissions['repositories_groups'].get('group2') == 'group.read'
 
        assert u2_auth.repository_group_permissions.get('group1') == 'group.read'
 
        assert u2_auth.repository_group_permissions.get('group2') == 'group.read'
 

	
 
        a1_auth = AuthUser(user_id=self.anon.user_id)
 
        assert a1_auth.permissions['repositories_groups'].get('group1') == 'group.none'
 
        assert a1_auth.permissions['repositories_groups'].get('group2') == 'group.none'
 
        assert a1_auth.repository_group_permissions.get('group1') == 'group.none'
 
        assert a1_auth.repository_group_permissions.get('group2') == 'group.none'
 

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

	
 
        assert a1_auth.permissions['repositories_groups'].get('group1') == 'group.read'
 
        assert a1_auth.repository_group_permissions.get('group1') == 'group.read'
 

	
 
        # set default permission to none
 
        RepoGroupModel().grant_user_permission(repo_group=self.g1,
 
@@ -251,10 +251,10 @@ class TestPermissions(base.TestControlle
 

	
 
        # check his permissions
 
        a1_auth = AuthUser(user_id=self.anon.user_id)
 
        assert a1_auth.permissions['repositories_groups'].get('group1') == 'group.none'
 
        assert a1_auth.repository_group_permissions.get('group1') == 'group.none'
 

	
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        assert u1_auth.permissions['repositories_groups'].get('group1') == 'group.none'
 
        assert u1_auth.repository_group_permissions.get('group1') == 'group.none'
 

	
 
        # grant ug1 read permissions for
 
        RepoGroupModel().grant_user_group_permission(repo_group=self.g1,
 
@@ -270,10 +270,10 @@ class TestPermissions(base.TestControlle
 

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

	
 
        assert a1_auth.permissions['repositories_groups'].get('group1') == 'group.none'
 
        assert a1_auth.repository_group_permissions.get('group1') == 'group.none'
 

	
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        assert u1_auth.permissions['repositories_groups'].get('group1') == 'group.read'
 
        assert u1_auth.repository_group_permissions.get('group1') == 'group.read'
 

	
 
    def test_inherit_nice_permissions_from_default_user(self):
 
        user_model = UserModel()
 
@@ -286,7 +286,7 @@ class TestPermissions(base.TestControlle
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        # this user will have inherited permissions from default user
 
        assert u1_auth.permissions['global'] == set(['hg.create.repository', 'hg.fork.repository',
 
        assert u1_auth.global_permissions == set(['hg.create.repository', 'hg.fork.repository',
 
                              'hg.register.manual_activate',
 
                              'hg.extern_activate.auto',
 
                              'repository.read', 'group.read',
 
@@ -303,7 +303,7 @@ class TestPermissions(base.TestControlle
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        # this user will have inherited permissions from default user
 
        assert u1_auth.permissions['global'] == set(['hg.create.none', 'hg.fork.none',
 
        assert u1_auth.global_permissions == set(['hg.create.none', 'hg.fork.none',
 
                              'hg.register.manual_activate',
 
                              'hg.extern_activate.auto',
 
                              'repository.read', 'group.read',
 
@@ -327,7 +327,7 @@ class TestPermissions(base.TestControlle
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        # this user will have inherited more permissions from default user
 
        assert u1_auth.permissions['global'] == set([
 
        assert u1_auth.global_permissions == set([
 
                              'hg.create.repository',
 
                              'hg.fork.repository',
 
                              'hg.register.manual_activate',
 
@@ -353,7 +353,7 @@ class TestPermissions(base.TestControlle
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        # this user will have inherited less permissions from default user
 
        assert u1_auth.permissions['global'] == set([
 
        assert u1_auth.global_permissions == set([
 
                              'hg.create.repository',
 
                              'hg.fork.repository',
 
                              'hg.register.manual_activate',
 
@@ -386,7 +386,7 @@ class TestPermissions(base.TestControlle
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 

	
 
        assert u1_auth.permissions['global'] == set(['hg.create.none', 'hg.fork.none',
 
        assert u1_auth.global_permissions == set(['hg.create.none', 'hg.fork.none',
 
                              'hg.register.manual_activate',
 
                              'hg.extern_activate.auto',
 
                              'repository.read', 'group.read',
 
@@ -418,7 +418,7 @@ class TestPermissions(base.TestControlle
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 

	
 
        assert u1_auth.permissions['global'] == set(['hg.create.repository', 'hg.fork.repository',
 
        assert u1_auth.global_permissions == set(['hg.create.repository', 'hg.fork.repository',
 
                              'hg.register.manual_activate',
 
                              'hg.extern_activate.auto',
 
                              'repository.read', 'group.read',
 
@@ -447,7 +447,7 @@ class TestPermissions(base.TestControlle
 
                                          perm='repository.write')
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        assert u1_auth.permissions['repositories']['myownrepo'] == 'repository.write'
 
        assert u1_auth.repository_permissions['myownrepo'] == 'repository.write'
 

	
 
    def test_inactive_user_group_does_not_affect_repo_permissions_inverse(self):
 
        self.ug1 = fixture.create_user_group('G1')
 
@@ -471,7 +471,7 @@ class TestPermissions(base.TestControlle
 
                                          perm='repository.admin')
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        assert u1_auth.permissions['repositories']['myownrepo'] == 'repository.admin'
 
        assert u1_auth.repository_permissions['myownrepo'] == 'repository.admin'
 

	
 
    def test_inactive_user_group_does_not_affect_repo_group_permissions(self):
 
        self.ug1 = fixture.create_user_group('G1')
 
@@ -491,7 +491,7 @@ class TestPermissions(base.TestControlle
 
                                               perm='group.write')
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        assert u1_auth.permissions['repositories_groups'].get('group1') == 'group.write'
 
        assert u1_auth.repository_group_permissions.get('group1') == 'group.write'
 

	
 
    def test_inactive_user_group_does_not_affect_repo_group_permissions_inverse(self):
 
        self.ug1 = fixture.create_user_group('G1')
 
@@ -511,7 +511,7 @@ class TestPermissions(base.TestControlle
 
                                               perm='group.admin')
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        assert u1_auth.permissions['repositories_groups'].get('group1') == 'group.admin'
 
        assert u1_auth.repository_group_permissions.get('group1') == 'group.admin'
 

	
 
    def test_inactive_user_group_does_not_affect_user_group_permissions(self):
 
        self.ug1 = fixture.create_user_group('G1')
 
@@ -531,8 +531,8 @@ class TestPermissions(base.TestControlle
 
                                               perm='usergroup.write')
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        assert u1_auth.permissions['user_groups']['G1'] == 'usergroup.read'
 
        assert u1_auth.permissions['user_groups']['G2'] == 'usergroup.write'
 
        assert u1_auth.user_group_permissions['G1'] == 'usergroup.read'
 
        assert u1_auth.user_group_permissions['G2'] == 'usergroup.write'
 

	
 
    def test_inactive_user_group_does_not_affect_user_group_permissions_inverse(self):
 
        self.ug1 = fixture.create_user_group('G1')
 
@@ -552,8 +552,8 @@ class TestPermissions(base.TestControlle
 
                                               perm='usergroup.admin')
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        assert u1_auth.permissions['user_groups']['G1'] == 'usergroup.read'
 
        assert u1_auth.permissions['user_groups']['G2'] == 'usergroup.admin'
 
        assert u1_auth.user_group_permissions['G1'] == 'usergroup.read'
 
        assert u1_auth.user_group_permissions['G2'] == 'usergroup.admin'
 

	
 
    def test_owner_permissions_doesnot_get_overwritten_by_group(self):
 
        # create repo as USER,
 
@@ -563,7 +563,7 @@ class TestPermissions(base.TestControlle
 

	
 
        # he has permissions of admin as owner
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        assert u1_auth.permissions['repositories']['myownrepo'] == 'repository.admin'
 
        assert u1_auth.repository_permissions['myownrepo'] == 'repository.admin'
 
        # set his permission as user group, he should still be admin
 
        self.ug1 = fixture.create_user_group('G1')
 
        UserGroupModel().add_user_to_group(self.ug1, self.u1)
 
@@ -573,7 +573,7 @@ class TestPermissions(base.TestControlle
 

	
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        assert u1_auth.permissions['repositories']['myownrepo'] == 'repository.admin'
 
        assert u1_auth.repository_permissions['myownrepo'] == 'repository.admin'
 

	
 
    def test_owner_permissions_doesnot_get_overwritten_by_others(self):
 
        # create repo as USER,
 
@@ -583,13 +583,13 @@ class TestPermissions(base.TestControlle
 

	
 
        # he has permissions of admin as owner
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        assert u1_auth.permissions['repositories']['myownrepo'] == 'repository.admin'
 
        assert u1_auth.repository_permissions['myownrepo'] == 'repository.admin'
 
        # set his permission as user, he should still be admin
 
        RepoModel().grant_user_permission(self.test_repo, user=self.u1,
 
                                          perm='repository.none')
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        assert u1_auth.permissions['repositories']['myownrepo'] == 'repository.admin'
 
        assert u1_auth.repository_permissions['myownrepo'] == 'repository.admin'
 

	
 
    def _test_def_perm_equal(self, user, change_factor=0):
 
        perms = UserToPerm.query() \
0 comments (0 inline, 0 general)