#!/usr/bin/env python3 """Create a stable, secret-scanned SigNoz metadata API export bundle.""" from __future__ import annotations import argparse import json import os import shutil import sys import tempfile from pathlib import Path from typing import Any from signoz_metadata_contract import ( DEFAULT_POLICY, ApiClient, ContractError, canonical_json_bytes, load_policy, receipt, require_no_symlink_components, sha256_bytes, sha256_file, utc_now, validate_asset_document, validate_base_url, validate_identity, validate_new_private_destination, validate_token_reference, write_new_private_atomic, write_private, ) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description=( "Export an allowlisted SigNoz metadata subset without raw SQLite or " "Docker volume access." ) ) mode = parser.add_mutually_exclusive_group(required=True) mode.add_argument("--check", action="store_true") mode.add_argument("--apply", action="store_true") parser.add_argument("--base-url", required=True) parser.add_argument("--output-dir", required=True) parser.add_argument("--credential-file") parser.add_argument("--policy", default=str(DEFAULT_POLICY)) parser.add_argument("--trace-id", required=True) parser.add_argument("--run-id", required=True) parser.add_argument("--work-item-id", default="P0-OBS-002") parser.add_argument("--receipt-file") return parser.parse_args() def validate_destination(path: Path) -> None: if not path.is_absolute(): raise ContractError("output_dir_must_be_absolute") if path.exists() or path.is_symlink(): raise ContractError("output_dir_must_not_exist") parent = path.parent require_no_symlink_components(parent, label="output_parent") try: metadata = parent.lstat() except OSError as exc: raise ContractError("output_parent_unavailable") from exc if not parent.is_dir() or parent.is_symlink(): raise ContractError("output_parent_must_be_real_directory") if metadata.st_uid != os.geteuid() or not os.access(parent, os.W_OK | os.X_OK): raise ContractError("output_parent_not_owned_and_writable") def emit_stdout(value: dict[str, Any]) -> None: print(json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))) def run_check(args: argparse.Namespace, policy: dict[str, Any]) -> dict[str, Any]: if args.credential_file: validate_token_reference(Path(args.credential_file), read_value=False) return receipt( trace_id=args.trace_id, run_id=args.run_id, work_item_id=args.work_item_id, phase="terminal", terminal="check_pass_no_write", detail=( f"policy_assets_{len(policy['assets'])}_raw_sqlite_0_" "raw_volume_0_output_write_0" ), ) def stable_read( client: ApiClient, policy: dict[str, Any], ) -> tuple[dict[str, Any], dict[str, int]]: pass_documents: list[dict[str, Any]] = [] pass_counts: list[dict[str, int]] = [] for _ in range(policy["limits"]["stable_read_count"]): documents: dict[str, Any] = {} counts: dict[str, int] = {} for asset in policy["assets"]: document = client.request_json(asset["endpoint"]) counts[asset["id"]] = validate_asset_document(document, asset, policy) documents[asset["id"]] = document pass_documents.append(documents) pass_counts.append(counts) if canonical_json_bytes(pass_documents[0]) != canonical_json_bytes( pass_documents[1] ): raise ContractError("source_metadata_drift_between_stable_reads") if pass_counts[0] != pass_counts[1]: raise ContractError("source_item_count_drift_between_stable_reads") return pass_documents[0], pass_counts[0] def create_bundle( args: argparse.Namespace, policy_path: Path, policy: dict[str, Any], documents: dict[str, Any], counts: dict[str, int], ) -> dict[str, Any]: output_dir = Path(args.output_dir) temp_dir: Path | None = Path( tempfile.mkdtemp(prefix=f".{output_dir.name}.partial.", dir=output_dir.parent) ) temp_dir.chmod(0o700) assets_dir = temp_dir / "assets" assets_dir.mkdir(mode=0o700) try: asset_receipts: list[dict[str, Any]] = [] for asset in policy["assets"]: asset_id = asset["id"] payload = canonical_json_bytes(documents[asset_id]) asset_path = assets_dir / f"{asset_id}.json" write_private(asset_path, payload) asset_receipts.append( { "id": asset_id, "classification": asset["classification"], "endpoint": asset["endpoint"], "file": f"assets/{asset_id}.json", "sha256": sha256_bytes(payload), "size_bytes": len(payload), "item_count": counts[asset_id], } ) portable_count = sum( item["classification"] == "portable" for item in asset_receipts ) manifest = { "schema": "awoooi_signoz_metadata_export_bundle_v1", "trace_id": args.trace_id, "run_id": args.run_id, "work_item_id": args.work_item_id, "created_at": utc_now(), "source_runtime_id": policy["source_runtime"]["canonical_id"], "source_version": policy["source_runtime"]["signoz_version"], "source_base_url_sha256": sha256_bytes( validate_base_url(args.base_url).encode("utf-8") ), "policy_sha256": sha256_file(policy_path), "stable_read_count": policy["limits"]["stable_read_count"], "stable_read_terminal": "pass", "raw_sqlite_read": False, "raw_volume_read": False, "sensitive_value_persisted": False, "assets": asset_receipts, "coverage": { "portable_asset_count": portable_count, "export_only_asset_count": len(asset_receipts) - portable_count, "forbidden_endpoint_count": len(policy["forbidden_endpoints"]), "full_sqlite_coverage": False, }, "terminal": "portable_metadata_export_created_restore_unverified", "completion_claim": False, } write_private(temp_dir / "manifest.json", canonical_json_bytes(manifest)) receipt_rows = [ receipt( trace_id=args.trace_id, run_id=args.run_id, work_item_id=args.work_item_id, phase="sensor_source", terminal="pass", detail="allowlisted_api_get_two_stable_reads", ), receipt( trace_id=args.trace_id, run_id=args.run_id, work_item_id=args.work_item_id, phase="normalized_asset_identity", terminal="pass", detail=f"canonical_asset_count_{len(asset_receipts)}", ), receipt( trace_id=args.trace_id, run_id=args.run_id, work_item_id=args.work_item_id, phase="source_of_truth_diff", terminal="pass", detail="second_read_matches_first_read_canonical_hash", ), receipt( trace_id=args.trace_id, run_id=args.run_id, work_item_id=args.work_item_id, phase="ai_decision", terminal="pass", detail="candidate_action_export_portable_metadata_subset", ), receipt( trace_id=args.trace_id, run_id=args.run_id, work_item_id=args.work_item_id, phase="risk_policy_decision", terminal="pass", detail="low_risk_read_only_source_no_raw_sqlite_no_secret_persistence", ), receipt( trace_id=args.trace_id, run_id=args.run_id, work_item_id=args.work_item_id, phase="check_mode", terminal="pass", detail="policy_destination_and_credential_reference_preconditions_pass", ), receipt( trace_id=args.trace_id, run_id=args.run_id, work_item_id=args.work_item_id, phase="bounded_execution", terminal="pass", detail=f"private_atomic_bundle_assets_{len(asset_receipts)}", ), receipt( trace_id=args.trace_id, run_id=args.run_id, work_item_id=args.work_item_id, phase="rollback", terminal="not_required", detail="production_source_read_only_output_created_atomically", ), receipt( trace_id=args.trace_id, run_id=args.run_id, work_item_id=args.work_item_id, phase="terminal", terminal="partial_degraded", detail="portable_export_created_independent_restore_verifier_pending", ), ] receipt_payload = b"".join(canonical_json_bytes(item) for item in receipt_rows) write_private(temp_dir / "receipts.jsonl", receipt_payload) os.replace(temp_dir, output_dir) try: parent_descriptor = os.open(output_dir.parent, os.O_RDONLY) try: os.fsync(parent_descriptor) finally: os.close(parent_descriptor) except OSError: pass temp_dir = None return manifest finally: if temp_dir is not None and temp_dir.exists(): shutil.rmtree(temp_dir) def run_apply( args: argparse.Namespace, policy_path: Path, policy: dict[str, Any] ) -> dict[str, Any]: if not args.receipt_file: raise ContractError("receipt_file_required_for_apply") if not args.credential_file: raise ContractError("credential_file_required_for_apply") token = validate_token_reference(Path(args.credential_file), read_value=True) if token is None: raise ContractError("credential_value_unavailable") client = ApiClient( base_url=args.base_url, header_name=policy["authentication"]["header_name"], token=token, timeout_seconds=policy["limits"]["request_timeout_seconds"], max_response_bytes=policy["limits"]["max_response_bytes"], ) documents, counts = stable_read(client, policy) manifest = create_bundle(args, policy_path, policy, documents, counts) return receipt( trace_id=args.trace_id, run_id=args.run_id, work_item_id=args.work_item_id, phase="terminal", terminal="partial_degraded", detail=( "bundle_created_manifest_hash_" f"{sha256_bytes(canonical_json_bytes(manifest))}_restore_pending" ), ) def persist_optional_receipt(args: argparse.Namespace, value: dict[str, Any]) -> None: if args.receipt_file: write_new_private_atomic( Path(args.receipt_file), canonical_json_bytes(value), label="export_terminal_receipt", ) def main() -> int: args = parse_args() try: validate_identity(args.trace_id, label="trace_id") validate_identity(args.run_id, label="run_id") validate_identity(args.work_item_id, label="work_item_id") validate_base_url(args.base_url) validate_destination(Path(args.output_dir)) if args.apply and not args.receipt_file: raise ContractError("receipt_file_required_for_apply") if args.receipt_file: validate_new_private_destination( Path(args.receipt_file), label="export_terminal_receipt" ) policy_path = Path(args.policy) policy = load_policy(policy_path) if policy["work_item_id"] != args.work_item_id: raise ContractError("work_item_policy_mismatch") if args.check: result = run_check(args, policy) else: result = run_apply(args, policy_path, policy) persist_optional_receipt(args, result) emit_stdout(result) except (ContractError, OSError) as exc: try: validate_identity(args.trace_id, label="trace_id") validate_identity(args.run_id, label="run_id") validate_identity(args.work_item_id, label="work_item_id") failure = receipt( trace_id=args.trace_id, run_id=args.run_id, work_item_id=args.work_item_id, phase="terminal", terminal=( "check_failed_no_write" if args.check else "failed_export_not_closable" ), detail=( f"metadata_export_failed_{exc}_output_" f"{'created_unverified' if Path(args.output_dir).is_dir() else 'absent'}" ), ) persist_optional_receipt(args, failure) emit_stdout(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())