fix the CSRF token on delete files and directories

flask_admin/templates/bootstrap4/admin/file/list.html

@@ -86,8 +86,11 @@
                         {% if name != '..' and admin_view.can_delete_dirs %}
                         <form class="icon" method="POST" action="{{ get_url('.delete') }}">
                             {{ delete_form.path(value=path) }}
+
                             {% if delete_form.csrf_token is defined and delete_form.csrf_token %}
-                            {{ delete_form.csrf_token }}
+                              {{ delete_form.csrf_token }}
+                            {% elif csrf_token is defined and csrf_token %}
+                              <input name="csrf_token" type="hidden" value="{{ csrf_token() }}"/>
                             {% endif %}
                             <button onclick="return confirm('{{ _gettext('Are you sure you want to delete \\\'%(name)s\\\' recursively?', name=name) }}')">
                                 <i class="fa fa-times glyphicon glyphicon-remove"></i>
@@ -97,8 +100,11 @@
                     {% else %}
                     <form class="icon" method="POST" action="{{ get_url('.delete') }}">
                         {{ delete_form.path(value=path) }}
+
                         {% if delete_form.csrf_token is defined and delete_form.csrf_token %}
-                        {{ delete_form.csrf_token }}
+                          {{ delete_form.csrf_token }}
+                        {% elif csrf_token is defined and csrf_token %}
+                          <input name="csrf_token" type="hidden" value="{{ csrf_token() }}"/>
                         {% endif %}
                         <button onclick="return confirm('{{ _gettext('Are you sure you want to delete \\\'%(name)s\\\'?', name=name) }}')">
                             <i class="fa fa-trash glyphicon glyphicon-trash"></i>

flask_admin/tests/fileadmin/test_fileadmin.py

@@ -2,8 +2,11 @@
 import os.path as op
 from io import BytesIO
 
+import pytest
+
 from flask_admin import Admin
 from flask_admin.contrib import fileadmin
+from flask_admin.form import SecureForm
 from flask_admin.theme import Bootstrap4Theme
 
 
@@ -222,6 +225,174 @@ class EditModalOff(fileadmin_class):  # type: ignore[valid-type, misc]
             data = rv.data.decode("utf-8")
             assert "fa_modal_window" not in data
 
