Merge commit '422791ea45713aaaa865bdca74addb9fffd93a71'

Author: ikus060

README.md

@@ -136,6 +136,12 @@ Breaking changes:
 * `session-dir` is deprecated and should be replace by `rate-limit-dir`. User's session are stored in database.
 * previous `.css` customization are not barkward compatible
 
+## 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.
@@ -146,7 +152,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_users.py

@@ -233,7 +233,9 @@ def default(self, username=None, 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,25 +22,30 @@
 # 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.dispatch import poppath
 from rdiffweb.controller.filter_authorization import is_maintainer
 from rdiffweb.controller.form import CherryForm
 from rdiffweb.core.librdiff import AccessDeniedError, DoesNotExistError
 from rdiffweb.core.model import RepoObject
+from rdiffweb.core.rdw_templating import url_for
 from rdiffweb.tools.i18n import gettext_lazy 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()
@@ -62,20 +67,18 @@ 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.
-        repo.expire()
-        scheduled = cherrypy.engine.publish('schedule_task', self.delete_repo_path, repo.repoid, path)
-        assert scheduled
-        raise cherrypy.HTTPRedirect(form.redirect.data)
 
-    def delete_repo_path(self, repoid, path):
-        repo = RepoObject.query.filter(RepoObject.repoid == repoid).first()
-        repo.delete(path)
+        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/tests/test_page_admin_users.py

@@ -278,6 +278,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
+        UserObject.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(UserObject.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

@@ -35,33 +35,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)
 
     @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',
+            ),
         ]
     )
     @skipIf(rdiff_backup_version() < (2, 0, 1), "rdiff-backup-delete is available since 2.0.1")
-    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)
@@ -77,6 +103,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)
         userobj.expire()
@@ -90,6 +117,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)
         userobj.expire()
@@ -132,10 +160,9 @@ def test_delete_repo_as_admin(self):
         user_obj.refresh_repos()
         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)
@@ -156,11 +183,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)
@@ -179,8 +205,8 @@ def test_delete_repo_as_user(self):
         # Login as maintainer
         self._login('user', 'password')
 
-        # Try to delete our own repo
-        self._delete('user', 'testcases', 'testcases', redirect='/admin/repos/')
+        # Try to delete own own repo
+        self._delete('user', 'testcases', 'testcases')
         self.assertStatus(403)
 
         # Check database don't change
@@ -194,3 +220,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 UserObject.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 UserObject.get_user('admin').repo_objs])