Changeset - c96e05599877
Parent rev.
Child rev.
[Not reviewed]
default
0 2 0
Mads Kiilerich - 9 years ago 2016-09-12 17:41:19
madski@unity3d.com
api: stop explicitly passing apiuser to auth methods - use the global user instead
2 files changed with 58 insertions and 72 deletions:
kallithea/controllers/api/api.py
50
57
kallithea/lib/auth.py
8
15
0 comments (0 inline, 0 general)
kallithea/controllers/api/api.py
➞
Show inline comments
 
@@ -279,17 +279,17 @@ class ApiController(JSONRPCController):
 
          error :  {
 
            'Error occurred during cache invalidation action'
 
          }
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        repo = get_repo_or_error(repoid)

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')(user=apiuser):

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')():

			




	
	
	
		
 
            # check if we have admin permission for this repo !

			




	
	
	
		
 
            if not HasRepoPermissionAnyApi('repository.admin',

			




	
	
	
		
 
                                           'repository.write')(

			




	
	
	
		
 
                    user=apiuser, repo_name=repo.repo_name):

			




	
	
	
		
 
                    repo_name=repo.repo_name):

			




	
	
	
		
 
                raise JSONRPCError('repository `%s` does not exist' % (repoid,))

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            ScmModel().mark_for_invalidation(repo.repo_name)

			




	
	
	
		
 
            return dict(

			




	
	
	
		
 
                msg='Cache for repository `%s` was invalidated' % (repoid,),

			




	
	
		
 
@@ -345,17 +345,16 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
          error :  {

			




	
	
	
		
 
            'Error occurred locking repository `<reponame>`

			




	
	
	
		
 
          }

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        repo = get_repo_or_error(repoid)

			




	
	
	
		
 
        if HasPermissionAnyApi('hg.admin')(user=apiuser):

			




	
	
	
		
 
        if HasPermissionAnyApi('hg.admin')():

			




	
	
	
		
 
            pass

			




	
	
	
		
 
        elif HasRepoPermissionAnyApi('repository.admin',

			




	
	
	
		
 
                                     'repository.write')(user=apiuser,

			




	
	
	
		
 
                                                         repo_name=repo.repo_name):

			




	
	
	
		
 
                                     'repository.write')(repo_name=repo.repo_name):

			




	
	
	
		
 
            # make sure normal user does not pass someone else userid,

			




	
	
	
		
 
            # he is not allowed to do that

			




	
	
	
		
 
            if not isinstance(userid, Optional) and userid != apiuser.user_id:

			




	
	
	
		
 
                raise JSONRPCError(

			




	
	
	
		
 
                    'userid is not the same as your user'

			




	
	
	
		
 
                )

			




	
	
		
 
