Mitigate CSRF on repository deletion and user deletion #214 #215

Author: ikus060

README.md

@@ -107,6 +107,12 @@ Professional support for Rdiffweb is available by contacting [IKUS Soft](https:/
 
 # Changelog
 
+## 2.4.5 (2002-09-16)
+
+This releases include a security fix. If you are using an earlier version, you should upgrade to this release immediately.
+
+* Mitigate CSRF on repository deletion and user deletion [CVE-2022-3232](https://nvd.nist.gov/vuln/detail/CVE-2022-3232) #214 #215
+
 ## 2.4.4 (2002-09-15)
 
 This releases include a security fix. If you are using an earlier version, you should upgrade to this release immediately.
@@ -117,7 +123,7 @@ This releases include a security fix. If you are using an earlier version, you s
 
 This releases include a security fix. If you are using an earlier version, you should upgrade to this release immediately.
 
-* Mitigate CSRF in profile's SSH Keys #212
+* Mitigate CSRF in profile's SSH Keys [CVE-2022-3221](https://nvd.nist.gov/vuln/detail/CVE-2022-3221) #212
 
 ## 2.4.2 (2022-09-12)
 

rdiffweb/controller/page_admin.py

@@ -190,15 +190,13 @@ class UserForm(CherryForm):
         _('Quota Used'), validators=[validators.optional()], description=_("Disk spaces (in bytes) used by this user.")
     )
 
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.password.validators += [
-            validators.length(
-                min=self.app.cfg.password_min_length,
-                max=self.app.cfg.password_max_length,
-                message=_('Password must have between %(min)d and %(max)d characters.'),
-            )
-        ]
+    def validate_password(self, field):
+        validator = validators.length(
+            min=self.app.cfg.password_min_length,
+            max=self.app.cfg.password_max_length,
+            message=_('Password must have between %(min)d and %(max)d characters.'),
+        )
+        validator(self, field)
 
     @property
     def app(self):
@@ -340,7 +338,9 @@ def users(self, username=None, criteria=u"", search=u"", action=u"", **kwargs):
             else:
                 flash(_("Cannot edit user `%s`: user doesn't exists") % username, level='error')
         elif action == 'delete':
-            self._delete_user(action, DeleteUserForm())
+            form = DeleteUserForm()
+            if form.validate_on_submit():
+                self._delete_user(action, form)
 
         params = {
             "add_form": UserForm(formdata=None),

rdiffweb/controller/page_delete.py

@@ -22,24 +22,29 @@
 # Define the logger
 
 import logging
+import os
 
 import cherrypy
-from wtforms import validators
 from wtforms.fields.core import StringField
+from wtforms.validators import DataRequired, ValidationError
 
 from rdiffweb.controller import Controller
 from rdiffweb.controller.cherrypy_wtf import CherryForm
 from rdiffweb.controller.dispatch import poppath
 from rdiffweb.controller.filter_authorization import is_maintainer
 from rdiffweb.core.librdiff import AccessDeniedError, DoesNotExistError
+from rdiffweb.core.rdw_templating import url_for
 from rdiffweb.tools.i18n import ugettext as _
 
 _logger = logging.getLogger(__name__)
 
 
 class DeleteRepoForm(CherryForm):
-    confirm = StringField(_('Confirmation'), validators=[validators.data_required()])
-    redirect = StringField(default='/')
+    confirm = StringField(_('Confirmation'), validators=[DataRequired()])
+
+    def validate_confirm(self, field):
+        if self.confirm.data != self.expected_confirm:
+            raise ValidationError(_('Invalid value, must be: %s') % self.expected_confirm)
 
 
 @poppath()
@@ -61,15 +66,17 @@ def default(self, path=b"", **kwargs):
 
         # validate form
         form = DeleteRepoForm()
-        if not form.validate():
-            raise cherrypy.HTTPError(400, form.error_message)
-
-        # Validate the name
-        if form.confirm.data != path_obj.display_name:
-            _logger.info("do not delete repo, bad confirmation %r != %r", form.confirm.data, path_obj.display_name)
-            raise cherrypy.HTTPError(400, 'bad confirmation')
-
-        # Delete repository in background using a schedule task.
-        scheduled = cherrypy.engine.publish('schedule_task', repo.delete, path)
-        assert scheduled
-        raise cherrypy.HTTPRedirect(form.redirect.data)
+        form.expected_confirm = path_obj.display_name
+        if form.is_submitted():
+            if form.validate():
+                cherrypy.engine.publish('schedule_task', repo.delete, path)
+                # Redirect to parent folder or to root if repo get deleted
+                if path_obj.isroot:
+                    raise cherrypy.HTTPRedirect(url_for('/'))
+                else:
+                    parent_path = repo.fstat(os.path.dirname(path_obj.path))
+                    raise cherrypy.HTTPRedirect(url_for('browse', repo, parent_path))
+            else:
+                raise cherrypy.HTTPError(400, form.error_message)
+        else:
+            raise cherrypy.HTTPError(405)

rdiffweb/controller/pref_general.py

@@ -54,15 +54,13 @@ class UserPasswordForm(CherryForm):
         _('Confirm new password'), validators=[InputRequired(_("Confirmation password is missing."))]
     )
 
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.new.validators += [
-            Length(
-                min=self.app.cfg.password_min_length,
-                max=self.app.cfg.password_max_length,
-                message=_('Password must have between %(min)d and %(max)d characters.'),
-            )
-        ]
+    def validate_new(self, field):
+        validator = Length(
+            min=self.app.cfg.password_min_length,
+            max=self.app.cfg.password_max_length,
+            message=_('Password must have between %(min)d and %(max)d characters.'),
+        )
+        validator(self, field)
 
     @property
     def app(self):

rdiffweb/controller/tests/test_page_admin.py

@@ -267,6 +267,16 @@ def test_delete_user_admin(self):
         self.assertStatus(200)
         self.assertInBody("can't delete admin user")
 
+    def test_delete_user_method_get(self):
+        # Given a user
+        self.app.store.add_user('newuser')
+        # When trying to delete this user using method GET
+        self.getPage("/admin/users/?action=delete&username=newuser", method='GET')
+        # Then page return without error
+        self.assertStatus(200)
+        # Then user is not deleted
+        self.assertIsNotNone(self.app.store.get_user('newuser'))
+
     def test_change_password_with_too_short(self):
         self._edit_user(self.USERNAME, password='short')
         self.assertInBody("Password must have between 8 and 128 characters.")

rdiffweb/controller/tests/test_page_delete.py

@@ -37,33 +37,59 @@ class DeleteRepoTest(rdiffweb.test.WebCase):
 
     login = True
 
-    def _delete(self, user, repo, confirm, redirect=None):
+    def _delete(self, user, repo, confirm):
         body = {}
         if confirm is not None:
             body.update({'confirm': confirm})
-        if redirect is not None:
-            body['redirect'] = redirect
         self.getPage("/delete/" + user + "/" + repo + "/", method="POST", body=body)
 
     @skipIf(RDIFF_BACKUP_VERSION < (2, 0, 1), "rdiff-backup-delete is available since 2.0.1")
     @parameterized.expand(
         [
-            ("with_dir", 'admin', '/testcases/Revisions', 'Revisions', 303, 404),
+            ("with_dir", 'admin', '/testcases/Revisions', 'Revisions', 303, 404, '/browse/admin/testcases'),
             ("with_dir_wrong_confirmation", 'admin', '/testcases/Revisions', 'invalid', 400, 200),
-            ("with_file", 'admin', '/testcases/Revisions/Data', 'Data', 303, 404),
+            ("with_file", 'admin', '/testcases/Revisions/Data', 'Data', 303, 404, '/browse/admin/testcases/Revisions'),
             ("with_file_wrong_confirmation", 'admin', '/testcases/Revisions/Data', 'invalid', 400, 200),
             ("with_invalid", 'admin', '/testcases/invalid', 'invalid', 404, 404),
-            ("with_broken_symlink", 'admin', '/testcases/BrokenSymlink', 'BrokenSymlink', 303, 404),
-            ("with_utf8", 'admin', '/testcases/R%C3%A9pertoire%20Existant', 'Répertoire Existant', 303, 404),
+            (
+                "with_broken_symlink",
+                'admin',
+                '/testcases/BrokenSymlink',
+                'BrokenSymlink',
+                303,
+                404,
+                '/browse/admin/testcases',
+            ),
+            (
+                "with_utf8",
+                'admin',
+                '/testcases/R%C3%A9pertoire%20Existant',
+                'Répertoire Existant',
+                303,
+                404,
+                '/browse/admin/testcases',
+            ),
             ("with_rdiff_backup_data", 'admin', '/testcases/rdiff-backup-data', 'rdiff-backup-data', 404, 404),
-            ("with_quoted_path", 'admin', '/testcases/Char%20%3B090%20to%20quote', 'Char Z to quote', 303, 404),
+            (
+                "with_quoted_path",
+                'admin',
+                '/testcases/Char%20%3B090%20to%20quote',
+                'Char Z to quote',
+                303,
+                404,
+                '/browse/admin/testcases',
+            ),
         ]
     )
-    def test_delete_path(self, unused, username, path, confirmation, expected_status, expected_history_status):
+    def test_delete_path(
+        self, unused, username, path, confirmation, expected_status, expected_history_status, expected_redirect=None
+    ):
         # When trying to delete a file or a folder with a confirmation
         self._delete(username, path, confirmation)
         # Then a status is returned
         self.assertStatus(expected_status)
+        if expected_redirect:
+            self.assertHeaderItemValue('Location', self.baseurl + expected_redirect)
         # Check filesystem
         sleep(1)
         self.getPage("/history/" + username + "/" + path)
@@ -78,6 +104,7 @@ def test_delete_repo(self):
         # Delete repo
         self._delete(self.USERNAME, self.REPO, 'testcases')
         self.assertStatus(303)
+        self.assertHeaderItemValue('Location', self.baseurl + '/')
         # Check filesystem
         sleep(1)
         self.assertEqual(['broker-repo'], [r.name for r in self.app.store.get_user('admin').repo_objs])
@@ -89,6 +116,7 @@ def test_delete_repo_with_slash(self):
         # Then delete it.
         self._delete(self.USERNAME, self.REPO, 'testcases')
         self.assertStatus(303)
+        self.assertHeaderItemValue('Location', self.baseurl + '/')
         # Check filesystem
         sleep(1)
         self.assertEqual(['broker-repo'], [r.name for r in self.app.store.get_user('admin').repo_objs])
@@ -125,10 +153,9 @@ def test_delete_repo_as_admin(self):
         user_obj.user_root = self.testcases
         self.assertEqual(['broker-repo', 'testcases'], [r.name for r in user_obj.repo_objs])
 
-        self._delete('anotheruser', 'testcases', 'testcases', redirect='/admin/repos/')
+        self._delete('anotheruser', 'testcases', 'testcases')
         self.assertStatus(303)
-        location = self.assertHeader('Location')
-        self.assertTrue(location.endswith('/admin/repos/'))
+        self.assertHeaderItemValue('Location', self.baseurl + '/')
 
         # Check filesystem
         sleep(1)
@@ -147,11 +174,10 @@ def test_delete_repo_as_maintainer(self):
         # Login as maintainer
         self._login('maintainer', 'password')
 
-        # Try to delete own own repo
-        self._delete('maintainer', 'testcases', 'testcases', redirect='/admin/repos/')
+        # Try to delete your own repo
+        self._delete('maintainer', 'testcases', 'testcases')
         self.assertStatus(303)
-        location = self.assertHeader('Location')
-        self.assertTrue(location.endswith('/admin/repos/'))
+        self.assertHeaderItemValue('Location', self.baseurl + '/')
 
         # Check filesystem
         sleep(1)
@@ -169,7 +195,7 @@ def test_delete_repo_as_user(self):
         self._login('user', 'password')
 
         # Try to delete own own repo
-        self._delete('user', 'testcases', 'testcases', redirect='/admin/repos/')
+        self._delete('user', 'testcases', 'testcases')
         self.assertStatus(403)
 
         # Check database don't change
@@ -183,3 +209,13 @@ def test_delete_repo_does_not_exists(self):
         self._delete(self.USERNAME, repo, repo)
         # Then a 404 is return to the user
         self.assertStatus(404)
+
+    def test_delete_method_get(self):
+        # Given a user with repo
+        self.assertEqual(['broker-repo', 'testcases'], [r.name for r in self.app.store.get_user('admin').repo_objs])
+        # When trying to deleted repo with GET method
+        self.getPage("/delete/" + self.USERNAME + "/" + self.REPO + "/?confirm=" + self.REPO, method="GET")
+        # Then An error is returned
+        self.assertStatus(405)
+        # Then repo still exists
+        self.assertEqual(['broker-repo', 'testcases'], [r.name for r in self.app.store.get_user('admin').repo_objs])