#!/usr/bin/env python3 """Repository-owned generation, policy, probe, model, and verification commands.""" from __future__ import annotations import argparse import json import os from pathlib import Path import platform import pwd import re import shutil import signal import stat import subprocess import sys import tempfile import tomllib ROOT = Path(__file__).resolve().parent.parent SUPPORT = ROOT / "support" LEVELS = ("V1", "V2", "V3", "V4", "V5") PIDFS_SMOKE_STEPS = ( "linux-x86_64-kernel-floor", "pidfd-operations", "pidfs-magic", "concurrent-same-process-identity", "last-close-live-reopen-identity", "simultaneous-process-distinct-inode", "stale-handle-cannot-affect-live-peer", "exit-reap-drop-all-later-process-non-reuse", ) PROCESS_DISALLOWED_TYPES = ( "std::process::Child", "std::process::Command", "tokio::process::Child", "tokio::process::Command", ) PROCESS_DISALLOWED_METHODS = ( "libc::clone", "libc::execve", "libc::execveat", "libc::fork", "libc::posix_spawn", "libc::sigaction", "libc::signal", "libc::vfork", "libc::wait", "libc::wait4", "libc::waitid", "libc::waitpid", "nix::sys::signal::sigaction", "nix::sys::wait::wait", "nix::sys::wait::waitid", "nix::sys::wait::waitpid", "nix::unistd::execve", "nix::unistd::execveat", "nix::unistd::fork", "nix::unistd::vfork", "rustix::process::kill_process", "rustix::process::kill_process_group", "rustix::process::pidfd_open", "rustix::process::pidfd_send_signal", "rustix::process::waitid", "rustix::process::waitpid", "std::process::Child::kill", "std::process::Child::try_wait", "std::process::Child::wait", "std::process::Child::wait_with_output", "std::process::Command::output", "std::process::Command::spawn", "std::process::Command::status", ) PROCESS_FORBIDDEN_SYSCALL_PATTERNS = ( r"^clone3?\(", r"^execveat\(", r"^fork\(", r"^vfork\(", r"^wait4\(", r"^waitid\(", r"^rt_sigaction\(SIGCHLD,", ) PROCESS_REVIEW_DEPENDENCIES = frozenset({"libc", "nix", "rustix", "tokio"}) class RegistryError(RuntimeError): pass def load_toml(path: Path) -> dict: with path.open("rb") as source: return tomllib.load(source) def require(condition: bool, message: str) -> None: if not condition: raise RegistryError(message) def validate_fields(item: object, required: set[str], optional: set[str], owner: str) -> dict: require(isinstance(item, dict), f"{owner}: entry must be a table") fields = set(item) missing = required - fields extra = fields - required - optional require(not missing, f"{owner}: missing fields {sorted(missing)}") require(not extra, f"{owner}: unexpected fields {sorted(extra)}") return item def validate_levels(levels: object, owner: str) -> None: require(isinstance(levels, list) and levels, f"{owner}: levels must be nonempty") require(all(isinstance(level, str) for level in levels), f"{owner}: levels must be strings") require(len(levels) == len(set(levels)), f"{owner}: duplicate levels") require(all(level in LEVELS for level in levels), f"{owner}: unknown level") def validate_packages(data: dict) -> list[dict]: require(data.get("schema") == 1, "work-package registry schema must be 1") packages = data.get("package") require(isinstance(packages, list) and packages, "work-package registry is empty") by_id = {} for package in packages: package_id = package.get("id") require(isinstance(package_id, str) and re.fullmatch(r"IP-\d{2}", package_id) is not None, "invalid package ID") require(package_id not in by_id, f"duplicate package {package_id}") require(isinstance(package.get("title"), str) and package["title"], f"{package_id}: title is empty") require(isinstance(package.get("depends_on"), list), f"{package_id}: depends_on must be a list") validate_levels(package.get("levels"), package_id) by_id[package_id] = package for package_id, package in by_id.items(): dependencies = package["depends_on"] require(len(dependencies) == len(set(dependencies)), f"{package_id}: duplicate dependency") for dependency in dependencies: require(dependency in by_id, f"{package_id}: unknown dependency {dependency}") require(dependency != package_id, f"{package_id}: self dependency") topological_packages(packages) return packages def topological_packages(packages: list[dict]) -> list[dict]: by_id = {package["id"]: package for package in packages} indegree = {package_id: 0 for package_id in by_id} consumers = {package_id: [] for package_id in by_id} for package_id, package in by_id.items(): for dependency in package["depends_on"]: indegree[package_id] += 1 consumers[dependency].append(package_id) ready = sorted(package_id for package_id, degree in indegree.items() if degree == 0) ordered = [] while ready: package_id = ready.pop(0) ordered.append(by_id[package_id]) for consumer in sorted(consumers[package_id]): indegree[consumer] -= 1 if indegree[consumer] == 0: ready.append(consumer) ready.sort() require(len(ordered) == len(packages), "work-package dependency cycle") return ordered def operator_blocks(source: str) -> dict[str, str]: definitions = list(re.finditer(r"(?m)^([A-Za-z][A-Za-z0-9_]*)(?:\([^\n=]*\))? ==", source)) blocks = {} for index, definition in enumerate(definitions): end = definitions[index + 1].start() if index + 1 < len(definitions) else len(source) blocks[definition.group(1)] = source[definition.end():end] return blocks def action_symbols(source: str, aggregates: list[str]) -> set[str]: blocks = operator_blocks(source) aggregate_set = set(aggregates) require("Next" in aggregate_set, "action aggregates must include Next") found = set() pending = ["Next"] visited = set() while pending: aggregate = pending.pop() require(aggregate in blocks, f"missing aggregate operator {aggregate}") if aggregate in visited: continue visited.add(aggregate) body = blocks[aggregate] calls = set(re.findall(r"\b([A-Z][A-Za-z0-9_]*)\s*\(", body)) calls.update(re.findall(r"(?m)^\s*\\/\s+([A-Z][A-Za-z0-9_]*)\s*$", body)) for symbol in calls: if symbol in aggregate_set: pending.append(symbol) elif symbol in blocks: found.add(symbol) require(visited == aggregate_set, f"unused action aggregates: {sorted(aggregate_set - visited)}") return found def configured_symbols(source: str, section: str) -> set[str]: match = re.search(rf"(?ms)^{section}\s*$\n(.*?)(?=^[A-Z]+\s*$|\Z)", source) if match is None: return set() return set(re.findall(r"(?m)^\s+([A-Za-z][A-Za-z0-9_]*)\s*$", match.group(1))) def validate_refinements(data: dict, package_ids: set[str]) -> None: require(data.get("schema") == 1, "model-refinement registry schema must be 1") models = data.get("model") coverage = data.get("coverage") require(isinstance(models, list) and models, "model registry is empty") require(isinstance(coverage, list) and coverage, "model coverage is empty") by_name = {} expected = {} for model in models: name = model.get("name") require(isinstance(name, str) and name not in by_name, "duplicate or invalid model name") module = ROOT / model.get("module", "") config = ROOT / model.get("config", "") require(module.is_file(), f"{name}: missing model module") require(config.is_file(), f"{name}: missing model configuration") aggregates = model.get("action_aggregates") require(isinstance(aggregates, list) and aggregates, f"{name}: action aggregates are empty") module_source = module.read_text() config_source = config.read_text() expected[(name, "action")] = action_symbols(module_source, aggregates) expected[(name, "invariant")] = configured_symbols(config_source, "INVARIANTS") expected[(name, "property")] = configured_symbols(config_source, "PROPERTIES") by_name[name] = model actual = {key: set() for key in expected} for group in coverage: model = group.get("model") kind = group.get("kind") key = (model, kind) require(key in actual, f"unknown model coverage kind {model}:{kind}") symbols = group.get("symbols") owners = group.get("owners") require(isinstance(symbols, list) and symbols, f"{model}:{kind}: symbols are empty") require(isinstance(owners, list) and owners, f"{model}:{kind}: owners are empty") require(all(owner in package_ids for owner in owners), f"{model}:{kind}: unknown owner") validate_levels(group.get("levels"), f"{model}:{kind}") require(isinstance(group.get("verifier"), str) and group["verifier"], f"{model}:{kind}: verifier is empty") for symbol in symbols: require(symbol not in actual[key], f"duplicate refinement {model}!{symbol}") actual[key].add(symbol) for key, symbols in expected.items(): missing = symbols - actual[key] extra = actual[key] - symbols require(not missing and not extra, f"{key[0]} {key[1]} coverage mismatch; missing={sorted(missing)}, extra={sorted(extra)}") def validate_external_tools(data: dict) -> None: validate_fields( data, { "schema", "levels", "pidfs_host_support", "host_probe", "kernel_qualification", "command", "feature", }, set(), "external manifest", ) require(data.get("schema") == 2, "external-tool manifest schema must be 2") require(data.get("levels") == list(LEVELS), "external-tool levels must be V1 through V5") validate_pidfs_host_support(data) commands = data.get("command") features = data.get("feature") require(isinstance(commands, list) and commands, "external command manifest is empty") require(isinstance(features, list) and features, "external feature manifest is empty") ids = set() for item in commands + features: require(isinstance(item, dict), "external capability entry must be a table") item_id = item.get("id") require(isinstance(item_id, str) and item_id not in ids, "duplicate or invalid external capability ID") ids.add(item_id) item_policy = item.get("policy") require(isinstance(item_policy, str) and item_policy in {"fail", "skip"}, f"{item_id}: policy must be fail or skip") validate_levels(item.get("levels"), item_id) require(isinstance(item.get("diagnostic"), str) and item["diagnostic"], f"{item_id}: diagnostic is empty") required = { "rustc", "cargo", "cargo-deny", "tlc", "runsc", "fuse-overlayfs", "slirp4netns", "strace", "linux-x86_64", "kernel", "pidfs-host-support", "user-namespaces", "subordinate-ids", "fuse-device", } require(required <= ids, f"external manifest missing {sorted(required - ids)}") for command in commands: command_id = command["id"] validate_fields( command, {"id", "program", "args", "version_pattern", "policy", "levels", "diagnostic"}, {"accepted_exit_codes"}, command_id, ) require(isinstance(command.get("program"), str) and command["program"], f"{command['id']}: program is empty") require(isinstance(command.get("args"), list) and all(isinstance(arg, str) for arg in command["args"]), f"{command['id']}: args must be strings") exit_codes = command.get("accepted_exit_codes", [0]) require(isinstance(exit_codes, list) and exit_codes and all(isinstance(code, int) and not isinstance(code, bool) for code in exit_codes), f"{command['id']}: invalid accepted exit codes") pattern = command.get("version_pattern") require(isinstance(pattern, str), f"{command['id']}: version pattern is missing") try: re.compile(pattern) except re.error as error: raise RegistryError(f"{command['id']}: invalid version pattern: {error}") from error common = {"id", "probe", "policy", "levels", "diagnostic"} probe_fields = { "linux-x86_64": set(), "kernel-at-least": {"minimum"}, "registered-host-support": {"contract"}, "user-namespace": {"program", "sysctl_path"}, "subordinate-ids": {"minimum_count", "uidmap_program", "gidmap_program", "unshare_program"}, "open-character-device": {"path", "access"}, } for feature in features: feature_id = feature["id"] probe_name = feature.get("probe") require(isinstance(probe_name, str) and probe_name in probe_fields, f"{feature_id}: unknown feature probe") validate_fields(feature, common | probe_fields[probe_name], set(), feature_id) if probe_name == "kernel-at-least": minimum = feature["minimum"] require(isinstance(minimum, str) and re.fullmatch(r"\d+(?:\.\d+)+", minimum) is not None, f"{feature_id}: minimum must be a dotted numeric version") if feature_id == "kernel": require(minimum == data["pidfs_host_support"]["minimum_kernel"], "kernel: minimum must match pidfs host contract") elif probe_name == "registered-host-support": require(feature["contract"] == data["pidfs_host_support"]["id"], f"{feature_id}: unknown host contract") elif probe_name == "user-namespace": require(isinstance(feature["program"], str) and feature["program"], f"{feature_id}: program is empty") require(isinstance(feature["sysctl_path"], str) and Path(feature["sysctl_path"]).is_absolute(), f"{feature_id}: sysctl_path must be absolute") elif probe_name == "subordinate-ids": count = feature["minimum_count"] require(isinstance(count, int) and not isinstance(count, bool) and count >= 65536, f"{feature_id}: minimum_count must be at least 65536") for field in ("uidmap_program", "gidmap_program", "unshare_program"): require(isinstance(feature[field], str) and feature[field], f"{feature_id}: {field} is empty") elif probe_name == "open-character-device": require(isinstance(feature["path"], str) and Path(feature["path"]).is_absolute(), f"{feature_id}: path must be absolute") require(feature["access"] == "read-write", f"{feature_id}: access must be read-write") command_contracts = { "rustc": ("rustc", "fail", list(LEVELS)), "cargo": ("cargo", "fail", list(LEVELS)), "cargo-deny": ("cargo-deny", "fail", list(LEVELS)), "tlc": ("tlc", "fail", list(LEVELS)), "runsc": ("runsc", "skip", ["V3", "V4", "V5"]), "fuse-overlayfs": ("fuse-overlayfs", "skip", ["V2", "V3", "V4", "V5"]), "slirp4netns": ("slirp4netns", "skip", ["V2", "V3", "V4", "V5"]), "strace": ("strace", "skip", ["V2", "V3", "V4", "V5"]), } commands_by_id = {command["id"]: command for command in commands} for command_id, (program, policy, levels) in command_contracts.items(): require(command_id in commands_by_id, f"{command_id}: mandatory command entry is missing") command = commands_by_id[command_id] require((command["program"], command["policy"], command["levels"]) == (program, policy, levels), f"{command_id}: mandatory command contract drift") feature_contracts = { "linux-x86_64": ("linux-x86_64", "skip", ["V2", "V3", "V4", "V5"]), "kernel": ("kernel-at-least", "skip", ["V2", "V3", "V4", "V5"]), "pidfs-host-support": ("registered-host-support", "skip", ["V2", "V3", "V4", "V5"]), "user-namespaces": ("user-namespace", "skip", ["V2", "V3", "V4", "V5"]), "subordinate-ids": ("subordinate-ids", "skip", ["V2", "V3", "V4", "V5"]), "fuse-device": ("open-character-device", "skip", ["V2", "V3", "V4", "V5"]), } features_by_id = {feature["id"]: feature for feature in features} for feature_id, (probe_name, policy, levels) in feature_contracts.items(): require(feature_id in features_by_id, f"{feature_id}: mandatory feature entry is missing") feature = features_by_id[feature_id] require((feature["probe"], feature["policy"], feature["levels"]) == (probe_name, policy, levels), f"{feature_id}: mandatory feature contract drift") def validate_pidfs_host_support(data: dict) -> None: contract = validate_fields( data.get("pidfs_host_support"), { "id", "platform", "minimum_kernel", "probe", "qualification_ids", "runtime_failure_policy", "verifier_unavailable_policy", "bounded_claim", "diagnostic_platform", "diagnostic_kernel", "diagnostic_qualification", "diagnostic_smoke", }, set(), "pidfs host support", ) require(contract["id"] == "pidfs-process-identity-v1", "pidfs host support: invalid contract ID") require(contract["platform"] == "linux-x86_64", "pidfs host support: platform must be linux-x86_64") require(contract["minimum_kernel"] == "6.9", "pidfs host support: minimum kernel must be 6.9") require(contract["runtime_failure_policy"] == "fail-before-generation", "pidfs host support: invalid runtime failure policy") require(contract["verifier_unavailable_policy"] == "skip", "pidfs host support: invalid verifier policy") require(contract["bounded_claim"] == "current-boot-smoke-only", "pidfs host support: invalid smoke claim") for diagnostic in ( "diagnostic_platform", "diagnostic_kernel", "diagnostic_qualification", "diagnostic_smoke", ): require(isinstance(contract[diagnostic], str) and contract[diagnostic].startswith("LZ-HOST-PIDFS-"), f"pidfs host support: invalid {diagnostic}") probes = data.get("host_probe") require(isinstance(probes, list) and len(probes) == 1, "pidfs host support: exactly one host probe is required") probe = validate_fields( probes[0], { "id", "owner", "cache_key", "result_claim", "attempts_per_step", "step_timeout_ms", "total_timeout_ms", "cleanup_timeout_ms", "steps", }, set(), "pidfs host probe", ) require(probe["id"] == contract["probe"], "pidfs host support: probe ID mismatch") require(probe["owner"] == "IP-03", "pidfs host probe: owner must be IP-03") require(probe["cache_key"] == "boot_id", "pidfs host probe: cache key must be boot_id") require(probe["result_claim"] == "bounded-current-boot-observations", "pidfs host probe: result claim must remain bounded") require(isinstance(probe["attempts_per_step"], int) and not isinstance(probe["attempts_per_step"], bool) and probe["attempts_per_step"] == 1, "pidfs host probe: each step must run exactly once") expected_timeouts = {"step_timeout_ms": 5000, "total_timeout_ms": 60000, "cleanup_timeout_ms": 5000} for field, expected in expected_timeouts.items(): value = probe[field] require(isinstance(value, int) and not isinstance(value, bool) and value == expected, f"pidfs host probe: {field} must be {expected}") require(probe["steps"] == list(PIDFS_SMOKE_STEPS), "pidfs host probe: smoke steps do not match the selected contract") qualifications = data.get("kernel_qualification") require(isinstance(qualifications, list) and qualifications, "kernel qualification registry is empty") by_id = {} hashes = { "source_archive_sha256", "build_config_sha256", "build_image_sha256", "probe_binary_sha256", "reuse_result_sha256", "acquisition_result_sha256", } common = { "id", "kind", "architecture", "release", "source_ref", "source_commit", "source_archive_sha256", "source_signer_fingerprint", "build_config_sha256", "build_image_sha256", "probe_binary_sha256", "reuse_result_sha256", "acquisition_result_sha256", "pidfs_inode_semantics", "status", } for qualification in qualifications: require(isinstance(qualification, dict), "kernel qualification entry must be a table") kind = qualification.get("kind") optional = {"vendor", "vendor_patchset_sha256"} if kind == "vendor" else set() qualification = validate_fields(qualification, common, optional, f"kernel qualification {qualification.get('id')}") qualification_id = qualification["id"] require(isinstance(qualification_id, str) and qualification_id and qualification_id not in by_id, "duplicate or invalid kernel qualification ID") require(isinstance(kind, str) and kind in {"upstream", "vendor"}, f"{qualification_id}: invalid qualification kind") require(qualification["architecture"] == "x86_64", f"{qualification_id}: architecture must be x86_64") release = qualification["release"] require(isinstance(release, str) and release, f"{qualification_id}: invalid release") release_version = kernel_version_tuple(release) require(release_version is not None, f"{qualification_id}: release lacks a numeric kernel prefix") require(release_version >= version_tuple(contract["minimum_kernel"]), f"{qualification_id}: release is below the pidfs floor") require(isinstance(qualification["source_ref"], str) and qualification["source_ref"], f"{qualification_id}: source ref is empty") source_commit = qualification["source_commit"] signer_fingerprint = qualification["source_signer_fingerprint"] require(isinstance(source_commit, str) and re.fullmatch(r"[0-9a-f]{40}", source_commit) is not None, f"{qualification_id}: invalid source commit") require(isinstance(signer_fingerprint, str) and re.fullmatch(r"[0-9A-F]{40}", signer_fingerprint) is not None, f"{qualification_id}: invalid signer fingerprint") for field in hashes: value = qualification[field] require(isinstance(value, str) and re.fullmatch(r"[0-9a-f]{64}", value) is not None, f"{qualification_id}: invalid {field}") require(qualification["pidfs_inode_semantics"] == "boot-monotonic-u64", f"{qualification_id}: invalid pidfs inode semantics") require(qualification["status"] == "qualified", f"{qualification_id}: build is not qualified") if kind == "vendor": require(isinstance(qualification.get("vendor"), str) and qualification["vendor"], f"{qualification_id}: vendor is required") patchset = qualification.get("vendor_patchset_sha256") require(isinstance(patchset, str) and re.fullmatch(r"[0-9a-f]{64}", patchset) is not None, f"{qualification_id}: invalid vendor patchset hash") by_id[qualification_id] = qualification qualification_ids = contract["qualification_ids"] require(isinstance(qualification_ids, list) and qualification_ids and all(isinstance(item, str) and item for item in qualification_ids), "pidfs host support: qualification IDs are empty or invalid") require(len(qualification_ids) == len(set(qualification_ids)), "pidfs host support: duplicate qualification IDs") require(set(qualification_ids) <= set(by_id), "pidfs host support: unknown qualification ID") def validate_integrations(data: dict, capability_ids: set[str]) -> None: validate_fields(data, {"schema", "levels", "scenarios"}, set(), "integration manifest") require(data.get("schema") == 1, "integration manifest schema must be 1") require(data.get("levels") == list(LEVELS[1:]), "integration levels must be V2 through V5") scenarios = data.get("scenarios") require(isinstance(scenarios, list), "integration scenarios must be a list") names = set() for scenario in scenarios: validate_fields(scenario, {"name", "level", "command", "capabilities"}, set(), "integration scenario") name = scenario.get("name") require(isinstance(name, str) and name and name not in names, "duplicate or invalid integration scenario") names.add(name) require(scenario.get("level") in LEVELS[1:], f"{name}: invalid integration level") command = scenario.get("command") require(isinstance(command, list) and command and all(isinstance(part, str) for part in command), f"{name}: invalid command") capabilities = scenario.get("capabilities", []) require(isinstance(capabilities, list) and all(isinstance(capability, str) for capability in capabilities), f"{name}: capabilities must be strings") require(all(capability in capability_ids for capability in capabilities), f"{name}: unknown capability") def validate_process_safety(data: dict, integrations: dict) -> None: validate_fields(data, {"schema", "policy", "trace_case", "runtime_dependency"}, set(), "process-safety manifest") require(data.get("schema") == 2, "process-safety manifest schema must be 2") policy = validate_fields( data.get("policy"), { "id", "broker_crate", "broker_state", "production_roots", "lint_config", "lint_allow_paths", "dynamic_scenario", "dynamic_coverage", "completion_owner", "trace_expression", "forbidden_syscall_patterns", "disallowed_types", "disallowed_methods", }, set(), "process-safety policy", ) require(policy["id"] == "causal-child-acquisition-v1", "process-safety policy: invalid ID") require(policy["broker_crate"] == "lazaret-process-broker", "process-safety policy: invalid broker crate") require(policy["broker_state"] == "unimplemented", "process-safety policy: IP-03 has not implemented the broker") require(policy["lint_config"] == ".clippy.toml", "process-safety policy: invalid lint config") require(policy["dynamic_coverage"] == "registered-exercised-paths-only", "process-safety policy: dynamic coverage is overstated") require(policy["dynamic_scenario"] == "process-safety-baseline-exercised-paths", "process-safety policy: dynamic scenario changed") require(policy["completion_owner"] == "IP-03", "process-safety policy: completion owner must be IP-03") for field in ("production_roots", "lint_allow_paths"): values = policy[field] require(isinstance(values, list) and all(isinstance(value, str) and value for value in values), f"process-safety policy: invalid {field}") require(len(values) == len(set(values)), f"process-safety policy: duplicate {field}") require(policy["production_roots"] == ["crates"], "process-safety policy: production roots must cover the complete crates tree") require(not policy["lint_allow_paths"], "process-safety policy: no lint exemption is allowed before IP-03 implements the broker crate") require(policy["trace_expression"] == "process,wait4,waitid,rt_sigaction", "process-safety policy: trace expression must cover process, wait, and signal-disposition syscalls") require(policy["forbidden_syscall_patterns"] == list(PROCESS_FORBIDDEN_SYSCALL_PATTERNS), "process-safety policy: forbidden syscall coverage drift") require(policy["disallowed_types"] == list(PROCESS_DISALLOWED_TYPES), "process-safety policy: disallowed type coverage drift") require(policy["disallowed_methods"] == list(PROCESS_DISALLOWED_METHODS), "process-safety policy: disallowed method coverage drift") for pattern in policy["forbidden_syscall_patterns"]: try: re.compile(pattern) except re.error as error: raise RegistryError(f"process-safety policy: invalid pattern {pattern}: {error}") from error trace_cases = data.get("trace_case") require(isinstance(trace_cases, list) and trace_cases, "process-safety trace cases are empty") binaries = set() for trace_case in trace_cases: trace_case = validate_fields(trace_case, {"binary", "args", "expected_exit_code", "initial_execve_count"}, set(), "process-safety trace case") binary = trace_case["binary"] require(isinstance(binary, str) and binary and binary not in binaries, "duplicate or invalid process-safety trace binary") require(isinstance(trace_case["args"], list) and all(isinstance(arg, str) for arg in trace_case["args"]), f"{binary}: invalid trace arguments") require(isinstance(trace_case["expected_exit_code"], int) and not isinstance(trace_case["expected_exit_code"], bool), f"{binary}: invalid expected exit code") initial_execve_count = trace_case["initial_execve_count"] require(isinstance(initial_execve_count, int) and not isinstance(initial_execve_count, bool) and initial_execve_count == 1, f"{binary}: exactly one tracer-launched execve is required") binaries.add(binary) dependencies = data.get("runtime_dependency") require(isinstance(dependencies, list) and dependencies, "process-safety dependency registry is empty") identities = set() for dependency in dependencies: dependency = validate_fields(dependency, {"name", "version", "checksum", "features", "reviewed_process_apis"}, set(), "process-safety dependency") identity = (dependency["name"], dependency["version"]) require(all(isinstance(item, str) and item for item in identity), "process-safety dependency: invalid identity") require(identity not in identities, f"process-safety dependency: duplicate {identity[0]} {identity[1]}") checksum = dependency["checksum"] require(isinstance(checksum, str) and re.fullmatch(r"[0-9a-f]{64}", checksum) is not None, f"{identity[0]} {identity[1]}: invalid source checksum") features = dependency["features"] require(isinstance(features, list) and all(isinstance(feature, str) and feature for feature in features), f"{identity[0]} {identity[1]}: invalid features") require(features == sorted(set(features)), f"{identity[0]} {identity[1]}: features must be unique and sorted") process_apis = dependency["reviewed_process_apis"] require(isinstance(process_apis, list) and all(isinstance(api, str) and api for api in process_apis), f"{identity[0]} {identity[1]}: invalid reviewed process APIs") require(process_apis == sorted(set(process_apis)), f"{identity[0]} {identity[1]}: reviewed process APIs must be unique and sorted") identities.add(identity) scenarios = {scenario["name"]: scenario for scenario in integrations["scenarios"]} scenario_name = policy["dynamic_scenario"] require(scenario_name in scenarios, "process-safety policy: dynamic syscall scenario is not registered") scenario = scenarios[scenario_name] require(scenario["level"] == "V2", "process-safety policy: dynamic syscall scenario must be V2") require(scenario["command"] == ["tools/dev", "process-audit", "--dynamic"], "process-safety policy: dynamic syscall command mismatch") require(set(scenario["capabilities"]) == {"linux-x86_64", "strace"}, "process-safety policy: dynamic syscall capabilities mismatch") def validated_data() -> tuple[list[dict], dict, dict, dict]: packages = validate_packages(load_toml(SUPPORT / "work-packages.toml")) package_ids = {package["id"] for package in packages} refinements = load_toml(SUPPORT / "model-refinements.toml") validate_refinements(refinements, package_ids) external = load_toml(SUPPORT / "external-tools.toml") validate_external_tools(external) capability_ids = {item["id"] for kind in ("command", "feature") for item in external[kind]} integrations = load_toml(SUPPORT / "integration-tests.toml") validate_integrations(integrations, capability_ids) process_safety = load_toml(SUPPORT / "process-safety.toml") validate_process_safety(process_safety, integrations) return packages, external, integrations, process_safety def dependency_view(packages: list[dict]) -> str: ordered = topological_packages(packages) lines = [ "# Implementation dependencies", "", "Generated from `support/work-packages.toml`; do not edit this file directly.", "", "| Package | Deliverable | Depends on | Verification |", "| --- | --- | --- | --- |", ] for package in ordered: dependencies = ", ".join(f"`{item}`" for item in package["depends_on"]) or "—" levels = ", ".join(package["levels"]) lines.append(f"| `{package['id']}` | {package['title']} | {dependencies} | {levels} |") lines.extend(["", "```mermaid", "graph TD"]) for package in ordered: label = package["title"].replace('"', "'") lines.append(f" {package['id'].replace('-', '_')}[\"{package['id']} {label}\"]") for package in ordered: for dependency in package["depends_on"]: lines.append(f" {dependency.replace('-', '_')} --> {package['id'].replace('-', '_')}") lines.extend(["```", ""]) return "\n".join(lines) def host_contract_view(external: dict, integrations: dict, process_safety: dict) -> str: contract = external["pidfs_host_support"] probe = external["host_probe"][0] lines = [ "# Host support and process-safety registry", "", "Generated from `support/external-tools.toml` and `support/process-safety.toml`; do not edit this file directly.", "", "## Pidfs host support", "", f"- Contract: `{contract['id']}`", f"- Platform floor: `{contract['platform']}` kernel `{contract['minimum_kernel']}` or newer", f"- Runtime refusal: `{contract['runtime_failure_policy']}`", f"- Unavailable verifier behavior: `{contract['verifier_unavailable_policy']}`", f"- Smoke claim: `{contract['bounded_claim']}` cached only by `{probe['cache_key']}`", f"- Smoke bounds: one attempt per step, `{probe['step_timeout_ms']}` ms per step, `{probe['total_timeout_ms']}` ms total, `{probe['cleanup_timeout_ms']}` ms cleanup", "- Positive support: unavailable until IP-03 implements an exact running-build matcher and the smoke runner", "", "The smoke steps are bounded observations and never qualify source or build provenance:", "", ] lines.extend(f"1. `{step}`" for step in probe["steps"]) lines.extend(["", "## Qualified kernel builds", "", "| ID | Kind | Release | Source commit | Build image SHA-256 |", "| --- | --- | --- | --- | --- |"]) by_id = {item["id"]: item for item in external["kernel_qualification"]} for qualification_id in contract["qualification_ids"]: item = by_id[qualification_id] lines.append(f"| `{qualification_id}` | {item['kind']} | `{item['release']}` | `{item['source_commit']}` | `{item['build_image_sha256']}` |") policy = process_safety["policy"] lines.extend( [ "", "## Process-safety audit", "", f"Policy `{policy['id']}` pins {len(process_safety['runtime_dependency'])} dependency source/version/feature/API-review records, generates compiler-resolved disallowed type/method lints, and registers `{policy['dynamic_scenario']}` as V2 exercised-path tracing.", "", f"The dedicated `{policy['broker_crate']}` boundary is `{policy['broker_state']}` and no lint exemption is authorized. Dynamic coverage is `{policy['dynamic_coverage']}`; it does not claim unexercised-path coverage.", "", "## Complete registered data", "", "Every authoritative host-qualification and process-safety field is reproduced below so generated-view drift remains complete.", "", "```json", ] ) host_capability_ids = {"linux-x86_64", "kernel", "pidfs-host-support"} registered_data = { "pidfs_host_support": contract, "host_probe": external["host_probe"], "kernel_qualification": external["kernel_qualification"], "host_capabilities": [feature for feature in external["feature"] if feature["id"] in host_capability_ids], "process_trace_capability": next(command for command in external["command"] if command["id"] == "strace"), "process_trace_scenario": next(scenario for scenario in integrations["scenarios"] if scenario["name"] == policy["dynamic_scenario"]), "process_safety": process_safety, "workspace_lints": load_toml(ROOT / "Cargo.toml")["workspace"]["lints"], "workspace_member_lints": workspace_member_lints(), } lines.extend([json.dumps(registered_data, indent=2, sort_keys=True), "```", ""]) return "\n".join(lines) def clippy_config_view(process_safety: dict) -> str: policy = process_safety["policy"] reason = "process control is confined to the reviewed Lazaret spawn/reap broker" lines = ["# Generated from support/process-safety.toml; do not edit.", "", "disallowed-types = ["] lines.extend(f" {{ path = {json.dumps(path)}, reason = {json.dumps(reason)} }}," for path in policy["disallowed_types"]) lines.extend(["]", "", "disallowed-methods = ["]) lines.extend(f" {{ path = {json.dumps(path)}, reason = {json.dumps(reason)} }}," for path in policy["disallowed_methods"]) lines.extend(["]", ""]) return "\n".join(lines) def generate(check: bool) -> None: packages, external, integrations, process_safety = validated_data() outputs = { ROOT / "docs/implementation-dependencies.md": dependency_view(packages), ROOT / "docs/host-support.md": host_contract_view(external, integrations, process_safety), ROOT / ".clippy.toml": clippy_config_view(process_safety), } for target, content in outputs.items(): if check: require(target.is_file(), f"generated file missing: {target.relative_to(ROOT)}") require(target.read_text() == content, f"generated file drift: {target.relative_to(ROOT)}") else: target.write_text(content) def version_tuple(value: str) -> tuple[int, ...]: return tuple(int(part) for part in value.split(".")) def kernel_version_tuple(release: str) -> tuple[int, ...] | None: match = re.match(r"\d+(?:\.\d+)+", release) return version_tuple(match.group()) if match is not None else None def subordinate_id_range(path: Path, principals: set[str], minimum_count: int) -> int | None: try: rows = path.read_text().splitlines() except OSError: return None for row in rows: fields = row.split(":") if len(fields) != 3 or fields[0] not in principals: continue try: start = int(fields[1]) count = int(fields[2]) except ValueError: continue if start > 0 and count >= minimum_count: return start return None def functional_user_namespace(program: str) -> bool: executable = shutil.which(program) if executable is None: return False try: result = subprocess.run( [executable, "--user", "--map-root-user", "true"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=5, check=False, ) except (OSError, subprocess.TimeoutExpired): return False return result.returncode == 0 def mapping_helpers_usable(feature: dict, uid_start: int, gid_start: int) -> bool: unshare = shutil.which(feature["unshare_program"]) newuidmap = shutil.which(feature["uidmap_program"]) newgidmap = shutil.which(feature["gidmap_program"]) if None in {unshare, newuidmap, newgidmap}: return False try: process = subprocess.Popen( [unshare, "--user", "--", "sh", "-c", "kill -STOP $$; exec sleep 5"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) except OSError: return False try: try: _, status = os.waitpid(process.pid, os.WUNTRACED) except OSError: return False if not os.WIFSTOPPED(status): return False count = feature["minimum_count"] uid_result = subprocess.run( [newuidmap, str(process.pid), "0", str(os.getuid()), "1", "1", str(uid_start), str(count)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False, ) gid_result = subprocess.run( [newgidmap, str(process.pid), "0", str(os.getgid()), "1", "1", str(gid_start), str(count)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False, ) return uid_result.returncode == 0 and gid_result.returncode == 0 finally: if process.poll() is None: os.kill(process.pid, signal.SIGKILL) process.wait() def subordinate_ids_available(feature: dict) -> bool: try: user = pwd.getpwuid(os.getuid()) except KeyError: return False principals = {user.pw_name, str(user.pw_uid)} minimum_count = feature["minimum_count"] uid_start = subordinate_id_range(Path("/etc/subuid"), principals, minimum_count) gid_start = subordinate_id_range(Path("/etc/subgid"), principals, minimum_count) if uid_start is None or gid_start is None: return False return mapping_helpers_usable(feature, uid_start, gid_start) def probe_feature(feature: dict) -> tuple[bool, str]: probe = feature["probe"] if probe == "linux-x86_64": passed = platform.system() == "Linux" and platform.machine() == "x86_64" return passed, f"{platform.system()} {platform.machine()}" if probe == "kernel-at-least": release = platform.release() observed = kernel_version_tuple(release) passed = platform.system() == "Linux" and observed is not None and observed >= version_tuple(feature["minimum"]) return passed, release if probe == "registered-host-support": return False, "registered runner pending IP-03" if probe == "user-namespace": try: value = int(Path(feature["sysctl_path"]).read_text().strip()) except (OSError, ValueError): value = 0 passed = value > 0 and functional_user_namespace(feature["program"]) return passed, "functional" if passed else "unavailable or nonfunctional" if probe == "subordinate-ids": passed = subordinate_ids_available(feature) return passed, "configured and usable" if passed else "unavailable or unusable" if probe == "open-character-device": try: path = Path(feature["path"]) passed = stat.S_ISCHR(path.stat().st_mode) descriptor = os.open(path, os.O_RDWR | os.O_CLOEXEC | os.O_NONBLOCK) if passed else None except OSError: passed = False descriptor = None if descriptor is not None: os.close(descriptor) return passed, "openable read-write" if passed else "unavailable or inaccessible" raise RegistryError(f"unknown feature probe {probe}") def probe(level: str, external: dict | None = None) -> tuple[set[str], set[str]]: if external is None: _, external, _, _ = validated_data() available = set() skipped = set() failed = False for command in external["command"]: if level not in command["levels"]: continue program = shutil.which(command["program"]) output = "" passed = program is not None if passed: result = subprocess.run([program, *command["args"]], text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=False) output = result.stdout.strip() passed = result.returncode in command.get("accepted_exit_codes", [0]) and re.search(command["version_pattern"], output) is not None if passed: available.add(command["id"]) print(f"PASS {command['id']}: {output.splitlines()[0]}") else: status = command["policy"].upper() print(f"{status} {command['id']}: {command['diagnostic']}") if command["policy"] == "fail": failed = True else: skipped.add(command["id"]) for feature in external["feature"]: if level not in feature["levels"]: continue passed, detail = probe_feature(feature) if passed: available.add(feature["id"]) print(f"PASS {feature['id']}: {detail}") else: status = feature["policy"].upper() print(f"{status} {feature['id']}: {feature['diagnostic']}") if feature["policy"] == "fail": failed = True else: skipped.add(feature["id"]) require(not failed, f"{level} external capability probe failed") return available, skipped def run(command: list[str], cwd: Path = ROOT) -> None: print(f"RUN {' '.join(command)}") subprocess.run(command, cwd=cwd, check=True) def dependency_audit() -> None: policy = load_toml(ROOT / "deny.toml") allowed_licenses = set(policy["licenses"]["allow"]) result = subprocess.run(["cargo", "metadata", "--locked", "--offline", "--format-version", "1"], cwd=ROOT, text=True, stdout=subprocess.PIPE, check=True) metadata = json.loads(result.stdout) workspace_members = set(metadata["workspace_members"]) for package in metadata["packages"]: if package["id"] in workspace_members: continue source = package.get("source") or "" require(source.startswith("registry+https://github.com/rust-lang/crates.io-index"), f"unapproved dependency source: {package['name']}") license_expression = package.get("license") or "" tokens = set(re.findall(r"[A-Za-z0-9.-]+", license_expression)) - {"AND", "OR", "WITH"} require(bool(tokens) and tokens <= allowed_licenses, f"unapproved dependency license: {package['name']} {license_expression}") for package in metadata["packages"]: for dependency in package["dependencies"]: source = dependency.get("source") require(source is None or source.startswith("registry+https://github.com/rust-lang/crates.io-index"), f"git or unknown dependency: {package['name']} -> {dependency['name']}") if source is None: dependency_path = dependency.get("path") require(dependency_path is not None and Path(dependency_path).resolve().is_relative_to(ROOT), f"external path dependency: {dependency['name']}") else: require(dependency["req"] != "*", f"wildcard dependency: {package['name']} -> {dependency['name']}") run(["cargo", "deny", "--frozen", "check", "bans", "licenses", "sources"]) print("PASS dependency source, wildcard, and license audit") def has_process_lint_exemption(source: str) -> bool: warnings_allow = re.compile(r"#\s*!?\s*\[[^]]*(?:allow|expect)\s*\([^]]*warnings") return "disallowed_methods" in source or "disallowed_types" in source or warnings_allow.search(source) is not None def validate_workspace_process_lints(workspace: dict) -> None: lints = workspace.get("workspace", {}).get("lints", {}) require(lints.get("rust", {}).get("unsafe_code") == "forbid", "workspace must forbid unsafe Rust") clippy = lints.get("clippy", {}) require(clippy.get("disallowed_methods") == "deny", "workspace must deny Clippy disallowed methods") require(clippy.get("disallowed_types") == "deny", "workspace must deny Clippy disallowed types") def validate_member_process_lints(manifest: dict, package: str) -> None: require(manifest.get("lints") == {"workspace": True}, f"{package}: package must inherit workspace lints without override") def workspace_member_lints() -> dict[str, dict]: workspace = load_toml(ROOT / "Cargo.toml")["workspace"] return {member: load_toml(ROOT / member / "Cargo.toml").get("lints", {}) for member in workspace["members"]} def process_source_audit(process_safety: dict) -> None: policy = process_safety["policy"] validate_workspace_process_lints(load_toml(ROOT / "Cargo.toml")) require((ROOT / policy["lint_config"]).read_text() == clippy_config_view(process_safety), "generated Clippy process policy drift") lint_allow_paths = set(policy["lint_allow_paths"]) for root in policy["production_roots"]: root_path = ROOT / root require(root_path.is_dir(), f"process-safety production root is missing: {root}") for path in sorted(root_path.rglob("*.rs")): relative = str(path.relative_to(ROOT)) has_allow = has_process_lint_exemption(path.read_text(errors="replace")) require(not has_allow or relative in lint_allow_paths, f"process-safety lint exemption outside the broker boundary: {relative}") result = subprocess.run( ["cargo", "metadata", "--locked", "--offline", "--all-features", "--format-version", "1"], cwd=ROOT, text=True, stdout=subprocess.PIPE, check=True, ) metadata = json.loads(result.stdout) packages = {package["id"]: package for package in metadata["packages"]} workspace_members = set(metadata["workspace_members"]) for package_id in workspace_members: package = packages[package_id] validate_member_process_lints(load_toml(Path(package["manifest_path"])), package["name"]) lock_packages = { (package["name"], package["version"]): package for package in load_toml(ROOT / "Cargo.lock")["package"] if "source" in package } actual_dependencies = {} for node in metadata["resolve"]["nodes"]: if node["id"] in workspace_members: continue package = packages[node["id"]] identity = (package["name"], package["version"]) require(identity not in actual_dependencies, f"duplicate resolved dependency {identity[0]} {identity[1]}") require(identity in lock_packages, f"resolved dependency is absent from Cargo.lock: {identity[0]} {identity[1]}") lock_package = lock_packages[identity] actual_dependencies[identity] = { "checksum": lock_package.get("checksum"), "features": sorted(node["features"]), } expected_dependencies = { (dependency["name"], dependency["version"]): { "checksum": dependency["checksum"], "features": dependency["features"], } for dependency in process_safety["runtime_dependency"] } require(actual_dependencies == expected_dependencies, f"process-safety dependency source/feature drift; expected={expected_dependencies}, actual={actual_dependencies}") for dependency in process_safety["runtime_dependency"]: if dependency["name"] in PROCESS_REVIEW_DEPENDENCIES: require(dependency["reviewed_process_apis"], f"{dependency['name']} {dependency['version']}: process APIs require explicit review") print("PASS process-safety dependency source/features and compiler lint-boundary audit") def validate_process_trace(lines: list[str], policy: dict, trace_case: dict) -> None: binary = trace_case["binary"] execve_count = sum(line.startswith("execve(") for line in lines) require(execve_count == trace_case["initial_execve_count"], f"{binary}: unexpected execve surface count {execve_count}") syscall_patterns = [re.compile(pattern) for pattern in policy["forbidden_syscall_patterns"]] for line in lines: matches = [pattern.pattern for pattern in syscall_patterns if pattern.search(line) is not None] require(not matches, f"{binary}: forbidden process syscall surface {line}: {matches}") def process_syscall_audit(process_safety: dict) -> None: policy = process_safety["policy"] strace = shutil.which("strace") require(strace is not None, "strace is required for the dynamic process-syscall audit") run(["cargo", "build", "--offline", "--workspace", "--all-features", "--bins", "--locked"]) metadata_result = subprocess.run( ["cargo", "metadata", "--locked", "--offline", "--all-features", "--format-version", "1"], cwd=ROOT, text=True, stdout=subprocess.PIPE, check=True, ) metadata = json.loads(metadata_result.stdout) workspace_members = set(metadata["workspace_members"]) binaries = { target["name"] for package in metadata["packages"] if package["id"] in workspace_members for target in package["targets"] if "bin" in target["kind"] } trace_cases = {trace_case["binary"]: trace_case for trace_case in process_safety["trace_case"]} require(set(trace_cases) == binaries, f"process-safety trace-case coverage drift; expected={sorted(binaries)}, actual={sorted(trace_cases)}") with tempfile.TemporaryDirectory(prefix="lazaret-process-syscalls-") as directory: for binary in sorted(binaries): trace_case = trace_cases[binary] trace_base = Path(directory) / binary result = subprocess.run( [ strace, "-ff", "-qq", "-e", f"trace={policy['trace_expression']}", "-o", str(trace_base), "--", str(ROOT / "target" / "debug" / binary), *trace_case["args"], ], cwd=ROOT, check=False, ) require(result.returncode == trace_case["expected_exit_code"], f"{binary}: exercised path exited {result.returncode}") traces = sorted(Path(directory).glob(f"{binary}.*")) require(traces, f"process-safety syscall audit produced no trace for {binary}") lines = [line for trace in traces for line in trace.read_text(errors="replace").splitlines()] validate_process_trace(lines, policy, trace_case) print("PASS registered final-binary exercised-path process-syscall audit") def refresh_advisories() -> None: run(["cargo", "deny", "fetch"]) run(["cargo", "deny", "--frozen", "check", "advisories"]) def run_models() -> None: refinements = load_toml(SUPPORT / "model-refinements.toml") with tempfile.TemporaryDirectory(prefix="lazaret-tlc-") as directory: for model in refinements["model"]: module = ROOT / model["module"] config = ROOT / model["config"] metadata = Path(directory) / model["name"] run(["tlc", "-workers", "1", "-metadir", str(metadata), "-config", str(config), str(module)], Path(directory)) def run_integrations(level: str, external: dict, integrations: dict) -> None: scenarios = [scenario for scenario in integrations["scenarios"] if scenario["level"] == level] if not scenarios: print(f"SKIP {level} integration harness: no scenarios registered") return available, skipped = probe(level, external) by_id = {item["id"]: item for kind in ("command", "feature") for item in external[kind]} for scenario in scenarios: missing = set(scenario.get("capabilities", [])) - available if missing: require(missing <= skipped and all(by_id[item]["policy"] == "skip" for item in missing), f"{scenario['name']}: undeclared capability failure") print(f"SKIP {scenario['name']}: missing {', '.join(sorted(missing))}") continue run(scenario["command"]) def verify(level: str) -> None: _, external, integrations, process_safety = validated_data() generate(check=True) probe("V1", external) run(["cargo", "fmt", "--all", "--", "--check"]) run(["cargo", "clippy", "--offline", "--workspace", "--all-targets", "--all-features", "--locked", "--", "-D", "warnings"]) run(["cargo", "test", "--offline", "--workspace", "--all-features", "--locked"]) run(["cargo", "test", "--offline", "--workspace", "--all-features", "--locked", "--doc"]) run([sys.executable, "-m", "unittest", "discover", "-s", "tests/tooling"]) dependency_audit() process_source_audit(process_safety) run_models() target_index = LEVELS.index(level) for current in LEVELS[1:target_index + 1]: probe(current, external) run_integrations(current, external, integrations) def main() -> int: parser = argparse.ArgumentParser() subparsers = parser.add_subparsers(dest="command", required=True) generate_parser = subparsers.add_parser("generate") generate_parser.add_argument("--check", action="store_true") probe_parser = subparsers.add_parser("probe") probe_parser.add_argument("level", choices=LEVELS) verify_parser = subparsers.add_parser("verify") verify_parser.add_argument("level", choices=LEVELS) subparsers.add_parser("validate") subparsers.add_parser("dependencies") subparsers.add_parser("refresh-advisories") subparsers.add_parser("models") process_audit_parser = subparsers.add_parser("process-audit") process_audit_parser.add_argument("--dynamic", action="store_true") args = parser.parse_args() try: if args.command == "generate": generate(args.check) elif args.command == "probe": probe(args.level) elif args.command == "verify": verify(args.level) elif args.command == "validate": validated_data() elif args.command == "dependencies": dependency_audit() elif args.command == "refresh-advisories": validated_data() refresh_advisories() elif args.command == "models": validated_data() run_models() elif args.command == "process-audit": _, _, _, process_safety = validated_data() process_source_audit(process_safety) if args.dynamic: process_syscall_audit(process_safety) except (RegistryError, OSError, subprocess.CalledProcessError) as error: print(f"ERROR: {error}", file=sys.stderr) return 1 return 0 if __name__ == "__main__": sys.exit(main())