"""Temporary local fixtures only. Identity codec tests do not certify zstd. The optional real-zstd test automatically runs when the executable is present. """ import contextlib import gzip import hashlib import importlib.util import io import json import os from pathlib import Path import shutil import sys import tarfile import tempfile import unittest from unittest.mock import patch SPEC = importlib.util.spec_from_file_location("lfs_archive_tested", Path(__file__).with_name("publish_archives.py")) M = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(M) def codec(*args, **kwargs): return [sys.executable, "-c", "import sys,shutil;shutil.copyfileobj(sys.stdin.buffer,sys.stdout.buffer)"] class ArchiveTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.base = Path(self.temp.name) self.root = self.base / "source" (self.root / "data").mkdir(parents=True) (self.root / "data" / "a.txt").write_text("public source\n", encoding="utf-8") (self.root / "data" / "b.bin").write_bytes(bytes(range(256)) * 80) (self.root / "data" / "zero.txt").write_bytes(b"") self.plan = {"schema": "lfs-publication-plan-v1", "source_root": str(self.root), "groups": [{"id": "data", "paths": ["data"], "license": "test-only", "provenance": {"source": "synthetic"}}], "exclude_paths": []} self.plan_path = self.base / "plan.json" self.plan_path.write_bytes(M.canonical(self.plan)) self.out, self.work = self.base / "out", self.base / "work" self.addCleanup(patch.stopall) patch.object(M, "SOURCE_ROOT", str(self.root)).start() patch.object(M, "CHUNK_BYTES", 4096).start() self.codec_patch = patch.object(M, "zstd_command", codec) self.codec_patch.start() def pack(self): with contextlib.redirect_stdout(io.StringIO()): return M.pack(self.plan_path, self.out, self.work) def catalog(self): return json.loads((self.out / "data" / "catalog.json").read_bytes()) def save_catalog(self, cat): (self.out / "data" / "catalog.json").write_bytes(M.canonical(cat)) def rewrite_archive(self, data): group = self.out / "data" cat = self.catalog() for row in cat["chunks"]: (group / row["path"]).unlink() new = self.base / "mutant.tar" new.write_bytes(data) cat["chunks"], cat["archive_sha256"] = M.split_archive(new, group) cat["archive_bytes"] = len(data) self.save_catalog(cat) def test_pack_verify_full_member_content_and_no_extraction(self): before = {p.relative_to(self.root).as_posix(): p.read_bytes() for p in self.root.rglob("*") if p.is_file()} summary = self.pack() result = M.verify(self.out / "data") self.assertTrue(result["pass_all"]) self.assertEqual(result["source_files"], 3) self.assertEqual(result["source_bytes"], sum(map(len, before.values()))) self.assertGreater(summary["groups"][0]["chunks"], 1) self.assertEqual(before, {p.relative_to(self.root).as_posix(): p.read_bytes() for p in self.root.rglob("*") if p.is_file()}) self.assertFalse((self.out / "data" / "a.txt").exists()) def test_plan_executes_first_read_bytes_only(self): with patch.object(M, "json_read", side_effect=AssertionError("Must not read plan a second time")): self.pack() self.assertEqual(self.catalog()["plan_sha256"], hashlib.sha256(self.plan_path.read_bytes()).hexdigest()) def test_exclusions_logged_without_descending(self): for name in (".git", "venv", "__pycache__", ".cache", "node_modules", "logs", "phase4ar-normal-conditioning"): d = self.root / "data" / name d.mkdir() (d / "hidden.txt").write_text("skip") (self.root / "data" / "private.pem").write_text("private") (self.root / "data" / "run.log").write_text("log") self.pack() with gzip.open(self.out / "data" / "exclusions.jsonl.gz", "rt") as f: rows = [json.loads(x) for x in f] self.assertEqual(len(rows), 9) self.assertEqual(self.catalog()["source_files"], 3) self.assertTrue(M.verify(self.out / "data")["pass_all"]) def test_unsafe_explicit_paths(self): for name in ("../data", "/data", "data/../data", "data//x", "data\\x", "data/.git", "data/phase4ar-normal-conditioning"): plan = json.loads(json.dumps(self.plan)) plan["groups"][0]["paths"] = [name] with self.subTest(name=name), self.assertRaises(ValueError): M.validate_plan(plan) def test_overlapping_groups_refused(self): self.plan["groups"].append({"id": "nested", "paths": ["data/a.txt"], "license": "test", "provenance": "test"}) with self.assertRaisesRegex(ValueError, "Overlapping"): M.validate_plan(self.plan) def test_symlink_refused(self): try: (self.root / "data" / "link").symlink_to(self.root / "data" / "a.txt") except OSError: self.skipTest("OS does not permit temporary symlinks") with self.assertRaisesRegex(ValueError, "Symlink"): self.pack() def test_exclusive_and_disjoint_outputs(self): self.out.mkdir() with self.assertRaises(FileExistsError): self.pack() with self.assertRaisesRegex(ValueError, "disjoint"): M.pack(self.plan_path, self.root / "output", self.work) def test_parent_components_cannot_bypass_disjoint_gate(self): malicious = self.base / "unused" / ".." / "source" / "new-output" with self.assertRaisesRegex(ValueError, "Parent path components"): M.pack(self.plan_path, malicious, self.work) self.assertFalse((self.root / "new-output").exists()) def test_secret_rules_hide_values(self): # Build synthetic fixtures at runtime; keep actual credential-like # strings out of Git blobs so the repository secret guard stays useful. examples = ["-----BEGIN " + "OPENSSH PRIVATE " + "KEY-----", "ghp_" + "a" * 36, "github_pat_" + "b" * 40, "AKIA" + "A" * 16, "https://" + "account:" + "privatevalue" + "@example.org/", 'password' + ' = "privatevalue"', 'api_key' + ': "privatevalue"', 'secret_key' + '=ABCDEF123456', 'pa\u017f\u017fword' + '="privatevalue"', 'ap\u0131_key' + '="privatevalue"', '\u017fecret_key' + '="privatevalue"'] path = self.root / "data" / "literal.txt" for value in examples: path.write_text(value) files, _ = M.enumerate_group(self.root, self.plan["groups"][0], []) with self.subTest(rule=value[:8]), self.assertRaisesRegex(ValueError, "Secret scan refused") as caught: M.scan_secrets(self.root, files) self.assertNotIn(value, str(caught.exception)) self.assertNotIn("privatevalue", str(caught.exception)) def test_mutated_source_metadata_refused(self): files, _ = M.enumerate_group(self.root, self.plan["groups"][0], []) (self.root / "data" / "a.txt").write_text("changed bytes") with self.assertRaisesRegex(ValueError, "Source changed"): M.make_archive(self.root, files, self.base / "bad.tar", 1) def test_payload_corruption_refused(self): self.pack() p = self.out / "data" / self.catalog()["chunks"][0]["path"] data = bytearray(p.read_bytes()) data[-1] ^= 1 p.write_bytes(data) with self.assertRaises((ValueError, tarfile.TarError)): M.verify(self.out / "data") def test_missing_extra_or_reordered_payload_refused(self): self.pack() extra = self.out / "data" / "extra.bin" extra.write_bytes(b"x") with self.assertRaisesRegex(ValueError, "Extra/missing"): M.verify(self.out / "data") extra.unlink() cat = self.catalog() cat["chunks"] = list(reversed(cat["chunks"])) self.save_catalog(cat) with self.assertRaisesRegex(ValueError, "chunk name/order"): M.verify(self.out / "data") def test_tar_traversal_rejected_even_with_rehashed_chunks(self): self.pack() buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w") as archive: info = tarfile.TarInfo("../escape") info.size = 1 archive.addfile(info, io.BytesIO(b"x")) self.rewrite_archive(buf.getvalue()) with self.assertRaisesRegex(ValueError, "Unsafe relative path"): M.verify(self.out / "data") def test_tar_link_rejected_even_with_rehashed_chunks(self): self.pack() buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w") as archive: info = tarfile.TarInfo("data/a.txt") info.type, info.linkname = tarfile.SYMTYPE, "/tmp/escape" archive.addfile(info) self.rewrite_archive(buf.getvalue()) with self.assertRaisesRegex(ValueError, "Nonregular"): M.verify(self.out / "data") def test_hidden_tar_after_end_refused(self): self.pack() raw = (self.work / "data.tar.zst").read_bytes() self.rewrite_archive(raw + b"nonzero hidden second archive") with self.assertRaisesRegex(ValueError, "Nonzero data after tar"): M.verify(self.out / "data") def test_second_tar_after_end_refused(self): self.pack() buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w") as archive: info = tarfile.TarInfo("unlisted-secret.txt") info.size = 1 archive.addfile(info, io.BytesIO(b"x")) self.rewrite_archive((self.work / "data.tar.zst").read_bytes() + buf.getvalue()) with self.assertRaisesRegex(ValueError, "Nonzero data after tar"): M.verify(self.out / "data") def test_member_hash_mutation_refused(self): self.pack() p = self.out / "data" / "members.jsonl.gz" with gzip.open(p, "rt") as f: rows = [json.loads(x) for x in f] rows[0]["sha256"] = "0" * 64 p.unlink() M.gzip_rows(p, rows) cat = self.catalog() cat["members"] = M.sha_file(p) self.save_catalog(cat) with self.assertRaisesRegex(ValueError, "member content mismatch"): M.verify(self.out / "data") @unittest.skipUnless(shutil.which("zstd"), "Real zstd unavailable; run this integration test on Linux") def test_real_zstd_roundtrip(self): self.codec_patch.stop() self.pack() self.assertTrue(M.verify(self.out / "data")["pass_all"]) if __name__ == "__main__": unittest.main()