#!/usr/bin/env python3 """Restore a verified SigNoz metadata subset into an isolated API target.""" from __future__ import annotations import argparse import hashlib import json import os import socket import stat # Child commands use fixed argv without a shell. import subprocess # nosec B404 import sys import urllib.error import urllib.parse import urllib.request from pathlib import Path from typing import Any from signoz_metadata_contract import ( DEFAULT_POLICY, ApiClient, ContractError, canonical_json_bytes, load_policy, read_json_object, receipt, semantic_items, sha256_bytes, validate_base_url, validate_asset_document, validate_identity, validate_new_private_destination, validate_token_reference, write_new_private_atomic, ) VERIFIER = Path(__file__).with_name("verify-signoz-metadata-export.py") def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() mode = parser.add_mutually_exclusive_group(required=True) mode.add_argument("--check", action="store_true") mode.add_argument("--apply", action="store_true") mode.add_argument("--verify-cleanup", action="store_true") parser.add_argument("--bundle-dir", required=True) parser.add_argument("--policy", default=str(DEFAULT_POLICY)) parser.add_argument("--target-base-url", required=True) parser.add_argument("--credential-file") parser.add_argument( "--auth-mode", choices=("signoz_api_key", "isolated_session_bearer"), default="signoz_api_key", ) parser.add_argument("--isolation-receipt", required=True) 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() EXPECTED_IMAGES = { "signoz": { "ref": "signoz/signoz:v0.113.0", "id": "sha256:c17991d10966aedcb0d4d83805a0baf5310f1553d90611629036fcfaee8d969b", }, "clickhouse": { "ref": "clickhouse/clickhouse-server:25.5.6", "id": "sha256:8248e5926d7304400e44ecdf0c7181fbde28e6a9b1d647780eca839f34c00cc6", }, "zookeeper": { "ref": "signoz/zookeeper:3.7.1", "id": "sha256:3ab0e8f032ab58f14ac1e929ac40305a4a464cf320bdfe4151d62d6922729ba0", }, } PRIMARY_LABEL = "awoooi.signoz.metadata.restore.run_id" IDENTITY_LABEL = "com.awoooi.restore.identity" def restore_identity(trace_id: str, run_id: str, work_item_id: str) -> str: return hashlib.sha256( trace_id.encode("utf-8") + b"\0" + run_id.encode("utf-8") + b"\0" + work_item_id.encode("utf-8") ).hexdigest() def expected_resources(identity: str) -> dict[str, Any]: short_id = identity[:16] prefix = f"awoooi-signoz-md-restore-{short_id}" return { "identity": identity, "prefix": prefix, "canonical_id": f"ephemeral-signoz-metadata-restore-{short_id}", "network": f"{prefix}-net", "containers": [ f"{prefix}-server", f"{prefix}-clickhouse", f"{prefix}-zookeeper", ], "volumes": [ f"{prefix}-zk-data", f"{prefix}-zk-log", f"{prefix}-ch-data", f"{prefix}-server-data", ], "workspace": f"/tmp/{prefix}-private", } def run_independent_bundle_verifier(args: argparse.Namespace) -> None: command = [ sys.executable, str(VERIFIER), "--bundle-dir", args.bundle_dir, "--policy", args.policy, "--expected-trace-id", args.trace_id, "--expected-run-id", args.run_id, "--expected-work-item-id", args.work_item_id, ] # The verifier executable and argument vector are fixed. result = subprocess.run( # nosec B603 command, text=True, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=60, check=False, ) if result.returncode != 0: raise ContractError("independent_bundle_verifier_failed") try: verifier_receipt = json.loads(result.stdout) except json.JSONDecodeError as exc: raise ContractError("independent_bundle_verifier_receipt_invalid") from exc if verifier_receipt.get("terminal") != "pass_export_verified_restore_pending": raise ContractError("independent_bundle_verifier_terminal_invalid") def validate_isolation_receipt( args: argparse.Namespace, policy: dict[str, Any], ) -> dict[str, Any]: isolation_path = Path(args.isolation_receipt) try: isolation_metadata = isolation_path.lstat() except OSError as exc: raise ContractError("isolation_receipt_unavailable") from exc if not stat.S_ISREG(isolation_metadata.st_mode): raise ContractError("isolation_receipt_not_regular_file") if ( isolation_metadata.st_uid != os.geteuid() or stat.S_IMODE(isolation_metadata.st_mode) != 0o600 ): raise ContractError("isolation_receipt_owner_or_mode_invalid") isolation = read_json_object(isolation_path, label="isolation_receipt") if isolation.get("schema") != "awoooi_signoz_metadata_restore_isolation_v1": raise ContractError("isolation_receipt_schema_invalid") for key, expected in ( ("trace_id", args.trace_id), ("run_id", args.run_id), ("work_item_id", args.work_item_id), ): if isolation.get(key) != expected: raise ContractError(f"isolation_receipt_{key}_mismatch") target = isolation.get("target") if not isinstance(target, dict): raise ContractError("isolation_target_missing") isolation_policy = policy["restore_isolation"] identity = restore_identity(args.trace_id, args.run_id, args.work_item_id) resources = expected_resources(identity) target_url = validate_base_url(args.target_base_url, loopback_only=True) if target.get("base_url_sha256") != sha256_bytes(target_url.encode("utf-8")): raise ContractError("isolation_target_url_hash_mismatch") if target.get("identity") != identity: raise ContractError("isolation_target_identity_mismatch") if target.get("prefix") != resources["prefix"]: raise ContractError("isolation_target_prefix_mismatch") if target.get("canonical_id") != resources["canonical_id"]: raise ContractError("isolation_target_canonical_id_invalid") if target.get("image_ref") != isolation_policy["required_image_ref"]: raise ContractError("isolation_target_image_ref_mismatch") if target.get("image_id") != isolation_policy["required_image_id"]: raise ContractError("isolation_target_image_id_mismatch") if target.get("images") != EXPECTED_IMAGES: raise ContractError("isolation_target_image_set_mismatch") if target.get("resources") != { "network": resources["network"], "containers": resources["containers"], "volumes": resources["volumes"], "workspace": resources["workspace"], }: raise ContractError("isolation_target_resource_names_mismatch") listener = target.get("listener") if ( not isinstance(listener, dict) or listener.get("host") != "127.0.0.1" or not isinstance(listener.get("port"), int) or listener["port"] <= 0 or listener["port"] > 65535 ): raise ContractError("isolation_target_listener_invalid") expected_url = f"http://127.0.0.1:{listener['port']}" if target_url != expected_url: raise ContractError("isolation_target_listener_url_mismatch") if args.auth_mode != "isolated_session_bearer": raise ContractError("restore_auth_mode_must_be_isolated_session_bearer") if target.get("authentication_mode") != args.auth_mode: raise ContractError("isolation_target_auth_mode_mismatch") required_values = { "image_present_locally": True, "image_pull_performed": False, "image_build_performed": False, "production_network_attached": False, "production_volume_attached": False, "ephemeral_volumes_only": True, "cleanup_controller_armed": True, "zero_residue_verifier_armed": True, "network_scope": "loopback_and_internal_only", "resource_label_key": PRIMARY_LABEL, "resource_label_value": args.run_id, "identity_label_key": IDENTITY_LABEL, "identity_label_value": identity, } for key, expected in required_values.items(): if target.get(key) != expected: raise ContractError(f"isolation_target_{key}_invalid") return isolation def load_portable_assets( bundle_dir: Path, policy: dict[str, Any], ) -> list[tuple[dict[str, Any], list[dict[str, Any]]]]: result: list[tuple[dict[str, Any], list[dict[str, Any]]]] = [] for asset in policy["assets"]: if asset["classification"] != "portable": continue path = bundle_dir / "assets" / f"{asset['id']}.json" try: document = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeError, json.JSONDecodeError) as exc: raise ContractError(f"portable_asset_read_failed_{asset['id']}") from exc result.append((asset, semantic_items(document, asset))) return result def count_actions( assets: list[tuple[dict[str, Any], list[dict[str, Any]]]], policy: dict[str, Any], ) -> int: action_count = sum(len(items) for _, items in assets) if action_count <= 0: raise ContractError("restore_source_portable_item_count_zero") if action_count > policy["limits"]["max_restore_actions"]: raise ContractError("restore_action_limit_exceeded") return action_count def make_client( args: argparse.Namespace, policy: dict[str, Any], *, read_token: bool, ) -> ApiClient | None: if not args.credential_file: if read_token: raise ContractError("credential_file_required_for_restore_apply") return None token = validate_token_reference(Path(args.credential_file), read_value=read_token) if not read_token: return None if token is None: raise ContractError("credential_value_unavailable") if args.auth_mode == "isolated_session_bearer": header_name = "Authorization" token = f"Bearer {token}" else: header_name = policy["authentication"]["header_name"] return ApiClient( base_url=validate_base_url(args.target_base_url, loopback_only=True), header_name=header_name, token=token, timeout_seconds=policy["limits"]["request_timeout_seconds"], max_response_bytes=policy["limits"]["max_response_bytes"], ) def run_check( args: argparse.Namespace, policy: dict[str, Any], assets: list[tuple[dict[str, Any], list[dict[str, Any]]]], ) -> dict[str, Any]: make_client(args, policy, read_token=False) action_count = count_actions(assets, policy) 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"isolated_target_contract_pass_restore_actions_{action_count}_" "production_write_0" ), ) def read_target_stable( client: ApiClient, assets: list[tuple[dict[str, Any], list[dict[str, Any]]]], policy: dict[str, Any], ) -> dict[str, list[dict[str, Any]]]: snapshots: list[dict[str, list[dict[str, Any]]]] = [] for _ in range(2): current: dict[str, list[dict[str, Any]]] = {} for asset, _ in assets: document = client.request_json( asset["endpoint"], query_parameters=asset.get("query_parameters"), ) validate_asset_document(document, asset, policy) current[asset["id"]] = semantic_items(document, asset) snapshots.append(current) if canonical_json_bytes(snapshots[0]) != canonical_json_bytes(snapshots[1]): raise ContractError("restore_target_drift_between_stable_reads") return snapshots[0] def run_apply( args: argparse.Namespace, policy: dict[str, Any], assets: list[tuple[dict[str, Any], list[dict[str, Any]]]], ) -> dict[str, Any]: action_count = count_actions(assets, policy) client = make_client(args, policy, read_token=True) if client is None: raise ContractError("restore_client_unavailable") before = read_target_stable(client, assets, policy) if any(before[asset["id"]] for asset, _ in assets): raise ContractError("restore_target_not_empty_fail_closed") writes = 0 for asset, items in assets: restore = asset["restore"] for item in items: client.request_write(restore["method"], restore["endpoint"], item) writes += 1 if writes != action_count: raise ContractError("restore_write_count_mismatch") after = read_target_stable(client, assets, policy) expected = {asset["id"]: items for asset, items in assets} if canonical_json_bytes(after) != canonical_json_bytes(expected): raise ContractError("restore_semantic_readback_mismatch") restored_asset_counts: dict[str, int] = {} semantic_sha256: dict[str, str] = {} for asset, items in assets: asset_id = asset["id"] restored_items = after[asset_id] expected_digest = sha256_bytes(canonical_json_bytes(items)) restored_digest = sha256_bytes(canonical_json_bytes(restored_items)) if len(restored_items) != len(items): raise ContractError(f"restore_semantic_count_mismatch_{asset_id}") if restored_digest != expected_digest: raise ContractError(f"restore_semantic_digest_mismatch_{asset_id}") restored_asset_counts[asset_id] = len(restored_items) semantic_sha256[asset_id] = restored_digest portable_semantic_sha256 = sha256_bytes( canonical_json_bytes( { asset_id: { "count": restored_asset_counts[asset_id], "sha256": semantic_sha256[asset_id], } for asset_id in sorted(restored_asset_counts) } ) ) terminal = receipt( trace_id=args.trace_id, run_id=args.run_id, work_item_id=args.work_item_id, phase="terminal", terminal="partial_degraded_cleanup_pending", detail=( f"isolated_restore_actions_{writes}_semantic_readback_pass_" "cleanup_and_zero_residue_verifier_pending" ), ) terminal["restored_asset_counts"] = restored_asset_counts terminal["restored_asset_semantic_sha256"] = semantic_sha256 terminal["portable_semantic_sha256"] = portable_semantic_sha256 terminal["completion_scope"] = "portable_assets_only" terminal["completion_claim"] = False terminal["full_sqlite_completion_claim"] = False terminal["roles_preferences_global_config_restore_claim"] = False terminal["secret_bearing_assets_restore_claim"] = False terminal["phase_receipts"] = [ receipt( trace_id=args.trace_id, run_id=args.run_id, work_item_id=args.work_item_id, phase="sensor_source", terminal="pass", detail="independently_verified_portable_export_bundle", ), 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"portable_assets_{len(assets)}_restore_actions_{writes}", ), 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="isolated_target_two_read_stable_and_empty", ), 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_restore_portable_subset_to_ephemeral_target", ), 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="high_risk_bounded_loopback_ephemeral_no_production_attachment", ), receipt( trace_id=args.trace_id, run_id=args.run_id, work_item_id=args.work_item_id, phase="check_mode", terminal="pass", detail="isolation_bundle_action_limit_and_empty_target_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"restore_actions_{writes}_delete_actions_0", ), receipt( trace_id=args.trace_id, run_id=args.run_id, work_item_id=args.work_item_id, phase="independent_post_verifier", terminal="pass", detail="two_read_semantic_parity_pass", ), receipt( trace_id=args.trace_id, run_id=args.run_id, work_item_id=args.work_item_id, phase="rollback", terminal="cleanup_pending", detail="ephemeral_target_cleanup_controller_armed", ), receipt( trace_id=args.trace_id, run_id=args.run_id, work_item_id=args.work_item_id, phase="learning_writeback", terminal="pending", detail="requires_zero_residue_terminal_before_learning_ack", ), ] return terminal def docker_label_count(resource: str, label: str) -> int: command_by_resource = { "container": ["docker", "ps", "-a", "--quiet", "--filter", f"label={label}"], "volume": ["docker", "volume", "ls", "--quiet", "--filter", f"label={label}"], "network": ["docker", "network", "ls", "--quiet", "--filter", f"label={label}"], } try: # The Docker inventory command is selected from the fixed map above. result = subprocess.run( # nosec B603 command_by_resource[resource], text=True, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=15, check=False, ) except (OSError, subprocess.TimeoutExpired) as exc: raise ContractError(f"cleanup_{resource}_inventory_failed") from exc if result.returncode != 0: raise ContractError(f"cleanup_{resource}_inventory_failed") return len([line for line in result.stdout.splitlines() if line.strip()]) def docker_inspect(resource: str, name: str) -> dict[str, Any] | None: command = ["docker", resource, "inspect", name] try: result = subprocess.run( # nosec B603 command, text=True, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=15, check=False, ) except (OSError, subprocess.TimeoutExpired) as exc: raise ContractError(f"cleanup_{resource}_inventory_failed") from exc if result.returncode != 0: return None try: value = json.loads(result.stdout) except json.JSONDecodeError as exc: raise ContractError(f"cleanup_{resource}_inventory_invalid") from exc if not isinstance(value, list) or len(value) != 1 or not isinstance(value[0], dict): raise ContractError(f"cleanup_{resource}_inventory_invalid") return value[0] def require_target_tcp_unreachable(base_url: str) -> None: base_url = validate_base_url(base_url, loopback_only=True) parsed = urllib.parse.urlsplit(base_url) if parsed.hostname != "127.0.0.1" or parsed.port is None: raise ContractError("cleanup_target_listener_invalid") try: with socket.create_connection((parsed.hostname, parsed.port), timeout=2): raise ContractError("cleanup_target_tcp_listener_still_present") except ContractError: raise except OSError: return def cleanup_production_snapshot() -> dict[str, dict[str, Any]]: expected = { "signoz": EXPECTED_IMAGES["signoz"], "signoz-clickhouse": EXPECTED_IMAGES["clickhouse"], "signoz-zookeeper-1": EXPECTED_IMAGES["zookeeper"], "signoz-otel-collector": { "ref": "signoz/signoz-otel-collector:v0.144.1", "id": "sha256:47cbed488288df7bac6aab3665275bec6d6e24fe5c545c66914827a891e538d8", }, } snapshot: dict[str, dict[str, Any]] = {} for name, image in expected.items(): data = docker_inspect("container", name) if data is None: raise ContractError("cleanup_production_container_missing") config = data.get("Config") state = data.get("State") if not isinstance(config, dict) or not isinstance(state, dict): raise ContractError("cleanup_production_container_shape_invalid") row = { "id": data.get("Id"), "image_id": data.get("Image"), "image_ref": config.get("Image"), "running": state.get("Running"), "started_at": state.get("StartedAt"), "restart_count": data.get("RestartCount"), } if row["image_id"] != image["id"] or row["image_ref"] != image["ref"]: raise ContractError("cleanup_production_image_identity_invalid") snapshot[name] = row return snapshot def cleanup_image_inventory() -> list[str]: try: result = subprocess.run( # nosec B603 [ "docker", "image", "ls", "--no-trunc", "--format={{.ID}}|{{.Repository}}:{{.Tag}}", ], text=True, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=15, check=False, ) except (OSError, subprocess.TimeoutExpired) as exc: raise ContractError("cleanup_image_inventory_failed") from exc if result.returncode != 0: raise ContractError("cleanup_image_inventory_failed") rows = sorted(line for line in result.stdout.splitlines() if line.strip()) if not rows: raise ContractError("cleanup_image_inventory_empty") return rows def run_cleanup_verifier( args: argparse.Namespace, policy: dict[str, Any], isolation: dict[str, Any], ) -> dict[str, Any]: identity = restore_identity(args.trace_id, args.run_id, args.work_item_id) resources = expected_resources(identity) for name in resources["containers"]: if docker_inspect("container", name) is not None: raise ContractError("cleanup_exact_container_residue_present") for name in resources["volumes"]: if docker_inspect("volume", name) is not None: raise ContractError("cleanup_exact_volume_residue_present") if docker_inspect("network", resources["network"]) is not None: raise ContractError("cleanup_exact_network_residue_present") labels = ( f"{PRIMARY_LABEL}={args.run_id}", f"{IDENTITY_LABEL}={identity}", ) for label in labels: counts = { resource: docker_label_count(resource, label) for resource in ("container", "volume", "network") } if any(counts.values()): raise ContractError("cleanup_resource_label_residue_present") require_target_tcp_unreachable(args.target_base_url) workspace = Path(resources["workspace"]) if workspace.exists() or workspace.is_symlink(): raise ContractError("cleanup_private_workspace_residue_present") production_before = isolation.get("production_before") if not isinstance(production_before, dict): raise ContractError("cleanup_production_before_missing") if cleanup_production_snapshot() != production_before: raise ContractError("cleanup_production_identity_continuity_failed") image_inventory_before = isolation.get("image_inventory_before") if not isinstance(image_inventory_before, list): raise ContractError("cleanup_image_inventory_before_missing") if cleanup_image_inventory() != image_inventory_before: raise ContractError("cleanup_image_inventory_drift") return receipt( trace_id=args.trace_id, run_id=args.run_id, work_item_id=args.work_item_id, phase="terminal", terminal="pass_zero_residue_verified", detail=( "exact_container_0_volume_0_network_0_label_inventory_0_" "tcp_listener_0_workspace_0_production_identity_unchanged" ), ) def write_receipt(path: Path, value: dict[str, Any]) -> None: write_new_private_atomic( path, canonical_json_bytes(value), label="restore_terminal_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.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.target_base_url, loopback_only=True) if args.apply and not args.receipt_file: raise ContractError("receipt_file_required_for_restore_apply") if args.receipt_file: validate_new_private_destination( Path(args.receipt_file), label="restore_terminal_receipt" ) policy = load_policy(Path(args.policy)) run_independent_bundle_verifier(args) isolation = validate_isolation_receipt(args, policy) assets = load_portable_assets(Path(args.bundle_dir), policy) if args.check: result = run_check(args, policy, assets) elif args.apply: result = run_apply(args, policy, assets) else: result = run_cleanup_verifier(args, policy, isolation) if args.receipt_file: write_receipt(Path(args.receipt_file), result) emit_result(result) except (ContractError, OSError, subprocess.TimeoutExpired) 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") if args.check: failure_terminal = "check_failed_no_write" elif args.apply: failure_terminal = "failed_cleanup_required" else: failure_terminal = "cleanup_verifier_failed_residue_unknown" failure = receipt( trace_id=args.trace_id, run_id=args.run_id, work_item_id=args.work_item_id, phase="terminal", terminal=failure_terminal, detail=f"metadata_restore_drill_failed_{exc}", ) if args.receipt_file: write_receipt(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())