rshr_201_step_293_gate.py (13457B)
1 #!/usr/bin/env python3 2 """Emit the source-bound RSHR-201 gate result for Lib Step 293.""" 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-293-gates.v1.json" 16 STEP = 293 17 GATE_DEFINITION = ( 18 "symlink, FIFO, replacement, archive-bomb, and " 19 "missing-failed-skipped lane vectors" 20 ) 21 GATE_DIGEST = "84d8022c579c03ed4192b0cd579bf105abf3f3e816cdafc3a53f53bc68460baf" 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_293_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_292 = { 45 "contracts/rshr-201-step-gates.v1.json": ( 46 "d787d9980e8fdcedbdf2fdf43e4eddf05220714598dfc44ec8d049acf05efd6c" 47 ), 48 "tools/rshr_201_step_gate.py": ( 49 "0ceefa6f116993106ceb94dc4ca9f64cc3c6de85b54a9a7aff327cef92e34b37" 50 ), 51 } 52 EXPECTED_UNIT_TESTS = { 53 "safe_artifact_io::tests": [ 54 "safe_artifact_io::tests::archive_rejects_duplicates_prefix_conflicts_and_concatenation", 55 "safe_artifact_io::tests::bounded_read_hash_and_copy_are_streaming_and_exact", 56 "safe_artifact_io::tests::diagnostics_are_redacted", 57 "safe_artifact_io::tests::hard_maximums_reject_invalid_requests", 58 "safe_artifact_io::tests::hardlinked_regular_inputs_are_rejected", 59 "safe_artifact_io::tests::no_follow_admission_rejects_symlink_fifo_and_replacement", 60 "safe_artifact_io::tests::tar_gzip_admission_is_parse_only_and_bounded", 61 "safe_artifact_io::tests::traversal_enforces_type_count_byte_and_depth_bounds", 62 ], 63 "release_preflight::tests": [ 64 "release_preflight::tests::aggregate_diagnostics_are_static", 65 "release_preflight::tests::every_nonpass_state_fails_closed_without_short_circuiting", 66 "release_preflight::tests::exact_inventory_and_order_are_closed", 67 "release_preflight::tests::missing_duplicate_and_unexpected_lanes_fail_closed", 68 ], 69 "service_release_artifacts::tests": [ 70 "service_release_artifacts::tests::absent_nix_material_is_preserved_without_invented_digest_evidence", 71 "service_release_artifacts::tests::binary_archive_is_reproducible_and_metadata_is_fixed", 72 "service_release_artifacts::tests::contract_matches_the_checked_in_decision", 73 "service_release_artifacts::tests::contract_rejects_every_independent_governed_field_drift", 74 "service_release_artifacts::tests::errors_are_stable_and_source_free", 75 "service_release_artifacts::tests::exact_inventory_and_limits_are_literal", 76 "service_release_artifacts::tests::file_admission_predicates_reject_each_independent_drift", 77 "service_release_artifacts::tests::full_artifact_set_is_reproducible_immutable_and_verifiable", 78 "service_release_artifacts::tests::identifier_predicates_reject_each_independent_boundary", 79 "service_release_artifacts::tests::identifiers_and_sources_are_closed", 80 "service_release_artifacts::tests::low_level_release_admission_and_comparison_branches_are_qualified", 81 "service_release_artifacts::tests::output_name_is_one_bounded_component", 82 "service_release_artifacts::tests::output_scope_and_target_admission_are_fail_closed", 83 "service_release_artifacts::tests::output_scope_remote_and_inventory_reject_each_drift", 84 "service_release_artifacts::tests::package_metadata_rejects_each_independent_field_drift", 85 "service_release_artifacts::tests::private_or_incomplete_dependency_evidence_is_rejected", 86 "service_release_artifacts::tests::protected_text_and_invalid_inventory_fail_closed", 87 "service_release_artifacts::tests::release_metadata_and_text_admission_reject_each_field_drift", 88 "service_release_artifacts::tests::release_service_and_workspace_binding_fail_closed", 89 "service_release_artifacts::tests::remaining_release_boundaries_fail_closed", 90 "service_release_artifacts::tests::secret_scanner_detects_a_pattern_across_chunk_boundaries", 91 "service_release_artifacts::tests::snapshot_backed_inventory_rejects_file_and_root_mode_races", 92 "service_release_artifacts::tests::source_lock_and_source_bundle_drift_fail_closed", 93 "service_release_artifacts::tests::supply_chain_documents_are_deterministic_and_complete", 94 "service_release_artifacts::tests::supply_chain_graph_rejects_each_independent_structural_drift", 95 ], 96 } 97 EXPECTED_DECISION_TESTS = [ 98 "aggregate_inventory_nonclaims_and_vectors_are_exact", 99 "filesystem_traversal_copy_and_archive_models_are_exact", 100 "scope_platform_and_hard_maximums_are_exact", 101 ] 102 103 104 def run_cargo(arguments: list[str], *, label: str) -> bytes: 105 return step_292.run( 106 [ 107 "cargo", 108 "+1.97.1", 109 *arguments, 110 ], 111 step_292.gate_environment(), 112 label=label, 113 ) 114 115 116 def require_listed_tests(output: bytes, expected: list[str], *, label: str) -> None: 117 try: 118 lines = output.decode("utf-8", "strict").splitlines() 119 except UnicodeError as error: 120 raise step_292.GateError(f"{label} inventory is not UTF-8") from error 121 observed = sorted( 122 line.removesuffix(": test") for line in lines if line.endswith(": test") 123 ) 124 if observed != sorted(expected): 125 raise step_292.GateError(f"{label} inventory differs") 126 127 128 def run_unit_test_lane(test_filter: str, expected: list[str], *, label: str) -> None: 129 base = [ 130 "test", 131 "--offline", 132 "--manifest-path", 133 "tools/xtask/Cargo.toml", 134 "--locked", 135 "--bin", 136 "xtask", 137 test_filter, 138 ] 139 listed = run_cargo( 140 [*base, "--", "--list", "--format=terse"], 141 label=f"{label} inventory", 142 ) 143 require_listed_tests(listed, expected, label=label) 144 run_cargo( 145 [*base, "--", "--test-threads=1"], 146 label=label, 147 ) 148 149 150 def run_step() -> None: 151 run_cargo( 152 [ 153 "fmt", 154 "--all", 155 "--", 156 "--check", 157 ], 158 label="Step 293 formatting check", 159 ) 160 run_cargo( 161 [ 162 "clippy", 163 "--offline", 164 "--manifest-path", 165 "tools/xtask/Cargo.toml", 166 "--locked", 167 "--all-targets", 168 "--", 169 "-D", 170 "warnings", 171 ], 172 label="Step 293 lint check", 173 ) 174 for test_filter, label in [ 175 ("safe_artifact_io::tests", "safe-artifact I/O vectors"), 176 ("release_preflight::tests", "release-preflight aggregate vectors"), 177 ("service_release_artifacts::tests", "release-artifact adoption vectors"), 178 ]: 179 run_unit_test_lane( 180 test_filter, 181 EXPECTED_UNIT_TESTS[test_filter], 182 label=label, 183 ) 184 run_cargo( 185 [ 186 "run", 187 "--offline", 188 "--manifest-path", 189 "tools/xtask/Cargo.toml", 190 "--locked", 191 "--", 192 "safe-artifact-io-self-test", 193 ], 194 label="safe-artifact and aggregate self-test", 195 ) 196 decision_base = [ 197 "test", 198 "--offline", 199 "--manifest-path", 200 "tools/xtask/Cargo.toml", 201 "--locked", 202 "--test", 203 "services_hardening_safe_artifact_io_decision", 204 ] 205 listed = run_cargo( 206 [*decision_base, "--", "--list", "--format=terse"], 207 label="safe-artifact I/O decision contract inventory", 208 ) 209 require_listed_tests( 210 listed, 211 EXPECTED_DECISION_TESTS, 212 label="safe-artifact I/O decision contract", 213 ) 214 run_cargo( 215 [*decision_base, "--", "--test-threads=1"], 216 label="safe-artifact I/O decision contract", 217 ) 218 if step_292.git_bytes("status", "--porcelain=v1", "-z", "--untracked-files=all"): 219 raise step_292.GateError("verification changed the tracked or untracked source state") 220 221 222 def parse_arguments() -> argparse.Namespace: 223 parser = step_292.RedactedArgumentParser(allow_abbrev=False) 224 parser.add_argument("--step", type=int, required=True) 225 parser.add_argument("--check-id") 226 parser.add_argument("--source-revision", required=True) 227 parser.add_argument("--source-tree", required=True) 228 parser.add_argument("--candidate-digest") 229 parser.add_argument("--platform", required=True) 230 parser.add_argument("--execution-request-sha256", required=True) 231 return parser.parse_args() 232 233 234 def expected_contract(verifier_digest: str) -> dict[str, object]: 235 return { 236 "argv_template": EXPECTED_ARGV_TEMPLATE, 237 "assertion_id": [ASSERTION_ID], 238 "check_id": CHECK_ID, 239 "environment_authority": step_292.EXPECTED_ENVIRONMENT_AUTHORITY, 240 "environment_names": step_292.EXPECTED_ENVIRONMENT_NAMES, 241 "gate_definition_sha256": GATE_DIGEST, 242 "required_platforms": ["macos_aarch64"], 243 "required_tools": ["uv", "python3", "git"], 244 "result_schema": "radroots.services-hardening.rshr-200-step-check-result.v1", 245 "schema": "radroots.services-hardening.rshr-200-step-check-command.v1", 246 "step": STEP, 247 "verifier_path": "tools/rshr_201_step_293_gate.py", 248 "verifier_sha256": verifier_digest, 249 } 250 251 252 def require_immutable_step_292() -> None: 253 for relative, expected in IMMUTABLE_STEP_292.items(): 254 if step_292.sha256_bytes(step_292.read_regular(ROOT / relative)) != expected: 255 raise step_292.GateError("Step 292 immutable authority differs") 256 257 258 def main() -> int: 259 arguments = parse_arguments() 260 if arguments.step != STEP: 261 raise step_292.GateError("step is outside the Lib gate authority") 262 step_292.validate_digest(arguments.source_revision, "source revision", 40) 263 step_292.validate_digest(arguments.source_tree, "source tree", 40) 264 step_292.validate_digest(arguments.execution_request_sha256, "execution request", 64) 265 if step_292.sha256_bytes(GATE_DEFINITION.encode("utf-8")) != GATE_DIGEST: 266 raise step_292.GateError("compiled gate definition digest differs") 267 if arguments.check_id != CHECK_ID: 268 raise step_292.GateError("check identity differs") 269 if arguments.candidate_digest != "none" or arguments.platform != "macos_aarch64": 270 raise step_292.GateError("candidate or platform scope differs") 271 272 require_immutable_step_292() 273 authority_bytes = step_292.read_regular(AUTHORITY_PATH, 256 * 1024) 274 try: 275 authority = json.loads(authority_bytes) 276 except (UnicodeError, json.JSONDecodeError) as error: 277 raise step_292.GateError("gate authority is not canonical JSON") from error 278 if step_292.canonical(authority) + b"\n" != authority_bytes: 279 raise step_292.GateError("gate authority is not canonical JSON") 280 if ( 281 not isinstance(authority, dict) 282 or set(authority) != {"schema", "step", "gate_command_contract"} 283 or authority.get("schema") != "radroots.lib.rshr-201-step-293-gates.v1" 284 or authority.get("step") != [STEP] 285 ): 286 raise step_292.GateError("gate authority step inventory differs") 287 contracts = authority.get("gate_command_contract") 288 if not isinstance(contracts, list) or len(contracts) != 1: 289 raise step_292.GateError("gate command authority is absent or duplicated") 290 291 verifier_digest = step_292.sha256_bytes(step_292.read_regular(Path(__file__).resolve())) 292 contract = contracts[0] 293 if contract != expected_contract(verifier_digest): 294 raise step_292.GateError("gate command authority differs from source bytes") 295 296 step_292.require_source_state(arguments.source_revision, arguments.source_tree) 297 run_step() 298 assertions = [{"id": ASSERTION_ID, "result": "pass"}] 299 result = { 300 "schema": "radroots.services-hardening.rshr-200-step-check-result.v1", 301 "step": STEP, 302 "check_id": CHECK_ID, 303 "gate_definition_sha256": GATE_DIGEST, 304 "source_revision": arguments.source_revision, 305 "source_tree": arguments.source_tree, 306 "candidate_generation": 0, 307 "candidate_digest": "none", 308 "command_contract_sha256": step_292.sha256_bytes(step_292.canonical(contract)), 309 "verifier_sha256": verifier_digest, 310 "execution_request": [ 311 { 312 "platform": arguments.platform, 313 "sha256": arguments.execution_request_sha256, 314 } 315 ], 316 "assertion_inventory_sha256": step_292.sha256_bytes( 317 step_292.canonical(assertions) 318 ), 319 "assertion": assertions, 320 "result": "pass", 321 } 322 sys.stdout.buffer.write(step_292.canonical(result) + b"\n") 323 return 0 324 325 326 if __name__ == "__main__": 327 try: 328 raise SystemExit(main()) 329 except step_292.GateError as error: 330 print(f"Lib RSHR-201 Step 293 gate failed: {error}", file=sys.stderr) 331 raise SystemExit(1) 332 except Exception: 333 print("Lib RSHR-201 Step 293 gate failed safely", file=sys.stderr) 334 raise SystemExit(1)