kallithea Changeset - 516a43cbd814

Changeset - 516a43cbd814

Parent rev.

Child rev.

[Not reviewed]

Merge default

0 3 0

Mads Kiilerich (mads) - 5 years ago 2021-01-07 03:28:37
mads@kiilerich.com

Merge stable

3 files changed with 46 insertions and 40 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

@@ @@ -1153,25 +1153,25 @@ class ApiController(JSONRPCController): @@
                 'dirs': _d,
+            }
             return _map[ret_type]
         except KeyError:
             raise JSONRPCError('ret_type must be one of %s'
                                % (','.join(sorted(_map))))
         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=None,
                     repo_type=None, description='',
                     private=False, clone_uri=None,
                     landing_rev='rev:tip',
                     enable_statistics=None,
                     enable_downloads=None,
                     copy_permissions=False):
         """
         Creates a repository. The repository name contains the full path, but the
         parent repository group must exist. For example "foo/bar/baz" require the groups
         "foo" and "bar" (with "foo" as parent), and create "baz" repository with
         "bar" as group. This command can be executed only using api_key
@@ @@ -1210,24 +1210,37 @@ class ApiController(JSONRPCController): @@
+                    }
             error:  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           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 = db.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 owner is not None:
                 # forbid setting owner for non-admins
                 raise JSONRPCError(
                     'Only Kallithea admin can specify `owner` param'
+                )
         if owner is None:
             owner = request.authuser.user_id
         owner = get_user_or_error(owner)
         if RepoModel().get_by_repo_name(repo_name):
@@ @@ -1235,39 +1248,32 @@ class ApiController(JSONRPCController): @@
         defs = db.Setting.get_default_repo_settings(strip_prefix=True)
         if private is None:
             private = defs.get('repo_private') or False
         if repo_type is None:
             repo_type = defs.get('repo_type')
         if enable_statistics is None:
             enable_statistics = defs.get('repo_enable_statistics')
         if enable_downloads is None:
             enable_downloads = defs.get('repo_enable_downloads')
         try:
             repo_name_parts = repo_name.split('/')
             repo_group = None
             if len(repo_name_parts) > 1:
                 group_name = '/'.join(repo_name_parts[:-1])
                 repo_group = db.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,
                 repo_private=private,
                 clone_uri=clone_uri,
-                repo_group=repo_group,
+                repo_group=group_name,
                 repo_landing_rev=landing_rev,
                 enable_statistics=enable_statistics,
                 enable_downloads=enable_downloads,
                 repo_copy_permissions=copy_permissions,
+            )
             RepoModel().create(form_data=data, cur_user=owner.username)
             # no commit, it's done in RepoModel, or async via celery
             return dict(
                 msg="Created new repository `%s`" % (repo_name,),
                 success=True,  # cannot return the repo data here since fork
                                # can be done async
@@ @@ -1297,61 +1303,65 @@ class ApiController(JSONRPCController): @@
         :param description:
         :param private:
         :param clone_uri:
         :param landing_rev:
         :param enable_statistics:
         :param enable_downloads:
         """
         repo = get_repo_or_error(repoid)
         if not HasPermissionAny('hg.admin')():
             if not HasRepoPermissionLevel('admin')(repo.repo_name):
                 raise JSONRPCError('repository `%s` does not exist' % (repoid,))
             if (name != repo.repo_name and
+            if (name != repo.repo_name and repo.group_id is None and
                 not HasPermissionAny('hg.create.repository')()
             ):
                 raise JSONRPCError('no permission to create (or move) repositories')
+                raise JSONRPCError('no permission to create (or move) top level repositories')
             if owner is not None:
                 # forbid setting owner for non-admins
                 raise JSONRPCError(
                     'Only Kallithea admin can specify `owner` param'
+                )
         updates = {}
         repo_group = group
         if repo_group is not None:
             repo_group = get_repo_group_or_error(repo_group)
             repo_group = get_repo_group_or_error(repo_group)  # TODO: repos can thus currently not be moved to root
             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')
             store_update(updates, private, 'repo_private')
             store_update(updates, clone_uri, 'clone_uri')
             store_update(updates, landing_rev, 'repo_landing_rev')
             store_update(updates, enable_statistics, 'repo_enable_statistics')
             store_update(updates, enable_downloads, 'repo_enable_downloads')
             RepoModel().update(repo, **updates)
             meta.Session().commit()
             return dict(
                 msg='updated repo ID:%s %s' % (repo.repo_id, repo.repo_name),
                 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=None,
                   description='', copy_permissions=False,
                   private=False, landing_rev='rev:tip'):
         """
         Creates a fork of given repo. In case of using celery this will
         immediately return success message, while fork is going to be created
         asynchronous. This command can be executed only using api_key belonging to
         user with admin rights or regular user that have fork permission, and at least
         read access to forking repository. Regular users cannot specify owner parameter.
@@ @@ -1388,56 +1398,58 @@ 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))
         group_name = None
         fork_name_parts = fork_name.split('/')
         if len(fork_name_parts) > 1:
             group_name = '/'.join(fork_name_parts[:-1])
             repo_group = db.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 owner is not None:
                 # 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 owner is None:
             owner = request.authuser.user_id
         owner = get_user_or_error(owner)
         try:
             fork_name_parts = fork_name.split('/')
             repo_group = None
             if len(fork_name_parts) > 1:
                 group_name = '/'.join(fork_name_parts[:-1])
                 repo_group = db.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=repo_group,
+                repo_group=group_name,
                 repo_type=repo.repo_type,
                 description=description,
                 private=private,
                 copy_permissions=copy_permissions,
                 landing_rev=landing_rev,
                 update_after_clone=False,
                 fork_parent_id=repo.repo_id,
+            )
             RepoModel().create_fork(form_data, cur_user=owner.username)
             # no commit, it's done in RepoModel, or async via celery
             return dict(
                 msg='Created fork of `%s` as `%s`' % (repo.repo_name,

kallithea/templates/admin/permissions/permissions_globals.html

➞

Show inline comments

@@ @@ -45,25 +45,24 @@ ${h.form(url('admin_permissions'), metho @@
                         <span data-toggle="tooltip" title="${_('All default permissions on each user group will be reset to chosen permission, note that all custom default permission on user groups will be lost')}">
                             ${_('Apply to all existing user groups')}
                         </span>
                     </label>
                     <span class="help-block">${_('Permissions for the Default user on new user groups.')}</span>
                 </div>
             </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="default_user_group_create">${_('User group creation')}:</label>
                 <div>
                     ${h.select('default_user_group_create','',c.user_group_create_choices,class_='form-control')}
                     <span class="help-block">${_('Enable this to allow non-admins to create user groups.')}</span>
                 </div>
             </div>
             <div class="form-group">
                 <label class="control-label" for="default_fork">${_('Repository forking')}:</label>
                 <div>

kallithea/tests/api/api_base.py

➞

Show inline comments

@@ @@ -878,37 +878,29 @@ class _BaseTestApi(object): @@
         meta.Session().commit()
         RepoGroupModel().grant_user_permission(repo_group_name,
                                                self.TEST_USER_LOGIN,
                                                'group.none')
         meta.Session().commit()
         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,
+        }
         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',
                                   repo_name=repo_name,
                                   owner=owner,
                                   repo_type=self.REPO_TYPE,
+        )
         response = api_call(self, params)
         expected = 'user `%s` does not exist' % owner
@@ @@ -1112,39 +1104,39 @@ class _BaseTestApi(object): @@
     def test_api_update_repo_exception_occurred(self):
         repo_name = 'api_update_me'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
         id_, params = _build_data(self.apikey, 'update_repo',
                                   repoid=repo_name, owner=base.TEST_USER_ADMIN_LOGIN,)
         response = api_call(self, params)
         try:
             expected = 'failed to update repo `%s`' % repo_name
             self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
-    def test_api_update_repo_regular_user_change_repo_name(self):
+    def test_api_update_repo_regular_user_change_top_level_repo_name(self):
         repo_name = 'admin_owned'
         new_repo_name = 'new_repo_name'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
         RepoModel().grant_user_permission(repo=repo_name,
                                           user=self.TEST_USER_LOGIN,
                                           perm='repository.admin')
         UserModel().revoke_perm('default', 'hg.create.repository')
         UserModel().grant_perm('default', 'hg.create.none')
         updates = {'name': new_repo_name}
         id_, params = _build_data(self.apikey_regular, 'update_repo',
                                   repoid=repo_name, **updates)
         response = api_call(self, params)
         try:
             expected = 'no permission to create (or move) repositories'
+            expected = 'no permission to create (or move) top level repositories'
             self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
             fixture.destroy_repo(new_repo_name)
     def test_api_update_repo_regular_user_change_repo_name_allowed(self):
         repo_name = 'admin_owned'
         new_repo_name = 'new_repo_name'
         repo = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
         RepoModel().grant_user_permission(repo=repo_name,
                                           user=self.TEST_USER_LOGIN,
                                           perm='repository.admin')
@@ @@ -1256,24 +1248,27 @@ class _BaseTestApi(object): @@
                                                      fork_name),
             'success': True,
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_repo(fork_name)
     @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)
         ret = {
             'msg': 'Created fork of `%s` as `%s`' % (self.REPO,
                                                      fork_name),
             'success': True,
+        }
         expected = ret
@@ @@ -1321,25 +1316,25 @@ class _BaseTestApi(object): @@
         # regardless of base repository permission, forking is disallowed
         # when repository creation is disabled
         RepoModel().grant_user_permission(repo=self.REPO,
                                           user=self.TEST_USER_LOGIN,
                                           perm=perm)
         UserModel().revoke_perm('default', 'hg.create.repository')
         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'
         id_, params = _build_data(self.apikey, 'fork_repo',
                                   repoid=self.REPO,
                                   fork_name=fork_name,
                                   owner=owner,
+        )
         response = api_call(self, params)

0 comments (0 inline, 0 general)