#!/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 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)