CVE-2022-24840: Improper Limitation of a Pathname to a Restricted Directory ('Path Traversal') (CWE-22)
value: django-s3file is a lightweight file upload input for Django and Amazon S3 . In versions prior to 5.5.1 it was possible to traverse the entire AWS S3 bucket and in most cases to access or delete files. If the `AWS_LOCATION` setting was set, traversal was limited to that location only. The issue was discovered by the maintainer. There were no reports of the vulnerability being known to or exploited by a third party, prior to the release of the patch. The vulnerability has been fixed in version 5.5.1 and above. There is no feasible workaround. We must urge all users to immediately updated to a patched version.
CWE description: The product uses external input to construct a pathname that is intended to identify a file or directory that is located underneath a restricted parent directory, but the product does not properly neutralize special elements within the pathname that can cause the pathname to resolve to a location that is outside of the restricted directory.
CVSS v3 base score: 9.8
Published: 2022-06-09T04:15Z
Repository: https://github.com/codingjoe/django-s3file
Language: Python
Files changed: s3file/forms.py, s3file/middleware.py, s3file/static/s3file/js/s3file.js, s3file/views.py, tests/conftest.py, tests/test_forms.py, tests/test_middleware.py
Commit date: 2022-06-06 12:38:02 +0200
Commit message: Fix CVE-XXXX-XXXX -- Fix Path Traversal security vulnerability

Diff: diff --git a/s3file/forms.py b/s3file/forms.py
index 308eba3..4d5915e 100644
--- a/s3file/forms.py
+++ b/s3file/forms.py
@@ -2,10 +2,11 @@ import base64
 import logging
 import pathlib
 import uuid
 
 from django.conf import settings
+from django.core import signing
 from django.utils.functional import cached_property
 from storages.utils import safe_join
 
 from s3file.storages import storage
 
@@ -14,14 +15,18 @@ logger = logging.getLogger("s3file")
 
 class S3FileInputMixin:
     """FileInput that uses JavaScript to directly upload to Amazon S3."""
 
     needs_multipart_form = False
