mirror of
https://github.com/django/django.git
synced 2024-12-22 17:16:24 +00:00
Fixed #35604, Refs #35326 -- Made FileSystemStorage.exists() behaviour independent from allow_overwrite.
Partially reverts 0b33a3abc2
.
Storage.exists(name) was documented to "return False if
the name is available for a new file." but return True if
the file exists. This is ambiguous in the overwrite file
case. It will now always return whether the file exists.
Thank you to Natalia Bidart and Josh Schneier for the
review.
This commit is contained in:
parent
5559011c2b
commit
8d6a20b656
@ -51,6 +51,10 @@ class Storage:
|
|||||||
validate_file_name(name, allow_relative_path=True)
|
validate_file_name(name, allow_relative_path=True)
|
||||||
return name
|
return name
|
||||||
|
|
||||||
|
def is_name_available(self, name, max_length=None):
|
||||||
|
exceeds_max_length = max_length and len(name) > max_length
|
||||||
|
return not self.exists(name) and not exceeds_max_length
|
||||||
|
|
||||||
# These methods are part of the public API, with default implementations.
|
# These methods are part of the public API, with default implementations.
|
||||||
|
|
||||||
def get_valid_name(self, name):
|
def get_valid_name(self, name):
|
||||||
@ -82,11 +86,11 @@ class Storage:
|
|||||||
validate_file_name(file_name)
|
validate_file_name(file_name)
|
||||||
file_ext = "".join(pathlib.PurePath(file_name).suffixes)
|
file_ext = "".join(pathlib.PurePath(file_name).suffixes)
|
||||||
file_root = file_name.removesuffix(file_ext)
|
file_root = file_name.removesuffix(file_ext)
|
||||||
# If the filename already exists, generate an alternative filename
|
# If the filename is not available, generate an alternative
|
||||||
# until it doesn't exist.
|
# filename until one is available.
|
||||||
# Truncate original name if required, so the new filename does not
|
# Truncate original name if required, so the new filename does not
|
||||||
# exceed the max_length.
|
# exceed the max_length.
|
||||||
while self.exists(name) or (max_length and len(name) > max_length):
|
while not self.is_name_available(name, max_length=max_length):
|
||||||
# file_ext includes the dot.
|
# file_ext includes the dot.
|
||||||
name = os.path.join(
|
name = os.path.join(
|
||||||
dir_name, self.get_alternative_name(file_root, file_ext)
|
dir_name, self.get_alternative_name(file_root, file_ext)
|
||||||
|
@ -4,7 +4,6 @@ from datetime import datetime, timezone
|
|||||||
from urllib.parse import urljoin
|
from urllib.parse import urljoin
|
||||||
|
|
||||||
from django.conf import settings
|
from django.conf import settings
|
||||||
from django.core.exceptions import SuspiciousFileOperation
|
|
||||||
from django.core.files import File, locks
|
from django.core.files import File, locks
|
||||||
from django.core.files.move import file_move_safe
|
from django.core.files.move import file_move_safe
|
||||||
from django.core.signals import setting_changed
|
from django.core.signals import setting_changed
|
||||||
@ -192,14 +191,18 @@ class FileSystemStorage(Storage, StorageSettingsMixin):
|
|||||||
# concurrently.
|
# concurrently.
|
||||||
pass
|
pass
|
||||||
|
|
||||||
def exists(self, name):
|
def is_name_available(self, name, max_length=None):
|
||||||
try:
|
|
||||||
exists = os.path.lexists(self.path(name))
|
|
||||||
except SuspiciousFileOperation:
|
|
||||||
raise
|
|
||||||
if self._allow_overwrite:
|
if self._allow_overwrite:
|
||||||
return False
|
return not (max_length and len(name) > max_length)
|
||||||
return exists
|
return super().is_name_available(name, max_length=max_length)
|
||||||
|
|
||||||
|
def get_alternative_name(self, file_root, file_ext):
|
||||||
|
if self._allow_overwrite:
|
||||||
|
return f"{file_root}{file_ext}"
|
||||||
|
return super().get_alternative_name(file_root, file_ext)
|
||||||
|
|
||||||
|
def exists(self, name):
|
||||||
|
return os.path.lexists(self.path(name))
|
||||||
|
|
||||||
def listdir(self, path):
|
def listdir(self, path):
|
||||||
path = self.path(path)
|
path = self.path(path)
|
||||||
|
@ -129,8 +129,7 @@ The ``Storage`` class
|
|||||||
.. method:: exists(name)
|
.. method:: exists(name)
|
||||||
|
|
||||||
Returns ``True`` if a file referenced by the given name already exists
|
Returns ``True`` if a file referenced by the given name already exists
|
||||||
in the storage system, or ``False`` if the name is available for a new
|
in the storage system.
|
||||||
file.
|
|
||||||
|
|
||||||
.. method:: get_accessed_time(name)
|
.. method:: get_accessed_time(name)
|
||||||
|
|
||||||
|
@ -80,11 +80,14 @@ class GenerateFilenameStorageTests(SimpleTestCase):
|
|||||||
("", ""),
|
("", ""),
|
||||||
]
|
]
|
||||||
s = FileSystemStorage()
|
s = FileSystemStorage()
|
||||||
|
s_overwrite = FileSystemStorage(allow_overwrite=True)
|
||||||
msg = "Could not derive file name from '%s'"
|
msg = "Could not derive file name from '%s'"
|
||||||
for file_name, base_name in candidates:
|
for file_name, base_name in candidates:
|
||||||
with self.subTest(file_name=file_name):
|
with self.subTest(file_name=file_name):
|
||||||
with self.assertRaisesMessage(SuspiciousFileOperation, msg % base_name):
|
with self.assertRaisesMessage(SuspiciousFileOperation, msg % base_name):
|
||||||
s.get_available_name(file_name)
|
s.get_available_name(file_name)
|
||||||
|
with self.assertRaisesMessage(SuspiciousFileOperation, msg % base_name):
|
||||||
|
s_overwrite.get_available_name(file_name)
|
||||||
with self.assertRaisesMessage(SuspiciousFileOperation, msg % base_name):
|
with self.assertRaisesMessage(SuspiciousFileOperation, msg % base_name):
|
||||||
s.generate_filename(file_name)
|
s.generate_filename(file_name)
|
||||||
|
|
||||||
@ -98,11 +101,14 @@ class GenerateFilenameStorageTests(SimpleTestCase):
|
|||||||
("\\tmp\\..\\path", "/tmp/.."),
|
("\\tmp\\..\\path", "/tmp/.."),
|
||||||
]
|
]
|
||||||
s = FileSystemStorage()
|
s = FileSystemStorage()
|
||||||
|
s_overwrite = FileSystemStorage(allow_overwrite=True)
|
||||||
for file_name, path in candidates:
|
for file_name, path in candidates:
|
||||||
msg = "Detected path traversal attempt in '%s'" % path
|
msg = "Detected path traversal attempt in '%s'" % path
|
||||||
with self.subTest(file_name=file_name):
|
with self.subTest(file_name=file_name):
|
||||||
with self.assertRaisesMessage(SuspiciousFileOperation, msg):
|
with self.assertRaisesMessage(SuspiciousFileOperation, msg):
|
||||||
s.get_available_name(file_name)
|
s.get_available_name(file_name)
|
||||||
|
with self.assertRaisesMessage(SuspiciousFileOperation, msg):
|
||||||
|
s_overwrite.get_available_name(file_name)
|
||||||
with self.assertRaisesMessage(SuspiciousFileOperation, msg):
|
with self.assertRaisesMessage(SuspiciousFileOperation, msg):
|
||||||
s.generate_filename(file_name)
|
s.generate_filename(file_name)
|
||||||
|
|
||||||
|
@ -95,18 +95,18 @@ class FileStorageTests(SimpleTestCase):
|
|||||||
"""
|
"""
|
||||||
Standard file access options are available, and work as expected.
|
Standard file access options are available, and work as expected.
|
||||||
"""
|
"""
|
||||||
self.assertFalse(os.path.exists(os.path.join(self.temp_dir, "storage_test")))
|
self.assertFalse(self.storage.exists("storage_test"))
|
||||||
f = self.storage.open("storage_test", "w")
|
f = self.storage.open("storage_test", "w")
|
||||||
f.write("storage contents")
|
f.write("storage contents")
|
||||||
f.close()
|
f.close()
|
||||||
self.assertTrue(os.path.exists(os.path.join(self.temp_dir, "storage_test")))
|
self.assertTrue(self.storage.exists("storage_test"))
|
||||||
|
|
||||||
f = self.storage.open("storage_test", "r")
|
f = self.storage.open("storage_test", "r")
|
||||||
self.assertEqual(f.read(), "storage contents")
|
self.assertEqual(f.read(), "storage contents")
|
||||||
f.close()
|
f.close()
|
||||||
|
|
||||||
self.storage.delete("storage_test")
|
self.storage.delete("storage_test")
|
||||||
self.assertFalse(os.path.exists(os.path.join(self.temp_dir, "storage_test")))
|
self.assertFalse(self.storage.exists("storage_test"))
|
||||||
|
|
||||||
def _test_file_time_getter(self, getter):
|
def _test_file_time_getter(self, getter):
|
||||||
# Check for correct behavior under both USE_TZ=True and USE_TZ=False.
|
# Check for correct behavior under both USE_TZ=True and USE_TZ=False.
|
||||||
@ -275,10 +275,10 @@ class FileStorageTests(SimpleTestCase):
|
|||||||
"""
|
"""
|
||||||
Saving a pathname should create intermediate directories as necessary.
|
Saving a pathname should create intermediate directories as necessary.
|
||||||
"""
|
"""
|
||||||
self.assertFalse(os.path.exists(os.path.join(self.temp_dir, "path/to")))
|
self.assertFalse(self.storage.exists("path/to"))
|
||||||
self.storage.save("path/to/test.file", ContentFile("file saved with path"))
|
self.storage.save("path/to/test.file", ContentFile("file saved with path"))
|
||||||
|
|
||||||
self.assertTrue(os.path.exists(os.path.join(self.temp_dir, "path/to")))
|
self.assertTrue(self.storage.exists("path/to"))
|
||||||
with self.storage.open("path/to/test.file") as f:
|
with self.storage.open("path/to/test.file") as f:
|
||||||
self.assertEqual(f.read(), b"file saved with path")
|
self.assertEqual(f.read(), b"file saved with path")
|
||||||
|
|
||||||
@ -692,12 +692,12 @@ class OverwritingStorageTests(FileStorageTests):
|
|||||||
stored_name_1 = self.storage.save(name, f_1)
|
stored_name_1 = self.storage.save(name, f_1)
|
||||||
try:
|
try:
|
||||||
self.assertEqual(stored_name_1, name)
|
self.assertEqual(stored_name_1, name)
|
||||||
self.assertTrue(os.path.exists(os.path.join(self.temp_dir, name)))
|
self.assertTrue(self.storage.exists(name))
|
||||||
with self.storage.open(name) as fp:
|
with self.storage.open(name) as fp:
|
||||||
self.assertEqual(fp.read(), content_1)
|
self.assertEqual(fp.read(), content_1)
|
||||||
stored_name_2 = self.storage.save(name, f_2)
|
stored_name_2 = self.storage.save(name, f_2)
|
||||||
self.assertEqual(stored_name_2, name)
|
self.assertEqual(stored_name_2, name)
|
||||||
self.assertTrue(os.path.exists(os.path.join(self.temp_dir, name)))
|
self.assertTrue(self.storage.exists(name))
|
||||||
with self.storage.open(name) as fp:
|
with self.storage.open(name) as fp:
|
||||||
self.assertEqual(fp.read(), content_2)
|
self.assertEqual(fp.read(), content_2)
|
||||||
finally:
|
finally:
|
||||||
@ -729,6 +729,22 @@ class OverwritingStorageTests(FileStorageTests):
|
|||||||
finally:
|
finally:
|
||||||
self.storage.delete(name)
|
self.storage.delete(name)
|
||||||
|
|
||||||
|
def test_file_name_truncation(self):
|
||||||
|
name = "test_long_file_name.txt"
|
||||||
|
file = ContentFile(b"content")
|
||||||
|
stored_name = self.storage.save(name, file, max_length=10)
|
||||||
|
self.addCleanup(self.storage.delete, stored_name)
|
||||||
|
self.assertEqual(stored_name, "test_l.txt")
|
||||||
|
self.assertEqual(len(stored_name), 10)
|
||||||
|
|
||||||
|
def test_file_name_truncation_extension_too_long(self):
|
||||||
|
name = "file_name.longext"
|
||||||
|
file = ContentFile(b"content")
|
||||||
|
with self.assertRaisesMessage(
|
||||||
|
SuspiciousFileOperation, "Storage can not find an available filename"
|
||||||
|
):
|
||||||
|
self.storage.save(name, file, max_length=5)
|
||||||
|
|
||||||
|
|
||||||
class DiscardingFalseContentStorage(FileSystemStorage):
|
class DiscardingFalseContentStorage(FileSystemStorage):
|
||||||
def _save(self, name, content):
|
def _save(self, name, content):
|
||||||
|
Loading…
Reference in New Issue
Block a user