#!/usr/bin/env python3 """Validate Railiance S1 receipts and reject secret-shaped content.""" from __future__ import annotations import argparse import json import re import sys import uuid from datetime import datetime, timezone from pathlib import Path from typing import Any SCHEMA_VERSION = "1.0" EVENT_TYPES = { "plan", "apply", "convergence", "verification", "rotation", "provisioning-chain", } STATUSES = {"pass", "fail", "not-run"} DIGEST_RE = re.compile(r"^[0-9a-f]{64}$") REVISION_RE = re.compile(r"^[0-9a-f]{7,40}$") FORBIDDEN_KEYS = { "api_key", "credential", "credentials", "password", "private_key", "secret", "secret_value", "token", "tokens", } FORBIDDEN_VALUE_PATTERNS = ( re.compile(r"-----BEGIN (?:OPENSSH|RSA|EC|AGE) PRIVATE KEY-----"), re.compile(r"\bhc_[A-Za-z0-9_-]{16,}\b"), ) class ReceiptError(ValueError): """A receipt is incomplete, ambiguous, or unsafe to retain.""" def utc_now() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def _parse_time(value: Any, label: str) -> datetime: if not isinstance(value, str): raise ReceiptError(f"{label} must be an RFC3339 string") try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError as exc: raise ReceiptError(f"{label} must be RFC3339") from exc if parsed.tzinfo is None: raise ReceiptError(f"{label} must include a timezone") return parsed def _scan_safe(value: Any, path: str = "$") -> None: if isinstance(value, dict): for key, child in value.items(): normalized = str(key).lower().replace("-", "_") if normalized in FORBIDDEN_KEYS or normalized.endswith( ("_api_key", "_credential", "_password", "_private_key", "_secret", "_token") ): raise ReceiptError(f"forbidden secret-shaped key at {path}.{key}") _scan_safe(child, f"{path}.{key}") elif isinstance(value, list): for index, child in enumerate(value): _scan_safe(child, f"{path}[{index}]") elif isinstance(value, str): for pattern in FORBIDDEN_VALUE_PATTERNS: if pattern.search(value): raise ReceiptError(f"forbidden credential-shaped value at {path}") def _require_digest(payload: dict[str, Any], key: str) -> None: if not DIGEST_RE.fullmatch(str(payload.get(key, ""))): raise ReceiptError(f"{key} must be a lowercase sha256 digest") def _validate_hosts(payload: dict[str, Any]) -> None: hosts = payload.get("hosts") profiles = payload.get("profiles") if not isinstance(hosts, list) or not hosts or not all( isinstance(host, str) and host for host in hosts ): raise ReceiptError("passing host receipts require hosts") if len(set(hosts)) != len(hosts): raise ReceiptError("hosts must be unique") if not isinstance(profiles, dict) or set(profiles) != set(hosts): raise ReceiptError("profiles must map every and only listed host") def validate_receipt(payload: Any) -> dict[str, Any]: if not isinstance(payload, dict): raise ReceiptError("receipt must be a JSON object") _scan_safe(payload) if payload.get("schema_version") != SCHEMA_VERSION: raise ReceiptError(f"schema_version must be {SCHEMA_VERSION}") try: uuid.UUID(str(payload.get("receipt_id"))) except (ValueError, TypeError, AttributeError) as exc: raise ReceiptError("receipt_id must be a UUID") from exc event_type = payload.get("event_type") status = payload.get("status") if event_type not in EVENT_TYPES: raise ReceiptError(f"event_type must be one of {sorted(EVENT_TYPES)}") if status not in STATUSES: raise ReceiptError(f"status must be one of {sorted(STATUSES)}") if not isinstance(payload.get("synthetic"), bool): raise ReceiptError("synthetic must be boolean") _parse_time(payload.get("created_at"), "created_at") if not REVISION_RE.fullmatch(str(payload.get("source_revision", ""))): raise ReceiptError("source_revision must be a 7-40 character git revision") _require_digest(payload, "inventory_sha256") if status == "pass" and event_type == "plan": _require_digest(payload, "plan_summary_sha256") elif status == "pass" and event_type == "apply": resource_ids = payload.get("provider_resource_ids") if not isinstance(resource_ids, list) or not resource_ids: raise ReceiptError("passing apply receipts require provider_resource_ids") elif status == "pass" and event_type in {"convergence", "verification"}: _validate_hosts(payload) if event_type == "verification": observed = _parse_time(payload.get("observed_at"), "observed_at") fresh = _parse_time(payload.get("fresh_until"), "fresh_until") if fresh <= observed: raise ReceiptError("fresh_until must be after observed_at") evidence = payload.get("evidence") if not isinstance(evidence, list) or len(evidence) < len(payload["hosts"]): raise ReceiptError("passing verification receipts require host evidence") for item in evidence: if not isinstance(item, dict) or not DIGEST_RE.fullmatch( str(item.get("sha256", "")) ): raise ReceiptError("verification evidence requires sha256 digests") elif status == "pass" and event_type == "rotation": if payload.get("decryption_verified") is not True: raise ReceiptError("passing rotation receipts require decryption_verified=true") for key in ("files", "before_recipients", "after_recipients"): if not isinstance(payload.get(key), list) or not payload[key]: raise ReceiptError(f"passing rotation receipts require {key}") elif status == "pass" and event_type == "provisioning-chain": if payload.get("synthetic") is not True: raise ReceiptError("combined provisioning-chain receipts are synthetic only") phases = payload.get("phases") expected = ["plan", "apply", "cloud-init", "convergence", "verification"] if not isinstance(phases, list) or [p.get("phase") for p in phases] != expected: raise ReceiptError(f"provisioning-chain phases must be {expected}") if any(phase.get("status") != "pass" for phase in phases): raise ReceiptError("passing provisioning-chain receipts require every phase to pass") return payload def load_receipt(path: Path) -> dict[str, Any]: try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise ReceiptError(f"cannot read {path}: {exc}") from exc return validate_receipt(payload) def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("receipts", nargs="+", type=Path) args = parser.parse_args() failures = [] for path in args.receipts: try: load_receipt(path) except ReceiptError as exc: failures.append(f"{path}: {exc}") if failures: print("receipt validation failed:\n- " + "\n- ".join(failures), file=sys.stderr) return 1 print(json.dumps({"ok": True, "validated": len(args.receipts)}, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())