-    upload_path = str(
-        getattr(settings, "S3FILE_UPLOAD_PATH", pathlib.PurePosixPath("tmp", "s3file"))
+    upload_path = safe_join(
+        str(storage.aws_location),
+        str(
+            getattr(
+                settings, "S3FILE_UPLOAD_PATH", pathlib.PurePosixPath("tmp", "s3file")
+            )
+        ),
     )
-    upload_path = safe_join(str(storage.location), upload_path)
     expires = settings.SESSION_COOKIE_AGE
 
     @property
     def bucket_name(self):
         return storage.bucket.name
@@ -43,10 +48,15 @@ class S3FileInputMixin:
 
         defaults = {
             "data-fields-%s" % key: value for key, value in response["fields"].items()
         }
         defaults["data-url"] = response["url"]
+        signer = signing.Signer(
+            salt=f"{S3FileInputMixin.__module__}.{S3FileInputMixin.__name__}"
+        )
+        print(self.upload_folder)
+        defaults["data-s3f-signature"] = signer.signature(self.upload_folder)
         defaults.update(attrs)
 
         try:
             defaults["class"] += " s3file"
         except KeyError:
diff --git a/s3file/middleware.py b/s3file/middleware.py
index 6c253ef..027c613 100644
--- a/s3file/middleware.py
+++ b/s3file/middleware.py
@@ -1,11 +1,15 @@
 import logging
 import pathlib
 
-from s3file.storages import local_dev, storage
+from django.core import signing
+from django.core.exceptions import PermissionDenied, SuspiciousFileOperation
+from django.utils.crypto import constant_time_compare
 
 from . import views
+from .forms import S3FileInputMixin
+from .storages import local_dev, storage
 
 logger = logging.getLogger("s3file")
 
 
 class S3FileMiddleware:
@@ -13,28 +17,53 @@ class S3FileMiddleware:
         self.get_response = get_response
 
     def __call__(self, request):
         file_fields = request.POST.getlist("s3file")
         for field_name in file_fields:
+
             paths = request.POST.getlist(field_name)
-            request.FILES.setlist(field_name, list(self.get_files_from_storage(paths)))
+            if paths:
+                try:
+                    signature = request.POST[f"{field_name}-s3f-signature"]
+                except KeyError:
+                    raise PermissionDenied("No signature provided.")
+                try:
+                    request.FILES.setlist(
+                        field_name, list(self.get_files_from_storage(paths, signature))
+                    )
+                except SuspiciousFileOperation as e:
+                    raise PermissionDenied("Illegal file name!") from e
 
         if local_dev and request.path == "/__s3_mock__/":
             return views.S3MockView.as_view()(request)
 
         return self.get_response(request)
 
     @staticmethod
-    def get_files_from_storage(paths):
+    def get_files_from_storage(paths, signature):
         """Return S3 file where the name does not include the path."""
+        try:
+            location = storage.aws_location
+        except AttributeError:
+            location = storage.location
+        signer = signing.Signer(
+            salt=f"{S3FileInputMixin.__module__}.{S3FileInputMixin.__name__}"
+        )
         for path in paths:
             path = pathlib.PurePosixPath(path)
+            print(path)
+            print(signer.signature(path.parent), signature)
+            if not constant_time_compare(signer.signature(path.parent), signature):
+                raise PermissionDenied("Illegal signature!")
             try:
-                location = storage.aws_location
-            except AttributeError:
-                location = storage.location
+                relative_path = str(path.relative_to(location))
+            except ValueError as e:
+                raise SuspiciousFileOperation(
+                    f"Path is not inside the designated upload location: {path}"
+                ) from e
+
             try:
-                f = storage.open(str(path.relative_to(location)))
+                f = storage.open(relative_path)
                 f.name = path.name
                 yield f
             except (OSError, ValueError):
                 logger.exception("File not found: %s", path)
diff --git a/s3file/static/s3file/js/s3file.js b/s3file/static/s3file/js/s3file.js
index 8e0bf47..4cd6aa5 100644
--- a/s3file/static/s3file/js/s3file.js
+++ b/s3file/static/s3file/js/s3file.js
@@ -92,10 +92,16 @@
         var hiddenFileInput = document.createElement('input')
         hiddenFileInput.type = 'hidden'
         hiddenFileInput.name = name
         hiddenFileInput.value = parseURL(result)
         form.appendChild(hiddenFileInput)
+        var hiddenSignatureInput = document.createElement('input')
+        hiddenSignatureInput.type = 'hidden'
+        hiddenSignatureInput.name = name + '-s3f-signature'
+        console.log(fileInput.dataset.s3fSignature)
+        hiddenSignatureInput.value = fileInput.dataset.s3fSignature
+        form.appendChild(hiddenSignatureInput)
       })
       fileInput.name = ''
       window.uploading -= 1
     }, function (err) {
       console.log(err)
diff --git a/s3file/views.py b/s3file/views.py
index f25d24b..54c054d 100644
--- a/s3file/views.py
+++ b/s3file/views.py
@@ -1,9 +1,10 @@
 import base64
 import hashlib
 import hmac
 import logging
+from pathlib import Path
 
 from django import http
 from django.conf import settings
 from django.core.files.storage import default_storage
 from django.views import generic
diff --git a/tests/conftest.py b/tests/conftest.py
index 4bfefd2..21fe640 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,14 +1,16 @@
-import os
 import tempfile
+from pathlib import Path
 
 import pytest
 from django.core.files.base import ContentFile
 from django.utils.encoding import force_str
 from selenium import webdriver
 from selenium.common.exceptions import WebDriverException
 
+from s3file.storages import storage
+
 
 @pytest.fixture(scope="session")
 def driver():
     chrome_options = webdriver.ChromeOptions()
     chrome_options.headless = True
@@ -20,34 +22,53 @@ def driver():
         yield b
         b.quit()
 
 
 @pytest.fixture
-def upload_file(request):
-    path = tempfile.mkdtemp()
-    file_name = os.path.join(path, "%s.txt" % request.node.name)
-    with open(file_name, "w") as f:
+def freeze_upload_folder(monkeypatch):
+    """Freeze datetime and UUID."""
+    upload_folder = Path(storage.aws_location) / "tmp" / "s3file"
+    monkeypatch.setattr(
+        "s3file.forms.S3FileInputMixin.upload_folder",
+        str(upload_folder),
+    )
+    return upload_folder
+
+
+@pytest.fixture
+def upload_file(request, freeze_upload_folder):
+    path = Path(tempfile.mkdtemp()) / freeze_upload_folder / f"{request.node.name}.txt"
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("w") as f:
         f.write(request.node.name)
-    return file_name
+    return str(path.absolute())
 
 
 @pytest.fixture
-def another_upload_file(request):
-    path = tempfile.mkdtemp()
-    file_name = os.path.join(path, "another_%s.txt" % request.node.name)
-    with open(file_name, "w") as f:
+def another_upload_file(request, freeze_upload_folder):
+    path = (
+        Path(tempfile.mkdtemp())
+        / freeze_upload_folder
+        / f"another_{request.node.name}.txt"
+    )
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("w") as f:
         f.write(request.node.name)
-    return file_name
+    return str(path.absolute())
 
 
 @pytest.fixture
-def yet_another_upload_file(request):
-    path = tempfile.mkdtemp()
-    file_name = os.path.join(path, "yet_another_%s.txt" % request.node.name)
-    with open(file_name, "w") as f:
+def yet_another_upload_file(request, freeze_upload_folder):
+    path = (
+        Path(tempfile.mkdtemp())
+        / freeze_upload_folder
+        / f"yet_another_{request.node.name}.txt"
+    )
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("w") as f:
         f.write(request.node.name)
-    return file_name
+    return str(path.absolute())
 
 
 @pytest.fixture
 def filemodel(request, db):
     from tests.testapp.models import FileModel
diff --git a/tests/test_forms.py b/tests/test_forms.py
index ec1a9c4..776e060 100644
--- a/tests/test_forms.py
+++ b/tests/test_forms.py
@@ -29,27 +29,19 @@ def wait_for_page_load(driver, timeout=30):
 class TestS3FileInput:
     @property
     def url(self):
         return reverse("upload")
 
-    @pytest.fixture
-    def freeze(self, monkeypatch):
-        """Freeze datetime and UUID."""
-        monkeypatch.setattr(
-            "s3file.forms.S3FileInputMixin.upload_folder",
-            os.path.join(storage.aws_location, "tmp"),
-        )
-
-    def test_value_from_datadict(self, client, upload_file):
-        print(storage.location)
+    def test_value_from_datadict(self, freeze_upload_folder, client, upload_file):
         with open(upload_file) as f:
-            uploaded_file = storage.save("test.jpg", f)
+            uploaded_file = storage.save(freeze_upload_folder / "test.jpg", f)
         response = client.post(
             reverse("upload"),
             {
-                "file": json.dumps([uploaded_file]),
-                "s3file": '["file"]',
+                "file": f"custom/location/{uploaded_file}",
+                "file-s3f-signature": "m94qBxBsnMIuIICiY133kX18KkllSPMVbhGAdAwNn1A",
+                "s3file": "file",
             },
         )
 
         assert response.status_code == 201
 
@@ -80,35 +72,40 @@ class TestS3FileInput:
     def test_clear(self, filemodel):
         form = UploadForm(data={"file-clear": "1"}, instance=filemodel)
         assert form.is_valid()
         assert not form.cleaned_data["file"]
 
-    def test_build_attr(self):
+    def test_build_attr(self, freeze_upload_folder):
         assert set(ClearableFileInput().build_attrs({}).keys()) == {
             "class",
             "data-url",
             "data-fields-x-amz-algorithm",
             "data-fields-x-amz-date",
             "data-fields-x-amz-signature",
             "data-fields-x-amz-credential",
             "data-fields-policy",
             "data-fields-key",
+            "data-s3f-signature",
         }
+        assert (
+            ClearableFileInput().build_attrs({})["data-s3f-signature"]
+            == "tFV9nGZlq9WX1I5Sotit18z1f4C_3lPnj33_zo4LZRc"
+        )
         assert ClearableFileInput().build_attrs({})["class"] == "s3file"
         assert (
             ClearableFileInput().build_attrs({"class": "my-class"})["class"]
             == "my-class s3file"
         )
 
-    def test_get_conditions(self, freeze):
+    def test_get_conditions(self, freeze_upload_folder):
         conditions = ClearableFileInput().get_conditions(None)
         assert all(
             condition in conditions
             for condition in [
                 {"bucket": "test-bucket"},
                 {"success_action_status": "201"},
-                ["starts-with", "$key", "custom/location/tmp"],
+                ["starts-with", "$key", "custom/location/tmp/s3file"],
                 ["starts-with", "$Content-Type", ""],
             ]
         ), conditions
 
     def test_accept(self):
@@ -143,24 +140,28 @@ class TestS3FileInput:
 
         with pytest.raises(NoSuchElementException):
             error = driver.find_element(By.XPATH, "//body[@JSError]")
             pytest.fail(error.get_attribute("JSError"))
 
-    def test_file_insert(self, request, driver, live_server, upload_file, freeze):
+    def test_file_insert(
+        self, request, driver, live_server, upload_file, freeze_upload_folder
+    ):
         driver.get(live_server + self.url)
         file_input = driver.find_element(By.XPATH, "//input[@name='file']")
         file_input.send_keys(upload_file)
         assert file_input.get_attribute("name") == "file"
         with wait_for_page_load(driver, timeout=10):
             file_input.submit()
-        assert storage.exists("tmp/%s.txt" % request.node.name)
+        assert storage.exists("tmp/s3file/%s.txt" % request.node.name)
 
         with pytest.raises(NoSuchElementException):
             error = driver.find_element(By.XPATH, "//body[@JSError]")
             pytest.fail(error.get_attribute("JSError"))
 
-    def test_file_insert_submit_value(self, driver, live_server, upload_file, freeze):
+    def test_file_insert_submit_value(
+        self, driver, live_server, upload_file, freeze_upload_folder
+    ):
         driver.get(live_server + self.url)
         file_input = driver.find_element(By.XPATH, "//input[@name='file']")
         file_input.send_keys(upload_file)
         assert file_input.get_attribute("name") == "file"
         save_button = driver.find_element(By.XPATH, "//input[@name='save']")
@@ -176,11 +177,11 @@ class TestS3FileInput:
         with wait_for_page_load(driver, timeout=10):
             save_button.click()
         assert "save_continue" in driver.page_source
         assert "continue_value" in driver.page_source
 
-    def test_progress(self, driver, live_server, upload_file, freeze):
+    def test_progress(self, driver, live_server, upload_file, freeze_upload_folder):
         driver.get(live_server + self.url)
         file_input = driver.find_element(By.XPATH, "//input[@name='file']")
         file_input.send_keys(upload_file)
         assert file_input.get_attribute("name") == "file"
         save_button = driver.find_element(By.XPATH, "//input[@name='save']")
@@ -200,20 +201,27 @@ class TestS3FileInput:
 
     def test_multi_file(
         self,
         driver,
         live_server,
-        freeze,
+        freeze_upload_folder,
         upload_file,
         another_upload_file,
         yet_another_upload_file,
     ):
         driver.get(live_server + self.url)
         file_input = driver.find_element(By.XPATH, "//input[@name='file']")
-        file_input.send_keys(" \n ".join([upload_file, another_upload_file]))
+        file_input.send_keys(
+            " \n ".join(
+                [
+                    str(freeze_upload_folder / upload_file),
+                    str(freeze_upload_folder / another_upload_file),
+                ]
+            )
+        )
         file_input = driver.find_element(By.XPATH, "//input[@name='other_file']")
-        file_input.send_keys(yet_another_upload_file)
+        file_input.send_keys(str(freeze_upload_folder / yet_another_upload_file))
         save_button = driver.find_element(By.XPATH, "//input[@name='save']")
         with wait_for_page_load(driver, timeout=10):
             save_button.click()
         response = json.loads(driver.find_elements(By.CSS_SELECTOR, "pre")[0].text)
         assert response["FILES"] == {
diff --git a/tests/test_middleware.py b/tests/test_middleware.py
index 9dc552d..dc2acc6 100644
--- a/tests/test_middleware.py
+++ b/tests/test_middleware.py
@@ -1,27 +1,30 @@
 import os
 
+import pytest
+from django.core.exceptions import PermissionDenied, SuspiciousFileOperation
 from django.core.files.base import ContentFile
 from django.core.files.uploadedfile import SimpleUploadedFile
 
 from s3file.middleware import S3FileMiddleware
 from s3file.storages import storage
 
 
 class TestS3FileMiddleware:
-    def test_get_files_from_storage(self):
+    def test_get_files_from_storage(self, freeze_upload_folder):
         content = b"test_get_files_from_storage"
         name = storage.save(
             "tmp/s3file/test_get_files_from_storage", ContentFile(content)
         )
         files = S3FileMiddleware.get_files_from_storage(
-            [os.path.join(storage.aws_location, name)]
+            [os.path.join(storage.aws_location, name)],
+            "tFV9nGZlq9WX1I5Sotit18z1f4C_3lPnj33_zo4LZRc",
         )
         file = next(files)
         assert file.read() == content
 
-    def test_process_request(self, rf):
+    def test_process_request(self, freeze_upload_folder, rf):
         uploaded_file = SimpleUploadedFile("uploaded_file.txt", b"uploaded")
         request = rf.post("/", data={"file": uploaded_file})
         S3FileMiddleware(lambda x: None)(request)
         assert request.FILES.getlist("file")
         assert request.FILES.get("file").read() == b"uploaded"
@@ -30,50 +33,101 @@ class TestS3FileMiddleware:
         request = rf.post(
             "/",
             data={
                 "file": "custom/location/tmp/s3file/s3_file.txt",
                 "s3file": "file",
+                "file-s3f-signature": "tFV9nGZlq9WX1I5Sotit18z1f4C_3lPnj33_zo4LZRc",
             },
         )
         S3FileMiddleware(lambda x: None)(request)
         assert request.FILES.getlist("file")
         assert request.FILES.get("file").read() == b"s3file"
 
-    def test_process_request__multiple_files(self, rf):
+    def test_process_request__location_escape(self, freeze_upload_folder, rf):
+        storage.save("secrets/passwords.txt", ContentFile(b"keep this secret"))
+        request = rf.post(
+            "/",
+            data={
+                "file": "custom/location/secrets/passwords.txt",
+                "s3file": "file",
+                "file-s3f-signature": "tFV9nGZlq9WX1I5Sotit18z1f4C_3lPnj33_zo4LZRc",
+            },
+        )
+        with pytest.raises(PermissionDenied) as e:
+            S3FileMiddleware(lambda x: None)(request)
+        assert "Illegal signature!" in str(e.value)
+
+    def test_process_request__multiple_files(self, freeze_upload_folder, rf):
         storage.save("tmp/s3file/s3_file.txt", ContentFile(b"s3file"))
         storage.save("tmp/s3file/s3_other_file.txt", ContentFile(b"other s3file"))
         request = rf.post(
             "/",
             data={
                 "file": [
                     "custom/location/tmp/s3file/s3_file.txt",
                     "custom/location/tmp/s3file/s3_other_file.txt",
                 ],
+                "file-s3f-signature": "tFV9nGZlq9WX1I5Sotit18z1f4C_3lPnj33_zo4LZRc",
+                "other_file-s3f-signature": "tFV9nGZlq9WX1I5Sotit18z1f4C_3lPnj33_zo4LZRc",
                 "s3file": ["file", "other_file"],
             },
         )
         S3FileMiddleware(lambda x: None)(request)
         files = request.FILES.getlist("file")
         assert files[0].read() == b"s3file"
         assert files[1].read() == b"other s3file"
 
-    def test_process_request__no_location(self, rf, settings):
+    def test_process_request__no_location(self, freeze_upload_folder, rf, settings):
         settings.AWS_LOCATION = ""
         uploaded_file = SimpleUploadedFile("uploaded_file.txt", b"uploaded")
         request = rf.post("/", data={"file": uploaded_file})
         S3FileMiddleware(lambda x: None)(request)
         assert request.FILES.getlist("file")
         assert request.FILES.get("file").read() == b"uploaded"
 
         storage.save("tmp/s3file/s3_file.txt", ContentFile(b"s3file"))
         request = rf.post(
-            "/", data={"file": "tmp/s3file/s3_file.txt", "s3file": "file"}
+            "/",
+            data={
+                "file": f"tmp/s3file/s3_file.txt",
+                "s3file": "file",
+                "file-s3f-signature": "scjzm3N8njBQIVSGEhOchtM0TkGyb2U6OXGLVlRUZhY",
+            },
         )
         S3FileMiddleware(lambda x: None)(request)
         assert request.FILES.getlist("file")
         assert request.FILES.get("file").read() == b"s3file"
 
-    def test_process_request__no_file(self, rf, caplog):
-        request = rf.post("/", data={"file": "does_not_exist.txt", "s3file": "file"})
+    def test_process_request__no_file(self, freeze_upload_folder, rf, caplog):
+        request = rf.post(
+            "/",
+            data={
+                "file": "custom/location/tmp/s3file/does_not_exist.txt",
+                "s3file": "file",
+                "file-s3f-signature": "tFV9nGZlq9WX1I5Sotit18z1f4C_3lPnj33_zo4LZRc",
+            },
+        )
         S3FileMiddleware(lambda x: None)(request)
         assert not request.FILES.getlist("file")
-        assert "File not found: does_not_exist.txt" in caplog.text
+        assert (
+            "File not found: custom/location/tmp/s3file/does_not_exist.txt"
+            in caplog.text
+        )
+
+    def test_process_request__no_signature(self, rf, caplog):
+        request = rf.post(
+            "/", data={"file": "tmp/s3file/does_not_exist.txt", "s3file": "file"}
+        )
+        with pytest.raises(PermissionDenied) as e:
+            S3FileMiddleware(lambda x: None)(request)
+
+    def test_process_request__wrong_signature(self, rf, caplog):
+        request = rf.post(
+            "/",
+            data={
+                "file": "tmp/s3file/does_not_exist.txt",
+                "s3file": "file",
+                "file-s3f-signature": "fake",
+            },
+        )
+        with pytest.raises(PermissionDenied) as e:
+            S3FileMiddleware(lambda x: None)(request)

Vulnerable code:     upload_path = str(
        getattr(settings, "S3FILE_UPLOAD_PATH", pathlib.PurePosixPath("tmp", "s3file"))
    upload_path = safe_join(str(storage.location), upload_path)
from s3file.storages import local_dev, storage
            request.FILES.setlist(field_name, list(self.get_files_from_storage(paths)))
    def get_files_from_storage(paths):
                location = storage.aws_location
            except AttributeError:
                location = storage.location
                f = storage.open(str(path.relative_to(location)))
import os
def upload_file(request):
    path = tempfile.mkdtemp()
    file_name = os.path.join(path, "%s.txt" % request.node.name)
    with open(file_name, "w") as f:
    return file_name
def another_upload_file(request):
    path = tempfile.mkdtemp()
    file_name = os.path.join(path, "another_%s.txt" % request.node.name)
    with open(file_name, "w") as f:
    return file_name
def yet_another_upload_file(request):
    path = tempfile.mkdtemp()
    file_name = os.path.join(path, "yet_another_%s.txt" % request.node.name)
    with open(file_name, "w") as f:
    return file_name
    @pytest.fixture
    def freeze(self, monkeypatch):
        """Freeze datetime and UUID."""
        monkeypatch.setattr(
            "s3file.forms.S3FileInputMixin.upload_folder",
            os.path.join(storage.aws_location, "tmp"),
        )

    def test_value_from_datadict(self, client, upload_file):
        print(storage.location)
            uploaded_file = storage.save("test.jpg", f)
                "file": json.dumps([uploaded_file]),
                "s3file": '["file"]',
    def test_build_attr(self):
    def test_get_conditions(self, freeze):
                ["starts-with", "$key", "custom/location/tmp"],
    def test_file_insert(self, request, driver, live_server, upload_file, freeze):
        assert storage.exists("tmp/%s.txt" % request.node.name)
    def test_file_insert_submit_value(self, driver, live_server, upload_file, freeze):
    def test_progress(self, driver, live_server, upload_file, freeze):
        freeze,
        file_input.send_keys(" \n ".join([upload_file, another_upload_file]))
        file_input.send_keys(yet_another_upload_file)
    def test_get_files_from_storage(self):
            [os.path.join(storage.aws_location, name)]
    def test_process_request(self, rf):
    def test_process_request__multiple_files(self, rf):
    def test_process_request__no_location(self, rf, settings):
            "/", data={"file": "tmp/s3file/s3_file.txt", "s3file": "file"}
    def test_process_request__no_file(self, rf, caplog):
        request = rf.post("/", data={"file": "does_not_exist.txt", "s3file": "file"})
        assert "File not found: does_not_exist.txt" in caplog.text
Fixed code: from django.core import signing
    upload_path = safe_join(
        str(storage.aws_location),
        str(
            getattr(
                settings, "S3FILE_UPLOAD_PATH", pathlib.PurePosixPath("tmp", "s3file")
            )
        ),
        signer = signing.Signer(
            salt=f"{S3FileInputMixin.__module__}.{S3FileInputMixin.__name__}"
        )
        print(self.upload_folder)
        defaults["data-s3f-signature"] = signer.signature(self.upload_folder)
from django.core import signing
from django.core.exceptions import PermissionDenied, SuspiciousFileOperation
from django.utils.crypto import constant_time_compare
from .forms import S3FileInputMixin
from .storages import local_dev, storage

            if paths:
                try:
                    signature = request.POST[f"{field_name}-s3f-signature"]
                except KeyError:
                    raise PermissionDenied("No signature provided.")
                try:
                    request.FILES.setlist(
                        field_name, list(self.get_files_from_storage(paths, signature))
                    )
                except SuspiciousFileOperation as e:
                    raise PermissionDenied("Illegal file name!") from e
    def get_files_from_storage(paths, signature):
        try:
            location = storage.aws_location
        except AttributeError:
            location = storage.location
        signer = signing.Signer(
            salt=f"{S3FileInputMixin.__module__}.{S3FileInputMixin.__name__}"
        )
            print(path)
            print(signer.signature(path.parent), signature)
            if not constant_time_compare(signer.signature(path.parent), signature):
                raise PermissionDenied("Illegal signature!")
                relative_path = str(path.relative_to(location))
            except ValueError as e:
                raise SuspiciousFileOperation(
                    f"Path is not inside the designated upload location: {path}"
                ) from e

                f = storage.open(relative_path)
        var hiddenSignatureInput = document.createElement('input')
        hiddenSignatureInput.type = 'hidden'
        hiddenSignatureInput.name = name + '-s3f-signature'
        console.log(fileInput.dataset.s3fSignature)
        hiddenSignatureInput.value = fileInput.dataset.s3fSignature
        form.appendChild(hiddenSignatureInput)
from pathlib import Path
from pathlib import Path
from s3file.storages import storage

def freeze_upload_folder(monkeypatch):
    """Freeze datetime and UUID."""
    upload_folder = Path(storage.aws_location) / "tmp" / "s3file"
    monkeypatch.setattr(
        "s3file.forms.S3FileInputMixin.upload_folder",
        str(upload_folder),
    )
    return upload_folder


@pytest.fixture
def upload_file(request, freeze_upload_folder):
    path = Path(tempfile.mkdtemp()) / freeze_upload_folder / f"{request.node.name}.txt"
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w") as f:
    return str(path.absolute())
def another_upload_file(request, freeze_upload_folder):
    path = (
        Path(tempfile.mkdtemp())
        / freeze_upload_folder
        / f"another_{request.node.name}.txt"
    )
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w") as f:
    return str(path.absolute())
def yet_another_upload_file(request, freeze_upload_folder):
    path = (
        Path(tempfile.mkdtemp())
        / freeze_upload_folder
        / f"yet_another_{request.node.name}.txt"
    )
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w") as f:
    return str(path.absolute())
    def test_value_from_datadict(self, freeze_upload_folder, client, upload_file):
            uploaded_file = storage.save(freeze_upload_folder / "test.jpg", f)
                "file": f"custom/location/{uploaded_file}",
                "file-s3f-signature": "m94qBxBsnMIuIICiY133kX18KkllSPMVbhGAdAwNn1A",
                "s3file": "file",
    def test_build_attr(self, freeze_upload_folder):
            "data-s3f-signature",
        assert (
            ClearableFileInput().build_attrs({})["data-s3f-signature"]
            == "tFV9nGZlq9WX1I5Sotit18z1f4C_3lPnj33_zo4LZRc"
        )
    def test_get_conditions(self, freeze_upload_folder):
                ["starts-with", "$key", "custom/location/tmp/s3file"],
    def test_file_insert(
        self, request, driver, live_server, upload_file, freeze_upload_folder
    ):
        assert storage.exists("tmp/s3file/%s.txt" % request.node.name)
    def test_file_insert_submit_value(
        self, driver, live_server, upload_file, freeze_upload_folder
    ):
    def test_progress(self, driver, live_server, upload_file, freeze_upload_folder):
        freeze_upload_folder,
        file_input.send_keys(
            " \n ".join(
                [
                    str(freeze_upload_folder / upload_file),
                    str(freeze_upload_folder / another_upload_file),
                ]
            )
        )
        file_input.send_keys(str(freeze_upload_folder / yet_another_upload_file))
import pytest
from django.core.exceptions import PermissionDenied, SuspiciousFileOperation
    def test_get_files_from_storage(self, freeze_upload_folder):
            [os.path.join(storage.aws_location, name)],
            "tFV9nGZlq9WX1I5Sotit18z1f4C_3lPnj33_zo4LZRc",
    def test_process_request(self, freeze_upload_folder, rf):
                "file-s3f-signature": "tFV9nGZlq9WX1I5Sotit18z1f4C_3lPnj33_zo4LZRc",
    def test_process_request__location_escape(self, freeze_upload_folder, rf):
        storage.save("secrets/passwords.txt", ContentFile(b"keep this secret"))
        request = rf.post(
            "/",
            data={
                "file": "custom/location/secrets/passwords.txt",
                "s3file": "file",
                "file-s3f-signature": "tFV9nGZlq9WX1I5Sotit18z1f4C_3lPnj33_zo4LZRc",
            },
        )
        with pytest.raises(PermissionDenied) as e:
            S3FileMiddleware(lambda x: None)(request)
        assert "Illegal signature!" in str(e.value)

    def test_process_request__multiple_files(self, freeze_upload_folder, rf):
                "file-s3f-signature": "tFV9nGZlq9WX1I5Sotit18z1f4C_3lPnj33_zo4LZRc",
                "other_file-s3f-signature": "tFV9nGZlq9WX1I5Sotit18z1f4C_3lPnj33_zo4LZRc",
    def test_process_request__no_location(self, freeze_upload_folder, rf, settings):
            "/",
            data={
                "file": f"tmp/s3file/s3_file.txt",
                "s3file": "file",
                "file-s3f-signature": "scjzm3N8njBQIVSGEhOchtM0TkGyb2U6OXGLVlRUZhY",
            },
    def test_process_request__no_file(self, freeze_upload_folder, rf, caplog):
        request = rf.post(
            "/",
            data={
                "file": "custom/location/tmp/s3file/does_not_exist.txt",
                "s3file": "file",
                "file-s3f-signature": "tFV9nGZlq9WX1I5Sotit18z1f4C_3lPnj33_zo4LZRc",
            },
        )
        assert (
            "File not found: custom/location/tmp/s3file/does_not_exist.txt"
            in caplog.text
        )

    def test_process_request__no_signature(self, rf, caplog):
        request = rf.post(
            "/", data={"file": "tmp/s3file/does_not_exist.txt", "s3file": "file"}
        )
        with pytest.raises(PermissionDenied) as e:
            S3FileMiddleware(lambda x: None)(request)

    def test_process_request__wrong_signature(self, rf, caplog):
        request = rf.post(
            "/",
            data={
                "file": "tmp/s3file/does_not_exist.txt",
                "s3file": "file",
                "file-s3f-signature": "fake",
            },
        )
        with pytest.raises(PermissionDenied) as e:
            S3FileMiddleware(lambda x: None)(request)