ms2k
a0055f67f1
Give the Git Sentry launch shape one Rust owner, update the physical corpus to supply every authority, and make the default repository gate distinguish unavailable physical prerequisites from corpus failures.
502 lines
20 KiB
Python
502 lines
20 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
import importlib.machinery
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
LOADER = importlib.machinery.SourceFileLoader(
|
|
"lazaret_verify", str(ROOT / "tools/lazaret_verify.py")
|
|
)
|
|
SPEC = importlib.util.spec_from_loader(LOADER.name, LOADER)
|
|
VERIFY = importlib.util.module_from_spec(SPEC)
|
|
LOADER.exec_module(VERIFY)
|
|
CORPUS_LOADER = importlib.machinery.SourceFileLoader(
|
|
"ignored_workspace_verifier", str(ROOT / "tools/ignored_workspace_verifier.py")
|
|
)
|
|
CORPUS_SPEC = importlib.util.spec_from_loader(CORPUS_LOADER.name, CORPUS_LOADER)
|
|
CORPUS = importlib.util.module_from_spec(CORPUS_SPEC)
|
|
CORPUS_LOADER.exec_module(CORPUS)
|
|
|
|
|
|
class VerificationTests(unittest.TestCase):
|
|
def test_current_workspace_lints_pass(self) -> None:
|
|
VERIFY.audit_workspace_lints(VERIFY.workspace_metadata())
|
|
|
|
def test_workspace_unsafe_policy_is_required(self) -> None:
|
|
workspace = VERIFY.load_toml(ROOT / "Cargo.toml")
|
|
workspace["workspace"]["lints"]["rust"]["unsafe_code"] = "deny"
|
|
original = VERIFY.load_toml
|
|
VERIFY.load_toml = lambda path: workspace if path == ROOT / "Cargo.toml" else original(path)
|
|
try:
|
|
with self.assertRaisesRegex(VERIFY.VerificationError, "forbid unsafe"):
|
|
VERIFY.audit_workspace_lints(VERIFY.workspace_metadata())
|
|
finally:
|
|
VERIFY.load_toml = original
|
|
|
|
def test_clippy_process_policy_is_exact(self) -> None:
|
|
config = VERIFY.load_toml(ROOT / ".clippy.toml")
|
|
config["disallowed-methods"] = config["disallowed-methods"][:-1]
|
|
original = VERIFY.load_toml
|
|
VERIFY.load_toml = lambda path: config if path == ROOT / ".clippy.toml" else original(path)
|
|
try:
|
|
with self.assertRaisesRegex(VERIFY.VerificationError, "reviewed boundary"):
|
|
VERIFY.audit_workspace_lints(VERIFY.workspace_metadata())
|
|
finally:
|
|
VERIFY.load_toml = original
|
|
|
|
def test_member_lint_inheritance_is_required(self) -> None:
|
|
metadata = VERIFY.workspace_metadata()
|
|
member = next(
|
|
package
|
|
for package in metadata["packages"]
|
|
if package["id"] in set(metadata["workspace_members"])
|
|
)
|
|
original = VERIFY.load_toml
|
|
|
|
def without_inheritance(path: Path) -> dict:
|
|
data = original(path)
|
|
if path == Path(member["manifest_path"]):
|
|
data = copy.deepcopy(data)
|
|
data.pop("lints")
|
|
return data
|
|
|
|
VERIFY.load_toml = without_inheritance
|
|
try:
|
|
with self.assertRaisesRegex(VERIFY.VerificationError, "inherit workspace lints"):
|
|
VERIFY.audit_workspace_lints(metadata)
|
|
finally:
|
|
VERIFY.load_toml = original
|
|
|
|
def test_lint_bypass_pattern_covers_restricted_lints(self) -> None:
|
|
forbidden = (
|
|
"#[allow(disallowed_methods)]",
|
|
"#![expect(clippy::disallowed_types)]",
|
|
"#[allow(warnings)]",
|
|
"#[allow(clippy::all)]",
|
|
"#[allow(clippy::style)]",
|
|
"#[expect(clippy::restriction)]",
|
|
"#![allow(clippy :: all)]",
|
|
"#[expect(clippy :: style)]",
|
|
)
|
|
for source in forbidden:
|
|
with self.subTest(source=source):
|
|
self.assertIsNotNone(VERIFY.LINT_BYPASS.search(source))
|
|
self.assertIsNone(VERIFY.LINT_BYPASS.search("#[allow(dead_code)]"))
|
|
|
|
def test_process_lint_allowances_are_exact_reviewed_functions(self) -> None:
|
|
self.assertEqual(
|
|
VERIFY.PROCESS_LINT_ALLOWANCES["crates/lazaret-linux/tests/capability-drop-helper.rs"],
|
|
{"main"},
|
|
)
|
|
self.assertNotIn("spawn_unreviewed", set().union(*VERIFY.PROCESS_LINT_ALLOWANCES.values()))
|
|
|
|
def test_process_lint_allowance_parser_rejects_false_exemptions(self) -> None:
|
|
with self.assertRaisesRegex(VERIFY.VerificationError, "whole-file"):
|
|
VERIFY.process_lint_allowances(
|
|
"example.rs", "#![allow(clippy::disallowed_methods)]\nfn reviewed() {}"
|
|
)
|
|
with self.assertRaisesRegex(VERIFY.VerificationError, "non-function"):
|
|
VERIFY.process_lint_allowances(
|
|
"example.rs",
|
|
"#[allow(clippy::disallowed_methods)]\nstruct Boundary;\nfn hidden() {}",
|
|
)
|
|
self.assertEqual(
|
|
VERIFY.process_lint_allowances(
|
|
"example.rs", "#[allow(clippy::disallowed_methods)]\nfn arbitrary() {}"
|
|
),
|
|
{"arbitrary"},
|
|
)
|
|
|
|
def test_license_audit_accepts_permitted_choice_only(self) -> None:
|
|
allowed = {"Apache-2.0", "MIT"}
|
|
self.assertTrue(
|
|
VERIFY.license_permitted(
|
|
"Apache-2.0 WITH LLVM-exception OR Apache-2.0 OR MIT", allowed
|
|
)
|
|
)
|
|
self.assertTrue(VERIFY.license_permitted("GPL-3.0-only OR MIT", allowed))
|
|
self.assertTrue(VERIFY.license_permitted("MIT/Apache-2.0", allowed))
|
|
self.assertTrue(VERIFY.license_permitted("Apache-2.0 / MIT", allowed))
|
|
self.assertFalse(VERIFY.license_permitted("GPL-3.0-only", allowed))
|
|
self.assertFalse(VERIFY.license_permitted("MIT AND GPL-3.0-only", allowed))
|
|
|
|
def test_only_the_exact_materialized_fuser_source_is_approved(self) -> None:
|
|
package = {
|
|
"name": "fuser",
|
|
"version": "0.17.0",
|
|
"manifest_path": str(VERIFY.MATERIALIZED_FUSER_MANIFEST),
|
|
}
|
|
self.assertTrue(VERIFY.is_materialized_fuser(package))
|
|
self.assertFalse(VERIFY.is_materialized_fuser({**package, "version": "0.18.0"}))
|
|
self.assertFalse(
|
|
VERIFY.is_materialized_fuser(
|
|
{**package, "manifest_path": str(ROOT / "fuser/Cargo.toml")}
|
|
)
|
|
)
|
|
|
|
def test_workspace_source_discovery_is_not_confined_to_crates(self) -> None:
|
|
with tempfile.TemporaryDirectory(dir=ROOT) as directory:
|
|
package_root = Path(directory)
|
|
source = package_root / "outside.rs"
|
|
source.write_text("fn main() {}")
|
|
metadata = {
|
|
"workspace_members": ["outside"],
|
|
"packages": [
|
|
{
|
|
"id": "outside",
|
|
"manifest_path": str(package_root / "Cargo.toml"),
|
|
}
|
|
],
|
|
"target_directory": str(ROOT / "target"),
|
|
}
|
|
self.assertEqual(VERIFY.workspace_rust_sources(metadata), [source])
|
|
|
|
def test_workspace_source_discovery_rejects_overlapping_target_directory(self) -> None:
|
|
with tempfile.TemporaryDirectory(dir=ROOT) as directory:
|
|
package_root = Path(directory)
|
|
metadata = {
|
|
"workspace_members": ["outside"],
|
|
"packages": [
|
|
{
|
|
"id": "outside",
|
|
"manifest_path": str(package_root / "Cargo.toml"),
|
|
}
|
|
],
|
|
"target_directory": str(ROOT),
|
|
}
|
|
with self.assertRaisesRegex(VERIFY.VerificationError, "must not contain"):
|
|
VERIFY.workspace_rust_sources(metadata)
|
|
metadata["target_directory"] = str(package_root)
|
|
with self.assertRaisesRegex(VERIFY.VerificationError, "must not contain"):
|
|
VERIFY.workspace_rust_sources(metadata)
|
|
|
|
def test_process_trace_requires_exactly_one_exec(self) -> None:
|
|
VERIFY.validate_process_trace("example", ['execve("example", [], []) = 0'])
|
|
with self.assertRaisesRegex(VERIFY.VerificationError, "execve surface count 0"):
|
|
VERIFY.validate_process_trace("example", [])
|
|
with self.assertRaisesRegex(VERIFY.VerificationError, "execve surface count 2"):
|
|
VERIFY.validate_process_trace("example", ["execve(a)", "execve(b)"])
|
|
|
|
def test_wrapper_trace_requires_its_exec_in_place(self) -> None:
|
|
VERIFY.validate_process_trace(
|
|
"lazaret-launch-wrapper",
|
|
["execve(wrapper)", "execve(/bin/true)"],
|
|
)
|
|
with self.assertRaisesRegex(VERIFY.VerificationError, "execve surface count 1"):
|
|
VERIFY.validate_process_trace("lazaret-launch-wrapper", ["execve(wrapper)"])
|
|
|
|
def test_process_trace_rejects_each_forbidden_surface(self) -> None:
|
|
forbidden = (
|
|
"clone(child_stack=NULL) = 1",
|
|
"clone3(NULL, 0) = 1",
|
|
"execveat(3, \"child\", [], [], 0) = 0",
|
|
"fork() = 1",
|
|
"vfork() = 1",
|
|
"wait4(1, NULL, 0, NULL) = 1",
|
|
"waitid(P_PIDFD, 3, {}, WEXITED, NULL) = 0",
|
|
"rt_sigaction(SIGCHLD, {}, NULL, 8) = 0",
|
|
)
|
|
for syscall in forbidden:
|
|
with self.subTest(syscall=syscall):
|
|
with self.assertRaisesRegex(VERIFY.VerificationError, "forbidden process syscall"):
|
|
VERIFY.validate_process_trace("example", ["execve(example)", syscall])
|
|
|
|
def test_process_syscall_audit_requires_strace(self) -> None:
|
|
with mock.patch.object(VERIFY.shutil, "which", return_value=None):
|
|
with self.assertRaisesRegex(VERIFY.VerificationError, "strace must be installed"):
|
|
VERIFY.audit_process_syscalls({})
|
|
|
|
def test_process_syscall_audit_rejects_failed_binary(self) -> None:
|
|
for binary, returncode in VERIFY.PROCESS_TRACE_EXIT_CODES.items():
|
|
VERIFY.validate_trace_exit(binary, returncode)
|
|
with self.assertRaisesRegex(VERIFY.VerificationError, "expected"):
|
|
VERIFY.validate_trace_exit(binary, returncode + 1)
|
|
with self.assertRaisesRegex(VERIFY.VerificationError, "expected None"):
|
|
VERIFY.validate_trace_exit("example", 0)
|
|
|
|
def test_workspace_binary_discovery_covers_every_current_target(self) -> None:
|
|
self.assertEqual(
|
|
set(VERIFY.workspace_binary_targets(VERIFY.workspace_metadata())),
|
|
{
|
|
"lazaret",
|
|
"lazaret-escalation-client",
|
|
"lazaret-host-coordinator",
|
|
"lazaret-git-sentry",
|
|
"lazaret-ignored-workspace",
|
|
"lazaret-launch-wrapper",
|
|
"lazaret-runtime-supervisor",
|
|
"lazaret-workspace-benchmark",
|
|
},
|
|
)
|
|
|
|
def test_workspace_binary_names_must_be_unique(self) -> None:
|
|
metadata = {
|
|
"workspace_members": ["one", "two"],
|
|
"packages": [
|
|
{"id": "one", "targets": [{"name": "same", "kind": ["bin"]}]},
|
|
{"id": "two", "targets": [{"name": "same", "kind": ["bin"]}]},
|
|
],
|
|
}
|
|
with self.assertRaisesRegex(VERIFY.VerificationError, "must be unique"):
|
|
VERIFY.workspace_binary_targets(metadata)
|
|
|
|
def test_binary_artifacts_use_cargo_reported_path(self) -> None:
|
|
metadata = {
|
|
"workspace_members": ["one"],
|
|
"packages": [
|
|
{"id": "one", "targets": [{"name": "example", "kind": ["bin"]}]}
|
|
],
|
|
}
|
|
executable = "/tmp/custom-cargo-target/debug/example"
|
|
message = (
|
|
'{"reason":"compiler-artifact","package_id":"one",'
|
|
f'"target":{{"name":"example","kind":["bin"]}},"executable":"{executable}"}}'
|
|
)
|
|
self.assertEqual(
|
|
VERIFY.binary_artifacts(metadata, message),
|
|
{"example": Path(executable)},
|
|
)
|
|
|
|
def test_tool_versions_are_exactly_pinned(self) -> None:
|
|
expected = {
|
|
"rustc": r"^rustc 1\.97\.0 ",
|
|
"cargo": r"^cargo 1\.97\.0 ",
|
|
"cargo-nextest": r"^cargo-nextest 0\.9\.140 ",
|
|
"cargo-deny": r"^cargo-deny 0\.20\.2$",
|
|
"tlc": r"Version 2\.19 ",
|
|
"uv": r"^uv 0\.11\.28(?: .*)?$",
|
|
}
|
|
self.assertEqual(
|
|
{program: pattern.pattern for program, _, _, pattern in VERIFY.REQUIRED_TOOL_VERSIONS},
|
|
expected,
|
|
)
|
|
|
|
def test_entrypoint_uses_dependency_free_offline_uv(self) -> None:
|
|
entrypoint = (ROOT / "tools/verify").read_text()
|
|
self.assertIn(
|
|
'"$(dirname "$0")/materialize-fuser" --test-dependencies', entrypoint
|
|
)
|
|
self.assertIn(
|
|
"uv run --python '>=3.11' --no-project --offline "
|
|
"--no-python-downloads --no-cache python3",
|
|
entrypoint,
|
|
)
|
|
self.assertIn('lazaret_verify.py" "$@"', entrypoint)
|
|
|
|
def test_fuser_materializer_matches_locked_archive(self) -> None:
|
|
packages = VERIFY.load_toml(ROOT / "Cargo.lock")["package"]
|
|
[fuser] = [package for package in packages if package["name"] == "fuser"]
|
|
materializer = (ROOT / "tools/materialize-fuser").read_text()
|
|
self.assertIn(f"version={fuser['version']}", materializer)
|
|
self.assertIn(f"crate_checksum={fuser['checksum']}", materializer)
|
|
self.assertTrue(
|
|
(ROOT / f"patches/fuser-{fuser['version']}-lock-flags.patch").is_file()
|
|
)
|
|
cargo_config = VERIFY.load_toml(ROOT / ".cargo/config.toml")
|
|
self.assertEqual(
|
|
cargo_config["paths"],
|
|
[f".lazaret/vendor/fuser-{fuser['version']}"],
|
|
)
|
|
|
|
def test_workspace_tests_use_nextest_and_cargo_retains_doctests(self) -> None:
|
|
source = (ROOT / "tools/lazaret_verify.py").read_text()
|
|
expected = [
|
|
[
|
|
"cargo",
|
|
"test",
|
|
"--offline",
|
|
"--locked",
|
|
"--manifest-path",
|
|
".lazaret/vendor/fuser-0.17.0/Cargo.toml",
|
|
"lock_request_preserves_kind_flags",
|
|
],
|
|
[
|
|
"cargo",
|
|
"nextest",
|
|
"run",
|
|
"--offline",
|
|
"--workspace",
|
|
"--all-features",
|
|
"--locked",
|
|
"--filterset",
|
|
(
|
|
"not binary(capability-drop-helper) and not binary(acquisition) "
|
|
"and not binary(recovery-process) and not binary(runtime-composition) "
|
|
"and not binary(network-runtime)"
|
|
),
|
|
],
|
|
[
|
|
"cargo",
|
|
"test",
|
|
"--offline",
|
|
"-p",
|
|
"lazaret-linux",
|
|
"--test",
|
|
"acquisition",
|
|
"--locked",
|
|
],
|
|
[
|
|
"cargo",
|
|
"test",
|
|
"--offline",
|
|
"-p",
|
|
"lazaret-host-coordinator",
|
|
"--test",
|
|
"recovery-process",
|
|
"--locked",
|
|
],
|
|
[
|
|
"cargo",
|
|
"test",
|
|
"--offline",
|
|
"-p",
|
|
"lazaret-linux",
|
|
"--test",
|
|
"capability-drop-helper",
|
|
"--locked",
|
|
],
|
|
[
|
|
"cargo",
|
|
"test",
|
|
"--offline",
|
|
"-p",
|
|
"lazaret-host-coordinator",
|
|
"--test",
|
|
"runtime-composition",
|
|
"--locked",
|
|
],
|
|
[
|
|
"cargo",
|
|
"test",
|
|
"--offline",
|
|
"-p",
|
|
"lazaret-runtime-supervisor",
|
|
"--test",
|
|
"network-runtime",
|
|
"--locked",
|
|
],
|
|
[
|
|
"cargo",
|
|
"test",
|
|
"--offline",
|
|
"--workspace",
|
|
"--all-features",
|
|
"--locked",
|
|
"--doc",
|
|
],
|
|
]
|
|
commands = []
|
|
original_run = VERIFY.run
|
|
original_physical = VERIFY.run_physical
|
|
VERIFY.run = lambda command, cwd=VERIFY.ROOT: commands.append(command)
|
|
VERIFY.run_physical = lambda command, cwd=VERIFY.ROOT: commands.append(command)
|
|
try:
|
|
VERIFY.run_rust_tests()
|
|
finally:
|
|
VERIFY.run = original_run
|
|
VERIFY.run_physical = original_physical
|
|
self.assertEqual(commands, expected)
|
|
self.assertIn("exact cargo-nextest 0.9.140", source)
|
|
self.assertIn("exact uv 0.11.28", source)
|
|
self.assertIn("Python 3.11 or newer", source)
|
|
|
|
def test_physical_test_accepts_only_pass_or_explicit_skip(self) -> None:
|
|
with mock.patch.object(
|
|
VERIFY.subprocess,
|
|
"run",
|
|
return_value=VERIFY.subprocess.CompletedProcess(["physical"], 77),
|
|
):
|
|
VERIFY.run_physical(["physical"])
|
|
with mock.patch.object(
|
|
VERIFY.subprocess,
|
|
"run",
|
|
return_value=VERIFY.subprocess.CompletedProcess(["physical"], 1),
|
|
):
|
|
with self.assertRaisesRegex(VERIFY.VerificationError, "physical test exited 1"):
|
|
VERIFY.run_physical(["physical"])
|
|
|
|
def test_ignored_workspace_corpus_inventory_has_an_executable_owner(self) -> None:
|
|
inventory = json.loads(
|
|
(
|
|
ROOT / "tests/fixtures/ignored-workspace-v1/corpus.json"
|
|
).read_text()
|
|
)["cases"]
|
|
implemented = set(CORPUS.CASE_GROUPS) | {"linked-worktree-index", "git-sentry"}
|
|
self.assertEqual({case["id"] for case in inventory}, implemented)
|
|
|
|
def test_git_sentry_launch_uses_every_production_authority(self) -> None:
|
|
self.assertEqual(
|
|
CORPUS.git_sentry_command(
|
|
"sentry", "workspace", "common", "hooks", "config"
|
|
),
|
|
["sentry", "workspace", "common", "hooks", "config"],
|
|
)
|
|
|
|
def test_ignored_workspace_prerequisite_absence_is_specific(self) -> None:
|
|
with mock.patch.object(CORPUS.sys, "platform", "darwin"):
|
|
self.assertEqual(CORPUS.unavailable_physical_prerequisite(), "Linux is required")
|
|
with (
|
|
mock.patch.object(CORPUS.sys, "platform", "linux"),
|
|
mock.patch.object(CORPUS.shutil, "which", return_value=None),
|
|
):
|
|
self.assertEqual(
|
|
CORPUS.unavailable_physical_prerequisite(), "git is unavailable"
|
|
)
|
|
|
|
def test_default_verifier_runs_the_complete_ignored_workspace_corpus(self) -> None:
|
|
binaries = {
|
|
"lazaret-ignored-workspace": Path("/build/router"),
|
|
"lazaret-git-sentry": Path("/build/sentry"),
|
|
}
|
|
commands = []
|
|
with (
|
|
mock.patch.object(VERIFY, "build_workspace_binaries", return_value=binaries),
|
|
mock.patch.object(
|
|
VERIFY,
|
|
"run_physical",
|
|
side_effect=lambda command, cwd=VERIFY.ROOT: commands.append(command),
|
|
),
|
|
):
|
|
VERIFY.run_ignored_workspace_corpus({})
|
|
[command] = commands
|
|
self.assertEqual(
|
|
command[0:2],
|
|
[VERIFY.sys.executable, str(ROOT / "tools/ignored_workspace_verifier.py")],
|
|
)
|
|
self.assertIn("corpus", command)
|
|
self.assertIn(str(binaries["lazaret-ignored-workspace"]), command)
|
|
self.assertIn(str(binaries["lazaret-git-sentry"]), command)
|
|
self.assertNotIn("--case", command)
|
|
|
|
def test_advisory_refresh_checks_versions_before_mutation(self) -> None:
|
|
events = []
|
|
original_versions = VERIFY.check_tool_versions
|
|
original_refresh = VERIFY.refresh_advisories
|
|
VERIFY.check_tool_versions = lambda: events.append("versions")
|
|
VERIFY.refresh_advisories = lambda: events.append("refresh")
|
|
try:
|
|
with mock.patch.object(VERIFY.sys, "argv", ["tools/verify", "--refresh-advisories"]):
|
|
self.assertEqual(VERIFY.main(), 0)
|
|
finally:
|
|
VERIFY.check_tool_versions = original_versions
|
|
VERIFY.refresh_advisories = original_refresh
|
|
self.assertEqual(events, ["versions", "refresh"])
|
|
|
|
def test_model_paths_are_current_and_complete(self) -> None:
|
|
self.assertEqual(
|
|
{module.name for module, _ in VERIFY.MODEL_CONFIGURATIONS},
|
|
{"Lazaret.tla", "PreparedLower.tla"},
|
|
)
|
|
for module, config in VERIFY.MODEL_CONFIGURATIONS:
|
|
self.assertTrue(module.is_file())
|
|
self.assertTrue(config.is_file())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|