kallithea Changeset - c0c8d12dc032

Changeset - c0c8d12dc032

Parent rev.

Child rev.

[Not reviewed]

default

0 2 0

Mads Kiilerich - 8 years ago 2018-05-07 00:49:44
mads@kiilerich.com

repos: add missing access control check for repository permission management

This issue was found and reported by
Kacper Szurek
https://security.szurek.pl/

2 files changed with 8 insertions and 6 deletions:

kallithea/controllers/admin/repos.py

kallithea/tests/functional/test_admin_permissions.py

0 comments (0 inline, 0 general)

kallithea/controllers/admin/repos.py

➞

Show inline comments

@@ @@ -297,59 +297,61 @@ class ReposController(BaseRepoController @@
     @HasRepoPermissionLevelDecorator('admin')
     def edit(self, repo_name):
         defaults = self.__load_data()
         c.repo_fields = RepositoryField.query() \
             .filter(RepositoryField.repository == c.repo_info).all()
         c.active = 'settings'
         return htmlfill.render(
             render('admin/repos/repo_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @HasRepoPermissionLevelDecorator('admin')
     def edit_permissions(self, repo_name):
         c.repo_info = self._load_repo()
         c.active = 'permissions'
         defaults = RepoModel()._get_defaults(repo_name)
         return htmlfill.render(
             render('admin/repos/repo_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @HasRepoPermissionLevelDecorator('admin')
     def edit_permissions_update(self, repo_name):
         form = RepoPermsForm()().to_python(request.POST)
         RepoModel()._update_permissions(repo_name, form['perms_new'],
                                         form['perms_updates'])
         # TODO: implement this
         #action_logger(request.authuser, 'admin_changed_repo_permissions',
         #              repo_name, request.ip_addr)
         Session().commit()
         h.flash(_('Repository permissions updated'), category='success')
         raise HTTPFound(location=url('edit_repo_perms', repo_name=repo_name))
     @HasRepoPermissionLevelDecorator('admin')
     def edit_permissions_revoke(self, repo_name):
         try:
             obj_type = request.POST.get('obj_type')
             obj_id = None
             if obj_type == 'user':
                 obj_id = safe_int(request.POST.get('user_id'))
             elif obj_type == 'user_group':
                 obj_id = safe_int(request.POST.get('user_group_id'))
             if obj_type == 'user':
                 RepoModel().revoke_user_permission(repo=repo_name, user=obj_id)
             elif obj_type == 'user_group':
                 RepoModel().revoke_user_group_permission(
                     repo=repo_name, group_name=obj_id
+                )
             # TODO: implement this
             #action_logger(request.authuser, 'admin_revoked_repo_permissions',
             #              repo_name, request.ip_addr)
             Session().commit()
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('An error occurred during revoking of permission'),
                     category='error')
             raise HTTPInternalServerError()

kallithea/tests/functional/test_admin_permissions.py

➞

Show inline comments

@@ @@ -61,69 +61,69 @@ class TestAdminPermissionsController(Tes @@
         response.mustcontain('127.0.0.0 - 127.0.0.255')
         ## now delete
         response = self.app.post(url('edit_user_ips_delete', id=default_user_id),
                                  params=dict(del_ip_id=ip_obj.ip_id,
                                              _authentication_token=self.authentication_token()),
                                  extra_environ={'REMOTE_ADDR': '127.0.0.1'})
         # IP permissions are cached, need to invalidate this cache explicitly
         invalidate_all_caches()
         response = self.app.get(url('admin_permissions_ips'))
         response.mustcontain('All IP addresses are allowed')
         response.mustcontain(no=['127.0.0.0/24'])
         response.mustcontain(no=['127.0.0.0 - 127.0.0.255'])
     def test_index_overview(self):
         self.log_user()
         response = self.app.get(url('admin_permissions_perms'))
         # Test response...
     def test_edit_permissions_permissions(self):
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         # Test unauthenticated access
         # FIXME: access without authentication
         # Test unauthenticated access - it will redirect to login page
         response = self.app.post(
             url('edit_repo_perms_update', repo_name=HG_REPO),
             params=dict(
                 perm_new_member_1='repository.read',
                 perm_new_member_name_1=user.username,
                 perm_new_member_type_1='user',
                 _authentication_token=self.authentication_token()),
             status=302)
         assert response.location.endswith(url('edit_repo_perms_update', repo_name=HG_REPO))
         assert not response.location.endswith(url('edit_repo_perms_update', repo_name=HG_REPO))
         assert response.location.endswith(url('login_home', came_from=url('edit_repo_perms_update', repo_name=HG_REPO)))
         # FIXME: access without authentication
         response = self.app.post(
             url('edit_repo_perms_revoke', repo_name=HG_REPO),
             params=dict(
                 obj_type='user',
                 user_id=user.user_id,
                 _authentication_token=self.authentication_token()),
             status=204) # success has no content
         assert not response.body
             status=302)
         assert response.location.endswith(url('login_home', came_from=url('edit_repo_perms_revoke', repo_name=HG_REPO)))
         # Test authenticated access
         self.log_user()
         response = self.app.post(
             url('edit_repo_perms_update', repo_name=HG_REPO),
             params=dict(
                 perm_new_member_1='repository.read',
                 perm_new_member_name_1=user.username,
                 perm_new_member_type_1='user',
                 _authentication_token=self.authentication_token()),
             status=302)
         assert response.location.endswith(url('edit_repo_perms_update', repo_name=HG_REPO))
         response = self.app.post(
             url('edit_repo_perms_revoke', repo_name=HG_REPO),
             params=dict(
                 obj_type='user',
                 user_id=user.user_id,
                 _authentication_token=self.authentication_token()),
             status=204) # success has no content
         assert not response.body

0 comments (0 inline, 0 general)