wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 3 Packages Projects Releases Wiki Activity

feat(observability): add safe SigNoz metadata backup lane

Browse Source
This commit is contained in:
ogt
2026-07-15 13:57:13 +08:00
parent 4dc5b51d0f
commit 1730c5bb71
12 changed files with 3485 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

371
scripts/backup/signoz-metadata-export.py Executable file
View File

@@ -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())

510
scripts/backup/signoz-metadata-restore-drill.py Executable file
View File

@@ -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())

551
scripts/backup/signoz_metadata_contract.py Executable file
View File

@@ -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

602
scripts/backup/tests/test_signoz_metadata_export_contract.py Normal file
View File

@@ -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()

308
scripts/backup/verify-signoz-metadata-export.py Executable file
View File

@@ -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())

505
scripts/ops/deploy-signoz-metadata-toolchain.sh Executable file
View File

@@ -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"

179
scripts/ops/tests/test_signoz_metadata_toolchain_deploy.py Normal file
View File

@@ -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

93
scripts/reboot-recovery/signoz-canonical-route-preflight.py
View File

@@ -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")

96
scripts/reboot-recovery/tests/test_signoz_canonical_route_migration.py
View File

@@ -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"