| """Adversarial archive validation and immutable/idempotent extraction checks.""" |
| import hashlib |
| import io |
| from pathlib import Path |
| import stat |
| import tempfile |
| import unittest |
| import warnings |
| import zipfile |
|
|
| import references as refs |
|
|
|
|
| def archive_bytes(entries): |
| output = io.BytesIO() |
| with warnings.catch_warnings(): |
| warnings.simplefilter("ignore", UserWarning) |
| with zipfile.ZipFile(output, "w", zipfile.ZIP_DEFLATED) as archive: |
| for name, data in entries: |
| archive.writestr(name, data) |
| return output.getvalue() |
|
|
|
|
| class ExtractionTests(unittest.TestCase): |
| def setUp(self): |
| self.temp = tempfile.TemporaryDirectory() |
| self.root = Path(self.temp.name).resolve() |
|
|
| def tearDown(self): |
| self.temp.cleanup() |
|
|
| def fixture(self, entries): |
| data = archive_bytes(entries) |
| path = self.root / "fixture.zip" |
| path.write_bytes(data) |
| package = {"build": "test", "filename": path.name, "bytes": len(data), |
| "sha256": hashlib.sha256(data).hexdigest(), "archive_entries": len(entries)} |
| return path, package |
|
|
| def test_extract_preserves_bytes_empty_dirs_and_is_idempotent(self): |
| entries = [("empty/", b""), ("data/sub/file.txt", b"\xff\r\nlegacy text"), |
| ("tools/binary.exe", b"MZ\0binary"), ("blank", b"")] |
| path, package = self.fixture(entries) |
| inventory = refs.inspect_archive(path, package) |
| destination = self.root / "extracted/test" |
| self.assertEqual(refs.extract(path, destination, inventory), "extracted") |
| self.assertEqual(refs.extract(path, destination, inventory), "unchanged") |
| for name, data in entries: |
| if not name.endswith("/"): |
| self.assertEqual((destination / name).read_bytes(), data) |
| self.assertFalse((destination / name).stat().st_mode & 0o111) |
| self.assertTrue((destination / "empty").is_dir()) |
| self.assertEqual(inventory["files"][0]["sha256"], hashlib.sha256(entries[1][1]).hexdigest()) |
|
|
| def test_hydrated_clone_restores_only_missing_empty_directories(self): |
| path, package = self.fixture([("data/models/", b""), ("file.txt", b"original")]) |
| inventory = refs.inspect_archive(path, package) |
| dest = self.root / "out" |
| refs.extract(path, dest, inventory) |
| (dest / "data/models").rmdir() |
| (dest / "data").rmdir() |
| self.assertEqual(refs.extract(path, dest, inventory), "restored directories") |
| self.assertTrue((dest / "data/models").is_dir()) |
| (dest / "data/models").rmdir() |
| (dest / "file.txt").write_bytes(b"edited") |
| with self.assertRaises(ValueError): |
| refs.extract(path, dest, inventory) |
| self.assertFalse((dest / "data/models").exists()) |
| self.assertEqual((dest / "file.txt").read_bytes(), b"edited") |
|
|
| def test_modified_output_is_never_overwritten(self): |
| path, package = self.fixture([("f.txt", b"raw")]) |
| inventory = refs.inspect_archive(path, package) |
| dest = self.root / "out" |
| refs.extract(path, dest, inventory) |
| (dest / "f.txt").write_bytes(b"edited") |
| with self.assertRaises(ValueError): |
| refs.extract(path, dest, inventory) |
| self.assertEqual((dest / "f.txt").read_bytes(), b"edited") |
|
|
| def test_unexpected_file_or_directory_is_rejected(self): |
| path, package = self.fixture([("f.txt", b"raw")]) |
| inventory = refs.inspect_archive(path, package) |
| dest = self.root / "out" |
| refs.extract(path, dest, inventory) |
| (dest / "unowned").mkdir() |
| with self.assertRaises(ValueError): |
| refs.check_extracted(dest, inventory) |
|
|
| def test_hash_and_entry_count_fail_before_extraction(self): |
| path, package = self.fixture([("file", b"raw")]) |
| for change in [{"sha256": "0" * 64}, {"bytes": 1}, {"archive_entries": 10}]: |
| with self.subTest(change=change), self.assertRaises(ValueError): |
| refs.inspect_archive(path, {**package, **change}) |
|
|
| def test_unsafe_paths_and_collisions(self): |
| cases = [[(name, b"data")] for name in ["../escape", "/absolute", "C:/drive", "a\\b", "a/../b", |
| "a/./b", "a//b", "CON.txt", "a/NUL", "a/trailing.", "a/end ", "bad\nname"]] |
| cases += [[("file", b"a"), ("file", b"b")], [("A", b"a"), ("a", b"b")], |
| [("A/f", b"a"), ("a/g", b"b")], [("file", b"a"), ("file/child", b"b")], |
| [("caf\u00e9/f", b"a"), ("cafe\u0301/g", b"b")]] |
| for entries in cases: |
| with self.subTest(entries=entries), zipfile.ZipFile(io.BytesIO(archive_bytes(entries))) as z: |
| with self.assertRaises(ValueError): |
| refs.checked_members(z) |
|
|
| def test_symlink_members_and_destination_are_rejected(self): |
| link = zipfile.ZipInfo("link") |
| link.create_system = 3 |
| link.external_attr = (stat.S_IFLNK | 0o777) << 16 |
| with zipfile.ZipFile(io.BytesIO(archive_bytes([(link, b"../elsewhere")]))) as z: |
| with self.assertRaises(ValueError): |
| refs.checked_members(z) |
| path, package = self.fixture([("file", b"data")]) |
| inventory = refs.inspect_archive(path, package) |
| (self.root / "link").symlink_to(self.root, target_is_directory=True) |
| with self.assertRaises(ValueError): |
| refs.extract(path, self.root / "link/out", inventory) |
|
|
| def test_expansion_limits(self): |
| with zipfile.ZipFile(io.BytesIO(archive_bytes([("large", b"0" * 1000000)]))) as z: |
| with self.assertRaises(ValueError): |
| refs.checked_members(z) |
|
|
| def test_lock_prevents_concurrent_writers_and_cleans_up(self): |
| with refs.exclusive(self.root / "refs"): |
| with self.assertRaises(ValueError): |
| with refs.exclusive(self.root / "refs"): |
| self.fail("Lock permitted a second writer") |
| self.assertFalse((self.root / "refs/.unpack.lock").exists()) |
|
|
| def test_inventory_write_is_atomic_idempotent_and_rejects_symlink(self): |
| path = self.root / "inventory/test.json" |
| refs.write_json(path, {"a": 1}) |
| before = path.stat().st_mtime_ns |
| refs.write_json(path, {"a": 1}) |
| self.assertEqual(before, path.stat().st_mtime_ns) |
| path.unlink() |
| path.symlink_to(self.root / "other") |
| with self.assertRaises(ValueError): |
| refs.write_json(path, {"a": 2}) |
|
|
