import copy import contextlib import importlib.util import io from pathlib import Path import subprocess import tempfile import unittest from unittest import mock ROOT = Path(__file__).resolve().parents[2] SPEC = importlib.util.spec_from_file_location("lazaret_dev", ROOT / "tools/lazaret_dev.py") DEV = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(DEV) class RegistryTests(unittest.TestCase): def setUp(self): self.packages = DEV.load_toml(ROOT / "support/work-packages.toml") self.refinements = DEV.load_toml(ROOT / "support/model-refinements.toml") self.external = DEV.load_toml(ROOT / "support/external-tools.toml") self.integrations = DEV.load_toml(ROOT / "support/integration-tests.toml") self.process_safety = DEV.load_toml(ROOT / "support/process-safety.toml") def test_committed_registries_are_complete(self): packages = DEV.validate_packages(self.packages) DEV.validate_refinements(self.refinements, {package["id"] for package in packages}) DEV.validate_external_tools(self.external) capability_ids = {item["id"] for kind in ("command", "feature") for item in self.external[kind]} DEV.validate_integrations(self.integrations, capability_ids) DEV.validate_process_safety(self.process_safety, self.integrations) def test_dependency_cycle_is_rejected(self): packages = copy.deepcopy(self.packages) packages["package"][0]["depends_on"] = ["IP-01"] with self.assertRaisesRegex(DEV.RegistryError, "cycle"): DEV.validate_packages(packages) def test_unknown_model_symbol_is_rejected(self): refinements = copy.deepcopy(self.refinements) refinements["coverage"][0]["symbols"].append("UnknownAction") package_ids = {package["id"] for package in DEV.validate_packages(self.packages)} with self.assertRaisesRegex(DEV.RegistryError, "coverage mismatch"): DEV.validate_refinements(refinements, package_ids) def test_unknown_probe_policy_is_rejected(self): external = copy.deepcopy(self.external) external["command"][0]["policy"] = "ignore" with self.assertRaisesRegex(DEV.RegistryError, "fail or skip"): DEV.validate_external_tools(external) def test_command_fields_are_strict(self): mutations = ( lambda command: command.pop("program"), lambda command: command.update(args=[1]), lambda command: command.update(unexpected=True), ) for mutation in mutations: with self.subTest(mutation=mutation): external = copy.deepcopy(self.external) mutation(external["command"][0]) with self.assertRaises(DEV.RegistryError): DEV.validate_external_tools(external) def test_each_probe_rejects_missing_fields(self): for index, feature in enumerate(self.external["feature"]): with self.subTest(probe=feature["probe"]): external = copy.deepcopy(self.external) removable = next((field for field in feature if field not in {"id", "probe", "policy", "levels", "diagnostic"}), "diagnostic") external["feature"][index].pop(removable) with self.assertRaises(DEV.RegistryError): DEV.validate_external_tools(external) def test_each_probe_rejects_malformed_fields(self): invalid = { "linux-x86_64": {"id": 1}, "kernel-at-least": {"minimum": 6.8}, "registered-host-support": {"contract": "unknown"}, "user-namespace": {"program": 1}, "subordinate-ids": {"minimum_count": "65536"}, "open-character-device": {"access": "read-only"}, } for index, feature in enumerate(self.external["feature"]): with self.subTest(probe=feature["probe"]): external = copy.deepcopy(self.external) external["feature"][index].update(invalid[feature["probe"]]) with self.assertRaises(DEV.RegistryError): DEV.validate_external_tools(external) def test_each_probe_rejects_extraneous_fields(self): for index, feature in enumerate(self.external["feature"]): with self.subTest(probe=feature["probe"]): external = copy.deepcopy(self.external) external["feature"][index]["unexpected"] = True with self.assertRaisesRegex(DEV.RegistryError, "unexpected fields"): DEV.validate_external_tools(external) def test_supported_platform_probe_is_mandatory(self): for capability in ("linux-x86_64", "kernel", "pidfs-host-support", "user-namespaces"): with self.subTest(capability=capability): external = copy.deepcopy(self.external) external["feature"] = [feature for feature in external["feature"] if feature["id"] != capability] with self.assertRaisesRegex(DEV.RegistryError, "external manifest missing"): DEV.validate_external_tools(external) def test_pidfs_contract_fields_are_strict(self): mutations = ( lambda external: external["pidfs_host_support"].pop("diagnostic_smoke"), lambda external: external["pidfs_host_support"].update(minimum_kernel=6.9), lambda external: external["pidfs_host_support"].update(unexpected=True), lambda external: external["host_probe"][0].update(steps=["kernel-version-only"]), lambda external: external["host_probe"][0].update(attempts_per_step=True), lambda external: external["host_probe"][0].update(total_timeout_ms=1000), lambda external: external["host_probe"][0].update(step_timeout_ms=4000), lambda external: external["kernel_qualification"][0].pop("source_commit"), lambda external: external["kernel_qualification"][0].update(build_image_sha256="bad"), lambda external: external["kernel_qualification"][0].update(build_image_sha256=1), lambda external: external["kernel_qualification"][0].update(id=[]), lambda external: external["kernel_qualification"][0].update(unexpected=True), ) for mutation in mutations: with self.subTest(mutation=mutation): external = copy.deepcopy(self.external) mutation(external) with self.assertRaises(DEV.RegistryError): DEV.validate_external_tools(external) def test_linux_68_is_refused_and_69_meets_only_the_floor(self): kernel = next(feature for feature in self.external["feature"] if feature["id"] == "kernel") with mock.patch.object(DEV.platform, "system", return_value="Linux"), mock.patch.object( DEV.platform, "release", return_value="6.8.99" ): self.assertEqual(DEV.probe_feature(kernel), (False, "6.8.99")) with mock.patch.object(DEV.platform, "system", return_value="Linux"), mock.patch.object( DEV.platform, "release", return_value="6.9.0" ): self.assertEqual(DEV.probe_feature(kernel), (True, "6.9.0")) with mock.patch.object(DEV.platform, "system", return_value="Linux"), mock.patch.object( DEV.platform, "release", return_value="6.12.38+deb13-amd64" ): self.assertEqual(DEV.probe_feature(kernel), (True, "6.12.38+deb13-amd64")) pidfs = next(feature for feature in self.external["feature"] if feature["id"] == "pidfs-host-support") self.assertEqual(DEV.probe_feature(pidfs), (False, "registered runner pending IP-03")) def test_pidfs_smoke_failure_is_a_hard_runtime_refusal(self): contract = self.external["pidfs_host_support"] self.assertTrue(contract["diagnostic_smoke"].startswith("LZ-HOST-PIDFS-SMOKE:")) self.assertEqual(contract["runtime_failure_policy"], "fail-before-generation") def test_mandatory_capabilities_cannot_be_remapped(self): for capability in ("linux-x86_64", "kernel", "pidfs-host-support", "user-namespaces"): with self.subTest(capability=capability): external = copy.deepcopy(self.external) feature = next(item for item in external["feature"] if item["id"] == capability) feature.update(policy="fail") with self.assertRaisesRegex(DEV.RegistryError, "mandatory feature contract drift"): DEV.validate_external_tools(external) external = copy.deepcopy(self.external) pidfs = next(item for item in external["feature"] if item["id"] == "pidfs-host-support") pidfs.pop("contract") pidfs["probe"] = "linux-x86_64" with self.assertRaisesRegex(DEV.RegistryError, "mandatory feature contract drift"): DEV.validate_external_tools(external) external = copy.deepcopy(self.external) rustc = next(item for item in external["command"] if item["id"] == "rustc") rustc["levels"] = ["V1"] with self.assertRaisesRegex(DEV.RegistryError, "mandatory command contract drift"): DEV.validate_external_tools(external) def test_unqualified_vendor_build_is_rejected(self): external = copy.deepcopy(self.external) vendor = copy.deepcopy(external["kernel_qualification"][0]) vendor.update( id="vendor-linux-6.9.0-x86_64", kind="vendor", vendor="Example Vendor", vendor_patchset_sha256="0" * 64, status="unqualified", ) external["kernel_qualification"].append(vendor) external["pidfs_host_support"]["qualification_ids"].append(vendor["id"]) with self.assertRaisesRegex(DEV.RegistryError, "not qualified"): DEV.validate_external_tools(external) def test_qualified_vendor_release_string_is_representable(self): external = copy.deepcopy(self.external) vendor = copy.deepcopy(external["kernel_qualification"][0]) vendor.update( id="vendor-linux-6.12-x86_64", kind="vendor", release="6.12.38-1-amd64", vendor="Example Vendor", vendor_patchset_sha256="0" * 64, ) external["kernel_qualification"].append(vendor) external["pidfs_host_support"]["qualification_ids"].append(vendor["id"]) DEV.validate_external_tools(external) def test_process_safety_registration_is_strict(self): mutations = ( lambda process: process["policy"].pop("dynamic_scenario"), lambda process: process["policy"].update(production_roots=["crates/lazaret-core"]), lambda process: process["policy"].update(lint_allow_paths=["crates/foreign.rs"]), lambda process: process["policy"].update(trace_expression="read"), lambda process: process["policy"].update(forbidden_syscall_patterns=["$^"]), lambda process: process["policy"].update(unexpected=True), lambda process: process["runtime_dependency"][0].update(features=["z", "a"]), lambda process: process["runtime_dependency"][0].update(checksum=1), lambda process: process["trace_case"][0].update(initial_execve_count=True), ) for mutation in mutations: with self.subTest(mutation=mutation): process = copy.deepcopy(self.process_safety) mutation(process) with self.assertRaises(DEV.RegistryError): DEV.validate_process_safety(process, self.integrations) integrations = copy.deepcopy(self.integrations) integrations["scenarios"] = [] with self.assertRaisesRegex(DEV.RegistryError, "dynamic syscall scenario"): DEV.validate_process_safety(self.process_safety, integrations) def test_process_lint_exemption_is_detected(self): self.assertTrue(DEV.has_process_lint_exemption("#![allow(clippy::disallowed_types)]\n")) self.assertTrue(DEV.has_process_lint_exemption("#[expect(clippy::{disallowed_methods})]\n")) def test_workspace_process_lints_are_mandatory(self): workspace = DEV.load_toml(ROOT / "Cargo.toml") DEV.validate_workspace_process_lints(workspace) for group, lint in (("rust", "unsafe_code"), ("clippy", "disallowed_methods"), ("clippy", "disallowed_types")): with self.subTest(lint=lint): changed = copy.deepcopy(workspace) changed["workspace"]["lints"][group].pop(lint) with self.assertRaises(DEV.RegistryError): DEV.validate_workspace_process_lints(changed) def test_every_workspace_member_inherits_process_lints(self): workspace = DEV.load_toml(ROOT / "Cargo.toml")["workspace"] for member in workspace["members"]: with self.subTest(member=member): manifest = DEV.load_toml(ROOT / member / "Cargo.toml") DEV.validate_member_process_lints(manifest, member) changed = copy.deepcopy(manifest) changed["lints"] = {"workspace": False} with self.assertRaises(DEV.RegistryError): DEV.validate_member_process_lints(changed, member) def test_compiler_resolves_grouped_process_import_alias(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) (root / "src").mkdir() (root / "Cargo.toml").write_text( '[package]\nname = "lint-probe"\nversion = "0.0.0"\nedition = "2024"\n' "[lints.clippy]\ndisallowed_types = \"deny\"\n" ) (root / ".clippy.toml").write_text(DEV.clippy_config_view(self.process_safety)) (root / "src/main.rs").write_text( 'use std::{process::Command as C};\nfn main() { let _ = C::new("true"); }\n' ) result = subprocess.run( ["cargo", "clippy", "--offline", "--", "-D", "warnings"], cwd=root, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=False, ) self.assertNotEqual(result.returncode, 0) self.assertIn("disallowed type", result.stdout) def test_process_syscall_trace_refuses_bypass(self): policy = self.process_safety["policy"] trace_case = self.process_safety["trace_case"][0] DEV.validate_process_trace(['execve("/tmp/lazaret", ["lazaret"], 0x0) = 0'], policy, trace_case) for syscall in ("clone", "execveat", "wait4", "waitid"): with self.subTest(syscall=syscall), self.assertRaisesRegex(DEV.RegistryError, "forbidden"): DEV.validate_process_trace( ['execve("/tmp/lazaret", ["lazaret"], 0x0) = 0', f"{syscall}(0) = 1"], policy, trace_case, ) with self.assertRaisesRegex(DEV.RegistryError, "SIGCHLD"): DEV.validate_process_trace( ['execve("/tmp/lazaret", ["lazaret"], 0x0) = 0', "rt_sigaction(SIGCHLD, {}, NULL, 8) = 0"], policy, trace_case, ) with self.assertRaisesRegex(DEV.RegistryError, "execve surface count"): DEV.validate_process_trace( ['execve("/tmp/lazaret", [], 0x0) = 0', 'execve("/tmp/other", [], 0x0) = 0'], policy, trace_case, ) def test_generated_host_view_covers_authoritative_fields(self): baseline = DEV.host_contract_view(self.external, self.integrations, self.process_safety) external = copy.deepcopy(self.external) external["kernel_qualification"][0]["probe_binary_sha256"] = "0" * 64 self.assertNotEqual(DEV.host_contract_view(external, self.integrations, self.process_safety), baseline) process = copy.deepcopy(self.process_safety) process["runtime_dependency"][0]["features"] = ["changed"] self.assertNotEqual(DEV.host_contract_view(self.external, self.integrations, process), baseline) def test_integration_manifest_rejects_top_level_extras(self): integrations = copy.deepcopy(self.integrations) integrations["unexpected"] = True capability_ids = {item["id"] for kind in ("command", "feature") for item in self.external[kind]} with self.assertRaisesRegex(DEV.RegistryError, "unexpected fields"): DEV.validate_integrations(integrations, capability_ids) def test_missing_skip_capability_skips_scenario(self): integrations = { "scenarios": [{"name": "fuse", "level": "V2", "command": ["false"], "capabilities": ["fuse-device"]}] } with mock.patch.object(DEV, "probe", return_value=(set(), {"fuse-device"})): with contextlib.redirect_stdout(io.StringIO()) as output: DEV.run_integrations("V2", self.external, integrations) self.assertIn("SKIP fuse", output.getvalue()) def test_missing_fail_capability_rejects_scenario(self): integrations = { "scenarios": [{"name": "compiler", "level": "V2", "command": ["false"], "capabilities": ["rustc"]}] } with mock.patch.object(DEV, "probe", return_value=(set(), set())): with self.assertRaisesRegex(DEV.RegistryError, "undeclared capability failure"): DEV.run_integrations("V2", self.external, integrations) def test_subordinate_range_accepts_name_or_numeric_uid(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "subids" path.write_text("developer:not-a-number:65536\n1000:524288:65536\n") self.assertEqual(DEV.subordinate_id_range(path, {"developer", "1000"}, 65536), 524288) def test_subordinate_range_rejects_short_or_malformed_rows(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "subids" path.write_text("developer:524288:65535\ndeveloper:bad:65536\ndeveloper:0:65536\n") self.assertIsNone(DEV.subordinate_id_range(path, {"developer", "1000"}, 65536)) if __name__ == "__main__": unittest.main()