CVE-2022-3232: Cross-Site Request Forgery (CSRF) (CWE-352)
value: Cross-Site Request Forgery (CSRF) in GitHub repository ikus060/rdiffweb prior to 2.4.5.
CWE description: The web application does not, or can not, sufficiently verify whether a well-formed, valid, consistent request was intentionally provided by the user who submitted the request.
CVSS v3 base score: 4.3
Published: 2022-09-17T20:15Z
Repository: https://github.com/ikus060/rdiffweb
Language: Python
Files changed: README.md, rdiffweb/controller/page_admin.py, rdiffweb/controller/page_delete.py, rdiffweb/controller/pref_general.py, rdiffweb/controller/tests/test_page_admin.py, rdiffweb/controller/tests/test_page_delete.py
Commit date: 2022-09-16 15:56:28 -0400
Commit message: Mitigate CSRF on repository deletion and user deletion #214 #215

Diff: diff --git a/README.md b/README.md
index b8bd9a47..84a0bfa5 100644
--- a/README.md
+++ b/README.md
@@ -105,21 +105,27 @@ Bug reports should be reported on the Rdiffweb Gitlab at https://gitlab.com/ikus
 
 Professional support for Rdiffweb is available by contacting [IKUS Soft](https://www.ikus-soft.com/en/support/#form).
 
 # 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.
 
 * Use `X-Real-IP` to identify client IP address to mitigate Brute-Force attack #213
 
 ## 2.4.3 (2022-09-14)
 
 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)
 
 This releases include a security fix. If you are using an earlier version, you should upgrade to this release immediately.
 