@@ -438,13 +437,13 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
          result : {

			




	
	
	
		
 
            [repo_object, repo_object,...]

			




	
	
	
		
 
          }

			




	
	
	
		
 
          error :  null

			




	
	
	
		
 
        """

			




	
	
	
		
 

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')(user=apiuser):

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')():

			




	
	
	
		
 
            # make sure normal user does not pass someone else userid,

			




	
	
	
		
 
            # he is not allowed to do that

			




	
	
	
		
 
            if not isinstance(userid, Optional) and userid != apiuser.user_id:

			




	
	
	
		
 
                raise JSONRPCError(

			




	
	
	
		
 
                    'userid is not the same as your user'

			




	
	
	
		
 
                )

			




	
	
		
 
@@ -572,13 +571,13 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
                         },

			




	
	
	
		
 
                    }

			




	
	
	
		
 

			




	
	
	
		
 
            error:  null

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')(user=apiuser):

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')():

			




	
	
	
		
 
            # make sure normal user does not pass someone else userid,

			




	
	
	
		
 
            # he is not allowed to do that

			




	
	
	
		
 
            if not isinstance(userid, Optional) and userid != apiuser.user_id:

			




	
	
	
		
 
                raise JSONRPCError(

			




	
	
	
		
 
                    'userid is not the same as your user'

			




	
	
	
		
 
                )

			




	
	
		
 
@@ -847,17 +846,17 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
                       "members" :  [<user_obj>,...]

			




	
	
	
		
 
                     }

			




	
	
	
		
 
            error : null

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        user_group = get_user_group_or_error(usergroupid)

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')(user=apiuser):

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')():

			




	
	
	
		
 
            # check if we have at least read permission for this user group !

			




	
	
	
		
 
            _perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)

			




	
	
	
		
 
            if not HasUserGroupPermissionAny(*_perms)(

			




	
	
	
		
 
                    user=apiuser, user_group_name=user_group.users_group_name):

			




	
	
	
		
 
                    user_group_name=user_group.users_group_name):

			




	
	
	
		
 
                raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))

			




	
	
	
		
 

			




	
	
	
		
 
        data = user_group.get_api_data()

			




	
	
	
		
 
        return data

			




	
	
	
		
 

			




	
	
	
		
 
    # permission check inside

			




	
	
		
 
@@ -876,15 +875,14 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
            result : [<user_group_obj>,...]

			




	
	
	
		
 
            error : null

			




	
	
	
		
 
        """

			




	
	
	
		
 

			




	
	
	
		
 
        result = []

			




	
	
	
		
 
        _perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)

			




	
	
	
		
 
        extras = {'user': apiuser}

			




	
	
	
		
 
        for user_group in UserGroupList(UserGroup.get_all(),

			




	
	
	
		
 
                                        perm_set=_perms, extra_kwargs=extras):

			




	
	
	
		
 
                                        perm_set=_perms):

			




	
	
	
		
 
            result.append(user_group.get_api_data())

			




	
	
	
		
 
        return result

			




	
	
	
		
 

			




	
	
	
		
 
    @HasPermissionAnyDecorator('hg.admin', 'hg.usergroup.create.true')

			




	
	
	
		
 
    def create_user_group(self, apiuser, group_name, description=Optional(''),

			




	
	
	
		
 
                          owner=Optional(OAttr('apiuser')), active=Optional(True)):

			




	
	
		
 
@@ -983,17 +981,17 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
          error :  {

			




	
	
	
		
 
            "failed to update user group `<user group name>`"

			




	
	
	
		
 
          }

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        user_group = get_user_group_or_error(usergroupid)

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')(user=apiuser):

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')():

			




	
	
	
		
 
            # check if we have admin permission for this user group !

			




	
	
	
		
 
            _perms = ('usergroup.admin',)

			




	
	
	
		
 
            if not HasUserGroupPermissionAny(*_perms)(

			




	
	
	
		
 
                    user=apiuser, user_group_name=user_group.users_group_name):

			




	
	
	
		
 
                    user_group_name=user_group.users_group_name):

			




	
	
	
		
 
                raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))

			




	
	
	
		
 

			




	
	
	
		
 
        if not isinstance(owner, Optional):

			




	
	
	
		
 
            owner = get_user_or_error(owner)

			




	
	
	
		
 

			




	
	
	
		
 
        updates = {}

			




	
	
		
 
