446 lines
18 KiB
Python
446 lines
18 KiB
Python
from __future__ import annotations
|
|
|
|
from contextlib import closing
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
import random
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import pyzipper
|
|
from sqlalchemy import event
|
|
|
|
from govoplan_core.core.change_sequence import ChangeSequenceEntry
|
|
from govoplan_files.backend.change_tracking import register_files_change_tracking
|
|
from govoplan_files.backend.storage.archives import (
|
|
_read_selected_zip_members,
|
|
_safe_member_path,
|
|
extract_archive_upload,
|
|
)
|
|
from govoplan_files.backend.storage.common import FileStorageError
|
|
from govoplan_files.backend.storage.files import (
|
|
archive_storage_backend_scope,
|
|
_archive_write_backend,
|
|
)
|
|
from govoplan_files.backend.storage.native_zip import (
|
|
native_zip_library,
|
|
read_native_zip_members,
|
|
)
|
|
from test_archives import _zip_bytes, _tar_bytes, _encrypted_zip_bytes
|
|
import test_storage_recovery as recovery_fixture
|
|
|
|
|
|
def legacy_archive(entries: dict[str, bytes]) -> bytes:
|
|
# Fixture-only known password, never an application secret.
|
|
with tempfile.TemporaryDirectory(prefix="files-legacy-fixture-") as temporary:
|
|
root = Path(temporary)
|
|
for name, content in entries.items():
|
|
path = root / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(content)
|
|
subprocess.run(
|
|
[shutil.which("zip"), "-q", "-P", "fixture-only", "fixture.zip", *entries],
|
|
cwd=root,
|
|
check=True,
|
|
)
|
|
return (root / "fixture.zip").read_bytes()
|
|
|
|
|
|
@unittest.skipUnless(
|
|
shutil.which("zip") and native_zip_library() is not None,
|
|
"optional libarchive and synthetic ZIP writer required",
|
|
)
|
|
class NativeArchivePerformanceTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.contents = {
|
|
"first.dat": random.Random(123).randbytes(8192),
|
|
"folder/second.txt": b"second",
|
|
}
|
|
self.archive = legacy_archive(self.contents)
|
|
|
|
def read(self, **kwargs):
|
|
return list(
|
|
_read_selected_zip_members(
|
|
self.archive,
|
|
selected_files={"first.dat"},
|
|
password="fixture-only",
|
|
max_file_bytes=100_000,
|
|
max_total_bytes=100_000,
|
|
**kwargs,
|
|
)
|
|
)
|
|
|
|
def test_native_selection_and_missing_library_python_fallback_match(self):
|
|
with patch(
|
|
"govoplan_files.backend.storage.archives._read_member",
|
|
side_effect=AssertionError("Python decoder must not run"),
|
|
):
|
|
native = self.read()
|
|
with patch(
|
|
"govoplan_files.backend.storage.archives.native_zip_library",
|
|
return_value=None,
|
|
):
|
|
fallback = self.read()
|
|
self.assertEqual(native, fallback)
|
|
self.assertEqual([("first.dat", self.contents["first.dat"])], native)
|
|
|
|
def test_utf8_flagged_german_names_use_native_reader(self):
|
|
# Info-ZIP stores UTF-8 names without the language flag on some hosts.
|
|
# Set the standard UTF-8 flag in local and central headers for this
|
|
# fixture; encryption header verification is unaffected by bit 11.
|
|
encoded_name = "Grüße.txt".encode("utf-8")
|
|
placeholder = "x" * len(encoded_name)
|
|
data = bytearray(
|
|
legacy_archive({placeholder: b"german"}).replace(
|
|
placeholder.encode(), encoded_name
|
|
)
|
|
)
|
|
for signature, offset in ((b"PK\x03\x04", 6), (b"PK\x01\x02", 8)):
|
|
position = data.index(signature) + offset
|
|
flags = int.from_bytes(data[position : position + 2], "little") | 0x800
|
|
data[position : position + 2] = flags.to_bytes(2, "little")
|
|
self.archive = bytes(data)
|
|
with patch(
|
|
"govoplan_files.backend.storage.archives._read_member",
|
|
side_effect=AssertionError("Python decoder must not run"),
|
|
):
|
|
result = list(
|
|
_read_selected_zip_members(
|
|
self.archive,
|
|
selected_files={"Grüße.txt"},
|
|
password="fixture-only",
|
|
max_file_bytes=100,
|
|
max_total_bytes=100,
|
|
)
|
|
)
|
|
self.assertEqual([("Grüße.txt", b"german")], result)
|
|
|
|
def test_wrong_password_and_corruption_never_fall_back(self):
|
|
with patch(
|
|
"govoplan_files.backend.storage.archives._read_member",
|
|
side_effect=AssertionError("No fallback after native error"),
|
|
):
|
|
with self.assertRaisesRegex(FileStorageError, "Native ZIP decoding failed"):
|
|
list(
|
|
_read_selected_zip_members(
|
|
self.archive,
|
|
selected_files={"first.dat"},
|
|
password="wrong",
|
|
max_file_bytes=100_000,
|
|
max_total_bytes=100_000,
|
|
)
|
|
)
|
|
corrupted = bytearray(self.archive)
|
|
with pyzipper.AESZipFile(BytesIO(self.archive)) as archive:
|
|
first = archive.infolist()[0]
|
|
payload_start = (
|
|
first.header_offset
|
|
+ 30
|
|
+ len(first.filename.encode())
|
|
+ len(first.extra)
|
|
)
|
|
corrupted[payload_start + first.compress_size - 1] ^= 0x40
|
|
self.archive = bytes(corrupted)
|
|
with self.assertRaises(FileStorageError):
|
|
self.read()
|
|
|
|
def test_independent_crc_check_rejects_mismatched_metadata(self):
|
|
with pyzipper.AESZipFile(BytesIO(self.archive)) as archive:
|
|
infos = {
|
|
_safe_member_path(info.filename): info for info in archive.infolist()
|
|
}
|
|
infos["first.dat"].CRC ^= 1
|
|
with self.assertRaisesRegex(FileStorageError, "CRC verification"):
|
|
list(
|
|
read_native_zip_members(
|
|
native_zip_library(),
|
|
self.archive,
|
|
infos_by_path=infos,
|
|
selected_files={"first.dat"},
|
|
normalize_name=_safe_member_path,
|
|
password="fixture-only",
|
|
max_file_bytes=100_000,
|
|
max_total_bytes=100_000,
|
|
)
|
|
)
|
|
|
|
def test_legacy_encoded_non_utf8_names_choose_python_before_decoding(self):
|
|
encoded_name = "ä.txt".encode("cp437")
|
|
placeholder = "x" * len(encoded_name)
|
|
self.archive = legacy_archive({placeholder: b"legacy"}).replace(
|
|
placeholder.encode(), encoded_name
|
|
)
|
|
with patch(
|
|
"govoplan_files.backend.storage.archives.native_zip_library",
|
|
side_effect=AssertionError("Legacy encoding must use original reader"),
|
|
):
|
|
result = list(
|
|
_read_selected_zip_members(
|
|
self.archive,
|
|
selected_files={"ä.txt"},
|
|
password="fixture-only",
|
|
max_file_bytes=100,
|
|
max_total_bytes=100,
|
|
)
|
|
)
|
|
self.assertEqual([("ä.txt", b"legacy")], result)
|
|
|
|
def test_suspended_native_reader_is_freed_after_destination_failure(self):
|
|
library = native_zip_library()
|
|
|
|
class LibraryProxy:
|
|
def __getattr__(self, name):
|
|
return getattr(library, name)
|
|
|
|
proxy = LibraryProxy()
|
|
with (
|
|
patch.object(
|
|
proxy, "archive_read_free", wraps=library.archive_read_free
|
|
) as close,
|
|
patch(
|
|
"govoplan_files.backend.storage.archives.native_zip_library",
|
|
return_value=proxy,
|
|
),
|
|
):
|
|
with self.assertRaisesRegex(FileStorageError, "Destination failure"):
|
|
# This unit test owns the native handle locally; public imports
|
|
# now own it in a fresh child (covered by worker cleanup tests).
|
|
with closing(_read_selected_zip_members(
|
|
self.archive, selected_files=set(self.contents), password="fixture-only",
|
|
max_file_bytes=100_000, max_total_bytes=100_000,
|
|
)) as members:
|
|
next(members)
|
|
raise FileStorageError("Destination failure")
|
|
close.assert_called_once()
|
|
|
|
def test_native_read_enforces_actual_member_and_total_limits(self):
|
|
for member_limit, total_limit in ((10, 100_000), (100_000, 10)):
|
|
with self.subTest(member_limit=member_limit, total_limit=total_limit):
|
|
with self.assertRaises(FileStorageError):
|
|
list(
|
|
_read_selected_zip_members(
|
|
self.archive,
|
|
selected_files={"first.dat"},
|
|
password="fixture-only",
|
|
max_file_bytes=member_limit,
|
|
max_total_bytes=total_limit,
|
|
)
|
|
)
|
|
|
|
def test_native_emits_progress_inside_a_large_member(self):
|
|
self.archive = legacy_archive(
|
|
{"first.dat": random.Random(234).randbytes(3 * 1024 * 1024)}
|
|
)
|
|
progress = []
|
|
result = list(
|
|
_read_selected_zip_members(
|
|
self.archive,
|
|
selected_files={"first.dat"},
|
|
password="fixture-only",
|
|
max_file_bytes=4 * 1024 * 1024,
|
|
max_total_bytes=4 * 1024 * 1024,
|
|
total_bytes=3 * 1024 * 1024,
|
|
progress=lambda *args: progress.append(args),
|
|
)
|
|
)
|
|
self.assertEqual(3 * 1024 * 1024, len(result[0][1]))
|
|
self.assertTrue(
|
|
any(item[1] == 0 and 0 < item[3] < 3 * 1024 * 1024 for item in progress)
|
|
)
|
|
self.assertEqual(
|
|
("extracting", 1, 1, 3 * 1024 * 1024, 3 * 1024 * 1024), progress[-1]
|
|
)
|
|
|
|
|
|
class ArchiveStreamingTests(unittest.TestCase):
|
|
def test_plain_zip_and_tar_store_one_member_before_reading_the_next(self):
|
|
entries = {"one.txt": b"one", "two.txt": b"two"}
|
|
for filename, archive in (
|
|
("files.zip", _zip_bytes(entries)),
|
|
("files.tar.gz", _tar_bytes(entries)),
|
|
):
|
|
with self.subTest(filename=filename):
|
|
progress = []
|
|
with patch(
|
|
"govoplan_files.backend.storage.archives.create_file_asset",
|
|
return_value=SimpleNamespace(asset=object()),
|
|
) as create:
|
|
result = extract_archive_upload(
|
|
object(),
|
|
tenant_id="tenant",
|
|
owner_type="user",
|
|
owner_id="user",
|
|
user_id="user",
|
|
archive_data=archive,
|
|
filename=filename,
|
|
folder="",
|
|
campaign_id=None,
|
|
progress=lambda *args: progress.append(args),
|
|
)
|
|
self.assertEqual(2, len(result))
|
|
self.assertEqual(2, create.call_count)
|
|
stored_first = progress.index(("storing", 1, 2, 3, 6))
|
|
extracted_second = progress.index(("extracting", 2, 2, 6, 6))
|
|
self.assertLess(stored_first, extracted_second)
|
|
|
|
def test_aes_archive_keeps_existing_decoder(self):
|
|
with patch(
|
|
"govoplan_files.backend.storage.archives.native_zip_library",
|
|
side_effect=AssertionError("AES must not use native path"),
|
|
):
|
|
result = list(
|
|
_read_selected_zip_members(
|
|
_encrypted_zip_bytes("correct"),
|
|
selected_files={"secure/report.txt"},
|
|
password="correct",
|
|
max_file_bytes=100,
|
|
max_total_bytes=100,
|
|
)
|
|
)
|
|
self.assertEqual([("secure/report.txt", b"classified")], result)
|
|
|
|
def test_late_unsafe_header_prevents_every_write(self):
|
|
with patch(
|
|
"govoplan_files.backend.storage.archives.create_file_asset"
|
|
) as create:
|
|
with self.assertRaisesRegex(FileStorageError, "Unsafe archive member"):
|
|
extract_archive_upload(
|
|
object(),
|
|
tenant_id="tenant",
|
|
owner_type="user",
|
|
owner_id="user",
|
|
user_id="user",
|
|
archive_data=_zip_bytes({"good.txt": b"good", "../unsafe": b"bad"}),
|
|
filename="unsafe.zip",
|
|
folder="",
|
|
campaign_id=None,
|
|
)
|
|
create.assert_not_called()
|
|
|
|
|
|
class ArchivePersistencePerformanceTests(unittest.TestCase):
|
|
def test_backend_scope_is_lazy_nested_and_reset_after_exception(self):
|
|
first_backend = object()
|
|
second_backend = object()
|
|
with patch(
|
|
"govoplan_files.backend.storage.files.get_storage_backend",
|
|
side_effect=[first_backend, second_backend],
|
|
) as factory:
|
|
with self.assertRaisesRegex(RuntimeError, "fixture failure"):
|
|
with archive_storage_backend_scope():
|
|
factory.assert_not_called()
|
|
self.assertIs(first_backend, _archive_write_backend())
|
|
with archive_storage_backend_scope():
|
|
self.assertIs(first_backend, _archive_write_backend())
|
|
factory.assert_called_once()
|
|
raise RuntimeError("fixture failure")
|
|
with archive_storage_backend_scope():
|
|
self.assertIs(second_backend, _archive_write_backend())
|
|
self.assertEqual(2, factory.call_count)
|
|
|
|
def test_backend_reuse_does_not_cache_tenant_identity_or_blob_dedup(self):
|
|
fixture = recovery_fixture.StorageRecoveryTests()
|
|
fixture.setUp()
|
|
self.addCleanup(fixture.doCleanups)
|
|
results = []
|
|
with patch(
|
|
"govoplan_files.backend.storage.files.get_storage_backend",
|
|
return_value=fixture.backend,
|
|
) as factory:
|
|
for tenant in ("tenant-one", "tenant-two"):
|
|
results.append(
|
|
extract_archive_upload(
|
|
fixture.session,
|
|
tenant_id=tenant,
|
|
owner_type="user",
|
|
owner_id="user-1",
|
|
user_id="user-1",
|
|
archive_data=_zip_bytes({"one.txt": b"one", "two.txt": b"two"}),
|
|
filename="files.zip",
|
|
folder="",
|
|
campaign_id=None,
|
|
)
|
|
)
|
|
fixture.session.commit()
|
|
self.assertEqual(2, factory.call_count)
|
|
self.assertEqual(
|
|
["tenant-one", "tenant-two"], [items[0].blob.tenant_id for items in results]
|
|
)
|
|
self.assertNotEqual(results[0][0].blob.id, results[1][0].blob.id)
|
|
|
|
def test_asset_and_version_are_published_once_without_redundant_update(self):
|
|
fixture = recovery_fixture.StorageRecoveryTests()
|
|
fixture.setUp()
|
|
self.addCleanup(fixture.doCleanups)
|
|
register_files_change_tracking()
|
|
queries = []
|
|
event.listen(
|
|
fixture.engine,
|
|
"before_cursor_execute",
|
|
lambda conn, cursor, statement, parameters, context, executemany: (
|
|
queries.append(statement)
|
|
),
|
|
)
|
|
with patch(
|
|
"govoplan_files.backend.storage.files.get_storage_backend",
|
|
return_value=fixture.backend,
|
|
):
|
|
stored = extract_archive_upload(
|
|
fixture.session,
|
|
tenant_id="tenant-1",
|
|
owner_type="user",
|
|
owner_id="user-1",
|
|
user_id="user-1",
|
|
archive_data=_zip_bytes({"one.txt": b"one", "two.txt": b"two"}),
|
|
filename="files.zip",
|
|
folder="",
|
|
campaign_id=None,
|
|
)
|
|
fixture.session.commit()
|
|
self.assertFalse(
|
|
any(statement.startswith("UPDATE file_assets") for statement in queries)
|
|
)
|
|
events = (
|
|
fixture.session.query(ChangeSequenceEntry)
|
|
.filter(ChangeSequenceEntry.module_id == "files")
|
|
.all()
|
|
)
|
|
self.assertEqual(["created", "created"], [entry.operation for entry in events])
|
|
self.assertTrue(
|
|
all(item.asset.current_version_id == item.version.id for item in stored)
|
|
)
|
|
|
|
def test_streamed_late_content_limit_failure_rolls_back_previous_blobs(self):
|
|
fixture = recovery_fixture.StorageRecoveryTests()
|
|
fixture.setUp()
|
|
self.addCleanup(fixture.doCleanups)
|
|
with patch(
|
|
"govoplan_files.backend.storage.files.get_storage_backend",
|
|
return_value=fixture.backend,
|
|
):
|
|
with self.assertRaisesRegex(FileStorageError, "per-file limit"):
|
|
extract_archive_upload(
|
|
fixture.session,
|
|
tenant_id="tenant-1",
|
|
owner_type="user",
|
|
owner_id="user-1",
|
|
user_id="user-1",
|
|
archive_data=_zip_bytes(
|
|
{"one.txt": b"one", "too-large.txt": b"x" * 200}
|
|
),
|
|
filename="files.zip",
|
|
folder="",
|
|
campaign_id=None,
|
|
max_file_bytes=100,
|
|
)
|
|
fixture.session.rollback()
|
|
self.assertEqual([], list(fixture.backend.root.rglob("*.blob")))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|