Spaces:
Running
Running
| #!/usr/bin/env python3 | |
| """Create, verify, compare or summarize bounded local supported-path receipts.""" | |
| from __future__ import annotations | |
| import argparse | |
| import hashlib | |
| import json | |
| import os | |
| import re | |
| import secrets | |
| import stat | |
| import sys | |
| from collections.abc import Mapping, Sequence | |
| from datetime import UTC, datetime | |
| from pathlib import Path | |
| MAX_INPUT_BYTES = 4_096 | |
| MAX_RECEIPT_BYTES = 8_192 | |
| MAX_COMPARE_OUTPUT_BYTES = 4_096 | |
| MAX_MANIFEST_BYTES = 16_384 | |
| MAX_SUMMARY_OUTPUT_BYTES = 4_096 | |
| SCHEMA_VERSION = "lumi-eggcracker-supported-path-receipt-v1" | |
| RECEIPT_FIELDS = { | |
| "schema_version", | |
| "recorded_at_utc", | |
| "supported_environment", | |
| "supported_path", | |
| "exact_command", | |
| "expected_result", | |
| "observed_result", | |
| "permission_to_quote", | |
| "privacy", | |
| } | |
| TIMESTAMP_PATTERN = re.compile( | |
| r"[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z" | |
| ) | |
| SUPPORTED_ENVIRONMENT = "disposable-native-ubuntu-24.04-cgroup-v2" | |
| SUPPORTED_PATH = "containment-primitive-probe" | |
| EXACT_COMMAND = ( | |
| "sudo /usr/bin/python3 -I -S scripts/containment_probe.py " | |
| "--i-understand-this-kills-a-test-tree" | |
| ) | |
| EXPECTED_RESULT = ( | |
| "result=TERMINATED,target_survivors=0,canary_survived=true," | |
| "root_populated=0,cleanup_complete=true,exit=0" | |
| ) | |
| SOURCE_COMMIT_PATTERN = re.compile(r"[0-9a-f]{40}") | |
| SOURCE_TREE_PATTERN = re.compile(r"[0-9a-f]{64}") | |
| BLOCKER_STAGES = {"host-preflight", "command-start", "probe-execution", "receipt-check"} | |
| BLOCKER_CODES = { | |
| "CGROUP_KILL_REQUIRED", | |
| "CGROUP_V2_REQUIRED", | |
| "COMMAND_FAILED", | |
| "OTHER_REDACTED_BLOCKER", | |
| "PIDFD_REQUIRED", | |
| "RECEIPT_INVALID", | |
| "RECEIPT_MISSING", | |
| "ROOT_REQUIRED", | |
| "SOURCE_IDENTITY_REQUIRED", | |
| "SYSTEMD_REQUIRED", | |
| "UNSUPPORTED_HOST", | |
| } | |
| OBSERVED_RESULT_COMPARE_FIELDS = ( | |
| "outcome", | |
| "result", | |
| "target_survivors", | |
| "canary_survived", | |
| "root_populated", | |
| "cleanup_complete", | |
| "blocker_stage", | |
| "blocker_code", | |
| ) | |
| INPUT_FIELDS = { | |
| "supported_environment", | |
| "supported_path", | |
| "exact_command", | |
| "expected_result", | |
| "observed_result", | |
| "permission_to_quote", | |
| } | |
| SUMMARY_MANIFEST_FIELDS = {"receipts"} | |
| SUMMARY_ENTRY_FIELDS = {"path", "source_commit", "source_tree_sha256"} | |
| SUMMARY_ENTRY_EXPECTED_FIELDS = { | |
| "path", | |
| "expected_source_commit", | |
| "expected_source_tree_sha256", | |
| } | |
| SUMMARY_OUTCOMES = ("reproducible-blocker", "supported-success") | |
| PRIVACY = { | |
| "redacted": True, | |
| "secrets_collected": False, | |
| "source_code_collected": False, | |
| "credentials_collected": False, | |
| "personal_data_collected": False, | |
| "raw_environment_collected": False, | |
| "raw_argv_collected": False, | |
| "raw_output_collected": False, | |
| "network_or_upload_used": False, | |
| } | |
| class ReceiptInputError(ValueError): | |
| """A bounded failure whose message never contains operator-supplied values.""" | |
| class SafeArgumentParser(argparse.ArgumentParser): | |
| def error(self, _message: str) -> None: | |
| raise ReceiptInputError("invalid command-line arguments") | |
| def _reject_nonstandard_number(_value: str) -> object: | |
| raise ReceiptInputError("receipt contains a non-standard number") | |
| def _unique_object(pairs: list[tuple[str, object]]) -> dict[str, object]: | |
| value: dict[str, object] = {} | |
| for name, item in pairs: | |
| if name in value: | |
| raise ReceiptInputError("input contains a duplicate object member") | |
| value[name] = item | |
| return value | |
| def _read_input(path: Path) -> object: | |
| try: | |
| metadata = path.lstat() | |
| if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode): | |
| raise ReceiptInputError("input must be a regular file, not a link") | |
| with path.open("rb") as handle: | |
| encoded = handle.read(MAX_INPUT_BYTES + 1) | |
| except ReceiptInputError: | |
| raise | |
| except OSError as error: | |
| raise ReceiptInputError("input could not be read") from error | |
| if len(encoded) > MAX_INPUT_BYTES: | |
| raise ReceiptInputError("input exceeds its safe size bound") | |
| try: | |
| return json.loads(encoded.decode("utf-8"), object_pairs_hook=_unique_object) | |
| except ReceiptInputError: | |
| raise | |
| except (UnicodeError, json.JSONDecodeError, RecursionError) as error: | |
| raise ReceiptInputError("input is not bounded UTF-8 JSON") from error | |
| def _read_manifest(path: Path) -> object: | |
| """Read one bounded summary manifest without exposing its path or values.""" | |
| _safe_receipt_parent(path) | |
| descriptor = -1 | |
| try: | |
| metadata = path.lstat() | |
| if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode): | |
| raise ReceiptInputError("manifest must be a regular file, not a link") | |
| flags = ( | |
| os.O_RDONLY | |
| | getattr(os, "O_NOFOLLOW", 0) | |
| | getattr(os, "O_NONBLOCK", 0) | |
| | getattr(os, "O_BINARY", 0) | |
| ) | |
| descriptor = os.open(path, flags) | |
| with os.fdopen(descriptor, "rb") as handle: | |
| descriptor = -1 | |
| current = os.fstat(handle.fileno()) | |
| if stat.S_ISLNK(current.st_mode) or not stat.S_ISREG(current.st_mode): | |
| raise ReceiptInputError("manifest must be a regular file, not a link") | |
| encoded = handle.read(MAX_MANIFEST_BYTES + 1) | |
| except ReceiptInputError: | |
| raise | |
| except (OSError, ValueError) as error: | |
| raise ReceiptInputError("manifest could not be read") from error | |
| finally: | |
| if descriptor >= 0: | |
| try: | |
| os.close(descriptor) | |
| except OSError: | |
| pass | |
| if len(encoded) > MAX_MANIFEST_BYTES: | |
| raise ReceiptInputError("manifest exceeds its safe size bound") | |
| try: | |
| return json.loads( | |
| encoded.decode("utf-8"), | |
| object_pairs_hook=_unique_object, | |
| parse_constant=_reject_nonstandard_number, | |
| ) | |
| except ReceiptInputError: | |
| raise | |
| except (UnicodeError, json.JSONDecodeError, RecursionError, ValueError) as error: | |
| raise ReceiptInputError("manifest is not bounded UTF-8 JSON") from error | |
| def _require_exact_fields(value: Mapping[str, object], expected: set[str], label: str) -> None: | |
| if set(value) != expected: | |
| raise ReceiptInputError(f"{label} does not match the closed schema") | |
| def _validate_source_identity(observed: Mapping[str, object]) -> None: | |
| source_commit = observed["source_commit"] | |
| source_tree = observed["source_tree_sha256"] | |
| if not isinstance(source_commit, str) or SOURCE_COMMIT_PATTERN.fullmatch(source_commit) is None: | |
| raise ReceiptInputError("source_commit is not an exact lowercase commit identity") | |
| if not isinstance(source_tree, str) or SOURCE_TREE_PATTERN.fullmatch(source_tree) is None: | |
| raise ReceiptInputError("source_tree_sha256 is not an exact lowercase tree identity") | |
| def _validate_observed_result(observed: object) -> dict[str, object]: | |
| if not isinstance(observed, dict): | |
| raise ReceiptInputError("observed_result must be an object") | |
| outcome = observed.get("outcome") | |
| if outcome == "supported-success": | |
| _require_exact_fields( | |
| observed, | |
| { | |
| "outcome", | |
| "result", | |
| "target_survivors", | |
| "canary_survived", | |
| "root_populated", | |
| "cleanup_complete", | |
| "source_commit", | |
| "source_tree_sha256", | |
| }, | |
| "supported success", | |
| ) | |
| required_observations = { | |
| "result": "TERMINATED", | |
| "target_survivors": 0, | |
| "canary_survived": True, | |
| "root_populated": 0, | |
| "cleanup_complete": True, | |
| } | |
| for field, expected in required_observations.items(): | |
| actual = observed[field] | |
| if type(actual) is not type(expected) or actual != expected: | |
| raise ReceiptInputError(f"observed success field {field} is not the required value") | |
| _validate_source_identity(observed) | |
| elif outcome == "reproducible-blocker": | |
| _require_exact_fields( | |
| observed, | |
| { | |
| "outcome", | |
| "blocker_stage", | |
| "blocker_code", | |
| "source_commit", | |
| "source_tree_sha256", | |
| }, | |
| "reproducible blocker", | |
| ) | |
| blocker_stage = observed["blocker_stage"] | |
| if not isinstance(blocker_stage, str) or blocker_stage not in BLOCKER_STAGES: | |
| raise ReceiptInputError("blocker_stage is not an allowed redacted value") | |
| blocker_code = observed["blocker_code"] | |
| if not isinstance(blocker_code, str) or blocker_code not in BLOCKER_CODES: | |
| raise ReceiptInputError("blocker_code is not an allowed redacted value") | |
| _validate_source_identity(observed) | |
| else: | |
| raise ReceiptInputError("observed_result outcome is not supported") | |
| return dict(observed) | |
| def _validate_input(value: object) -> dict[str, object]: | |
| if not isinstance(value, dict): | |
| raise ReceiptInputError("input root must be an object") | |
| _require_exact_fields(value, INPUT_FIELDS, "input") | |
| fixed_strings = { | |
| "supported_environment": SUPPORTED_ENVIRONMENT, | |
| "supported_path": SUPPORTED_PATH, | |
| "exact_command": EXACT_COMMAND, | |
| "expected_result": EXPECTED_RESULT, | |
| } | |
| for field, expected in fixed_strings.items(): | |
| if not isinstance(value[field], str) or value[field] != expected: | |
| raise ReceiptInputError(f"input field {field} is not the supported fixed value") | |
| permission = value["permission_to_quote"] | |
| if not isinstance(permission, bool): | |
| raise ReceiptInputError("permission_to_quote must be an explicit boolean") | |
| _validate_observed_result(value["observed_result"]) | |
| return dict(value) | |
| def _safe_receipt_parent(path: Path) -> Path: | |
| parent = Path(os.path.abspath(path.parent)) | |
| try: | |
| for candidate in (parent, *parent.parents): | |
| metadata = candidate.lstat() | |
| if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode): | |
| raise ReceiptInputError("receipt parent must contain no links or non-directories") | |
| except ReceiptInputError: | |
| raise | |
| except (OSError, ValueError) as error: | |
| raise ReceiptInputError("receipt parent is not an existing regular directory") from error | |
| return parent | |
| def _read_final_receipt(path: Path) -> tuple[object, str]: | |
| _safe_receipt_parent(path) | |
| descriptor = -1 | |
| try: | |
| metadata = path.lstat() | |
| if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode): | |
| raise ReceiptInputError("receipt must be a regular file, not a link") | |
| flags = ( | |
| os.O_RDONLY | |
| | getattr(os, "O_NOFOLLOW", 0) | |
| | getattr(os, "O_NONBLOCK", 0) | |
| | getattr(os, "O_BINARY", 0) | |
| ) | |
| descriptor = os.open(path, flags) | |
| with os.fdopen(descriptor, "rb") as handle: | |
| descriptor = -1 | |
| current = os.fstat(handle.fileno()) | |
| if stat.S_ISLNK(current.st_mode) or not stat.S_ISREG(current.st_mode): | |
| raise ReceiptInputError("receipt must be a regular file, not a link") | |
| encoded = handle.read(MAX_RECEIPT_BYTES + 1) | |
| except ReceiptInputError: | |
| raise | |
| except (OSError, ValueError) as error: | |
| raise ReceiptInputError("receipt could not be read") from error | |
| finally: | |
| if descriptor >= 0: | |
| try: | |
| os.close(descriptor) | |
| except OSError: | |
| pass | |
| if len(encoded) > MAX_RECEIPT_BYTES: | |
| raise ReceiptInputError("receipt exceeds its safe size bound") | |
| receipt_sha256 = hashlib.sha256(encoded).hexdigest() | |
| try: | |
| value = json.loads( | |
| encoded.decode("utf-8"), | |
| object_pairs_hook=_unique_object, | |
| parse_constant=_reject_nonstandard_number, | |
| ) | |
| except ReceiptInputError: | |
| raise | |
| except (UnicodeError, json.JSONDecodeError, RecursionError, ValueError) as error: | |
| raise ReceiptInputError("receipt is not bounded UTF-8 JSON") from error | |
| return value, receipt_sha256 | |
| def _validate_expected_source_identity(value: object, pattern: re.Pattern[str], label: str) -> str: | |
| if not isinstance(value, str) or pattern.fullmatch(value) is None: | |
| raise ReceiptInputError(f"{label} is not an exact lowercase source identity") | |
| return value | |
| def _validate_final_receipt( | |
| value: object, | |
| *, | |
| expected_source_commit: str, | |
| expected_source_tree_sha256: str, | |
| ) -> str: | |
| if not isinstance(value, dict): | |
| raise ReceiptInputError("receipt root must be an object") | |
| _require_exact_fields(value, RECEIPT_FIELDS, "receipt") | |
| fixed_strings = { | |
| "schema_version": SCHEMA_VERSION, | |
| "supported_environment": SUPPORTED_ENVIRONMENT, | |
| "supported_path": SUPPORTED_PATH, | |
| "exact_command": EXACT_COMMAND, | |
| "expected_result": EXPECTED_RESULT, | |
| } | |
| for field, expected in fixed_strings.items(): | |
| if not isinstance(value[field], str) or value[field] != expected: | |
| raise ReceiptInputError(f"receipt field {field} is not the supported fixed value") | |
| recorded_at = value["recorded_at_utc"] | |
| if not isinstance(recorded_at, str) or TIMESTAMP_PATTERN.fullmatch(recorded_at) is None: | |
| raise ReceiptInputError("recorded_at_utc is not an exact UTC timestamp") | |
| try: | |
| datetime.strptime(recorded_at, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=UTC) | |
| except ValueError as error: | |
| raise ReceiptInputError("recorded_at_utc is not an exact UTC timestamp") from error | |
| permission = value["permission_to_quote"] | |
| if not isinstance(permission, bool): | |
| raise ReceiptInputError("permission_to_quote must be an explicit boolean") | |
| privacy = value["privacy"] | |
| if not isinstance(privacy, dict): | |
| raise ReceiptInputError("privacy does not match the closed schema") | |
| _require_exact_fields(privacy, set(PRIVACY), "privacy") | |
| for field, expected in PRIVACY.items(): | |
| actual = privacy[field] | |
| if type(actual) is not type(expected) or actual != expected: | |
| raise ReceiptInputError("privacy does not match the required redacted values") | |
| observed = _validate_observed_result(value["observed_result"]) | |
| _validate_expected_source_identity( | |
| expected_source_commit, SOURCE_COMMIT_PATTERN, "expected source_commit" | |
| ) | |
| _validate_expected_source_identity( | |
| expected_source_tree_sha256, SOURCE_TREE_PATTERN, "expected source_tree_sha256" | |
| ) | |
| if ( | |
| observed["source_commit"] != expected_source_commit | |
| or observed["source_tree_sha256"] != expected_source_tree_sha256 | |
| ): | |
| raise ReceiptInputError("receipt source identity does not match the expected identities") | |
| outcome = observed["outcome"] | |
| if not isinstance(outcome, str): | |
| raise ReceiptInputError("observed_result outcome is not supported") | |
| return outcome | |
| def _read_validated_receipt( | |
| path: Path, | |
| *, | |
| expected_source_commit: str, | |
| expected_source_tree_sha256: str, | |
| ) -> tuple[dict[str, object], str]: | |
| value, receipt_sha256 = _read_final_receipt(path) | |
| _validate_final_receipt( | |
| value, | |
| expected_source_commit=expected_source_commit, | |
| expected_source_tree_sha256=expected_source_tree_sha256, | |
| ) | |
| if not isinstance(value, dict) or not isinstance(value.get("observed_result"), dict): | |
| raise ReceiptInputError("receipt observed_result is not a bounded object") | |
| return dict(value["observed_result"]), receipt_sha256 | |
| def _normalize_summary_path(raw_path: object) -> tuple[Path, str]: | |
| if not isinstance(raw_path, str) or not raw_path or "\x00" in raw_path: | |
| raise ReceiptInputError("manifest receipt path is not a usable text path") | |
| try: | |
| normalized_path = os.path.abspath(os.path.normpath(raw_path)) | |
| normalized_key = os.path.normcase(normalized_path) | |
| except (OSError, TypeError, ValueError) as error: | |
| raise ReceiptInputError("manifest receipt path is not a usable text path") from error | |
| if not normalized_key: | |
| raise ReceiptInputError("manifest receipt path is not a usable text path") | |
| return Path(normalized_path), normalized_key | |
| def _validate_summary_manifest(value: object) -> list[tuple[Path, str, str]]: | |
| if not isinstance(value, dict): | |
| raise ReceiptInputError("summary manifest root must be an object") | |
| _require_exact_fields(value, SUMMARY_MANIFEST_FIELDS, "summary manifest") | |
| entries = value["receipts"] | |
| if not isinstance(entries, list) or len(entries) < 2 or len(entries) > 8: | |
| raise ReceiptInputError("summary manifest must contain between two and eight receipts") | |
| validated: list[tuple[Path, str, str]] = [] | |
| normalized_paths: set[str] = set() | |
| for entry in entries: | |
| if not isinstance(entry, dict): | |
| raise ReceiptInputError("summary manifest receipt entry must be an object") | |
| entry_fields = set(entry) | |
| if entry_fields == SUMMARY_ENTRY_FIELDS: | |
| source_commit_field = "source_commit" | |
| source_tree_field = "source_tree_sha256" | |
| elif entry_fields == SUMMARY_ENTRY_EXPECTED_FIELDS: | |
| source_commit_field = "expected_source_commit" | |
| source_tree_field = "expected_source_tree_sha256" | |
| else: | |
| raise ReceiptInputError("summary manifest receipt entry does not match the closed schema") | |
| receipt_path, normalized_key = _normalize_summary_path(entry["path"]) | |
| if normalized_key in normalized_paths: | |
| raise ReceiptInputError("summary manifest contains a duplicate normalized path") | |
| normalized_paths.add(normalized_key) | |
| source_commit = _validate_expected_source_identity( | |
| entry[source_commit_field], SOURCE_COMMIT_PATTERN, "manifest source_commit" | |
| ) | |
| source_tree_sha256 = _validate_expected_source_identity( | |
| entry[source_tree_field], SOURCE_TREE_PATTERN, "manifest source_tree_sha256" | |
| ) | |
| validated.append((receipt_path, source_commit, source_tree_sha256)) | |
| return validated | |
| def summarize_receipts(manifest_path: str | os.PathLike[str]) -> dict[str, object]: | |
| """Validate and summarize two through eight source-bound receipts.""" | |
| entries = _validate_summary_manifest(_read_manifest(Path(manifest_path))) | |
| rows: list[dict[str, str]] = [] | |
| receipt_digests: set[str] = set() | |
| outcome_counts = dict.fromkeys(SUMMARY_OUTCOMES, 0) | |
| for receipt_path, source_commit, source_tree_sha256 in entries: | |
| observed, receipt_sha256 = _read_validated_receipt( | |
| receipt_path, | |
| expected_source_commit=source_commit, | |
| expected_source_tree_sha256=source_tree_sha256, | |
| ) | |
| if receipt_sha256 in receipt_digests: | |
| raise ReceiptInputError("summary manifest contains a duplicate receipt digest") | |
| receipt_digests.add(receipt_sha256) | |
| outcome = observed["outcome"] | |
| if not isinstance(outcome, str) or outcome not in outcome_counts: | |
| raise ReceiptInputError("receipt outcome is not supported") | |
| outcome_counts[outcome] += 1 | |
| rows.append( | |
| { | |
| "outcome": outcome, | |
| "receipt_sha256": receipt_sha256, | |
| "source_commit": source_commit, | |
| "source_tree_sha256": source_tree_sha256, | |
| } | |
| ) | |
| return { | |
| "outcome": "supported-receipt-set-summary", | |
| "rows": rows, | |
| "aggregate": { | |
| "receipt_count": len(rows), | |
| "outcome_counts": dict(outcome_counts), | |
| }, | |
| } | |
| def compare_receipts( | |
| before_path: str | os.PathLike[str], | |
| after_path: str | os.PathLike[str], | |
| *, | |
| before_source_commit: str, | |
| before_source_tree_sha256: str, | |
| after_source_commit: str, | |
| after_source_tree_sha256: str, | |
| ) -> dict[str, object]: | |
| """Validate and compare two final receipts without writing either one.""" | |
| before_observed, before_receipt_sha256 = _read_validated_receipt( | |
| Path(before_path), | |
| expected_source_commit=before_source_commit, | |
| expected_source_tree_sha256=before_source_tree_sha256, | |
| ) | |
| after_observed, after_receipt_sha256 = _read_validated_receipt( | |
| Path(after_path), | |
| expected_source_commit=after_source_commit, | |
| expected_source_tree_sha256=after_source_tree_sha256, | |
| ) | |
| missing = object() | |
| observed_result_changes: dict[str, dict[str, object | None]] = {} | |
| for field in OBSERVED_RESULT_COMPARE_FIELDS: | |
| before_value = before_observed.get(field, missing) | |
| after_value = after_observed.get(field, missing) | |
| if before_value is missing and after_value is missing: | |
| continue | |
| if before_value == after_value: | |
| continue | |
| observed_result_changes[field] = { | |
| "before": None if before_value is missing else before_value, | |
| "after": None if after_value is missing else after_value, | |
| } | |
| return { | |
| "changed": before_receipt_sha256 != after_receipt_sha256, | |
| "before_receipt_sha256": before_receipt_sha256, | |
| "after_receipt_sha256": after_receipt_sha256, | |
| "source_transition": { | |
| "before": { | |
| "source_commit": before_observed["source_commit"], | |
| "source_tree_sha256": before_observed["source_tree_sha256"], | |
| }, | |
| "after": { | |
| "source_commit": after_observed["source_commit"], | |
| "source_tree_sha256": after_observed["source_tree_sha256"], | |
| }, | |
| }, | |
| "observed_result_changes": observed_result_changes, | |
| } | |
| def _encode_comparison(value: Mapping[str, object]) -> bytes: | |
| try: | |
| encoded = ( | |
| json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False) + "\n" | |
| ).encode("utf-8") | |
| except (TypeError, ValueError, OverflowError) as error: | |
| raise ReceiptInputError("comparison output is not bounded JSON") from error | |
| if len(encoded) > MAX_COMPARE_OUTPUT_BYTES: | |
| raise ReceiptInputError("comparison output exceeds its safe size bound") | |
| return encoded | |
| def _encode_summary(value: Mapping[str, object]) -> bytes: | |
| try: | |
| encoded = ( | |
| json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False) + "\n" | |
| ).encode("utf-8") | |
| except (TypeError, ValueError, OverflowError) as error: | |
| raise ReceiptInputError("summary output is not bounded JSON") from error | |
| if len(encoded) > MAX_SUMMARY_OUTPUT_BYTES: | |
| raise ReceiptInputError("summary output exceeds its safe size bound") | |
| return encoded | |
| def _safe_output_parent(destination: Path) -> Path: | |
| parent = Path(os.path.abspath(destination.parent)) | |
| try: | |
| for candidate in (parent, *parent.parents): | |
| metadata = candidate.lstat() | |
| if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode): | |
| raise ReceiptInputError("output parent must contain no links or non-directories") | |
| except ReceiptInputError: | |
| raise | |
| except OSError as error: | |
| raise ReceiptInputError("output parent is not an existing regular directory") from error | |
| return parent | |
| def _write_new_receipt(destination: Path, receipt: Mapping[str, object]) -> None: | |
| parent = _safe_output_parent(destination) | |
| destination = parent / destination.name | |
| try: | |
| metadata = destination.lstat() | |
| except FileNotFoundError: | |
| pass | |
| except OSError as error: | |
| raise ReceiptInputError("output could not be checked safely") from error | |
| else: | |
| if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode): | |
| raise ReceiptInputError("output must not be a link or non-regular file") | |
| raise ReceiptInputError("output already exists; refusing to overwrite it") | |
| encoded = (json.dumps(receipt, indent=2, sort_keys=True) + "\n").encode("utf-8") | |
| if len(encoded) > MAX_RECEIPT_BYTES: | |
| raise ReceiptInputError("receipt exceeds its safe size bound") | |
| temporary = parent / f".{destination.name}.{secrets.token_hex(8)}.tmp" | |
| flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | |
| flags |= getattr(os, "O_NOFOLLOW", 0) | |
| flags |= getattr(os, "O_BINARY", 0) | |
| linked = False | |
| try: | |
| descriptor = os.open(temporary, flags, 0o600) | |
| with os.fdopen(descriptor, "wb") as handle: | |
| handle.write(encoded) | |
| handle.flush() | |
| os.fsync(handle.fileno()) | |
| os.link(temporary, destination, follow_symlinks=False) | |
| linked = True | |
| except OSError as error: | |
| raise ReceiptInputError("receipt could not be written safely") from error | |
| finally: | |
| try: | |
| temporary.unlink(missing_ok=True) | |
| except OSError: | |
| if not linked: | |
| raise ReceiptInputError("temporary receipt could not be removed safely") from None | |
| def _build_receipt(value: Mapping[str, object]) -> dict[str, object]: | |
| return { | |
| "schema_version": SCHEMA_VERSION, | |
| "recorded_at_utc": datetime.now(UTC) | |
| .isoformat(timespec="seconds") | |
| .replace("+00:00", "Z"), | |
| **value, | |
| "privacy": dict(PRIVACY), | |
| } | |
| def main(argv: Sequence[str] | None = None) -> int: | |
| parser = SafeArgumentParser( | |
| description="write, verify, compare or summarize bounded local supported-path receipts" | |
| ) | |
| raw_arguments = list(sys.argv[1:] if argv is None else argv) | |
| if raw_arguments[:1] in (["compare-receipts"], ["compare-receipt"], ["compare"]): | |
| raw_arguments = ["--compare-receipts", *raw_arguments[1:]] | |
| if raw_arguments[:1] in ( | |
| ["summary"], | |
| ["summary-receipts"], | |
| ["summarize-receipts"], | |
| ["receipt-set-summary"], | |
| ): | |
| raw_arguments = ["--summary-receipts", *raw_arguments[1:]] | |
| verification_requested = "--verify-receipt" in raw_arguments | |
| comparison_requested = any( | |
| argument | |
| in { | |
| "--compare-receipts", | |
| "--compare-receipt", | |
| "--compare", | |
| "compare-receipts", | |
| "compare-receipt", | |
| "compare", | |
| } | |
| for argument in raw_arguments | |
| ) | |
| summary_requested = any( | |
| argument | |
| in { | |
| "--summary-receipts", | |
| "--summarize-receipts", | |
| "--summarize-supported-path-receipts", | |
| "--receipt-set-summary", | |
| "--summary", | |
| "summary", | |
| "summary-receipts", | |
| "summarize-receipts", | |
| "receipt-set-summary", | |
| "--receipt-manifest", | |
| "--summary-manifest", | |
| "--i-opt-in-to-summarize-receipts", | |
| "--i-opt-in-to-summarize-receipt-set", | |
| "--i-opt-in-to-summary", | |
| "--require-all-supported", | |
| } | |
| for argument in raw_arguments | |
| ) | |
| modes = parser.add_mutually_exclusive_group() | |
| modes.add_argument( | |
| "--i-opt-in-to-write-local-receipt", | |
| action="store_true", | |
| help="explicitly allow this command to write the local redacted receipt", | |
| ) | |
| modes.add_argument( | |
| "--verify-receipt", | |
| type=Path, | |
| help="read and verify one final local JSON receipt without writing", | |
| ) | |
| modes.add_argument( | |
| "--compare-receipts", | |
| "--compare-receipt", | |
| "--compare", | |
| nargs="*", | |
| dest="compare_paths", | |
| metavar="FILE", | |
| help="compare two final local JSON receipts without writing", | |
| ) | |
| modes.add_argument( | |
| "--summary-receipts", | |
| "--summarize-receipts", | |
| "--summarize-supported-path-receipts", | |
| "--receipt-set-summary", | |
| "--summary", | |
| action="store_true", | |
| dest="summary_receipts", | |
| help="summarize two through eight final local JSON receipts without writing", | |
| ) | |
| parser.add_argument( | |
| "--i-opt-in-to-summarize-receipts", | |
| "--i-opt-in-to-summarize-receipt-set", | |
| "--i-opt-in-to-summary", | |
| action="store_true", | |
| dest="i_opt_in_to_summarize_receipts", | |
| help="explicitly allow a bounded local receipt-set summary", | |
| ) | |
| parser.add_argument( | |
| "--require-all-supported", | |
| action="store_true", | |
| help="exit 1 when a valid receipt-set summary contains a reproducible blocker", | |
| ) | |
| parser.add_argument( | |
| "--expect-source-commit", | |
| help="expected exact lowercase source commit identity for verification", | |
| ) | |
| parser.add_argument( | |
| "--expect-source-tree-sha256", | |
| help="expected exact lowercase source tree identity for verification", | |
| ) | |
| parser.add_argument( | |
| "--before", | |
| "--before-receipt", | |
| dest="before_receipt", | |
| type=Path, | |
| help="first final local JSON receipt for comparison", | |
| ) | |
| parser.add_argument( | |
| "--after", | |
| "--after-receipt", | |
| dest="after_receipt", | |
| type=Path, | |
| help="second final local JSON receipt for comparison", | |
| ) | |
| parser.add_argument( | |
| "--before-source-commit", | |
| "--expect-before-source-commit", | |
| "--before-expect-source-commit", | |
| "--before-expected-source-commit", | |
| dest="before_source_commit", | |
| help="expected source commit identity for the first receipt", | |
| ) | |
| parser.add_argument( | |
| "--before-source-tree-sha256", | |
| "--expect-before-source-tree-sha256", | |
| "--before-expect-source-tree-sha256", | |
| "--before-expected-source-tree-sha256", | |
| dest="before_source_tree_sha256", | |
| help="expected source tree identity for the first receipt", | |
| ) | |
| parser.add_argument( | |
| "--after-source-commit", | |
| "--expect-after-source-commit", | |
| "--after-expect-source-commit", | |
| "--after-expected-source-commit", | |
| dest="after_source_commit", | |
| help="expected source commit identity for the second receipt", | |
| ) | |
| parser.add_argument( | |
| "--after-source-tree-sha256", | |
| "--expect-after-source-tree-sha256", | |
| "--after-expect-source-tree-sha256", | |
| "--after-expected-source-tree-sha256", | |
| dest="after_source_tree_sha256", | |
| help="expected source tree identity for the second receipt", | |
| ) | |
| parser.add_argument("--input", type=Path, help="bounded local JSON input") | |
| parser.add_argument("--output", type=Path, help="new local JSON receipt") | |
| parser.add_argument( | |
| "--manifest", | |
| "--receipt-manifest", | |
| "--summary-manifest", | |
| dest="summary_manifest", | |
| type=Path, | |
| help="bounded local JSON manifest for a receipt-set summary", | |
| ) | |
| try: | |
| arguments = parser.parse_args(raw_arguments) | |
| verification_requested = arguments.verify_receipt is not None | |
| comparison_requested = ( | |
| arguments.compare_paths is not None | |
| or arguments.before_receipt is not None | |
| or arguments.after_receipt is not None | |
| or arguments.before_source_commit is not None | |
| or arguments.before_source_tree_sha256 is not None | |
| or arguments.after_source_commit is not None | |
| or arguments.after_source_tree_sha256 is not None | |
| ) | |
| summary_requested = ( | |
| arguments.summary_receipts | |
| or arguments.summary_manifest is not None | |
| or arguments.i_opt_in_to_summarize_receipts | |
| or arguments.require_all_supported | |
| ) | |
| if arguments.verify_receipt is not None: | |
| if ( | |
| arguments.i_opt_in_to_write_local_receipt | |
| or arguments.i_opt_in_to_summarize_receipts | |
| or arguments.require_all_supported | |
| or arguments.input is not None | |
| or arguments.output is not None | |
| or arguments.summary_manifest is not None | |
| or arguments.compare_paths is not None | |
| or arguments.before_receipt is not None | |
| or arguments.after_receipt is not None | |
| or arguments.before_source_commit is not None | |
| or arguments.before_source_tree_sha256 is not None | |
| or arguments.after_source_commit is not None | |
| or arguments.after_source_tree_sha256 is not None | |
| ): | |
| raise ReceiptInputError("verification mode cannot be combined with write mode") | |
| if ( | |
| arguments.expect_source_commit is None | |
| or arguments.expect_source_tree_sha256 is None | |
| ): | |
| raise ReceiptInputError("verification requires both expected source identities") | |
| _validate_expected_source_identity( | |
| arguments.expect_source_commit, SOURCE_COMMIT_PATTERN, "expected source_commit" | |
| ) | |
| _validate_expected_source_identity( | |
| arguments.expect_source_tree_sha256, | |
| SOURCE_TREE_PATTERN, | |
| "expected source_tree_sha256", | |
| ) | |
| receipt, receipt_sha256 = _read_final_receipt(arguments.verify_receipt) | |
| outcome = _validate_final_receipt( | |
| receipt, | |
| expected_source_commit=arguments.expect_source_commit, | |
| expected_source_tree_sha256=arguments.expect_source_tree_sha256, | |
| ) | |
| print( | |
| json.dumps( | |
| {"outcome": outcome, "receipt_sha256": receipt_sha256}, | |
| sort_keys=True, | |
| separators=(",", ":"), | |
| ) | |
| ) | |
| return 0 | |
| if summary_requested: | |
| if not arguments.summary_receipts: | |
| raise ReceiptInputError("summary manifest or opt-in requires summary mode") | |
| if ( | |
| not arguments.i_opt_in_to_summarize_receipts | |
| or arguments.input is not None | |
| or arguments.output is not None | |
| or arguments.expect_source_commit is not None | |
| or arguments.expect_source_tree_sha256 is not None | |
| or arguments.compare_paths is not None | |
| or arguments.before_receipt is not None | |
| or arguments.after_receipt is not None | |
| or arguments.before_source_commit is not None | |
| or arguments.before_source_tree_sha256 is not None | |
| or arguments.after_source_commit is not None | |
| or arguments.after_source_tree_sha256 is not None | |
| ): | |
| raise ReceiptInputError("summary mode requires its opt-in and cannot be combined with other modes") | |
| if arguments.summary_manifest is None: | |
| raise ReceiptInputError("summary mode requires --manifest") | |
| summary = summarize_receipts(arguments.summary_manifest) | |
| encoded = _encode_summary(summary) | |
| sys.stdout.write(encoded.decode("utf-8")) | |
| if arguments.require_all_supported: | |
| outcome_counts = summary["aggregate"]["outcome_counts"] | |
| return 1 if outcome_counts["reproducible-blocker"] else 0 | |
| return 0 | |
| if comparison_requested: | |
| if ( | |
| arguments.i_opt_in_to_write_local_receipt | |
| or arguments.i_opt_in_to_summarize_receipts | |
| or arguments.verify_receipt is not None | |
| or arguments.input is not None | |
| or arguments.output is not None | |
| or arguments.summary_manifest is not None | |
| or arguments.expect_source_commit is not None | |
| or arguments.expect_source_tree_sha256 is not None | |
| ): | |
| raise ReceiptInputError("comparison mode cannot be combined with write or verify mode") | |
| positional_paths = arguments.compare_paths | |
| if positional_paths: | |
| if ( | |
| len(positional_paths) != 2 | |
| or arguments.before_receipt is not None | |
| or arguments.after_receipt is not None | |
| ): | |
| raise ReceiptInputError("comparison requires exactly two receipt files") | |
| before_receipt = Path(positional_paths[0]) | |
| after_receipt = Path(positional_paths[1]) | |
| else: | |
| before_receipt = arguments.before_receipt | |
| after_receipt = arguments.after_receipt | |
| if before_receipt is None or after_receipt is None: | |
| raise ReceiptInputError("comparison requires both receipt files") | |
| if ( | |
| arguments.before_source_commit is None | |
| or arguments.before_source_tree_sha256 is None | |
| or arguments.after_source_commit is None | |
| or arguments.after_source_tree_sha256 is None | |
| ): | |
| raise ReceiptInputError("comparison requires both expected source identity pairs") | |
| _validate_expected_source_identity( | |
| arguments.before_source_commit, | |
| SOURCE_COMMIT_PATTERN, | |
| "before source_commit", | |
| ) | |
| _validate_expected_source_identity( | |
| arguments.before_source_tree_sha256, | |
| SOURCE_TREE_PATTERN, | |
| "before source_tree_sha256", | |
| ) | |
| _validate_expected_source_identity( | |
| arguments.after_source_commit, | |
| SOURCE_COMMIT_PATTERN, | |
| "after source_commit", | |
| ) | |
| _validate_expected_source_identity( | |
| arguments.after_source_tree_sha256, | |
| SOURCE_TREE_PATTERN, | |
| "after source_tree_sha256", | |
| ) | |
| comparison = compare_receipts( | |
| before_receipt, | |
| after_receipt, | |
| before_source_commit=arguments.before_source_commit, | |
| before_source_tree_sha256=arguments.before_source_tree_sha256, | |
| after_source_commit=arguments.after_source_commit, | |
| after_source_tree_sha256=arguments.after_source_tree_sha256, | |
| ) | |
| sys.stdout.write(_encode_comparison(comparison).decode("utf-8")) | |
| return 0 | |
| if ( | |
| arguments.expect_source_commit is not None | |
| or arguments.expect_source_tree_sha256 is not None | |
| ): | |
| raise ReceiptInputError("expected source identities require --verify-receipt") | |
| if not arguments.i_opt_in_to_write_local_receipt: | |
| raise ReceiptInputError("explicit opt-in is required; no receipt was written") | |
| if arguments.input is None or arguments.output is None: | |
| raise ReceiptInputError("write mode requires --input and --output") | |
| value = _validate_input(_read_input(arguments.input)) | |
| _write_new_receipt(arguments.output, _build_receipt(value)) | |
| except ReceiptInputError as error: | |
| if verification_requested: | |
| prefix = "NOT VERIFIED" | |
| elif summary_requested: | |
| prefix = "NOT SUMMARIZED" | |
| elif comparison_requested: | |
| prefix = "NOT COMPARED" | |
| else: | |
| prefix = "NOT WRITTEN" | |
| print(f"{prefix}: {error}", file=sys.stderr) | |
| return 2 | |
| print("WROTE local redacted supported-path receipt") | |
| return 0 | |
| if __name__ == "__main__": | |
| raise SystemExit(main()) | |