rshr_201_step_294_gate.py (10435B)
1 #!/usr/bin/env python3 2 """Emit the source-bound RSHR-201 gate result for Lib Step 294.""" 3 4 from __future__ import annotations 5 6 import argparse 7 import json 8 import sys 9 from pathlib import Path 10 11 import rshr_201_step_gate as step_292 12 13 14 ROOT = Path(__file__).resolve().parent.parent 15 AUTHORITY_PATH = ROOT / "contracts/rshr-201-step-294-gates.v1.json" 16 STEP = 294 17 GATE_DEFINITION = ( 18 "known-vulnerable fixtures, missing inventory, unavailable, stale, timeout, " 19 "and expired-suppression vectors" 20 ) 21 GATE_DIGEST = "d61e7135b7f468f32bc765be844399ade531649d85a4f29e57c37c83eced8bcf" 22 CHECK_ID = f"gate-01-{GATE_DIGEST}" 23 ASSERTION_ID = f"step_{STEP:03d}_gate_01_{GATE_DIGEST}" 24 EXPECTED_ARGV_TEMPLATE = [ 25 "cargo", 26 "extbuild", 27 "run", 28 "--", 29 "uv", 30 "run", 31 "--offline", 32 "--no-project", 33 "python3", 34 "-B", 35 "tools/rshr_201_step_294_gate.py", 36 "--step={step}", 37 "--check-id={check_id}", 38 "--source-revision={source_revision}", 39 "--source-tree={source_tree}", 40 "--candidate-digest={candidate_digest}", 41 "--platform=macos_aarch64", 42 "--execution-request-sha256={execution_request_sha256}", 43 ] 44 IMMUTABLE_STEP_293 = { 45 "contracts/rshr-201-step-293-gates.v1.json": ( 46 "456753ae166022205ee0f0f4722a4ea424adcf1f51f9852045749fd38b436c4e" 47 ), 48 "tools/rshr_201_step_293_gate.py": ( 49 "060ce66d435d7bb77bd7fb7c068e242c87bb4768819d85fbf45da324e1afac39" 50 ), 51 } 52 EXPECTED_UNIT_TESTS = { 53 "safe_artifact_io::step_294_tests": [ 54 "safe_artifact_io::step_294_tests::canonical_tar_gzip_is_exactly_reencoded_before_admission", 55 "safe_artifact_io::step_294_tests::materialization_retains_exact_member_and_parent_bindings", 56 ], 57 "advisory_snapshot::tests": [ 58 "advisory_snapshot::tests::expired_suppression_is_rejected", 59 "advisory_snapshot::tests::known_vulnerable_fixture", 60 "advisory_snapshot::tests::missing_inventory_is_rejected", 61 "advisory_snapshot::tests::stale_snapshot_is_rejected", 62 "advisory_snapshot::tests::timed_out_operation_is_nonpass", 63 "advisory_snapshot::tests::unavailable_provider_is_nonpass", 64 ], 65 } 66 EXPECTED_DECISION_TESTS = [ 67 "authority_scope_and_complete_file_are_exact", 68 "freshness_suppressions_results_and_required_vectors_are_exact", 69 "immutable_archives_reports_and_execution_bounds_are_exact", 70 "workload_inventory_and_tool_pins_are_exact", 71 ] 72 73 74 def run_cargo(arguments: list[str], *, label: str) -> bytes: 75 return step_292.run( 76 ["cargo", "+1.97.1", *arguments], 77 step_292.gate_environment(), 78 label=label, 79 ) 80 81 82 def require_listed_tests(output: bytes, expected: list[str], *, label: str) -> None: 83 try: 84 lines = output.decode("utf-8", "strict").splitlines() 85 except UnicodeError as error: 86 raise step_292.GateError(f"{label} inventory is not UTF-8") from error 87 observed = sorted( 88 line.removesuffix(": test") for line in lines if line.endswith(": test") 89 ) 90 if observed != sorted(expected): 91 raise step_292.GateError(f"{label} inventory differs") 92 93 94 def run_unit_test_lane(test_filter: str, expected: list[str], *, label: str) -> None: 95 base = [ 96 "test", 97 "--offline", 98 "--manifest-path", 99 "tools/xtask/Cargo.toml", 100 "--locked", 101 "--bin", 102 "xtask", 103 test_filter, 104 ] 105 listed = run_cargo( 106 [*base, "--", "--list", "--format=terse"], 107 label=f"{label} inventory", 108 ) 109 require_listed_tests(listed, expected, label=label) 110 run_cargo([*base, "--", "--test-threads=1"], label=label) 111 112 113 def run_step() -> None: 114 run_cargo(["fmt", "--all", "--", "--check"], label="Step 294 formatting check") 115 run_cargo( 116 [ 117 "clippy", 118 "--offline", 119 "--manifest-path", 120 "tools/xtask/Cargo.toml", 121 "--locked", 122 "--all-targets", 123 "--", 124 "-D", 125 "warnings", 126 ], 127 label="Step 294 lint check", 128 ) 129 for test_filter, label in [ 130 ("safe_artifact_io::step_294_tests", "deterministic archive vectors"), 131 ("advisory_snapshot::tests", "advisory snapshot vectors"), 132 ]: 133 run_unit_test_lane( 134 test_filter, 135 EXPECTED_UNIT_TESTS[test_filter], 136 label=label, 137 ) 138 run_cargo( 139 [ 140 "run", 141 "--offline", 142 "--manifest-path", 143 "tools/xtask/Cargo.toml", 144 "--locked", 145 "--", 146 "advisory-snapshot-self-test", 147 ], 148 label="advisory snapshot self-test", 149 ) 150 decision_base = [ 151 "test", 152 "--offline", 153 "--manifest-path", 154 "tools/xtask/Cargo.toml", 155 "--locked", 156 "--test", 157 "services_hardening_advisory_snapshot_decision", 158 ] 159 listed = run_cargo( 160 [*decision_base, "--", "--list", "--format=terse"], 161 label="advisory snapshot decision contract inventory", 162 ) 163 require_listed_tests( 164 listed, 165 EXPECTED_DECISION_TESTS, 166 label="advisory snapshot decision contract", 167 ) 168 run_cargo( 169 [*decision_base, "--", "--test-threads=1"], 170 label="advisory snapshot decision contract", 171 ) 172 if step_292.git_bytes("status", "--porcelain=v1", "-z", "--untracked-files=all"): 173 raise step_292.GateError("verification changed the tracked or untracked source state") 174 175 176 def parse_arguments() -> argparse.Namespace: 177 parser = step_292.RedactedArgumentParser(allow_abbrev=False) 178 parser.add_argument("--step", type=int, required=True) 179 parser.add_argument("--check-id") 180 parser.add_argument("--source-revision", required=True) 181 parser.add_argument("--source-tree", required=True) 182 parser.add_argument("--candidate-digest") 183 parser.add_argument("--platform", required=True) 184 parser.add_argument("--execution-request-sha256", required=True) 185 return parser.parse_args() 186 187 188 def expected_contract(verifier_digest: str) -> dict[str, object]: 189 return { 190 "argv_template": EXPECTED_ARGV_TEMPLATE, 191 "assertion_id": [ASSERTION_ID], 192 "check_id": CHECK_ID, 193 "environment_authority": step_292.EXPECTED_ENVIRONMENT_AUTHORITY, 194 "environment_names": step_292.EXPECTED_ENVIRONMENT_NAMES, 195 "gate_definition_sha256": GATE_DIGEST, 196 "required_platforms": ["macos_aarch64"], 197 "required_tools": ["uv", "python3", "git"], 198 "result_schema": "radroots.services-hardening.rshr-200-step-check-result.v1", 199 "schema": "radroots.services-hardening.rshr-200-step-check-command.v1", 200 "step": STEP, 201 "verifier_path": "tools/rshr_201_step_294_gate.py", 202 "verifier_sha256": verifier_digest, 203 } 204 205 206 def require_immutable_step_293() -> None: 207 for relative, expected in IMMUTABLE_STEP_293.items(): 208 if step_292.sha256_bytes(step_292.read_regular(ROOT / relative)) != expected: 209 raise step_292.GateError("Step 293 immutable authority differs") 210 211 212 def main() -> int: 213 arguments = parse_arguments() 214 if arguments.step != STEP: 215 raise step_292.GateError("step is outside the Lib gate authority") 216 step_292.validate_digest(arguments.source_revision, "source revision", 40) 217 step_292.validate_digest(arguments.source_tree, "source tree", 40) 218 step_292.validate_digest(arguments.execution_request_sha256, "execution request", 64) 219 if step_292.sha256_bytes(GATE_DEFINITION.encode("utf-8")) != GATE_DIGEST: 220 raise step_292.GateError("compiled gate definition digest differs") 221 if arguments.check_id != CHECK_ID: 222 raise step_292.GateError("check identity differs") 223 if arguments.candidate_digest != "none" or arguments.platform != "macos_aarch64": 224 raise step_292.GateError("candidate or platform scope differs") 225 226 require_immutable_step_293() 227 authority_bytes = step_292.read_regular(AUTHORITY_PATH, 256 * 1024) 228 try: 229 authority = json.loads(authority_bytes) 230 except (UnicodeError, json.JSONDecodeError) as error: 231 raise step_292.GateError("gate authority is not canonical JSON") from error 232 if step_292.canonical(authority) + b"\n" != authority_bytes: 233 raise step_292.GateError("gate authority is not canonical JSON") 234 if ( 235 not isinstance(authority, dict) 236 or set(authority) != {"schema", "step", "gate_command_contract"} 237 or authority.get("schema") != "radroots.lib.rshr-201-step-294-gates.v1" 238 or authority.get("step") != [STEP] 239 ): 240 raise step_292.GateError("gate authority step inventory differs") 241 contracts = authority.get("gate_command_contract") 242 if not isinstance(contracts, list) or len(contracts) != 1: 243 raise step_292.GateError("gate command authority is absent or duplicated") 244 245 verifier_digest = step_292.sha256_bytes(step_292.read_regular(Path(__file__).resolve())) 246 contract = contracts[0] 247 if contract != expected_contract(verifier_digest): 248 raise step_292.GateError("gate command authority differs from source bytes") 249 250 step_292.require_source_state(arguments.source_revision, arguments.source_tree) 251 run_step() 252 assertions = [{"id": ASSERTION_ID, "result": "pass"}] 253 result = { 254 "schema": "radroots.services-hardening.rshr-200-step-check-result.v1", 255 "step": STEP, 256 "check_id": CHECK_ID, 257 "gate_definition_sha256": GATE_DIGEST, 258 "source_revision": arguments.source_revision, 259 "source_tree": arguments.source_tree, 260 "candidate_generation": 0, 261 "candidate_digest": "none", 262 "command_contract_sha256": step_292.sha256_bytes(step_292.canonical(contract)), 263 "verifier_sha256": verifier_digest, 264 "execution_request": [ 265 {"platform": arguments.platform, "sha256": arguments.execution_request_sha256} 266 ], 267 "assertion_inventory_sha256": step_292.sha256_bytes( 268 step_292.canonical(assertions) 269 ), 270 "assertion": assertions, 271 "result": "pass", 272 } 273 sys.stdout.buffer.write(step_292.canonical(result) + b"\n") 274 return 0 275 276 277 if __name__ == "__main__": 278 try: 279 raise SystemExit(main()) 280 except step_292.GateError as error: 281 print(f"Lib RSHR-201 Step 294 gate failed: {error}", file=sys.stderr) 282 raise SystemExit(1) 283 except Exception: 284 print("Lib RSHR-201 Step 294 gate failed safely", file=sys.stderr) 285 raise SystemExit(1)