File size: 6,494 Bytes
29ed697 81d1607 29ed697 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 | """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})
|