@@ -1042,17 +1040,17 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
            or

			




	
	
	
		
 
            "RepoGroup assigned to <repo_groups_list>"

			




	
	
	
		
 
          }

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        user_group = get_user_group_or_error(usergroupid)

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')(user=apiuser):

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')():

			




	
	
	
		
 
            # check if we have admin permission for this user group !

			




	
	
	
		
 
            _perms = ('usergroup.admin',)

			




	
	
	
		
 
            if not HasUserGroupPermissionAny(*_perms)(

			




	
	
	
		
 
                    user=apiuser, user_group_name=user_group.users_group_name):

			




	
	
	
		
 
                    user_group_name=user_group.users_group_name):

			




	
	
	
		
 
                raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            UserGroupModel().delete(user_group)

			




	
	
	
		
 
            Session().commit()

			




	
	
	
		
 
            return dict(

			




	
	
		
 
@@ -1103,17 +1101,17 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
            "failed to add member to user group `<user_group_name>`"

			




	
	
	
		
 
          }

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        user = get_user_or_error(userid)

			




	
	
	
		
 
        user_group = get_user_group_or_error(usergroupid)

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')(user=apiuser):

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')():

			




	
	
	
		
 
            # check if we have admin permission for this user group !

			




	
	
	
		
 
            _perms = ('usergroup.admin',)

			




	
	
	
		
 
            if not HasUserGroupPermissionAny(*_perms)(

			




	
	
	
		
 
                    user=apiuser, user_group_name=user_group.users_group_name):

			




	
	
	
		
 
                    user_group_name=user_group.users_group_name):

			




	
	
	
		
 
                raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            ugm = UserGroupModel().add_user_to_group(user_group, user)

			




	
	
	
		
 
            success = True if ugm != True else False

			




	
	
	
		
 
            msg = 'added member `%s` to user group `%s`' % (

			




	
	
		
 
@@ -1157,17 +1155,17 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
                    }

			




	
	
	
		
 
            error:  null

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        user = get_user_or_error(userid)

			




	
	
	
		
 
        user_group = get_user_group_or_error(usergroupid)

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')(user=apiuser):

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')():

			




	
	
	
		
 
            # check if we have admin permission for this user group !

			




	
	
	
		
 
            _perms = ('usergroup.admin',)

			




	
	
	
		
 
            if not HasUserGroupPermissionAny(*_perms)(

			




	
	
	
		
 
                    user=apiuser, user_group_name=user_group.users_group_name):

			




	
	
	
		
 
                    user_group_name=user_group.users_group_name):

			




	
	
	
		
 
                raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            success = UserGroupModel().remove_user_from_group(user_group, user)

			




	
	
	
		
 
            msg = 'removed member `%s` from user group `%s`' % (

			




	
	
	
		
 
                user.username, user_group.users_group_name

			




	
	
		
 
@@ -1242,16 +1240,16 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
          }

			




	
	
	
		
 
          error :  null

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        repo = get_repo_or_error(repoid)

			




	
	
	
		
 

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')(user=apiuser):

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')():

			




	
	
	
		
 
            # check if we have admin permission for this repo !

			




	
	
	
		
 
            perms = ('repository.admin', 'repository.write', 'repository.read')

			




	
	
	
		
 
            if not HasRepoPermissionAnyApi(*perms)(user=apiuser, repo_name=repo.repo_name):

			




	
	
	
		
 
            if not HasRepoPermissionAnyApi(*perms)(repo_name=repo.repo_name):

			




	
	
	
		
 
                raise JSONRPCError('repository `%s` does not exist' % (repoid,))

			




	
	
	
		
 

			




	
	
	
		
 
        members = []

			




	
	
	
		
 
        followers = []

			




	
	
	
		
 
        for user in repo.repo_to_perm:

			




	
	
	
		
 
            perm = user.permission.permission_name

			




	
	
		
 
