#!/usr/bin/env python3 """Build a confined, digest-bound SIM S0 evidence pack from accepted worker runs.""" from __future__ import annotations import argparse import hashlib import json import os import platform import re import shutil import subprocess from pathlib import Path from typing import Any import yaml EVIDENCE_SCHEMA = "missioncore.simulation-s0-evidence/v1" class EvidenceError(RuntimeError): """Evidence inputs are incomplete, inconsistent, or outside the reviewed boundary.""" def _run(command: list[str]) -> str: result = subprocess.run( command, check=False, capture_output=True, text=True, timeout=15, ) if result.returncode != 0: raise EvidenceError( f"command failed ({result.returncode}): {' '.join(command)}; " f"stderr={result.stderr.strip()!r}" ) return result.stdout.strip() def _load_json(path: Path) -> dict[str, Any]: loaded = json.loads(path.read_text(encoding="utf-8")) if not isinstance(loaded, dict): raise EvidenceError(f"expected a JSON object: {path}") return loaded def _load_yaml(path: Path) -> dict[str, Any]: documents = [ document for document in yaml.safe_load_all(path.read_text(encoding="utf-8")) if document is not None ] if len(documents) != 1 or not isinstance(documents[0], dict): raise EvidenceError(f"expected a YAML mapping: {path}") return documents[0] def _result_env(path: Path) -> dict[str, str]: result: dict[str, str] = {} for line in path.read_text(encoding="utf-8").splitlines(): key, separator, value = line.partition("=") if not separator or not key: raise EvidenceError(f"invalid result line in {path}: {line!r}") result[key] = value required_passes = { "px4_stock_rover", "uxrce_dds", "ros2_vehicle_status", "gazebo_clock", "pause_step_speed", "resource_baseline", "clean_lifecycle", } failed = sorted(key for key in required_passes if result.get(key) != "pass") if failed: raise EvidenceError(f"run did not pass required checks: {', '.join(failed)}") if result.get("actuator_authority") != "false": raise EvidenceError("run widened actuator authority") return result def _sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def _write_json(path: Path, payload: dict[str, Any]) -> None: path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") def _package_version(package: str) -> str: return _run(["dpkg-query", "-W", "-f=${Version}", package]) def _git_commit(path: Path) -> str: commit = _run( [ "git", "-c", f"safe.directory={path}", "-C", str(path), "rev-parse", "HEAD", ] ) if re.fullmatch(r"[a-f0-9]{40}", commit) is None: raise EvidenceError(f"invalid Git commit at {path}: {commit!r}") return commit def _memory_inventory() -> dict[str, int]: values: dict[str, int] = {} for line in Path("/proc/meminfo").read_text(encoding="utf-8").splitlines(): key, raw = line.split(":", maxsplit=1) values[key] = int(raw.strip().split()[0]) * 1024 return { "total_bytes": values["MemTotal"], "available_bytes": values["MemAvailable"], "swap_total_bytes": values["SwapTotal"], "swap_free_bytes": values["SwapFree"], } def _gpu_inventory() -> dict[str, str | int] | None: if shutil.which("nvidia-smi") is None: return None output = _run( [ "nvidia-smi", "--query-gpu=name,driver_version,memory.total", "--format=csv,noheader,nounits", ] ) name, driver, memory_mib = (part.strip() for part in output.splitlines()[0].split(",")) return { "name": name, "driver": driver, "memory_total_mib": int(memory_mib), } def _cpu_model() -> str: for line in Path("/proc/cpuinfo").read_text(encoding="utf-8").splitlines(): if line.startswith("model name"): return line.split(":", maxsplit=1)[1].strip() raise EvidenceError("CPU model was not available") def _component_inventory(profile: dict[str, Any], source_root: Path) -> dict[str, Any]: components = profile.get("components") if not isinstance(components, list) or not components: raise EvidenceError("profile components are missing") if any( not isinstance(component, dict) or component.get("qualification_state") != "accepted" for component in components ): raise EvidenceError("all profile components must be accepted before evidence admission") packages = { "ros-jazzy-ros-base": _package_version("ros-jazzy-ros-base"), "ros-jazzy-ros-gz": _package_version("ros-jazzy-ros-gz"), "gz-harmonic": _package_version("gz-harmonic"), "gz-sim8-cli": _package_version("gz-sim8-cli"), "libsdformat14": _package_version("libsdformat14"), "ros-jazzy-navigation2": _package_version("ros-jazzy-navigation2"), "ros-jazzy-nav2-bringup": _package_version("ros-jazzy-nav2-bringup"), } commits = { "px4-autopilot": _git_commit(source_root / "PX4-Autopilot"), "px4-msgs": _git_commit(source_root / "px4_msgs"), "px4-gazebo-models": _git_commit( source_root / "PX4-Autopilot" / "Tools" / "simulation" / "gz" ), "micro-xrce-dds-agent": _git_commit(source_root / "Micro-XRCE-DDS-Agent"), } declared = { str(component["id"]): { "resolved_version": component.get("resolved_version"), "resolved_commit": component.get("resolved_commit"), "license": component.get("license"), "qualification_state": component.get("qualification_state"), } for component in components } for identifier, commit in commits.items(): if declared[identifier]["resolved_commit"] != commit: raise EvidenceError(f"worker commit drift for {identifier}") nav2_prefix = _run(["ros2", "pkg", "prefix", "nav2_bringup"]) px4_msgs_prefix = _run(["ros2", "pkg", "prefix", "px4_msgs"]) nav2_executables = _run(["ros2", "pkg", "executables", "nav2_controller"]).splitlines() if not nav2_executables: raise EvidenceError("Nav2 controller executable is unavailable") return { "declared": declared, "installed_packages": packages, "observed_commits": commits, "runtime_discovery": { "nav2_bringup_prefix": nav2_prefix, "nav2_controller_executables": nav2_executables, "px4_msgs_prefix": px4_msgs_prefix, }, } def _world_model_from_log(path: Path) -> dict[str, str]: text = path.read_text(encoding="utf-8", errors="replace") world_model = re.search(r"world:\s*([^,\s]+),\s*model:\s*([^\s]+)", text) commit = re.search(r"PX4 git-hash:\s*([a-f0-9]{40})", text) version = re.search(r"PX4 version:\s*(.+)", text) if world_model is None or commit is None or version is None: raise EvidenceError(f"PX4 identity is incomplete in {path}") return { "world": world_model.group(1), "model": world_model.group(2), "px4_commit": commit.group(1), "px4_version": version.group(1).strip(), } def _ensure_confined(path: Path, root: Path, *, label: str) -> Path: resolved = path.expanduser().resolve(strict=True) if not resolved.is_relative_to(root): raise EvidenceError(f"{label} escaped reviewed root {root}: {resolved}") return resolved def _arguments() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--profile", type=Path, required=True) parser.add_argument("--run-1x", type=Path, required=True) parser.add_argument("--run-2x", type=Path, required=True) parser.add_argument("--output-dir", type=Path, required=True) parser.add_argument("--d-root", type=Path, required=True) parser.add_argument("--source-root", type=Path, required=True) parser.add_argument("--repository-commit", required=True) parser.add_argument("--wsl-windows-path", required=True) return parser.parse_args() def main() -> int: args = _arguments() d_root = args.d_root.expanduser().resolve(strict=True) profile_path = _ensure_confined(args.profile, d_root, label="profile") run_1x = _ensure_confined(args.run_1x, d_root, label="1x run") run_2x = _ensure_confined(args.run_2x, d_root, label="2x run") source_root = _ensure_confined(args.source_root, d_root, label="source root") output_dir = args.output_dir.expanduser().resolve(strict=False) if not output_dir.is_relative_to(d_root): raise EvidenceError("evidence output escaped the reviewed D-only root") if output_dir.exists(): raise EvidenceError(f"evidence output already exists: {output_dir}") if re.fullmatch(r"[a-f0-9]{40}", args.repository_commit) is None: raise EvidenceError("repository commit must be a full Git SHA") profile = _load_yaml(profile_path) profile_sha256 = _sha256(profile_path) result_1x = _result_env(run_1x / "result.env") result_2x = _result_env(run_2x / "result.env") if result_1x.get("speed_factor") != "1" or result_2x.get("speed_factor") != "2": raise EvidenceError("run speed factors are not the declared 1x/2x pair") time_1x = _load_json(run_1x / "time-control.json") time_2x = _load_json(run_2x / "time-control.json") resource_1x = _load_json(run_1x / "resource-baseline.json") resource_2x = _load_json(run_2x / "resource-baseline.json") if any( report.get("verdict") != "pass" for report in (time_1x, time_2x, resource_1x, resource_2x) ): raise EvidenceError("time or resource report is not a pass") output_dir.mkdir(parents=True) disk = shutil.disk_usage(d_root) target_host = profile["target_host"] worker_inventory = { "redacted": True, "repository_commit": args.repository_commit, "profile_sha256": profile_sha256, "target": { "system": platform.system(), "kernel_release": platform.release(), "architecture": platform.machine(), "wsl_distribution": os.environ.get("WSL_DISTRO_NAME"), "linux_pretty_name": _run( ["bash", "-lc", ". /etc/os-release && printf '%s' \"$PRETTY_NAME\""] ), }, "cpu": { "model": _cpu_model(), "logical_processors": os.cpu_count(), }, "memory": _memory_inventory(), "gpu": _gpu_inventory(), "disk": { "total_bytes": disk.total, "free_bytes": disk.free, }, "components": _component_inventory(profile, source_root), } if worker_inventory["target"]["wsl_distribution"] != target_host["wsl_distribution"]: raise EvidenceError("worker distribution does not match the profile") mutable_paths = profile["storage"]["mutable_paths"] placement: list[dict[str, Any]] = [] for item in mutable_paths: path = Path(str(item["wsl_path"])).resolve(strict=True) if not path.is_relative_to(d_root): raise EvidenceError(f"mutable path escaped D-only root: {path}") placement.append( { "id": item["id"], "windows_path": item["windows_path"], "wsl_path": str(path), "exists": path.is_dir(), } ) d_only = { "redacted": True, "profile_sha256": profile_sha256, "wsl_distribution_windows_path": args.wsl_windows_path, "wsl_mount_root": str(d_root), "container_runtime": target_host["container_runtime"], "existing_docker_desktop_used_or_modified": False, "mutable_paths": placement, "disk": { "total_bytes": disk.total, "free_bytes": disk.free, "minimum_free_gib": profile["storage"]["minimum_free_gib"], "stop_below_gib": profile["storage"]["stop_below_gib"], }, } px4_launch = { "repository_commit": args.repository_commit, "runs": [ { "run_id": result_1x["run_id"], "speed_factor": 1, "identity": _world_model_from_log(run_1x / "px4-rover.log"), "result": result_1x, }, { "run_id": result_2x["run_id"], "speed_factor": 2, "identity": _world_model_from_log(run_2x / "px4-rover.log"), "result": result_2x, }, ], } telemetry = { "runs": [ { "run_id": result["run_id"], "vehicle_status": _load_yaml(run / "vehicle-status.yaml"), "vehicle_status_topic_present": ( "/fmu/out/vehicle_status_v1" in (run / "ros-topics.txt").read_text(encoding="utf-8") ), } for result, run in ((result_1x, run_1x), (result_2x, run_2x)) ] } authoritative_clock = { "authority": profile["clock"], "runs": [ { "run_id": result["run_id"], "clock_sample": _load_yaml(run / "clock.yaml"), "pause_advance_ns": report["pause"]["observed_advance_ns"], } for result, run, report in ( (result_1x, run_1x, time_1x), (result_2x, run_2x, time_2x), ) ], } pause_step_speed = { "contract": profile["runtime_acceptance"], "runs": [ { "run_id": result_1x["run_id"], "speed_factor": 1, "pause": time_1x["pause"], "single_step": time_1x["single_step"], "speed": time_1x["speed"], "verdict": time_1x["verdict"], }, { "run_id": result_2x["run_id"], "speed_factor": 2, "pause": time_2x["pause"], "single_step": time_2x["single_step"], "speed": time_2x["speed"], "verdict": time_2x["verdict"], }, ], } clean_stop = { "runs": [ { "run_id": result_1x["run_id"], "clean_lifecycle": result_1x["clean_lifecycle"], }, { "run_id": result_2x["run_id"], "clean_lifecycle": result_2x["clean_lifecycle"], }, ], "owned_process_residue": False, } artifacts: dict[str, tuple[str, dict[str, Any]]] = { "worker-inventory": ( "worker-inventory.redacted.json", worker_inventory, ), "d-only-physical-placement": ( "d-only-physical-placement.redacted.json", d_only, ), "px4-stock-rover-launch": ( "px4-stock-rover-launch.redacted.json", px4_launch, ), "ros2-telemetry": ( "ros2-telemetry.redacted.json", telemetry, ), "authoritative-simulation-clock": ( "authoritative-simulation-clock.redacted.json", authoritative_clock, ), "pause-step-speed": ( "pause-step-speed.redacted.json", pause_step_speed, ), "clean-stop-no-orphans": ( "clean-stop-no-orphans.redacted.json", clean_stop, ), "resource-baseline-1x": ( "resource-baseline-1x.redacted.json", resource_1x, ), "resource-baseline-2x": ( "resource-baseline-2x.redacted.json", resource_2x, ), } required = profile.get("required_evidence") if not isinstance(required, list) or set(required) != set(artifacts): raise EvidenceError("profile required_evidence does not match the generator") manifest_checks: list[dict[str, str]] = [] for identifier in required: filename, payload = artifacts[str(identifier)] artifact_path = output_dir / filename _write_json(artifact_path, payload) manifest_checks.append( { "id": str(identifier), "status": "pass", "artifact": filename, "sha256": _sha256(artifact_path), "note": f"Redacted factual worker evidence for {identifier}.", } ) manifest = { "schema_version": EVIDENCE_SCHEMA, "profile_sha256": profile_sha256, "checks": manifest_checks, } manifest_path = output_dir / "evidence.yaml" manifest_path.write_text( yaml.safe_dump(manifest, sort_keys=False), encoding="utf-8", ) print( json.dumps( { "evidence_manifest": str(manifest_path), "profile_sha256": profile_sha256, "checks": len(manifest_checks), }, sort_keys=True, ) ) return 0 if __name__ == "__main__": raise SystemExit(main())