917 lines
34 KiB
Python
917 lines
34 KiB
Python
#!/usr/bin/env python3
|
|
"""Fixed Host110 target for Agent99 SigNoz metadata toolchain deployment."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import fcntl
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import signal
|
|
import stat
|
|
import subprocess # nosec B404 - fixed executable and argv vectors only
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
|
|
SHA256 = re.compile(r"^[0-9a-f]{64}$")
|
|
LABELS = (
|
|
"metadata-export-policy.json",
|
|
"signoz_metadata_contract.py",
|
|
"signoz-metadata-export.py",
|
|
"verify-signoz-metadata-export.py",
|
|
"signoz-metadata-restore-drill.py",
|
|
"signoz-metadata-isolated-restore.py",
|
|
"isolated-cluster.xml",
|
|
)
|
|
MODES = (0o644, 0o644, 0o755, 0o755, 0o755, 0o755, 0o644)
|
|
ROOT_UID = 0
|
|
REMOTE_ROOT = Path("/backup/toolchains/signoz-metadata")
|
|
RECEIPT_ROOT = Path("/backup/deploy-receipts/signoz-metadata-toolchain")
|
|
OPERATION_LOCK = Path("/tmp/awoooi-signoz-backup-operation.lock")
|
|
SIGNOZ_IMAGE_REF = "signoz/signoz:v0.113.0"
|
|
SIGNOZ_IMAGE_ID = (
|
|
"sha256:c17991d10966aedcb0d4d83805a0baf5310f1553d90611629036fcfaee8d969b"
|
|
)
|
|
ACTIVE_PATTERN = (
|
|
r"([s]ignoz-metadata-export[.]py|[s]ignoz-metadata-restore-drill[.]py|"
|
|
r"[s]ignoz-metadata-isolated-restore[.]py|(^|/|[[:space:]])"
|
|
r"([b]ackup-signoz|[c]lickhouse-native-backup|"
|
|
r"[c]lickhouse-native-restore-drill|[r]un-signoz-backup-canary)"
|
|
r"[.]sh([[:space:]]|$))"
|
|
)
|
|
MUTATION_STATE = {
|
|
"source_write_attempted": False,
|
|
"source_write_performed": False,
|
|
"quarantine_performed": False,
|
|
"stage_cleanup_performed": False,
|
|
}
|
|
|
|
|
|
class TargetError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def canonical_bytes(value: Any) -> bytes:
|
|
return (
|
|
json.dumps(value, ensure_ascii=True, 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 parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument(
|
|
"--mode",
|
|
choices=("check", "prepare", "apply", "verify", "reconcile"),
|
|
required=True,
|
|
)
|
|
parser.add_argument("--trace-id", required=True)
|
|
parser.add_argument("--run-id", required=True)
|
|
parser.add_argument("--work-item-id", required=True)
|
|
parser.add_argument("--source-revision", required=True)
|
|
parser.add_argument("--bundle-id", required=True)
|
|
parser.add_argument("--manifest-b64", required=True)
|
|
parser.add_argument("--stage-dir")
|
|
parser.add_argument("--attempt-id", default="")
|
|
return parser.parse_args()
|
|
|
|
|
|
def validate_args(args: argparse.Namespace) -> list[dict[str, Any]]:
|
|
for value in (args.trace_id, args.run_id, args.work_item_id):
|
|
if not SAFE_ID.fullmatch(value):
|
|
raise TargetError("identity_invalid")
|
|
if args.mode == "reconcile":
|
|
if not SAFE_ID.fullmatch(args.attempt_id):
|
|
raise TargetError("reconcile_attempt_id_invalid")
|
|
elif args.attempt_id:
|
|
raise TargetError("attempt_id_not_allowed")
|
|
if args.work_item_id != "P0-OBS-002":
|
|
raise TargetError("work_item_not_allowlisted")
|
|
if not re.fullmatch(r"[0-9a-f]{40}", args.source_revision):
|
|
raise TargetError("source_revision_invalid")
|
|
if not SHA256.fullmatch(args.bundle_id):
|
|
raise TargetError("bundle_id_invalid")
|
|
try:
|
|
raw = base64.b64decode(args.manifest_b64, validate=True)
|
|
manifest = json.loads(raw)
|
|
except (ValueError, json.JSONDecodeError) as exc:
|
|
raise TargetError("manifest_invalid") from exc
|
|
if not isinstance(manifest, list) or len(manifest) != len(LABELS):
|
|
raise TargetError("manifest_shape_invalid")
|
|
normalized: list[dict[str, Any]] = []
|
|
for index, row in enumerate(manifest):
|
|
if not isinstance(row, dict):
|
|
raise TargetError("manifest_row_invalid")
|
|
expected = {
|
|
"name": LABELS[index],
|
|
"sha256": row.get("sha256"),
|
|
"mode": f"{MODES[index]:04o}",
|
|
}
|
|
if (
|
|
row.get("name") != expected["name"]
|
|
or not isinstance(expected["sha256"], str)
|
|
or not SHA256.fullmatch(expected["sha256"])
|
|
or row.get("mode") != expected["mode"]
|
|
or set(row) != {"name", "sha256", "mode"}
|
|
):
|
|
raise TargetError("manifest_row_contract_invalid")
|
|
normalized.append(expected)
|
|
bundle_id = sha256_bytes(
|
|
"".join(f"{row['sha256']}\n" for row in normalized).encode("ascii")
|
|
)
|
|
if bundle_id != args.bundle_id:
|
|
raise TargetError("bundle_id_manifest_mismatch")
|
|
expected_stage = Path(f"/tmp/awoooi-signoz-metadata-toolchain.{args.run_id}")
|
|
if args.mode in {"prepare", "apply", "reconcile"}:
|
|
if args.stage_dir != str(expected_stage):
|
|
raise TargetError("stage_path_not_exact")
|
|
elif args.stage_dir:
|
|
raise TargetError("stage_path_not_allowed")
|
|
return normalized
|
|
|
|
|
|
def run(argv: list[str], timeout: int = 20) -> str:
|
|
try:
|
|
result = subprocess.run( # nosec B603
|
|
argv,
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.DEVNULL,
|
|
text=True,
|
|
timeout=timeout,
|
|
check=False,
|
|
env={"PATH": "/usr/sbin:/usr/bin:/sbin:/bin", "LANG": "C.UTF-8"},
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired) as exc:
|
|
raise TargetError("bounded_command_failed") from exc
|
|
if result.returncode != 0:
|
|
raise TargetError("bounded_command_failed")
|
|
return result.stdout.strip()
|
|
|
|
|
|
def production_snapshot() -> dict[str, Any]:
|
|
output = run(["/usr/bin/docker", "inspect", "signoz"])
|
|
try:
|
|
rows = json.loads(output)
|
|
except json.JSONDecodeError as exc:
|
|
raise TargetError("signoz_runtime_readback_invalid") from exc
|
|
if not isinstance(rows, list) or len(rows) != 1 or not isinstance(rows[0], dict):
|
|
raise TargetError("signoz_runtime_readback_invalid")
|
|
row = rows[0]
|
|
config = row.get("Config")
|
|
state = row.get("State")
|
|
value = {
|
|
"id": row.get("Id"),
|
|
"image_id": row.get("Image"),
|
|
"image_ref": config.get("Image") if isinstance(config, dict) else None,
|
|
"running": state.get("Running") if isinstance(state, dict) else None,
|
|
"started_at": state.get("StartedAt") if isinstance(state, dict) else None,
|
|
"restart_count": row.get("RestartCount"),
|
|
}
|
|
if (
|
|
not isinstance(value["id"], str)
|
|
or len(value["id"]) != 64
|
|
or value["image_id"] != SIGNOZ_IMAGE_ID
|
|
or value["image_ref"] != SIGNOZ_IMAGE_REF
|
|
or value["running"] is not True
|
|
or not isinstance(value["started_at"], str)
|
|
or not isinstance(value["restart_count"], int)
|
|
):
|
|
raise TargetError("signoz_runtime_identity_drift")
|
|
return value
|
|
|
|
|
|
def require_health() -> None:
|
|
request = urllib.request.Request("http://127.0.0.1:8080/api/v1/health")
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=10) as response: # nosec B310
|
|
if response.status != 200:
|
|
raise TargetError("signoz_health_not_200")
|
|
response.read(4097)
|
|
except (urllib.error.URLError, TimeoutError, OSError) as exc:
|
|
raise TargetError("signoz_health_unavailable") from exc
|
|
|
|
|
|
def require_no_active_operations() -> None:
|
|
try:
|
|
result = subprocess.run( # nosec B603
|
|
["/usr/bin/pgrep", "-af", ACTIVE_PATTERN],
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.DEVNULL,
|
|
text=True,
|
|
timeout=10,
|
|
check=False,
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired) as exc:
|
|
raise TargetError("active_operation_readback_failed") from exc
|
|
if result.returncode != 1 or result.stdout.strip():
|
|
raise TargetError("active_operation_present_or_unverified")
|
|
|
|
|
|
def require_directory(
|
|
path: Path,
|
|
*,
|
|
owner: int | None = ROOT_UID,
|
|
mode: int | tuple[int, ...] | None = None,
|
|
) -> None:
|
|
try:
|
|
metadata = path.lstat()
|
|
except OSError as exc:
|
|
raise TargetError("directory_unavailable") from exc
|
|
if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
|
|
raise TargetError("directory_not_safe")
|
|
if owner is not None and metadata.st_uid != owner:
|
|
raise TargetError("directory_owner_invalid")
|
|
if mode is not None:
|
|
allowed_modes = (mode,) if isinstance(mode, int) else mode
|
|
if stat.S_IMODE(metadata.st_mode) not in allowed_modes:
|
|
raise TargetError("directory_mode_invalid")
|
|
|
|
|
|
def ensure_private_root(path: Path) -> None:
|
|
if os.path.lexists(path):
|
|
require_directory(path, owner=ROOT_UID, mode=0o700)
|
|
return
|
|
parent = path.parent
|
|
require_directory(parent, owner=ROOT_UID)
|
|
path.mkdir(mode=0o700)
|
|
|
|
|
|
def target_state(bundle_id: str, manifest: list[dict[str, Any]]) -> str:
|
|
if os.path.lexists(REMOTE_ROOT / "current"):
|
|
raise TargetError("active_pointer_must_remain_absent")
|
|
target = REMOTE_ROOT / bundle_id
|
|
if not os.path.lexists(target):
|
|
return "absent"
|
|
require_directory(target, owner=ROOT_UID, mode=0o750)
|
|
if {item.name for item in target.iterdir()} != set(LABELS):
|
|
raise TargetError("target_tree_drift")
|
|
for row in manifest:
|
|
path = target / row["name"]
|
|
metadata = path.lstat()
|
|
if (
|
|
stat.S_ISLNK(metadata.st_mode)
|
|
or not stat.S_ISREG(metadata.st_mode)
|
|
or metadata.st_uid != ROOT_UID
|
|
or stat.S_IMODE(metadata.st_mode) != int(row["mode"], 8)
|
|
or sha256_file(path) != row["sha256"]
|
|
):
|
|
raise TargetError("target_file_drift")
|
|
return "exact"
|
|
|
|
|
|
def base_receipt(
|
|
args: argparse.Namespace, terminal: str, **extra: Any
|
|
) -> dict[str, Any]:
|
|
value = {
|
|
"schema": "awoooi_agent99_signoz_toolchain_target_v1",
|
|
"trace_id": args.trace_id,
|
|
"run_id": args.run_id,
|
|
"work_item_id": args.work_item_id,
|
|
"source_revision": args.source_revision,
|
|
"mode": args.mode,
|
|
"bundle_id": args.bundle_id,
|
|
"terminal": terminal,
|
|
"production_service_mutation_performed": False,
|
|
"active_pointer_created": False,
|
|
"credential_value_read": False,
|
|
"completion_claim": False,
|
|
**MUTATION_STATE,
|
|
}
|
|
if args.attempt_id:
|
|
value["attempt_id"] = args.attempt_id
|
|
value.update(extra)
|
|
return value
|
|
|
|
|
|
def reset_mutation_state() -> None:
|
|
for key in MUTATION_STATE:
|
|
MUTATION_STATE[key] = False
|
|
|
|
|
|
def write_private_new(path: Path, value: bytes, mode: int) -> None:
|
|
descriptor = os.open(
|
|
path,
|
|
os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0),
|
|
mode,
|
|
)
|
|
try:
|
|
with os.fdopen(descriptor, "wb", closefd=False) as stream:
|
|
stream.write(value)
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
finally:
|
|
os.close(descriptor)
|
|
os.chmod(path, mode)
|
|
|
|
|
|
def validate_stage(stage: Path, manifest: list[dict[str, Any]]) -> None:
|
|
sudo_uid = os.environ.get("SUDO_UID", "")
|
|
if not sudo_uid.isdigit() or int(sudo_uid) <= 0:
|
|
raise TargetError("sudo_transport_identity_unavailable")
|
|
require_directory(stage, owner=int(sudo_uid), mode=0o700)
|
|
if {item.name for item in stage.iterdir()} != set(LABELS):
|
|
raise TargetError("stage_tree_invalid")
|
|
for row in manifest:
|
|
path = stage / row["name"]
|
|
metadata = path.lstat()
|
|
if (
|
|
stat.S_ISLNK(metadata.st_mode)
|
|
or not stat.S_ISREG(metadata.st_mode)
|
|
or metadata.st_size <= 0
|
|
or metadata.st_size > 2 * 1024 * 1024
|
|
or sha256_file(path) != row["sha256"]
|
|
):
|
|
raise TargetError("stage_file_invalid")
|
|
|
|
|
|
def validate_candidate(candidate: Path) -> None:
|
|
for name in LABELS:
|
|
if name.endswith(".py"):
|
|
source = (candidate / name).read_text(encoding="utf-8")
|
|
compile(source, str(candidate / name), "exec")
|
|
policy = json.loads((candidate / "metadata-export-policy.json").read_text("utf-8"))
|
|
if (
|
|
policy.get("work_item_id") != "P0-OBS-002"
|
|
or policy.get("source_runtime", {}).get("raw_database_access_allowed")
|
|
is not False
|
|
or policy.get("source_runtime", {}).get("raw_volume_access_allowed")
|
|
is not False
|
|
or policy.get("completion_contract", {}).get(
|
|
"full_sqlite_completion_claim_allowed"
|
|
)
|
|
is not False
|
|
):
|
|
raise TargetError("candidate_policy_contract_invalid")
|
|
|
|
|
|
def acquire_operation_lock() -> Any:
|
|
metadata = OPERATION_LOCK.lstat()
|
|
if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode):
|
|
raise TargetError("operation_lock_unsafe")
|
|
stream = OPERATION_LOCK.open("rb")
|
|
try:
|
|
fcntl.flock(stream.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except OSError as exc:
|
|
stream.close()
|
|
raise TargetError("operation_lock_busy") from exc
|
|
return stream
|
|
|
|
|
|
def prepare_stage(
|
|
args: argparse.Namespace, manifest: list[dict[str, Any]]
|
|
) -> dict[str, Any]:
|
|
lock = acquire_operation_lock()
|
|
try:
|
|
before = production_snapshot()
|
|
require_health()
|
|
require_no_active_operations()
|
|
state = target_state(args.bundle_id, manifest)
|
|
if state == "exact":
|
|
return base_receipt(args, "stage_not_required_exact", target_state=state)
|
|
stage = Path(args.stage_dir)
|
|
if os.path.lexists(stage):
|
|
raise TargetError("stage_identity_collision")
|
|
sudo_uid = os.environ.get("SUDO_UID", "")
|
|
sudo_gid = os.environ.get("SUDO_GID", "")
|
|
if (
|
|
not sudo_uid.isdigit()
|
|
or not sudo_gid.isdigit()
|
|
or int(sudo_uid) <= 0
|
|
or int(sudo_gid) <= 0
|
|
):
|
|
raise TargetError("sudo_transport_identity_unavailable")
|
|
MUTATION_STATE["source_write_attempted"] = True
|
|
stage.mkdir(mode=0o700)
|
|
os.chown(stage, int(sudo_uid), int(sudo_gid))
|
|
MUTATION_STATE["source_write_performed"] = True
|
|
try:
|
|
if production_snapshot() != before:
|
|
raise TargetError("production_identity_changed_during_prepare")
|
|
require_health()
|
|
except Exception:
|
|
shutil.rmtree(stage)
|
|
MUTATION_STATE["stage_cleanup_performed"] = True
|
|
raise
|
|
return base_receipt(args, "stage_prepared", target_state=state)
|
|
finally:
|
|
lock.close()
|
|
|
|
|
|
def read_private_json(path: Path, error_prefix: str) -> dict[str, Any]:
|
|
metadata = path.lstat()
|
|
if (
|
|
stat.S_ISLNK(metadata.st_mode)
|
|
or not stat.S_ISREG(metadata.st_mode)
|
|
or metadata.st_uid != ROOT_UID
|
|
or stat.S_IMODE(metadata.st_mode) != 0o600
|
|
or metadata.st_size <= 0
|
|
or metadata.st_size > 65536
|
|
):
|
|
raise TargetError(f"{error_prefix}_metadata_invalid")
|
|
try:
|
|
raw = path.read_bytes()
|
|
value = json.loads(raw)
|
|
except (OSError, UnicodeError, json.JSONDecodeError) as exc:
|
|
raise TargetError(f"{error_prefix}_invalid") from exc
|
|
if not isinstance(value, dict) or raw != canonical_bytes(value):
|
|
raise TargetError(f"{error_prefix}_not_canonical")
|
|
return value
|
|
|
|
|
|
def apply_intent(
|
|
args: argparse.Namespace, state: str, continuity: str
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"schema": "awoooi_agent99_signoz_toolchain_apply_intent_v1",
|
|
"trace_id": args.trace_id,
|
|
"run_id": args.run_id,
|
|
"work_item_id": args.work_item_id,
|
|
"source_revision": args.source_revision,
|
|
"bundle_id": args.bundle_id,
|
|
"target_state_before": state,
|
|
"production_continuity_sha256": continuity,
|
|
}
|
|
|
|
|
|
def load_apply_intent(args: argparse.Namespace) -> dict[str, Any] | None:
|
|
path = RECEIPT_ROOT / args.run_id / "apply-intent.json"
|
|
if not os.path.lexists(path):
|
|
return None
|
|
value = read_private_json(path, "apply_intent")
|
|
expected = {
|
|
"schema": "awoooi_agent99_signoz_toolchain_apply_intent_v1",
|
|
"trace_id": args.trace_id,
|
|
"run_id": args.run_id,
|
|
"work_item_id": args.work_item_id,
|
|
"source_revision": args.source_revision,
|
|
"bundle_id": args.bundle_id,
|
|
}
|
|
if (
|
|
set(value)
|
|
!= set(expected) | {"target_state_before", "production_continuity_sha256"}
|
|
or any(
|
|
value.get(key) != expected_value for key, expected_value in expected.items()
|
|
)
|
|
or value.get("target_state_before") not in {"absent", "exact"}
|
|
or not isinstance(value.get("production_continuity_sha256"), str)
|
|
or not SHA256.fullmatch(value["production_continuity_sha256"])
|
|
):
|
|
raise TargetError("apply_intent_identity_invalid")
|
|
return value
|
|
|
|
|
|
def deployment_receipt(
|
|
args: argparse.Namespace,
|
|
*,
|
|
continuity: str,
|
|
method: str,
|
|
target_mutated: bool,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"schema": "awoooi_agent99_signoz_toolchain_deployment_v1",
|
|
"trace_id": args.trace_id,
|
|
"run_id": args.run_id,
|
|
"work_item_id": args.work_item_id,
|
|
"source_revision": args.source_revision,
|
|
"bundle_id": args.bundle_id,
|
|
"target_state": "exact",
|
|
"deployment_method": method,
|
|
"finalized_by_mode": args.mode,
|
|
"reconcile_attempt_id": args.attempt_id if args.mode == "reconcile" else "",
|
|
"production_continuity_sha256": continuity,
|
|
"production_service_mutation_performed": False,
|
|
"content_addressed_target_mutation_performed": target_mutated,
|
|
"active_pointer_created": False,
|
|
"credential_value_read": False,
|
|
"stage_cleanup_verified": True,
|
|
"candidate_cleanup_verified": True,
|
|
"completion_claim": False,
|
|
}
|
|
|
|
|
|
def write_deployment_receipt(
|
|
args: argparse.Namespace,
|
|
*,
|
|
continuity: str,
|
|
method: str,
|
|
target_mutated: bool,
|
|
) -> dict[str, Any]:
|
|
value = deployment_receipt(
|
|
args,
|
|
continuity=continuity,
|
|
method=method,
|
|
target_mutated=target_mutated,
|
|
)
|
|
path = RECEIPT_ROOT / args.run_id / "deploy-receipt.json"
|
|
write_private_new(path, canonical_bytes(value), 0o600)
|
|
return value
|
|
|
|
|
|
def load_deploy_receipt(args: argparse.Namespace) -> dict[str, Any] | None:
|
|
path = RECEIPT_ROOT / args.run_id / "deploy-receipt.json"
|
|
if not os.path.lexists(path):
|
|
return None
|
|
value = read_private_json(path, "deploy_receipt")
|
|
expected = {
|
|
"schema": "awoooi_agent99_signoz_toolchain_deployment_v1",
|
|
"trace_id": args.trace_id,
|
|
"run_id": args.run_id,
|
|
"work_item_id": args.work_item_id,
|
|
"source_revision": args.source_revision,
|
|
"bundle_id": args.bundle_id,
|
|
"target_state": "exact",
|
|
"production_service_mutation_performed": False,
|
|
"active_pointer_created": False,
|
|
"credential_value_read": False,
|
|
"stage_cleanup_verified": True,
|
|
"candidate_cleanup_verified": True,
|
|
"completion_claim": False,
|
|
}
|
|
additional = {
|
|
"deployment_method",
|
|
"finalized_by_mode",
|
|
"reconcile_attempt_id",
|
|
"production_continuity_sha256",
|
|
"content_addressed_target_mutation_performed",
|
|
}
|
|
if (
|
|
set(value) != set(expected) | additional
|
|
or any(
|
|
value.get(key) != expected_value for key, expected_value in expected.items()
|
|
)
|
|
or value.get("deployment_method")
|
|
not in {"created", "adopted_existing_exact", "reconciled_exact"}
|
|
or value.get("finalized_by_mode") not in {"apply", "reconcile"}
|
|
or not isinstance(
|
|
value.get("content_addressed_target_mutation_performed"), bool
|
|
)
|
|
or not isinstance(value.get("production_continuity_sha256"), str)
|
|
or not SHA256.fullmatch(value["production_continuity_sha256"])
|
|
):
|
|
raise TargetError("deploy_receipt_identity_invalid")
|
|
attempt_id = value.get("reconcile_attempt_id")
|
|
if value["finalized_by_mode"] == "apply":
|
|
if attempt_id != "":
|
|
raise TargetError("deploy_receipt_attempt_identity_invalid")
|
|
elif not isinstance(attempt_id, str) or not SAFE_ID.fullmatch(attempt_id):
|
|
raise TargetError("deploy_receipt_attempt_identity_invalid")
|
|
return value
|
|
|
|
|
|
def quarantine_owned_directory(
|
|
source: Path,
|
|
destination: Path,
|
|
*,
|
|
owner: int,
|
|
modes: tuple[int, ...] = (0o700,),
|
|
) -> None:
|
|
if not os.path.lexists(source):
|
|
if os.path.lexists(destination):
|
|
require_directory(destination, owner=owner, mode=modes)
|
|
return
|
|
require_directory(source, owner=owner, mode=modes)
|
|
if os.path.lexists(destination):
|
|
raise TargetError("quarantine_identity_collision")
|
|
MUTATION_STATE["source_write_attempted"] = True
|
|
os.rename(source, destination)
|
|
MUTATION_STATE["source_write_performed"] = True
|
|
MUTATION_STATE["quarantine_performed"] = True
|
|
|
|
|
|
def apply_bundle(
|
|
args: argparse.Namespace, manifest: list[dict[str, Any]]
|
|
) -> dict[str, Any]:
|
|
stage = Path(args.stage_dir)
|
|
target = REMOTE_ROOT / args.bundle_id
|
|
candidate = REMOTE_ROOT / f".{args.bundle_id}.candidate.{args.run_id}"
|
|
receipt_dir = RECEIPT_ROOT / args.run_id
|
|
lock = acquire_operation_lock()
|
|
try:
|
|
before = production_snapshot()
|
|
continuity = sha256_bytes(canonical_bytes(before))
|
|
require_health()
|
|
require_no_active_operations()
|
|
state = target_state(args.bundle_id, manifest)
|
|
ensure_private_root(REMOTE_ROOT)
|
|
ensure_private_root(RECEIPT_ROOT)
|
|
if os.path.lexists(receipt_dir) or os.path.lexists(candidate):
|
|
raise TargetError("deployment_identity_collision")
|
|
MUTATION_STATE["source_write_attempted"] = True
|
|
receipt_dir.mkdir(mode=0o700)
|
|
MUTATION_STATE["source_write_performed"] = True
|
|
write_private_new(
|
|
receipt_dir / "apply-intent.json",
|
|
canonical_bytes(apply_intent(args, state, continuity)),
|
|
0o600,
|
|
)
|
|
try:
|
|
if state == "exact":
|
|
if os.path.lexists(stage):
|
|
raise TargetError("idempotent_apply_stage_must_be_absent")
|
|
after = production_snapshot()
|
|
require_health()
|
|
if after != before:
|
|
raise TargetError("production_identity_changed_during_apply")
|
|
durable = write_deployment_receipt(
|
|
args,
|
|
continuity=continuity,
|
|
method="adopted_existing_exact",
|
|
target_mutated=False,
|
|
)
|
|
return base_receipt(
|
|
args,
|
|
"apply_pass_idempotent_exact",
|
|
target_state="exact",
|
|
durable_deploy_receipt_valid=True,
|
|
durable_deploy_receipt_sha256=sha256_bytes(
|
|
canonical_bytes(durable)
|
|
),
|
|
production_continuity_sha256=continuity,
|
|
)
|
|
|
|
candidate.mkdir(mode=0o700)
|
|
validate_stage(stage, manifest)
|
|
for row in manifest:
|
|
source = stage / row["name"]
|
|
destination = candidate / row["name"]
|
|
write_private_new(destination, source.read_bytes(), int(row["mode"], 8))
|
|
validate_candidate(candidate)
|
|
# Publish the final directory mode atomically with the rename so a
|
|
# signal cannot expose an otherwise exact shared target as drift.
|
|
os.chmod(candidate, 0o750)
|
|
os.rename(candidate, target)
|
|
if target_state(args.bundle_id, manifest) != "exact":
|
|
raise TargetError("post_apply_target_not_exact")
|
|
after = production_snapshot()
|
|
require_health()
|
|
if after != before:
|
|
raise TargetError("production_identity_changed_during_apply")
|
|
manifest_rows = "".join(
|
|
f"{row['name']}\t{row['sha256']}\t{row['mode']}\n" for row in manifest
|
|
).encode("ascii")
|
|
write_private_new(
|
|
receipt_dir / "deployed-manifest.tsv", manifest_rows, 0o600
|
|
)
|
|
shutil.rmtree(stage)
|
|
if os.path.lexists(stage) or os.path.lexists(candidate):
|
|
raise TargetError("owned_live_residue_after_apply")
|
|
MUTATION_STATE["stage_cleanup_performed"] = True
|
|
durable = write_deployment_receipt(
|
|
args,
|
|
continuity=continuity,
|
|
method="created",
|
|
target_mutated=True,
|
|
)
|
|
return base_receipt(
|
|
args,
|
|
"apply_pass_deployed_inactive",
|
|
target_state="exact",
|
|
durable_deploy_receipt_valid=True,
|
|
durable_deploy_receipt_sha256=sha256_bytes(canonical_bytes(durable)),
|
|
production_continuity_sha256=continuity,
|
|
)
|
|
except Exception:
|
|
sudo_uid = os.environ.get("SUDO_UID", "")
|
|
if receipt_dir.is_dir() and sudo_uid.isdigit() and int(sudo_uid) > 0:
|
|
quarantine_owned_directory(
|
|
stage,
|
|
receipt_dir / "transport-stage",
|
|
owner=int(sudo_uid),
|
|
)
|
|
quarantine_owned_directory(
|
|
candidate,
|
|
receipt_dir / "quarantine-candidate",
|
|
owner=ROOT_UID,
|
|
modes=(0o700, 0o750),
|
|
)
|
|
# A content-addressed target is shared once visible. Never move or delete it.
|
|
raise
|
|
finally:
|
|
lock.close()
|
|
|
|
|
|
def reconcile_apply(
|
|
args: argparse.Namespace, manifest: list[dict[str, Any]]
|
|
) -> dict[str, Any]:
|
|
stage = Path(args.stage_dir)
|
|
target = REMOTE_ROOT / args.bundle_id
|
|
candidate = REMOTE_ROOT / f".{args.bundle_id}.candidate.{args.run_id}"
|
|
receipt_dir = RECEIPT_ROOT / args.run_id
|
|
lock = acquire_operation_lock()
|
|
try:
|
|
before = production_snapshot()
|
|
continuity = sha256_bytes(canonical_bytes(before))
|
|
require_health()
|
|
require_no_active_operations()
|
|
ensure_private_root(RECEIPT_ROOT)
|
|
if not os.path.lexists(receipt_dir):
|
|
MUTATION_STATE["source_write_attempted"] = True
|
|
receipt_dir.mkdir(mode=0o700)
|
|
MUTATION_STATE["source_write_performed"] = True
|
|
else:
|
|
require_directory(receipt_dir, owner=ROOT_UID, mode=0o700)
|
|
|
|
state_error = ""
|
|
try:
|
|
state = target_state(args.bundle_id, manifest)
|
|
except TargetError as exc:
|
|
state = "drift_or_unverified"
|
|
state_error = str(exc)
|
|
|
|
intent_error = ""
|
|
try:
|
|
intent = load_apply_intent(args)
|
|
except TargetError as exc:
|
|
intent = None
|
|
intent_error = str(exc)
|
|
|
|
deploy_error = ""
|
|
try:
|
|
durable = load_deploy_receipt(args)
|
|
except TargetError as exc:
|
|
durable = None
|
|
deploy_error = str(exc)
|
|
|
|
sudo_uid = os.environ.get("SUDO_UID", "")
|
|
if not sudo_uid.isdigit() or int(sudo_uid) <= 0:
|
|
raise TargetError("sudo_transport_identity_unavailable")
|
|
quarantine_owned_directory(
|
|
stage,
|
|
receipt_dir / "transport-stage",
|
|
owner=int(sudo_uid),
|
|
)
|
|
quarantine_owned_directory(
|
|
candidate,
|
|
receipt_dir / "quarantine-candidate",
|
|
owner=ROOT_UID,
|
|
modes=(0o700, 0o750),
|
|
)
|
|
if os.path.lexists(stage) or os.path.lexists(candidate):
|
|
raise TargetError("reconcile_owned_live_residue_remaining")
|
|
|
|
if deploy_error:
|
|
invalid = receipt_dir / "invalid-deploy-receipt.json"
|
|
current = receipt_dir / "deploy-receipt.json"
|
|
if os.path.lexists(current):
|
|
if os.path.lexists(invalid):
|
|
raise TargetError("invalid_deploy_receipt_quarantine_collision")
|
|
os.rename(current, invalid)
|
|
MUTATION_STATE["source_write_attempted"] = True
|
|
MUTATION_STATE["source_write_performed"] = True
|
|
MUTATION_STATE["quarantine_performed"] = True
|
|
|
|
if production_snapshot() != before:
|
|
raise TargetError("production_identity_changed_during_reconcile")
|
|
require_health()
|
|
if intent is not None and intent["production_continuity_sha256"] != continuity:
|
|
raise TargetError("apply_intent_production_continuity_mismatch")
|
|
|
|
durable_sha = ""
|
|
if state == "exact" and durable is not None:
|
|
if durable["production_continuity_sha256"] != continuity:
|
|
raise TargetError("deploy_receipt_production_continuity_mismatch")
|
|
terminal = "reconcile_exact_deployment_verified"
|
|
durable_sha = sha256_bytes(canonical_bytes(durable))
|
|
durable_valid = True
|
|
elif state == "exact" and intent is not None and not intent_error:
|
|
durable = write_deployment_receipt(
|
|
args,
|
|
continuity=continuity,
|
|
method="reconciled_exact",
|
|
target_mutated=intent["target_state_before"] == "absent",
|
|
)
|
|
terminal = "reconcile_exact_deployment_verified"
|
|
durable_sha = sha256_bytes(canonical_bytes(durable))
|
|
durable_valid = True
|
|
elif state == "exact":
|
|
terminal = "reconcile_exact_target_preserved_unverified"
|
|
durable_valid = False
|
|
elif state == "absent":
|
|
terminal = "reconcile_zero_owned_residue"
|
|
durable_valid = False
|
|
else:
|
|
terminal = "reconcile_shared_target_drift_preserved"
|
|
durable_valid = False
|
|
|
|
result = base_receipt(
|
|
args,
|
|
terminal,
|
|
target_state=state,
|
|
target_state_error=state_error,
|
|
apply_intent_error=intent_error,
|
|
deploy_receipt_error=deploy_error,
|
|
durable_deploy_receipt_valid=durable_valid,
|
|
durable_deploy_receipt_sha256=durable_sha,
|
|
production_continuity_sha256=continuity,
|
|
shared_target_preserved=os.path.lexists(target),
|
|
owned_live_residue_verified_absent=True,
|
|
)
|
|
reconcile_path = receipt_dir / f"reconcile-{args.attempt_id}.json"
|
|
write_private_new(reconcile_path, canonical_bytes(result), 0o600)
|
|
return result
|
|
finally:
|
|
lock.close()
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
reset_mutation_state()
|
|
old_handlers: dict[int, Any] = {}
|
|
|
|
def abort_on_signal(signum: int, _frame: Any) -> None:
|
|
for handled in (signal.SIGHUP, signal.SIGINT, signal.SIGTERM):
|
|
signal.signal(handled, signal.SIG_IGN)
|
|
raise TargetError(f"signal_{signum}_reconcile_required")
|
|
|
|
try:
|
|
manifest = validate_args(args)
|
|
if args.mode in {"prepare", "apply", "reconcile"}:
|
|
for signum in (signal.SIGHUP, signal.SIGINT, signal.SIGTERM):
|
|
old_handlers[signum] = signal.getsignal(signum)
|
|
signal.signal(signum, abort_on_signal)
|
|
if args.mode == "prepare":
|
|
result = prepare_stage(args, manifest)
|
|
elif args.mode == "apply":
|
|
result = apply_bundle(args, manifest)
|
|
elif args.mode == "reconcile":
|
|
result = reconcile_apply(args, manifest)
|
|
else:
|
|
before = production_snapshot()
|
|
require_health()
|
|
require_no_active_operations()
|
|
state = target_state(args.bundle_id, manifest)
|
|
deploy_receipt = None
|
|
if args.mode == "verify":
|
|
if state != "exact":
|
|
raise TargetError("verify_target_not_exact")
|
|
expected_stage = Path(
|
|
f"/tmp/awoooi-signoz-metadata-toolchain.{args.run_id}"
|
|
)
|
|
expected_candidate = REMOTE_ROOT / (
|
|
f".{args.bundle_id}.candidate.{args.run_id}"
|
|
)
|
|
if os.path.lexists(expected_stage) or os.path.lexists(
|
|
expected_candidate
|
|
):
|
|
raise TargetError("verify_owned_residue_present")
|
|
deploy_receipt = load_deploy_receipt(args)
|
|
if deploy_receipt is not None and deploy_receipt[
|
|
"production_continuity_sha256"
|
|
] != sha256_bytes(canonical_bytes(before)):
|
|
raise TargetError("deploy_receipt_production_continuity_mismatch")
|
|
if production_snapshot() != before:
|
|
raise TargetError("production_identity_changed_during_readback")
|
|
require_health()
|
|
result = base_receipt(
|
|
args,
|
|
"verify_pass" if args.mode == "verify" else "check_pass_no_write",
|
|
target_state=state,
|
|
durable_deploy_receipt_valid=(deploy_receipt is not None),
|
|
production_continuity_sha256=sha256_bytes(canonical_bytes(before)),
|
|
)
|
|
print(json.dumps(result, sort_keys=True, separators=(",", ":")))
|
|
return 0
|
|
except (TargetError, OSError, UnicodeError, json.JSONDecodeError) as exc:
|
|
failure = base_receipt(
|
|
args,
|
|
"failed",
|
|
error_code=str(exc),
|
|
)
|
|
if args.mode == "apply":
|
|
failure_path = RECEIPT_ROOT / args.run_id / "failure-receipt.json"
|
|
try:
|
|
if failure_path.parent.is_dir() and not os.path.lexists(failure_path):
|
|
write_private_new(failure_path, canonical_bytes(failure), 0o600)
|
|
except OSError:
|
|
pass
|
|
print(json.dumps(failure, sort_keys=True, separators=(",", ":")))
|
|
return 1
|
|
finally:
|
|
for signum, handler in old_handlers.items():
|
|
signal.signal(signum, handler)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|