diff --git a/rdiffweb/controller/page_admin.py b/rdiffweb/controller/page_admin.py
index dc367fc3..b67100d5 100644
--- a/rdiffweb/controller/page_admin.py
+++ b/rdiffweb/controller/page_admin.py
@@ -188,19 +188,17 @@ class UserForm(CherryForm):
     )
     disk_usage = SizeField(
         _('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):
         return cherrypy.request.app
 
@@ -338,11 +336,13 @@ class AdminPage(Controller):
                 else:
                     flash(form.error_message, level='error')
             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),
             "edit_form": EditUserForm(formdata=None),
             "criteria": criteria,
diff --git a/rdiffweb/controller/page_delete.py b/rdiffweb/controller/page_delete.py
index 7fb31ca6..cd2f97fa 100644
--- a/rdiffweb/controller/page_delete.py
+++ b/rdiffweb/controller/page_delete.py
@@ -20,28 +20,33 @@ Created on Apr. 5, 2021
 @author: Patrik Dufresne <patrik@ikus-soft.com>
 '''
 # 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()
 class DeletePage(Controller):
     @cherrypy.expose
@@ -59,17 +64,19 @@ class DeletePage(Controller):
         # Check user's permissions
         is_maintainer()
 
         # 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)
diff --git a/rdiffweb/controller/pref_general.py b/rdiffweb/controller/pref_general.py
index 2720b6b1..3bc01ee9 100644
--- a/rdiffweb/controller/pref_general.py
+++ b/rdiffweb/controller/pref_general.py
@@ -52,19 +52,17 @@ class UserPasswordForm(CherryForm):
     )
     confirm = PasswordField(
         _('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):
         return cherrypy.request.app
 
diff --git a/rdiffweb/controller/tests/test_page_admin.py b/rdiffweb/controller/tests/test_page_admin.py
index 76d1d8fd..aa7b1cc5 100644
--- a/rdiffweb/controller/tests/test_page_admin.py
+++ b/rdiffweb/controller/tests/test_page_admin.py
@@ -265,10 +265,20 @@ class AdminUsersAsAdminTest(AbstractAdminTest):
         # Try deleting admin user
         self._delete_user(self.USERNAME)
         self.assertStatus(200)
         self.assertInBody("can&#39;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.")
 
     def test_change_password_with_too_long(self):
diff --git a/rdiffweb/controller/tests/test_page_delete.py b/rdiffweb/controller/tests/test_page_delete.py
index 1eb23731..f8f41a8d 100644
--- a/rdiffweb/controller/tests/test_page_delete.py
+++ b/rdiffweb/controller/tests/test_page_delete.py
@@ -35,37 +35,63 @@ RDIFF_BACKUP_VERSION = rdiff_backup_version()
 
 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)
         self.assertStatus(expected_history_status)
 
@@ -76,10 +102,11 @@ class DeleteRepoTest(rdiffweb.test.WebCase):
         # Check initial list of repo
         self.assertEqual(['broker-repo', 'testcases'], [r.name for r in self.app.store.get_user('admin').repo_objs])
         # 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])
         self.assertFalse(os.path.isdir(os.path.join(self.testcases, 'testcases')))
 
@@ -87,10 +114,11 @@ class DeleteRepoTest(rdiffweb.test.WebCase):
         # Check initial list of repo
         self.assertEqual(['broker-repo', 'testcases'], [r.name for r in self.app.store.get_user('admin').repo_objs])
         # 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])
         self.assertFalse(os.path.isdir(os.path.join(self.testcases, 'testcases')))
 
@@ -123,14 +151,13 @@ class DeleteRepoTest(rdiffweb.test.WebCase):
         # Create a another user with admin right
         user_obj = self.app.store.add_user('anotheruser', 'password')
         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)
         self.assertEqual(['broker-repo'], [r.name for r in user_obj.repo_objs])
         self.assertFalse(os.path.isdir(os.path.join(self.testcases, 'testcases')))
@@ -145,15 +172,14 @@ class DeleteRepoTest(rdiffweb.test.WebCase):
         self.assertEqual(['broker-repo', 'testcases'], [r.name for r in user_obj.repo_objs])
 
         # 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)
         self.assertEqual(['broker-repo'], [r.name for r in user_obj.repo_objs])
         self.assertFalse(os.path.isdir(os.path.join(self.testcases, 'testcases')))
@@ -167,11 +193,11 @@ class DeleteRepoTest(rdiffweb.test.WebCase):
 
         # Login as maintainer
         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
         self.assertEqual(['broker-repo', 'testcases'], [r.name for r in user_obj.repo_objs])
         self.assertTrue(os.path.isdir(os.path.join(self.testcases, 'testcases')))
@@ -181,5 +207,15 @@ class DeleteRepoTest(rdiffweb.test.WebCase):
         repo = 'invalid'
         # When trying to delete this repo
         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])

Vulnerable code: * Mitigate CSRF in profile's SSH Keys #212
    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.'),
            )
        ]
            self._delete_user(action, DeleteUserForm())
from wtforms import validators
    confirm = StringField(_('Confirmation'), validators=[validators.data_required()])
    redirect = StringField(default='/')
        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)
    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 _delete(self, user, repo, confirm, redirect=None):
        if redirect is not None:
            body['redirect'] = redirect
            ("with_dir", 'admin', '/testcases/Revisions', 'Revisions', 303, 404),
            ("with_file", 'admin', '/testcases/Revisions/Data', 'Data', 303, 404),
            ("with_broken_symlink", 'admin', '/testcases/BrokenSymlink', 'BrokenSymlink', 303, 404),
            ("with_utf8", 'admin', '/testcases/R%C3%A9pertoire%20Existant', 'Répertoire Existant', 303, 404),
            ("with_quoted_path", 'admin', '/testcases/Char%20%3B090%20to%20quote', 'Char Z to quote', 303, 404),
    def test_delete_path(self, unused, username, path, confirmation, expected_status, expected_history_status):
        self._delete('anotheruser', 'testcases', 'testcases', redirect='/admin/repos/')
        location = self.assertHeader('Location')
        self.assertTrue(location.endswith('/admin/repos/'))
        # Try to delete own own repo
        self._delete('maintainer', 'testcases', 'testcases', redirect='/admin/repos/')
        location = self.assertHeader('Location')
        self.assertTrue(location.endswith('/admin/repos/'))
        self._delete('user', 'testcases', 'testcases', redirect='/admin/repos/')
Fixed code: ## 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

* Mitigate CSRF in profile's SSH Keys [CVE-2022-3221](https://nvd.nist.gov/vuln/detail/CVE-2022-3221) #212
    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)
            form = DeleteUserForm()
            if form.validate_on_submit():
                self._delete_user(action, form)
import os
from wtforms.validators import DataRequired, ValidationError
from rdiffweb.core.rdw_templating import url_for
    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)
        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)
    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)
    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 _delete(self, user, repo, confirm):
            ("with_dir", 'admin', '/testcases/Revisions', 'Revisions', 303, 404, '/browse/admin/testcases'),
            ("with_file", 'admin', '/testcases/Revisions/Data', 'Data', 303, 404, '/browse/admin/testcases/Revisions'),
            (
                "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_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, expected_redirect=None
    ):
        if expected_redirect:
            self.assertHeaderItemValue('Location', self.baseurl + expected_redirect)
        self.assertHeaderItemValue('Location', self.baseurl + '/')
        self.assertHeaderItemValue('Location', self.baseurl + '/')
        self._delete('anotheruser', 'testcases', 'testcases')
        self.assertHeaderItemValue('Location', self.baseurl + '/')
        # Try to delete your own repo
        self._delete('maintainer', 'testcases', 'testcases')
        self.assertHeaderItemValue('Location', self.baseurl + '/')
        self._delete('user', 'testcases', 'testcases')

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