Changeset - 591effa1fc4d
Parent rev.
Child rev.
[Not reviewed]
default
0 2 0
Mads Kiilerich - 9 years ago 2016-09-12 17:41:20
madski@unity3d.com
api: drop the old Api auth methods and use the normal methods for access control
2 files changed with 46 insertions and 124 deletions:
kallithea/controllers/api/api.py
46
46
kallithea/lib/auth.py
78
0 comments (0 inline, 0 general)
kallithea/controllers/api/api.py
➞
Show inline comments
 
@@ -30,14 +30,14 @@ import traceback
 
import logging
 
from sqlalchemy import or_
 

			




	
	
	
		
 
from kallithea.controllers.api import JSONRPCController, JSONRPCError

			




	
	
	
		
 
from kallithea.lib.auth import (

			




	
	
	
		
 
    PasswordGenerator, AuthUser, HasPermissionAnyDecorator,

			




	
	
	
		
 
    HasPermissionAnyDecorator, HasPermissionAnyApi, HasRepoPermissionAnyApi,

			




	
	
	
		
 
    HasRepoGroupPermissionAnyApi, HasUserGroupPermissionAny)

			




	
	
	
		
 
    HasPermissionAnyDecorator, HasPermissionAny, HasRepoPermissionAny,

			




	
	
	
		
 
    HasRepoGroupPermissionAny, HasUserGroupPermissionAny)

			




	
	
	
		
 
from kallithea.lib.utils import map_groups, repo2db_mapper

			




	
	
	
		
 
from kallithea.lib.utils2 import (

			




	
	
	
		
 
    str2bool, time_to_datetime, safe_int, Optional, OAttr)

			




	
	
	
		
 
from kallithea.model.meta import Session

			




	
	
	
		
 
from kallithea.model.repo_group import RepoGroupModel

			




	
	
	
		
 
from kallithea.model.scm import ScmModel, UserGroupList

			




	
	
		
 
