kallithea Changeset - a041321d2aa1

Changeset - a041321d2aa1

Parent rev.

Child rev.

[Not reviewed]

default

0 10 0

Søren Løvborg - 10 years ago 2015-09-04 00:01:20
sorenl@unity3d.com

security: apply CSRF check to all non-GET requests

The automatic CSRF protection was broken for POST requests with no
request payload parameters (but possibly containing request URI
parameters); a security hole was narrowly avoided because the code
base quite consistently checks the request method in the same way,
and because of browser protection against PUT/DELETE CSRF attacks.

Since explicit is better than implicit, the better way of checking
the HTTP request method is to simply check request.method, instead
of checking if request.POST is non-empty, which is subtly different
(it doesn't catch POST requests if all parameters are in the query
string) and non-obvious (because it also applies to PUT requests).

The commit also fixes some tests which relied on the CSRF protection
being broken. It does not fix all the controllers that still does
the misleading request.POST check, but since the CSRF check has now
been tightened, those are no longer a potential security issue.

10 files changed with 60 insertions and 34 deletions:

kallithea/lib/auth.py

kallithea/tests/functional/test_admin_defaults.py

kallithea/tests/functional/test_admin_gists.py

kallithea/tests/functional/test_admin_notifications.py

kallithea/tests/functional/test_admin_repos.py

kallithea/tests/functional/test_admin_user_groups.py

kallithea/tests/functional/test_admin_users.py

kallithea/tests/functional/test_changeset_comments.py

kallithea/tests/functional/test_forks.py

kallithea/tests/functional/test_my_account.py

0 comments (0 inline, 0 general)

kallithea/lib/auth.py

➞

Show inline comments

@@ @@ -757,14 +757,19 @@ class LoginRequired(object): @@
                     return redirect_to_login(_('Invalid API key'))
             else:
                 # controller does not allow API access
                 log.warning('API access to %s is not allowed', loc)
                 return abort(403)
         # CSRF protection - POSTs with session auth must contain correct token
         if request.POST and user.is_authenticated:
         # CSRF protection: Whenever a request has ambient authority (whether
         # through a session cookie or its origin IP address), it must include
         # the correct token, unless the HTTP method is GET or HEAD (and thus
         # guaranteed to be side effect free.
         # Note that the 'is_authenticated' flag is True for anonymous users too,
         # but not when the user is authenticated by API key.
         if user.is_authenticated and request.method not in ['GET', 'HEAD']:
             token = request.POST.get(secure_form.token_key)
             if not token or token != secure_form.authentication_token():
                 log.error('CSRF check failed')
                 return abort(403)
         # regular user authentication

kallithea/tests/functional/test_admin_defaults.py

➞

Show inline comments

