145 lines
6.3 KiB
Python
145 lines
6.3 KiB
Python
"""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",
|
|
"crates/argand-site-registry/examples/observer.env": "ARGAND_REGISTRY_PIN=fixture\n",
|
|
"crates/argand-site-registry/tests/fixtures/v03-contract.json": "{}\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):
|
|
release.safe_path("crates/argand-site-registry/examples/observer.env")
|
|
release.safe_path("crates/argand-site-registry/tests/fixtures/v03-contract.json")
|
|
for path in ("/tmp/a.rs", "a/../b.rs", "a//b.rs", "a\\b.rs", "data/a.md",
|
|
".git/config.toml", "secret.key", "secrets.env",
|
|
"config/observer.env", "other.json", "fixtures/provider.json",
|
|
"x\na.rs"):
|
|
with self.subTest(path=path), self.assertRaises(ValueError):
|
|
release.safe_path(path)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|