diff --git a/scripts/pre1_harness_binding.py b/scripts/pre1_harness_binding.py index e7a0653..d0d68b6 100644 --- a/scripts/pre1_harness_binding.py +++ b/scripts/pre1_harness_binding.py @@ -13,14 +13,21 @@ TOOL_ONLY_PATHS = frozenset({ + "qualification/pre1-cases.json", + "scripts/pre1_comparative_topology.py", + "scripts/pre1_installed_discovery.py", "scripts/run_pre1_qualification_case.py", "scripts/test_pre1_qualification_case.py", "scripts/pre1_harness_binding.py", "scripts/pre1_environment_contract.py", "scripts/pre1_package_execution.py", "scripts/test_pre1_package_execution.py", + "scripts/test_pre1_prepared_home.py", + "scripts/test_pre1_comparative_topology.py", + "scripts/test_pre1_installed_discovery.py", "scripts/run_php83_local_ci.py", "scripts/test_php83_local_ci.py", + "scripts/prepare_pre1_minimum_runtime.py", }) @@ -32,6 +39,26 @@ }) +# This is an exact append-only qualification note, not a documentation wildcard. +# Both the released prefix and the reviewed addition remain immutable here. +REVIEWED_DOCUMENT_APPENDICES = { + "OPERATIONS.md": { + "base_sha256": "79602ed2ddc66383fe7777415e9cb125b29134e3bc4a553edce5f3aab796ea45", + "append_sha256": "bc0a713ae89a67c8cb244d25edc15c7b6fd384ead6b0c54f4d39e59ffdc0f8b7", + }, +} + + +def validate_document_appendix(root: Path, path: str, product: str, harness: str) -> None: + expected = REVIEWED_DOCUMENT_APPENDICES[path] + before = git(root, "show", f"{product}:{path}") + after = git(root, "show", f"{harness}:{path}") + if (hashlib.sha256(before).hexdigest() != expected["base_sha256"] + or not after.startswith(before) + or hashlib.sha256(after[len(before):]).hexdigest() != expected["append_sha256"]): + raise ValueError("qualification documentation differs from reviewed append-only note") + + def git(root: Path, *argv: str) -> bytes: return subprocess.check_output( ["git", *argv], cwd=root, stderr=subprocess.PIPE, timeout=30 @@ -75,5 +102,9 @@ def validate_harness_source( 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 | CI_ONLY_PATHS): + remaining = set(filter(None, changes)) - (TOOL_ONLY_PATHS | CI_ONLY_PATHS) + for path in remaining & REVIEWED_DOCUMENT_APPENDICES.keys(): + validate_document_appendix(root, path, product_source_commit, binding["harness_source_commit"]) + remaining.remove(path) + if remaining: raise ValueError("qualification tooling revision changes frozen product source") diff --git a/scripts/pre1_package_execution.py b/scripts/pre1_package_execution.py index 44320e2..60475b2 100644 --- a/scripts/pre1_package_execution.py +++ b/scripts/pre1_package_execution.py @@ -341,6 +341,13 @@ def validate_binding_context(value: dict, context: dict) -> None: if any(value.get(k) != context[k] for k in ("component", "runtime", "target")) or value.get("bindings") != {k: context[k] for k in BINDINGS}: raise ValueError("package execution candidate/environment/runtime binding differs") +def prepared_package_home() -> Path: + """Keep the immutable preparation boundary separate from per-case HOME.""" + return safe_path( + Path(os.environ.get("IICP_PRE1_PREPARED_PACKAGE_HOME", os.environ["HOME"])) + ) + + def validate_binding(value: dict, context: dict, artifact: Path, root: Path, vendor_artifact: Path | None = None) -> Path: if context["component"] == "management": @@ -350,7 +357,7 @@ def validate_binding(value: dict, context: dict, artifact: Path, root: Path, validate_binding_identity(value) validate_binding_context(value, context) workspace = safe_path(Path(value["workspace"])) - home = safe_path(Path(os.environ["HOME"])) + home = prepared_package_home() installed = safe_path(Path(value["installed_package"])) validate_workspace_boundary(workspace, home, installed, root) if value["artifact_sha256"] != file_digest(artifact) or value["installed_payload_sha256"] != digest(installed_payload(artifact, installed, context["component"])): @@ -505,7 +512,7 @@ def validate_summary_identity(summary: dict) -> None: def write_case_proof(value: dict) -> Path: """Publish a complete sidecar atomically, without overwriting earlier evidence.""" path = Path(os.environ["IICP_PRE1_CASE_PROOF_OUTPUT"]) - home = safe_path(Path(os.environ["HOME"])) + home = prepared_package_home() parent = safe_path(path.parent) if not path.is_absolute() or not parent.is_relative_to(home) or path.exists() or path.is_symlink(): raise ValueError("case proof output is unsafe or already exists") @@ -597,7 +604,7 @@ def rust_fixtures(root: Path, installed: Path) -> dict[str, str]: def create_rust_binding(root, workspace, installed, artifact, runtime, target, bindings, vendor_artifact): validate_immutable_bindings(bindings) - validate_workspace_boundary(safe_path(workspace), safe_path(Path(os.environ["HOME"])), + validate_workspace_boundary(safe_path(workspace), prepared_package_home(), safe_path(installed), root) payload, deps = verify_rust_payload(workspace, installed, artifact, vendor_artifact) value = {"schema": SCHEMA, "component": "client-rust", "runtime": runtime, @@ -618,7 +625,7 @@ def validate_rust_binding(value, context, artifact, root, vendor_artifact): validate_binding_context(value, context) workspace = safe_path(Path(value["workspace"])) installed = safe_path(Path(value["installed_package"])) - validate_workspace_boundary(workspace, safe_path(Path(os.environ["HOME"])), installed, root) + validate_workspace_boundary(workspace, prepared_package_home(), installed, root) payload, deps = verify_rust_payload(workspace, installed, artifact, vendor_artifact) expected = {"artifact_sha256": file_digest(artifact), "vendor_artifact_sha256": file_digest(vendor_artifact), @@ -796,7 +803,7 @@ def management_assertions(name, source): def stage_management_consumer(root: Path, artifact: Path, workspace: Path) -> dict: """Prepare only; the caller owns dependency acquisition and isolation.""" root, artifact, workspace = safe_path(root), safe_path(artifact), safe_path(workspace) - home = safe_path(Path(os.environ["HOME"])) + home = prepared_package_home() if workspace == home or not workspace.is_relative_to(home) or workspace.is_relative_to(root) or any(workspace.iterdir()): raise ValueError("Management consumer workspace is not empty and run-isolated") expected = rust_archive_files(artifact, artifact.stem + "/") @@ -820,7 +827,7 @@ def stage_management_consumer(root: Path, artifact: Path, workspace: Path) -> di def validate_management_consumer(root: Path, artifact: Path, workspace: Path, binding: dict) -> Path: """Recheck artifact and reviewed assertions, even after a forged rehash.""" workspace = safe_path(workspace) - home = safe_path(Path(os.environ["HOME"])) + home = prepared_package_home() if workspace == home or not workspace.is_relative_to(home) or workspace.is_relative_to(root.resolve()): raise ValueError("Management consumer workspace is not run-isolated") payload = safe_path(workspace / "payload") diff --git a/scripts/run_pre1_qualification_case.py b/scripts/run_pre1_qualification_case.py index 8e7c28d..c12ba1e 100755 --- a/scripts/run_pre1_qualification_case.py +++ b/scripts/run_pre1_qualification_case.py @@ -433,7 +433,7 @@ def validate_runtime(runtime: str, runtime_row: dict, manifest: dict) -> None: def command_environment(runtime_row: dict, runtime: str) -> dict[str, str]: - allowed = {"PATH", "HOME", "IICP_HOME", "LANG", "LC_ALL", "SYSTEMROOT", "WINDIR", "COMSPEC", "PATHEXT", "TEMP", "TMP", "TMPDIR", "RUSTUP_HOME"} + allowed = {"PATH", "HOME", "IICP_HOME", "LANG", "LC_ALL", "SYSTEMROOT", "WINDIR", "COMSPEC", "PATHEXT", "TEMP", "TMP", "TMPDIR", "RUSTUP_HOME", "IICP_PRE1_PREPARED_PACKAGE_HOME"} env = {key: value for key, value in os.environ.items() if key.upper() in allowed} for name in ( "IICP_PRE1_CELL_ID", diff --git a/scripts/test_pre1_prepared_home.py b/scripts/test_pre1_prepared_home.py new file mode 100644 index 0000000..e3f2600 --- /dev/null +++ b/scripts/test_pre1_prepared_home.py @@ -0,0 +1,81 @@ +"""Preparation ownership must not inherit the private execution HOME.""" +import ast +import json +import os +from pathlib import Path +import tempfile +import unittest +from unittest.mock import patch +import pre1_package_execution as adapter + + +class PreparedHomeTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.root = Path(self.tmp.name).resolve() + self.prepared = self.root / "prepared" + self.case = self.root / "case" + self.prepared.mkdir(mode=0o700) + self.case.mkdir(mode=0o700) + + def test_legacy_preparation_defaults_to_home(self): + with patch.dict(os.environ, {"HOME": str(self.prepared)}, clear=True): + self.assertEqual(adapter.prepared_package_home(), self.prepared) + + def test_execution_home_does_not_replace_preparation_owner(self): + workspace = self.prepared / "workspace" + installed = workspace / "payload" + installed.mkdir(parents=True) + source = self.root / "source" + source.mkdir() + env = {"HOME": str(self.case), "IICP_PRE1_PREPARED_PACKAGE_HOME": str(self.prepared)} + with patch.dict(os.environ, env, clear=True): + self.assertEqual(adapter.prepared_package_home(), self.prepared) + adapter.validate_workspace_boundary(workspace, adapter.prepared_package_home(), installed, source) + with self.assertRaises(ValueError): + adapter.validate_workspace_boundary(workspace, self.case, installed, source) + + def test_wrong_explicit_preparation_boundary_is_rejected(self): + workspace = self.prepared / "workspace" + installed = workspace / "payload" + installed.mkdir(parents=True) + with patch.dict(os.environ, {"HOME": str(self.prepared), + "IICP_PRE1_PREPARED_PACKAGE_HOME": str(self.case)}, clear=True): + with self.assertRaises(ValueError): + adapter.validate_workspace_boundary(workspace, adapter.prepared_package_home(), installed, self.root / "source") + + def test_symlink_preparation_boundary_is_rejected(self): + alias = self.root / "alias" + alias.symlink_to(self.prepared, target_is_directory=True) + with patch.dict(os.environ, {"HOME": str(self.case), + "IICP_PRE1_PREPARED_PACKAGE_HOME": str(alias)}, clear=True): + with self.assertRaises(ValueError): + adapter.prepared_package_home() + + def test_driver_preserves_preparation_boundary(self): + # Execute the actual owning function without importing its driver (which + # performs unrelated qualification-map initialization at import time). + path = Path(adapter.__file__).with_name("run_pre1_qualification_case.py") + tree = ast.parse(path.read_text()) + function = next(n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name == "command_environment") + namespace = {"os": os, "Path": Path, "json": json, "COMPONENT": "fixture"} + exec(compile(ast.Module(body=[function], type_ignores=[]), str(path), "exec"), namespace) + manifest = self.root / "candidate.json" + manifest.write_text(json.dumps({"components": [{"id": "fixture", "artifacts": [{"kind": "release-manifest", "name": "release.json"}]}]})) + env = {"HOME": str(self.case), "IICP_HOME": str(self.case), + "IICP_PRE1_PREPARED_PACKAGE_HOME": str(self.prepared)} + for node in ast.walk(function): + if isinstance(node, ast.For) and isinstance(node.iter, ast.Tuple): + for value in node.iter.elts: + if isinstance(value, ast.Constant) and isinstance(value.value, str): + env[value.value] = "diagnostic-fixture" + env.update(IICP_PRE1_CANDIDATE_MANIFEST=str(manifest), IICP_PRE1_ARTIFACT_ROOT=str(self.root)) + with patch.dict(os.environ, env, clear=True): + result = namespace["command_environment"]({"env": {}, "programs": {name: "/runtime/" + name for name in ("python", "node", "cargo", "rustc", "php")}}, "fixture") + self.assertEqual(result["HOME"], str(self.case)) + self.assertEqual(result["IICP_PRE1_PREPARED_PACKAGE_HOME"], str(self.prepared)) + + +if __name__ == "__main__": + unittest.main()