Changeset - a041321d2aa1
[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:
0 comments (0 inline, 0 general)
kallithea/lib/auth.py
Show inline comments
 
@@ -760,8 +760,13 @@ class LoginRequired(object):
 
                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')
kallithea/tests/functional/test_admin_defaults.py
Show inline comments
 
@@ -16,7 +16,8 @@ class TestDefaultsController(TestControl
 
        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'))
 
@@ -62,7 +63,8 @@ class TestDefaultsController(TestControl
 
        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()))
kallithea/tests/functional/test_admin_gists.py
Show inline comments
 
@@ -136,19 +136,20 @@ class TestGistsController(TestController
 
    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')
kallithea/tests/functional/test_admin_notifications.py
Show inline comments
 
@@ -56,9 +56,9 @@ class TestNotificationsController(TestCo
 
        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)
kallithea/tests/functional/test_admin_repos.py
Show inline comments
 
@@ -398,7 +398,8 @@ class _BaseTest(object):
 
        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))
 

	
 
@@ -450,7 +451,8 @@ class _BaseTest(object):
 
        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()
 

	
kallithea/tests/functional/test_admin_user_groups.py
Show inline comments
 
@@ -32,7 +32,7 @@ class TestAdminUsersGroupsController(Tes
 
        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),
 
@@ -54,7 +54,8 @@ class TestAdminUsersGroupsController(Tes
 
        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()
 
@@ -97,7 +98,8 @@ class TestAdminUsersGroupsController(Tes
 

	
 
        ## 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)
 
@@ -119,7 +121,8 @@ class TestAdminUsersGroupsController(Tes
 
        # 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()
 
@@ -167,8 +170,8 @@ class TestAdminUsersGroupsController(Tes
 
                    [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)
 
@@ -189,7 +192,8 @@ class TestAdminUsersGroupsController(Tes
 
        # 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 ==
kallithea/tests/functional/test_admin_users.py
Show inline comments
 
@@ -167,7 +167,8 @@ class TestAdminUsersController(TestContr
 

	
 
        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')
 

	
 
@@ -181,16 +182,19 @@ class TestAdminUsersController(TestContr
 

	
 
        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):
 
@@ -203,7 +207,8 @@ class TestAdminUsersController(TestContr
 

	
 
        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: '
 
@@ -213,10 +218,12 @@ class TestAdminUsersController(TestContr
 
        # 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):
 
@@ -229,7 +236,8 @@ class TestAdminUsersController(TestContr
 

	
 
        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: '
 
@@ -241,7 +249,8 @@ class TestAdminUsersController(TestContr
 

	
 
        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):
kallithea/tests/functional/test_changeset_comments.py
Show inline comments
 
@@ -132,10 +132,11 @@ class TestChangeSetCommentsController(Te
 
        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)
kallithea/tests/functional/test_forks.py
Show inline comments
 
@@ -96,7 +96,8 @@ class _BaseTestCase(object):
 
        )
 

	
 
        # 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()
kallithea/tests/functional/test_my_account.py
Show inline comments
 
@@ -57,7 +57,8 @@ class TestMyAccountController(TestContro
 
        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):
0 comments (0 inline, 0 general)