commit a47a923ebe8a589f5ca996646776e3508bcc94bd
parent d00b35c0e3d84b87f140ac2fb3080be96c409dc9
Author: triesap <tyson@radroots.org>
Date: Sat, 5 Sep 2026 02:15:30 +0000
contracts: bind RSHR 201 process gate
Diffstat:
2 files changed, 626 insertions(+), 0 deletions(-)
diff --git a/contracts/rshr-201-step-gates.v1.json b/contracts/rshr-201-step-gates.v1.json
@@ -0,0 +1 @@
+{"gate_command_contract":[{"argv_template":["cargo","extbuild","run","--","uv","run","--offline","--no-project","python3","-B","tools/rshr_201_step_gate.py","--step={step}","--check-id={check_id}","--source-revision={source_revision}","--source-tree={source_tree}","--candidate-digest={candidate_digest}","--platform=macos_aarch64","--execution-request-sha256={execution_request_sha256}"],"assertion_id":["step_292_gate_01_f65be73a73a7ab8b0c8e02f0695ee5e910c73e0017572b82861c0c7f0d4fa454"],"check_id":"gate-01-f65be73a73a7ab8b0c8e02f0695ee5e910c73e0017572b82861c0c7f0d4fa454","environment_authority":{"cache_policy_id":"rshr-200-step-287-cache-policy.v1","cache_policy_sha256":"3e81d178bce97b6c349dfbb00c68fd6f620ac00b1a1c8d37b12e9998f3c9eaaa","cadence_policy_id":"rshr-200-step-287-cadence-policy.v1","cadence_policy_sha256":"d24903df8659ee3772297c84994911efe7d21cb8b988320ddc6ddce0431892a1","isolation":"extbuild_host_constrained","network":"disabled","network_policy_id":"none","network_policy_sha256":"none","resource_policy_id":"rshr-200-step-287-resource-policy.v1","resource_policy_sha256":"05d3c7a89185d3c55678d97955193fce2ed92b1eee5af99083d77ea64c98d14e"},"environment_names":["EXT_BUILD_CONFIG","EXT_BUILD_MACHINE_CONFIG","EXT_BUILD_ROOT","HOME","PATH","RUSTUP_TOOLCHAIN","TMPDIR"],"gate_definition_sha256":"f65be73a73a7ab8b0c8e02f0695ee5e910c73e0017572b82861c0c7f0d4fa454","required_platforms":["macos_aarch64"],"required_tools":["uv","python3","git"],"result_schema":"radroots.services-hardening.rshr-200-step-check-result.v1","schema":"radroots.services-hardening.rshr-200-step-check-command.v1","step":292,"verifier_path":"tools/rshr_201_step_gate.py","verifier_sha256":"0ceefa6f116993106ceb94dc4ca9f64cc3c6de85b54a9a7aff327cef92e34b37"}],"schema":"radroots.lib.rshr-201-step-gates.v1","step":[292]}
diff --git a/tools/rshr_201_step_gate.py b/tools/rshr_201_step_gate.py
@@ -0,0 +1,625 @@
+#!/usr/bin/env python3
+"""Emit the source-bound RSHR-201 gate result for Lib Step 292."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import re
+import selectors
+import signal
+import stat
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parent.parent
+AUTHORITY_PATH = ROOT / "contracts/rshr-201-step-gates.v1.json"
+ORIGIN = "ssh://git@github.com/radrootslabs/lib.git"
+BRANCH = "rshr/rcld-201"
+STEP = 292
+GATE_DEFINITION = (
+ "timeout, orphan, output-cap, inherited-build-environment, "
+ "loader-injection, and redaction vectors"
+)
+GATE_DIGEST = "f65be73a73a7ab8b0c8e02f0695ee5e910c73e0017572b82861c0c7f0d4fa454"
+CHECK_ID = f"gate-01-{GATE_DIGEST}"
+ASSERTION_ID = f"step_{STEP:03d}_gate_01_{GATE_DIGEST}"
+MAX_SOURCE_BYTES = 64 * 1024 * 1024
+MAX_STDOUT_BYTES = 16 * 1024 * 1024
+MAX_STDERR_BYTES = 16 * 1024 * 1024
+COMMAND_TIMEOUT_SECONDS = 3600.0
+STREAM_CHUNK_BYTES = 64 * 1024
+SELECT_INTERVAL_SECONDS = 0.1
+TERM_GRACE_SECONDS = 0.25
+KILL_GRACE_SECONDS = 0.5
+EXPECTED_ENVIRONMENT_AUTHORITY = {
+ "cache_policy_id": "rshr-200-step-287-cache-policy.v1",
+ "cache_policy_sha256": (
+ "3e81d178bce97b6c349dfbb00c68fd6f620ac00b1a1c8d37b12e9998f3c9eaaa"
+ ),
+ "cadence_policy_id": "rshr-200-step-287-cadence-policy.v1",
+ "cadence_policy_sha256": (
+ "d24903df8659ee3772297c84994911efe7d21cb8b988320ddc6ddce0431892a1"
+ ),
+ "isolation": "extbuild_host_constrained",
+ "network": "disabled",
+ "network_policy_id": "none",
+ "network_policy_sha256": "none",
+ "resource_policy_id": "rshr-200-step-287-resource-policy.v1",
+ "resource_policy_sha256": (
+ "05d3c7a89185d3c55678d97955193fce2ed92b1eee5af99083d77ea64c98d14e"
+ ),
+}
+EXPECTED_ENVIRONMENT_NAMES = [
+ "EXT_BUILD_CONFIG",
+ "EXT_BUILD_MACHINE_CONFIG",
+ "EXT_BUILD_ROOT",
+ "HOME",
+ "PATH",
+ "RUSTUP_TOOLCHAIN",
+ "TMPDIR",
+]
+EXPECTED_ARGV_TEMPLATE = [
+ "cargo",
+ "extbuild",
+ "run",
+ "--",
+ "uv",
+ "run",
+ "--offline",
+ "--no-project",
+ "python3",
+ "-B",
+ "tools/rshr_201_step_gate.py",
+ "--step={step}",
+ "--check-id={check_id}",
+ "--source-revision={source_revision}",
+ "--source-tree={source_tree}",
+ "--candidate-digest={candidate_digest}",
+ "--platform=macos_aarch64",
+ "--execution-request-sha256={execution_request_sha256}",
+]
+
+
+class GateError(RuntimeError):
+ """A fail-closed gate error whose message contains no protected data."""
+
+
+class RedactedArgumentParser(argparse.ArgumentParser):
+ """Reject malformed input without reflecting argument values."""
+
+ def error(self, _message: str) -> None:
+ raise GateError("arguments are invalid")
+
+
+def canonical(value: object) -> bytes:
+ return json.dumps(
+ value,
+ sort_keys=True,
+ separators=(",", ":"),
+ ensure_ascii=False,
+ allow_nan=False,
+ ).encode("utf-8")
+
+
+def sha256_bytes(contents: bytes) -> str:
+ return hashlib.sha256(contents).hexdigest()
+
+
+def read_regular(path: Path, maximum: int = MAX_SOURCE_BYTES) -> bytes:
+ try:
+ relative = path.relative_to(ROOT)
+ except ValueError as error:
+ raise GateError("source path is outside the repository") from error
+ if maximum < 0 or path.is_symlink():
+ raise GateError(f"source path is not a bounded regular file: {relative}")
+ try:
+ descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
+ except OSError as error:
+ raise GateError(f"source path cannot be opened safely: {relative}") from error
+ try:
+ metadata = os.fstat(descriptor)
+ if not stat.S_ISREG(metadata.st_mode) or metadata.st_size > maximum:
+ raise GateError(f"source path is not a bounded regular file: {relative}")
+ contents = bytearray()
+ while len(contents) <= maximum:
+ chunk = os.read(
+ descriptor,
+ min(STREAM_CHUNK_BYTES, maximum - len(contents) + 1),
+ )
+ if not chunk:
+ break
+ contents.extend(chunk)
+ if len(contents) > maximum:
+ raise GateError(f"source path exceeds its byte bound: {relative}")
+ after = os.fstat(descriptor)
+ if (
+ metadata.st_dev,
+ metadata.st_ino,
+ metadata.st_size,
+ metadata.st_mtime_ns,
+ ) != (
+ after.st_dev,
+ after.st_ino,
+ after.st_size,
+ after.st_mtime_ns,
+ ):
+ raise GateError(f"source path changed during its bounded read: {relative}")
+ return bytes(contents)
+ except OSError as error:
+ raise GateError(f"source path cannot be read safely: {relative}") from error
+ finally:
+ os.close(descriptor)
+
+
+def _process_group_exists(process_group: int) -> bool:
+ try:
+ os.killpg(process_group, 0)
+ except ProcessLookupError:
+ return False
+ except PermissionError:
+ return True
+ return True
+
+
+def _signal_process_group(
+ process: subprocess.Popen[bytes], process_signal: signal.Signals
+) -> None:
+ try:
+ os.killpg(process.pid, process_signal)
+ except ProcessLookupError:
+ return
+ except OSError:
+ try:
+ process.send_signal(process_signal)
+ except OSError:
+ return
+
+
+def _bounded_reap(process: subprocess.Popen[bytes], timeout: float) -> None:
+ if process.poll() is not None:
+ return
+ try:
+ process.wait(timeout=timeout)
+ except (OSError, subprocess.TimeoutExpired):
+ return
+
+
+def _terminate_process_group(process: subprocess.Popen[bytes]) -> None:
+ """Apply bounded TERM/KILL cleanup to the isolated process group."""
+
+ _signal_process_group(process, signal.SIGTERM)
+ term_deadline = time.monotonic() + TERM_GRACE_SECONDS
+ while _process_group_exists(process.pid) and time.monotonic() < term_deadline:
+ _bounded_reap(
+ process,
+ min(SELECT_INTERVAL_SECONDS, max(0.001, term_deadline - time.monotonic())),
+ )
+ if process.poll() is not None:
+ time.sleep(min(0.01, max(0.0, term_deadline - time.monotonic())))
+ if _process_group_exists(process.pid):
+ _signal_process_group(process, signal.SIGKILL)
+ kill_deadline = time.monotonic() + KILL_GRACE_SECONDS
+ while _process_group_exists(process.pid) and time.monotonic() < kill_deadline:
+ _bounded_reap(
+ process,
+ min(SELECT_INTERVAL_SECONDS, max(0.001, kill_deadline - time.monotonic())),
+ )
+ if process.poll() is not None:
+ time.sleep(min(0.01, max(0.0, kill_deadline - time.monotonic())))
+ _bounded_reap(process, max(0.001, kill_deadline - time.monotonic()))
+ if process.poll() is None or _process_group_exists(process.pid):
+ raise GateError("bounded command cleanup did not reach a terminal state")
+
+
+def run(
+ arguments: list[str],
+ environment: dict[str, str],
+ *,
+ label: str,
+ maximum_stdout: int = MAX_STDOUT_BYTES,
+ maximum_stderr: int = MAX_STDERR_BYTES,
+ timeout_seconds: float = COMMAND_TIMEOUT_SECONDS,
+) -> bytes:
+ """Run one isolated command with live dual-stream caps and redacted errors."""
+
+ if (
+ not arguments
+ or not label
+ or maximum_stdout < 0
+ or maximum_stderr < 0
+ or timeout_seconds <= 0
+ or any(
+ not isinstance(argument, str)
+ or not argument
+ or "\x00" in argument
+ or "\r" in argument
+ or "\n" in argument
+ for argument in arguments
+ )
+ or any(
+ not isinstance(name, str)
+ or not name
+ or "=" in name
+ or "\x00" in name
+ or not isinstance(value, str)
+ or "\x00" in value
+ for name, value in environment.items()
+ )
+ ):
+ raise GateError(f"{label} request is invalid")
+
+ process: subprocess.Popen[bytes] | None = None
+ selector = selectors.DefaultSelector()
+ streams: list[object] = []
+ try:
+ try:
+ process = subprocess.Popen(
+ arguments,
+ cwd=ROOT,
+ env=dict(environment),
+ stdin=subprocess.DEVNULL,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ close_fds=True,
+ start_new_session=True,
+ )
+ except OSError as error:
+ raise GateError(f"{label} could not start") from error
+
+ assert process.stdout is not None
+ assert process.stderr is not None
+ streams.extend((process.stdout, process.stderr))
+ destinations = {
+ process.stdout: ("stdout", maximum_stdout, bytearray()),
+ process.stderr: ("stderr", maximum_stderr, bytearray()),
+ }
+ for stream in destinations:
+ os.set_blocking(stream.fileno(), False)
+ selector.register(stream, selectors.EVENT_READ)
+
+ deadline = time.monotonic() + timeout_seconds
+ while selector.get_map():
+ remaining = deadline - time.monotonic()
+ if remaining <= 0:
+ raise GateError(f"{label} exceeded its deadline")
+ if process.poll() is not None and _process_group_exists(process.pid):
+ raise GateError(f"{label} left a running process group")
+ events = selector.select(min(SELECT_INTERVAL_SECONDS, remaining))
+ if not events:
+ if process.poll() is not None and _process_group_exists(process.pid):
+ raise GateError(f"{label} left a running process group")
+ continue
+ for key, _mask in events:
+ stream = key.fileobj
+ stream_name, maximum, destination = destinations[stream]
+ read_size = min(
+ STREAM_CHUNK_BYTES,
+ max(1, maximum - len(destination) + 1),
+ )
+ try:
+ chunk = os.read(stream.fileno(), read_size)
+ except BlockingIOError:
+ continue
+ except OSError as error:
+ raise GateError(f"{label} stream read failed") from error
+ if not chunk:
+ selector.unregister(stream)
+ continue
+ destination.extend(chunk)
+ if len(destination) > maximum:
+ raise GateError(f"{label} {stream_name} exceeded its byte bound")
+
+ remaining = deadline - time.monotonic()
+ if remaining <= 0:
+ raise GateError(f"{label} exceeded its deadline")
+ try:
+ return_code = process.wait(timeout=remaining)
+ except subprocess.TimeoutExpired as error:
+ raise GateError(f"{label} exceeded its deadline") from error
+
+ if _process_group_exists(process.pid):
+ raise GateError(f"{label} left a running process group")
+ if return_code != 0:
+ raise GateError(f"{label} failed with exit status {return_code}")
+ return bytes(destinations[process.stdout][2])
+ except GateError:
+ if process is not None:
+ _terminate_process_group(process)
+ raise
+ except (OSError, subprocess.SubprocessError) as error:
+ if process is not None:
+ _terminate_process_group(process)
+ raise GateError(f"{label} failed safely") from error
+ except BaseException:
+ if process is not None:
+ _terminate_process_group(process)
+ raise
+ finally:
+ selector.close()
+ for stream in streams:
+ try:
+ stream.close() # type: ignore[attr-defined]
+ except OSError:
+ pass
+
+
+def git_environment() -> dict[str, str]:
+ return {
+ "GIT_ATTR_NOSYSTEM": "1",
+ "GIT_CONFIG_GLOBAL": os.devnull,
+ "GIT_CONFIG_NOSYSTEM": "1",
+ "GIT_NO_LAZY_FETCH": "1",
+ "GIT_NO_REPLACE_OBJECTS": "1",
+ "GIT_OPTIONAL_LOCKS": "0",
+ "GIT_PAGER": "cat",
+ "GIT_TERMINAL_PROMPT": "0",
+ "HOME": os.environ.get("HOME", os.devnull),
+ "LANG": "C",
+ "LC_ALL": "C",
+ "PATH": os.environ.get("PATH", os.defpath),
+ }
+
+
+def git_bytes(*arguments: str, maximum: int = MAX_SOURCE_BYTES) -> bytes:
+ return run(
+ ["git", "--no-pager", "--literal-pathspecs", *arguments],
+ git_environment(),
+ label="source identity inspection",
+ maximum_stdout=maximum,
+ maximum_stderr=0,
+ timeout_seconds=60.0,
+ )
+
+
+def git(*arguments: str) -> str:
+ try:
+ return git_bytes(*arguments).decode("utf-8", "strict").strip()
+ except UnicodeError as error:
+ raise GateError("source identity is not UTF-8") from error
+
+
+def require_source_state(source_revision: str, source_tree: str) -> None:
+ if (
+ git("rev-parse", "HEAD") != source_revision
+ or git("rev-parse", "HEAD^{tree}") != source_tree
+ or git("symbolic-ref", "--short", "HEAD") != BRANCH
+ or git("remote", "get-url", "origin") != ORIGIN
+ or git("rev-parse", f"refs/remotes/origin/{BRANCH}") != source_revision
+ or git_bytes("status", "--porcelain=v1", "-z", "--untracked-files=all")
+ ):
+ raise GateError("Lib source is not clean and tracking-exact")
+
+ tracked = git_bytes("ls-files", "-z").split(b"\0")
+ if any(path == b".github" or path.startswith(b".github/") for path in tracked):
+ raise GateError("forbidden .github surface is tracked")
+ if os.path.lexists(ROOT / ".github"):
+ raise GateError("forbidden .github surface is present")
+
+
+def gate_environment() -> dict[str, str]:
+ allowed = {
+ "CARGO_HOME",
+ "CARGO_PROFILE_DEV_DEBUG",
+ "CARGO_TARGET_DIR",
+ "EXT_BUILD_CONFIG",
+ "EXT_BUILD_CARGO_TIMINGS_DIR",
+ "EXT_BUILD_MACHINE_CONFIG",
+ "EXT_BUILD_ROOT",
+ "HOME",
+ "LANG",
+ "LC_ALL",
+ "PATH",
+ "RUSTUP_HOME",
+ "RUSTUP_TOOLCHAIN",
+ "SCCACHE_DIR",
+ "SDKROOT",
+ "TMPDIR",
+ }
+ environment = {name: value for name, value in os.environ.items() if name in allowed}
+ environment.update(
+ {
+ "CARGO_NET_OFFLINE": "true",
+ "CARGO_TERM_COLOR": "never",
+ "LANG": "C",
+ "LC_ALL": "C",
+ }
+ )
+ if not environment.get("PATH"):
+ environment["PATH"] = os.defpath
+ extbuild_root_value = environment.get("EXT_BUILD_ROOT")
+ routed_path_names = (
+ "CARGO_TARGET_DIR",
+ "EXT_BUILD_CARGO_TIMINGS_DIR",
+ "SCCACHE_DIR",
+ )
+ routed_values = [environment[name] for name in routed_path_names if name in environment]
+ if routed_values:
+ if not extbuild_root_value:
+ raise GateError("extbuild routing environment is incomplete")
+ extbuild_root_input = Path(extbuild_root_value)
+ if not extbuild_root_input.is_absolute():
+ raise GateError("extbuild routing environment is invalid")
+ extbuild_root = Path(os.path.abspath(extbuild_root_value))
+ for value in routed_values:
+ destination_input = Path(value)
+ if not destination_input.is_absolute():
+ raise GateError("extbuild routing environment is invalid")
+ destination = Path(os.path.abspath(value))
+ if (
+ destination != extbuild_root and extbuild_root not in destination.parents
+ ):
+ raise GateError("extbuild routing environment is invalid")
+ return environment
+
+
+def run_step() -> None:
+ environment = gate_environment()
+ run(
+ [
+ "cargo",
+ "+1.97.1",
+ "test",
+ "--offline",
+ "--manifest-path",
+ "tools/xtask/Cargo.toml",
+ "--locked",
+ "--bin",
+ "xtask",
+ "bounded_process",
+ "--",
+ "--test-threads=1",
+ ],
+ environment,
+ label="bounded-process vector tests",
+ )
+ run(
+ [
+ "cargo",
+ "+1.97.1",
+ "run",
+ "--offline",
+ "--manifest-path",
+ "tools/xtask/Cargo.toml",
+ "--locked",
+ "--",
+ "bounded-process-self-test",
+ ],
+ environment,
+ label="bounded-process self-test",
+ )
+ run(
+ [
+ "cargo",
+ "+1.97.1",
+ "test",
+ "--offline",
+ "--manifest-path",
+ "tools/xtask/Cargo.toml",
+ "--locked",
+ "--test",
+ "services_hardening_bounded_process_decision",
+ "--",
+ "--test-threads=1",
+ ],
+ environment,
+ label="bounded-process decision contract",
+ )
+ if git_bytes("status", "--porcelain=v1", "-z", "--untracked-files=all"):
+ raise GateError("verification changed the tracked or untracked source state")
+
+
+def parse_arguments() -> argparse.Namespace:
+ parser = RedactedArgumentParser(allow_abbrev=False)
+ parser.add_argument("--step", type=int, required=True)
+ parser.add_argument("--check-id")
+ parser.add_argument("--source-revision", required=True)
+ parser.add_argument("--source-tree", required=True)
+ parser.add_argument("--candidate-digest")
+ parser.add_argument("--platform", required=True)
+ parser.add_argument("--execution-request-sha256", required=True)
+ return parser.parse_args()
+
+
+def validate_digest(value: str, label: str, length: int) -> None:
+ if re.fullmatch(rf"[0-9a-f]{{{length}}}", value) is None:
+ raise GateError(f"{label} is not canonical")
+
+
+def expected_contract(verifier_digest: str) -> dict[str, object]:
+ return {
+ "argv_template": EXPECTED_ARGV_TEMPLATE,
+ "assertion_id": [ASSERTION_ID],
+ "check_id": CHECK_ID,
+ "environment_authority": EXPECTED_ENVIRONMENT_AUTHORITY,
+ "environment_names": EXPECTED_ENVIRONMENT_NAMES,
+ "gate_definition_sha256": GATE_DIGEST,
+ "required_platforms": ["macos_aarch64"],
+ "required_tools": ["uv", "python3", "git"],
+ "result_schema": "radroots.services-hardening.rshr-200-step-check-result.v1",
+ "schema": "radroots.services-hardening.rshr-200-step-check-command.v1",
+ "step": STEP,
+ "verifier_path": "tools/rshr_201_step_gate.py",
+ "verifier_sha256": verifier_digest,
+ }
+
+
+def main() -> int:
+ arguments = parse_arguments()
+ if arguments.step != STEP:
+ raise GateError("step is outside the Lib gate authority")
+ validate_digest(arguments.source_revision, "source revision", 40)
+ validate_digest(arguments.source_tree, "source tree", 40)
+ validate_digest(arguments.execution_request_sha256, "execution request", 64)
+ if sha256_bytes(GATE_DEFINITION.encode("utf-8")) != GATE_DIGEST:
+ raise GateError("compiled gate definition digest differs")
+ if arguments.check_id != CHECK_ID:
+ raise GateError("check identity differs")
+ if arguments.candidate_digest != "none" or arguments.platform != "macos_aarch64":
+ raise GateError("candidate or platform scope differs")
+
+ authority_bytes = read_regular(AUTHORITY_PATH, 256 * 1024)
+ try:
+ authority = json.loads(authority_bytes)
+ except (UnicodeError, json.JSONDecodeError) as error:
+ raise GateError("gate authority is not canonical JSON") from error
+ if canonical(authority) + b"\n" != authority_bytes:
+ raise GateError("gate authority is not canonical JSON")
+ if (
+ not isinstance(authority, dict)
+ or set(authority) != {"schema", "step", "gate_command_contract"}
+ or authority.get("schema") != "radroots.lib.rshr-201-step-gates.v1"
+ or authority.get("step") != [STEP]
+ ):
+ raise GateError("gate authority step inventory differs")
+ contracts = authority.get("gate_command_contract")
+ if not isinstance(contracts, list) or len(contracts) != 1:
+ raise GateError("gate command authority is absent or duplicated")
+
+ verifier_digest = sha256_bytes(read_regular(Path(__file__).resolve()))
+ contract = contracts[0]
+ if contract != expected_contract(verifier_digest):
+ raise GateError("gate command authority differs from source bytes")
+
+ require_source_state(arguments.source_revision, arguments.source_tree)
+ run_step()
+ assertions = [{"id": ASSERTION_ID, "result": "pass"}]
+ result = {
+ "schema": "radroots.services-hardening.rshr-200-step-check-result.v1",
+ "step": STEP,
+ "check_id": CHECK_ID,
+ "gate_definition_sha256": GATE_DIGEST,
+ "source_revision": arguments.source_revision,
+ "source_tree": arguments.source_tree,
+ "candidate_generation": 0,
+ "candidate_digest": "none",
+ "command_contract_sha256": sha256_bytes(canonical(contract)),
+ "verifier_sha256": verifier_digest,
+ "execution_request": [
+ {
+ "platform": arguments.platform,
+ "sha256": arguments.execution_request_sha256,
+ }
+ ],
+ "assertion_inventory_sha256": sha256_bytes(canonical(assertions)),
+ "assertion": assertions,
+ "result": "pass",
+ }
+ sys.stdout.buffer.write(canonical(result) + b"\n")
+ return 0
+
+
+if __name__ == "__main__":
+ try:
+ raise SystemExit(main())
+ except GateError as error:
+ print(f"Lib RSHR-201 gate failed: {error}", file=sys.stderr)
+ raise SystemExit(1)
+ except Exception:
+ print("Lib RSHR-201 gate failed safely", file=sys.stderr)
+ raise SystemExit(1)