@@ -1312,13 +1310,13 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
                      },

			




	
	
	
		
 
                      …

			




	
	
	
		
 
                    ]

			




	
	
	
		
 
            error:  null

			




	
	
	
		
 
        """

			




	
	
	
		
 
        result = []

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')(user=apiuser):

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')():

			




	
	
	
		
 
            repos = RepoModel().get_all_user_repos(user=apiuser)

			




	
	
	
		
 
        else:

			




	
	
	
		
 
            repos = Repository.get_all()

			




	
	
	
		
 

			




	
	
	
		
 
        for repo in repos:

			




	
	
	
		
 
            result.append(repo.get_api_data())

			




	
	
		
 
@@ -1356,16 +1354,16 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
                      …

			




	
	
	
		
 
                    ]

			




	
	
	
		
 
            error:  null

			




	
	
	
		
 
        """

			




	
	
	
		
 
        repo = get_repo_or_error(repoid)

			




	
	
	
		
 

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')(user=apiuser):

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')():

			




	
	
	
		
 
            # check if we have admin permission for this repo !

			




	
	
	
		
 
            perms = ('repository.admin', 'repository.write', 'repository.read')

			




	
	
	
		
 
            if not HasRepoPermissionAnyApi(*perms)(user=apiuser, repo_name=repo.repo_name):

			




	
	
	
		
 
            if not HasRepoPermissionAnyApi(*perms)(repo_name=repo.repo_name):

			




	
	
	
		
 
                raise JSONRPCError('repository `%s` does not exist' % (repoid,))

			




	
	
	
		
 

			




	
	
	
		
 
        ret_type = Optional.extract(ret_type)

			




	
	
	
		
 
        _map = {}

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            _d, _f = ScmModel().get_nodes(repo, revision, root_path,

			




	
	
		
 
@@ -1444,13 +1442,13 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
          result : null

			




	
	
	
		
 
          error :  {

			




	
	
	
		
 
             'failed to create repository `<repo_name>`

			




	
	
	
		
 
          }

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')(user=apiuser):

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')():

			




	
	
	
		
 
            if not isinstance(owner, Optional):

			




	
	
	
		
 
                # forbid setting owner for non-admins

			




	
	
	
		
 
                raise JSONRPCError(

			




	
	
	
		
 
                    'Only Kallithea admin can specify `owner` param'

			




	
	
	
		
 
                )

			




	
	
	
		
 
        if isinstance(owner, Optional):

			




	
	
		
 
@@ -1538,20 +1536,19 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
        :param landing_rev:

			




	
	
	
		
 
        :param enable_statistics:

			




	
	
	
		
 
        :param enable_locking:

			




	
	
	
		
 
        :param enable_downloads:

			




	
	
	
		
 
        """

			




	
	
	
		
 
        repo = get_repo_or_error(repoid)

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')(user=apiuser):

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')():

			




	
	
	
		
 
            # check if we have admin permission for this repo !

			




	
	
	
		
 
            if not HasRepoPermissionAnyApi('repository.admin')(user=apiuser,

			




	
	
	
		
 
                                                               repo_name=repo.repo_name):

			




	
	
	
		
 
            if not HasRepoPermissionAnyApi('repository.admin')(repo_name=repo.repo_name):

			




	
	
	
		
 
                raise JSONRPCError('repository `%s` does not exist' % (repoid,))

			




	
	
	
		
 

			




	
	
	
		
 
            if (name != repo.repo_name and

			




	
	
	
		
 
                not HasPermissionAnyApi('hg.create.repository')(user=apiuser)

			




	
	
	
		
 
                not HasPermissionAnyApi('hg.create.repository')()

			




	
	
	
		
 
                ):

			




	
	
	
		
 
                raise JSONRPCError('no permission to create (or move) repositories')

			




	
	
	
		
 

			




	
	
	
		
 
            if not isinstance(owner, Optional):

			




	
	
	
		
 
                # forbid setting owner for non-admins

			




	
	
	
		
 
                raise JSONRPCError(

			




	
	
		
 
@@ -1638,25 +1635,24 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 

			




	
	
	
		
 
        _repo = RepoModel().get_by_repo_name(fork_name)

			




	
	
	
		
 
        if _repo:

			




	
	
	
		
 
            type_ = 'fork' if _repo.fork else 'repo'

			




	
	
	
		
 
            raise JSONRPCError("%s `%s` already exist" % (type_, fork_name))

			




	
	
	
		
 

			




	
	
	
		
 
        if HasPermissionAnyApi('hg.admin')(user=apiuser):

			




	
	
	
		
 
        if HasPermissionAnyApi('hg.admin')():

			




	
	
	
		
 
            pass

			




	
	
	
		
 
        elif HasRepoPermissionAnyApi('repository.admin',

			




	
	
	
		
 
                                     'repository.write',

			




	
	
	
		
 
                                     'repository.read')(user=apiuser,

			




	
	
	
		
 
                                                        repo_name=repo.repo_name):

			




	
	
	
		
 
                                     'repository.read')(repo_name=repo.repo_name):

			




	
	
	
		
 
            if not isinstance(owner, Optional):

			




	
	
	
		
 
                # forbid setting owner for non-admins

			




	
	
	
		
 
                raise JSONRPCError(

			




	
	
	
		
 
                    'Only Kallithea admin can specify `owner` param'

			




	
	
	
		
 
                )

			




	
	
	
		
 

			




	
	
	
		
 
            if not HasPermissionAnyApi('hg.create.repository')(user=apiuser):

			




	
	
	
		
 
            if not HasPermissionAnyApi('hg.create.repository')():

			




	
	
	
		
 
                raise JSONRPCError('no permission to create repositories')

			




	
	
	
		
 
        else:

			




	
	
	
		
 
            raise JSONRPCError('repository `%s` does not exist' % (repoid,))

			




	
	
	
		
 

			




	
	
	
		
 
        if isinstance(owner, Optional):

			




	
	
	
		
 
            owner = apiuser.user_id

			




	
	
		
 
@@ -1721,16 +1717,15 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
                    }

			




	
	
	
		
 
            error:  null

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        repo = get_repo_or_error(repoid)

			




	
	
	
		
 

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')(user=apiuser):

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')():

			




	
	
	
		
 
            # check if we have admin permission for this repo !

			




	
	
	
		
 
            if not HasRepoPermissionAnyApi('repository.admin')(user=apiuser,

			




	
	
	
		
 
                                                               repo_name=repo.repo_name):

			




	
	
	
		
 
            if not HasRepoPermissionAnyApi('repository.admin')(repo_name=repo.repo_name):

			




	
	
	
		
 
                raise JSONRPCError('repository `%s` does not exist' % (repoid,))

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            handle_forks = Optional.extract(forks)

			




	
	
	
		
 
            _forks_msg = ''

			




	
	
	
		
 
            _forks = [f for f in repo.forks]

			




	
	
		
 
@@ -1880,23 +1875,23 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
          }

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        repo = get_repo_or_error(repoid)

			




	
	
	
		
 
        perm = get_perm_or_error(perm)

			




	
	
	
		
 
        user_group = get_user_group_or_error(usergroupid)

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')(user=apiuser):

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')():

			




	
	
	
		
 
            # check if we have admin permission for this repo !

			




	
	
	
		
 
            _perms = ('repository.admin',)

			




	
	
	
		
 
            if not HasRepoPermissionAnyApi(*_perms)(

			




	
	
	
		
 
                    user=apiuser, repo_name=repo.repo_name):

			




	
	
	
		
 
                    repo_name=repo.repo_name):

			




	
	
	
		
 
                raise JSONRPCError('repository `%s` does not exist' % (repoid,))

			




	
	
	
		
 

			




	
	
	
		
 
            # check if we have at least read permission for this user group !

			




	
	
	
		
 
            _perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)

			




	
	
	
		
 
            if not HasUserGroupPermissionAny(*_perms)(

			




	
	
	
		
 
                    user=apiuser, user_group_name=user_group.users_group_name):

			




	
	
	
		
 
                    user_group_name=user_group.users_group_name):

			




	
	
	
		
 
                raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            RepoModel().grant_user_group_permission(

			




	
	
	
		
 
                repo=repo, group_name=user_group, perm=perm)

			




	
	
	
		
 

			




	
	
		
 
@@ -1938,23 +1933,23 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
                      "success": true

			




	
	
	
		
 
                    }

			




	
	
	
		
 
            error:  null

			




	
	
	
		
 
        """

			




	
	
	
		
 
        repo = get_repo_or_error(repoid)

			




	
	
	
		
 
        user_group = get_user_group_or_error(usergroupid)

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')(user=apiuser):

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')():

			




	
	
	
		
 
            # check if we have admin permission for this repo !

			




	
	
	
		
 
            _perms = ('repository.admin',)

			




	
	
	
		
 
            if not HasRepoPermissionAnyApi(*_perms)(

			




	
	
	
		
 
                    user=apiuser, repo_name=repo.repo_name):

			




	
	
	
		
 
                    repo_name=repo.repo_name):

			




	
	
	
		
 
                raise JSONRPCError('repository `%s` does not exist' % (repoid,))

			




	
	
	
		
 

			




	
	
	
		
 
            # check if we have at least read permission for this user group !

			




	
	
	
		
 
            _perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)

			




	
	
	
		
 
            if not HasUserGroupPermissionAny(*_perms)(

			




	
	
	
		
 
                    user=apiuser, user_group_name=user_group.users_group_name):

			




	
	
	
		
 
                    user_group_name=user_group.users_group_name):

			




	
	
	
		
 
                raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            RepoModel().revoke_user_group_permission(

			




	
	
	
		
 
                repo=repo, group_name=user_group)

			




	
	
	
		
 

			




	
	
		
 