@@ @@ -13,13 +13,14 @@ class TestDefaultsController(TestControl @@
         response.mustcontain('default_repo_enable_locking')
     def test_index_as_xml(self):
         response = self.app.get(url('formatted_defaults', format='xml'))
     def test_create(self):
         response = self.app.post(url('defaults'))
         response = self.app.post(url('defaults'),
             {'_authentication_token': self.authentication_token()})
     def test_new(self):
         response = self.app.get(url('new_default'))
     def test_new_as_xml(self):
         response = self.app.get(url('formatted_new_default', format='xml'))
@@ @@ -59,13 +60,14 @@ class TestDefaultsController(TestControl @@
         self.assertEqual(params, defs)
     def test_update_browser_fakeout(self):
         response = self.app.post(url('default', id=1), params=dict(_method='put', _authentication_token=self.authentication_token()))
     def test_delete(self):
         response = self.app.delete(url('default', id=1))
         # Not possible due to CSRF protection.
         response = self.app.delete(url('default', id=1), status=403)
     def test_delete_browser_fakeout(self):
         response = self.app.post(url('default', id=1), params=dict(_method='delete', _authentication_token=self.authentication_token()))
     def test_show(self):
         response = self.app.get(url('default', id=1))

kallithea/tests/functional/test_admin_gists.py

➞

Show inline comments

@@ @@ -133,25 +133,26 @@ class TestGistsController(TestController @@
         self.skipTest('not implemented')
         response = self.app.put(url('gist', gist_id=1))
     def test_delete(self):
         self.log_user()
         gist = _create_gist('delete-me')
         response = self.app.delete(url('gist', gist_id=gist.gist_id))
         self.checkSessionFlash(response, 'Deleted gist %s' % gist.gist_id)
         response = self.app.post(url('gist', gist_id=gist.gist_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
     def test_delete_normal_user_his_gist(self):
         self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         gist = _create_gist('delete-me', owner=TEST_USER_REGULAR_LOGIN)
         response = self.app.delete(url('gist', gist_id=gist.gist_id))
         self.checkSessionFlash(response, 'Deleted gist %s' % gist.gist_id)
         response = self.app.post(url('gist', gist_id=gist.gist_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
     def test_delete_normal_user_not_his_own_gist(self):
         self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         gist = _create_gist('delete-me')
         response = self.app.delete(url('gist', gist_id=gist.gist_id), status=403)
         response = self.app.post(url('gist', gist_id=gist.gist_id), status=403,
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
     def test_show(self):
         gist = _create_gist('gist-show-me')
         response = self.app.get(url('gist', gist_id=gist.gist_access_id))
         response.mustcontain('added file: gist-show-me<')
         response.mustcontain('%s - created' % TEST_USER_ADMIN_LOGIN)

kallithea/tests/functional/test_admin_notifications.py

➞

Show inline comments

@@ @@ -53,15 +53,15 @@ class TestNotificationsController(TestCo @@
         get_notif = lambda un: [x.notification for x in un]
         self.assertEqual(get_notif(cur_user.notifications), [notification])
         self.assertEqual(get_notif(u1.notifications), [notification])
         self.assertEqual(get_notif(u2.notifications), [notification])
         cur_usr_id = cur_user.user_id
         response = self.app.delete(url('notification',
                                        notification_id=
                                        notification.notification_id))
         response = self.app.post(
             url('notification', notification_id=notification.notification_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         self.assertEqual(response.body, 'ok')
         cur_user = User.get(cur_usr_id)
         self.assertEqual(cur_user.notifications, [])
     def test_show(self):

kallithea/tests/functional/test_admin_repos.py

➞

Show inline comments

@@ @@ -395,13 +395,14 @@ class _BaseTest(object): @@
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(os.path.join(TESTS_TMP_PATH, repo_name))
         except vcs.exceptions.VCSError:
             self.fail('no repo %s in filesystem' % repo_name)
         response = self.app.delete(url('delete_repo', repo_name=repo_name))
         response = self.app.post(url('delete_repo', repo_name=repo_name),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'Deleted repository %s' % (repo_name))
         response.follow()
         #check if repo was deleted from db
@@ @@ -447,13 +448,14 @@ class _BaseTest(object): @@
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(os.path.join(TESTS_TMP_PATH, repo_name))
         except vcs.exceptions.VCSError:
             self.fail('no repo %s in filesystem' % repo_name)
         response = self.app.delete(url('delete_repo', repo_name=repo_name))
         response = self.app.post(url('delete_repo', repo_name=repo_name),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'Deleted repository %s' % (repo_name_unicode))
         response.follow()
         #check if repo was deleted from db
         deleted_repo = Session().query(Repository)\
             .filter(Repository.repo_name == repo_name_unicode).scalar()

kallithea/tests/functional/test_admin_user_groups.py

➞

Show inline comments

@@ @@ -29,13 +29,13 @@ class TestAdminUsersGroupsController(Tes @@
                                '/edit">%s</a>' % TEST_USER_GROUP)
     def test_new(self):
         response = self.app.get(url('new_users_group'))
     def test_update(self):
         response = self.app.put(url('users_group', id=1))
+        response = self.app.put(url('users_group', id=1), status=403)
     def test_update_browser_fakeout(self):
         response = self.app.post(url('users_group', id=1),
                                  params=dict(_method='put', _authentication_token=self.authentication_token()))
     def test_delete(self):
@@ @@ -51,13 +51,14 @@ class TestAdminUsersGroupsController(Tes @@
         self.checkSessionFlash(response,
                                'Created user group ')
         gr = Session().query(UserGroup)\
             .filter(UserGroup.users_group_name == users_group_name).one()
         response = self.app.delete(url('users_group', id=gr.users_group_id))
         response = self.app.post(url('users_group', id=gr.users_group_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         gr = Session().query(UserGroup)\
             .filter(UserGroup.users_group_name == users_group_name).scalar()
         self.assertEqual(gr, None)
@@ @@ -94,13 +95,14 @@ class TestAdminUsersGroupsController(Tes @@
             sorted([[ug.users_group_id, p.permission_id],
                     [ug.users_group_id, p2.permission_id],
                     [ug.users_group_id, p3.permission_id]]))
         ## DISABLE REPO CREATE ON A GROUP
         response = self.app.put(
             url('edit_user_group_default_perms', id=ug.users_group_id), {})
             url('edit_user_group_default_perms', id=ug.users_group_id),
             params={'_authentication_token': self.authentication_token()})
         response.follow()
         ug = UserGroup.get_by_group_name(users_group_name)
         p = Permission.get_by_key('hg.create.none')
         p2 = Permission.get_by_key('hg.usergroup.create.false')
         p3 = Permission.get_by_key('hg.fork.none')
@@ @@ -116,13 +118,14 @@ class TestAdminUsersGroupsController(Tes @@
                     [ug.users_group_id, p2.permission_id],
                     [ug.users_group_id, p3.permission_id]]))
         # DELETE !
         ug = UserGroup.get_by_group_name(users_group_name)
         ugid = ug.users_group_id
         response = self.app.delete(url('users_group', id=ug.users_group_id))
         response = self.app.post(url('users_group', id=ug.users_group_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         response = response.follow()
         gr = Session().query(UserGroup)\
             .filter(UserGroup.users_group_name == users_group_name).scalar()
         self.assertEqual(gr, None)
         p = Permission.get_by_key('hg.create.repository')
@@ @@ -164,14 +167,14 @@ class TestAdminUsersGroupsController(Tes @@
             sorted([[x.users_group_id, x.permission_id, ] for x in perms]),
             sorted([[ug.users_group_id, p.permission_id],
                     [ug.users_group_id, p2.permission_id],
                     [ug.users_group_id, p3.permission_id]]))
         ## DISABLE REPO CREATE ON A GROUP
         response = self.app.put(
             url('edit_user_group_default_perms', id=ug.users_group_id), {})
         response = self.app.put(url('edit_user_group_default_perms', id=ug.users_group_id),
             params={'_authentication_token': self.authentication_token()})
         response.follow()
         ug = UserGroup.get_by_group_name(users_group_name)
         p = Permission.get_by_key('hg.create.none')
         p2 = Permission.get_by_key('hg.usergroup.create.false')
         p3 = Permission.get_by_key('hg.fork.none')
@@ @@ -186,13 +189,14 @@ class TestAdminUsersGroupsController(Tes @@
                     [ug.users_group_id, p2.permission_id],
                     [ug.users_group_id, p3.permission_id]]))
         # DELETE !
         ug = UserGroup.get_by_group_name(users_group_name)
         ugid = ug.users_group_id
         response = self.app.delete(url('users_group', id=ug.users_group_id))
         response = self.app.post(url('users_group', id=ug.users_group_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         response = response.follow()
         gr = Session().query(UserGroup)\
                            .filter(UserGroup.users_group_name ==
                                    users_group_name).scalar()
         self.assertEqual(gr, None)

kallithea/tests/functional/test_admin_users.py

➞

Show inline comments

@@ @@ -164,13 +164,14 @@ class TestAdminUsersController(TestContr @@
         username = 'newtestuserdeleteme'
         fixture.create_user(name=username)
         new_user = Session().query(User)\
             .filter(User.username == username).one()
         response = self.app.delete(url('user', id=new_user.user_id))
         response = self.app.post(url('user', id=new_user.user_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'Successfully deleted user')
     def test_delete_repo_err(self):
         self.log_user()
         username = 'repoerr'
@@ @@ -178,73 +179,81 @@ class TestAdminUsersController(TestContr @@
         fixture.create_user(name=username)
         fixture.create_repo(name=reponame, cur_user=username)
         new_user = Session().query(User)\
             .filter(User.username == username).one()
         response = self.app.delete(url('user', id=new_user.user_id))
         response = self.app.post(url('user', id=new_user.user_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'User "%s" still '
                                'owns 1 repositories and cannot be removed. '
                                'Switch owners or remove those repositories: '
                                '%s' % (username, reponame))
         response = self.app.delete(url('delete_repo', repo_name=reponame))
         response = self.app.post(url('delete_repo', repo_name=reponame),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'Deleted repository %s' % reponame)
         response = self.app.delete(url('user', id=new_user.user_id))
         response = self.app.post(url('user', id=new_user.user_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'Successfully deleted user')
     def test_delete_repo_group_err(self):
         self.log_user()
         username = 'repogrouperr'
         groupname = 'repogroup_fail'
         fixture.create_user(name=username)
         fixture.create_repo_group(name=groupname, cur_user=username)
         new_user = Session().query(User)\
             .filter(User.username == username).one()
         response = self.app.delete(url('user', id=new_user.user_id))
         response = self.app.post(url('user', id=new_user.user_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'User "%s" still '
                                'owns 1 repository groups and cannot be removed. '
                                'Switch owners or remove those repository groups: '
                                '%s' % (username, groupname))
         # Relevant _if_ the user deletion succeeded to make sure we can render groups without owner
         # rg = RepoGroup.get_by_group_name(group_name=groupname)
         # response = self.app.get(url('repos_groups', id=rg.group_id))
         response = self.app.delete(url('delete_repo_group', group_name=groupname))
         response = self.app.post(url('delete_repo_group', group_name=groupname),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'Removed repository group %s' % groupname)
         response = self.app.delete(url('user', id=new_user.user_id))
         response = self.app.post(url('user', id=new_user.user_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'Successfully deleted user')
     def test_delete_user_group_err(self):
         self.log_user()
         username = 'usergrouperr'
         groupname = 'usergroup_fail'
         fixture.create_user(name=username)
         ug = fixture.create_user_group(name=groupname, cur_user=username)
         new_user = Session().query(User)\
             .filter(User.username == username).one()
         response = self.app.delete(url('user', id=new_user.user_id))
         response = self.app.post(url('user', id=new_user.user_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'User "%s" still '
                                'owns 1 user groups and cannot be removed. '
                                'Switch owners or remove those user groups: '
                                '%s' % (username, groupname))
         # TODO: why do this fail?
         #response = self.app.delete(url('delete_users_group', id=groupname))
         #self.checkSessionFlash(response, 'Removed user group %s' % groupname)
         fixture.destroy_user_group(ug.users_group_id)
         response = self.app.delete(url('user', id=new_user.user_id))
         response = self.app.post(url('user', id=new_user.user_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'Successfully deleted user')
     def test_show(self):
         response = self.app.get(url('user', id=1))
     def test_edit(self):

kallithea/tests/functional/test_changeset_comments.py

➞

Show inline comments

@@ @@ -129,16 +129,17 @@ class TestChangeSetCommentsController(Te @@
                                      params=params)
         comments = ChangesetComment.query().all()
         self.assertEqual(len(comments), 1)
         comment_id = comments[0].comment_id
-        self.app.delete(url(controller='changeset',
+        self.app.post(url(controller='changeset',
                                     action='delete_comment',
                                     repo_name=HG_REPO,
                                     comment_id=comment_id))
                                     comment_id=comment_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         comments = ChangesetComment.query().all()
         self.assertEqual(len(comments), 0)
         response = self.app.get(url(controller='changeset', action='index',
                                 repo_name=HG_REPO, revision=rev))

kallithea/tests/functional/test_forks.py

➞

Show inline comments

@@ @@ -93,13 +93,14 @@ class _BaseTestCase(object): @@
         response.mustcontain(
             """<a href="/%s">%s</a>""" % (fork_name, fork_name)
+        )
         # remove this fork
         response = self.app.delete(url('delete_repo', repo_name=fork_name))
         response = self.app.post(url('delete_repo', repo_name=fork_name),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
     def test_fork_create_into_group(self):
         self.log_user()
         group = fixture.create_repo_group('vc')
         group_id = group.group_id
         fork_name = self.REPO_FORK

kallithea/tests/functional/test_my_account.py

➞

Show inline comments

@@ @@ -54,13 +54,14 @@ class TestMyAccountController(TestContro @@
         self.checkSessionFlash(response, 'This email address is already in use')
     def test_my_account_my_emails_add_mising_email_in_form(self):
         self.log_user()
         response = self.app.get(url('my_account_emails'))
         response.mustcontain('No additional emails specified')
         response = self.app.post(url('my_account_emails'),)
         response = self.app.post(url('my_account_emails'),
             {'_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'Please enter an email address')
     def test_my_account_my_emails_add_remove(self):
         self.log_user()
         response = self.app.get(url('my_account_emails'))
         response.mustcontain('No additional emails specified')

0 comments (0 inline, 0 general)