kallithea Changeset - 7643d8ecbb20

Changeset - 7643d8ecbb20

Parent rev.

Child rev.

[Not reviewed]

stable

0 3 0

Mads Kiilerich (mads) - 5 years ago 2021-01-02 23:41:37
mads@kiilerich.com

Grafted from: 217062ecb161

api: fix repo group permission check for repo creation

hg.create.repository is only controlling whether all users have write access at
top level. For other repo locations, check for write permission to the repo
group.

Note: This also covers "repo creation" by forking or by moving another repo.

3 files changed with 39 insertions and 34 deletions:

kallithea/controllers/api/api.py

kallithea/templates/admin/permissions/permissions_globals.html

kallithea/tests/api/api_base.py

0 comments (0 inline, 0 general)

kallithea/controllers/api/api.py

➞

Show inline comments

@@ @@ -1164,13 +1164,13 @@ class ApiController(JSONRPCController): @@
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to get repo: `%s` nodes' % repo.repo_name
+            )
-    @HasPermissionAnyDecorator('hg.admin', 'hg.create.repository')
+    # permission check inside
     def create_repo(self, repo_name, owner=Optional(OAttr('apiuser')),
                     repo_type=Optional('hg'), description=Optional(''),
                     private=Optional(False), clone_uri=Optional(None),
                     landing_rev=Optional('rev:tip'),
                     enable_statistics=Optional(False),
                     enable_downloads=Optional(False),
@@ @@ -1221,12 +1221,25 @@ class ApiController(JSONRPCController): @@
           result : null
           error :  {
              'failed to create repository `<repo_name>`
+          }
         """
         group_name = None
         repo_name_parts = repo_name.split('/')
         if len(repo_name_parts) > 1:
             group_name = '/'.join(repo_name_parts[:-1])
             repo_group = RepoGroup.get_by_group_name(group_name)
             if repo_group is None:
                 raise JSONRPCError("repo group `%s` not found" % group_name)
             if not(HasPermissionAny('hg.admin')() or HasRepoGroupPermissionLevel('write')(group_name)):
                 raise JSONRPCError("no permission to create repo in %s" % group_name)
         else:
             if not HasPermissionAny('hg.admin', 'hg.create.repository')():
                 raise JSONRPCError("no permission to create top level repo")
         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'
+                )
@@ @@ -1251,19 +1264,12 @@ class ApiController(JSONRPCController): @@
         clone_uri = Optional.extract(clone_uri)
         description = Optional.extract(description)
         landing_rev = Optional.extract(landing_rev)
         copy_permissions = Optional.extract(copy_permissions)
         try:
             repo_name_parts = repo_name.split('/')
             group_name = None
             if len(repo_name_parts) > 1:
                 group_name = '/'.join(repo_name_parts[:-1])
                 repo_group = RepoGroup.get_by_group_name(group_name)
                 if repo_group is None:
                     raise JSONRPCError("repo group `%s` not found" % group_name)
             data = dict(
                 repo_name=repo_name_parts[-1],
                 repo_name_full=repo_name,
                 repo_type=repo_type,
                 repo_description=description,
                 owner=owner,
@@ @@ -1331,12 +1337,15 @@ class ApiController(JSONRPCController): @@
+                )
         updates = {}
         repo_group = group
         if not isinstance(repo_group, Optional):
             repo_group = get_repo_group_or_error(repo_group)
             if repo_group.group_id != repo.group_id:
                 if not(HasPermissionAny('hg.admin')() or HasRepoGroupPermissionLevel('write')(repo_group.group_name)):
                     raise JSONRPCError("no permission to create (or move) repo in %s" % repo_group.group_name)
             repo_group = repo_group.group_id
         try:
             store_update(updates, name, 'repo_name')
             store_update(updates, repo_group, 'repo_group')
             store_update(updates, owner, 'owner')
             store_update(updates, description, 'repo_description')
@@ @@ -1353,12 +1362,13 @@ class ApiController(JSONRPCController): @@
                 repository=repo.get_api_data()
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to update repo `%s`' % repoid)
     # permission check inside
     @HasPermissionAnyDecorator('hg.admin', 'hg.fork.repository')
     def fork_repo(self, repoid, fork_name,
                   owner=Optional(OAttr('apiuser')),
                   description=Optional(''), copy_permissions=Optional(False),
                   private=Optional(False), landing_rev=Optional('rev:tip')):
         """
@@ @@ -1407,40 +1417,42 @@ 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))
         group_name = None
         fork_name_parts = fork_name.split('/')
         if len(fork_name_parts) > 1:
             group_name = '/'.join(fork_name_parts[:-1])
             repo_group = RepoGroup.get_by_group_name(group_name)
             if repo_group is None:
                 raise JSONRPCError("repo group `%s` not found" % group_name)
             if not(HasPermissionAny('hg.admin')() or HasRepoGroupPermissionLevel('write')(group_name)):
                 raise JSONRPCError("no permission to create repo in %s" % group_name)
         else:
             if not HasPermissionAny('hg.admin', 'hg.create.repository')():
                 raise JSONRPCError("no permission to create top level repo")
         if HasPermissionAny('hg.admin')():
             pass
         elif HasRepoPermissionLevel('read')(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 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 = request.authuser.user_id
         owner = get_user_or_error(owner)
         try:
             fork_name_parts = fork_name.split('/')
             group_name = None
             if len(fork_name_parts) > 1:
                 group_name = '/'.join(fork_name_parts[:-1])
                 repo_group = RepoGroup.get_by_group_name(group_name)
                 if repo_group is None:
                     raise JSONRPCError("repo group `%s` not found" % group_name)
             form_data = dict(
                 repo_name=fork_name_parts[-1],
                 repo_name_full=fork_name,
                 repo_group=group_name,
                 repo_type=repo.repo_type,
                 description=Optional.extract(description),

kallithea/templates/admin/permissions/permissions_globals.html

➞

Show inline comments

@@ @@ -51,13 +51,12 @@ ${h.form(url('admin_permissions'), metho @@
             </div>
             <div class="form-group">
                 <label class="control-label" for="default_repo_create">${_('Top level repository creation')}:</label>
                 <div>
                     ${h.select('default_repo_create','',c.repo_create_choices,class_='form-control')}
                     <span class="help-block">${_('Enable this to allow non-admins to create repositories at the top level.')}</span>
                     <span class="help-block">${_('Note: This will also give all users API access to create repositories everywhere. That might change in future versions.')}</span>
                 </div>
             </div>
             <div class="form-group">
                 <label class="control-label" for="create_on_write">${_('Repository creation with group write access')}:</label>
                 <div>
                     ${h.select('create_on_write','',c.repo_create_on_write_choices,class_='form-control')}

kallithea/tests/api/api_base.py

➞

Show inline comments

@@ @@ -913,26 +913,17 @@ class _BaseTestApi(object): @@
         id_, params = _build_data(self.apikey_regular, 'create_repo',
                                   repo_name=repo_name,
                                   repo_type=self.REPO_TYPE,
+        )
         response = api_call(self, params)
         # Current result when API access control is different from Web:
         ret = {
             'msg': 'Created new repository `%s`' % repo_name,
             'success': True,
             'task': None,
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         # API access control match Web access control:
         expected = 'no permission to create repo in test_repo_group/api-repo-repo'
         self._compare_error(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
         # Expected and arguably more correct result:
         #expected = 'failed to create repository `%s`' % repo_name
         #self._compare_error(id_, expected, given=response.body)
         fixture.destroy_repo_group(repo_group_name)
     def test_api_create_repo_unknown_owner(self):
         repo_name = 'api-repo'
         owner = 'i-dont-exist'
         id_, params = _build_data(self.apikey, 'create_repo',
@@ @@ -1295,12 +1286,15 @@ class _BaseTestApi(object): @@
     @base.parametrize('fork_name', [
         'api-repo-fork',
         '%s/api-repo-fork' % TEST_REPO_GROUP,
     ])
     def test_api_fork_repo_non_admin(self, fork_name):
         RepoGroupModel().grant_user_permission(TEST_REPO_GROUP,
                                                self.TEST_USER_LOGIN,
                                                'group.write')
         id_, params = _build_data(self.apikey_regular, 'fork_repo',
                                   repoid=self.REPO,
                                   fork_name=fork_name,
+        )
         response = api_call(self, params)
@@ @@ -1361,13 +1355,13 @@ class _BaseTestApi(object): @@
         UserModel().grant_perm('default', 'hg.create.none')
         id_, params = _build_data(self.apikey_regular, 'fork_repo',
                                   repoid=self.REPO,
                                   fork_name=fork_name,
+        )
         response = api_call(self, params)
-        expected = 'no permission to create repositories'
+        expected = 'no permission to create top level repo'
         self._compare_error(id_, expected, given=response.body)
         fixture.destroy_repo(fork_name)
     def test_api_fork_repo_unknown_owner(self):
         fork_name = 'api-repo-fork'
         owner = 'i-dont-exist'

0 comments (0 inline, 0 general)