@@ -2200,16 +2195,15 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
          }

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 

			




	
	
	
		
 
        repo_group = get_repo_group_or_error(repogroupid)

			




	
	
	
		
 

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')(user=apiuser):

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')():

			




	
	
	
		
 
            # check if we have admin permission for this repo group !

			




	
	
	
		
 
            if not HasRepoGroupPermissionAnyApi('group.admin')(user=apiuser,

			




	
	
	
		
 
                                                               group_name=repo_group.group_name):

			




	
	
	
		
 
            if not HasRepoGroupPermissionAnyApi('group.admin')(group_name=repo_group.group_name):

			




	
	
	
		
 
                raise JSONRPCError('repository group `%s` does not exist' % (repogroupid,))

			




	
	
	
		
 

			




	
	
	
		
 
        user = get_user_or_error(userid)

			




	
	
	
		
 
        perm = get_perm_or_error(perm, prefix='group.')

			




	
	
	
		
 
        apply_to_children = Optional.extract(apply_to_children)

			




	
	
	
		
 

			




	
	
		
 
@@ -2267,16 +2261,15 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
          }

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 

			




	
	
	
		
 
        repo_group = get_repo_group_or_error(repogroupid)

			




	
	
	
		
 

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')(user=apiuser):

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')():

			




	
	
	
		
 
            # check if we have admin permission for this repo group !

			




	
	
	
		
 
            if not HasRepoGroupPermissionAnyApi('group.admin')(user=apiuser,

			




	
	
	
		
 
                                                               group_name=repo_group.group_name):

			




	
	
	
		
 
            if not HasRepoGroupPermissionAnyApi('group.admin')(group_name=repo_group.group_name):

			




	
	
	
		
 
                raise JSONRPCError('repository group `%s` does not exist' % (repogroupid,))

			




	
	
	
		
 

			




	
	
	
		
 
        user = get_user_or_error(userid)

			




	
	
	
		
 
        apply_to_children = Optional.extract(apply_to_children)

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
		
 
@@ -2338,24 +2331,24 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
          }

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        repo_group = get_repo_group_or_error(repogroupid)

			




	
	
	
		
 
        perm = get_perm_or_error(perm, prefix='group.')

			




	
	
	
		
 
        user_group = get_user_group_or_error(usergroupid)

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')(user=apiuser):

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')():

			




	
	
	
		
 
            # check if we have admin permission for this repo group !

			




	
	
	
		
 
            _perms = ('group.admin',)

			




	
	
	
		
 
            if not HasRepoGroupPermissionAnyApi(*_perms)(

			




	
	
	
		
 
                    user=apiuser, group_name=repo_group.group_name):

			




	
	
	
		
 
                    group_name=repo_group.group_name):

			




	
	
	
		
 
                raise JSONRPCError(

			




	
	
	
		
 
                    'repository group `%s` does not exist' % (repogroupid,))

			




	
	
	
		
 

			




	
	
	
		
 
            # check if we have at least read permission for this user group !

			




	
	
	
		
 
            _perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)

			




	
	
	
		
 
            if not HasUserGroupPermissionAny(*_perms)(

			




	
	
	
		
 
                    user=apiuser, user_group_name=user_group.users_group_name):

			




	
	
	
		
 
                    user_group_name=user_group.users_group_name):

			




	
	
	
		
 
                raise JSONRPCError(

			




	
	
	
		
 
                    'user group `%s` does not exist' % (usergroupid,))

			




	
	
	
		
 

			




	
	
	
		
 
        apply_to_children = Optional.extract(apply_to_children)

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
		
 
@@ -2416,24 +2409,24 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
          }

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        repo_group = get_repo_group_or_error(repogroupid)

			




	
	
	
		
 
        user_group = get_user_group_or_error(usergroupid)

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')(user=apiuser):

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')():

			




	
	
	
		
 
            # check if we have admin permission for this repo group !

			




	
	
	
		
 
            _perms = ('group.admin',)

			




	
	
	
		
 
            if not HasRepoGroupPermissionAnyApi(*_perms)(

			




	
	
	
		
 
                    user=apiuser, group_name=repo_group.group_name):

			




	
	
	
		
 
                    group_name=repo_group.group_name):

			




	
	
	
		
 
                raise JSONRPCError(

			




	
	
	
		
 
                    'repository group `%s` does not exist' % (repogroupid,))

			




	
	
	
		
 

			




	
	
	
		
 
            # check if we have at least read permission for this user group !

			




	
	
	
		
 
            _perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)

			




	
	
	
		
 
            if not HasUserGroupPermissionAny(*_perms)(

			




	
	
	
		
 
                    user=apiuser, user_group_name=user_group.users_group_name):

			




	
	
	
		
 
                    user_group_name=user_group.users_group_name):

			




	
	
	
		
 
                raise JSONRPCError(

			




	
	
	
		
 
                    'user group `%s` does not exist' % (usergroupid,))

			




	
	
	
		
 

			




	
	
	
		
 
        apply_to_children = Optional.extract(apply_to_children)

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
		
 
@@ -2463,13 +2456,13 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
        :param apiuser: filled automatically from apikey

			




	
	
	
		
 
        :type apiuser: AuthUser

			




	
	
	
		
 
        :param gistid: id of private or public gist

			




	
	
	
		
 
        :type gistid: str

			




	
	
	
		
 
        """

			




	
	
	
		
 
        gist = get_gist_or_error(gistid)

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')(user=apiuser):

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')():

			




	
	
	
		
 
            if gist.gist_owner != apiuser.user_id:

			




	
	
	
		
 
                raise JSONRPCError('gist `%s` does not exist' % (gistid,))

			




	
	
	
		
 
        return gist.get_api_data()

			




	
	
	
		
 

			




	
	
	
		
 
    def get_gists(self, apiuser, userid=Optional(OAttr('apiuser'))):

			




	
	
	
		
 
        """

			




	
	
		
 
