#!/usr/bin/env python3 """Independently verify a SigNoz metadata export bundle.""" from __future__ import annotations import argparse import json import os import stat import sys from pathlib import Path from typing import Any from signoz_metadata_contract import ( DEFAULT_POLICY, SHA256, ContractError, canonical_json_bytes, load_policy, receipt, require_no_symlink_components, sha256_file, validate_asset_document, validate_identity, write_new_private_atomic, ) EXPECTED_PHASES = [ "sensor_source", "normalized_asset_identity", "source_of_truth_diff", "ai_decision", "risk_policy_decision", "check_mode", "bounded_execution", "rollback", "terminal", ] def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--bundle-dir", required=True) parser.add_argument("--policy", default=str(DEFAULT_POLICY)) parser.add_argument("--expected-trace-id", required=True) parser.add_argument("--expected-run-id", required=True) parser.add_argument("--expected-work-item-id", default="P0-OBS-002") parser.add_argument("--receipt-file") return parser.parse_args() def require_private_regular(path: Path, *, label: str) -> None: require_no_symlink_components(path, label=label) try: metadata = path.lstat() except OSError as exc: raise ContractError(f"{label}_unavailable") from exc if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode): raise ContractError(f"{label}_not_private_regular_file") if metadata.st_uid != os.geteuid() or stat.S_IMODE(metadata.st_mode) != 0o600: raise ContractError(f"{label}_owner_or_mode_invalid") def require_private_directory(path: Path, *, label: str) -> None: require_no_symlink_components(path, label=label) try: metadata = path.lstat() except OSError as exc: raise ContractError(f"{label}_unavailable") from exc if not stat.S_ISDIR(metadata.st_mode): raise ContractError(f"{label}_not_directory") if metadata.st_uid != os.geteuid() or stat.S_IMODE(metadata.st_mode) != 0o700: raise ContractError(f"{label}_owner_or_mode_invalid") def load_bundle_json(path: Path, *, label: str) -> tuple[Any, bytes]: require_private_regular(path, label=label) try: raw = path.read_bytes() value = json.loads(raw) except (OSError, UnicodeError, json.JSONDecodeError) as exc: raise ContractError(f"{label}_json_invalid") from exc if raw != canonical_json_bytes(value): raise ContractError(f"{label}_not_canonical_json") return value, raw def validate_manifest_identity( manifest: dict[str, Any], args: argparse.Namespace ) -> None: expected = { "trace_id": args.expected_trace_id, "run_id": args.expected_run_id, "work_item_id": args.expected_work_item_id, } for key, value in expected.items(): if manifest.get(key) != value: raise ContractError(f"manifest_{key}_mismatch") if manifest.get("schema") != "awoooi_signoz_metadata_export_bundle_v1": raise ContractError("manifest_schema_invalid") if manifest.get("stable_read_count") != 2: raise ContractError("manifest_stable_read_count_invalid") if manifest.get("stable_read_terminal") != "pass": raise ContractError("manifest_stable_read_not_pass") if manifest.get("raw_sqlite_read") is not False: raise ContractError("manifest_raw_sqlite_read_not_false") if manifest.get("raw_volume_read") is not False: raise ContractError("manifest_raw_volume_read_not_false") if manifest.get("sensitive_value_persisted") is not False: raise ContractError("manifest_secret_persistence_not_false") if manifest.get("completion_claim") is not False: raise ContractError("manifest_illegal_completion_claim") source_url_hash = manifest.get("source_base_url_sha256") if not isinstance(source_url_hash, str) or not SHA256.fullmatch(source_url_hash): raise ContractError("manifest_source_url_hash_invalid") if manifest.get("terminal") != ( "portable_metadata_export_created_restore_unverified" ): raise ContractError("manifest_terminal_invalid") def verify_assets( bundle_dir: Path, manifest: dict[str, Any], policy: dict[str, Any], ) -> tuple[int, int]: manifest_assets = manifest.get("assets") if not isinstance(manifest_assets, list): raise ContractError("manifest_assets_invalid") policy_assets = policy["assets"] if [item.get("id") for item in manifest_assets if isinstance(item, dict)] != [ item["id"] for item in policy_assets ]: raise ContractError("manifest_asset_identity_or_order_mismatch") assets_dir = bundle_dir / "assets" require_private_directory(assets_dir, label="assets_dir") expected_asset_files = {f"{item['id']}.json" for item in policy_assets} if {item.name for item in assets_dir.iterdir()} != expected_asset_files: raise ContractError("bundle_assets_tree_contains_unexpected_entries") total_items = 0 portable_items = 0 for asset, recorded in zip(policy_assets, manifest_assets, strict=True): if not isinstance(recorded, dict): raise ContractError("manifest_asset_entry_invalid") asset_id = asset["id"] expected_relative = f"assets/{asset_id}.json" if recorded.get("file") != expected_relative: raise ContractError(f"manifest_asset_file_mismatch_{asset_id}") if recorded.get("endpoint") != asset["endpoint"]: raise ContractError(f"manifest_asset_endpoint_mismatch_{asset_id}") if recorded.get("classification") != asset["classification"]: raise ContractError(f"manifest_asset_classification_mismatch_{asset_id}") asset_path = bundle_dir / "assets" / f"{asset_id}.json" value, raw = load_bundle_json(asset_path, label=f"asset_{asset_id}") if recorded.get("sha256") != sha256_file(asset_path): raise ContractError(f"manifest_asset_hash_mismatch_{asset_id}") if recorded.get("size_bytes") != len(raw): raise ContractError(f"manifest_asset_size_mismatch_{asset_id}") item_count = validate_asset_document(value, asset, policy) if recorded.get("item_count") != item_count: raise ContractError(f"manifest_asset_item_count_mismatch_{asset_id}") total_items += item_count if asset["classification"] == "portable": portable_items += item_count return total_items, portable_items def verify_export_receipts( path: Path, args: argparse.Namespace, ) -> None: require_private_regular(path, label="export_receipts") try: raw = path.read_bytes() lines = raw.decode("utf-8").splitlines() rows = [json.loads(line) for line in lines] except (OSError, UnicodeError, json.JSONDecodeError) as exc: raise ContractError("export_receipts_invalid") from exc if len(rows) != len(EXPECTED_PHASES): raise ContractError("export_receipt_count_invalid") if raw != b"".join(canonical_json_bytes(row) for row in rows): raise ContractError("export_receipts_not_canonical_jsonl") if [row.get("phase") for row in rows if isinstance(row, dict)] != EXPECTED_PHASES: raise ContractError("export_receipt_phase_order_invalid") for row in rows: if not isinstance(row, dict): raise ContractError("export_receipt_not_object") if row.get("schema") != "awoooi_signoz_metadata_receipt_v1": raise ContractError("export_receipt_schema_invalid") if row.get("trace_id") != args.expected_trace_id: raise ContractError("export_receipt_trace_id_mismatch") if row.get("run_id") != args.expected_run_id: raise ContractError("export_receipt_run_id_mismatch") if row.get("work_item_id") != args.expected_work_item_id: raise ContractError("export_receipt_work_item_id_mismatch") if rows[-1].get("terminal") != "partial_degraded": raise ContractError("export_receipt_terminal_must_remain_partial") def write_receipt_file(path: Path, value: dict[str, Any]) -> None: write_new_private_atomic( path, canonical_json_bytes(value), label="verifier_receipt", ) def emit_result(value: dict[str, Any]) -> None: print( json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ) ) def main() -> int: args = parse_args() try: validate_identity(args.expected_trace_id, label="trace_id") validate_identity(args.expected_run_id, label="run_id") validate_identity(args.expected_work_item_id, label="work_item_id") policy_path = Path(args.policy) policy = load_policy(policy_path) bundle_dir = Path(args.bundle_dir) if not bundle_dir.is_absolute(): raise ContractError("bundle_dir_must_be_absolute_real_directory") require_private_directory(bundle_dir, label="bundle_dir") if {item.name for item in bundle_dir.iterdir()} != { "assets", "manifest.json", "receipts.jsonl", }: raise ContractError("bundle_tree_contains_unexpected_entries") manifest_value, _ = load_bundle_json( bundle_dir / "manifest.json", label="manifest" ) if not isinstance(manifest_value, dict): raise ContractError("manifest_not_object") validate_manifest_identity(manifest_value, args) if manifest_value.get("policy_sha256") != sha256_file(policy_path): raise ContractError("manifest_policy_hash_mismatch") if ( manifest_value.get("source_version") != (policy["source_runtime"]["signoz_version"]) ): raise ContractError("manifest_source_version_mismatch") total_items, portable_items = verify_assets(bundle_dir, manifest_value, policy) coverage = manifest_value.get("coverage") if not isinstance(coverage, dict): raise ContractError("manifest_coverage_invalid") portable_assets = sum( item["classification"] == "portable" for item in policy["assets"] ) if coverage != { "portable_asset_count": portable_assets, "export_only_asset_count": len(policy["assets"]) - portable_assets, "forbidden_endpoint_count": len(policy["forbidden_endpoints"]), "full_sqlite_coverage": False, }: raise ContractError("manifest_coverage_mismatch") verify_export_receipts(bundle_dir / "receipts.jsonl", args) result = receipt( trace_id=args.expected_trace_id, run_id=args.expected_run_id, work_item_id=args.expected_work_item_id, phase="independent_export_verifier", terminal="pass_export_verified_restore_pending", detail=( f"asset_count_{len(policy['assets'])}_items_{total_items}_" f"portable_items_{portable_items}_secret_scan_pass" ), ) if args.receipt_file: write_receipt_file(Path(args.receipt_file), result) emit_result(result) except (ContractError, OSError) as exc: try: validate_identity(args.expected_trace_id, label="trace_id") validate_identity(args.expected_run_id, label="run_id") validate_identity(args.expected_work_item_id, label="work_item_id") failure = receipt( trace_id=args.expected_trace_id, run_id=args.expected_run_id, work_item_id=args.expected_work_item_id, phase="independent_export_verifier", terminal="failed", detail=f"bundle_verification_failed_{exc}", ) if args.receipt_file: write_receipt_file(Path(args.receipt_file), failure) emit_result(failure) except (ContractError, OSError) as receipt_exc: print(f"RECEIPT_ERROR={receipt_exc}", file=sys.stderr) print(f"ERROR={exc}", file=sys.stderr) return 1 return 0 if __name__ == "__main__": raise SystemExit(main())