tegwick
8e310b5c65
Some checks failed
Use-case regression gate / usecase-regressions (push) Has been cancelled
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3
243 lines
14 KiB
Python
243 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""Rehearse controller + exact helper wire on clones of retained recovered data.
|
|
|
|
Kubernetes, root ownership and fresh provider capture are explicit simulations.
|
|
The packaged Django/PostgreSQL/content operations and helper protocol are real.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import runpy
|
|
import subprocess
|
|
import tempfile
|
|
import uuid
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def sha(data):
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--restore-receipt", type=Path, required=True)
|
|
parser.add_argument("--plan", type=Path, required=True)
|
|
parser.add_argument("--candidate-image", required=True)
|
|
parser.add_argument("--output", type=Path, required=True)
|
|
parser.add_argument("--work-root", type=Path, default=Path("/tmp"),
|
|
help="Private parent for restored data and rehearsal copies")
|
|
args = parser.parse_args()
|
|
if not args.work_root.is_absolute() or not args.work_root.is_dir():
|
|
raise ValueError("Absolute existing work root required")
|
|
if args.output.exists() or args.output.is_symlink():
|
|
raise ValueError("Fresh output required")
|
|
previous = runpy.run_path(str(ROOT / "scripts/rehearse-source-date-journal.py"))
|
|
fixture = runpy.run_path(str(ROOT / "tests/test_source_date_supervisor.py"))
|
|
prepare = runpy.run_path(str(ROOT / "scripts/prepare-source-date-capsule.py"))["prepare"]
|
|
control, FakeNode = fixture["supervisor"], fixture["FakeNode"]
|
|
run, hashes = previous["PREVIOUS"]["run"], previous["PREVIOUS"]["hashes"]
|
|
raw_receipt, plan_bytes = args.restore_receipt.read_bytes(), args.plan.read_bytes()
|
|
state = json.loads(raw_receipt)
|
|
if (state["status"] != "restored_verified"
|
|
or not re.fullmatch(r"csoc-recovery-\d{14}", state["database_container"])
|
|
or state["network"] != state["database_container"] + "-net"):
|
|
raise ValueError("Retained isolated restore required")
|
|
net = json.loads(run(["docker", "network", "inspect", state["network"]]))[0]
|
|
db = json.loads(run(["docker", "inspect", state["database_container"]]))[0]
|
|
if (not net["Internal"] or set(net["Containers"]) != {db["Id"]}
|
|
or not db["State"]["Running"] or db["Image"] != state["postgres_image_id"]
|
|
or set(db["NetworkSettings"]["Networks"]) != {state["network"]}
|
|
or "recovery-db" not in db["NetworkSettings"]["Networks"][state["network"]]["Aliases"]
|
|
or any(db["NetworkSettings"]["Ports"].values())):
|
|
raise ValueError("Original recovery isolation changed")
|
|
image = run(["docker", "image", "inspect", args.candidate_image,
|
|
"--format", "{{.Id}}"]).decode().strip()
|
|
if not re.fullmatch(r"sha256:[a-f0-9]{64}", image):
|
|
raise ValueError("Retained image required")
|
|
pinned_image = "forgejo.coulomb.social/coulomb/coulomb-social@" + image
|
|
root = Path(state["restored_directory"]) / "content"
|
|
if any(p.is_symlink() for p in [root, *root.parents]) or not root.resolve().is_relative_to(
|
|
args.work_root.resolve()):
|
|
raise ValueError("Temporary recovery directory required")
|
|
originals = hashes(root)
|
|
wire = (ROOT / "scripts/source_date_wire.py").read_bytes()
|
|
modules = {name: (ROOT / "coulomb_social/apps/spaces" / name).read_bytes() for name in
|
|
("date_backfill.py", "date_backfill_journal.py", "date_backfill_operation.py")}
|
|
payload_modules = {"modules": {n: base64.b64encode(v).decode() for n, v in modules.items()},
|
|
"module_hashes": {n: sha(v) for n, v in modules.items()}}
|
|
outcomes = []
|
|
for scenario in ("normal", "lost_apply_reply"):
|
|
clone = "csoc_date_" + uuid.uuid4().hex
|
|
with tempfile.TemporaryDirectory(prefix="csoc-date-supervisor-rehearsal-",
|
|
dir=args.work_root) as temporary:
|
|
import shutil
|
|
|
|
work = Path(temporary)
|
|
shutil.copytree(root, work / "content")
|
|
(work / "plan.json").write_bytes(plan_bytes)
|
|
(work / "probe").mkdir(mode=0o700)
|
|
for name, data in modules.items():
|
|
(work / name).write_bytes(data)
|
|
|
|
def application(database, code, payload=None, operation=None):
|
|
name = "csoc-date-supervisor-" + uuid.uuid4().hex
|
|
command = ["docker", "run", "--rm", "-i", "--name", name,
|
|
"--network", state["network"], "--read-only",
|
|
"--tmpfs", "/tmp:size=33554432", "--cap-drop=ALL",
|
|
"--security-opt=no-new-privileges", "--user",
|
|
f"{os.getuid()}:{os.getgid()}",
|
|
"--entrypoint", "/app/.venv/bin/python",
|
|
"-e", "DJANGO_SETTINGS_MODULE=coulomb_social.settings.base",
|
|
"-e", f"DATABASE_URL=postgres://postgres@recovery-db/{database}",
|
|
"-e", "CONTENT_ROOT=/app/var/content", "-e", "FORGEJO_BASE_URL=",
|
|
"-e", "PYTHONDONTWRITEBYTECODE=1",
|
|
"-v", str(work / "content") + ":/app/var/content",
|
|
"-v", str(work / "content") + ":/rehearsal-content:ro",
|
|
"-v", str(work / "plan.json") + ":/review/plan.json:ro",
|
|
"-v", str(operation or work / "probe") + ":/operation"]
|
|
for member in modules:
|
|
command += ["-v", str(work / member)
|
|
+ ":/app/coulomb_social/apps/spaces/" + member + ":ro"]
|
|
command += [image, "-c", code]
|
|
try:
|
|
result = subprocess.run(command, input=payload, capture_output=True,
|
|
timeout=210)
|
|
finally:
|
|
subprocess.run(["docker", "rm", "-f", name], capture_output=True,
|
|
timeout=30, check=False)
|
|
if result.returncode:
|
|
fd, diagnostic = tempfile.mkstemp(prefix="csoc-supervisor-private-diagnostics-",
|
|
dir=args.work_root)
|
|
with os.fdopen(fd, "wb") as stream:
|
|
stream.write(result.stdout + result.stderr)
|
|
print(json.dumps({"private_diagnostics": diagnostic}))
|
|
raise ValueError("Private helper refused")
|
|
rows = [json.loads(line) for line in result.stdout.splitlines()
|
|
if line.startswith(b"{")]
|
|
if len(rows) != 1:
|
|
raise ValueError("Single helper result required")
|
|
return rows[0]
|
|
|
|
before = application("coulomb_social_restore", previous["SNAPSHOT"])
|
|
run(["docker", "exec", state["database_container"], "createdb", "-U", "postgres",
|
|
"-T", "coulomb_social_restore", clone])
|
|
try:
|
|
migration = ("import django; django.setup(); "
|
|
"from django.db import connection; "
|
|
"from django.db.migrations.executor import MigrationExecutor; "
|
|
"MigrationExecutor(connection).migrate(["
|
|
"('spaces','0003_space_slug_global_uniq')]);\n")
|
|
baseline = application(clone, migration + wire.decode(), control.canonical({
|
|
**payload_modules, "mode": "probe", "database_name": clone}))["fingerprint"]
|
|
|
|
class NativeHelper(FakeNode):
|
|
def command(self, arguments, payload=None, **options):
|
|
if arguments[:2] != ["kubectl", "exec"]:
|
|
return super().command(arguments, payload, **options)
|
|
message = json.loads(payload)
|
|
# Translate the synthetic Kubernetes transport to the
|
|
# actual isolated clone; live Capsule.payload always pins
|
|
# coulomb_social_db. No production endpoint is accepted.
|
|
message["database_name"] = clone
|
|
private = self.supervisor.directory / "private"
|
|
result = application(clone, arguments[-1], control.canonical(message),
|
|
private if private.exists() else None)
|
|
if message["mode"] == "apply":
|
|
self.apply_count += 1
|
|
if scenario == "lost_apply_reply":
|
|
self.fail = "apply_reply"
|
|
self.trip("apply_reply")
|
|
return control.canonical(result)
|
|
|
|
def capture(self, target):
|
|
self.capture_count += 1
|
|
now = datetime.now(UTC).isoformat()
|
|
key = ("platform-pg/application-archives/coulomb-social/"
|
|
"synthetic-control.tar.age")
|
|
return {"schema": "platform.application-primary-archive.v1",
|
|
"application": "coulomb-social",
|
|
"status": "offsite_fetched_verified",
|
|
"captured_at": now, "completed_at": now, "image": pinned_image,
|
|
"age_recipient": "age1synthetic", "immutable_create": True,
|
|
"remote_retrieval_verified": True,
|
|
"version_id": "synthetic-version",
|
|
"object_key": key, "destination": "s3://railiance-platform-pg-backup/"
|
|
+ key, "encrypted_sha256": "a" * 64, "ciphertext_sha256": "a" * 64,
|
|
"database_sha256": "b" * 64, "content_sha256": "c" * 64,
|
|
"encrypted_bytes": 1, "ciphertext_bytes": 1,
|
|
"capture_and_provider_simulated": True}
|
|
|
|
node = NativeHelper()
|
|
node.dep["spec"]["template"]["spec"]["containers"][0]["image"] = pinned_image
|
|
target = {"cluster_uid": "cluster-id", "deployment_uid": "deployment-id",
|
|
"node_uid": "node-id", "node_name": "synthetic-node", "pvc_uid": "pvc-id",
|
|
"image": pinned_image, "deployment_spec_sha256": sha(control.canonical(
|
|
node.dep["spec"])), "helm_revision": 18,
|
|
"helm_manifest_sha256": sha(b"synthetic helm manifest"),
|
|
"public_recipient": "age1synthetic", "installed_helpers": {
|
|
n: "f" * 64 for n in control.HELPERS}}
|
|
(work / "baseline.json").write_bytes(control.canonical(baseline))
|
|
(work / "target.json").write_bytes(control.canonical(target))
|
|
prepared = prepare(work / "plan.json", work / "baseline.json", work / "target.json",
|
|
work / "capsule")
|
|
capsule = control.Capsule(work / "capsule", prepared["proposal_digest"])
|
|
operation = control.Supervisor(capsule, node, state_root=work / "state")
|
|
node.supervisor = operation
|
|
try:
|
|
result = operation.execute()
|
|
except control.Refused:
|
|
if scenario != "lost_apply_reply":
|
|
raise
|
|
assert node.dep["spec"]["replicas"] == 0
|
|
result = operation.recover()
|
|
assert result["phase"] == "completed" and result["correction_status"] == "completed"
|
|
assert node.capture_count == node.apply_count == 1
|
|
assert node.dep["spec"]["replicas"] == 1 and not node.helper
|
|
after = application("coulomb_social_restore", previous["SNAPSHOT"])
|
|
if before["tables_sha256"] != after["tables_sha256"] or hashes(root) != originals:
|
|
raise ValueError("Original recovery source changed")
|
|
outcomes.append({"scenario": scenario, "status": "settled",
|
|
"actual_helper_wire_and_data_operation": True,
|
|
"capture_calls": 1, "apply_calls": 1,
|
|
"all_tables_compared": baseline["tables"],
|
|
"all_files_compared": baseline["files"]})
|
|
finally:
|
|
run(["docker", "exec", state["database_container"], "dropdb", "-U", "postgres",
|
|
"--force", clone])
|
|
proof = {"schema": "coulomb.supervised-date-isolated-rehearsal.v1",
|
|
"status": "controller_and_real_helper_rehearsed", "scenarios": outcomes,
|
|
"kubernetes_and_root_operations_simulated": True,
|
|
"capture_and_provider_simulated": True, "production_access": False,
|
|
"native_observer_not_executed": True,
|
|
"production_writes": False, "live_fence_or_native_backup_proved": False,
|
|
"source_overlay_used": True, "candidate_image_id": image,
|
|
"retained_recovery_source_unchanged": True, "disposable_sql_clones_removed": True,
|
|
"sources_sha256": {p: sha((ROOT / p).read_bytes()) for p in (
|
|
"scripts/source_date_supervisor.py", "scripts/source_date_wire.py",
|
|
"scripts/prepare-source-date-capsule.py",
|
|
"scripts/observe-source-date-target.py",
|
|
"scripts/rehearse-source-date-supervisor.py",
|
|
"tests/test_source_date_supervisor.py")},
|
|
"modules_sha256": {n: sha(v) for n, v in modules.items()},
|
|
"private_plan_sha256": sha(plan_bytes), "restore_receipt_sha256": sha(raw_receipt)}
|
|
fd = os.open(args.output, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
|
with os.fdopen(fd, "w") as stream:
|
|
json.dump(proof, stream, indent=2)
|
|
stream.write("\n")
|
|
print(json.dumps(proof))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except (ValueError, OSError, KeyError, subprocess.SubprocessError):
|
|
raise SystemExit("Supervisor rehearsal refused; inspect private diagnostics locally.")
|