Commander1024
cdc4d4a8df
Record the finite dependency search, Wi-Fi maintenance blocker, and pinned icon provenance. Add bounded host orchestration and fixture coverage, and update release documentation with current evidence.
299 lines
13 KiB
Python
299 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
# SPDX-License-Identifier: GPL-3.0-only
|
|
"""Temporary-fixture contract tests; no installed dependencies or device required."""
|
|
|
|
import copy
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
sys.dont_write_bytecode = True
|
|
from icon_provenance import IconProvenanceTests
|
|
|
|
PROJECT = Path(__file__).absolute().parents[2]
|
|
SPEC = importlib.util.spec_from_file_location("release_notices", PROJECT / "tools/release_notices.py")
|
|
notices = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(notices)
|
|
|
|
|
|
class BundleTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory(prefix="release-notices-test-")
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.base = Path(self.temp.name)
|
|
self.roots = {name: self.base / name for name in notices.ROOTS}
|
|
for root in self.roots.values():
|
|
root.mkdir()
|
|
self.catalog = {"schema": 1, "snapshot": {"fixture": "1"}, "inputs": []}
|
|
self.add_input("project", "LICENSE", b"Full license\nCopyright holder\n")
|
|
self.add_input("sdk", "nested/COPYING", b"First grant\nSecond grant\nDisclaimer\n")
|
|
self.add_input("toolchain", "source.c", b"/* full notice */\nint code;\n", [0, 18])
|
|
self.output = self.base / "bundle"
|
|
|
|
def add_input(self, root, path, data, span=None):
|
|
target = self.roots[root] / path
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_bytes(data)
|
|
payload = data if span is None else data[span[0]:span[1]]
|
|
self.catalog["inputs"].append({
|
|
"root": root, "path": path, "size": len(data),
|
|
"sha256": notices.digest(data), "range": span,
|
|
"output_sha256": notices.digest(payload), "purpose": "test fixture",
|
|
})
|
|
|
|
def run_bundle(self, output=None, catalog=None):
|
|
return notices.assemble(self.roots, output or self.output,
|
|
notices.json_bytes(catalog or self.catalog))
|
|
|
|
def assert_preflight_failure(self):
|
|
with self.assertRaises((notices.NoticeError, OSError)):
|
|
self.run_bundle()
|
|
self.assertFalse(self.output.exists())
|
|
|
|
@staticmethod
|
|
def contents(root):
|
|
return {p.relative_to(root).as_posix(): p.read_bytes()
|
|
for p in root.rglob("*") if p.is_file()}
|
|
|
|
def test_exact_bytes_and_manifest(self):
|
|
manifest = self.run_bundle()
|
|
self.assertEqual(json.loads((self.output / "manifest.json").read_bytes()), manifest)
|
|
for entry in manifest["inputs"]:
|
|
data = (self.output / entry["output"]).read_bytes()
|
|
self.assertEqual(notices.digest(data), entry["output_sha256"])
|
|
self.assertEqual(len(data), entry["output_size"])
|
|
self.assertEqual((self.output / "inputs/sdk/nested/COPYING").read_bytes(),
|
|
b"First grant\nSecond grant\nDisclaimer\n")
|
|
self.assertEqual((self.output / "inputs/toolchain/source.c.notice.txt").read_bytes(),
|
|
b"/* full notice */\n")
|
|
self.assertNotIn(str(self.base), (self.output / "manifest.json").read_text())
|
|
self.assertEqual(self.output.stat().st_mode & 0o777, 0o700)
|
|
|
|
def test_deterministic_order_paths_and_mtime(self):
|
|
self.run_bundle()
|
|
first = self.contents(self.output)
|
|
# Moving the input roots and changing source mtimes must not affect bytes.
|
|
for name, root in list(self.roots.items()):
|
|
moved = self.base / (name + "-moved")
|
|
root.rename(moved)
|
|
self.roots[name] = moved
|
|
for file in moved.rglob("*"):
|
|
os.utime(file, (123456789, 123456789))
|
|
self.run_bundle(self.base / "second")
|
|
self.assertEqual(first, self.contents(self.base / "second"))
|
|
|
|
def test_missing_source(self):
|
|
(self.roots["sdk"] / "nested/COPYING").unlink()
|
|
self.assert_preflight_failure()
|
|
|
|
def test_changed_source_same_size(self):
|
|
path = self.roots["project"] / "LICENSE"
|
|
path.write_bytes(b"x" * path.stat().st_size)
|
|
self.assert_preflight_failure()
|
|
|
|
def test_changed_non_notice_source_body(self):
|
|
path = self.roots["toolchain"] / "source.c"
|
|
path.write_bytes(path.read_bytes().replace(b"code", b"evil"))
|
|
self.assert_preflight_failure()
|
|
|
|
def test_empty_source(self):
|
|
(self.roots["project"] / "LICENSE").write_bytes(b"")
|
|
self.assert_preflight_failure()
|
|
|
|
def test_growing_source(self):
|
|
path = self.roots["project"] / "LICENSE"
|
|
with path.open("ab") as stream:
|
|
stream.write(b"unexpected")
|
|
self.assert_preflight_failure()
|
|
|
|
def test_existing_directory_never_modified(self):
|
|
self.output.mkdir()
|
|
marker = self.output / "user-data"
|
|
marker.write_bytes(b"keep me")
|
|
with self.assertRaises(FileExistsError):
|
|
self.run_bundle()
|
|
self.assertEqual(self.contents(self.output), {"user-data": b"keep me"})
|
|
|
|
def test_existing_empty_directory_rejected(self):
|
|
self.output.mkdir()
|
|
with self.assertRaises(FileExistsError):
|
|
self.run_bundle()
|
|
self.assertEqual(list(self.output.iterdir()), [])
|
|
|
|
def test_existing_file_never_modified(self):
|
|
self.output.write_bytes(b"user data")
|
|
with self.assertRaises(FileExistsError):
|
|
self.run_bundle()
|
|
self.assertEqual(self.output.read_bytes(), b"user data")
|
|
|
|
def test_output_symlink_rejected(self):
|
|
self.output.symlink_to(self.base / "absent")
|
|
with self.assertRaises(FileExistsError):
|
|
self.run_bundle()
|
|
self.assertTrue(self.output.is_symlink())
|
|
self.assertFalse((self.base / "absent").exists())
|
|
|
|
def test_output_parent_symlink_rejected(self):
|
|
link = self.base / "link"
|
|
link.symlink_to(self.base, target_is_directory=True)
|
|
with self.assertRaises(OSError):
|
|
self.run_bundle(link / "bundle")
|
|
self.assertFalse(self.output.exists())
|
|
|
|
def test_output_inside_inputs_rejected(self):
|
|
for root in self.roots.values():
|
|
with self.subTest(root=root), self.assertRaises(notices.NoticeError):
|
|
self.run_bundle(root / "bundle")
|
|
self.assertFalse((root / "bundle").exists())
|
|
|
|
def test_parent_traversal_output_rejected(self):
|
|
with self.assertRaises(notices.NoticeError):
|
|
self.run_bundle(self.base / "project/../bundle")
|
|
self.assertFalse(self.output.exists())
|
|
|
|
def test_missing_parent_not_created(self):
|
|
with self.assertRaises(FileNotFoundError):
|
|
self.run_bundle(self.base / "absent/bundle")
|
|
self.assertFalse((self.base / "absent").exists())
|
|
|
|
def test_source_file_symlink_even_to_identical_bytes_rejected(self):
|
|
path = self.roots["project"] / "LICENSE"
|
|
other = self.base / "other"
|
|
path.rename(other)
|
|
path.symlink_to(other)
|
|
self.assert_preflight_failure()
|
|
|
|
def test_source_directory_symlink_rejected(self):
|
|
path = self.roots["sdk"] / "nested"
|
|
other = self.base / "other"
|
|
path.rename(other)
|
|
path.symlink_to(other, target_is_directory=True)
|
|
self.assert_preflight_failure()
|
|
|
|
def test_input_root_symlink_rejected(self):
|
|
root = self.roots["sdk"]
|
|
other = self.base / "other"
|
|
root.rename(other)
|
|
root.symlink_to(other, target_is_directory=True)
|
|
self.assert_preflight_failure()
|
|
|
|
def test_fifo_does_not_block(self):
|
|
path = self.roots["project"] / "LICENSE"
|
|
path.unlink()
|
|
os.mkfifo(path)
|
|
self.assert_preflight_failure()
|
|
|
|
def test_directory_is_not_a_notice(self):
|
|
path = self.roots["project"] / "LICENSE"
|
|
path.unlink()
|
|
path.mkdir()
|
|
self.assert_preflight_failure()
|
|
|
|
def test_catalog_paths_rejected(self):
|
|
for path in ("../secret", "/etc/passwd", "nested/../../secret", "a//b",
|
|
"./LICENSE", "a\\b", "", "a/./b", "a\x00b"):
|
|
with self.subTest(path=path):
|
|
catalog = copy.deepcopy(self.catalog)
|
|
catalog["inputs"][0]["path"] = path
|
|
with self.assertRaises(notices.NoticeError):
|
|
self.run_bundle(catalog=catalog)
|
|
self.assertFalse(self.output.exists())
|
|
|
|
def test_invalid_catalog_entries(self):
|
|
for field, value in (("root", "unknown"), ("size", 0),
|
|
("size", notices.MAX_FILE + 1), ("size", True),
|
|
("sha256", "wrong"), ("range", [-1, 2]),
|
|
("range", [0, 99999]), ("range", [2, 1]),
|
|
("output_sha256", "0" * 64)):
|
|
with self.subTest(field=field, value=value):
|
|
catalog = copy.deepcopy(self.catalog)
|
|
catalog["inputs"][0][field] = value
|
|
with self.assertRaises(notices.NoticeError):
|
|
self.run_bundle(catalog=catalog)
|
|
self.assertFalse(self.output.exists())
|
|
|
|
def test_duplicate_input(self):
|
|
self.catalog["inputs"].append(self.catalog["inputs"][0])
|
|
self.assert_preflight_failure()
|
|
|
|
def test_entry_count_bound(self):
|
|
self.catalog["inputs"] *= notices.MAX_ENTRIES
|
|
self.assert_preflight_failure()
|
|
|
|
def test_total_size_bound(self):
|
|
with mock.patch.object(notices, "MAX_TOTAL", 1):
|
|
self.assert_preflight_failure()
|
|
|
|
def test_unlisted_secrets_builds_and_configs_never_read(self):
|
|
for name in ("sdkconfig", ".env", "device.pem", ".pio/build/firmware.bin",
|
|
"backups/credentials.json"):
|
|
path = self.roots["project"] / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(b"SECRET_DO_NOT_COPY")
|
|
original = notices.read_bounded
|
|
calls = []
|
|
def tracked(fd, path, limit=notices.MAX_FILE):
|
|
calls.append(path)
|
|
return original(fd, path, limit)
|
|
with mock.patch.object(notices, "read_bounded", side_effect=tracked):
|
|
self.run_bundle()
|
|
self.assertEqual(set(calls), {e["path"] for e in self.catalog["inputs"]})
|
|
self.assertFalse(any(b"SECRET_DO_NOT_COPY" in b for b in self.contents(self.output).values()))
|
|
|
|
def test_failed_write_has_no_completion_marker_or_cleanup(self):
|
|
original = notices.write_new
|
|
def fail(fd, path, data):
|
|
if path == "README.txt":
|
|
raise OSError("injected write failure")
|
|
original(fd, path, data)
|
|
with mock.patch.object(notices, "write_new", side_effect=fail):
|
|
with self.assertRaises(OSError):
|
|
self.run_bundle()
|
|
self.assertTrue(self.output.exists())
|
|
self.assertFalse((self.output / "manifest.json").exists())
|
|
# Retrying cannot overwrite or delete even this partial output.
|
|
with self.assertRaises(FileExistsError):
|
|
self.run_bundle()
|
|
|
|
def test_binary_and_non_utf8_output_rejected(self):
|
|
for data in (b"binary\x00notice", b"invalid\xffnotice"):
|
|
with self.subTest(data=data):
|
|
self.catalog["inputs"] = []
|
|
self.add_input("project", "bad", data)
|
|
self.assert_preflight_failure()
|
|
|
|
def test_cli_requires_explicit_output(self):
|
|
result = subprocess.run([sys.executable, str(PROJECT / "tools/release_notices.py"),
|
|
"--sdk-root", str(self.roots["sdk"]),
|
|
"--toolchain-root", str(self.roots["toolchain"])],
|
|
capture_output=True, timeout=10)
|
|
self.assertEqual(result.returncode, 2)
|
|
self.assertIn(b"--output", result.stderr)
|
|
|
|
def test_cli_with_isolated_policy_and_sources(self):
|
|
policy = self.base / "policy"
|
|
(policy / "tools").mkdir(parents=True)
|
|
(policy / "third_party/release-notices").mkdir(parents=True)
|
|
script = policy / "tools/release_notices.py"
|
|
script.write_bytes((PROJECT / "tools/release_notices.py").read_bytes())
|
|
(policy / notices.CATALOG).write_bytes(notices.json_bytes(self.catalog))
|
|
command = [sys.executable, str(script), "--project-root", str(self.roots["project"]),
|
|
"--sdk-root", str(self.roots["sdk"]), "--toolchain-root",
|
|
str(self.roots["toolchain"]), "--output", str(self.output)]
|
|
result = subprocess.run(command, capture_output=True, timeout=10)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
before = self.contents(self.output)
|
|
result = subprocess.run(command, capture_output=True, timeout=10)
|
|
self.assertEqual(result.returncode, 1)
|
|
self.assertEqual(before, self.contents(self.output))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|