kallithea Changeset - c96e05599877

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

kallithea/lib/auth.py

0 comments (0 inline, 0 general)

kallithea/controllers/api/api.py

➞

Show inline comments

@@ @@ -273,29 +273,29 @@ class ApiController(JSONRPCController): @@
           error :  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           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,),
                 repository=repo.repo_name
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'Error occurred during cache invalidation action'
@@ @@ -339,29 +339,28 @@ class ApiController(JSONRPCController): @@
           error :  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           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'
+                )
         else:
             raise JSONRPCError('repository `%s` does not exist' % (repoid,))
         if isinstance(userid, Optional):
             userid = apiuser.user_id
@@ @@ -432,25 +431,25 @@ class ApiController(JSONRPCController): @@
         :param userid: User to get locks for
         :type userid: Optional(str or int)
         OUTPUT::
           id : <id_given_in_input>
           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'
+                )
         ret = []
         if isinstance(userid, Optional):
             user = None
         else:
             user = get_user_or_error(userid)
@@ @@ -566,25 +565,25 @@ class ApiController(JSONRPCController): @@
                         "permissions": {
                             "global": ["hg.create.repository",
                                        "repository.read",
                                        "hg.register.manual_activate"],
                             "repositories": {"repo1": "repository.none"},
                             "repositories_groups": {"Group1": "group.read"}
                          },
+                    }
             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'
+                )
         if isinstance(userid, Optional):
             userid = apiuser.user_id
         user = get_user_or_error(userid)
         data = user.get_api_data()
@@ @@ -841,56 +840,55 @@ class ApiController(JSONRPCController): @@
             id : <id_given_in_input>
             result : None if group not exist
+                     {
                        "users_group_id" : "<id>",
                        "group_name" :     "<groupname>",
                        "active":          "<bool>",
                        "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
     def get_user_groups(self, apiuser):
         """
         Lists all existing user groups. This command can be executed only using
         api_key belonging to user with admin rights or user who has at least
         read access to user group.
         :param apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         OUTPUT::
             id : <id_given_in_input>
             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)):
         """
         Creates new user group. This command can be executed only using api_key
         belonging to user with admin rights or an user who has create user group
         permission
         :param apiuser: filled automatically from apikey
@@ @@ -977,29 +975,29 @@ class ApiController(JSONRPCController): @@
           error :  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           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 = {}
         store_update(updates, group_name, 'users_group_name')
         store_update(updates, description, 'user_group_description')
         store_update(updates, owner, 'user')
         store_update(updates, active, 'users_group_active')
         try:
             UserGroupModel().update(user_group, updates)
@@ @@ -1036,29 +1034,29 @@ class ApiController(JSONRPCController): @@
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             "failed to delete user group ID:<user_group_id> <user_group_name>"
             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(
                 msg='deleted user group ID:%s %s' %
                     (user_group.users_group_id, user_group.users_group_name),
                 user_group=None
+            )
         except UserGroupsAssignedException as e:
             log.error(traceback.format_exc())
@@ @@ -1097,29 +1095,29 @@ class ApiController(JSONRPCController): @@
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             "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`' % (
                 user.username, user_group.users_group_name
+            )
             msg = msg if success else 'User is already in that group'
             Session().commit()
             return dict(
@@ @@ -1151,29 +1149,29 @@ class ApiController(JSONRPCController): @@
             id : <id_given_in_input>
             result: {
                       "success":  True|False,  # depends on if member is in group
                       "msg": "removed member <username> from user group <groupname> |
                               User wasn't in group"
+                    }
             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
+            )
             msg = msg if success else "User wasn't in group"
             Session().commit()
             return dict(success=success, msg=msg)
         except Exception:
             log.error(traceback.format_exc())
@@ @@ -1236,28 +1234,28 @@ class ApiController(JSONRPCController): @@
                                   },
                                   …
+                                ]
                  "followers":   [<user_obj>, ...]
+                 ]
+            }
+          }
           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
             user = user.user
             user_data = {
                 'name': user.username,
                 'type': "user",
                 'permission': perm
+            }
@@ @@ -1306,25 +1304,25 @@ class ApiController(JSONRPCController): @@
                         "landing_rev":       "<landing_rev>",
                         "owner":             "<repo_owner>",
                         "fork_of":           "<name_of_fork_parent>",
                         "enable_downloads":  "<bool>",
                         "enable_locking":    "<bool>",
                         "enable_statistics": "<bool>",
                       },
                       …
+                    ]
             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())
         return result
     # permission check inside
     def get_repo_nodes(self, apiuser, repoid, revision, root_path,
                        ret_type=Optional('all')):
         """
@@ @@ -1350,28 +1348,28 @@ class ApiController(JSONRPCController): @@
             id : <id_given_in_input>
             result: [
+                      {
                         "name" :        "<name>"
                         "type" :        "<type>",
                       },
                       …
+                    ]
             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,
                                           flat=False)
             _map = {
                 'all': _d + _f,
                 'files': _f,
                 'dirs': _d,
+            }
@@ @@ -1438,25 +1436,25 @@ class ApiController(JSONRPCController): @@
+                    }
             error:  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           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):
             owner = apiuser.user_id
         owner = get_user_or_error(owner)
         if RepoModel().get_by_repo_name(repo_name):
             raise JSONRPCError("repo `%s` already exist" % repo_name)
@@ @@ -1532,32 +1530,31 @@ class ApiController(JSONRPCController): @@
         :param name:
         :param owner:
         :param group:
         :param description:
         :param private:
         :param clone_uri:
         :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(
                     'Only Kallithea admin can specify `owner` param'
+                )
         updates = {}
         repo_group = group
         if not isinstance(repo_group, Optional):
@@ @@ -1632,37 +1629,36 @@ class ApiController(JSONRPCController): @@
+                    }
             error:  null
         """
         repo = get_repo_or_error(repoid)
         repo_name = repo.repo_name
         _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
         owner = get_user_or_error(owner)
         try:
             # create structure of groups and return the last group
             group = map_groups(fork_name)
