From c86e9e35ca8f05b589d2f5ea15c34ca10634b161 Mon Sep 17 00:00:00 2001 From: RobLe3 Date: Wed, 16 Sep 2026 10:32:31 +0200 Subject: [PATCH] fix: bind qualification tooling separately and validate v3 environments --- scripts/pre1_environment_contract.py | 306 ++++++++++++++++++++++++ scripts/pre1_harness_binding.py | 67 ++++++ scripts/run_pre1_qualification_case.py | 31 ++- scripts/test_pre1_qualification_case.py | 65 +++++ 4 files changed, 465 insertions(+), 4 deletions(-) create mode 100644 scripts/pre1_environment_contract.py create mode 100644 scripts/pre1_harness_binding.py diff --git a/scripts/pre1_environment_contract.py b/scripts/pre1_environment_contract.py new file mode 100644 index 0000000..6fdb5fc --- /dev/null +++ b/scripts/pre1_environment_contract.py @@ -0,0 +1,306 @@ +"""Portable qualification environment contract; mirrored from the hub validator. + +Only qualification tooling imports this module. No product runtime dependency. +""" +from __future__ import annotations +import hashlib +import json +import re + + +def canonical_sha256(value: object) -> str: + data = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() + return "sha256:" + hashlib.sha256(data).hexdigest() + + +def valid_digest(value: object) -> bool: + return isinstance(value, str) and re.fullmatch(r"sha256:[0-9a-f]{64}", value) is not None + + +RUNTIME_FIELDS = { + "lock_inputs_sha256", + "dependency_cache_sha256", + "online_prepare_status", + "offline_install_status", + "package_artifact_smoke_status", + "egress_disabled_during_offline", + "empty_volatile_cache_at_start", +} +TOP_LEVEL_FIELDS = { + "schema", + "status", + "target", + "bindings", + "network", + "source_state", + "runtimes", + "content_free", + "secrets_present", + "non_authorizing", + "environment_sha256", +} +V2_TOP_LEVEL_FIELDS = TOP_LEVEL_FIELDS | {"execution"} +V3_TOP_LEVEL_FIELDS = V2_TOP_LEVEL_FIELDS | {"isolation"} +EXECUTION_FIELDS = { + "execution_kind", + "evidence_scope", + "host_architecture", + "guest_architecture", + "container_engine", + "container_engine_version", + "emulation_mechanism", + "image_digest", +} +EXECUTION_KINDS = { + "native", + "isolated_vm", + "container_native", + "container_emulated", +} + + +def _architecture(value: object) -> str: + normalized = str(value).lower() + if normalized in {"amd64", "x86_64"}: + return "x86_64" + if normalized in {"arm64", "aarch64"}: + return "aarch64" + return normalized + + +def _target_architecture(target: str) -> str: + return "x86_64" if target.endswith("x86_64") else "aarch64" + + +def _validate_container_execution(execution: dict, kind: object) -> list[str]: + container_fields = ( + "container_engine", + "container_engine_version", + "image_digest", + ) + if kind in {"container_native", "container_emulated"}: + errors = ( + ["qualification environment container provenance is incomplete"] + if not all( + isinstance(execution.get(field), str) and execution[field] + for field in container_fields + ) + else [] + ) + if not valid_digest(execution.get("image_digest")): + errors.append("qualification environment image digest is invalid") + return errors + if any(execution.get(field) is not None for field in container_fields): + return ["non-container qualification environment claims container provenance"] + return [] + + +def _validate_execution_architecture( + execution: dict, target: str, kind: object +) -> list[str]: + host = _architecture(execution.get("host_architecture")) + guest = _architecture(execution.get("guest_architecture")) + errors = ( + [] + if guest == _target_architecture(target) + else ["qualification environment guest architecture differs"] + ) + if kind == "container_emulated": + if target != "linux-x86_64": + errors.append("emulated qualification is limited to Linux x86-64") + if (host, guest) != ("aarch64", "x86_64"): + errors.append("Linux x86-64 emulation architecture provenance differs") + elif kind in {"native", "container_native"} and host != guest: + errors.append("native qualification host and guest architectures differ") + return errors + + +def _validate_emulation(execution: dict, kind: object) -> list[str]: + mechanism = execution.get("emulation_mechanism") + if kind == "container_emulated": + if not isinstance(mechanism, str) or not mechanism: + return ["emulated qualification lacks its mechanism"] + elif mechanism is not None: + return ["non-emulated qualification claims an emulation mechanism"] + return [] + + +def _validate_isolation(value: dict, schema: object) -> list[str]: + if schema != "iicp.pre1-qualification-environment.v3": + return [] + isolation = value.get("isolation") + if not isinstance(isolation, dict) or set(isolation) != { + "kind", + "identity", + "evidence_sha256", + }: + return ["qualification environment isolation provenance differs"] + if not _valid_isolation_values(isolation): + return ["qualification environment isolation evidence is invalid"] + return [] + + +def _valid_isolation_values(isolation: dict) -> bool: + text_values = (isolation.get("kind"), isolation.get("identity")) + return all(map(_nonempty_string, text_values)) and valid_digest( + isolation.get("evidence_sha256") + ) + + +def _nonempty_string(value: object) -> bool: + return isinstance(value, str) and bool(value) + + +def _validate_execution( + value: dict, target: str, expected_kind: str | None +) -> list[str]: + execution = value.get("execution") + if not isinstance(execution, dict) or set(execution) != EXECUTION_FIELDS: + return ["qualification environment execution provenance differs"] + errors: list[str] = [] + kind = execution.get("execution_kind") + if kind not in EXECUTION_KINDS: + errors.append("qualification environment execution kind is invalid") + if expected_kind is not None and kind != expected_kind: + errors.append("qualification environment execution kind differs") + scope = execution.get("evidence_scope") + expected_scope = ( + "functional-only" + if kind == "container_emulated" + else "functional-and-performance" + ) + if scope != expected_scope: + errors.append("qualification environment evidence scope differs") + return ( + errors + + _validate_container_execution(execution, kind) + + _validate_execution_architecture(execution, target, kind) + + _validate_emulation(execution, kind) + ) + + +def validate_runtime(runtime: str, value: object) -> list[str]: + if not isinstance(value, dict) or set(value) != RUNTIME_FIELDS: + return [f"qualification environment runtime fields differ: {runtime}"] + errors: list[str] = [] + for field in ("lock_inputs_sha256", "dependency_cache_sha256"): + if not valid_digest(value.get(field)): + errors.append(f"qualification environment has invalid {field}: {runtime}") + for field in ( + "online_prepare_status", + "offline_install_status", + "package_artifact_smoke_status", + ): + if value.get(field) != "PASS": + errors.append( + f"qualification environment gate is not PASS: {runtime}:{field}" + ) + for field in ("egress_disabled_during_offline", "empty_volatile_cache_at_start"): + if value.get(field) is not True: + errors.append( + f"qualification environment safety gate differs: {runtime}:{field}" + ) + return errors + + +def _validate_header(value: dict, target: str, execution_kind: str | None) -> list[str]: + errors: list[str] = [] + schema = value.get("schema") + expected_fields = ( + V3_TOP_LEVEL_FIELDS + if schema == "iicp.pre1-qualification-environment.v3" + else ( + V2_TOP_LEVEL_FIELDS + if schema == "iicp.pre1-qualification-environment.v2" + else TOP_LEVEL_FIELDS + ) + ) + if set(value) != expected_fields: + errors.append("qualification environment fields differ") + if schema not in { + "iicp.pre1-qualification-environment.v1", + "iicp.pre1-qualification-environment.v2", + "iicp.pre1-qualification-environment.v3", + }: + errors.append("unexpected qualification-environment schema") + if execution_kind == "container_emulated" and schema not in { + "iicp.pre1-qualification-environment.v2", + "iicp.pre1-qualification-environment.v3", + }: + errors.append("emulated qualification requires execution provenance") + errors.extend(_validate_isolation(value, schema)) + if value.get("status") != "READY": + errors.append("qualification environment is not ready") + if value.get("target") != target: + errors.append("qualification environment target differs") + if ( + value.get("content_free") is not True + or value.get("secrets_present") is not False + ): + errors.append("qualification environment must be content-free and secret-free") + if value.get("non_authorizing") is not True: + errors.append("qualification environment must remain non-authorizing") + if schema in { + "iicp.pre1-qualification-environment.v2", + "iicp.pre1-qualification-environment.v3", + }: + errors.extend(_validate_execution(value, target, execution_kind)) + return errors + + +def _validate_bindings(value: dict, expected_bindings: dict) -> list[str]: + errors: list[str] = [] + bindings = value.get("bindings") + if not isinstance(bindings, dict) or set(bindings) != set(expected_bindings): + return ["qualification environment bindings differ"] + for field, expected in expected_bindings.items(): + if not valid_digest(bindings.get(field)): + errors.append(f"qualification environment has invalid binding: {field}") + if expected is not None and bindings.get(field) != expected: + errors.append(f"qualification environment binds a different {field}") + return errors + + +def _validate_boundaries(value: dict) -> list[str]: + errors: list[str] = [] + if value.get("network") != { + "preparation_egress": "dependency-download-only", + "qualification_egress": "disabled", + "loopback_fixtures": True, + }: + errors.append("qualification environment network boundary differs") + if value.get("source_state") != { + "clean_checkout": True, + "empty_volatile_caches_at_start": True, + "product_artifacts_separate": True, + }: + errors.append("qualification environment source-state boundary differs") + return errors + + +def _validate_digest(value: dict) -> list[str]: + copy = json.loads(json.dumps(value)) + claimed = copy.get("environment_sha256") + copy["environment_sha256"] = None + if not valid_digest(claimed) or claimed != canonical_sha256(copy): + return ["qualification environment digest differs"] + return [] + + +def validate_modern_environment(value: dict, *, target: str, bindings: dict) -> None: + if value.get("schema") not in { + "iicp.pre1-qualification-environment.v2", + "iicp.pre1-qualification-environment.v3", + }: + raise ValueError("qualification modern environment schema differs") + errors = (_validate_header(value, target, None) + + _validate_bindings(value, bindings) + + _validate_boundaries(value) + _validate_digest(value)) + runtimes = value.get("runtimes") + if not isinstance(runtimes, dict) or not runtimes: + errors.append("qualification environment runtimes are missing") + else: + for runtime, row in runtimes.items(): + errors.extend(validate_runtime(runtime, row)) + if errors: + raise ValueError("; ".join(errors)) diff --git a/scripts/pre1_harness_binding.py b/scripts/pre1_harness_binding.py new file mode 100644 index 0000000..9a3a0db --- /dev/null +++ b/scripts/pre1_harness_binding.py @@ -0,0 +1,67 @@ +"""Separate qualification-tool identity from unchanged frozen product source. + +This module is copied verbatim into component-owned qualification tooling. It +does not participate in a product runtime or grant qualification credit. +""" + +from __future__ import annotations + +import hashlib +import re +import subprocess +from pathlib import Path + + +TOOL_ONLY_PATHS = frozenset({ + "scripts/run_pre1_qualification_case.py", + "scripts/test_pre1_qualification_case.py", + "scripts/pre1_harness_binding.py", + "scripts/pre1_environment_contract.py", +}) + + +def git(root: Path, *argv: str) -> bytes: + return subprocess.check_output( + ["git", *argv], cwd=root, stderr=subprocess.PIPE, timeout=30 + ) + + +def harness_identity(root: Path) -> dict[str, str]: + """Hash the tracked Git tree manifest, including all assertions/fixtures. + + The manifest contains modes, Git blob IDs and paths. Its SHA-256 and exact + commit together identify tooling; product payloads retain their own SHA-256. + """ + commit = git(root, "rev-parse", "HEAD").decode().strip() + tree = git(root, "ls-tree", "-r", "--full-tree", commit) + return { + "harness_source_commit": commit, + "harness_sha256": "sha256:" + hashlib.sha256(tree).hexdigest(), + } + + +def validate_harness_source( + root: Path, product_source_commit: str, binding: dict[str, str] +) -> None: + """Refuse dirty tooling, pin substitution or changes to any product file.""" + if not isinstance(binding, dict) or set(binding) != { + "harness_source_commit", "harness_sha256" + }: + raise ValueError("qualification harness binding fields differ") + if not re.fullmatch(r"[0-9a-f]{40}", str(binding["harness_source_commit"])): + raise ValueError("qualification harness source commit is invalid") + if not re.fullmatch(r"sha256:[0-9a-f]{64}", str(binding["harness_sha256"])): + raise ValueError("qualification harness digest is invalid") + if not re.fullmatch(r"[0-9a-f]{40}", str(product_source_commit)): + raise ValueError("qualification product source commit is invalid") + if harness_identity(root) != binding: + raise ValueError("qualification harness source or digest differs") + if git(root, "status", "--porcelain=v1", "--untracked-files=normal").strip(): + raise ValueError("qualification harness checkout is dirty") + # --no-renames makes a move appear as both a deletion and an addition. + changes = git( + root, "diff", "--no-renames", "--name-only", "-z", + product_source_commit, binding["harness_source_commit"], "--" + ).decode().split("\0") + if set(filter(None, changes)) - TOOL_ONLY_PATHS: + raise ValueError("qualification tooling revision changes frozen product source") diff --git a/scripts/run_pre1_qualification_case.py b/scripts/run_pre1_qualification_case.py index 67f7e65..865ce76 100755 --- a/scripts/run_pre1_qualification_case.py +++ b/scripts/run_pre1_qualification_case.py @@ -13,6 +13,11 @@ import sys from pathlib import Path +sys.dont_write_bytecode = True +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from pre1_harness_binding import harness_identity, validate_harness_source +from pre1_environment_contract import validate_modern_environment + ROOT = Path(__file__).resolve().parents[1] COMPONENT = 'client-python' RUNTIMES = ['cpython-3.11', 'cpython-3.12', 'cpython-3.13', 'cpython-3.14'] @@ -90,6 +95,12 @@ def description() -> dict: "evidence_policy": "digest-only", "artifact_consumption": "verified-candidate-root", "source_commit_binding": True, + "supported_environment_schemas": [ + "iicp.pre1-qualification-environment.v1", + "iicp.pre1-qualification-environment.v2", + "iicp.pre1-qualification-environment.v3", + ], + **harness_identity(ROOT), "semantic_binding": { "contract": "iicp.pre1-semantic-assertion-binding.v1", "exact_assertion_per_scenario": True, @@ -251,7 +262,11 @@ def _validate_environment_manifest( runtime_map_digest: str, ) -> str: if ( - value.get("schema") != "iicp.pre1-qualification-environment.v1" + value.get("schema") not in { + "iicp.pre1-qualification-environment.v1", + "iicp.pre1-qualification-environment.v2", + "iicp.pre1-qualification-environment.v3", + } or value.get("status") != "READY" or value.get("target") != target or value.get("content_free") is not True @@ -287,6 +302,8 @@ def _validate_environment_manifest( copy["environment_sha256"] = None if not valid_digest(claimed) or claimed != canonical_sha256(copy): raise ValueError("qualification environment digest differs") + if value.get("schema") != "iicp.pre1-qualification-environment.v1": + validate_modern_environment(value, target=target, bindings=expected) return str(claimed) @@ -318,9 +335,13 @@ def validate_context(cell: str, scenario: str | None) -> tuple[str, dict, dict, component = next((row for row in manifest.get("components", []) if row.get("id") == COMPONENT), None) if not isinstance(component, dict) or component.get("state") != "BUILT": raise ValueError("qualification component artifact set is not built") - head = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip() - if head != component.get("source_commit"): - raise ValueError("qualification source commit differs from the candidate") + harness_text = os.environ.get("IICP_PRE1_HARNESS_BINDING") + if harness_text is not None: + validate_harness_source(ROOT, component.get("source_commit"), json.loads(harness_text)) + else: + head = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip() + if head != component.get("source_commit"): + raise ValueError("qualification source commit differs from the candidate") artifact_root = Path(os.environ.get("IICP_PRE1_ARTIFACT_ROOT", "")) component_root = artifact_root / COMPONENT @@ -431,6 +452,8 @@ def command_environment(runtime_row: dict, runtime: str) -> dict[str, str]: "IICP_PRE1_CONTEXT_SHA256", ): env[name] = os.environ[name] + if "IICP_PRE1_HARNESS_BINDING" in os.environ: + env["IICP_PRE1_HARNESS_BINDING"] = os.environ["IICP_PRE1_HARNESS_BINDING"] env.update(runtime_row.get("env", {})) cache = Path(env["IICP_HOME"]) / "qualification-cache" / COMPONENT / runtime cache.mkdir(parents=True, exist_ok=True) diff --git a/scripts/test_pre1_qualification_case.py b/scripts/test_pre1_qualification_case.py index 4cecc71..ac4e9bd 100755 --- a/scripts/test_pre1_qualification_case.py +++ b/scripts/test_pre1_qualification_case.py @@ -168,5 +168,70 @@ def test_python_runtime_is_exactly_bound_to_cell(self) -> None: self.assertEqual(module.expected_runtime_version("cpython-3.11", {}), "3.11") + +class ModernEnvironmentTests(unittest.TestCase): + def fixture(self): + runtime = module.RUNTIMES[0] + value = { + "schema": "iicp.pre1-qualification-environment.v3", + "status": "READY", "target": "linux-x86_64", + "bindings": {key: "sha256:" + char * 64 for key, char in ( + ("candidate_manifest_sha256", "a"), + ("artifact_materialization_sha256", "b"), + ("runtime_map_sha256", "c"))}, + "network": {"preparation_egress": "dependency-download-only", + "qualification_egress": "disabled", "loopback_fixtures": True}, + "source_state": {"clean_checkout": True, "empty_volatile_caches_at_start": True, + "product_artifacts_separate": True}, + "execution": {"execution_kind": "native", "evidence_scope": "functional-and-performance", + "host_architecture": "x86_64", "guest_architecture": "x86_64", + "container_engine": None, "container_engine_version": None, + "emulation_mechanism": None, "image_digest": None}, + "isolation": {"kind": "isolated-fixture", "identity": "test-only", + "evidence_sha256": "sha256:" + "d" * 64}, + "runtimes": {runtime: { + "lock_inputs_sha256": "sha256:" + "e" * 64, + "dependency_cache_sha256": "sha256:" + "f" * 64, + "online_prepare_status": "PASS", "offline_install_status": "PASS", + "package_artifact_smoke_status": "PASS", "egress_disabled_during_offline": True, + "empty_volatile_cache_at_start": True}}, + "content_free": True, "secrets_present": False, "non_authorizing": True, + "environment_sha256": None, + } + return runtime, value + + def check(self, value, runtime): + value["environment_sha256"] = None + value["environment_sha256"] = module.canonical_sha256(value) + return module._validate_environment_manifest( + value, target="linux-x86_64", runtime=runtime, + candidate_digest="sha256:" + "a" * 64, + materialization_digest="sha256:" + "b" * 64, + runtime_map_digest="sha256:" + "c" * 64) + + def test_v3_preserves_provenance_and_package_gates(self): + runtime, value = self.fixture() + self.assertEqual(self.check(value, runtime), value["environment_sha256"]) + + def test_rehashed_invalid_provenance_is_not_accepted(self): + mutations = [("execution", "guest_architecture", "aarch64"), + ("execution", "evidence_scope", "functional-only"), + ("execution", "image_digest", "sha256:" + "0" * 64), + ("isolation", "evidence_sha256", "invalid"), + ("network", "qualification_egress", "enabled"), + ("source_state", "product_artifacts_separate", False)] + for section, key, replacement in mutations: + runtime, value = self.fixture() + value[section][key] = replacement + with self.subTest(key=key), self.assertRaises(ValueError): + self.check(value, runtime) + + def test_rehashed_invalid_dependency_hash_is_not_accepted(self): + runtime, value = self.fixture() + value["runtimes"][runtime]["dependency_cache_sha256"] = "missing" + with self.assertRaises(ValueError): + self.check(value, runtime) + + if __name__ == "__main__": unittest.main()