From 1730c5bb71cd50f3353e8f3f16652c3c687781bd Mon Sep 17 00:00:00 2001 From: ogt Date: Wed, 15 Jul 2026 13:57:13 +0800 Subject: [PATCH] feat(observability): add safe SigNoz metadata backup lane --- config/signoz/metadata-export-policy.json | 204 ++++++ ...rganization-canonical-route-migration.yaml | 12 +- .../p0-obs-002-post-closure-regression.yaml | 82 ++- scripts/backup/signoz-metadata-export.py | 371 +++++++++++ .../backup/signoz-metadata-restore-drill.py | 510 +++++++++++++++ scripts/backup/signoz_metadata_contract.py | 551 ++++++++++++++++ .../test_signoz_metadata_export_contract.py | 602 ++++++++++++++++++ .../backup/verify-signoz-metadata-export.py | 308 +++++++++ .../ops/deploy-signoz-metadata-toolchain.sh | 505 +++++++++++++++ .../test_signoz_metadata_toolchain_deploy.py | 179 ++++++ .../signoz-canonical-route-preflight.py | 93 ++- .../test_signoz_canonical_route_migration.py | 96 ++- 12 files changed, 3485 insertions(+), 28 deletions(-) create mode 100644 config/signoz/metadata-export-policy.json create mode 100755 scripts/backup/signoz-metadata-export.py create mode 100755 scripts/backup/signoz-metadata-restore-drill.py create mode 100755 scripts/backup/signoz_metadata_contract.py create mode 100644 scripts/backup/tests/test_signoz_metadata_export_contract.py create mode 100755 scripts/backup/verify-signoz-metadata-export.py create mode 100755 scripts/ops/deploy-signoz-metadata-toolchain.sh create mode 100644 scripts/ops/tests/test_signoz_metadata_toolchain_deploy.py diff --git a/config/signoz/metadata-export-policy.json b/config/signoz/metadata-export-policy.json new file mode 100644 index 000000000..d26b0ac77 --- /dev/null +++ b/config/signoz/metadata-export-policy.json @@ -0,0 +1,204 @@ +{ + "schema": "awoooi_signoz_metadata_export_policy_v1", + "policy_version": "2026-07-15", + "work_item_id": "P0-OBS-002", + "source_runtime": { + "canonical_id": "signoz-110-control-plane", + "signoz_version": "v0.113.0", + "minimum_python_version": "3.10", + "raw_database_access_allowed": false, + "raw_volume_access_allowed": false, + "github_supply_chain_allowed": false + }, + "authentication": { + "header_name": "SIGNOZ-API-KEY", + "secret_value_in_arguments_allowed": false, + "secret_reference_type": "root_owned_regular_file", + "accepted_file_modes": [ + "0400", + "0600" + ] + }, + "limits": { + "request_timeout_seconds": 15, + "max_response_bytes": 4194304, + "max_asset_count": 16, + "max_items_per_asset": 1000, + "max_restore_actions": 500, + "stable_read_count": 2 + }, + "forbidden_key_names": [ + "access_key", + "access_token", + "api_key", + "authorization", + "client_secret", + "cookie", + "credential", + "password", + "private_key", + "refresh_token", + "routing_key", + "secret", + "service_account_json", + "session", + "token", + "webhook_url" + ], + "forbidden_value_patterns": [ + "-----BEGIN [A-Z ]*PRIVATE KEY-----", + "(?i)\\bBearer\\s+[A-Za-z0-9._~+/-]{16,}={0,2}\\b", + "(?i)https?://[^/@\\s]+:[^/@\\s]+@" + ], + "forbidden_endpoints": [ + { + "id": "personal_access_tokens", + "path": "/api/v1/pats", + "reason": "response may contain an API token" + }, + { + "id": "authentication_domains", + "path": "/api/v1/domains", + "reason": "response may contain OIDC or Google client secrets" + }, + { + "id": "notification_channels", + "path": "/api/v1/channels", + "reason": "response may contain webhook and routing credentials" + }, + { + "id": "users_and_credentials", + "path": "/api/v1/user", + "reason": "credential lifecycle is reconstructed, not exported" + }, + { + "id": "invitations", + "path": "/api/v1/invite", + "reason": "invitation tokens are non-portable credentials" + }, + { + "id": "sessions", + "path": "/api/v1/sessions", + "reason": "sessions are explicitly outside backup scope" + } + ], + "assets": [ + { + "id": "dashboards", + "endpoint": "/api/v1/dashboards", + "method": "GET", + "response_pointer": "/data", + "response_kind": "collection", + "classification": "portable", + "restore": { + "method": "POST", + "endpoint": "/api/v1/dashboards", + "strategy": "collection_items", + "strip_top_level_fields": [ + "createdAt", + "id", + "orgId", + "updatedAt", + "uuid" + ] + } + }, + { + "id": "alert_rules", + "endpoint": "/api/v1/rules", + "method": "GET", + "response_pointer": "/data", + "response_kind": "collection", + "classification": "portable", + "restore": { + "method": "POST", + "endpoint": "/api/v1/rules", + "strategy": "collection_items", + "strip_top_level_fields": [ + "createdAt", + "id", + "orgId", + "updatedAt" + ] + } + }, + { + "id": "explorer_views", + "endpoint": "/api/v1/explorer/views", + "method": "GET", + "response_pointer": "/data", + "response_kind": "collection", + "classification": "portable", + "restore": { + "method": "POST", + "endpoint": "/api/v1/explorer/views", + "strategy": "collection_items", + "strip_top_level_fields": [ + "createdAt", + "id", + "orgId", + "updatedAt", + "userId" + ] + } + }, + { + "id": "roles", + "endpoint": "/api/v1/roles", + "method": "GET", + "response_pointer": "/data", + "response_kind": "collection", + "classification": "export_only", + "coverage_gap": "role relation objects require a separately bounded enumerator" + }, + { + "id": "organization_preferences", + "endpoint": "/api/v1/org/preferences", + "method": "GET", + "response_pointer": "/data", + "response_kind": "collection", + "classification": "export_only", + "coverage_gap": "preference names require a reviewed per-name restore allowlist" + }, + { + "id": "user_preferences", + "endpoint": "/api/v1/user/preferences", + "method": "GET", + "response_pointer": "/data", + "response_kind": "collection", + "classification": "export_only", + "coverage_gap": "principal-scoped preferences are not portable without identity reconstruction" + }, + { + "id": "global_config", + "endpoint": "/api/v1/global/config", + "method": "GET", + "response_pointer": "/data", + "response_kind": "document", + "classification": "export_only", + "coverage_gap": "readback-only configuration has no supported write endpoint" + } + ], + "restore_isolation": { + "target_url_scope": "loopback_only", + "required_image_ref": "signoz/signoz:v0.113.0", + "image_id_sha256_required": true, + "image_pull_allowed": false, + "production_network_attachment_allowed": false, + "production_volume_attachment_allowed": false, + "ephemeral_volumes_only": true, + "cleanup_controller_required": true, + "zero_residue_verifier_required": true, + "resource_label_key": "awoooi.signoz.metadata.restore.run_id" + }, + "completion_contract": { + "portable_export_alone_is_completion": false, + "independent_bundle_verifier_required": true, + "isolated_restore_semantic_readback_required": true, + "zero_residue_terminal_required": true, + "durable_terminal_receipt_required_for_apply": true, + "failure_terminal_receipt_required": true, + "secret_assets_reissue_required": true, + "full_sqlite_completion_claim_allowed": false + } +} diff --git a/ops/signoz/organization-canonical-route-migration.yaml b/ops/signoz/organization-canonical-route-migration.yaml index 4810622a4..d0b2a6426 100644 --- a/ops/signoz/organization-canonical-route-migration.yaml +++ b/ops/signoz/organization-canonical-route-migration.yaml @@ -202,7 +202,17 @@ runtime_observations: clickhouse_native_failed_artifact_retention_status: pending_bounded_retention_decision clickhouse_native_resume_inventory_status: pending_durable_submitted_inventory_contract signoz_sqlite_application_consistent_backup_status: pending - service_registry_runtime_mirror_status: partial_degraded + signoz_metadata_export_source_contract_status: implemented_tested_not_deployed + signoz_metadata_export_policy: config/signoz/metadata-export-policy.json + signoz_metadata_exporter: scripts/backup/signoz-metadata-export.py + signoz_metadata_export_verifier: scripts/backup/verify-signoz-metadata-export.py + signoz_metadata_restore_driver: scripts/backup/signoz-metadata-restore-drill.py + signoz_metadata_controlled_deployer: scripts/ops/deploy-signoz-metadata-toolchain.sh + signoz_metadata_runtime_export_terminal: pending_authenticated_stable_export + signoz_metadata_runtime_restore_terminal: pending_isolated_same_version_target + signoz_metadata_zero_residue_terminal: pending + signoz_metadata_full_sqlite_completion_claim: false + service_registry_runtime_mirror_status: source_manifest_exact_production_readback_pending service_registry_runtime_mirror_work_item_id: P0-OBS-002-ASSET-DRIFT-001 github_freeze_legacy_asset_status: pending_controlled_retirement github_freeze_legacy_asset_work_item_id: P0-OBS-002-ASSET-DRIFT-002 diff --git a/ops/signoz/p0-obs-002-post-closure-regression.yaml b/ops/signoz/p0-obs-002-post-closure-regression.yaml index 48f1f5f8c..67292d6df 100644 --- a/ops/signoz/p0-obs-002-post-closure-regression.yaml +++ b/ops/signoz/p0-obs-002-post-closure-regression.yaml @@ -58,21 +58,20 @@ asset_reconciliation: source: ops/config/service-registry.yaml runtime_mirror: k8s/awoooi-prod/15-service-registry-configmap.yaml source_service_count: 27 - runtime_mirror_service_count: 24 - exact_shared_service_count: 24 + runtime_mirror_service_count: 27 + exact_shared_service_count: 27 shared_service_drift_count: 0 - source_only_services: - - bitan-app - - sentry - - signoz + source_only_services: [] + source_runtime_manifest_exact: true + production_runtime_readback_status: pending_current_sha_readback signoz_clickhouse_exact_match: true live_signoz_query_host: 192.168.0.110 source_signoz_host: 192.168.0.188 terminal: partial_degraded safe_next_action: >- - Reconcile the three source-only services and the stale SignOz query host - against production runtime before atomically regenerating the ConfigMap; - do not copy the stale 188 SignOz row into the runtime mirror. + Verify the exact 27-service source/runtime manifest against production, + then reconcile the SignOz query-host identity against live 110 readback; + source-manifest parity alone does not close the production drift item. github_freeze_legacy_asset_drift: drift_work_item_id: P0-OBS-002-ASSET-DRIFT-002 superseded_by: global_product_governance_v2 @@ -761,9 +760,57 @@ pending_verifiers: status: pending_supported_non_raw_export_or_coordinated_snapshot raw_sqlite_read_or_sync_allowed: false sqlite_cli_present_in_runtime: false + source_contract_status: implemented_tested_not_deployed + source_contract: + policy: config/signoz/metadata-export-policy.json + exporter: scripts/backup/signoz-metadata-export.py + shared_contract: scripts/backup/signoz_metadata_contract.py + independent_export_verifier: scripts/backup/verify-signoz-metadata-export.py + isolated_restore_driver: scripts/backup/signoz-metadata-restore-drill.py + contract_tests: scripts/backup/tests/test_signoz_metadata_export_contract.py + controlled_deployer: scripts/ops/deploy-signoz-metadata-toolchain.sh + deployer_tests: scripts/ops/tests/test_signoz_metadata_toolchain_deploy.py + expected_file_count: 8 + hashes: + policy_sha256: df09dbf91d30bcf4d1a2119b1c301240a252fddd446c56ecba253d5399526533 + exporter_sha256: b184d55bd5601377d25e78e422de172d01c3f8b8ea2de948b65c915201094956 + shared_contract_sha256: f719d3f2ec86acfafe12be1883e57e6675b8b64caab0ee16d6054c8b383fa213 + independent_export_verifier_sha256: cbab56dd3919866a96550c1dce46ec67783cd6f00473e4491861e17d1430048c + isolated_restore_driver_sha256: 38ce70b3f891a3ce8edfaeb880ccf69d0bb6817c3f801a0aefa0f06341901997 + contract_tests_sha256: 474d9dfcfe170c62425356ea9f06ed0dc1629b39215ee80f395f721c7d43a923 + controlled_deployer_sha256: b3edcb3629aa3069fcecde5812cc4f21e83e324eec89b3c75f61cc73444adb97 + deployer_tests_sha256: cae135b2075d845f13d762c751d9e6de88bf806b37dcd5bd8837487211a20880 + test_terminal: 19_passed + portable_assets: + - dashboards + - alert_rules + - explorer_views + export_only_assets: + - roles + - organization_preferences + - user_preferences + - global_config + forbidden_assets: + - personal_access_tokens + - authentication_domains + - notification_channels + - users_and_credentials + - invitations + - sessions + raw_sqlite_read: false + raw_volume_read: false + secret_value_persistence: false + full_sqlite_completion_claim: false + runtime_export_terminal: pending_authenticated_stable_export + runtime_restore_terminal: pending_isolated_same_version_target + runtime_zero_residue_terminal: pending required_preconditions: - - supported_metadata_export_or_application_consistent_snapshot_design - - independent_restore_verifier_without_raw_session_or_secret_read + - deploy_exact_source_contract_with_hash_and_mode_readback + - trusted_service_account_secret_reference_without_value_readback + - authenticated_two_read_stable_production_export + - isolated_same_version_restore_semantic_readback + - zero_container_volume_network_listener_residue + - reissue_secret_bearing_assets_from_secret_references signoz_backup_offsite_dr: status: pending_offsite_snapshot_and_restore_verifier current_repository_scope: local_same_failure_domain @@ -816,7 +863,7 @@ terminal: - signoz_backup_offsite_dr_pending - clickhouse_native_failed_artifact_retention_pending - clickhouse_native_resume_submitted_inventory_pending - - service_registry_runtime_mirror_reconciliation_pending + - service_registry_production_and_live_host_readback_pending - github_freeze_legacy_asset_retirement_pending - monitoring_generator_duplicate_awoooi_api_identity_pending safe_next_action: >- @@ -826,8 +873,9 @@ terminal: target and restore it independently, then apply an explicit failed-artifact retention decision. Persist the submitted source inventory before BACKUP so same-run resume cannot bind an old artifact to newer live counters. Keep raw - SQLite reads and syncs disabled. Reconcile the source-only service registry - rows against live runtime before regenerating its K8s ConfigMap mirror, and - retire legacy GitHub runner assets only after the Gitea replacement verifier - is ready. Deduplicate the monitoring registry's awoooi-api identity before - applying regenerated scrape configuration. + SQLite reads and syncs disabled. Verify the exact source/runtime service + registry manifest against production and reconcile its SignOz query-host + identity against live runtime. Retire legacy GitHub runner assets only after + the Gitea replacement verifier is ready. Deduplicate the monitoring + registry's awoooi-api identity before applying regenerated scrape + configuration. diff --git a/scripts/backup/signoz-metadata-export.py b/scripts/backup/signoz-metadata-export.py new file mode 100755 index 000000000..50f9e123e --- /dev/null +++ b/scripts/backup/signoz-metadata-export.py @@ -0,0 +1,371 @@ +#!/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()) diff --git a/scripts/backup/signoz-metadata-restore-drill.py b/scripts/backup/signoz-metadata-restore-drill.py new file mode 100755 index 000000000..364e959c9 --- /dev/null +++ b/scripts/backup/signoz-metadata-restore-drill.py @@ -0,0 +1,510 @@ +#!/usr/bin/env python3 +"""Restore a verified SigNoz metadata subset into an isolated API target.""" + +from __future__ import annotations + +import argparse +import json +import os +import stat + +# Child commands use fixed argv without a shell. +import subprocess # nosec B404 +import sys +import urllib.error +import urllib.request +from pathlib import Path +from typing import Any + +from signoz_metadata_contract import ( + DEFAULT_POLICY, + SHA256, + 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("--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() + + +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"] + 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") + canonical_id = target.get("canonical_id") + if not isinstance(canonical_id, str) or not canonical_id.startswith( + "ephemeral-signoz-metadata-restore-" + ): + raise ContractError("isolation_target_canonical_id_invalid") + if target.get("image_ref") != isolation_policy["required_image_ref"]: + raise ContractError("isolation_target_image_ref_mismatch") + image_id = target.get("image_id") + if ( + not isinstance(image_id, str) + or not image_id.startswith("sha256:") + or not SHA256.fullmatch(image_id.removeprefix("sha256:")) + ): + raise ContractError("isolation_target_image_id_invalid") + required_values = { + "image_present_locally": True, + "image_pull_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": isolation_policy["resource_label_key"], + "resource_label_value": args.run_id, + } + 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") + return ApiClient( + base_url=validate_base_url(args.target_base_url, loopback_only=True), + header_name=policy["authentication"]["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"]) + 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") + 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["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 require_target_unreachable(base_url: str) -> None: + base_url = validate_base_url(base_url, loopback_only=True) + request = urllib.request.Request(base_url, method="GET") + try: + # base_url is validated as loopback HTTP(S) immediately above. + with urllib.request.urlopen( # nosec B310 + request, timeout=2 + ): + pass + except urllib.error.HTTPError as exc: + raise ContractError("cleanup_target_listener_still_present") from exc + except (urllib.error.URLError, TimeoutError, OSError): + return + raise ContractError("cleanup_target_listener_still_present") + + +def run_cleanup_verifier( + args: argparse.Namespace, + policy: dict[str, Any], +) -> dict[str, Any]: + label = f"{policy['restore_isolation']['resource_label_key']}={args.run_id}" + counts = { + resource: docker_label_count(resource, label) + for resource in ("container", "volume", "network") + } + if any(counts.values()): + raise ContractError("cleanup_resource_residue_present") + require_target_unreachable(args.target_base_url) + 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="container_0_volume_0_network_0_listener_0", + ) + + +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) + 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) + 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()) diff --git a/scripts/backup/signoz_metadata_contract.py b/scripts/backup/signoz_metadata_contract.py new file mode 100755 index 000000000..f8907962b --- /dev/null +++ b/scripts/backup/signoz_metadata_contract.py @@ -0,0 +1,551 @@ +#!/usr/bin/env python3 +"""Shared fail-closed primitives for the SigNoz metadata backup lane. + +The contract intentionally works only through authenticated HTTP APIs. It +never opens the SigNoz SQLite database or a Docker volume and never accepts a +secret value on the command line. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +import stat +import urllib.error +import urllib.parse +import urllib.request +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + + +REPO_ROOT = Path(__file__).resolve().parents[2] +DEFAULT_POLICY = REPO_ROOT / "config" / "signoz" / "metadata-export-policy.json" +SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$") +SAFE_ASSET_ID = re.compile(r"^[a-z][a-z0-9_]{0,63}$") +SHA256 = re.compile(r"^[0-9a-f]{64}$") +ALLOWED_TOKEN_MODES = {0o400, 0o600} +LOOPBACK_HOSTS = {"127.0.0.1", "::1", "localhost"} + + +class ContractError(ValueError): + """A bounded policy, transport, or artifact contract failure.""" + + +class _NoRedirect(urllib.request.HTTPRedirectHandler): + def redirect_request( + self, + req: urllib.request.Request, + fp: Any, + code: int, + msg: str, + headers: Any, + newurl: str, + ) -> None: + return None + + +def utc_now() -> str: + return ( + datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z") + ) + + +def canonical_json_bytes(value: Any) -> bytes: + return ( + json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + + "\n" + ).encode("utf-8") + + +def sha256_bytes(value: bytes) -> str: + return hashlib.sha256(value).hexdigest() + + +def sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def read_json_object(path: Path, *, label: str) -> dict[str, Any]: + require_no_symlink_components(path, label=label) + try: + metadata = path.stat() + if not stat.S_ISREG(metadata.st_mode) or metadata.st_size > 1024 * 1024: + raise ContractError(f"{label}_size_or_type_invalid") + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError) as exc: + raise ContractError(f"{label}_read_failed") from exc + if not isinstance(value, dict): + raise ContractError(f"{label}_not_object") + return value + + +def _normalized_key(value: str) -> str: + return re.sub(r"[^a-z0-9]", "", value.casefold()) + + +def require_no_symlink_components(path: Path, *, label: str) -> None: + if not path.is_absolute(): + path = Path.cwd() / path + current = path + while True: + try: + metadata = current.lstat() + except OSError as exc: + raise ContractError(f"{label}_path_component_unavailable") from exc + if stat.S_ISLNK(metadata.st_mode): + raise ContractError(f"{label}_symlink_component_forbidden") + if current == current.parent: + break + current = current.parent + + +def load_policy(path: Path) -> dict[str, Any]: + policy = read_json_object(path, label="policy") + if policy.get("schema") != "awoooi_signoz_metadata_export_policy_v1": + raise ContractError("unsupported_policy_schema") + + runtime = policy.get("source_runtime") + if not isinstance(runtime, dict): + raise ContractError("policy_source_runtime_missing") + if runtime.get("raw_database_access_allowed") is not False: + raise ContractError("policy_raw_database_must_be_forbidden") + if runtime.get("raw_volume_access_allowed") is not False: + raise ContractError("policy_raw_volume_must_be_forbidden") + if runtime.get("github_supply_chain_allowed") is not False: + raise ContractError("policy_github_supply_chain_must_be_forbidden") + if runtime.get("minimum_python_version") != "3.10": + raise ContractError("policy_minimum_python_version_invalid") + + limits = policy.get("limits") + if not isinstance(limits, dict): + raise ContractError("policy_limits_missing") + for key in ( + "request_timeout_seconds", + "max_response_bytes", + "max_asset_count", + "max_items_per_asset", + "max_restore_actions", + "stable_read_count", + ): + value = limits.get(key) + if not isinstance(value, int) or isinstance(value, bool) or value <= 0: + raise ContractError(f"policy_limit_invalid_{key}") + if limits["stable_read_count"] != 2: + raise ContractError("policy_requires_exactly_two_stable_reads") + + authentication = policy.get("authentication") + if not isinstance(authentication, dict): + raise ContractError("policy_authentication_missing") + if authentication.get("header_name") != "SIGNOZ-API-KEY": + raise ContractError("policy_auth_header_not_fixed") + if authentication.get("secret_value_in_arguments_allowed") is not False: + raise ContractError("policy_secret_argument_must_be_forbidden") + if authentication.get("accepted_file_modes") != ["0400", "0600"]: + raise ContractError("policy_credential_modes_invalid") + + assets = policy.get("assets") + if not isinstance(assets, list) or not assets: + raise ContractError("policy_assets_missing") + if len(assets) > limits["max_asset_count"]: + raise ContractError("policy_asset_count_exceeds_limit") + + forbidden_paths = { + item.get("path") + for item in policy.get("forbidden_endpoints", []) + if isinstance(item, dict) + } + asset_ids: set[str] = set() + asset_paths: set[str] = set() + for asset in assets: + if not isinstance(asset, dict): + raise ContractError("policy_asset_not_object") + asset_id = asset.get("id") + endpoint = asset.get("endpoint") + if not isinstance(asset_id, str) or not SAFE_ASSET_ID.fullmatch(asset_id): + raise ContractError("policy_asset_id_unsafe") + if asset_id in asset_ids: + raise ContractError(f"policy_asset_duplicate_{asset_id}") + asset_ids.add(asset_id) + if not isinstance(endpoint, str) or not endpoint.startswith("/api/v1/"): + raise ContractError(f"policy_asset_endpoint_unsafe_{asset_id}") + if "?" in endpoint or "#" in endpoint or ".." in endpoint: + raise ContractError(f"policy_asset_endpoint_unsafe_{asset_id}") + if endpoint in forbidden_paths: + raise ContractError(f"policy_forbidden_endpoint_allowlisted_{asset_id}") + if endpoint in asset_paths: + raise ContractError(f"policy_asset_endpoint_duplicate_{asset_id}") + asset_paths.add(endpoint) + if asset.get("method") != "GET": + raise ContractError(f"policy_asset_export_must_be_get_{asset_id}") + if asset.get("response_kind") not in {"collection", "document"}: + raise ContractError(f"policy_asset_response_kind_invalid_{asset_id}") + if asset.get("classification") not in {"portable", "export_only"}: + raise ContractError(f"policy_asset_classification_invalid_{asset_id}") + if asset.get("classification") == "portable": + restore = asset.get("restore") + if not isinstance(restore, dict): + raise ContractError(f"policy_restore_missing_{asset_id}") + if restore.get("method") not in {"POST", "PUT", "PATCH"}: + raise ContractError(f"policy_restore_method_invalid_{asset_id}") + if restore.get("strategy") != "collection_items": + raise ContractError(f"policy_restore_strategy_invalid_{asset_id}") + if restore.get("endpoint") != endpoint: + raise ContractError(f"policy_restore_endpoint_mismatch_{asset_id}") + if asset.get("response_kind") != "collection": + raise ContractError(f"policy_portable_asset_not_collection_{asset_id}") + + completion = policy.get("completion_contract") + if not isinstance(completion, dict): + raise ContractError("policy_completion_contract_missing") + if completion.get("portable_export_alone_is_completion") is not False: + raise ContractError("policy_export_must_not_claim_completion") + if completion.get("full_sqlite_completion_claim_allowed") is not False: + raise ContractError("policy_full_sqlite_claim_must_be_forbidden") + if completion.get("durable_terminal_receipt_required_for_apply") is not True: + raise ContractError("policy_durable_apply_receipt_must_be_required") + if completion.get("failure_terminal_receipt_required") is not True: + raise ContractError("policy_failure_receipt_must_be_required") + restore_isolation = policy.get("restore_isolation") + if not isinstance(restore_isolation, dict): + raise ContractError("policy_restore_isolation_missing") + if restore_isolation.get("image_id_sha256_required") is not True: + raise ContractError("policy_restore_image_id_must_be_required") + + forbidden_names = policy.get("forbidden_key_names") + if not isinstance(forbidden_names, list) or not forbidden_names: + raise ContractError("policy_forbidden_keys_missing") + normalized = [_normalized_key(str(item)) for item in forbidden_names] + if not all(normalized) or len(set(normalized)) != len(normalized): + raise ContractError("policy_forbidden_keys_invalid") + for pattern in policy.get("forbidden_value_patterns", []): + try: + re.compile(str(pattern)) + except re.error as exc: + raise ContractError("policy_forbidden_value_pattern_invalid") from exc + return policy + + +def validate_identity(value: str, *, label: str) -> str: + if not SAFE_ID.fullmatch(value): + raise ContractError(f"{label}_unsafe") + return value + + +def validate_base_url(value: str, *, loopback_only: bool = False) -> str: + parsed = urllib.parse.urlsplit(value) + if parsed.username or parsed.password or parsed.query or parsed.fragment: + raise ContractError("base_url_contains_forbidden_components") + if parsed.path not in {"", "/"}: + raise ContractError("base_url_path_must_be_empty") + host = (parsed.hostname or "").casefold() + is_loopback = host in LOOPBACK_HOSTS + if loopback_only and not is_loopback: + raise ContractError("restore_target_must_be_loopback") + if parsed.scheme == "http" and not is_loopback: + raise ContractError("plain_http_allowed_only_for_loopback") + if parsed.scheme not in {"http", "https"} or not host: + raise ContractError("base_url_scheme_or_host_invalid") + try: + parsed.port + except ValueError as exc: + raise ContractError("base_url_port_invalid") from exc + return value.rstrip("/") + + +def validate_token_reference(path: Path, *, read_value: bool) -> str | None: + if not path.is_absolute(): + raise ContractError("credential_reference_must_be_absolute") + require_no_symlink_components(path, label="credential_reference") + try: + metadata = path.lstat() + except OSError as exc: + raise ContractError("credential_reference_unavailable") from exc + if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode): + raise ContractError("credential_reference_not_regular_file") + if metadata.st_uid != os.geteuid(): + raise ContractError("credential_reference_owner_mismatch") + if stat.S_IMODE(metadata.st_mode) not in ALLOWED_TOKEN_MODES: + raise ContractError("credential_reference_mode_must_be_0400_or_0600") + if metadata.st_size <= 0 or metadata.st_size > 4096: + raise ContractError("credential_reference_size_invalid") + if not read_value: + return None + try: + raw = path.read_bytes() + except OSError as exc: + raise ContractError("credential_reference_read_failed") from exc + token = raw.rstrip(b"\r\n") + if not token or len(token) > 4096 or any(byte < 33 or byte > 126 for byte in token): + raise ContractError("credential_reference_value_shape_invalid") + return token.decode("ascii") + + +def scan_forbidden(value: Any, policy: dict[str, Any]) -> None: + forbidden_keys = { + _normalized_key(str(item)) for item in policy["forbidden_key_names"] + } + value_patterns = [ + re.compile(str(item)) for item in policy.get("forbidden_value_patterns", []) + ] + + def walk(item: Any, pointer: str) -> None: + if isinstance(item, dict): + for key, child in item.items(): + if not isinstance(key, str): + raise ContractError("json_object_key_not_string") + if _normalized_key(key) in forbidden_keys: + raise ContractError( + f"forbidden_key_detected_at_{pointer or 'root'}" + ) + walk(child, f"{pointer}/{key}") + elif isinstance(item, list): + for index, child in enumerate(item): + walk(child, f"{pointer}/{index}") + elif isinstance(item, str): + if any(pattern.search(item) for pattern in value_patterns): + raise ContractError( + f"forbidden_value_pattern_detected_at_{pointer or 'root'}" + ) + + walk(value, "") + + +def resolve_pointer(value: Any, pointer: str) -> Any: + if pointer == "": + return value + if not pointer.startswith("/"): + raise ContractError("response_pointer_invalid") + current = value + for raw_part in pointer[1:].split("/"): + part = raw_part.replace("~1", "/").replace("~0", "~") + if isinstance(current, dict) and part in current: + current = current[part] + else: + raise ContractError("response_pointer_not_found") + return current + + +def validate_asset_document( + value: Any, + asset: dict[str, Any], + policy: dict[str, Any], +) -> int: + scan_forbidden(value, policy) + selected = resolve_pointer(value, str(asset.get("response_pointer", ""))) + if asset["response_kind"] == "collection": + if not isinstance(selected, list): + raise ContractError(f"asset_collection_shape_invalid_{asset['id']}") + if len(selected) > policy["limits"]["max_items_per_asset"]: + raise ContractError(f"asset_item_limit_exceeded_{asset['id']}") + if any(not isinstance(item, dict) for item in selected): + raise ContractError(f"asset_collection_item_invalid_{asset['id']}") + return len(selected) + if not isinstance(selected, dict): + raise ContractError(f"asset_document_shape_invalid_{asset['id']}") + return 1 + + +def semantic_items(value: Any, asset: dict[str, Any]) -> list[dict[str, Any]]: + selected = resolve_pointer(value, str(asset.get("response_pointer", ""))) + if not isinstance(selected, list): + raise ContractError(f"asset_collection_shape_invalid_{asset['id']}") + restore = asset.get("restore") + if not isinstance(restore, dict): + raise ContractError(f"asset_not_portable_{asset['id']}") + strip_fields = set(restore.get("strip_top_level_fields", [])) + normalized: list[dict[str, Any]] = [] + for item in selected: + if not isinstance(item, dict): + raise ContractError(f"asset_collection_item_invalid_{asset['id']}") + normalized.append( + {key: value for key, value in item.items() if key not in strip_fields} + ) + return sorted(normalized, key=lambda item: canonical_json_bytes(item)) + + +class ApiClient: + def __init__( + self, + *, + base_url: str, + header_name: str, + token: str, + timeout_seconds: int, + max_response_bytes: int, + ) -> None: + self.base_url = validate_base_url(base_url) + self.header_name = header_name + self.token = token + self.timeout_seconds = timeout_seconds + self.max_response_bytes = max_response_bytes + self.opener = urllib.request.build_opener(_NoRedirect()) + + def _url(self, endpoint: str) -> str: + if not endpoint.startswith("/api/v1/") or any( + item in endpoint for item in ("..", "?", "#") + ): + raise ContractError("api_endpoint_unsafe") + return f"{self.base_url}{endpoint}" + + def request_json(self, endpoint: str) -> Any: + request = urllib.request.Request( + self._url(endpoint), + method="GET", + headers={ + self.header_name: self.token, + "Accept": "application/json", + "User-Agent": "AWOOOI-P0-OBS-002-metadata-export/1", + }, + ) + try: + with self.opener.open(request, timeout=self.timeout_seconds) as response: + status_code = int(response.status) + content_type = response.headers.get_content_type() + payload = response.read(self.max_response_bytes + 1) + except urllib.error.HTTPError as exc: + raise ContractError(f"api_http_status_{exc.code}") from exc + except (urllib.error.URLError, TimeoutError, OSError) as exc: + raise ContractError("api_transport_failed") from exc + if status_code != 200: + raise ContractError(f"api_http_status_{status_code}") + if content_type != "application/json": + raise ContractError("api_content_type_not_json") + if len(payload) > self.max_response_bytes: + raise ContractError("api_response_size_exceeded") + try: + return json.loads(payload) + except (UnicodeError, json.JSONDecodeError) as exc: + raise ContractError("api_response_json_invalid") from exc + + def request_write(self, method: str, endpoint: str, payload: Any) -> None: + if method not in {"POST", "PUT", "PATCH"}: + raise ContractError("api_write_method_forbidden") + body = canonical_json_bytes(payload) + if len(body) > self.max_response_bytes: + raise ContractError("api_request_size_exceeded") + request = urllib.request.Request( + self._url(endpoint), + method=method, + data=body, + headers={ + self.header_name: self.token, + "Accept": "application/json", + "Content-Type": "application/json", + "User-Agent": "AWOOOI-P0-OBS-002-metadata-restore/1", + }, + ) + try: + with self.opener.open(request, timeout=self.timeout_seconds) as response: + status_code = int(response.status) + response_payload = response.read(self.max_response_bytes + 1) + except urllib.error.HTTPError as exc: + raise ContractError(f"api_write_http_status_{exc.code}") from exc + except (urllib.error.URLError, TimeoutError, OSError) as exc: + raise ContractError("api_write_transport_failed") from exc + if status_code not in {200, 201, 202, 204}: + raise ContractError(f"api_write_http_status_{status_code}") + if len(response_payload) > self.max_response_bytes: + raise ContractError("api_write_response_size_exceeded") + + +def receipt( + *, + trace_id: str, + run_id: str, + work_item_id: str, + phase: str, + terminal: str, + detail: str, +) -> dict[str, Any]: + return { + "schema": "awoooi_signoz_metadata_receipt_v1", + "trace_id": trace_id, + "run_id": run_id, + "work_item_id": work_item_id, + "observed_at": utc_now(), + "phase": phase, + "terminal": terminal, + "detail": detail, + } + + +def write_private(path: Path, payload: bytes) -> None: + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + descriptor = os.open(path, flags, 0o600) + try: + os.fchmod(descriptor, 0o600) + with os.fdopen(descriptor, "wb", closefd=True) as stream: + stream.write(payload) + stream.flush() + os.fsync(stream.fileno()) + except BaseException: + try: + os.close(descriptor) + except OSError: + pass + raise + + +def validate_new_private_destination(path: Path, *, label: str) -> None: + if not path.is_absolute(): + raise ContractError(f"{label}_path_must_be_absolute") + require_no_symlink_components(path.parent, label=f"{label}_parent") + try: + parent_metadata = path.parent.lstat() + except OSError as exc: + raise ContractError(f"{label}_parent_unavailable") from exc + if not stat.S_ISDIR(parent_metadata.st_mode): + raise ContractError(f"{label}_parent_not_directory") + if parent_metadata.st_uid != os.geteuid(): + raise ContractError(f"{label}_parent_owner_mismatch") + if path.exists() or path.is_symlink(): + raise ContractError(f"{label}_path_exists") + + +def write_new_private_atomic(path: Path, payload: bytes, *, label: str) -> None: + validate_new_private_destination(path, label=label) + temporary = path.with_name(f".{path.name}.partial.{os.getpid()}") + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + try: + descriptor = os.open(temporary, flags, 0o600) + try: + with os.fdopen(descriptor, "wb", closefd=True) as stream: + stream.write(payload) + stream.flush() + os.fsync(stream.fileno()) + except BaseException: + try: + os.close(descriptor) + except OSError: + pass + raise + os.replace(temporary, path) + try: + parent_descriptor = os.open(path.parent, os.O_RDONLY) + try: + os.fsync(parent_descriptor) + finally: + os.close(parent_descriptor) + except OSError: + pass + finally: + try: + temporary.unlink() + except FileNotFoundError: + pass diff --git a/scripts/backup/tests/test_signoz_metadata_export_contract.py b/scripts/backup/tests/test_signoz_metadata_export_contract.py new file mode 100644 index 000000000..f53283dbd --- /dev/null +++ b/scripts/backup/tests/test_signoz_metadata_export_contract.py @@ -0,0 +1,602 @@ +from __future__ import annotations + +import hashlib +import json +import os +import socket +import subprocess +import sys +import threading +from contextlib import contextmanager +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import Any, Iterator + + +ROOT = Path(__file__).resolve().parents[3] +EXPORTER = ROOT / "scripts" / "backup" / "signoz-metadata-export.py" +VERIFIER = ROOT / "scripts" / "backup" / "verify-signoz-metadata-export.py" +RESTORE = ROOT / "scripts" / "backup" / "signoz-metadata-restore-drill.py" +POLICY = ROOT / "config" / "signoz" / "metadata-export-policy.json" +TRACE_ID = "trace-P0-OBS-002-metadata" +RUN_ID = "run-P0-OBS-002-metadata" +WORK_ITEM_ID = "P0-OBS-002" +TOKEN = "test-only-signoz-token-reference" + + +SOURCE_ASSETS: dict[str, Any] = { + "/api/v1/dashboards": {"data": [{"id": "dash-1", "title": "Ops"}]}, + "/api/v1/rules": {"data": [{"id": "rule-1", "name": "Latency"}]}, + "/api/v1/explorer/views": {"data": [{"id": "view-1", "name": "Errors"}]}, + "/api/v1/roles": {"data": [{"id": "role-1", "name": "Viewer"}]}, + "/api/v1/org/preferences": {"data": [{"name": "theme", "value": "dark"}]}, + "/api/v1/user/preferences": {"data": [{"name": "timezone", "value": "UTC"}]}, + "/api/v1/global/config": {"data": {"setupCompleted": True}}, +} + + +class MetadataHandler(BaseHTTPRequestHandler): + assets: dict[str, Any] = {} + drift_path: str | None = None + get_counts: dict[str, int] = {} + post_count = 0 + + def log_message(self, format: str, *args: object) -> None: + return + + def _authorized(self) -> bool: + return self.headers.get("SIGNOZ-API-KEY") == TOKEN + + def _send_json(self, status: int, value: Any) -> None: + payload = json.dumps(value, separators=(",", ":")).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + def do_GET(self) -> None: + if not self._authorized(): + self._send_json(401, {"error": "unauthorized"}) + return + path = self.path.split("?", 1)[0] + if path not in self.assets: + self._send_json(404, {"error": "not found"}) + return + self.get_counts[path] = self.get_counts.get(path, 0) + 1 + value = json.loads(json.dumps(self.assets[path])) + if self.drift_path == path and self.get_counts[path] >= 2: + value["data"].append({"id": "drift", "title": "changed"}) + self._send_json(200, value) + + def do_POST(self) -> None: + if not self._authorized(): + self._send_json(401, {"error": "unauthorized"}) + return + path = self.path.split("?", 1)[0] + if path not in self.assets or not isinstance( + self.assets[path].get("data"), list + ): + self._send_json(404, {"error": "not found"}) + return + length = int(self.headers.get("Content-Length", "0")) + value = json.loads(self.rfile.read(length)) + if not isinstance(value, dict): + self._send_json(400, {"error": "object required"}) + return + type(self).post_count += 1 + restored = {"id": f"server-{type(self).post_count}", **value} + self.assets[path]["data"].append(restored) + self._send_json(201, {"data": restored}) + + +@contextmanager +def metadata_server( + assets: dict[str, Any], + *, + drift_path: str | None = None, +) -> Iterator[tuple[str, type[MetadataHandler]]]: + handler = type("BoundMetadataHandler", (MetadataHandler,), {}) + handler.assets = json.loads(json.dumps(assets)) + handler.drift_path = drift_path + handler.get_counts = {} + handler.post_count = 0 + server = ThreadingHTTPServer(("127.0.0.1", 0), handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + host, port = server.server_address + yield f"http://{host}:{port}", handler + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + + +def credential_file(tmp_path: Path, *, mode: int = 0o600) -> Path: + path = tmp_path / "signoz-api-key.ref" + path.write_text(f"{TOKEN}\n", encoding="ascii") + path.chmod(mode) + return path + + +def export_command( + *, + base_url: str, + output_dir: Path, + credential: Path | None, + apply: bool, + receipt_file: Path | None = None, +) -> list[str]: + command = [ + sys.executable, + str(EXPORTER), + "--apply" if apply else "--check", + "--base-url", + base_url, + "--output-dir", + str(output_dir), + "--trace-id", + TRACE_ID, + "--run-id", + RUN_ID, + "--work-item-id", + WORK_ITEM_ID, + ] + if credential is not None: + command.extend(["--credential-file", str(credential)]) + if apply: + terminal_path = receipt_file or output_dir.with_name( + f"{output_dir.name}.terminal.json" + ) + command.extend(["--receipt-file", str(terminal_path)]) + return command + + +def create_bundle(tmp_path: Path) -> tuple[Path, Path]: + credential = credential_file(tmp_path) + bundle = tmp_path / "bundle" + with metadata_server(SOURCE_ASSETS) as (base_url, _): + result = subprocess.run( + export_command( + base_url=base_url, + output_dir=bundle, + credential=credential, + apply=True, + ), + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=20, + check=False, + ) + assert result.returncode == 0, result.stderr + return bundle, credential + + +def isolation_receipt(tmp_path: Path, target_base_url: str) -> Path: + policy = json.loads(POLICY.read_text(encoding="utf-8")) + value = { + "schema": "awoooi_signoz_metadata_restore_isolation_v1", + "trace_id": TRACE_ID, + "run_id": RUN_ID, + "work_item_id": WORK_ITEM_ID, + "target": { + "canonical_id": f"ephemeral-signoz-metadata-restore-{RUN_ID}", + "base_url_sha256": hashlib.sha256( + target_base_url.encode("utf-8") + ).hexdigest(), + "image_ref": "signoz/signoz:v0.113.0", + "image_id": f"sha256:{'a' * 64}", + "image_present_locally": True, + "image_pull_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": policy["restore_isolation"]["resource_label_key"], + "resource_label_value": RUN_ID, + }, + } + path = tmp_path / "isolation.json" + path.write_text(json.dumps(value), encoding="utf-8") + path.chmod(0o600) + return path + + +def restore_command( + *, + mode: str, + bundle: Path, + target_base_url: str, + credential: Path | None, + isolation: Path, + receipt_file: Path | None = None, +) -> list[str]: + command = [ + sys.executable, + str(RESTORE), + mode, + "--bundle-dir", + str(bundle), + "--target-base-url", + target_base_url, + "--isolation-receipt", + str(isolation), + "--trace-id", + TRACE_ID, + "--run-id", + RUN_ID, + "--work-item-id", + WORK_ITEM_ID, + ] + if credential is not None: + command.extend(["--credential-file", str(credential)]) + if mode == "--apply": + terminal_path = receipt_file or bundle.parent / "restore-terminal.json" + command.extend(["--receipt-file", str(terminal_path)]) + return command + + +def test_policy_is_explicitly_non_raw_and_non_completion() -> None: + policy = json.loads(POLICY.read_text(encoding="utf-8")) + assert policy["source_runtime"]["raw_database_access_allowed"] is False + assert policy["source_runtime"]["raw_volume_access_allowed"] is False + assert policy["source_runtime"]["github_supply_chain_allowed"] is False + assert policy["completion_contract"]["portable_export_alone_is_completion"] is False + assert ( + policy["completion_contract"]["full_sqlite_completion_claim_allowed"] is False + ) + assert {item["path"] for item in policy["forbidden_endpoints"]} >= { + "/api/v1/pats", + "/api/v1/domains", + "/api/v1/channels", + } + + +def test_check_mode_writes_nothing(tmp_path: Path) -> None: + output = tmp_path / "must-not-exist" + result = subprocess.run( + export_command( + base_url="https://signoz.invalid", + output_dir=output, + credential=None, + apply=False, + ), + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=10, + check=False, + ) + assert result.returncode == 0, result.stderr + assert json.loads(result.stdout)["terminal"] == "check_pass_no_write" + assert not output.exists() + + +def test_apply_and_independent_verifier_pass(tmp_path: Path) -> None: + bundle, _ = create_bundle(tmp_path) + result = subprocess.run( + [ + sys.executable, + str(VERIFIER), + "--bundle-dir", + str(bundle), + "--expected-trace-id", + TRACE_ID, + "--expected-run-id", + RUN_ID, + "--expected-work-item-id", + WORK_ITEM_ID, + ], + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=10, + check=False, + ) + assert result.returncode == 0, result.stderr + assert json.loads(result.stdout)["terminal"] == ( + "pass_export_verified_restore_pending" + ) + manifest = json.loads((bundle / "manifest.json").read_text(encoding="utf-8")) + assert manifest["stable_read_count"] == 2 + assert manifest["raw_sqlite_read"] is False + assert manifest["sensitive_value_persisted"] is False + assert manifest["completion_claim"] is False + assert oct((bundle / "manifest.json").stat().st_mode & 0o777) == "0o600" + + +def test_secret_key_fails_closed_without_bundle(tmp_path: Path) -> None: + assets = json.loads(json.dumps(SOURCE_ASSETS)) + assets["/api/v1/global/config"]["data"]["clientSecret"] = "must-not-persist" + output = tmp_path / "bundle" + credential = credential_file(tmp_path) + with metadata_server(assets) as (base_url, _): + result = subprocess.run( + export_command( + base_url=base_url, + output_dir=output, + credential=credential, + apply=True, + ), + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=20, + check=False, + ) + assert result.returncode == 1 + assert "forbidden_key_detected" in result.stderr + assert "must-not-persist" not in result.stderr + assert not output.exists() + + +def test_source_drift_fails_closed_without_bundle(tmp_path: Path) -> None: + output = tmp_path / "bundle" + credential = credential_file(tmp_path) + with metadata_server(SOURCE_ASSETS, drift_path="/api/v1/dashboards") as ( + base_url, + _, + ): + result = subprocess.run( + export_command( + base_url=base_url, + output_dir=output, + credential=credential, + apply=True, + ), + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=20, + check=False, + ) + assert result.returncode == 1 + assert "source_metadata_drift_between_stable_reads" in result.stderr + assert not output.exists() + + +def test_tamper_breaks_independent_verifier(tmp_path: Path) -> None: + bundle, _ = create_bundle(tmp_path) + asset = bundle / "assets" / "dashboards.json" + asset.write_text('{"data":[]}\n', encoding="utf-8") + asset.chmod(0o600) + result = subprocess.run( + [ + sys.executable, + str(VERIFIER), + "--bundle-dir", + str(bundle), + "--expected-trace-id", + TRACE_ID, + "--expected-run-id", + RUN_ID, + ], + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=10, + check=False, + ) + assert result.returncode == 1 + assert "manifest_asset_hash_mismatch_dashboards" in result.stderr + + +def test_unexpected_bundle_file_breaks_independent_verifier(tmp_path: Path) -> None: + bundle, _ = create_bundle(tmp_path) + unexpected = bundle / "unlisted.json" + unexpected.write_text('{"not":"manifested"}\n', encoding="utf-8") + unexpected.chmod(0o600) + result = subprocess.run( + [ + sys.executable, + str(VERIFIER), + "--bundle-dir", + str(bundle), + "--expected-trace-id", + TRACE_ID, + "--expected-run-id", + RUN_ID, + ], + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=10, + check=False, + ) + assert result.returncode == 1 + assert "bundle_tree_contains_unexpected_entries" in result.stderr + + +def test_restore_check_and_semantic_apply_are_bounded(tmp_path: Path) -> None: + bundle, credential = create_bundle(tmp_path) + target_assets = { + "/api/v1/dashboards": {"data": []}, + "/api/v1/rules": {"data": []}, + "/api/v1/explorer/views": {"data": []}, + } + with metadata_server(target_assets) as (target_base_url, handler): + isolation = isolation_receipt(tmp_path, target_base_url) + check_result = subprocess.run( + restore_command( + mode="--check", + bundle=bundle, + target_base_url=target_base_url, + credential=credential, + isolation=isolation, + ), + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=20, + check=False, + ) + assert check_result.returncode == 0, check_result.stderr + assert json.loads(check_result.stdout)["terminal"] == "check_pass_no_write" + assert handler.post_count == 0 + + apply_result = subprocess.run( + restore_command( + mode="--apply", + bundle=bundle, + target_base_url=target_base_url, + credential=credential, + isolation=isolation, + ), + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=20, + check=False, + ) + assert apply_result.returncode == 0, apply_result.stderr + assert json.loads(apply_result.stdout)["terminal"] == ( + "partial_degraded_cleanup_pending" + ) + phase_receipts = json.loads(apply_result.stdout)["phase_receipts"] + assert [item["phase"] for item in phase_receipts] == [ + "sensor_source", + "normalized_asset_identity", + "source_of_truth_diff", + "ai_decision", + "risk_policy_decision", + "check_mode", + "bounded_execution", + "independent_post_verifier", + "rollback", + "learning_writeback", + ] + assert phase_receipts[-1]["terminal"] == "pending" + assert handler.post_count == 3 + + +def test_restore_rejects_non_loopback_target_before_writes(tmp_path: Path) -> None: + bundle, credential = create_bundle(tmp_path) + target = "https://signoz.example.invalid" + isolation = isolation_receipt(tmp_path, target) + result = subprocess.run( + restore_command( + mode="--apply", + bundle=bundle, + target_base_url=target, + credential=credential, + isolation=isolation, + ), + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=20, + check=False, + ) + assert result.returncode == 1 + assert "restore_target_must_be_loopback" in result.stderr + + +def test_cleanup_verifier_requires_zero_labeled_residue(tmp_path: Path) -> None: + bundle, _ = create_bundle(tmp_path) + fake_bin = tmp_path / "bin" + fake_bin.mkdir() + docker = fake_bin / "docker" + docker.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8") + docker.chmod(0o755) + + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + target = f"http://127.0.0.1:{port}" + isolation = isolation_receipt(tmp_path, target) + env = {**os.environ, "PATH": f"{fake_bin}:{os.environ['PATH']}"} + result = subprocess.run( + restore_command( + mode="--verify-cleanup", + bundle=bundle, + target_base_url=target, + credential=None, + isolation=isolation, + ), + env=env, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=20, + check=False, + ) + assert result.returncode == 0, result.stderr + assert json.loads(result.stdout)["terminal"] == "pass_zero_residue_verified" + + +def test_credential_reference_mode_fails_before_network(tmp_path: Path) -> None: + credential = credential_file(tmp_path, mode=0o644) + output = tmp_path / "bundle" + result = subprocess.run( + export_command( + base_url="https://signoz.invalid", + output_dir=output, + credential=credential, + apply=True, + ), + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=10, + check=False, + ) + assert result.returncode == 1 + assert "credential_reference_mode_must_be_0400_or_0600" in result.stderr + assert not output.exists() + + +def test_credential_symlink_fails_before_network(tmp_path: Path) -> None: + credential = credential_file(tmp_path) + link = tmp_path / "credential-link" + link.symlink_to(credential) + output = tmp_path / "bundle" + result = subprocess.run( + export_command( + base_url="https://signoz.invalid", + output_dir=output, + credential=link, + apply=True, + ), + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=10, + check=False, + ) + assert result.returncode == 1 + assert "credential_reference_symlink_component_forbidden" in result.stderr + assert not output.exists() + + +def test_apply_failure_writes_terminal_receipt_without_secret(tmp_path: Path) -> None: + assets = json.loads(json.dumps(SOURCE_ASSETS)) + secret_value = "must-not-appear-in-terminal-receipt" + assets["/api/v1/global/config"]["data"]["clientSecret"] = secret_value + output = tmp_path / "bundle" + receipt_file = tmp_path / "terminal.json" + credential = credential_file(tmp_path) + with metadata_server(assets) as (base_url, _): + command = export_command( + base_url=base_url, + output_dir=output, + credential=credential, + apply=True, + receipt_file=receipt_file, + ) + result = subprocess.run( + command, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=20, + check=False, + ) + assert result.returncode == 1 + terminal = json.loads(receipt_file.read_text(encoding="utf-8")) + assert terminal["terminal"] == "failed_export_not_closable" + assert secret_value not in receipt_file.read_text(encoding="utf-8") + assert not output.exists() diff --git a/scripts/backup/verify-signoz-metadata-export.py b/scripts/backup/verify-signoz-metadata-export.py new file mode 100755 index 000000000..ce630a618 --- /dev/null +++ b/scripts/backup/verify-signoz-metadata-export.py @@ -0,0 +1,308 @@ +#!/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()) diff --git a/scripts/ops/deploy-signoz-metadata-toolchain.sh b/scripts/ops/deploy-signoz-metadata-toolchain.sh new file mode 100755 index 000000000..d9981c188 --- /dev/null +++ b/scripts/ops/deploy-signoz-metadata-toolchain.sh @@ -0,0 +1,505 @@ +#!/usr/bin/env bash +# P0-OBS-002 - deploy an immutable, inactive SigNoz metadata toolchain to host110. +# +# The deployer never reads credentials, SQLite, or Docker volumes. It does not +# change an active pointer or invoke an authenticated metadata export. Failed +# candidates are moved into the run receipt directory rather than deleted. + +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" +TARGET_HOST="${SIGNOZ_METADATA_TARGET_HOST:-wooo@192.168.0.110}" +REMOTE_ROOT="${SIGNOZ_METADATA_REMOTE_ROOT:-/backup/toolchains/signoz-metadata}" +RECEIPT_ROOT="${SIGNOZ_METADATA_RECEIPT_ROOT:-/backup/deploy-receipts/signoz-metadata-toolchain}" +STAGE_ROOT="${SIGNOZ_METADATA_STAGE_ROOT:-/tmp}" +SSH_TIMEOUT_SECONDS="${SIGNOZ_METADATA_SSH_TIMEOUT_SECONDS:-120}" +SCP_TIMEOUT_SECONDS="${SIGNOZ_METADATA_SCP_TIMEOUT_SECONDS:-120}" +REMOTE_TIMEOUT_SECONDS="${SIGNOZ_METADATA_REMOTE_TIMEOUT_SECONDS:-60}" +KILL_AFTER_SECONDS="${SIGNOZ_METADATA_KILL_AFTER_SECONDS:-10}" +LOCAL_PYTHON_BIN="${SIGNOZ_METADATA_LOCAL_PYTHON_BIN:-python3.11}" + +LABELS=( + metadata-export-policy.json + signoz_metadata_contract.py + signoz-metadata-export.py + verify-signoz-metadata-export.py + signoz-metadata-restore-drill.py +) +LOCAL_SOURCES=( + "${ROOT_DIR}/config/signoz/metadata-export-policy.json" + "${ROOT_DIR}/scripts/backup/signoz_metadata_contract.py" + "${ROOT_DIR}/scripts/backup/signoz-metadata-export.py" + "${ROOT_DIR}/scripts/backup/verify-signoz-metadata-export.py" + "${ROOT_DIR}/scripts/backup/signoz-metadata-restore-drill.py" +) +MODES=(0644 0644 0755 0755 0755) +SSH_OPTIONS=( + -o BatchMode=yes + -o ConnectTimeout=8 + -o ConnectionAttempts=1 + -o ServerAliveInterval=5 + -o ServerAliveCountMax=2 +) + +usage() { + cat <<'USAGE' +Usage: deploy-signoz-metadata-toolchain.sh [--check|--apply] + +Required environment: + TRACE_ID Path-safe controlled-apply trace identifier + RUN_ID Unique path-safe execution identifier + WORK_ITEM_ID Must be P0-OBS-002 + +--check validates the five local sources and performs a no-write host110 diff. +--apply creates one immutable exact-hash directory. It does not create or +change an active pointer and does not run an authenticated metadata export. +USAGE +} + +MODE="${1:---check}" +case "${MODE}" in + --check|--apply) ;; + -h|--help) usage; exit 0 ;; + *) usage >&2; exit 64 ;; +esac +[ "$#" -le 1 ] || { usage >&2; exit 64; } + +TRACE_ID="${TRACE_ID:-}" +RUN_ID="${RUN_ID:-}" +WORK_ITEM_ID="${WORK_ITEM_ID:-}" + +valid_identity() { + [[ "$1" =~ ^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$ ]] +} + +valid_positive_integer() { + [[ "$1" =~ ^[1-9][0-9]*$ ]] +} + +safe_absolute_path() { + local path="$1" + [[ "${path}" == /* ]] || return 1 + [[ "${path}" != / && "${path}" != */ ]] || return 1 + [[ "${path}" =~ ^/[A-Za-z0-9._/-]+$ ]] || return 1 + case "/${path#/}/" in + *'//'*|*'/../'*|*'/./'*) return 1 ;; + esac +} + +valid_identity "${TRACE_ID}" || { echo "invalid TRACE_ID" >&2; exit 64; } +valid_identity "${RUN_ID}" || { echo "invalid RUN_ID" >&2; exit 64; } +[ "${WORK_ITEM_ID}" = "P0-OBS-002" ] || { + echo "WORK_ITEM_ID must be P0-OBS-002" >&2 + exit 64 +} +for value in "${SSH_TIMEOUT_SECONDS}" "${SCP_TIMEOUT_SECONDS}" \ + "${REMOTE_TIMEOUT_SECONDS}" "${KILL_AFTER_SECONDS}"; do + valid_positive_integer "${value}" || { echo "invalid timeout" >&2; exit 64; } +done +for path in "${REMOTE_ROOT}" "${RECEIPT_ROOT}" "${STAGE_ROOT}"; do + safe_absolute_path "${path}" || { echo "unsafe remote path" >&2; exit 64; } +done + +bounded_ssh() { + timeout --kill-after="${KILL_AFTER_SECONDS}" "${SSH_TIMEOUT_SECONDS}" ssh "$@" +} + +bounded_scp() { + timeout --kill-after="${KILL_AFTER_SECONDS}" "${SCP_TIMEOUT_SECONDS}" scp "$@" +} + +hash_file() { + sha256sum "$1" | awk '{print $1}' +} + +emit() { + local phase="$1" terminal="$2" detail="$3" + printf '{"schema":"awoooi_controlled_apply_receipt_v1","trace_id":"%s","run_id":"%s","work_item_id":"%s","asset_id":"host110-signoz-metadata-toolchain","phase":"%s","terminal":"%s","detail":"%s"}\n' \ + "${TRACE_ID}" "${RUN_ID}" "${WORK_ITEM_ID}" "${phase}" "${terminal}" "${detail}" +} + +validate_local_sources() { + local source + command -v "${LOCAL_PYTHON_BIN}" >/dev/null 2>&1 + "${LOCAL_PYTHON_BIN}" -c 'import sys; assert sys.version_info >= (3, 10)' + for source in "${LOCAL_SOURCES[@]}"; do + [ -f "${source}" ] && [ ! -L "${source}" ] && [ -s "${source}" ] + done + "${LOCAL_PYTHON_BIN}" - "${LOCAL_SOURCES[@]:1}" <<'PY' +from pathlib import Path +import sys + +for value in sys.argv[1:]: + path = Path(value) + compile(path.read_text(encoding="utf-8"), str(path), "exec") +PY + PYTHONPATH="${ROOT_DIR}/scripts/backup" "${LOCAL_PYTHON_BIN}" - \ + "${LOCAL_SOURCES[0]}" <<'PY' +from pathlib import Path +import sys +from signoz_metadata_contract import load_policy + +policy = load_policy(Path(sys.argv[1])) +assert policy["work_item_id"] == "P0-OBS-002" +assert policy["source_runtime"]["raw_database_access_allowed"] is False +assert policy["source_runtime"]["raw_volume_access_allowed"] is False +assert policy["completion_contract"]["full_sqlite_completion_claim_allowed"] is False +PY +} + +validate_local_sources +SOURCE_HASHES=() +SOURCE_DETAIL="" +for index in "${!LOCAL_SOURCES[@]}"; do + source_hash="$(hash_file "${LOCAL_SOURCES[index]}")" + SOURCE_HASHES+=("${source_hash}") + SOURCE_DETAIL+="${LABELS[index]}_${source_hash}_" +done +SOURCE_DETAIL="${SOURCE_DETAIL%_}" +BUNDLE_ID="$(printf '%s\n' "${SOURCE_HASHES[@]}" | sha256sum | awk '{print $1}')" +TARGET_DIR="${REMOTE_ROOT}/${BUNDLE_ID}" +STAGE_DIR="${STAGE_ROOT}/awoooi-signoz-metadata-toolchain.${RUN_ID}" + +emit sensor_source pass "local_exact_sources_5" +emit normalized_asset_identity pass "bundle_${BUNDLE_ID}" +emit ai_decision pass "candidate_additive_immutable_inactive_source_deploy" +emit risk_policy_decision pass "medium_no_active_pointer_no_secret_no_raw_sqlite" + +remote_check() { + bounded_ssh "${SSH_OPTIONS[@]}" "${TARGET_HOST}" bash -s -- \ + "${REMOTE_ROOT}" "${BUNDLE_ID}" "${REMOTE_TIMEOUT_SECONDS}" \ + "${KILL_AFTER_SECONDS}" "${SOURCE_HASHES[@]}" <<'REMOTE_CHECK' +set -euo pipefail + +REMOTE_ROOT="$1" +BUNDLE_ID="$2" +REMOTE_TIMEOUT_SECONDS="$3" +KILL_AFTER_SECONDS="$4" +shift 4 +EXPECTED_HASHES=("$@") +LABELS=( + metadata-export-policy.json + signoz_metadata_contract.py + signoz-metadata-export.py + verify-signoz-metadata-export.py + signoz-metadata-restore-drill.py +) +MODES=(0644 0644 0755 0755 0755) +TARGET_DIR="${REMOTE_ROOT}/${BUNDLE_ID}" + +[ "${#EXPECTED_HASHES[@]}" -eq 5 ] +[ -d /backup ] && [ ! -L /backup ] +[ ! -e "${REMOTE_ROOT}/current" ] && [ ! -L "${REMOTE_ROOT}/current" ] +for command_name in curl docker pgrep python3 sha256sum stat timeout; do + command -v "${command_name}" >/dev/null 2>&1 +done +python3 -c 'import sys; assert sys.version_info >= (3, 10)' + +container_state="$(timeout --kill-after="${KILL_AFTER_SECONDS}" \ + "${REMOTE_TIMEOUT_SECONDS}" docker inspect --format \ + '{{.Id}}\t{{.State.Running}}\t{{.State.StartedAt}}\t{{.RestartCount}}\t{{if .State.Health}}{{.State.Health.Status}}{{else}}none{{end}}' signoz)" +api_status="$(timeout --kill-after="${KILL_AFTER_SECONDS}" \ + "${REMOTE_TIMEOUT_SECONDS}" curl -sS -o /dev/null -w '%{http_code}' \ + http://127.0.0.1:8080/api/v1/health)" +[ "${api_status}" = "200" ] + +active_operations="$( + (pgrep -af '(^|/|[[:space:]])(backup-signoz|clickhouse-native-backup|clickhouse-native-restore-drill|run-signoz-backup-canary)[.]sh([[:space:]]|$)' || true) \ + | wc -l | tr -d ' ' +)" +[ "${active_operations}" = "0" ] + +state=absent +drift=0 +if [ -e "${TARGET_DIR}" ] || [ -L "${TARGET_DIR}" ]; then + [ -d "${TARGET_DIR}" ] && [ ! -L "${TARGET_DIR}" ] || drift=$((drift + 1)) + for index in "${!LABELS[@]}"; do + target="${TARGET_DIR}/${LABELS[index]}" + if [ ! -f "${target}" ] || [ -L "${target}" ]; then + drift=$((drift + 1)) + continue + fi + [ "$(sha256sum "${target}" | awk '{print $1}')" = "${EXPECTED_HASHES[index]}" ] || drift=$((drift + 1)) + [ "0$(stat -c '%a' "${target}")" = "${MODES[index]}" ] || drift=$((drift + 1)) + done + state=exact + [ "${drift}" -eq 0 ] || state=drift +fi + +candidate_count=0 +for candidate in "${REMOTE_ROOT}"/."${BUNDLE_ID}".candidate.*; do + [ -e "${candidate}" ] || [ -L "${candidate}" ] || continue + candidate_count=$((candidate_count + 1)) +done +[ "${candidate_count}" -eq 0 ] +[ "${drift}" -eq 0 ] +printf 'REMOTE_CHECK terminal=check_pass_no_write state=%s drift=%s candidate_count=%s api=%s active_operations=%s container_state_sha256=%s\n' \ + "${state}" "${drift}" "${candidate_count}" "${api_status}" "${active_operations}" \ + "$(printf '%s' "${container_state}" | sha256sum | awk '{print $1}')" +REMOTE_CHECK +} + +quarantine_transport_stage() { + bounded_ssh "${SSH_OPTIONS[@]}" "${TARGET_HOST}" sudo -n bash -s -- \ + "${STAGE_DIR}" "${RECEIPT_ROOT}" "${RUN_ID}" <<'REMOTE_QUARANTINE' +set -euo pipefail +stage_dir="$1" +receipt_root="$2" +run_id="$3" +receipt_dir="${receipt_root}/${run_id}" +if [ ! -e "${stage_dir}" ] && [ ! -L "${stage_dir}" ]; then + exit 0 +fi +[ -d "${stage_dir}" ] && [ ! -L "${stage_dir}" ] +mkdir -p "${receipt_root}" +if [ ! -e "${receipt_dir}" ] && [ ! -L "${receipt_dir}" ]; then + mkdir -m 0700 "${receipt_dir}" +fi +[ -d "${receipt_dir}" ] && [ ! -L "${receipt_dir}" ] +[ ! -e "${receipt_dir}/transport-stage" ] && [ ! -L "${receipt_dir}/transport-stage" ] +mv "${stage_dir}" "${receipt_dir}/transport-stage" +REMOTE_QUARANTINE +} + +if ! REMOTE_CHECK_OUTPUT="$(remote_check)"; then + emit source_of_truth_diff failed "remote_read_only_check_failed" + emit terminal failed_no_write "remote_check_failed" + exit 1 +fi +case "${REMOTE_CHECK_OUTPUT}" in + *"terminal=check_pass_no_write state=absent "*) REMOTE_STATE=absent ;; + *"terminal=check_pass_no_write state=exact "*) REMOTE_STATE=exact ;; + *) + emit source_of_truth_diff failed "remote_check_receipt_invalid" + emit terminal failed_no_write "remote_check_receipt_invalid" + exit 1 + ;; +esac +emit source_of_truth_diff pass "remote_state_${REMOTE_STATE}_drift_0" + +if [ "${MODE}" = "--check" ]; then + emit check_mode check_pass_no_write "bundle_${BUNDLE_ID}_remote_state_${REMOTE_STATE}" + emit rollback no_write "active_pointer_unchanged" + emit terminal check_pass_no_write "source_ready_remote_state_${REMOTE_STATE}" + exit 0 +fi + +if [ "${REMOTE_STATE}" = "exact" ]; then + emit check_mode pass "immutable_bundle_already_exact" + emit bounded_execution idempotent_no_write "bundle_${BUNDLE_ID}" + emit independent_post_verifier pass "remote_exact_hash_mode_api_identity" + emit rollback no_write "active_pointer_absent" + emit learning_writeback partial_degraded "runtime_export_not_executed" + emit terminal pass_source_deployed_inactive "idempotent_runtime_export_pending" + exit 0 +fi + +if ! bounded_ssh "${SSH_OPTIONS[@]}" "${TARGET_HOST}" bash -s -- \ + "${STAGE_DIR}" <<'REMOTE_STAGE'; then +set -euo pipefail +stage_dir="$1" +[ ! -e "${stage_dir}" ] && [ ! -L "${stage_dir}" ] +stage_root="${stage_dir%/*}" +[ -d "${stage_root}" ] && [ ! -L "${stage_root}" ] && [ -w "${stage_root}" ] +umask 077 +mkdir -m 0700 "${stage_dir}" +REMOTE_STAGE + emit bounded_execution failed "transport_stage_create_failed" + emit terminal failed_no_active_pointer_write "stage_create_failed" + exit 1 +fi + +for index in "${!LOCAL_SOURCES[@]}"; do + if ! bounded_scp -q "${SSH_OPTIONS[@]}" "${LOCAL_SOURCES[index]}" \ + "${TARGET_HOST}:${STAGE_DIR}/${LABELS[index]}"; then + quarantine_transport_stage >/dev/null 2>&1 || true + emit bounded_execution failed "transport_failed_index_${index}" + emit terminal failed_no_active_pointer_write "transport_stage_quarantine_attempted" + exit 1 + fi +done + +if ! REMOTE_APPLY_OUTPUT="$(bounded_ssh "${SSH_OPTIONS[@]}" "${TARGET_HOST}" \ + sudo -n bash -s -- "${TRACE_ID}" "${RUN_ID}" "${WORK_ITEM_ID}" \ + "${REMOTE_ROOT}" "${RECEIPT_ROOT}" "${STAGE_DIR}" "${BUNDLE_ID}" \ + "${REMOTE_TIMEOUT_SECONDS}" "${KILL_AFTER_SECONDS}" \ + "${SOURCE_HASHES[@]}" <<'REMOTE_APPLY' +set -euo pipefail + +TRACE_ID="$1" +RUN_ID="$2" +WORK_ITEM_ID="$3" +REMOTE_ROOT="$4" +RECEIPT_ROOT="$5" +STAGE_DIR="$6" +BUNDLE_ID="$7" +REMOTE_TIMEOUT_SECONDS="$8" +KILL_AFTER_SECONDS="$9" +shift 9 +EXPECTED_HASHES=("$@") +LABELS=( + metadata-export-policy.json + signoz_metadata_contract.py + signoz-metadata-export.py + verify-signoz-metadata-export.py + signoz-metadata-restore-drill.py +) +MODES=(0644 0644 0755 0755 0755) +TARGET_DIR="${REMOTE_ROOT}/${BUNDLE_ID}" +CANDIDATE_DIR="${REMOTE_ROOT}/.${BUNDLE_ID}.candidate.${RUN_ID}" +RECEIPT_DIR="${RECEIPT_ROOT}/${RUN_ID}" +RECEIPT_LOG="${RECEIPT_DIR}/receipts.jsonl" +DEPLOYED=0 +TERMINAL=failed + +emit_remote() { + local phase="$1" terminal="$2" detail="$3" + printf '{"schema":"awoooi_controlled_apply_receipt_v1","trace_id":"%s","run_id":"%s","work_item_id":"%s","asset_id":"host110-signoz-metadata-toolchain","phase":"%s","terminal":"%s","detail":"%s"}\n' \ + "${TRACE_ID}" "${RUN_ID}" "${WORK_ITEM_ID}" "${phase}" "${terminal}" "${detail}" | tee -a "${RECEIPT_LOG}" +} + +finalize() { + local rc=$? quarantine_terminal=not_required + trap - EXIT HUP INT TERM + set +e + if [ -d "${RECEIPT_DIR}" ] && [ ! -L "${RECEIPT_DIR}" ]; then + if [ -d "${STAGE_DIR}" ] && [ ! -L "${STAGE_DIR}" ]; then + mv "${STAGE_DIR}" "${RECEIPT_DIR}/transport-stage" + fi + if [ "${rc}" -ne 0 ]; then + quarantine_terminal=inactive_candidates_preserved + if [ -d "${CANDIDATE_DIR}" ] && [ ! -L "${CANDIDATE_DIR}" ]; then + mv "${CANDIDATE_DIR}" "${RECEIPT_DIR}/quarantine-candidate" + fi + if [ "${DEPLOYED}" -eq 1 ] && [ -d "${TARGET_DIR}" ] && [ ! -L "${TARGET_DIR}" ]; then + mv "${TARGET_DIR}" "${RECEIPT_DIR}/quarantine-target" + fi + fi + emit_remote rollback "${quarantine_terminal}" "active_pointer_never_created" + emit_remote terminal "${TERMINAL}" "exit_${rc}_runtime_export_not_executed" + fi + exit "${rc}" +} +trap finalize EXIT HUP INT TERM + +[ "${#EXPECTED_HASHES[@]}" -eq 5 ] +[ -d "${STAGE_DIR}" ] && [ ! -L "${STAGE_DIR}" ] +[ ! -e "${TARGET_DIR}" ] && [ ! -L "${TARGET_DIR}" ] +[ ! -e "${CANDIDATE_DIR}" ] && [ ! -L "${CANDIDATE_DIR}" ] +[ ! -e "${REMOTE_ROOT}/current" ] && [ ! -L "${REMOTE_ROOT}/current" ] +[ ! -e "${RECEIPT_DIR}" ] && [ ! -L "${RECEIPT_DIR}" ] +python3 -c 'import sys; assert sys.version_info >= (3, 10)' +umask 077 +mkdir -p "${REMOTE_ROOT}" "${RECEIPT_ROOT}" +mkdir -m 0700 "${RECEIPT_DIR}" "${CANDIDATE_DIR}" +: > "${RECEIPT_LOG}" +chmod 0600 "${RECEIPT_LOG}" + +exec 8>>/tmp/awoooi-signoz-backup-operation.lock +flock -n 8 +active_operations="$( + (pgrep -af '(^|/|[[:space:]])(backup-signoz|clickhouse-native-backup|clickhouse-native-restore-drill|run-signoz-backup-canary)[.]sh([[:space:]]|$)' || true) \ + | wc -l | tr -d ' ' +)" +[ "${active_operations}" = "0" ] + +runtime_before="$(timeout --kill-after="${KILL_AFTER_SECONDS}" \ + "${REMOTE_TIMEOUT_SECONDS}" docker inspect --format \ + '{{.Id}}\t{{.State.Running}}\t{{.State.StartedAt}}\t{{.RestartCount}}\t{{if .State.Health}}{{.State.Health.Status}}{{else}}none{{end}}' signoz)" +api_before="$(timeout --kill-after="${KILL_AFTER_SECONDS}" \ + "${REMOTE_TIMEOUT_SECONDS}" curl -sS -o /dev/null -w '%{http_code}' \ + http://127.0.0.1:8080/api/v1/health)" +[ "${api_before}" = "200" ] +emit_remote sensor_source pass "staged_sources_5_runtime_api_200" + +for index in "${!LABELS[@]}"; do + staged="${STAGE_DIR}/${LABELS[index]}" + [ -f "${staged}" ] && [ ! -L "${staged}" ] + [ "$(sha256sum "${staged}" | awk '{print $1}')" = "${EXPECTED_HASHES[index]}" ] + install -o root -g root -m "${MODES[index]}" "${staged}" \ + "${CANDIDATE_DIR}/${LABELS[index]}" +done + +python3 - "${CANDIDATE_DIR}" <<'PY' +from pathlib import Path +import sys + +root = Path(sys.argv[1]) +for name in ( + "signoz_metadata_contract.py", + "signoz-metadata-export.py", + "verify-signoz-metadata-export.py", + "signoz-metadata-restore-drill.py", +): + path = root / name + compile(path.read_text(encoding="utf-8"), str(path), "exec") +PY +PYTHONPATH="${CANDIDATE_DIR}" python3 "${CANDIDATE_DIR}/signoz-metadata-export.py" \ + --check --base-url https://signoz.invalid \ + --output-dir "${RECEIPT_DIR}/offline-output-must-not-exist" \ + --policy "${CANDIDATE_DIR}/metadata-export-policy.json" \ + --trace-id "${TRACE_ID}" --run-id "${RUN_ID}" \ + --work-item-id "${WORK_ITEM_ID}" >/dev/null +[ ! -e "${RECEIPT_DIR}/offline-output-must-not-exist" ] +emit_remote check_mode pass "candidate_hash_mode_compile_policy_check" + +mv "${CANDIDATE_DIR}" "${TARGET_DIR}" +chmod 0750 "${TARGET_DIR}" +DEPLOYED=1 + +for index in "${!LABELS[@]}"; do + target="${TARGET_DIR}/${LABELS[index]}" + [ -f "${target}" ] && [ ! -L "${target}" ] + [ "$(sha256sum "${target}" | awk '{print $1}')" = "${EXPECTED_HASHES[index]}" ] + [ "0$(stat -c '%a' "${target}")" = "${MODES[index]}" ] + printf '%s\t%s\t%s\n' "${LABELS[index]}" "${EXPECTED_HASHES[index]}" "${MODES[index]}" \ + >> "${RECEIPT_DIR}/deployed-manifest.tsv" +done +chmod 0600 "${RECEIPT_DIR}/deployed-manifest.tsv" +runtime_after="$(timeout --kill-after="${KILL_AFTER_SECONDS}" \ + "${REMOTE_TIMEOUT_SECONDS}" docker inspect --format \ + '{{.Id}}\t{{.State.Running}}\t{{.State.StartedAt}}\t{{.RestartCount}}\t{{if .State.Health}}{{.State.Health.Status}}{{else}}none{{end}}' signoz)" +api_after="$(timeout --kill-after="${KILL_AFTER_SECONDS}" \ + "${REMOTE_TIMEOUT_SECONDS}" curl -sS -o /dev/null -w '%{http_code}' \ + http://127.0.0.1:8080/api/v1/health)" +[ "${runtime_after}" = "${runtime_before}" ] +[ "${api_after}" = "200" ] +emit_remote bounded_execution pass "immutable_bundle_deployed_active_pointer_0" +emit_remote independent_post_verifier pass "hash_mode_compile_api_identity_unchanged" +emit_remote learning_writeback partial_degraded "authenticated_export_restore_pending" +TERMINAL=pass_source_deployed_inactive +REMOTE_APPLY +)"; then + quarantine_transport_stage >/dev/null 2>&1 || true + emit bounded_execution failed "remote_apply_failed_inactive_candidate_quarantined" + emit terminal failed_no_active_pointer_write "runtime_export_not_executed" + exit 1 +fi +case "${REMOTE_APPLY_OUTPUT}" in + *'"phase":"terminal","terminal":"pass_source_deployed_inactive"'*) ;; + *) + emit independent_post_verifier failed "remote_apply_terminal_invalid" + emit terminal failed_no_active_pointer_write "runtime_export_not_executed" + exit 1 + ;; +esac + +if ! POSTCHECK_OUTPUT="$(remote_check)"; then + emit independent_post_verifier failed "remote_postcheck_failed" + emit terminal failed_source_deploy_unverified "active_pointer_0" + exit 1 +fi +case "${POSTCHECK_OUTPUT}" in + *"terminal=check_pass_no_write state=exact "*) ;; + *) + emit independent_post_verifier failed "remote_postcheck_receipt_invalid" + emit terminal failed_source_deploy_unverified "active_pointer_0" + exit 1 + ;; +esac +emit check_mode pass "precheck_absent" +emit bounded_execution pass "immutable_bundle_${BUNDLE_ID}_active_pointer_0" +emit independent_post_verifier pass "postcheck_exact_hash_mode_api_identity" +emit rollback not_required "inactive_additive_bundle" +emit learning_writeback partial_degraded "runtime_export_restore_zero_residue_pending" +emit terminal pass_source_deployed_inactive "runtime_export_not_executed" diff --git a/scripts/ops/tests/test_signoz_metadata_toolchain_deploy.py b/scripts/ops/tests/test_signoz_metadata_toolchain_deploy.py new file mode 100644 index 000000000..b2d1d130e --- /dev/null +++ b/scripts/ops/tests/test_signoz_metadata_toolchain_deploy.py @@ -0,0 +1,179 @@ +from __future__ import annotations + +import json +import os +import subprocess +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[3] +DEPLOYER = ROOT / "scripts" / "ops" / "deploy-signoz-metadata-toolchain.sh" + + +def _write_executable(path: Path, text: str) -> None: + path.write_text(text, encoding="utf-8") + path.chmod(0o755) + + +def fake_environment(tmp_path: Path, *, mode: str) -> dict[str, str]: + fake_bin = tmp_path / "bin" + fake_bin.mkdir() + event_log = tmp_path / "events.log" + ssh_count = tmp_path / "ssh.count" + event_log.touch() + ssh_count.write_text("0\n", encoding="utf-8") + + _write_executable( + fake_bin / "timeout", + """#!/bin/bash +set -eu +while [[ "${1:-}" == --kill-after=* ]]; do shift; done +shift +exec "$@" +""", + ) + _write_executable( + fake_bin / "ssh", + """#!/bin/bash +set -eu +count="$(cat "${TEST_SSH_COUNT:?}")" +count=$((count + 1)) +printf '%s\n' "$count" > "${TEST_SSH_COUNT:?}" +printf 'ssh %s\n' "$*" >> "${TEST_EVENT_LOG:?}" +cat >/dev/null || true +case "${FAKE_SSH_MODE:?}" in + check_absent) + printf 'REMOTE_CHECK terminal=check_pass_no_write state=absent drift=0 candidate_count=0 api=200 active_operations=0 container_state_sha256=%064d\n' 0 + ;; + check_drift) + exit 3 + ;; + apply) + case "$count" in + 1) + printf 'REMOTE_CHECK terminal=check_pass_no_write state=absent drift=0 candidate_count=0 api=200 active_operations=0 container_state_sha256=%064d\n' 0 + ;; + 2) ;; + 3) + printf '{"phase":"terminal","terminal":"pass_source_deployed_inactive"}\n' + ;; + 4) + printf 'REMOTE_CHECK terminal=check_pass_no_write state=exact drift=0 candidate_count=0 api=200 active_operations=0 container_state_sha256=%064d\n' 0 + ;; + *) exit 90 ;; + esac + ;; +esac +""", + ) + _write_executable( + fake_bin / "scp", + """#!/bin/bash +set -eu +printf 'scp %s\n' "$*" >> "${TEST_EVENT_LOG:?}" +""", + ) + return { + **os.environ, + "PATH": f"{fake_bin}:{os.environ['PATH']}", + "TEST_EVENT_LOG": str(event_log), + "TEST_SSH_COUNT": str(ssh_count), + "FAKE_SSH_MODE": mode, + "TRACE_ID": "trace-P0-OBS-002-metadata-deploy", + "RUN_ID": f"run-{mode}", + "WORK_ITEM_ID": "P0-OBS-002", + } + + +def run_deployer( + tmp_path: Path, *, mode: str, apply: bool +) -> subprocess.CompletedProcess[str]: + return subprocess.run( + ["bash", str(DEPLOYER), "--apply" if apply else "--check"], + env=fake_environment(tmp_path, mode=mode), + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=20, + check=False, + ) + + +def receipt_rows(stdout: str) -> list[dict[str, object]]: + return [json.loads(line) for line in stdout.splitlines() if line.startswith("{")] + + +def test_help_documents_inactive_immutable_contract() -> None: + result = subprocess.run( + ["bash", str(DEPLOYER), "--help"], + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=10, + check=False, + ) + assert result.returncode == 0 + assert "immutable exact-hash directory" in result.stdout + assert "does not create or" in result.stdout + assert "active pointer" in result.stdout + + +def test_check_mode_is_no_write_and_ready_when_bundle_absent(tmp_path: Path) -> None: + result = run_deployer(tmp_path, mode="check_absent", apply=False) + assert result.returncode == 0, result.stderr + rows = receipt_rows(result.stdout) + assert rows[-1]["terminal"] == "check_pass_no_write" + assert rows[-1]["detail"].endswith("remote_state_absent") + events = (tmp_path / "events.log").read_text(encoding="utf-8").splitlines() + assert len([line for line in events if line.startswith("ssh ")]) == 1 + assert not [line for line in events if line.startswith("scp ")] + + +def test_check_mode_fails_closed_on_remote_drift(tmp_path: Path) -> None: + result = run_deployer(tmp_path, mode="check_drift", apply=False) + assert result.returncode == 1 + rows = receipt_rows(result.stdout) + assert rows[-1]["terminal"] == "failed_no_write" + + +def test_apply_uses_five_transfers_and_exact_postcheck(tmp_path: Path) -> None: + result = run_deployer(tmp_path, mode="apply", apply=True) + assert result.returncode == 0, result.stderr + rows = receipt_rows(result.stdout) + assert rows[-1]["terminal"] == "pass_source_deployed_inactive" + assert rows[-1]["detail"] == "runtime_export_not_executed" + events = (tmp_path / "events.log").read_text(encoding="utf-8").splitlines() + assert len([line for line in events if line.startswith("scp ")]) == 5 + assert len([line for line in events if line.startswith("ssh ")]) == 4 + + +def test_deployer_has_no_active_pointer_or_destructive_fallback() -> None: + source = DEPLOYER.read_text(encoding="utf-8") + assert '"${REMOTE_ROOT}/current"' in source + assert "active_pointer_0" in source + assert "pass_source_deployed_inactive" in source + assert "quarantine-candidate" in source + assert "quarantine-target" in source + assert "docker stop" not in source + assert "docker start" not in source + assert "docker restart" not in source + assert "sqlite3" not in source + assert "github.com" not in source.casefold() + assert "curl |" not in source + assert "curl -fsSL" not in source + assert "\nrm " not in source + assert "ln -s" not in source + + +def test_exact_five_file_supply_chain_and_modes_are_fixed() -> None: + source = DEPLOYER.read_text(encoding="utf-8") + for path in ( + "config/signoz/metadata-export-policy.json", + "scripts/backup/signoz_metadata_contract.py", + "scripts/backup/signoz-metadata-export.py", + "scripts/backup/verify-signoz-metadata-export.py", + "scripts/backup/signoz-metadata-restore-drill.py", + ): + assert path in source + assert "MODES=(0644 0644 0755 0755 0755)" in source + assert '[ "${#EXPECTED_HASHES[@]}" -eq 5 ]' in source diff --git a/scripts/reboot-recovery/signoz-canonical-route-preflight.py b/scripts/reboot-recovery/signoz-canonical-route-preflight.py index 6694cda2a..6b69c217f 100644 --- a/scripts/reboot-recovery/signoz-canonical-route-preflight.py +++ b/scripts/reboot-recovery/signoz-canonical-route-preflight.py @@ -31,7 +31,7 @@ EXPECTED_POST_CLOSURE_BLOCKERS = [ "signoz_backup_offsite_dr_pending", "clickhouse_native_failed_artifact_retention_pending", "clickhouse_native_resume_submitted_inventory_pending", - "service_registry_runtime_mirror_reconciliation_pending", + "service_registry_production_and_live_host_readback_pending", "github_freeze_legacy_asset_retirement_pending", "monitoring_generator_duplicate_awoooi_api_identity_pending", ] @@ -46,6 +46,24 @@ TOOLCHAIN_SOURCE_PATHS = { "ops/signoz/clickhouse/restore-drill/config.d/isolated-cluster.xml" ), } +METADATA_EXPORT_SOURCE_PATHS = { + "policy_sha256": "config/signoz/metadata-export-policy.json", + "exporter_sha256": "scripts/backup/signoz-metadata-export.py", + "shared_contract_sha256": "scripts/backup/signoz_metadata_contract.py", + "independent_export_verifier_sha256": ( + "scripts/backup/verify-signoz-metadata-export.py" + ), + "isolated_restore_driver_sha256": ( + "scripts/backup/signoz-metadata-restore-drill.py" + ), + "contract_tests_sha256": ( + "scripts/backup/tests/test_signoz_metadata_export_contract.py" + ), + "controlled_deployer_sha256": ("scripts/ops/deploy-signoz-metadata-toolchain.sh"), + "deployer_tests_sha256": ( + "scripts/ops/tests/test_signoz_metadata_toolchain_deploy.py" + ), +} HISTORICAL_PRODUCTION_TOOLCHAIN_HASHES = { "backup_orchestrator_sha256": ( "2fff0ded4ece9104b910f9e2db6deb4eebdc173d12c2be2bad2a59596e0d7520" @@ -101,6 +119,19 @@ def _current_toolchain_source_hashes(root: Path) -> dict[str, str]: return hashes +def _current_metadata_export_source_hashes(root: Path) -> dict[str, str]: + hashes: dict[str, str] = {} + for field, relative in METADATA_EXPORT_SOURCE_PATHS.items(): + source = root / relative + if not source.is_file() or source.is_symlink(): + return {} + try: + hashes[field] = _sha256_file(source) + except OSError: + return {} + return hashes + + def evaluate(root: Path, contract_path: Path) -> dict[str, Any]: contract = _load_yaml(contract_path) post_closure = contract.get("runtime_observations", {}).get( @@ -318,14 +349,17 @@ def evaluate(root: Path, contract_path: Path) -> dict[str, Any]: == 27 and asset_reconciliation.get("runtime_mirror_service_count") == len(runtime_services) - == 24 + == 27 and asset_reconciliation.get("exact_shared_service_count") == len(exact_shared_services) - == 24 + == 27 and asset_reconciliation.get("shared_service_drift_count") == 0 and asset_reconciliation.get("source_only_services") == source_only_services - == ["bitan-app", "sentry", "signoz"] + == [] + and asset_reconciliation.get("source_runtime_manifest_exact") is True + and asset_reconciliation.get("production_runtime_readback_status") + == "pending_current_sha_readback" and source_services.get("signoz-clickhouse") == runtime_services.get("signoz-clickhouse") and asset_reconciliation.get("signoz_clickhouse_exact_match") is True @@ -823,6 +857,16 @@ def evaluate(root: Path, contract_path: Path) -> dict[str, Any]: sqlite_backup = pending_verifiers.get( "signoz_sqlite_application_consistent_backup", {} ) + metadata_export_source_contract = sqlite_backup.get("source_contract", {}) + declared_metadata_export_hashes = metadata_export_source_contract.get("hashes", {}) + actual_metadata_export_hashes = _current_metadata_export_source_hashes(root) + checks["signoz_metadata_export_source_hashes_bound"] = ( + metadata_export_source_contract.get("expected_file_count") + == len(METADATA_EXPORT_SOURCE_PATHS) + and set(declared_metadata_export_hashes) == set(METADATA_EXPORT_SOURCE_PATHS) + and declared_metadata_export_hashes == actual_metadata_export_hashes + and metadata_export_source_contract.get("test_terminal") == "19_passed" + ) offsite_backup = pending_verifiers.get("signoz_backup_offsite_dr", {}) failed_artifacts = pending_verifiers.get( "clickhouse_native_failed_artifact_retention", {} @@ -844,6 +888,26 @@ def evaluate(root: Path, contract_path: Path) -> dict[str, Any]: == "pending_supported_non_raw_export_or_coordinated_snapshot" and sqlite_backup.get("raw_sqlite_read_or_sync_allowed") is False and sqlite_backup.get("sqlite_cli_present_in_runtime") is False + and sqlite_backup.get("source_contract_status") + == "implemented_tested_not_deployed" + and metadata_export_source_contract.get("policy") + == "config/signoz/metadata-export-policy.json" + and metadata_export_source_contract.get("exporter") + == "scripts/backup/signoz-metadata-export.py" + and metadata_export_source_contract.get("independent_export_verifier") + == "scripts/backup/verify-signoz-metadata-export.py" + and metadata_export_source_contract.get("isolated_restore_driver") + == "scripts/backup/signoz-metadata-restore-drill.py" + and metadata_export_source_contract.get("raw_sqlite_read") is False + and metadata_export_source_contract.get("raw_volume_read") is False + and metadata_export_source_contract.get("secret_value_persistence") is False + and metadata_export_source_contract.get("full_sqlite_completion_claim") is False + and sqlite_backup.get("runtime_export_terminal") + == "pending_authenticated_stable_export" + and sqlite_backup.get("runtime_restore_terminal") + == "pending_isolated_same_version_target" + and sqlite_backup.get("runtime_zero_residue_terminal") == "pending" + and len(sqlite_backup.get("required_preconditions", [])) == 6 and offsite_backup.get("status") == "pending_offsite_snapshot_and_restore_verifier" and offsite_backup.get("current_repository_scope") @@ -974,8 +1038,27 @@ def evaluate(root: Path, contract_path: Path) -> dict[str, Any]: == "pending_durable_submitted_inventory_contract" and post_closure.get("signoz_sqlite_application_consistent_backup_status") == "pending" + and post_closure.get("signoz_metadata_export_source_contract_status") + == sqlite_backup.get("source_contract_status") + and post_closure.get("signoz_metadata_export_policy") + == sqlite_backup.get("source_contract", {}).get("policy") + and post_closure.get("signoz_metadata_exporter") + == sqlite_backup.get("source_contract", {}).get("exporter") + and post_closure.get("signoz_metadata_export_verifier") + == sqlite_backup.get("source_contract", {}).get("independent_export_verifier") + and post_closure.get("signoz_metadata_restore_driver") + == sqlite_backup.get("source_contract", {}).get("isolated_restore_driver") + and post_closure.get("signoz_metadata_controlled_deployer") + == sqlite_backup.get("source_contract", {}).get("controlled_deployer") + and post_closure.get("signoz_metadata_runtime_export_terminal") + == sqlite_backup.get("runtime_export_terminal") + and post_closure.get("signoz_metadata_runtime_restore_terminal") + == sqlite_backup.get("runtime_restore_terminal") + and post_closure.get("signoz_metadata_zero_residue_terminal") + == sqlite_backup.get("runtime_zero_residue_terminal") + and post_closure.get("signoz_metadata_full_sqlite_completion_claim") is False and post_closure.get("service_registry_runtime_mirror_status") - == "partial_degraded" + == "source_manifest_exact_production_readback_pending" and post_closure.get("service_registry_runtime_mirror_work_item_id") == "P0-OBS-002-ASSET-DRIFT-001" and post_closure.get("github_freeze_legacy_asset_status") diff --git a/scripts/reboot-recovery/tests/test_signoz_canonical_route_migration.py b/scripts/reboot-recovery/tests/test_signoz_canonical_route_migration.py index 410f5e68e..6ac48442b 100644 --- a/scripts/reboot-recovery/tests/test_signoz_canonical_route_migration.py +++ b/scripts/reboot-recovery/tests/test_signoz_canonical_route_migration.py @@ -834,6 +834,62 @@ def test_post_release_verifier_and_native_backup_close_with_remaining_gaps() -> ] assert sqlite_pending["raw_sqlite_read_or_sync_allowed"] is False assert sqlite_pending["sqlite_cli_present_in_runtime"] is False + assert sqlite_pending["source_contract_status"] == ( + "implemented_tested_not_deployed" + ) + source_contract = sqlite_pending["source_contract"] + assert source_contract["policy"] == "config/signoz/metadata-export-policy.json" + assert source_contract["exporter"] == ("scripts/backup/signoz-metadata-export.py") + assert source_contract["independent_export_verifier"] == ( + "scripts/backup/verify-signoz-metadata-export.py" + ) + assert source_contract["isolated_restore_driver"] == ( + "scripts/backup/signoz-metadata-restore-drill.py" + ) + assert source_contract["controlled_deployer"] == ( + "scripts/ops/deploy-signoz-metadata-toolchain.sh" + ) + assert source_contract["deployer_tests"] == ( + "scripts/ops/tests/test_signoz_metadata_toolchain_deploy.py" + ) + assert source_contract["expected_file_count"] == 8 + metadata_paths = { + "policy_sha256": "config/signoz/metadata-export-policy.json", + "exporter_sha256": "scripts/backup/signoz-metadata-export.py", + "shared_contract_sha256": "scripts/backup/signoz_metadata_contract.py", + "independent_export_verifier_sha256": ( + "scripts/backup/verify-signoz-metadata-export.py" + ), + "isolated_restore_driver_sha256": ( + "scripts/backup/signoz-metadata-restore-drill.py" + ), + "contract_tests_sha256": ( + "scripts/backup/tests/test_signoz_metadata_export_contract.py" + ), + "controlled_deployer_sha256": ( + "scripts/ops/deploy-signoz-metadata-toolchain.sh" + ), + "deployer_tests_sha256": ( + "scripts/ops/tests/test_signoz_metadata_toolchain_deploy.py" + ), + } + assert source_contract["hashes"] == { + field: hashlib.sha256((ROOT / relative).read_bytes()).hexdigest() + for field, relative in metadata_paths.items() + } + assert source_contract["test_terminal"] == "19_passed" + assert source_contract["raw_sqlite_read"] is False + assert source_contract["raw_volume_read"] is False + assert source_contract["secret_value_persistence"] is False + assert source_contract["full_sqlite_completion_claim"] is False + assert sqlite_pending["runtime_export_terminal"] == ( + "pending_authenticated_stable_export" + ) + assert sqlite_pending["runtime_restore_terminal"] == ( + "pending_isolated_same_version_target" + ) + assert sqlite_pending["runtime_zero_residue_terminal"] == "pending" + assert len(sqlite_pending["required_preconditions"]) == 6 offsite = receipt["pending_verifiers"]["signoz_backup_offsite_dr"] assert offsite["current_repository_scope"] == "local_same_failure_domain" assert offsite["offsite_verified"] is False @@ -878,10 +934,14 @@ def test_post_release_verifier_and_native_backup_close_with_remaining_gaps() -> assets = receipt["asset_reconciliation"] assert assets["drift_work_item_id"] == "P0-OBS-002-ASSET-DRIFT-001" assert assets["source_service_count"] == 27 - assert assets["runtime_mirror_service_count"] == 24 - assert assets["exact_shared_service_count"] == 24 + assert assets["runtime_mirror_service_count"] == 27 + assert assets["exact_shared_service_count"] == 27 assert assets["shared_service_drift_count"] == 0 - assert assets["source_only_services"] == ["bitan-app", "sentry", "signoz"] + assert assets["source_only_services"] == [] + assert assets["source_runtime_manifest_exact"] is True + assert assets["production_runtime_readback_status"] == ( + "pending_current_sha_readback" + ) assert assets["signoz_clickhouse_exact_match"] is True assert assets["live_signoz_query_host"] == "192.168.0.110" assert assets["source_signoz_host"] == "192.168.0.188" @@ -905,7 +965,7 @@ def test_post_release_verifier_and_native_backup_close_with_remaining_gaps() -> "signoz_backup_offsite_dr_pending", "clickhouse_native_failed_artifact_retention_pending", "clickhouse_native_resume_submitted_inventory_pending", - "service_registry_runtime_mirror_reconciliation_pending", + "service_registry_production_and_live_host_readback_pending", "github_freeze_legacy_asset_retirement_pending", "monitoring_generator_duplicate_awoooi_api_identity_pending", ] @@ -1002,8 +1062,34 @@ def test_post_closure_contract_mirror_and_program_blockers_are_consistent() -> N assert ( post_closure["signoz_sqlite_application_consistent_backup_status"] == "pending" ) + assert post_closure["signoz_metadata_export_source_contract_status"] == ( + "implemented_tested_not_deployed" + ) + assert post_closure["signoz_metadata_export_policy"] == ( + "config/signoz/metadata-export-policy.json" + ) + assert post_closure["signoz_metadata_exporter"] == ( + "scripts/backup/signoz-metadata-export.py" + ) + assert post_closure["signoz_metadata_export_verifier"] == ( + "scripts/backup/verify-signoz-metadata-export.py" + ) + assert post_closure["signoz_metadata_restore_driver"] == ( + "scripts/backup/signoz-metadata-restore-drill.py" + ) + assert post_closure["signoz_metadata_controlled_deployer"] == ( + "scripts/ops/deploy-signoz-metadata-toolchain.sh" + ) + assert post_closure["signoz_metadata_runtime_export_terminal"] == ( + "pending_authenticated_stable_export" + ) + assert post_closure["signoz_metadata_runtime_restore_terminal"] == ( + "pending_isolated_same_version_target" + ) + assert post_closure["signoz_metadata_zero_residue_terminal"] == "pending" + assert post_closure["signoz_metadata_full_sqlite_completion_claim"] is False assert post_closure["service_registry_runtime_mirror_status"] == ( - "partial_degraded" + "source_manifest_exact_production_readback_pending" ) assert post_closure["service_registry_runtime_mirror_work_item_id"] == ( "P0-OBS-002-ASSET-DRIFT-001"