@@ -2478,13 +2471,13 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 

			




	
	
	
		
 
        :param apiuser: filled automatically from apikey

			




	
	
	
		
 
        :type apiuser: AuthUser

			




	
	
	
		
 
        :param userid: user to get gists for

			




	
	
	
		
 
        :type userid: Optional(str or int)

			




	
	
	
		
 
        """

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')(user=apiuser):

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')():

			




	
	
	
		
 
            # make sure normal user does not pass someone else userid,

			




	
	
	
		
 
            # he is not allowed to do that

			




	
	
	
		
 
            if not isinstance(userid, Optional) and userid != apiuser.user_id:

			




	
	
	
		
 
                raise JSONRPCError(

			




	
	
	
		
 
                    'userid is not the same as your user'

			




	
	
	
		
 
                )

			




	
	
		
 
@@ -2598,13 +2591,13 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
          error :  {

			




	
	
	
		
 
            "failed to delete gist ID:<gist_id>"

			




	
	
	
		
 
          }

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        gist = get_gist_or_error(gistid)

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')(user=apiuser):

			




	
	
	
		
 
        if not HasPermissionAnyApi('hg.admin')():

			




	
	
	
		
 
            if gist.gist_owner != apiuser.user_id:

			




	
	
	
		
 
                raise JSONRPCError('gist `%s` does not exist' % (gistid,))

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            GistModel().delete(gist)

			




	
	
	
		
 
            Session().commit()

			




        

    


    
    

    

        

                

                    kallithea/lib/auth.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
		
 
@@ -936,16 +936,13 @@ class PermsFunction(object):

			




	
	
	
		
 
        """ Defend against accidentally forgetting to call the object

			




	
	
	
		
 
            and instead evaluating it directly in a boolean context,

			




	
	
	
		
 
            which could have security implications.

			




	
	
	
		
 
        """

			




	
	
	
		
 
        raise AssertionError(self.__class__.__name__ + ' is not a bool and must be called!')

			




	
	
	
		
 

			




	
	
	
		
 
    def __call__(self, check_location='unspecified location', user=None):

			




	
	
	
		
 
        if user:

			




	
	
	
		
 
            assert user.user_id == request.user.user_id, (user, request.user)

			




	
	
	
		
 

			




	
	
	
		
 
    def __call__(self, check_location='unspecified location'):

			




	
	
	
		
 
        user = request.user

			




	
	
	
		
 
        assert user

			




	
	
	
		
 
        assert isinstance(user, AuthUser), user

			




	
	
	
		
 

			




	
	
	
		
 
        cls_name = self.__class__.__name__

			




	
	
	
		
 
        check_scope = self._scope()

			




	
	
		
 
@@ -973,15 +970,15 @@ class HasPermissionAny(PermsFunction):

			




	
	
	
		
 
        if self.required_perms.intersection(self.user_perms.get('global')):

			




	
	
	
		
 
            return True

			




	
	
	
		
 
        return False

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class HasRepoPermissionAny(PermsFunction):

			




	
	
	
		
 
    def __call__(self, repo_name=None, check_location='', user=None):

			




	
	
	
		
 
    def __call__(self, repo_name=None, check_location=''):

			




	
	
	
		
 
        self.repo_name = repo_name

			




	
	
	
		
 
        return super(HasRepoPermissionAny, self).__call__(check_location, user)

			




	
	
	
		
 
        return super(HasRepoPermissionAny, self).__call__(check_location)

			




	
	
	
		
 

			




	
	
	
		
 
    def check_permissions(self):

			




	
	
	
		
 
        if not self.repo_name:

			




	
	
	
		
 
            self.repo_name = get_repo_slug(request)

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
		
 
@@ -996,15 +993,15 @@ class HasRepoPermissionAny(PermsFunction

			




	
	
	
		
 

			




	
	
	
		
 
    def _scope(self):

			




	
	
	
		
 
        return 'repo:%s' % self.repo_name

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class HasRepoGroupPermissionAny(PermsFunction):

			




	
	
	
		
 
    def __call__(self, group_name=None, check_location='', user=None):

			




	
	
	
		
 
    def __call__(self, group_name=None, check_location=''):

			




	
	
	
		
 
        self.group_name = group_name

			




	
	
	
		
 
        return super(HasRepoGroupPermissionAny, self).__call__(check_location, user)

			




	
	
	
		
 
        return super(HasRepoGroupPermissionAny, self).__call__(check_location)

			




	
	
	
		
 

			




	
	
	
		
 
    def check_permissions(self):

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            self._user_perms = set(

			




	
	
	
		
 
                [self.user_perms['repositories_groups'][self.group_name]]

			




	
	
	
		
 
            )

			




	
	
		
 
@@ -1016,15 +1013,15 @@ class HasRepoGroupPermissionAny(PermsFun

			




	
	
	
		
 

			




	
	
	
		
 
    def _scope(self):

			




	
	
	
		
 
        return 'repogroup:%s' % self.group_name

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class HasUserGroupPermissionAny(PermsFunction):

			




	
	
	
		
 
    def __call__(self, user_group_name=None, check_location='', user=None):

			




	
	
	
		
 
    def __call__(self, user_group_name=None, check_location=''):

			




	
	
	
		
 
        self.user_group_name = user_group_name

			




	
	
	
		
 
        return super(HasUserGroupPermissionAny, self).__call__(check_location, user)

			




	
	
	
		
 
        return super(HasUserGroupPermissionAny, self).__call__(check_location)

			




	
	
	
		
 

			




	
	
	
		
 
    def check_permissions(self):

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            self._user_perms = set(

			




	
	
	
		
 
                [self.user_perms['user_groups'][self.user_group_name]]

			




	
	
	
		
 
            )

			




	
	
		
 
@@ -1072,17 +1069,13 @@ class HasPermissionAnyMiddleware(object)

			




	
	
	
		
 
# SPECIAL VERSION TO HANDLE API AUTH

			




	
	
	
		
 
#==============================================================================

			




	
	
	
		
 
class _BaseApiPerm(object):

			




	
	
	
		
 
    def __init__(self, *perms):

			




	
	
	
		
 
        self.required_perms = set(perms)

			




	
	
	
		
 

			




	
	
	
		
 
    def __call__(self, check_location=None, user=None, repo_name=None,

			




	
	
	
		
 
                 group_name=None):

			




	
	
	
		
 
        assert user

			




	
	
	
		
 
        assert user.user_id == request.user.user_id, (user, request.user)

			




	
	
	
		
 

			




	
	
	
		
 
    def __call__(self, check_location=None, repo_name=None, group_name=None):

			




	
	
	
		
 
        user = request.user

			




	
	
	
		
 
        assert user

			




	
	
	
		
 
        assert isinstance(user, AuthUser), user

			




	
	
	
		
 

			




	
	
	
		
 
        cls_name = self.__class__.__name__

			




	
	
	
		
 
        check_scope = 'user:%s' % (user)

			




        

    



    


    



    



      

      





    
    0 comments (0 inline, 0 general)
    





    


  

  







  


    




    








    
        
    
    
        This site is powered by
            Kallithea 0.7.0,
        which is
        © 2010–2025 by various authors & licensed under GPLv3.