feat: establish standalone Argand Site Registry

tests/test_source_release.py

@@ -0,0 +1,139 @@
+"""Exercise the shipped source packager against actual Git trees and corrupt archives."""
+
+import importlib.util
+import io
+import json
+import subprocess
+import tarfile
+import tempfile
+import unittest
+from pathlib import Path
+
+MODULE = Path(__file__).resolve().parents[1] / "scripts" / "source_release.py"
+SPEC = importlib.util.spec_from_file_location("source_release", MODULE)
+release = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(release)
+
+
+class SourceReleaseTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.keys = tempfile.TemporaryDirectory()
+        cls.key = Path(cls.keys.name) / "fixture"
+        subprocess.run(["ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-f", str(cls.key)],
+                       check=True)
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.keys.cleanup()
+
+    def setUp(self):
+        self.temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temporary.cleanup)
+        self.root = Path(self.temporary.name) / "source with spaces"
+        self.root.mkdir()
+        self.git("init", "-q", "-b", "main")
+        self.git("config", "user.name", "Synthetic test")
+        self.git("config", "user.email", "test@example.invalid")
+        self.git("config", "core.excludesFile", "/dev/null")
+        self.git("config", "gpg.format", "ssh")
+        self.git("config", "user.signingkey", str(self.key))
+        self.git("config", "commit.gpgsign", "true")
+        files = {
+            "Cargo.toml": '[workspace.package]\nversion="0.1.0"\n',
+            "Cargo.lock": "# Synthetic lock\n",
+            "LICENSE": "Synthetic code license fixture\n",
+            "UPSTREAM.json": "{}\n",
+            "crates/argand-site-registry/LICENSE_SOURCES.md": "Synthetic source terms\n",
+            "src/lib.rs": "// Synthetic Rust source\n",
+            "src/.gitignore": "*.temporary\n",
+        }
+        for name, text in files.items():
+            path = self.root / name
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text(text)
+        self.commit()
+        self.output = self.root.parent / "release"
+
+    def git(self, *args):
+        return subprocess.check_output(["git", "-C", str(self.root), *args])
+
+    def commit(self):
+        self.git("add", "--all")
+        self.git("commit", "-q", "-m", "synthetic fixture")
+
+    def test_reproducible_archive_verifies_and_keeps_licenses(self):
+        pin = release.create(self.root, self.output)
+        second = self.root.parent / "second"
+        self.assertEqual(pin, release.create(self.root, second))
+        self.assertEqual((self.output / release.ARCHIVE).read_bytes(),
+                         (second / release.ARCHIVE).read_bytes())
+        receipt = release.verify(self.output, pin)
+        self.assertIn("LICENSE", [item["path"] for item in receipt["files"]])
+
+    def test_dirty_and_untracked_source_refused(self):
+        (self.root / "new.rs").write_text("// Not committed\n")
+        with self.assertRaisesRegex(ValueError, "clean committed"):
+            release.create(self.root, self.output)
+        self.assertFalse(self.output.exists())
+
+    def test_output_is_never_replaced(self):
+        pin = release.create(self.root, self.output)
+        with self.assertRaises(FileExistsError):
+            release.create(self.root, self.output)
+        release.verify(self.output, pin)
+
+    def test_tampering_and_wrong_trust_pin_refused(self):
+        pin = release.create(self.root, self.output)
+        with self.assertRaisesRegex(ValueError, "pin mismatch"):
+            release.verify(self.output, "0" * 64)
+        archive = self.output / release.ARCHIVE
+        archive.write_bytes(archive.read_bytes() + b"tampered")
+        with self.assertRaisesRegex(ValueError, "digest mismatch"):
+            release.verify(self.output, pin)
+
+    def test_link_and_dataset_excluded_before_publication(self):
+        (self.root / "link.rs").symlink_to("src/lib.rs")
+        self.commit()
+        with self.assertRaisesRegex(ValueError, "links or submodules"):
+            release.create(self.root, self.output)
+        (self.root / "link.rs").unlink()
+        (self.root / "private.sqlite").write_bytes(b"not a dataset release")
+        self.commit()
+        with self.assertRaisesRegex(ValueError, "allowlist"):
+            release.create(self.root, self.output)
+
+    def test_unsafe_member_refused_even_with_matching_archive_digest(self):
+        release.create(self.root, self.output)
+        receipt_path = self.output / release.RECEIPT
+        receipt = json.loads(receipt_path.read_text())
+        buffer = io.BytesIO()
+        with tarfile.open(fileobj=buffer, mode="w:gz") as archive:
+            member = tarfile.TarInfo(receipt["prefix"] + "../../escape.rs")
+            member.size = 4
+            archive.addfile(member, io.BytesIO(b"oops"))
+        data = buffer.getvalue()
+        (self.output / release.ARCHIVE).write_bytes(data)
+        receipt.update(sha256=release.digest(data), bytes=len(data))
+        encoded = json.dumps(receipt).encode()
+        receipt_path.write_bytes(encoded)
+        with self.assertRaisesRegex(ValueError, "excluded archive path"):
+            release.verify(self.output, release.digest(encoded))
+
+    def test_duplicate_receipt_keys_refused(self):
+        release.create(self.root, self.output)
+        receipt_path = self.output / release.RECEIPT
+        encoded = receipt_path.read_bytes().replace(b'"schema":', b'"schema":"duplicate","schema":', 1)
+        receipt_path.write_bytes(encoded)
+        with self.assertRaisesRegex(ValueError, "duplicate JSON key"):
+            release.verify(self.output, release.digest(encoded))
+
+    def test_special_source_paths_refused(self):
+        for path in ("/tmp/a.rs", "a/../b.rs", "a//b.rs", "a\\b.rs", "data/a.md",
+                     ".git/config.toml", "secret.key", "x\na.rs"):
+            with self.subTest(path=path), self.assertRaises(ValueError):
+                release.safe_path(path)
+
+
+if __name__ == "__main__":
+    unittest.main()