511 lines
18 KiB
Python
Executable File
511 lines
18 KiB
Python
Executable File
#!/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())
|