78 lines
3.4 KiB
Python
78 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
import tarfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from govoplan_files.backend.storage.archives import _safe_member_path, inspect_archive
|
|
from govoplan_files.backend.storage.common import FileStorageError
|
|
from test_archives import _tar_bytes, _zip_bytes
|
|
|
|
|
|
class _HeadersOnly:
|
|
def __init__(self, count: int, size: int = 0):
|
|
self.count = count
|
|
self.size = size
|
|
self.seen = 0
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
return False
|
|
|
|
def getmembers(self):
|
|
raise AssertionError("TAR preview must not inflate every member before checking limits")
|
|
|
|
def __iter__(self):
|
|
for index in range(self.count):
|
|
member = tarfile.TarInfo(f"file-{index}.txt")
|
|
member.size = self.size
|
|
self.seen += 1
|
|
yield member
|
|
raise AssertionError("Inspection advanced beyond the rejecting header")
|
|
|
|
|
|
class ArchiveInspectionBoundTests(unittest.TestCase):
|
|
def test_tar_entry_limit_stops_at_first_excess_header(self):
|
|
headers = _HeadersOnly(4)
|
|
with patch("govoplan_files.backend.storage.archives._open_tar", return_value=headers):
|
|
with self.assertRaisesRegex(FileStorageError, "too many entries"):
|
|
inspect_archive(b"fixture", filename="fixture.tar.gz", max_entries=2)
|
|
self.assertEqual(3, headers.seen)
|
|
|
|
def test_tar_expanded_size_rejected_before_payload_decompression(self):
|
|
headers = _HeadersOnly(1, size=100)
|
|
with patch("govoplan_files.backend.storage.archives._open_tar", return_value=headers):
|
|
with self.assertRaisesRegex(FileStorageError, "too large after extraction"):
|
|
inspect_archive(b"fixture", filename="fixture.tar.gz", max_expanded_bytes=10)
|
|
self.assertEqual(1, headers.seen)
|
|
|
|
def test_tar_ratio_rejected_before_payload_decompression(self):
|
|
headers = _HeadersOnly(1, size=100)
|
|
with patch("govoplan_files.backend.storage.archives._open_tar", return_value=headers):
|
|
with self.assertRaisesRegex(FileStorageError, "expansion ratio"):
|
|
inspect_archive(b"fixture", filename="fixture.tar.gz", max_expansion_ratio=2)
|
|
|
|
def test_derived_directories_are_included_in_entry_limit(self):
|
|
for filename, payload in (("fixture.zip", _zip_bytes({"a/b/c/file.txt": b"x"})), ("fixture.tar.gz", _tar_bytes({"a/b/c/file.txt": b"x"}))):
|
|
with self.subTest(filename=filename), self.assertRaisesRegex(FileStorageError, "including parent directories"):
|
|
inspect_archive(payload, filename=filename, max_entries=3)
|
|
|
|
def test_extreme_path_depth_and_byte_lengths_fail_before_deriving_directories(self):
|
|
for path, error in (("a/" * 128 + "file.txt", "components"), ("ü" * 2049, "UTF-8 bytes")):
|
|
with self.subTest(path_length=len(path)), self.assertRaisesRegex(FileStorageError, error):
|
|
inspect_archive(_zip_bytes({path: b"x"}), filename="fixture.zip")
|
|
|
|
def test_directory_at_exact_depth_limit_accepts_trailing_separator(self):
|
|
path = "a/" * 128
|
|
self.assertEqual(path.rstrip("/"), _safe_member_path(path))
|
|
|
|
def test_invalid_unicode_member_name_has_controlled_validation_error(self):
|
|
with self.assertRaisesRegex(FileStorageError, "not valid Unicode"):
|
|
_safe_member_path("invalid-\udcff.txt")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|