diff --git a/coulomb_social/apps/members/client_export.py b/coulomb_social/apps/members/client_export.py index 90a8d5e..359639e 100644 --- a/coulomb_social/apps/members/client_export.py +++ b/coulomb_social/apps/members/client_export.py @@ -88,6 +88,12 @@ def schema_inventory(): "django_admin_log": "Export only events attributable to selected client objects.", "process_logs": "Operator retention applies; outside application data export.", "backup_copies": "Independent retention/expiry requires operator disposition.", + "date_correction_journals": ( + "Operator-private date-operations journals retain original index bytes, plan and " + "backup metadata outside CONTENT_ROOT. Include retained client bytes in operator " + "offboarding handoff and explicit expiry; the app export cannot read " + "that root store." + ), }, "operational_schema": [ {"model": model._meta.label_lower, "table": model._meta.db_table, diff --git a/docs/release/client-data-export-review-2026-10-30.md b/docs/release/client-data-export-review-2026-10-30.md index 06309c8..45b8f5c 100644 --- a/docs/release/client-data-export-review-2026-10-30.md +++ b/docs/release/client-data-export-review-2026-10-30.md @@ -20,6 +20,12 @@ than copying or silently discarding it. Independent backups and process logs sta under operator retention. Exporting data does not delete any of these stores or revoke sessions. +The operational inventory also names the prepared supervisor's private +`date-operations` journals outside CONTENT_ROOT. They can retain original private +index bytes, plan and backup metadata. The app cannot read that root store; +operator offboarding must account for any retained client bytes and their explicit +expiry separately. The supervisor rehearsal creates no production journal. + The inventory derives field names, database columns, types, nullability and relations from the loaded Django models. Every application field has an explicit classification. A new model or unclassified field refuses export until reviewed. diff --git a/docs/release/source-date-journal-review-2026-10-30.md b/docs/release/source-date-journal-review-2026-10-30.md index f8f48aa..36af1a2 100644 --- a/docs/release/source-date-journal-review-2026-10-30.md +++ b/docs/release/source-date-journal-review-2026-10-30.md @@ -110,10 +110,10 @@ The candidate image uses explicit source overlays; no image is released here. The focused planner/journal/operation suite passes 50 cases; `make check` passes 364 cases including protected browsers, with zero skips. -Remaining wrapper requirements: bind exact reviewed source/plan/target and fresh -backup/readback hashes; verify the external writer fence and full live SQL/content -baseline; invoke this operation in the pinned helper; supervise the privileged -action; refuse outstanding journal state; -retain the journal on durable private storage; perform full unrelated-row/file -postflight and resolve every pending state before resuming writers. Rehearse that -wrapper against the independently recovered real pair before productive use. +The external supervisor source and recovered-pair wrapper rehearsal are now +described in source-date-supervisor-review-2026-10-30.md. They add exact capsule +binding, native capture/source bookends, pinned helper/fence checks, private durable +journals and explicit recovery/resume handling. The rehearsal's control plane and +provider remain simulated. Actual storage release, native observation/admission, +fresh verified backup, external fence and productive correction/postflight remain +required. Source tests and isolated rehearsal do not establish those live gates. diff --git a/docs/release/source-date-supervisor-rehearsal-2026-10-30.json b/docs/release/source-date-supervisor-rehearsal-2026-10-30.json new file mode 100644 index 0000000..586f62f --- /dev/null +++ b/docs/release/source-date-supervisor-rehearsal-2026-10-30.json @@ -0,0 +1,58 @@ +{ + "schema": "coulomb.supervised-date-isolated-rehearsal.v1", + "status": "controller_and_real_helper_rehearsed", + "scenarios": [ + { + "scenario": "normal", + "status": "settled", + "actual_helper_wire_and_data_operation": true, + "capture_calls": 1, + "apply_calls": 1, + "all_tables_compared": 13, + "all_files_compared": 82 + }, + { + "scenario": "lost_apply_reply", + "status": "settled", + "actual_helper_wire_and_data_operation": true, + "capture_calls": 1, + "apply_calls": 1, + "all_tables_compared": 13, + "all_files_compared": 82 + } + ], + "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": "sha256:512819463a1902b43a615beee14be7bf97e6a05e016385082b9380692ad1a8cd", + "retained_recovery_source_unchanged": true, + "disposable_sql_clones_removed": true, + "sources_sha256": { + "scripts/source_date_supervisor.py": "a212bac3307611d27791eb857323b6344ab138a31a03120f908bda0ef1bf30c7", + "scripts/source_date_wire.py": "044480950df86194a91136284de734801a463ebbdbbfa2cb1b0b6ffd1e91bc82", + "scripts/prepare-source-date-capsule.py": "1d1b788f7e1abeb076832bf57779245cb13c16d581048fbb5b7b39957ff5a8e8", + "scripts/observe-source-date-target.py": "0e55360f092e2975cb547413dbc5b414088f3a6e56882005c422013bb13796a4", + "scripts/rehearse-source-date-supervisor.py": "51eff1511b7c961a0761723abcbd2076a648f564c7d4f06b35fa38fac0763d62", + "tests/test_source_date_supervisor.py": "ebe06e24e91b0e7c9b275442fc77f79cfcb810eb00a51d8b5b7622afaf7966a7" + }, + "modules_sha256": { + "date_backfill.py": "84ce5d3369389e1af9ab999aad3804456fd2f66e654e290496be6930cd4885e4", + "date_backfill_journal.py": "47af997c02b3910058e5dc8c593dd8135a68be94e909c672a946f54601eb621e", + "date_backfill_operation.py": "50f962f6c42fd47dbbbdea3f06d38ccfef063e784bdb6361b8a2325cbe825e18" + }, + "private_plan_sha256": "44369d5abbf92d3e6d52e909cdc1e1c3a90b8817be7027c3a43846fbfc381ef8", + "restore_receipt_sha256": "5f745cdae0a34c4a98f0e4429ceea49475b4e5cd9d2f8f09dea6be000e200258", + "acceptance_gate": { + "command": "make check", + "passed": 383, + "failures": 0, + "errors": 0, + "skipped": 0, + "required_browser_journeys_included": true, + "junit_sha256": "89b349cec61070d1c51faf20056c9e55ff4f9c274d30298fe3fac4f9d8e008dd" + } +} diff --git a/docs/release/source-date-supervisor-review-2026-10-30.md b/docs/release/source-date-supervisor-review-2026-10-30.md new file mode 100644 index 0000000..5c457eb --- /dev/null +++ b/docs/release/source-date-supervisor-review-2026-10-30.md @@ -0,0 +1,127 @@ +# Attended creation-date supervisor + +Prepared source, not a productive correction or an admitted production capsule. +The separate storage release proposal `3389c506` is unchanged and still requires +its founder-terminal execution. Observe and seal the date target only after that +release: migration 0003 changes the complete-store baseline. + +## Bound operation + +`scripts/observe-source-date-target.py` performs read-only native observation on +the content node. It requires the expected cluster/deployment/PVC UIDs and exact +image digest, verifies the local machine against the Kubernetes node, reads only +the installed helper hashes and public recipient, and probes SQL/content metadata. +It checks the stable Helm release, sole Ready application writer, global slug +constraint and a server-dry-run helper. It publishes a new private capsule under +`/tmp`; it does not capture, scale, create a live helper or change dates. + +`scripts/prepare-source-date-capsule.py` also supports offline preparation from +private plan, target and baseline files. Neither preparation path grants approval. +Each attempt has a new random identifier. Its digest binds the manifest, exact +target, complete-store baseline, plan, supervisor, helper protocol and all three +date modules. Changing any reviewed byte requires another digest. + +`scripts/source_date_supervisor.py` defaults to read-only preflight. Execution +requires node root, a terminal and the founder's exact `APPLY-DATE ` +confirmation. It rechecks the deployment, actual running image, physical node, +PVC, Helm release, absent autoscaling/GitOps adoption, installed backup hashes and +inactive backup timers. It invokes the existing admitted paired capture worker, +checks immutable primary readback and local ciphertext metadata/hash, and compares +the complete SQL/content baseline before and after capture. + +The supervisor then holds backup/snapshot locks, scales the pinned deployment to +zero with resource-version/current-scale preconditions, waits for application +pods to disappear and refuses unknown PVC users. A single pinned helper mounts +the existing content PVC and an operator-private journal directory on that same +node. It uses application Secret references without retrieving their values, +no service-account token, a read-only root filesystem, dropped capabilities and +bounded memory/CPU/tmp storage. Server admission must preserve the reviewed +process, environment, mounts and security fields; the actual helper UID/spec are +checked again before use. + +The helper rechecks the complete baseline, takes the existing content lock and +PostgreSQL table locks, and invokes the reviewed durable date operation. Only the +selected SQL creation date and index date insertion may change. All tables and +content hashes must match the derived result before the exact helper is deleted +with a UID precondition and the application is resumed. This is an attended +sole-writer maintenance operation; native observation and execution must establish +the real writer boundary. Synthetic control-plane fixtures do not establish it. + +## Durable state and recovery + +The root receipt and frozen capsule live under +`/var/lib/coulomb-social-backup/date-operations//`. +The helper's `private/` directory is owned by application UID 10001 with mode 0700; +the root ancestor is private and the application does not mount this store. +The journal contains original private index bytes, the plan and original backup +binding. Include it in operator client-offboarding handoff and explicit expiry. +The app's client export cannot read that root store. No expiry is performed here. + +Uncertain outcomes retain the stopped application, helper and durable state. +Never replay `--execute` or replace the original backup with a new capture. +Explicit founder-attended `--recover` uses the frozen capsule and original digest: + +| Observed state | Recovery | +| --- | --- | +| Original helper operation still holds its kernel lock | Refuse; keep writers fenced. | +| Pending journal | Reconcile against committed SQL using the original backup binding. | +| Completed, rolled-back or absent journal | Require its exact expected complete-store fingerprint. | +| Lost helper-create/delete reply | Inspect the pinned name, UID and spec; recreate only the same helper when absent. | +| Original helper has terminated | Delete the matching terminal UID and recreate only for inspection/reconciliation. | +| Replaced helper, unknown writer or unrelated data drift | Refuse for operator review. | +| Lost scale-up reply, exact application already running | Settle rollout/receipt without restoring historical data or overwriting later writes. | + +Remote helper work is time-bounded. An interrupted process releases SQL/kernel +locks; its durable journal determines reconciliation. Normal pod deletion is +followed by disappearance checks. No forced deletion or generic data restore is +used. A receipt phase of `completed` means the attempt is settled; correction is +proved only by `correction_status: completed` and native postflight. An absent or +rolled-back correction does not finish WP-0014-T04. + +## Operator sequence after the storage release + +Run the observer from the reviewed source checkout on the content node, supplying +the private accepted-cohort plan and a fresh output directory: + +```sh +sudo python3 scripts/observe-source-date-target.py \ + --plan /private/date-backfill-plan.json \ + --output /tmp/csoc-date-capsule-NEW \ + --image forgejo.coulomb.social/coulomb/coulomb-social@sha256:512819463a1902b43a615beee14be7bf97e6a05e016385082b9380692ad1a8cd \ + --cluster-uid a553c742-0115-43d4-99a4-a5ca56fe0786 \ + --deployment-uid 2dbe661a-d232-4477-bbd1-833813077fbe \ + --pvc-uid 9888abeb-9d52-4c70-a23d-b5494780d9c3 +``` + +Review that newly observed capsule and exact proposal before founder execution: + +```sh +sudo python3 /tmp/csoc-date-capsule-NEW/supervisor.py \ + --capsule /tmp/csoc-date-capsule-NEW \ + --expect-digest sha256:REVIEWED_DATE_PROPOSAL --execute +``` + +These placeholders are deliberately not an executable admitted proposal. For an +uncertain attempt, run its frozen root `capsule/supervisor.py` with that capsule, +the same digest and `--recover`, then inspect the durable receipt and native +postflight. Do not type the founder confirmation on the founder's behalf. +The deployment-owner rule is recorded in the storage release review: “The agent +cannot approve itself.” No timer, mail, grant, production restore, bulk import, +apex change or Bubble retirement is authorized by this preparation. + +## Evidence + +`tests/test_source_date_supervisor.py` covers 19 focused lifecycle/refusal cases. +`scripts/rehearse-source-date-supervisor.py` uses the real prepared capsule, +supervisor, helper protocol and operation against a disposable clone of the +independently recovered PostgreSQL/content pair. It exercises normal completion +and a lost reply after the actual SQL/file commit, then explicit reconciliation +without another capture or apply. All 13 tables and 82 files are compared; the +retained original is unchanged and the clone is removed. + +The companion `source-date-supervisor-rehearsal-2026-10-30.json` binds reviewed +source hashes and records the boundaries: Kubernetes/root fencing and provider +capture are simulated, and the native observer was not executed. No new live +backup, production capsule, runtime admission or productive correction is claimed. +The full `make check` gate passes 383 cases, including mandatory browser journeys, +with zero failures, errors or skips; Ruff and whitespace checks pass. diff --git a/scripts/observe-source-date-target.py b/scripts/observe-source-date-target.py new file mode 100644 index 0000000..c4537dc --- /dev/null +++ b/scripts/observe-source-date-target.py @@ -0,0 +1,122 @@ +#!/usr/bin/env python3 +"""Read-only native observation and private capsule preparation, after storage release. + +Run on the content node. No backup, scale, pod creation, schema/date update, +credential delivery or timer activation is performed. Server dry-run is used. +""" + +import argparse +import base64 +import json +import os +import runpy +import tempfile +from pathlib import Path + +from source_date_supervisor import ( + CLAIM, + NS, + Capsule, + Node, + Refused, + Supervisor, + canonical, + sha, + sync_directory, +) + +ROOT = Path(__file__).resolve().parents[1] + + +def observe(plan, output, *, image, cluster_uid, deployment_uid, pvc_uid, node=None): + if os.geteuid() != 0: + raise Refused("native_node_root_read_context_required") + output = output.absolute() + if (output.exists() or any(p.is_symlink() for p in [output, *output.parents]) + or output.resolve().is_relative_to(ROOT) + or not output.resolve().is_relative_to(Path("/tmp"))): + raise Refused("fresh_private_tmp_output_outside_checkout_required") + node = node or Node() + dep = node.doc("get", "deployment", NS, "-n", NS) + spec = dep["spec"] + if (dep["metadata"]["uid"] != deployment_uid or spec["replicas"] != 1 + or len(spec["template"]["spec"]["containers"]) != 1 + or spec["template"]["spec"]["containers"][0]["image"] != image): + raise Refused("storage_release_or_target_prerequisite_missing") + if node.doc("get", "namespace", "kube-system")["metadata"]["uid"] != cluster_uid: + raise Refused("cluster_changed") + pvc = node.doc("get", "pvc", CLAIM, "-n", NS) + if pvc["metadata"]["uid"] != pvc_uid or pvc["status"]["phase"] != "Bound": + raise Refused("pvc_changed") + pods = [p for p in node.doc("get", "pods", "-n", NS)["items"] + if p.get("status", {}).get("phase") not in {"Succeeded", "Failed"} + and any(v.get("persistentVolumeClaim", {}).get("claimName") == CLAIM + for v in p["spec"].get("volumes", []))] + if len(pods) != 1: + raise Refused("single_application_pod_required") + host = node.doc("get", "node", pods[0]["spec"]["nodeName"]) + node.verify_local_node(host) + pins = node.installation_pins() + history = json.loads(node.command(["helm", "history", NS, "-n", NS, "-o", "json"])) + if not history or history[-1]["status"] != "deployed": + raise Refused("stable_helm_release_required") + target = {"cluster_uid": cluster_uid, "deployment_uid": deployment_uid, "pvc_uid": pvc_uid, + "node_name": host["metadata"]["name"], "node_uid": host["metadata"]["uid"], + "image": image, "deployment_spec_sha256": sha(canonical(spec)), + "helm_revision": history[-1]["revision"], + "helm_manifest_sha256": sha(node.command(["helm", "get", "manifest", NS, "-n", NS])), + **pins} + 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 = {"mode": "probe", "database_name": "coulomb_social_db", + "modules": {n: base64.b64encode(v).decode() for n, v in modules.items()}, + "module_hashes": {n: sha(v) for n, v in modules.items()}} + baseline = json.loads(node.command( + ["kubectl", "exec", "-i", "-n", NS, pods[0]["metadata"]["name"], "--", "env", + "PYTHONDONTWRITEBYTECODE=1", "python", "-c", + (ROOT / "scripts/source_date_wire.py").read_text()], canonical(payload)))["fingerprint"] + prepare = runpy.run_path(str(ROOT / "scripts/prepare-source-date-capsule.py"))["prepare"] + with tempfile.TemporaryDirectory(prefix="csoc-native-date-observation-") as temporary: + work = Path(temporary) + (work / "baseline.json").write_bytes(canonical(baseline)) + (work / "target.json").write_bytes(canonical(target)) + prepared = prepare(plan, work / "baseline.json", work / "target.json", work / "capsule") + capsule = Capsule(work / "capsule", prepared["proposal_digest"]) + check = Supervisor(capsule, node) + _, proof = check.preflight() + output.mkdir(mode=0o700) + # Publish exactly the already checked capsule, retaining its attempt id. + for name in [*capsule.content, "manifest.json"]: + data = (work / "capsule" / name).read_bytes() + fd = os.open(output / name, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, "wb") as stream: + stream.write(data) + stream.flush() + os.fsync(stream.fileno()) + sync_directory(output) + sync_directory(output.parent) + return {**prepared, "status": "native_read_only_capsule_prepared_not_admitted", + "baseline_rechecked": proof["fingerprint"] == baseline, + "helper_server_dry_run_sha256": check.preview_helper_spec, + "capture_performed": False, "live_changes": False} + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--plan", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--image", required=True) + parser.add_argument("--cluster-uid", required=True) + parser.add_argument("--deployment-uid", required=True) + parser.add_argument("--pvc-uid", required=True) + args = parser.parse_args() + print(json.dumps(observe(args.plan, args.output, image=args.image, cluster_uid=args.cluster_uid, + deployment_uid=args.deployment_uid, pvc_uid=args.pvc_uid))) + + +if __name__ == "__main__": + try: + main() + except Exception as exc: + print(json.dumps({"status": "observation_refused", "failure_type": type(exc).__name__})) + raise SystemExit(1) diff --git a/scripts/prepare-source-date-capsule.py b/scripts/prepare-source-date-capsule.py new file mode 100644 index 0000000..956d751 --- /dev/null +++ b/scripts/prepare-source-date-capsule.py @@ -0,0 +1,79 @@ +#!/usr/bin/env python3 +"""Seal existing private date plan, current target pins and complete-store baseline. + +Offline preparation only. It creates no approval and never contacts the cluster. +Acquire the target/baseline AFTER the separately supervised storage release. +""" + +import argparse +import json +import os +import uuid +from pathlib import Path + +from source_date_supervisor import Capsule, canonical, sha, sync_directory + +ROOT = Path(__file__).resolve().parents[1] + + +def prepare(plan, baseline, target, output): + output = output.absolute() + if (output.exists() or any(p.is_symlink() for p in [output, *output.parents]) + or output.resolve().is_relative_to(ROOT)): + raise ValueError("Fresh private destination outside checkout required") + sources = { + "supervisor.py": ROOT / "scripts/source_date_supervisor.py", + "wire.py": ROOT / "scripts/source_date_wire.py", + **{name: ROOT / "coulomb_social/apps/spaces" / name for name in + ("date_backfill.py", "date_backfill_journal.py", "date_backfill_operation.py")}, + "plan.json": plan, + } + if any(p.is_symlink() or not p.is_file() or p.resolve().is_relative_to(output) + for p in [*sources.values(), baseline, target]): + raise ValueError("Regular existing inputs required") + content = {name: path.read_bytes() for name, path in sources.items()} + manifest = {"schema": "coulomb.supervised-date-capsule.v1", + "attempt_id": uuid.uuid4().hex, + "target": json.loads(target.read_bytes()), + "baseline": json.loads(baseline.read_bytes()), + "files": {name: sha(data) for name, data in content.items()}, + "requires_founder_attended_execution": True, + "approval_status": "prepared_not_admitted"} + digest = "sha256:" + sha(canonical(manifest)) + # A private temporary validation directory avoids publishing an invalid or + # partial capsule under the requested final name. + import tempfile + + with tempfile.TemporaryDirectory(prefix="csoc-date-capsule-") as temporary: + stage = Path(temporary) + for name, data in {**content, "manifest.json": canonical(manifest)}.items(): + (stage / name).write_bytes(data) + Capsule(stage, digest) + output.mkdir(mode=0o700) + for name, data in {**content, "manifest.json": canonical(manifest)}.items(): + fd = os.open(output / name, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, "wb") as stream: + stream.write(data) + stream.flush() + os.fsync(stream.fileno()) + sync_directory(output) + sync_directory(output.parent) + return {"status": "prepared_not_admitted", "proposal_digest": digest, + "file_count": len(content) + 1, "production_access": False, "live_changes": False} + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--plan", type=Path, required=True) + parser.add_argument("--baseline", type=Path, required=True) + parser.add_argument("--target", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args() + print(json.dumps(prepare(args.plan, args.baseline, args.target, args.output))) + + +if __name__ == "__main__": + try: + main() + except Exception as exc: + raise SystemExit("Date capsule preparation refused: " + type(exc).__name__) diff --git a/scripts/rehearse-source-date-supervisor.py b/scripts/rehearse-source-date-supervisor.py new file mode 100644 index 0000000..a36cc75 --- /dev/null +++ b/scripts/rehearse-source-date-supervisor.py @@ -0,0 +1,237 @@ +#!/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) + args = parser.parse_args() + 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( + Path("/tmp")): + 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-") 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-") + 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.") diff --git a/scripts/source_date_supervisor.py b/scripts/source_date_supervisor.py new file mode 100644 index 0000000..e59954f --- /dev/null +++ b/scripts/source_date_supervisor.py @@ -0,0 +1,662 @@ +"""Node supervisor for an exact founder-reviewed date-correction capsule. + +Default is read-only. This tool is not an approval service: only the founder may +run execute/recover in their terminal under the deployment owner's supervision. +Uncertain attempts retain their helper, journals and stopped application. +""" + +from __future__ import annotations + +import argparse +import base64 +import copy +import fcntl +import hashlib +import json +import os +import re +import stat +import subprocess +import sys +import tempfile +from contextlib import ExitStack, contextmanager +from datetime import UTC, datetime +from pathlib import Path + +NS = "coulomb-social" +CLAIM = "coulomb-social-content" +STATE_ROOT = Path("/var/lib/coulomb-social-backup/date-operations") +MEMBERS = {"supervisor.py", "wire.py", "date_backfill.py", "date_backfill_journal.py", + "date_backfill_operation.py", "plan.json"} +HELPERS = {"/opt/coulomb-social-backup/paired-backup.py", + "/opt/coulomb-social-backup/snapshot-content-and-db.py", + "/opt/railiance-platform/scripts/primary_application_archive.py"} + + +def canonical(value): + return json.dumps(value, sort_keys=True, separators=(",", ":"), default=str).encode() + + +def sha(data): + return hashlib.sha256(data).hexdigest() + + +class Refused(RuntimeError): + pass + + +def private_file(path): + info = path.lstat() + if (not stat.S_ISREG(info.st_mode) or info.st_uid != 0 or info.st_mode & 0o077 + or info.st_nlink != 1 or any(p.is_symlink() for p in path.parents)): + raise Refused("private_root_file_required") + return path.read_bytes() + + +def sync_directory(path): + descriptor = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + try: + os.fsync(descriptor) + finally: + os.close(descriptor) + + +def save(path, value): + fd, temporary = tempfile.mkstemp(prefix=".state-", dir=path.parent) + try: + with os.fdopen(fd, "wb") as stream: + stream.write(canonical(value)) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, path) + directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(directory) + finally: + os.close(directory) + finally: + if os.path.exists(temporary): + os.unlink(temporary) + + +class Capsule: + def __init__(self, path, expected): + self.manifest = json.loads((path / "manifest.json").read_bytes()) + if "sha256:" + sha(canonical(self.manifest)) != expected: + raise Refused("proposal_digest_changed") + self.digest = expected + if (self.manifest.get("schema") != "coulomb.supervised-date-capsule.v1" + or set(self.manifest["files"]) != MEMBERS + or self.manifest.get("requires_founder_attended_execution") is not True + or not re.fullmatch(r"[a-f0-9]{32}", self.manifest.get("attempt_id", ""))): + raise Refused("exact_capsule_members_required") + self.content = {} + for name, digest in self.manifest["files"].items(): + member = path / name + if not member.is_file() or any(p.is_symlink() for p in [member, *member.parents]): + raise Refused("capsule_links_refused") + data = member.read_bytes() + if sha(data) != digest: + raise Refused("capsule_member_changed") + self.content[name] = data + self.target = self.manifest["target"] + fields = {"cluster_uid", "deployment_uid", "pvc_uid", "node_name", "node_uid", "image", + "deployment_spec_sha256", "helm_revision", "helm_manifest_sha256", + "public_recipient", "installed_helpers"} + if (set(self.target) != fields or set(self.target["installed_helpers"]) != HELPERS + or not re.fullmatch(r"[a-z0-9.-]+", self.target["node_name"]) + or not re.fullmatch(r"forgejo\.coulomb\.social/coulomb/" + r"coulomb-social@sha256:[a-f0-9]{64}", self.target["image"])): + raise Refused("exact_application_target_required") + hashes = [self.target[n] for n in ("deployment_spec_sha256", "helm_manifest_sha256")] + hashes += list(self.target["installed_helpers"].values()) + if (any(not re.fullmatch(r"[a-f0-9]{64}", h) for h in hashes) + or type(self.target["helm_revision"]) is not int or self.target["helm_revision"] < 1 + or not self.target["public_recipient"].startswith("age1")): + raise Refused("exact_target_pins_required") + self.plan = json.loads(self.content["plan.json"]) + self.baseline = self.manifest["baseline"] + if (self.plan.get("schema") != "coulomb.source-date-backfill-plan.v1" + or self.plan.get("blockers") or not self.plan.get("prepared_change_set_valid") + or set(self.baseline) != {"tables_sha256", "files_sha256", + "tables", "rows", "files"} + or any(not re.fullmatch(r"[a-f0-9]{64}", self.baseline[n]) + for n in ("tables_sha256", "files_sha256")) + or any(type(self.baseline[n]) is not int or self.baseline[n] < 1 + for n in ("tables", "rows", "files"))): + raise Refused("exact_reviewed_date_plan_and_baseline_required") + + def payload(self, mode, backup=None): + names = ("date_backfill.py", "date_backfill_journal.py", "date_backfill_operation.py") + return {"mode": mode, "plan": self.plan, "baseline": self.baseline, + "database_name": "coulomb_social_db", + "image": self.target["image"], "recipient": self.target["public_recipient"], + "backup": backup, + "modules": {n: base64.b64encode(self.content[n]).decode() for n in names}, + "module_hashes": {n: self.manifest["files"][n] for n in names}} + + +class Node: + def __init__(self): + self.env = {**os.environ, "KUBECONFIG": "/etc/rancher/k3s/k3s.yaml"} + + def command(self, arguments, payload=None, *, env=None, timeout=210): + with tempfile.TemporaryFile() as stdout, tempfile.TemporaryFile() as stderr: + process = subprocess.Popen(arguments, + stdin=subprocess.PIPE if payload else subprocess.DEVNULL, + stdout=stdout, stderr=stderr, env=env or self.env) + try: + process.communicate(payload, timeout=timeout) + except BaseException: + process.terminate() + try: + process.wait(timeout=400) # permit admitted backup's resume cleanup + except subprocess.TimeoutExpired: + process.kill() + process.wait() + raise Refused("child_outcome_requires_review") from None + stdout.seek(0) + data = stdout.read(8 * 1024 * 1024 + 1) + if process.returncode or len(data) > 8 * 1024 * 1024: + raise Refused("private_child_refused") + return data + + def doc(self, *arguments): + return json.loads(self.command(["kubectl", *arguments, "-o", "json"])) + + def verify_local_node(self, node): + machine = Path("/etc/machine-id").read_text().strip() + if not machine or machine != node["status"]["nodeInfo"]["machineID"]: + raise Refused("run_on_the_pinned_content_node") + + def installation_pins(self): + helpers = {} + for name in HELPERS: + path = Path(name) + info = path.lstat() + if (not stat.S_ISREG(info.st_mode) or info.st_uid != 0 or info.st_mode & 0o022 + or info.st_nlink != 1 or any(p.is_symlink() for p in path.parents)): + raise Refused("installed_backup_helper_changed") + helpers[name] = sha(path.read_bytes()) + fields = dict(line.split("=", 1) for line in private_file( + Path("/etc/coulomb-social-backup/recipient.env")).decode().splitlines() + if line and not line.startswith("#")) + if set(fields) != {"COULOMB_SOCIAL_BACKUP_AGE_RECIPIENT"}: + raise Refused("public_recipient_environment_required") + return {"installed_helpers": helpers, + "public_recipient": fields["COULOMB_SOCIAL_BACKUP_AGE_RECIPIENT"]} + + def verify_backup_installation(self, target): + expected = {name: target[name] for name in ("installed_helpers", "public_recipient")} + if self.installation_pins() != expected: + raise Refused("installed_backup_configuration_changed") + for name in ("paired-backup", "backup-freshness"): + result = subprocess.run(["systemctl", "is-active", f"coulomb-social-{name}.timer"], + capture_output=True, timeout=15) + if result.returncode not in {3, 4}: + raise Refused("timer_state_requires_owner_review") + + def capture(self, target): + env = dict(self.env) + allowed = {"AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", + "COULOMB_SOCIAL_BACKUP_AGE_RECIPIENT"} + for name in ("primary.env", "recipient.env"): + raw = private_file(Path("/etc/coulomb-social-backup") / name) + for line in raw.decode().splitlines(): + if line and not line.startswith("#"): + key, value = line.split("=", 1) + if key not in allowed: + raise Refused("unexpected_backup_environment_field") + env[key] = value + if env.get("COULOMB_SOCIAL_BACKUP_AGE_RECIPIENT") != target["public_recipient"]: + raise Refused("recipient_binding_changed") + point = json.loads(self.command( + ["python3", "/opt/coulomb-social-backup/paired-backup.py", "--execute"], + env=env, timeout=1200)) + # Validate before scaling or executing any date code. The installed + # admitted helper owns the actual immutable publication/readback. + if (point.get("status") != "offsite_fetched_verified" + or point.get("remote_retrieval_verified") is not True + or point.get("immutable_create") is not True + or point.get("image") != target["image"] + or point.get("age_recipient") != target["public_recipient"] + or not point.get("version_id") or point["version_id"] == "null"): + raise Refused("fresh_native_capture_required") + name = point.get("local_archive", "") + if not re.fullmatch(r"\d{8}T\d{6}Z-[a-f0-9]{24}\.tar\.age", name): + raise Refused("own_local_ciphertext_required") + ciphertext = Path("/var/lib/coulomb-social-backup") / name + info = ciphertext.lstat() + if (not stat.S_ISREG(info.st_mode) or info.st_uid != 0 or info.st_mode & 0o077 + or info.st_nlink != 1 or info.st_size != point["ciphertext_bytes"]): + raise Refused("private_complete_ciphertext_required") + with ciphertext.open("rb") as stream: + digest = hashlib.file_digest(stream, "sha256").hexdigest() + captured = datetime.fromisoformat(point["captured_at"].replace("Z", "+00:00")) + completed = datetime.fromisoformat(point["completed_at"].replace("Z", "+00:00")) + now = datetime.now(UTC) + prefix = "platform-pg/application-archives/coulomb-social/" + if (point.get("application") != NS or digest != point["ciphertext_sha256"] + or digest != point["encrypted_sha256"] or captured.tzinfo is None + or completed.tzinfo is None or not captured <= completed <= now + or (now - captured).total_seconds() > 600 + or not point["object_key"].startswith(prefix) + or not point["object_key"].endswith("/" + name) + or point["destination"] != "s3://railiance-platform-pg-backup/" + + point["object_key"]): + raise Refused("fresh_native_ciphertext_binding_changed") + return point + + def prepare_root(self, root): + root.mkdir(mode=0o700, parents=True, exist_ok=True) + info = root.lstat() + if (not stat.S_ISDIR(info.st_mode) or info.st_uid != os.geteuid() or info.st_mode & 0o077 + or any(p.is_symlink() for p in root.parents)): + raise Refused("private_supervisor_directory_required") + sync_directory(root) + sync_directory(root.parent) + + @contextmanager + def operation_lock(self, root): + self.prepare_root(root) + fd = os.open(root / "supervisor.lock", os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600) + with os.fdopen(fd, "w"): + info = os.fstat(fd) + if (not stat.S_ISREG(info.st_mode) or info.st_uid != os.geteuid() + or info.st_mode & 0o077 or info.st_nlink != 1): + raise Refused("private_supervisor_lock_required") + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + yield + + def lock_backups(self, stack): + for name in ("backup.lock", "snapshot.lock"): + path = Path("/var/lib/coulomb-social-backup") / name + private_file(path) + stream = stack.enter_context(path.open("r+")) + fcntl.flock(stream.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + + def helper_directory(self, path): + if not path.exists(): + path.mkdir(mode=0o700) + os.chown(path, 10001, 10001) + info = path.lstat() + if not stat.S_ISDIR(info.st_mode) or info.st_uid != 10001 or info.st_mode & 0o077: + raise Refused("private_helper_journal_directory_required") + sync_directory(path) + sync_directory(path.parent) + + def read_state(self, path): + return json.loads(private_file(path)) + + +class Supervisor: + def __init__(self, capsule, node, *, state_root=STATE_ROOT): + self.capsule, self.node = capsule, node + self.target, self.state_root = capsule.target, state_root + self.directory = state_root / capsule.digest.removeprefix("sha256:") + self.helper = "csoc-date-" + capsule.digest.removeprefix("sha256:")[:20] + self.state = None + + def record(self, phase, **changes): + self.state.update(changes, phase=phase, updated_at=datetime.now(UTC).isoformat()) + save(self.directory / "receipt.json", self.state) + + def deployment(self, replicas): + dep = self.node.doc("get", "deployment", NS, "-n", NS) + spec = copy.deepcopy(dep["spec"]) + if dep["metadata"]["uid"] != self.target["deployment_uid"] or spec["replicas"] != replicas: + raise Refused("deployment_identity_or_scale_changed") + spec["replicas"] = 1 + if sha(canonical(spec)) != self.target["deployment_spec_sha256"]: + raise Refused("deployment_spec_changed") + if (len(spec["template"]["spec"]["containers"]) != 1 + or spec["template"]["spec"]["containers"][0]["image"] != self.target["image"] + or spec["strategy"]["type"] != "Recreate"): + raise Refused("single_pinned_recreate_writer_required") + claims = [(v["name"], v["persistentVolumeClaim"]["claimName"]) + for v in spec["template"]["spec"].get("volumes", []) + if "persistentVolumeClaim" in v] + if claims != [("content", CLAIM)]: + raise Refused("exact_content_claim_required") + return dep + + def writers(self): + return [pod for pod in self.node.doc("get", "pods", "-n", NS)["items"] + if pod.get("status", {}).get("phase") not in {"Succeeded", "Failed"} + and any(v.get("persistentVolumeClaim", {}).get("claimName") == CLAIM + for v in pod["spec"].get("volumes", []))] + + def identity(self): + if self.node.doc("get", "namespace", "kube-system")["metadata"]["uid"] != self.target[ + "cluster_uid"]: + raise Refused("cluster_changed") + node = self.node.doc("get", "node", self.target["node_name"]) + self.node.verify_local_node(node) + if (node["metadata"]["uid"] != self.target["node_uid"] or node["spec"].get("unschedulable") + or not any(c["type"] == "Ready" and c["status"] == "True" + for c in node["status"]["conditions"])): + raise Refused("node_changed_or_not_ready") + pvc = self.node.doc("get", "pvc", CLAIM, "-n", NS) + if pvc["metadata"]["uid"] != self.target["pvc_uid"] or pvc["status"]["phase"] != "Bound": + raise Refused("pvc_changed") + if self.node.doc("get", "hpa", "-n", NS)["items"]: + raise Refused("autoscaling_requires_owner_path") + if any(a.get("spec", {}).get("destination", {}).get("namespace") == NS for a in + self.node.doc("get", "applications.argoproj.io", "-A")["items"]): + raise Refused("gitops_requires_owner_path") + history = json.loads(self.node.command(["helm", "history", NS, "-n", NS, "-o", "json"])) + if (not history or history[-1]["status"] != "deployed" + or history[-1]["revision"] != self.target["helm_revision"] + or sha(self.node.command(["helm", "get", "manifest", NS, "-n", NS])) + != self.target["helm_manifest_sha256"]): + raise Refused("helm_release_changed") + + def wire(self, pod, mode, backup=None): + return json.loads(self.node.command( + ["kubectl", "exec", "-i", "-n", NS, pod, "--", "env", + "PYTHONDONTWRITEBYTECODE=1", "python", "-c", self.capsule.content["wire.py"].decode()], + canonical(self.capsule.payload(mode, backup)))) + + def preflight(self): + self.identity() + dep = self.deployment(1) + pods = self.writers() + if (len(pods) != 1 or pods[0]["spec"].get("nodeName") != self.target["node_name"] + or pods[0].get("metadata", {}).get("deletionTimestamp") + or not any(c["type"] == "Ready" and c["status"] == "True" + for c in pods[0]["status"].get("conditions", []))): + raise Refused("single_ready_application_writer_required") + containers = pods[0]["spec"].get("containers", []) + statuses = pods[0]["status"].get("containerStatuses", []) + if (len(containers) != 1 or containers[0]["image"] != self.target["image"] + or len(statuses) != 1 or not statuses[0].get("imageID", "").endswith( + self.target["image"].split("@", 1)[1])): + raise Refused("actual_running_image_changed") + owner = pods[0]["metadata"]["ownerReferences"] + if len(owner) != 1 or owner[0]["kind"] != "ReplicaSet": + raise Refused("unexpected_application_writer") + rs = self.node.doc("get", "replicaset", owner[0]["name"], "-n", NS) + if not any(r["uid"] == self.target["deployment_uid"] + for r in rs["metadata"]["ownerReferences"]): + raise Refused("unexpected_application_controller") + self.node.verify_backup_installation(self.target) + result = self.wire(pods[0]["metadata"]["name"], "prepare") + if (result["fingerprint"] != self.capsule.baseline + or not result["storage_constraint_verified"]): + raise Refused("reviewed_dataset_or_storage_prerequisite_changed") + requested = self.helper_manifest(dep) + dry = json.loads(self.node.command(["kubectl", "create", "--dry-run=server", + "-f", "-", "-o", "json"], canonical(requested))) + self.validate_helper(dry["spec"], requested["spec"]) + self.preview_helper_spec = sha(canonical(dry["spec"])) + return dep, result + + def validate_helper(self, actual, requested): + if (len(actual.get("containers", [])) != 1 or actual.get("initContainers") + or actual.get("hostNetwork") or actual.get("hostPID") or actual.get("hostIPC") + or actual.get("automountServiceAccountToken") is not False + or actual.get("nodeName") != self.target["node_name"] + or actual.get("securityContext") != requested["securityContext"] + or actual.get("volumes") != requested["volumes"]): + raise Refused("unreviewed_helper_pod_admission") + container = actual["containers"][0] + for key in ("image", "command", "env", "envFrom", "securityContext", "resources", + "volumeMounts"): + if container.get(key) != requested["containers"][0].get(key): + raise Refused("unreviewed_helper_container_admission") + if container.get("args") or container.get("lifecycle"): + raise Refused("unreviewed_helper_process_admission") + + def create_helper(self, dep): + self.node.helper_directory(self.directory / "private") + helper = self.helper_manifest(dep) + dry = json.loads(self.node.command(["kubectl", "create", "--dry-run=server", + "-f", "-", "-o", "json"], canonical(helper))) + self.validate_helper(dry["spec"], helper["spec"]) + digest = sha(canonical(dry["spec"])) + if self.state.get("helper_spec_sha256", digest) != digest: + raise Refused("helper_admission_changed") + self.record("helper_create_started", helper_spec_sha256=digest, helper_uid=None) + created = json.loads(self.node.command( + ["kubectl", "create", "-f", "-", "-o", "json"], canonical(helper))) + self.record("helper_created", helper_uid=created["metadata"]["uid"]) + self.node.command(["kubectl", "wait", "--for=condition=Ready", f"pod/{self.helper}", + "-n", NS, "--timeout=120s"]) + self.assert_fenced(helper=True) + + def helper_manifest(self, dep): + app = dep["spec"]["template"]["spec"] + container = app["containers"][0] + mounts = [v for v in container.get("volumeMounts", []) + if v["mountPath"] == "/app/var/content"] + if len(mounts) != 1 or mounts[0]["name"] != "content": + raise Refused("expected_content_mount_required") + return {"apiVersion": "v1", "kind": "Pod", "metadata": { + "name": self.helper, "namespace": NS, + "annotations": {"csoc-date-proposal": self.capsule.digest}}, "spec": { + "nodeName": self.target["node_name"], "restartPolicy": "Never", + "automountServiceAccountToken": False, + "securityContext": {"runAsNonRoot": True, "runAsUser": 10001, + "runAsGroup": 10001, "fsGroup": 10001, + "seccompProfile": {"type": "RuntimeDefault"}}, + "imagePullSecrets": copy.deepcopy(app.get("imagePullSecrets", [])), + "containers": [{"name": "operation", "image": self.target["image"], + "imagePullPolicy": "IfNotPresent", + "command": ["python", "-c", "import time; time.sleep(1800)"], + "envFrom": copy.deepcopy(container.get("envFrom", [])), + "env": copy.deepcopy(container.get("env", [])), + "securityContext": {"allowPrivilegeEscalation": False, + "readOnlyRootFilesystem": True, + "capabilities": {"drop": ["ALL"]}}, + "resources": {"requests": {"cpu": "10m", "memory": "64Mi"}, + "limits": {"cpu": "100m", "memory": "256Mi"}}, + "volumeMounts": [{"name": "content", "mountPath": "/app/var/content"}, + {"name": "operation", "mountPath": "/operation"}, + {"name": "temporary", "mountPath": "/tmp"}]}], + "volumes": [{"name": "content", "persistentVolumeClaim": {"claimName": CLAIM}}, + {"name": "operation", "hostPath": { + "path": str(self.directory / "private"), "type": "Directory"}}, + {"name": "temporary", "emptyDir": { + "medium": "Memory", "sizeLimit": "32Mi"}}]}} + + def assert_fenced(self, helper=False): + self.identity() + self.deployment(0) + writers = self.writers() + if helper: + if (len(writers) != 1 or writers[0]["metadata"]["name"] != self.helper + or writers[0]["metadata"]["uid"] != self.state["helper_uid"] + or sha(canonical(writers[0]["spec"])) != self.state["helper_spec_sha256"]): + raise Refused("helper_identity_or_writer_fence_changed") + elif writers: + raise Refused("application_writers_not_gone") + + def finish(self, expected): + self.assert_fenced(helper=True) + if self.wire(self.helper, "probe")["fingerprint"] != expected: + raise Refused("full_dataset_postflight_changed") + self.record("helper_delete_started", expected_final=expected) + options = {"apiVersion": "v1", "kind": "DeleteOptions", "gracePeriodSeconds": 30, + "preconditions": {"uid": self.state["helper_uid"]}} + self.node.command(["kubectl", "delete", "--raw", + f"/api/v1/namespaces/{NS}/pods/{self.helper}", "-f", "-"], + canonical(options)) + self.node.command(["kubectl", "wait", "--for=delete", f"pod/{self.helper}", "-n", NS, + "--timeout=120s"]) + self.assert_fenced() + dep = self.deployment(0) + self.record("resume_started") + self.node.command(["kubectl", "scale", f"deployment/{NS}", "-n", NS, + "--replicas=1", "--current-replicas=0", + "--resource-version=" + dep["metadata"]["resourceVersion"]]) + self.node.command(["kubectl", "rollout", "status", f"deployment/{NS}", "-n", NS, + "--timeout=180s"]) + self.deployment(1) + self.record("completed", application_resumed=True) + return self.state + + def execute(self): + with self.node.operation_lock(self.state_root): + return self._execute() + + def _execute(self): + if self.directory.exists(): + raise Refused("prior_attempt_requires_recovery_not_replay") + dep, before = self.preflight() + self.directory.mkdir(mode=0o700, parents=True) + sync_directory(self.state_root) + self.state = {"schema": "coulomb.supervised-date-attempt.v1", + "proposal_digest": self.capsule.digest, "application_resumed": False, + "pid": os.getpid(), "expected_after": before["expected_after"], + "helper_spec_sha256": self.preview_helper_spec} + frozen = self.directory / "capsule" + frozen.mkdir(mode=0o700) + for name, data in {**self.capsule.content, + "manifest.json": canonical(self.capsule.manifest)}.items(): + fd = os.open(frozen / name, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, "wb") as stream: + stream.write(data) + stream.flush() + os.fsync(stream.fileno()) + sync_directory(frozen) + self.record("capture_started") + try: + point = self.node.capture(self.target) + self.record("captured", backup=point) + _, after = self.preflight() + if after != before or self.preview_helper_spec != self.state["helper_spec_sha256"]: + raise Refused("dataset_changed_across_capture") + with ExitStack() as locks: + self.node.lock_backups(locks) + dep = self.deployment(1) + self.record("fencing") + self.node.command(["kubectl", "scale", f"deployment/{NS}", "-n", NS, + "--replicas=0", "--current-replicas=1", + "--resource-version=" + dep["metadata"]["resourceVersion"]]) + self.node.command(["kubectl", "wait", "--for=delete", "pod", "-n", NS, + "-l", "app.kubernetes.io/instance=" + NS, "--timeout=120s"]) + self.assert_fenced() + self.create_helper(dep) + if self.wire(self.helper, "prepare") != before: + raise Refused("dataset_changed_before_correction") + self.record("apply_started") + result = self.wire(self.helper, "apply", point) + if (result.get("status") != "completed" or not result.get("whole_store_postflight") + or result.get("fingerprint") != before["expected_after"]): + raise Refused("date_result_unconfirmed") + self.record("applied", operation=result, correction_status="completed") + return self.finish(before["expected_after"]) + except BaseException as exc: + self.record("requires_review", failed_phase=self.state["phase"], + failure_reason=str(exc) if isinstance(exc, Refused) else type(exc).__name__) + raise + + def recover(self): + """Explicit attended reconciliation; never captures or replays apply.""" + with self.node.operation_lock(self.state_root), ExitStack() as locks: + self.node.lock_backups(locks) + self.state = self.node.read_state(self.directory / "receipt.json") + if (self.state.get("schema") != "coulomb.supervised-date-attempt.v1" + or self.state["proposal_digest"] != self.capsule.digest): + raise Refused("original_attempt_required") + self.identity() + dep = self.node.doc("get", "deployment", NS, "-n", NS) + if dep["spec"]["replicas"] == 1: + # A lost response after scale-up must never undo later writes. + if (self.state.get("failed_phase", self.state["phase"]) != "resume_started" + or "expected_final" not in self.state): + raise Refused("running_application_requires_owner_review") + self.deployment(1) + self.node.command(["kubectl", "rollout", "status", f"deployment/{NS}", + "-n", NS, "--timeout=180s"]) + self.record("completed", application_resumed=True, + resumed_reply_reconciled=True, post_resume_writes_preserved=True) + return self.state + self.deployment(0) + writers = self.writers() + if not writers: + # Lost deletion reply or interruption before helper creation: + # recreate only the same pinned helper against retained state. + terminal = [p for p in self.node.doc("get", "pods", "-n", NS)["items"] + if p["metadata"]["name"] == self.helper] + if terminal: + old = terminal[0] + if (len(terminal) != 1 or old.get("status", {}).get("phase") not in { + "Succeeded", "Failed"} + or sha(canonical(old["spec"])) != self.state.get("helper_spec_sha256") + or old["metadata"].get("annotations", {}).get("csoc-date-proposal") + != self.capsule.digest + or (self.state.get("helper_uid") and self.state["helper_uid"] + != old["metadata"]["uid"])): + raise Refused("terminal_helper_identity_requires_review") + self.record("terminal_helper_delete_started", helper_uid=old["metadata"]["uid"]) + self.node.command(["kubectl", "delete", "--raw", + f"/api/v1/namespaces/{NS}/pods/{self.helper}", "-f", "-"], + canonical({"apiVersion": "v1", "kind": "DeleteOptions", + "gracePeriodSeconds": 30, + "preconditions": { + "uid": self.state["helper_uid"]}})) + self.node.command(["kubectl", "wait", "--for=delete", f"pod/{self.helper}", + "-n", NS, "--timeout=120s"]) + self.create_helper(dep) + writers = self.writers() + if (len(writers) != 1 or writers[0]["metadata"]["name"] != self.helper + or sha(canonical(writers[0]["spec"])) != self.state.get("helper_spec_sha256")): + raise Refused("original_helper_requires_owner_review") + if (self.state.get("helper_uid") + and self.state["helper_uid"] != writers[0]["metadata"]["uid"]): + raise Refused("helper_replaced_requires_owner_review") + self.state["helper_uid"] = writers[0]["metadata"]["uid"] + self.assert_fenced(helper=True) + observed = self.wire(self.helper, "inspect", self.state.get("backup")) + status = observed.get("status") + if status == "operation_busy": + raise Refused("original_operation_still_running") + if status == "pending": + self.record("reconcile_started") + observed = self.wire(self.helper, "reconcile", self.state["backup"]) + status = observed.get("status") + expected = (self.state["expected_after"] if status == "completed" + else self.capsule.baseline) + if status not in {"completed", "rolled_back", "absent"} or observed.get( + "fingerprint") != expected: + raise Refused("reconciled_dataset_requires_owner_review") + self.record("reconciled", correction_status=status) + return self.finish(expected) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--capsule", type=Path, required=True) + parser.add_argument("--expect-digest", required=True) + modes = parser.add_mutually_exclusive_group() + modes.add_argument("--execute", action="store_true") + modes.add_argument("--recover", action="store_true") + args = parser.parse_args() + capsule = Capsule(args.capsule, args.expect_digest) + if sha(Path(__file__).read_bytes()) != capsule.manifest["files"]["supervisor.py"]: + raise Refused("run_the_reviewed_supervisor_source") + if os.geteuid() != 0: + raise Refused("node_root_context_required") + supervisor = Supervisor(capsule, Node()) + if args.execute or args.recover: + if not sys.stdin.isatty(): + raise Refused("founder_attended_terminal_required") + action = "RECOVER-DATE" if args.recover else "APPLY-DATE" + if input("Type " + action + " " + capsule.digest + " to continue: ").strip() != ( + action + " " + capsule.digest): + raise Refused("not_confirmed") + print(json.dumps(supervisor.recover() if args.recover else supervisor.execute())) + else: + _, proof = supervisor.preflight() + print(json.dumps({"status": "read_only_preflight", "proposal_digest": capsule.digest, + "before": proof, "live_changes": False})) + + +if __name__ == "__main__": + try: + main() + except Exception as exc: + print(json.dumps({"status": "refused", "failure_type": type(exc).__name__, + "reason": str(exc) if isinstance(exc, Refused) else type(exc).__name__})) + raise SystemExit(1) diff --git a/scripts/source_date_wire.py b/scripts/source_date_wire.py new file mode 100644 index 0000000..fce3ed5 --- /dev/null +++ b/scripts/source_date_wire.py @@ -0,0 +1,139 @@ +"""Pinned helper protocol. Stdin/private journals hold data; stdout is metadata only.""" + +import base64 +import copy +import fcntl +import hashlib +import json +import os +import signal +import stat +import sys +import tempfile +import types +from pathlib import Path + + +def canonical(value): + return json.dumps(value, sort_keys=True, separators=(",", ":"), default=str).encode() + + +def run(payload): + import django + + django.setup() + for filename in ("date_backfill.py", "date_backfill_journal.py", "date_backfill_operation.py"): + source = base64.b64decode(payload["modules"][filename], validate=True) + if hashlib.sha256(source).hexdigest() != payload["module_hashes"][filename]: + raise ValueError("Reviewed module changed") + name = "coulomb_social.apps.spaces." + filename[:-3] + module = types.ModuleType(name) + module.__package__ = "coulomb_social.apps.spaces" + module.__file__ = "" + sys.modules[name] = module + exec(compile(source, module.__file__, "exec"), module.__dict__) + + from django.conf import settings + from django.db import connection + + from coulomb_social.apps.spaces.date_backfill import prepare_index_date_patch + from coulomb_social.apps.spaces.date_backfill_operation import ( + _set_date, + _snapshot, + execute_date_operation, + fingerprint, + ) + + if (connection.vendor != "postgresql" or settings.CONTENT_ROOT != "/app/var/content" + or connection.settings_dict["NAME"] != payload["database_name"]): + raise ValueError("Pinned application database/content context required") + with connection.cursor() as cursor: + cursor.execute("SET lock_timeout='5s'") + cursor.execute("SET statement_timeout='60s'") + cursor.execute("SELECT conname FROM pg_constraint " + "WHERE conrelid='public.spaces_space'::regclass") + constraints = {r[0] for r in cursor.fetchall()} + cursor.execute("SELECT count(*) FROM django_migrations WHERE app='spaces' " + "AND name='0003_space_slug_global_uniq'") + migrated = cursor.fetchone()[0] == 1 + if "spaces_space_slug_global_uniq" not in constraints or not migrated: + raise ValueError("Storage release prerequisite missing") + root = Path(settings.CONTENT_ROOT) + mode = payload["mode"] + if mode in {"probe", "prepare"}: + snapshot = _snapshot(root) + result = {"fingerprint": fingerprint(snapshot), "storage_constraint_verified": True} + if mode == "prepare": + if result["fingerprint"] != payload["baseline"]: + raise ValueError("Reviewed dataset differs") + selected = payload["plan"]["selection"] + name = f"spaces/{selected['space_slug']}/pages/index.md" + patched = prepare_index_date_patch(payload["plan"], (root / name).read_bytes()) + expected = copy.deepcopy(snapshot) + _set_date(expected, payload["plan"], payload["plan"]["source_created_at"]) + expected["files"][name] = hashlib.sha256(patched).hexdigest() + result["expected_after"] = fingerprint(expected) + return result + if mode not in {"apply", "inspect", "reconcile"}: + raise ValueError("Unsupported operation") + journal = Path("/operation/journal.json") + lock_fd = os.open("/operation/execute.lock", os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600) + with os.fdopen(lock_fd, "w"): + try: + fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + return {"status": "operation_busy", "writers_must_remain_fenced": True} + if mode == "inspect": + status = "absent" + if journal.exists() or journal.is_symlink(): + fd = os.open(journal, os.O_RDONLY | os.O_NOFOLLOW) + with os.fdopen(fd, "rb") as stream: + info = os.fstat(stream.fileno()) + raw = stream.read(4 * 1024 * 1024 + 1) + if (not stat.S_ISREG(info.st_mode) or info.st_uid != os.geteuid() + or info.st_mode & 0o077 or info.st_nlink != 1 + or len(raw) > 4 * 1024 * 1024): + raise ValueError("Private bound journal required") + record = json.loads(raw) + original = base64.b64decode(record["original"], validate=True) + patched = prepare_index_date_patch(payload["plan"], original) + if (record["schema"] != "coulomb.source-date-journal.v1" + or record["plan"] != payload["plan"] + or record["backup_receipt_sha256"] != hashlib.sha256( + canonical(payload["backup"])).hexdigest() + or record["patched_sha256"] != hashlib.sha256(patched).hexdigest() + or record["state"] not in {"pending", "completed", "rolled_back"}): + raise ValueError("Original operation journal binding changed") + status = record["state"] + return {"status": status, "fingerprint": fingerprint(_snapshot(root))} + proof_path = Path("/operation/result.json") + result = execute_date_operation( + root, payload["plan"], journal, payload["baseline"], payload["backup"], + image=payload["image"], recipient=payload["recipient"], reconcile=mode == "reconcile") + result["fingerprint"] = fingerprint(_snapshot(root)) + fd, temporary = tempfile.mkstemp(dir="/operation", prefix=".result-") + with os.fdopen(fd, "wb") as stream: + stream.write(canonical(result)) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, proof_path) + directory = os.open("/operation", os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(directory) + finally: + os.close(directory) + return result + + +if __name__ == "__main__": + # A lost kubectl client must not leave an unbounded helper operation behind. + # SIGALRM kills this process/SQL connection; the durable journal reconciles it. + signal.alarm(150) + try: + raw = sys.stdin.buffer.read(4 * 1024 * 1024 + 1) + if len(raw) > 4 * 1024 * 1024: + raise ValueError("Bounded input required") + print(json.dumps(run(json.loads(raw)))) + except Exception as exc: + print(json.dumps({"status": "refused", "failure_type": type(exc).__name__})) + raise SystemExit(1) diff --git a/tests/test_source_date_supervisor.py b/tests/test_source_date_supervisor.py new file mode 100644 index 0000000..ebe26b7 --- /dev/null +++ b/tests/test_source_date_supervisor.py @@ -0,0 +1,367 @@ +"""Supervisor lifecycle against a synthetic Kubernetes/backup adapter, no network.""" + +import copy +import importlib.util +import json +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +spec = importlib.util.spec_from_file_location("source_date_supervisor", + ROOT / "scripts/source_date_supervisor.py") +supervisor = importlib.util.module_from_spec(spec) +spec.loader.exec_module(supervisor) +IMAGE = "forgejo.coulomb.social/coulomb/coulomb-social@sha256:" + "a" * 64 +BASE = {"tables_sha256": "b" * 64, "files_sha256": "c" * 64, + "tables": 13, "rows": 112, "files": 82} +AFTER = {**BASE, "tables_sha256": "d" * 64, "files_sha256": "e" * 64} + + +class FakeNode(supervisor.Node): + """Only control-plane/capture effects are simulated; production paths never open.""" + + def __init__(self): + self.dep = {"metadata": {"uid": "deployment-id", "resourceVersion": "17"}, "spec": { + "replicas": 1, "strategy": {"type": "Recreate"}, "template": {"spec": { + "containers": [{"name": "app", "image": IMAGE, + "env": [{"name": "CONTENT_ROOT", "value": "/app/var/content"}], + "envFrom": [{"secretRef": {"name": "synthetic-app-env"}}], + "volumeMounts": [{"name": "content", + "mountPath": "/app/var/content"}]}], + "volumes": [{"name": "content", "persistentVolumeClaim": { + "claimName": supervisor.CLAIM}}]}}}} + self.fp, self.helper, self.journal = BASE, None, "absent" + self.calls, self.fail, self.capture_count, self.apply_count = [], None, 0, 0 + self.next_uid = 1 + self.busy = False + self.sidecar = False + self.extra_writer = False + self.running_image = None + + def trip(self, name): + if self.fail == name: + self.fail = None + raise supervisor.Refused("synthetic lost response: " + name) + + def doc(self, *args): + kind = args[1] + if kind == "namespace": + return {"metadata": {"uid": "cluster-id"}} + if kind == "node": + return {"metadata": {"uid": "node-id", "name": "synthetic-node"}, + "spec": {}, "status": { + "conditions": [{"type": "Ready", "status": "True"}]}} + if kind == "pvc": + return {"metadata": {"uid": "pvc-id"}, "status": {"phase": "Bound"}} + if kind in {"hpa", "applications.argoproj.io"}: + return {"items": []} + if kind == "deployment": + return copy.deepcopy(self.dep) + if kind == "replicaset": + return {"metadata": {"ownerReferences": [{"uid": "deployment-id"}]}} + if kind == "pods": + pods = [] + if self.dep["spec"]["replicas"]: + pods.append({"metadata": {"name": "synthetic-app", "ownerReferences": [ + {"kind": "ReplicaSet", "name": "synthetic-rs"}]}, "spec": { + "nodeName": "synthetic-node", "containers": [{"image": self.running_image + or self.dep["spec"]["template"]["spec"]["containers"][0]["image"]}], + "volumes": [{"persistentVolumeClaim": { + "claimName": supervisor.CLAIM}}]}, "status": { + "phase": "Running", "conditions": [ + {"type": "Ready", "status": "True"}], + "containerStatuses": [{"imageID": self.dep["spec"]["template"][ + "spec"]["containers"][0]["image"]}]}}) + if self.helper: + pods.append(copy.deepcopy(self.helper)) + if self.extra_writer: + pods.append({"metadata": {"name": "unreviewed-writer"}, "spec": {"volumes": [ + {"persistentVolumeClaim": {"claimName": supervisor.CLAIM}}]}, + "status": {"phase": "Running"}}) + return {"items": pods} + raise AssertionError(args) + + def command(self, args, payload=None, **kwargs): + self.calls.append(args[:4]) + if args[:2] == ["helm", "history"]: + return supervisor.canonical([{"revision": 18, "status": "deployed"}]) + if args[:3] == ["helm", "get", "manifest"]: + return b"synthetic helm manifest" + if args[:2] == ["kubectl", "exec"]: + message = json.loads(payload) + mode = message["mode"] + if mode == "prepare": + return supervisor.canonical({"fingerprint": self.fp, + "storage_constraint_verified": True, + "expected_after": AFTER}) + if mode == "probe": + return supervisor.canonical({"fingerprint": self.fp}) + if mode == "inspect": + status = "operation_busy" if self.busy else self.journal + return supervisor.canonical({"status": status, + "fingerprint": self.fp}) + if mode == "reconcile": + assert self.journal == "pending" + self.journal, self.fp = "rolled_back", BASE + return supervisor.canonical({"status": self.journal, "fingerprint": self.fp}) + assert mode == "apply" + self.apply_count += 1 + if self.fail == "apply_pending": + self.journal = "pending" + self.trip("apply_pending") + self.journal, self.fp = "completed", AFTER + self.trip("apply_reply") + return supervisor.canonical({"status": "completed", "whole_store_postflight": True, + "fingerprint": self.fp}) + if args[:2] == ["kubectl", "scale"]: + current = next(a for a in args if a.startswith("--current-replicas=")).split("=")[1] + assert int(current) == self.dep["spec"]["replicas"] + wanted = next(a for a in args if a.startswith("--replicas=")).split("=")[1] + self.dep["spec"]["replicas"] = int(wanted) + return b"" + if args[:2] == ["kubectl", "create"]: + pod = json.loads(payload) + if self.sidecar: + pod["spec"]["containers"].append({"name": "injected"}) + if "--dry-run=server" not in args: + assert self.dep["spec"]["replicas"] == 0 and not self.helper + pod["metadata"]["uid"] = "helper-" + str(self.next_uid) + self.next_uid += 1 + pod["status"] = {"phase": "Running"} + self.helper = copy.deepcopy(pod) + self.trip("create_reply") + return supervisor.canonical(pod) + if args[:3] == ["kubectl", "delete", "--raw"]: + assert json.loads(payload)["preconditions"]["uid"] == self.helper["metadata"]["uid"] + self.helper = None + self.trip("delete_reply") + return b"" + if args[:3] == ["kubectl", "rollout", "status"]: + self.trip("resume_reply") + return b"" + assert args[:2] == ["kubectl", "wait"], args + return b"" + + def verify_backup_installation(self, target): + self.trip("helper_hash") + + def verify_local_node(self, node): + pass # synthetic cluster/root adapter, never a native machine-id claim + + def installation_pins(self): + return {"installed_helpers": {name: "f" * 64 for name in supervisor.HELPERS}, + "public_recipient": "age1synthetic"} + + def capture(self, target): + self.capture_count += 1 + self.trip("capture") + return {"status": "synthetic-capture"} + + def lock_backups(self, stack): + self.trip("backup_lock") + + def helper_directory(self, path): + path.mkdir(mode=0o700, exist_ok=True) + + def read_state(self, path): + return json.loads(path.read_bytes()) + + +@pytest.fixture +def prepared(tmp_path): + node = FakeNode() + target = {"cluster_uid": "cluster-id", "deployment_uid": "deployment-id", "node_uid": "node-id", + "node_name": "synthetic-node", "pvc_uid": "pvc-id", "image": IMAGE, + "deployment_spec_sha256": supervisor.sha(supervisor.canonical(node.dep["spec"])), + "helm_revision": 18, + "helm_manifest_sha256": supervisor.sha(b"synthetic helm manifest"), + "public_recipient": "age1synthetic", "installed_helpers": { + name: "f" * 64 for name in supervisor.HELPERS}} + capsule = tmp_path / "capsule" + capsule.mkdir(mode=0o700) + for name in supervisor.MEMBERS: + (capsule / name).write_text("synthetic source") + (capsule / "plan.json").write_text(json.dumps({ + "schema": "coulomb.source-date-backfill-plan.v1", "blockers": [], + "prepared_change_set_valid": True})) + manifest = {"schema": "coulomb.supervised-date-capsule.v1", "baseline": BASE, "target": target, + "attempt_id": "1" * 32, "requires_founder_attended_execution": True, + "files": {name: supervisor.sha((capsule / name).read_bytes()) + for name in supervisor.MEMBERS}} + (capsule / "manifest.json").write_bytes(supervisor.canonical(manifest)) + digest = "sha256:" + supervisor.sha(supervisor.canonical(manifest)) + frozen = supervisor.Capsule(capsule, digest) + root = tmp_path / "state" + return supervisor.Supervisor(frozen, node, state_root=root), node, capsule, digest + + +def test_success_captures_fences_applies_checks_and_resumes_once(prepared): + operation, node, _, _ = prepared + assert not operation.state_root.exists() + operation.preflight() + assert not operation.state_root.exists() and node.capture_count == 0 + result = operation.execute() + 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 + with pytest.raises(supervisor.Refused, match="prior_attempt"): + operation.execute() + assert node.apply_count == 1 + + +@pytest.mark.parametrize("failure", ["apply_reply", "apply_pending", + "create_reply", "delete_reply"]) +def test_uncertainty_preserves_fence_and_recovers_without_new_capture_or_apply(prepared, failure): + operation, node, _, _ = prepared + node.fail = failure + with pytest.raises(supervisor.Refused): + operation.execute() + assert node.dep["spec"]["replicas"] == 0 + applied = node.apply_count + result = operation.recover() + assert result["phase"] == "completed" and node.dep["spec"]["replicas"] == 1 + assert node.capture_count == 1 and node.apply_count == applied + assert not node.helper + if failure == "apply_pending": + assert result["correction_status"] == "rolled_back" + + +def test_live_original_operation_blocks_recovery_and_never_restarts(prepared): + operation, node, _, _ = prepared + node.fail = "apply_pending" + with pytest.raises(supervisor.Refused): + operation.execute() + node.busy = True + with pytest.raises(supervisor.Refused, match="still_running"): + operation.recover() + assert node.capture_count == node.apply_count == 1 and node.dep["spec"]["replicas"] == 0 + + +def test_lost_resume_reply_preserves_new_productive_writes(prepared): + operation, node, _, _ = prepared + node.fail = "resume_reply" + with pytest.raises(supervisor.Refused): + operation.execute() + assert node.dep["spec"]["replicas"] == 1 + node.fp = {**AFTER, "rows": 113} # legitimate write after the application resumed + result = operation.recover() + assert result["post_resume_writes_preserved"] and node.fp["rows"] == 113 + assert node.capture_count == node.apply_count == 1 + + +@pytest.mark.parametrize("problem", ["pod_injection", "foreign_writer", + "spec_drift", "helper_hash", "pod_image"]) +def test_preflight_refusals_never_capture_or_scale(prepared, problem): + operation, node, _, _ = prepared + if problem == "pod_injection": + node.sidecar = True + elif problem == "foreign_writer": + node.extra_writer = True + elif problem == "spec_drift": + node.dep["spec"]["template"]["spec"]["containers"][0]["image"] += "0" + elif problem == "pod_image": + node.running_image = IMAGE + "0" + else: + node.fail = "helper_hash" + with pytest.raises(supervisor.Refused): + operation.execute() + assert node.capture_count == node.apply_count == 0 and node.dep["spec"]["replicas"] == 1 + + +def test_changed_capsule_refuses_before_any_backend_access(prepared): + _, _, path, digest = prepared + (path / "wire.py").write_text("changed") + with pytest.raises(supervisor.Refused, match="member_changed"): + supervisor.Capsule(path, digest) + + +def test_unrelated_data_drift_after_uncertain_result_never_resumes(prepared): + operation, node, _, _ = prepared + node.fail = "apply_reply" + with pytest.raises(supervisor.Refused): + operation.execute() + node.fp = {**AFTER, "rows": 999} + with pytest.raises(supervisor.Refused, match="dataset_requires_owner_review"): + operation.recover() + assert node.dep["spec"]["replicas"] == 0 and node.helper + assert node.capture_count == node.apply_count == 1 + + +def test_replaced_helper_is_never_deleted_or_used_for_recovery(prepared): + operation, node, _, _ = prepared + node.fail = "apply_reply" + with pytest.raises(supervisor.Refused): + operation.execute() + node.helper["metadata"]["uid"] = "foreign-helper" + with pytest.raises(supervisor.Refused, match="helper_replaced"): + operation.recover() + assert node.dep["spec"]["replicas"] == 0 + assert node.helper["metadata"]["uid"] == "foreign-helper" + + +def test_terminal_original_helper_is_uid_deleted_and_recreated_only_for_reconciliation(prepared): + operation, node, _, _ = prepared + node.fail = "apply_reply" + with pytest.raises(supervisor.Refused): + operation.execute() + node.helper["status"]["phase"] = "Succeeded" + result = operation.recover() + assert result["correction_status"] == "completed" and not node.helper + assert node.next_uid == 3 and node.capture_count == node.apply_count == 1 + + +def test_capsule_preparation_is_private_fresh_and_has_a_new_attempt_id(prepared, tmp_path): + import runpy + import sys + + operation, _, source, _ = prepared + sys.path.insert(0, str(ROOT / "scripts")) + try: + prepare = runpy.run_path(str(ROOT / "scripts/prepare-source-date-capsule.py"))["prepare"] + finally: + sys.path.pop(0) + baseline, target = tmp_path / "baseline.json", tmp_path / "target.json" + baseline.write_bytes(supervisor.canonical(BASE)) + target.write_bytes(supervisor.canonical(operation.target)) + proofs = [] + for name in ("first", "second"): + output = tmp_path / name + proof = prepare(source / "plan.json", baseline, target, output) + capsule = supervisor.Capsule(output, proof["proposal_digest"]) + assert capsule.manifest["requires_founder_attended_execution"] + assert output.stat().st_mode & 0o777 == 0o700 + assert all(p.stat().st_mode & 0o777 == 0o600 for p in output.iterdir()) + proofs.append(proof["proposal_digest"]) + with pytest.raises(ValueError, match="Fresh private"): + prepare(source / "plan.json", baseline, target, output) + assert proofs[0] != proofs[1] + + +@pytest.mark.parametrize("changed_image", [False, True]) +def test_observer_prepares_without_capturing_or_refuses_missing_release( + prepared, tmp_path, monkeypatch, changed_image): + import runpy + import sys + + operation, node, source, _ = prepared + sys.path.insert(0, str(ROOT / "scripts")) + try: + observer = runpy.run_path(str(ROOT / "scripts/observe-source-date-target.py"))["observe"] + finally: + sys.path.pop(0) + monkeypatch.setattr(supervisor.os, "geteuid", lambda: 0) # simulated root context only + output = tmp_path / "observed-capsule" + if changed_image: + node.dep["spec"]["template"]["spec"]["containers"][0]["image"] += "0" + with pytest.raises(RuntimeError, match="prerequisite_missing"): + observer(source / "plan.json", output, image=IMAGE, cluster_uid="cluster-id", + deployment_uid="deployment-id", pvc_uid="pvc-id", node=node) + assert not output.exists() + else: + proof = observer(source / "plan.json", output, image=IMAGE, cluster_uid="cluster-id", + deployment_uid="deployment-id", pvc_uid="pvc-id", node=node) + capsule = supervisor.Capsule(output, proof["proposal_digest"]) + assert capsule.baseline == BASE and capsule.target == operation.target + assert proof["baseline_rechecked"] and not proof["live_changes"] + assert node.capture_count == node.apply_count == 0 and node.dep["spec"]["replicas"] == 1 diff --git a/workplans/CSOC-WP-0012-durable-recovery-service-readiness.md b/workplans/CSOC-WP-0012-durable-recovery-service-readiness.md index 7fde34a..8a9ce3a 100644 --- a/workplans/CSOC-WP-0012-durable-recovery-service-readiness.md +++ b/workplans/CSOC-WP-0012-durable-recovery-service-readiness.md @@ -262,6 +262,13 @@ unchanged. Temporary private archive removed. Proof: docs/release/client-data-export-rehearsal-2026-10-30.json. Full make check340passes, zero skips. No deployment, production export, teardown or expiry was performed. +2026-10-30 operational inventory: the prepared date supervisor introduces an +operator-private root date-operation store containing original client index bytes, +plan and backup binding. Client export inventory now names this store explicitly; +the app cannot read/export it. Include any future retained journals in operator +offboarding handoff and explicit expiry under the open service contract work. +No production date journal or expiry was created by the isolated rehearsal. + ## Completion Finish when every task above has its stated evidence and acceptance, or hand off any newly discovered residual to a live work record first. Cross-owner dependencies remain their existing canonical records. diff --git a/workplans/CSOC-WP-0014-cohort-transfer-bubble-offramp.md b/workplans/CSOC-WP-0014-cohort-transfer-bubble-offramp.md index 6135d85..887edb4 100644 --- a/workplans/CSOC-WP-0014-cohort-transfer-bubble-offramp.md +++ b/workplans/CSOC-WP-0014-cohort-transfer-bubble-offramp.md @@ -210,6 +210,24 @@ productive correction is implied.50 focused and364 full gate cases pass, zero sk The exact supervised node wrapper, live backup/source bookends, pinned helper/PVC admission, durable journal placement and operator postflight/resume remain open. +2026-10-30 supervisor follow-through: added exact private capsule preparation, +read-only native target observer and founder-terminal node supervisor. The wrapper +binds the physical node/PVC/deployment/Helm/image, source and full dataset, checks +fresh admitted capture/readback bookends, fences application/PVC writers, and runs +the operation through a pinned helper with a durable private host journal. Explicit +recovery handles original-operation busy state, pending/terminal journal outcomes, +lost create/delete/apply/resume replies and terminal helper replacement without +replaying apply or overwriting post-resume writes.19 focused cases pass. Review: +docs/release/source-date-supervisor-review-2026-10-30.md. Native observation and an +admitted date capsule must follow the still-pending storage release; none exists +for production yet. Source/rehearsal completion does not finish this task. +The real helper protocol and controller now pass normal completion and a lost +post-commit reply on disposable recovered-pair clones. Both cases compare all13 +tables/82files and recover without a second apply/capture; originals are unchanged +and clones removed. Kubernetes/root/provider actions are explicitly simulated. +Proof: docs/release/source-date-supervisor-rehearsal-2026-10-30.json. +`make check`:383 passed, zero skips, including required browser journeys. + ## Completion Finish when every task above has its stated evidence and acceptance, or hand off any newly discovered residual to a live work record first. Cross-owner dependencies remain their existing canonical records.