@@ @@ -1715,28 +1711,27 @@ class ApiController(JSONRPCController): @@
         OUTPUT::
             id : <id_given_in_input>
             result: {
                       "msg": "Deleted repository `<reponame>`",
                       "success": true
+                    }
             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]
             if handle_forks == 'detach':
                 _forks_msg = ' ' + 'Detached %s forks' % len(_forks)
             elif handle_forks == 'delete':
                 _forks_msg = ' ' + 'Deleted %s forks' % len(_forks)
             elif _forks:
                 raise JSONRPCError(
@@ @@ -1874,35 +1869,35 @@ class ApiController(JSONRPCController): @@
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             "failed to edit permission for user group: `<usergroup>` in repo `<repo>`'
+          }
         """
         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)
             Session().commit()
             return dict(
                 msg='Granted perm: `%s` for user group: `%s` in '
                     'repo: `%s`' % (
                         perm.permission_name, user_group.users_group_name,
                         repo.repo_name
@@ @@ -1932,35 +1927,35 @@ class ApiController(JSONRPCController): @@
         OUTPUT::
             id : <id_given_in_input>
             result: {
                       "msg" : "Revoked perm for group: `<usersgroupname>` in repo: `<reponame>`",
                       "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)
             Session().commit()
             return dict(
                 msg='Revoked perm for user group: `%s` in repo: `%s`' % (
                     user_group.users_group_name, repo.repo_name
                 ),
                 success=True
@@ @@ -2194,28 +2189,27 @@ class ApiController(JSONRPCController): @@
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             "failed to edit permission for user: `<userid>` in repo group: `<repo_group_name>`"
+          }
         """
         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)
         try:
             RepoGroupModel().add_permission(repo_group=repo_group,
                                             obj=user,
                                             obj_type="user",
                                             perm=perm,
                                             recursive=apply_to_children)
@@ @@ -2261,28 +2255,27 @@ class ApiController(JSONRPCController): @@
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             "failed to edit permission for user: `<userid>` in repo group: `<repo_group_name>`"
+          }
         """
         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:
             RepoGroupModel().delete_permission(repo_group=repo_group,
                                                obj=user,
                                                obj_type="user",
                                                recursive=apply_to_children)
             Session().commit()
@@ @@ -2332,36 +2325,36 @@ class ApiController(JSONRPCController): @@
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             "failed to edit permission for user group: `<usergroup>` in repo group: `<repo_group_name>`"
+          }
         """
         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:
             RepoGroupModel().add_permission(repo_group=repo_group,
                                             obj=user_group,
                                             obj_type="user_group",
                                             perm=perm,
                                             recursive=apply_to_children)
             Session().commit()
@@ @@ -2410,36 +2403,36 @@ class ApiController(JSONRPCController): @@
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             "failed to edit permission for user group: `<usergroup>` in repo group: `<repo_group_name>`"
+          }
         """
         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:
             RepoGroupModel().delete_permission(repo_group=repo_group,
                                                obj=user_group,
                                                obj_type="user_group",
                                                recursive=apply_to_children)
             Session().commit()
             return dict(
@@ @@ -2457,40 +2450,40 @@ class ApiController(JSONRPCController): @@
+            )
     def get_gist(self, apiuser, gistid):
         """
         Get given gist by id
         :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'))):
         """
         Get all gists for given user. If userid is empty returned gists
         are for user who called the api
         :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'
+                )
         if isinstance(userid, Optional):
             user_id = apiuser.user_id
         else:
             user_id = get_user_or_error(userid).user_id
@@ @@ -2592,25 +2585,25 @@ class ApiController(JSONRPCController): @@
           error :  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           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()
             return dict(
                 msg='deleted gist ID:%s' % (gist.gist_access_id,),
                 gist=None
+            )
         except Exception:
             log.error(traceback.format_exc())

kallithea/lib/auth.py

➞

Show inline comments

@@ @@ -930,28 +930,25 @@ class PermsFunction(object): @@
         self.required_perms = set(perms)
         self.user_perms = None
         self.repo_name = None
         self.group_name = None
     def __nonzero__(self):
         """ 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()
         log.debug('checking cls:%s %s usr:%s %s @ %s', cls_name,
                   self.required_perms, user, check_scope,
                   check_location)
         self.user_perms = user.permissions
         result = self.check_permissions()
@@ @@ -967,70 +964,70 @@ class PermsFunction(object): @@
     def _scope(self):
         return '(unknown scope)'
 class HasPermissionAny(PermsFunction):
     def check_permissions(self):
         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:
             self._user_perms = set(
                 [self.user_perms['repositories'][self.repo_name]]
+            )
         except KeyError:
             return False
         if self.required_perms.intersection(self._user_perms):
             return True
         return False
     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]]
+            )
         except KeyError:
             return False
         if self.required_perms.intersection(self._user_perms):
             return True
         return False
     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]]
+            )
         except KeyError:
             return False
         if self.required_perms.intersection(self._user_perms):
             return True
         return False
@@ @@ -1066,29 +1063,25 @@ class HasPermissionAnyMiddleware(object) @@
         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, 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)
         if repo_name:
             check_scope += ', repo:%s' % (repo_name)
         if group_name:
             check_scope += ', repo group:%s' % (group_name)

0 comments (0 inline, 0 general)