@@ -271,16 +271,16 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
          error :  {

			




	
	
	
		
 
            'Error occurred during cache invalidation action'

			




	
	
	
		
 
          }

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        repo = get_repo_or_error(repoid)

			




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

			




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

			




	
	
	
		
 
            # check if we have admin permission for this repo !

			




	
	
	
		
 
            if not HasRepoPermissionAnyApi('repository.admin',

			




	
	
	
		
 
                                           'repository.write')(

			




	
	
	
		
 
            if not HasRepoPermissionAny('repository.admin',

			




	
	
	
		
 
                                        'repository.write')(

			




	
	
	
		
 
                    repo_name=repo.repo_name):

			




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

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            ScmModel().mark_for_invalidation(repo.repo_name)

			




	
	
	
		
 
            return dict(

			




	
	
		
 
@@ -335,16 +335,16 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
          error :  {

			




	
	
	
		
 
            'Error occurred locking repository `<reponame>`

			




	
	
	
		
 
          }

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        repo = get_repo_or_error(repoid)

			




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

			




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

			




	
	
	
		
 
            pass

			




	
	
	
		
 
        elif HasRepoPermissionAnyApi('repository.admin',

			




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

			




	
	
	
		
 
        elif HasRepoPermissionAny('repository.admin',

			




	
	
	
		
 
                                  '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 != self.authuser.user_id:

			




	
	
	
		
 
                raise JSONRPCError(

			




	
	
	
		
 
                    'userid is not the same as your user'

			




	
	
	
		
 
                )

			




	
	
		
 
@@ -425,13 +425,13 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
          result : {

			




	
	
	
		
 
            [repo_object, repo_object,...]

			




	
	
	
		
 
          }

			




	
	
	
		
 
          error :  null

			




	
	
	
		
 
        """

			




	
	
	
		
 

			




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

			




	
	
	
		
 
        if not HasPermissionAny('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 != self.authuser.user_id:

			




	
	
	
		
 
                raise JSONRPCError(

			




	
	
	
		
 
                    'userid is not the same as your user'

			




	
	
	
		
 
                )

			




	
	
		
 
@@ -553,13 +553,13 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
                         },

			




	
	
	
		
 
                    }

			




	
	
	
		
 

			




	
	
	
		
 
            error:  null

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




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

			




	
	
	
		
 
        if not HasPermissionAny('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 != self.authuser.user_id:

			




	
	
	
		
 
                raise JSONRPCError(

			




	
	
	
		
 
                    'userid is not the same as your user'

			




	
	
	
		
 
                )

			




	
	
		
 
@@ -818,13 +818,13 @@ class ApiController(JSONRPCController):

			




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

			




	
	
	
		
 
                     }

			




	
	
	
		
 
            error : null

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        user_group = get_user_group_or_error(usergroupid)

			




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

			




	
	
	
		
 
        if not HasPermissionAny('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_group_name=user_group.users_group_name):

			




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

			




	
	
	
		
 

			




	
	
		
 
@@ -947,13 +947,13 @@ 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')():

			




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

			




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

			




	
	
	
		
 
            _perms = ('usergroup.admin',)

			




	
	
	
		
 
            if not HasUserGroupPermissionAny(*_perms)(

			




	
	
	
		
 
                    user_group_name=user_group.users_group_name):

			




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

			




	
	
	
		
 

			




	
	
		
 
@@ -1004,13 +1004,13 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
            or

			




	
	
	
		
 
            "RepoGroup assigned to <repo_groups_list>"

			




	
	
	
		
 
          }

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        user_group = get_user_group_or_error(usergroupid)

			




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

			




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

			




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

			




	
	
	
		
 
            _perms = ('usergroup.admin',)

			




	
	
	
		
 
            if not HasUserGroupPermissionAny(*_perms)(

			




	
	
	
		
 
                    user_group_name=user_group.users_group_name):

			




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

			




	
	
	
		
 

			




	
	
		
 
@@ -1063,13 +1063,13 @@ 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')():

			




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

			




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

			




	
	
	
		
 
            _perms = ('usergroup.admin',)

			




	
	
	
		
 
            if not HasUserGroupPermissionAny(*_perms)(

			




	
	
	
		
 
                    user_group_name=user_group.users_group_name):

			




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

			




	
	
	
		
 

			




	
	
		
 
@@ -1115,13 +1115,13 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
                    }

			




	
	
	
		
 
            error:  null

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        user = get_user_or_error(userid)

			




	
	
	
		
 
        user_group = get_user_group_or_error(usergroupid)

			




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

			




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

			




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

			




	
	
	
		
 
            _perms = ('usergroup.admin',)

			




	
	
	
		
 
            if not HasUserGroupPermissionAny(*_perms)(

			




	
	
	
		
 
                    user_group_name=user_group.users_group_name):

			




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

			




	
	
	
		
 

			




	
	
		
 
@@ -1198,16 +1198,16 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
          }

			




	
	
	
		
 
          error :  null

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        repo = get_repo_or_error(repoid)

			




	
	
	
		
 

			




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

			




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

			




	
	
	
		
 
            # check if we have admin permission for this repo !

			




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

			




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

			




	
	
	
		
 
            if not HasRepoPermissionAny(*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

			




	
	
		
 
@@ -1266,13 +1266,13 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
                      },

			




	
	
	
		
 
                      …

			




	
	
	
		
 
                    ]

			




	
	
	
		
 
            error:  null

			




	
	
	
		
 
        """

			




	
	
	
		
 
        result = []

			




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

			




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

			




	
	
	
		
 
            repos = RepoModel().get_all_user_repos(user=self.authuser.user_id)

			




	
	
	
		
 
        else:

			




	
	
	
		
 
            repos = Repository.get_all()

			




	
	
	
		
 

			




	
	
	
		
 
        for repo in repos:

			




	
	
	
		
 
            result.append(repo.get_api_data())

			




	
	
		
 
@@ -1308,16 +1308,16 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
                      …

			




	
	
	
		
 
                    ]

			




	
	
	
		
 
            error:  null

			




	
	
	
		
 
        """

			




	
	
	
		
 
        repo = get_repo_or_error(repoid)

			




	
	
	
		
 

			




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

			




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

			




	
	
	
		
 
            # check if we have admin permission for this repo !

			




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

			




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

			




	
	
	
		
 
            if not HasRepoPermissionAny(*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,

			




	
	
		
 
@@ -1394,13 +1394,13 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
          result : null

			




	
	
	
		
 
          error :  {

			




	
	
	
		
 
             'failed to create repository `<repo_name>`

			




	
	
	
		
 
          }

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




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

			




	
	
	
		
 
        if not HasPermissionAny('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):

			




	
	
		
 
@@ -1486,19 +1486,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')():

			




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

			




	
	
	
		
 
            # check if we have admin permission for this repo !

			




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

			




	
	
	
		
 
            if not HasRepoPermissionAny('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')()

			




	
	
	
		
 
                not HasPermissionAny('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(

			




	
	
		
 
@@ -1583,24 +1583,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')():

			




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

			




	
	
	
		
 
            pass

			




	
	
	
		
 
        elif HasRepoPermissionAnyApi('repository.admin',

			




	
	
	
		
 
                                     'repository.write',

			




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

			




	
	
	
		
 
        elif HasRepoPermissionAny('repository.admin',

			




	
	
	
		
 
                                  'repository.write',

			




	
	
	
		
 
                                  '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')():

			




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

			




	
	
	
		
 
                raise JSONRPCError('no permission to create repositories')

			




	
	
	
		
 
        else:

			




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

			




	
	
	
		
 

			




	
	
	
		
 
        if isinstance(owner, Optional):

			




	
	
	
		
 
            owner = self.authuser.user_id

			




	
	
		
 
@@ -1663,15 +1663,15 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
                    }

			




	
	
	
		
 
            error:  null

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        repo = get_repo_or_error(repoid)

			




	
	
	
		
 

			




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

			




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

			




	
	
	
		
 
            # check if we have admin permission for this repo !

			




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

			




	
	
	
		
 
            if not HasRepoPermissionAny('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]

			




	
	
		
 
@@ -1815,16 +1815,16 @@ 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')():

			




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

			




	
	
	
		
 
            # check if we have admin permission for this repo !

			




	
	
	
		
 
            _perms = ('repository.admin',)

			




	
	
	
		
 
            if not HasRepoPermissionAnyApi(*_perms)(

			




	
	
	
		
 
            if not HasRepoPermissionAny(*_perms)(

			




	
	
	
		
 
                    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)(

			




	
	
		
 
@@ -1871,16 +1871,16 @@ 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')():

			




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

			




	
	
	
		
 
            # check if we have admin permission for this repo !

			




	
	
	
		
 
            _perms = ('repository.admin',)

			




	
	
	
		
 
            if not HasRepoPermissionAnyApi(*_perms)(

			




	
	
	
		
 
            if not HasRepoPermissionAny(*_perms)(

			




	
	
	
		
 
                    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)(

			




	
	
		
 
@@ -2123,15 +2123,15 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
          }

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 

			




	
	
	
		
 
        repo_group = get_repo_group_or_error(repogroupid)

			




	
	
	
		
 

			




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

			




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

			




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

			




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

			




	
	
	
		
 
            if not HasRepoGroupPermissionAny('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)

			




	
	
	
		
 

			




	
	
		
 
@@ -2187,15 +2187,15 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
          }

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 

			




	
	
	
		
 
        repo_group = get_repo_group_or_error(repogroupid)

			




	
	
	
		
 

			




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

			




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

			




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

			




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

			




	
	
	
		
 
            if not HasRepoGroupPermissionAny('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:

			




	
	
		
 
@@ -2255,16 +2255,16 @@ 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')():

			




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

			




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

			




	
	
	
		
 
            _perms = ('group.admin',)

			




	
	
	
		
 
            if not HasRepoGroupPermissionAnyApi(*_perms)(

			




	
	
	
		
 
            if not HasRepoGroupPermissionAny(*_perms)(

			




	
	
	
		
 
                    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',)

			




	
	
		
 
@@ -2331,16 +2331,16 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
          }

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        repo_group = get_repo_group_or_error(repogroupid)

			




	
	
	
		
 
        user_group = get_user_group_or_error(usergroupid)

			




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

			




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

			




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

			




	
	
	
		
 
            _perms = ('group.admin',)

			




	
	
	
		
 
            if not HasRepoGroupPermissionAnyApi(*_perms)(

			




	
	
	
		
 
            if not HasRepoGroupPermissionAny(*_perms)(

			




	
	
	
		
 
                    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',)

			




	
	
		
 
@@ -2376,26 +2376,26 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
        Get given gist by id

			




	
	
	
		
 

			




	
	
	
		
 
        :param gistid: id of private or public gist

			




	
	
	
		
 
        :type gistid: str

			




	
	
	
		
 
        """

			




	
	
	
		
 
        gist = get_gist_or_error(gistid)

			




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

			




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

			




	
	
	
		
 
            if gist.gist_owner != self.authuser.user_id:

			




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

			




	
	
	
		
 
        return gist.get_api_data()

			




	
	
	
		
 

			




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

			




	
	
	
		
 
        """

			




	
	
	
		
 
        Get all gists for given user. If userid is empty returned gists

			




	
	
	
		
 
        are for user who called the api

			




	
	
	
		
 

			




	
	
	
		
 
        :param userid: user to get gists for

			




	
	
	
		
 
        :type userid: Optional(str or int)

			




	
	
	
		
 
        """

			




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

			




	
	
	
		
 
        if not HasPermissionAny('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 != self.authuser.user_id:

			




	
	
	
		
 
                raise JSONRPCError(

			




	
	
	
		
 
                    'userid is not the same as your user'

			




	
	
	
		
 
                )

			




	
	
		
 
@@ -2505,13 +2505,13 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
          error :  {

			




	
	
	
		
 
            "failed to delete gist ID:<gist_id>"

			




	
	
	
		
 
          }

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        gist = get_gist_or_error(gistid)

			




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

			




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

			




	
	
	
		
 
            if gist.gist_owner != self.authuser.user_id:

			




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

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            GistModel().delete(gist)

			




	
	
	
		
 
            Session().commit()

			




        

    


    
    

    

        

                

                    kallithea/lib/auth.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
		
 
@@ -1062,90 +1062,12 @@ class HasPermissionAnyMiddleware(object)

			




	
	
	
		
 
            return True

			




	
	
	
		
 
        log.debug('Permission to repo: %s denied for user: %s @ %s',

			




	
	
	
		
 
                  self.repo_name, self.username, 'PermissionMiddleware')

			




	
	
	
		
 
        return False

			




	
	
	
		
 

			




	
	
	
		
 

			




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

			




	
	
	
		
 
# SPECIAL VERSION TO HANDLE API AUTH

			




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

			




	
	
	
		
 
class _BaseApiPerm(object):

			




	
	
	
		
 
    def __init__(self, *perms):

			




	
	
	
		
 
        self.required_perms = set(perms)

			




	
	
	
		
 

			




	
	
	
		
 
    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)

			




	
	
	
		
 
        if repo_name:

			




	
	
	
		
 
            check_scope += ', repo:%s' % (repo_name)

			




	
	
	
		
 

			




	
	
	
		
 
        if group_name:

			




	
	
	
		
 
            check_scope += ', repo group:%s' % (group_name)

			




	
	
	
		
 

			




	
	
	
		
 
        log.debug('checking cls:%s %s %s @ %s',

			




	
	
	
		
 
                  cls_name, self.required_perms, check_scope, check_location)

			




	
	
	
		
 

			




	
	
	
		
 
        ## process user

			




	
	
	
		
 
        if not check_location:

			




	
	
	
		
 
            check_location = 'unspecified'

			




	
	
	
		
 
        if self.check_permissions(user.permissions, repo_name, group_name):

			




	
	
	
		
 
            log.debug('Permission to %s granted for user: %s @ %s',

			




	
	
	
		
 
                      check_scope, user, check_location)

			




	
	
	
		
 
            return True

			




	
	
	
		
 

			




	
	
	
		
 
        else:

			




	
	
	
		
 
            log.debug('Permission to %s denied for user: %s @ %s',

			




	
	
	
		
 
                      check_scope, user, check_location)

			




	
	
	
		
 
            return False

			




	
	
	
		
 

			




	
	
	
		
 
    def check_permissions(self, perm_defs, repo_name=None, group_name=None):

			




	
	
	
		
 
        """

			




	
	
	
		
 
        implement in child class should return True if permissions are ok,

			




	
	
	
		
 
        False otherwise

			




	
	
	
		
 

			




	
	
	
		
 
        :param perm_defs: dict with permission definitions

			




	
	
	
		
 
        :param repo_name: repo name

			




	
	
	
		
 
        """

			




	
	
	
		
 
        raise NotImplementedError()

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class HasPermissionAnyApi(_BaseApiPerm):

			




	
	
	
		
 
    def check_permissions(self, perm_defs, repo_name=None, group_name=None):

			




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

			




	
	
	
		
 
            return True

			




	
	
	
		
 
        return False

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class HasRepoPermissionAnyApi(_BaseApiPerm):

			




	
	
	
		
 
    def check_permissions(self, perm_defs, repo_name=None, group_name=None):

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            _user_perms = set([perm_defs['repositories'][repo_name]])

			




	
	
	
		
 
        except KeyError:

			




	
	
	
		
 
            log.warning(traceback.format_exc())

			




	
	
	
		
 
            return False

			




	
	
	
		
 
        if self.required_perms.intersection(_user_perms):

			




	
	
	
		
 
            return True

			




	
	
	
		
 
        return False

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class HasRepoGroupPermissionAnyApi(_BaseApiPerm):

			




	
	
	
		
 
    def check_permissions(self, perm_defs, repo_name=None, group_name=None):

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            _user_perms = set([perm_defs['repositories_groups'][group_name]])

			




	
	
	
		
 
        except KeyError:

			




	
	
	
		
 
            log.warning(traceback.format_exc())

			




	
	
	
		
 
            return False

			




	
	
	
		
 
        if self.required_perms.intersection(_user_perms):

			




	
	
	
		
 
            return True

			




	
	
	
		
 
        return False

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def check_ip_access(source_ip, allowed_ips=None):

			




	
	
	
		
 
    """

			




	
	
	
		
 
    Checks if source_ip is a subnet of any of allowed_ips.

			




	
	
	
		
 

			




	
	
	
		
 
    :param source_ip:

			




	
	
	
		
 
    :param allowed_ips: list of allowed ips together with mask

			




        

    



    


    



    



      

      





    
    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.