Bind supplemental probes to an explicit ready contract

reports/releases/v0.9.0/execution-v1/contract.json

@@ -40,7 +40,7 @@
   },
   "control_inventory": {
     "path": "reports/releases/v0.9.0/execution-v1/control-inventory.json",
-    "sha256": "4179a3617cd95bb950417171e1a5c9e8490577f98d559d340000a821789d8d6f"
+    "sha256": "d40d1a26797cc0392c0200c65778415206e78485c61821e18fbfaf0f5a5ed1d5"
   },
   "controls": [
     {
@@ -14472,7 +14472,7 @@
     "reports/releases/v0.9.0/execution-v1/codeql-compatibility/summary.json": "7bc68a9bc11994525c8dd536fc9c419c795003228ee285e288ea3bc7e9dd6adc",
     "reports/releases/v0.9.0/execution-v1/command-parsing-check.json": "73b9849ac390763dd65e26d829148323ec79f24fbae34f0572ae63a64bb306ab",
     "reports/releases/v0.9.0/execution-v1/control-integration-validation.json": "fe09cac1435971ff4535f0091f53568f5e8bac32675353eb8e755d9f86d38028",
-    "reports/releases/v0.9.0/execution-v1/control-inventory.json": "4179a3617cd95bb950417171e1a5c9e8490577f98d559d340000a821789d8d6f",
+    "reports/releases/v0.9.0/execution-v1/control-inventory.json": "d40d1a26797cc0392c0200c65778415206e78485c61821e18fbfaf0f5a5ed1d5",
     "reports/releases/v0.9.0/execution-v1/dependency-version-witnesses.json": "50d7d8566cb24d1fdee789bfc7932463f3d014c8702f69822101db46ba916480",
     "reports/releases/v0.9.0/execution-v1/held-tool-digests.json": "9d748428932fdf7922ae62a066304226166a8f7105ebd5a8be9a42c15a4ffab2",
     "reports/releases/v0.9.0/execution-v1/native-codeql-packs.json": "7ab00154d9b4b16197a4103ea25d16a69f5bd0a0a2623301249fe3182649a9d4",
@@ -14505,13 +14505,13 @@
     "scripts/probe-opentaint-modeling-surface.sh": "0415d7838cc0d0cd9a3e338cdd9d6e11f3dfc8ca8703f38347e87044c4e71de7",
     "scripts/probe-opentaint-native-activation.sh": "5400cc43cded6c84b17544a7affeefdff103f8ad70df39602ccd3b865c18328f",
     "scripts/probe-opentaint-primitive-tracking.sh": "88592a10525c2fd1c7de3c0e37e3e56f9a82a20b7b2ae72b2f6d9c7b92caba6c",
-    "scripts/probe-opentaint-product-v090.py": "0eab04452ecb434701400529b1a44bb21ff6596078d0c83295757055de09fced",
+    "scripts/probe-opentaint-product-v090.py": "0589faee6fae872d8991bb5f3d30db5d00c95ea11c5a56fbca8e89dfe726c864",
     "scripts/probe-opentaint-scan-activation.sh": "d8c7074105c644ec9990d96eb281288e9795fd76f7a01ebddafe7f97ea9d384b",
     "scripts/probe-opentaint-value-kind.sh": "1f1aae2c22a238011cc7061e9faf47cd075acad783264a1564582389de9fe22a",
     "scripts/probe-pysa-callee-resolution.sh": "ec4501bda36ad2a017af84d609eedfe8bfd706f7c5de3c37565116750f225cbe",
     "scripts/probe-pysa-modeling-load-bearing.sh": "315b8a46993392bd93ab211bad1b90318e3fcfdd313652229c15bc5285d9e15a",
     "scripts/probe-pysa-native-activation.sh": "cd55635af84a86806ddd21e5726b19315330349e1cb122a09417fa36d8e43871",
-    "scripts/probe-python-modeling-load-bearing-v090.py": "0331eba3b003d455f1cee163c4f47d3bd648551f07473c5b7eeb3ad0e2900073",
+    "scripts/probe-python-modeling-load-bearing-v090.py": "a3862528578f74ac99acf43ef6af6a6f9d72d1411b9fa9ae4116b03f31a77536",
     "scripts/probe-semgrep-jsjava-native.sh": "d09e3bab859030d4679af455a341e7a155119fd48cba248abdc11b0fdf8d44f2",
     "scripts/probe-warm-observability-v090.py": "c2c31b70154f25287488f262a3bd23e5d029e784d1c7c60173212a113b7346a8",
     "scripts/release_control_attempt_v090.py": "7e876cfd9c0be39c2c0a85bec99c1282366f3c69fffe3eb524b5aa2ffacf41b5",

reports/releases/v0.9.0/execution-v1/control-inventory.json

@@ -1511,7 +1511,7 @@
       "script_identity": [
         {
           "path": "scripts/probe-python-modeling-load-bearing-v090.py",
-          "sha256": "0331eba3b003d455f1cee163c4f47d3bd648551f07473c5b7eeb3ad0e2900073"
+          "sha256": "a3862528578f74ac99acf43ef6af6a6f9d72d1411b9fa9ae4116b03f31a77536"
         }
       ],
       "stage": "supplemental",
@@ -1591,7 +1591,7 @@
       "script_identity": [
         {
           "path": "scripts/probe-opentaint-product-v090.py",
-          "sha256": "0eab04452ecb434701400529b1a44bb21ff6596078d0c83295757055de09fced"
+          "sha256": "0589faee6fae872d8991bb5f3d30db5d00c95ea11c5a56fbca8e89dfe726c864"
         }
       ],
       "stage": "supplemental",

scripts/probe-opentaint-product-v090.py

@@ -143,9 +143,12 @@ def _verify_v090_contract(root: Path, contract_path: Path, probe_script_path: Pa
     if not isinstance(identities, dict):
         raise ProbeError("contract input_identities must bind release inputs")
     control_ref = contract.get("control_inventory")
-    control_rel = "reports/releases/v0.9.0/execution-v1/control-inventory.json"
-    if not isinstance(control_ref, dict) or control_ref.get("path") != control_rel:
+    control_rel = control_ref.get('path') if isinstance(control_ref, dict) else None
+    if (not isinstance(control_rel, str) or '..' in Path(control_rel).parts or
+            not control_rel.startswith('reports/releases/v0.9.0/execution-v1/')):
         raise ProbeError("contract must bind the v0.9 control inventory")
+    if (root / control_rel).resolve().is_relative_to(root.resolve()) is not True:
+        raise ProbeError('control inventory escapes repository')
     control_inventory, control_inventory_raw = _read_json(root / control_rel, "v0.9 control inventory")
     control_sha = _require_digest(control_ref.get("sha256"), "control inventory SHA-256")
     if _sha_bytes(control_inventory_raw) != control_sha or identities.get(control_rel) != control_sha:
@@ -155,8 +158,13 @@ def _verify_v090_contract(root: Path, contract_path: Path, probe_script_path: Pa
     if len(control_rows) != 1:
         raise ProbeError("control inventory must register this probe exactly once")
     control_record = control_rows[0]
-    if control_record.get("argv") != ["/usr/bin/python3", SCRIPT_REL.as_posix()]:
-        raise ProbeError("control inventory must bind the exact versioned probe command")
+    relative_contract = contract_path.resolve().relative_to(root.resolve()).as_posix()
+    if not relative_contract.startswith('reports/releases/v0.9.0/execution-v1/'):
+        raise ProbeError('contract must be inside the versioned execution plan')
+    expected = ['/usr/bin/python3', SCRIPT_REL.as_posix(), '--contract', relative_contract]
+    legacy = ['/usr/bin/python3', SCRIPT_REL.as_posix()]
+    if control_record.get('argv') != expected and not (relative_contract == CONTRACT_REL.as_posix() and control_record.get('argv') == legacy):
+        raise ProbeError('control inventory must bind this exact contract path')
     script_rows = control_record.get("script_identity")
     if not isinstance(script_rows, list) or len(script_rows) != 1 or script_rows[0].get("path") != SCRIPT_REL.as_posix():
         raise ProbeError("control inventory must bind this probe's script identity")

scripts/probe-python-modeling-load-bearing-v090.py

@@ -53,7 +53,7 @@ def _bound_contract_input(root: Path, reference: Mapping[str, object], label: st
     return candidate
 
 
-def _check_control_environment(contract: Mapping[str, object], root: Path) -> dict[str, str]:
+def _check_control_environment(contract: Mapping[str, object], root: Path, contract_path: Path) -> dict[str, str]:
     reference = contract.get("control_inventory")
     if not isinstance(reference, dict):
         raise ValueError("v0.9 contract has no bound control inventory")
@@ -88,8 +88,13 @@ def _check_control_environment(contract: Mapping[str, object], root: Path) -> di
     ]:
         raise ValueError("Python modeling probe differs from control inventory script identity")
     argv = control.get("argv")
-    if not isinstance(argv, list) or not argv or argv[-1] != "scripts/probe-python-modeling-load-bearing-v090.py":
-        raise ValueError("Python modeling control inventory points at a different script")
+    relative_contract = contract_path.resolve().relative_to(root.resolve()).as_posix()
+    if not relative_contract.startswith('reports/releases/v0.9.0/execution-v1/'):
+        raise ValueError('contract must be inside the versioned execution plan')
+    expected = ['/usr/bin/python3', script_path, '--contract', relative_contract]
+    legacy = ['/usr/bin/python3', script_path]
+    if argv != expected and not (relative_contract == 'reports/releases/v0.9.0/execution-v1/contract.json' and argv == legacy):
+        raise ValueError('Python modeling argv must bind this exact contract path')
     roots = control.get("output_roots")
     scratch_relative = "reports/raw/control-scratch/probe-python-modeling-load-bearing"
     if not isinstance(roots, list) or scratch_relative not in roots or "reports/raw/load-bearing-python-modeling-v090" not in roots:
@@ -540,7 +545,7 @@ def capture(
         raise ValueError("execution authorization required: reviewed executable contract required")
 
     # Complete every authorization and identity check before creating output.
-    control_environment = _check_control_environment(contract, root)
+    control_environment = _check_control_environment(contract, root, contract_path)
     tools = _check_held_tool_digests(contract, contract_path, root)
     output_root.mkdir(parents=True, exist_ok=False)
     workspace = output_root / "workspace"

scripts/test-probe-opentaint-product-v090.py

@@ -139,6 +139,32 @@ def _json(path: Path, value: object) -> None:
 
 
 class ProductProbeV090Tests(unittest.TestCase):
+    def test_explicit_ready_contract_inventory_and_wrong_path_guard(self):
+        contract = json.loads(self.fx.contract_path.read_bytes())
+        self.fx._json(self.fx.contract_path, {**contract, 'execution_authorized': False})
+        original = self.fx.contract_path.read_bytes()
+        inventory = json.loads(self.fx.control_path.read_text())
+        ready_rel = Path('reports/releases/v0.9.0/execution-v1/final-01/contract.json')
+        inventory_rel = Path('reports/releases/v0.9.0/execution-v1/final-01/control-inventory.json')
+        inventory['controls'][0]['argv'] += ['--contract', ready_rel.as_posix()]
+        self.fx._json(self.fx.root/inventory_rel, inventory)
+        digest = sha((self.fx.root/inventory_rel).read_bytes())
+        contract['control_inventory'] = {'path': inventory_rel.as_posix(), 'sha256': digest}
+        contract['input_identities'][inventory_rel.as_posix()] = digest
+        self.fx._json(self.fx.root/ready_rel, contract)
+        self.assertEqual(MODULE.capture(self.fx.root, ready_rel, MODULE.OUTPUT_REL,
+                                       runner=self.successful_runner([])), 0)
+        self.assertEqual(self.fx.contract_path.read_bytes(), original)
+        inventory['controls'][0]['argv'][-1] = 'reports/releases/v0.9.0/execution-v1/other.json'
+        self.fx._json(self.fx.root/inventory_rel, inventory)
+        digest = sha((self.fx.root/inventory_rel).read_bytes())
+        contract['control_inventory']['sha256'] = digest
+        contract['input_identities'][inventory_rel.as_posix()] = digest
+        self.fx._json(self.fx.root/ready_rel, contract)
+        with self.assertRaisesRegex(MODULE.ProbeError, 'exact contract path'):
+            MODULE.capture(self.fx.root, ready_rel, 'reports/raw/rejected', runner=self.successful_runner([]))
+        self.assertFalse((self.fx.root/'reports/raw/rejected').exists())
+
     def setUp(self) -> None:
         self.temp = tempfile.TemporaryDirectory()
         self.addCleanup(self.temp.cleanup)

scripts/test-probe-python-modeling-load-bearing-v090.py

@@ -158,6 +158,32 @@ def prepared_root(root: Path) -> tuple[Path, Path, Path]:
 
 
 class PythonModelingV090Tests(unittest.TestCase):
+    def test_explicit_ready_contract_preserves_disabled_canonical_and_rejects_wrong_path(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            canonical, root, scratch = prepared_root(Path(temporary))
+            contract = json.loads(canonical.read_text())
+            write_json(canonical, {**contract, 'execution_authorized': False})
+            original = canonical.read_bytes()
+            inventory = json.loads((root/contract['control_inventory']['path']).read_text())
+            ready_rel = 'reports/releases/v0.9.0/execution-v1/final-01/contract.json'
+            inventory_rel = 'reports/releases/v0.9.0/execution-v1/final-01/control-inventory.json'
+            inventory['controls'][0]['argv'] += ['--contract', ready_rel]
+            digest = write_json(root/inventory_rel, inventory)
+            contract['control_inventory'] = {'path': inventory_rel, 'sha256': digest}
+            contract['input_identities'][inventory_rel] = digest
+            ready = root/ready_rel
+            write_json(ready, contract)
+            self.assertEqual(MODULE.capture(ready, root/'output', root=root, runner=lambda *a, **k: 0), 0)
+            self.assertEqual(canonical.read_bytes(), original)
+            inventory['controls'][0]['argv'][-1] = 'reports/releases/v0.9.0/execution-v1/other.json'
+            digest = write_json(root/inventory_rel, inventory)
+            contract['control_inventory']['sha256'] = digest
+            contract['input_identities'][inventory_rel] = digest
+            write_json(ready, contract)
+            with self.assertRaisesRegex(ValueError, 'exact contract path'):
+                MODULE.capture(ready, root/'rejected', root=root, runner=lambda *a, **k: 0)
+            self.assertFalse((root/'rejected').exists())
+
     def test_denied_execution_creates_no_output_and_never_calls_runner(self) -> None:
         with tempfile.TemporaryDirectory() as temporary:
             root = Path(temporary)