+        def get_csrf_token(self, data):
+            data = data.split('name="csrf_token" type="hidden" value="')[1]
+            token = data.split('"')[0]
+            return token
+
+        @pytest.mark.parametrize(
+            "page",
+            [
+                "/admin/fileadmin/",
+                "/admin/fileadmin/upload",
+                "/admin/fileadmin/rename/?path=dummy.txt",
+                "/admin/fileadmin/rename/?path=d1",
+                "/admin/fileadmin/mkdir",
+            ],
+        )
+        def test_csrf_token(self, app, admin, page):
+            # Cross-Site-Request-Forgery (CSRF) Protection
+            app.config["WTF_CSRF_ENABLED"] = True
+
+            os.makedirs(op.join(self._test_files_root, "d1"), exist_ok=True)
+
+            fileadmin_class = self.fileadmin_class()
+            fileadmin_args, fileadmin_kwargs = self.fileadmin_args()
+
+            class SecureFileAdmin(fileadmin_class):  # type: ignore[valid-type, misc]
+                form_base_class = SecureForm
+
+            fileadmin_kwargs["endpoint"] = "fileadmin"
+            view = SecureFileAdmin(*fileadmin_args, **fileadmin_kwargs)
+            admin.add_view(view)
+
+            client = app.test_client()
+
+            rv = client.get(page, follow_redirects=True)
+            data = rv.data.decode("utf-8")
+            assert rv.status_code == 200
+            assert 'name="csrf_token"' in data
+            assert len(self.get_csrf_token(data)) == 56
+
+        def test_csrf_submit(self, app, admin, request):
+            # Cross-Site-Request-Forgery (CSRF) Protection
+            app.config["WTF_CSRF_ENABLED"] = True
+
+            def finalizer():
+                try:
+                    os.remove(op.join(self._test_files_root, "d1/dum.txt"))
+                    os.remove(op.join(self._test_files_root, "dummy_renamed.txt"))
+                    os.remove(op.join(self._test_files_root, "dummy2.txt"))
+                except OSError:
+                    pass
+
+            request.addfinalizer(finalizer)
+
+            fileadmin_class = self.fileadmin_class()
+            fileadmin_args, fileadmin_kwargs = self.fileadmin_args()
+
+            class SecureFileAdmin(fileadmin_class):  # type: ignore[valid-type, misc]
+                form_base_class = SecureForm
+
+                def is_accessible(self):
+                    return True
+
+            fileadmin_kwargs["endpoint"] = "myfileadmin"
+            view = SecureFileAdmin(*fileadmin_args, **fileadmin_kwargs)
+            admin.add_view(view)
+
+            client = app.test_client()
+
+            assert "dummy.txt" in client.get(
+                "/admin/myfileadmin/", follow_redirects=True
+            ).data.decode("utf-8")
+
+            # read the token
+            rv = client.get("/admin/myfileadmin", follow_redirects=True)
+            data = rv.data.decode("utf-8")
+            assert rv.status_code == 200
+            assert 'name="csrf_token"' in data
+            assert len(self.get_csrf_token(data)) == 56
+            csrf_token = self.get_csrf_token(data)
+
+            # rename
+            rv = client.post(
+                "/admin/myfileadmin/rename/?path=dummy.txt",
+                data=dict(
+                    name="dummy_renamed.txt",
+                    path="dummy.txt",
+                ),
+            )
+            data = rv.data.decode("utf-8")
+            data = data.split("</nav>")[1]
+            assert rv.status_code == 200
+            assert "CSRF token missing" in data
+
+            rv = client.post(
+                "/admin/myfileadmin/rename/?path=dummy.txt",
+                data=dict(
+                    name="dummy_renamed.txt", path="dummy.txt", csrf_token=csrf_token
+                ),
+                follow_redirects=True,
+            )
+            assert rv.status_code == 200
+            assert "dummy_renamed.txt" in client.get(
+                "/admin/myfileadmin/", follow_redirects=True
+            ).data.decode("utf-8")
+
+            rv = client.post(
+                "/admin/myfileadmin/upload/",
+                data=dict(
+                    upload=(BytesIO(b""), "dummy_renamed.txt"), csrf_token=csrf_token
+                ),
+                follow_redirects=True,
+            )
+            data = rv.data.decode("utf-8")
+            assert rv.status_code == 200
+            assert "dummy_renamed.txt" in client.get(
+                "/admin/myfileadmin/", follow_redirects=True
+            ).data.decode("utf-8")
+            assert "already exists." in data
+
+            with open(op.join(self._test_files_root, "dummy.txt"), "w") as fp:
+                fp.write("new_string\n")
+
+            # delete
+            rv = client.post(
+                "/admin/myfileadmin/delete/",
+                data=dict(path="dummy_renamed.txt", csrf_token=csrf_token),
+                follow_redirects=True,
+            )
+            assert rv.status_code == 200
+            assert "dummy_renamed.txt" not in client.get(
+                "/admin/myfileadmin/", follow_redirects=True
+            ).data.decode("utf-8")
+
+            # mkdir
+            rv = client.post(
+                "/admin/myfileadmin/mkdir/",
+                data=dict(name="dummy_dir", csrf_token=csrf_token),
+                follow_redirects=True,
+            )
+            assert rv.status_code == 200
+            assert "dummy_dir" in client.get(
+                "/admin/myfileadmin/", follow_redirects=True
+            ).data.decode("utf-8")
+
+            # rename - dir
+            rv = client.post(
+                "/admin/myfileadmin/rename/?path=dummy_dir",
+                data=dict(
+                    name="dummy_renamed_dir", path="dummy_dir", csrf_token=csrf_token
+                ),
+                follow_redirects=True,
+            )
+            assert rv.status_code == 200
+            assert "dummy_renamed_dir" in client.get(
+                "/admin/myfileadmin/", follow_redirects=True
+            ).data.decode("utf-8")
+
+            # delete - directory
+            rv = client.post(
+                "/admin/myfileadmin/delete/",
+                data=dict(path="dummy_renamed_dir", csrf_token=csrf_token),
+                follow_redirects=True,
+            )
+            assert rv.status_code == 200
+            assert "dummy_renamed_dir" not in client.get(
+                "/admin/myfileadmin/", follow_redirects=True
+            ).data.decode("utf-8")
+
 
 class TestLocalFileAdmin(Base.FileAdminTests):
     def fileadmin_class(self):

flask_admin/tests/fileadmin/test_fileadmin_azure.py

@@ -24,11 +24,16 @@ def setup_and_teardown(self):
             azure_connection_string
         )
         self._client.create_container(self._container_name)
-        file_name = "dummy.txt"
-        file_path = os.path.join(self._test_files_root, file_name)
-        blob_client = self._client.get_blob_client(self._container_name, file_name)
-        with open(file_path, "rb") as file:
-            blob_client.upload_blob(file)
+        for file_name in [["dummy.txt"], ["d1", "dum.txt"]]:
+            file_path = os.path.join(self._test_files_root, *file_name)
+            os.makedirs(os.path.dirname(file_path), exist_ok=True)
+            with open(file_path, "w") as f:
+                f.write("")
+
+            blob_path = "/".join(file_name)
+            blob_client = self._client.get_blob_client(self._container_name, blob_path)
+            with open(file_path, "rb") as file:
+                blob_client.upload_blob(file)
 
         yield
 

flask_admin/tests/fileadmin/test_fileadmin_s3.py

@@ -18,6 +18,7 @@ def mock_s3_client():
         client = boto3.client("s3")
         client.create_bucket(Bucket=_bucket_name)
         client.upload_fileobj(BytesIO(b""), _bucket_name, "dummy.txt")
+        client.upload_fileobj(BytesIO(b""), _bucket_name, "d1/dum.txt")
         yield client