railiance-apps/tools/coulomb_image_release_node.py

"""Attended founder-only node release; argv/data are sealed by the reviewed capsule.

This is a supervisor command, not a new agent credential or approval boundary.
No Secret object/output is fetched. Existing approved backup custody stays local.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
import stat
import subprocess
import tempfile
import time
import urllib.request
from contextlib import ExitStack
from datetime import UTC, datetime
from pathlib import Path

JOURNAL_ROOT = Path("/var/lib/coulomb-social-backup/releases")
SNAPSHOT = Path("/opt/coulomb-social-backup/snapshot-content-and-db.py")
INSTALL_RECEIPT = Path(
    "/var/lib/coulomb-social-backup-installations/20261031bacc0001/receipt.json"
)


def canonical(value):
    return json.dumps(value, sort_keys=True, separators=(",", ":")).encode()


def sha(data):
    return hashlib.sha256(data).hexdigest()


class Refused(RuntimeError):
    pass


def private_directories(root, relative):
    """Create every component privately; never repair existing unsafe state."""
    if not root.is_absolute() or relative.is_absolute() or ".." in relative.parts:
        raise Refused("private_directory_path_required")
    with ExitStack() as stack:
        def opened(name, parent=None):
            fd = os.open(name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=parent)
            stack.callback(os.close, fd)
            return fd

        fd = opened("/")
        for component in root.parts[1:]:
            fd = opened(component, fd)
        for component in (*relative.parts, None):
            info = os.fstat(fd)
            if info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) != 0o700:
                raise Refused("private_owned_directory_required")
            if component is None:
                break
            try:
                os.mkdir(component, mode=0o700, dir_fd=fd)
                os.fsync(fd)
            except FileExistsError:
                pass
            fd = opened(component, fd)


class Release:
    def __init__(self, capsule, expected, read_only=False):
        self.root = capsule
        self.plan = json.loads((capsule / "manifest.json").read_text())
        frozen = {}
        if "sha256:" + sha(canonical(self.plan)) != expected:
            raise Refused("proposal_digest_differs")
        for name, digest in self.plan["files"].items():
            path = capsule / name
            if (
                not path.resolve().is_relative_to(capsule.resolve())
                or Path(name).is_absolute()
                or ".." in Path(name).parts
                or path.is_symlink()
                or not path.is_file()
                or sha(path.read_bytes()) != digest
            ):
                raise Refused("capsule_file_differs")
            frozen[name] = path.read_bytes()
            if sha(frozen[name]) != digest:
                raise Refused("capsule_changed_during_verification")
        if os.geteuid() != 0:
            raise Refused("attended_founder_root_execution_required")
        self.env = dict(os.environ, KUBECONFIG="/etc/rancher/k3s/k3s.yaml")
        self.read_only = read_only
        for field in ("values_file", "snapshot_file"):
            if field in self.plan:
                name = self.plan[field]
                if (not isinstance(name, str) or not name.startswith("release/")
                        or ".." in Path(name).parts or name not in frozen):
                    raise Refused("reviewed_release_artifact_required")
        self.receipt = {
            "schema": "coulomb.guarded-image-release.v1",
            "proposal_digest": expected,
            "started_at": datetime.now(UTC).isoformat(),
            "status": "preflight",
            "production_data_restored": False,
            "timers_activated": False,
        }
        self.journal = JOURNAL_ROOT / expected.removeprefix("sha256:")
        self.verify_installation_prerequisite()
        if read_only:
            return
        if self.journal.exists():
            raise Refused("prior_execution_requires_review")
        private_directories(JOURNAL_ROOT.parent, Path(JOURNAL_ROOT.name))
        self.journal.mkdir(mode=0o700)
        self.root = self.journal / "capsule"
        self.root.mkdir(mode=0o700)
        for name, data in frozen.items():
            target = self.root / name
            private_directories(self.root, Path(name).parent)
            fd = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
            with os.fdopen(fd, "wb") as stream:
                stream.write(data)
        self.receipt["pid"] = os.getpid()
        self.save()

    def save(self):
        if self.read_only:
            return
        fd, name = tempfile.mkstemp(prefix=".receipt-", dir=self.journal)
        with os.fdopen(fd, "w") as stream:
            json.dump(self.receipt, stream, sort_keys=True, indent=2)
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(name, self.journal / "receipt.json")
        directory = os.open(self.journal, os.O_RDONLY | os.O_DIRECTORY)
        try:
            os.fsync(directory)
        finally:
            os.close(directory)

    def command(self, arguments, payload=None, env=None, timeout=180, allowed=(0,)):
        # In particular, give the paired capture its existing app-resume window
        # on timeout; subprocess.run would SIGKILL it before that cleanup.
        with tempfile.TemporaryFile() as stdout, tempfile.TemporaryFile() as stderr:
            process = subprocess.Popen(
                arguments,
                env=env or self.env,
                stdin=subprocess.PIPE if payload is not None else subprocess.DEVNULL,
                stdout=stdout,
                stderr=stderr,
            )
            try:
                process.communicate(payload, timeout=timeout)
            except subprocess.TimeoutExpired:
                process.terminate()
                try:
                    process.wait(timeout=450)
                except subprocess.TimeoutExpired:
                    process.kill()
                    process.wait()
                raise Refused("private_child_timed_out_after_cleanup_window") from None
            stdout.seek(0)
            body = stdout.read(8 * 1024**2 + 1)
            if len(body) > 8 * 1024**2:
                raise Refused("bounded_child_output_required")
            result = subprocess.CompletedProcess(arguments, process.returncode, body)
        if result.returncode not in allowed:
            raise Refused("private_child_refused")
        return result

    def kdoc(self, *arguments):
        return json.loads(self.command(["kubectl", *arguments, "-o", "json"]).stdout)

    def history(self):
        return json.loads(
            self.command(
                [
                    "helm",
                    "history",
                    "coulomb-social",
                    "-n",
                    "coulomb-social",
                    "-o",
                    "json",
                ]
            ).stdout
        )

    def deployment(self):
        return self.kdoc("get", "deployment", "coulomb-social", "-n", "coulomb-social")

    def probe(self, mode="data"):
        source = (self.root / "tools/coulomb_image_probe.py").read_text()
        result = self.command(
            [
                "kubectl",
                "exec",
                "-n",
                "coulomb-social",
                "deploy/coulomb-social",
                "--",
                "python",
                "-c",
                source,
                mode,
            ]
        )
        return json.loads(result.stdout)

    def same_data(self, actual):
        for name in (
            "business_tables",
            "business_rows",
            "business_rows_sha256",
            "content_files",
            "content_files_sha256",
            "migration_rows_sha256",
        ):
            if actual[name] != self.plan["expected_data"][name]:
                raise Refused("approved_recovered_dataset_drifted")
        if actual["duplicate_slug_groups"] != 0 or actual["tenant_slug_constraint"] is not True:
            raise Refused("storage_invariant_refused")
        if (actual.get("global_slug_constraint") is not True
                or actual.get("migration_0003_applied") is not True):
            raise Refused("already_migrated_baseline_required")

    def artifact_name(self, field):
        return self.plan.get(field, {
            "values_file": "release/identity-export-values.yaml",
            "snapshot_file": "release/identity-export-snapshot.py",
        }[field])

    def verify_installation_prerequisite(self):
        expected = self.plan.get("prerequisite_installation_digest")
        if expected is None:
            return
        if not re.fullmatch(r"sha256:[a-f0-9]{64}", expected):
            raise Refused("reviewed_installation_digest_required")
        try:
            info = INSTALL_RECEIPT.lstat()
        except FileNotFoundError:
            raise Refused("inactive_backup_installation_required_before_image_release") from None
        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 INSTALL_RECEIPT.parents)):
            raise Refused("private_installation_receipt_required")
        receipt = json.loads(INSTALL_RECEIPT.read_bytes())
        if (receipt.get("schema") != "coulomb.backup-source-supervised-installation.v1"
                or receipt.get("proposal_digest") != expected
                or receipt.get("status") != "sources_installed_units_inactive_verified"
                or receipt.get("verified_targets") != 19
                or receipt.get("timers_activated") is not False):
            raise Refused("verified_inactive_backup_installation_required")
        for name in ("paired-backup", "backup-freshness"):
            for suffix in ("service", "timer"):
                unit = f"coulomb-social-{name}.{suffix}"
                raw = self.command(["systemctl", "show", unit,
                                    "--property=LoadState,ActiveState,UnitFileState"]).stdout
                state = dict(line.split("=", 1) for line in raw.decode().splitlines())
                if (state.get("LoadState") != "loaded" or state.get("ActiveState") != "inactive"
                        or state.get("UnitFileState") not in {"disabled", "static"}):
                    raise Refused("backup_units_must_remain_inactive_for_image_release")
        jobs = self.command(["systemctl", "list-jobs", "--no-legend", "--no-pager",
                             "--plain"]).stdout.decode().splitlines()
        for line in jobs:
            fields = line.split()
            if (len(fields) != 4 or not fields[0].isdecimal()
                    or fields[1].startswith("coulomb-social-")):
                raise Refused("queued_backup_job_requires_review")

    def preflight(self):
        self.verify_installation_prerequisite()
        self.verify_installed_helpers()
        if (
            self.kdoc("get", "namespace", "kube-system")["metadata"]["uid"]
            != self.plan["cluster_uid"]
        ):
            raise Refused("wrong_cluster")
        node = self.kdoc("get", "nodes")["items"]
        if len(node) != 1 or not any(
            a["address"] == "92.205.62.239" for a in node[0]["status"]["addresses"]
        ):
            raise Refused("wrong_node")
        conditions = {v["type"]: v["status"] for v in node[0]["status"]["conditions"]}
        if conditions.get("Ready") != "True" or any(
            conditions.get(k) != "False"
            for k in ("MemoryPressure", "DiskPressure", "PIDPressure")
        ):
            raise Refused("node_pressure")
        allocated = self.command(
            ["kubectl", "describe", "node", node[0]["metadata"]["name"]]
        ).stdout.decode()
        requests = {}
        for resource in ("cpu", "memory"):
            found = re.search(
                r"(?m)^\s+" + resource + r"\s+\S+\s+\((\d+)%\)", allocated
            )
            if not found or int(found[1]) > 90:
                raise Refused("node_request_headroom_refused")
            requests[resource] = int(found[1])
        self.receipt["node_request_percent"] = requests
        history = self.history()
        if (
            history[-1]["revision"] != self.plan["baseline_revision"]
            or history[-1]["status"] != "deployed"
        ):
            raise Refused("helm_baseline_changed")
        manifest = self.command(
            ["helm", "get", "manifest", "coulomb-social", "-n", "coulomb-social"]
        ).stdout
        if sha(manifest) != self.plan["baseline_helm_manifest_sha256"]:
            raise Refused("helm_manifest_changed")
        deployment = self.deployment()
        if (
            deployment["metadata"]["uid"] != self.plan["deployment_uid"]
            or sha(canonical(deployment["spec"])) != self.plan["baseline_spec_sha256"]
        ):
            raise Refused("live_spec_changed")
        if (
            deployment["status"].get("readyReplicas") != 1
            or deployment["spec"]["strategy"]["type"] != "Recreate"
        ):
            raise Refused("single_ready_writer_required")
        # Helm's three-way merge preserves unmanaged pod metadata. Bind the
        # complete desired live spec rather than treating kubectl apply's
        # last-applied projection as Helm's result.
        image_patch = [
            {"op": "test", "path": "/spec", "value": deployment["spec"]},
            {"op": "replace", "path": "/spec/template/spec/containers/0/image",
             "value": self.plan["candidate_image"]},
        ]
        preview = self.kdoc(
            "patch", "deployment", "coulomb-social", "-n", "coulomb-social",
            "--type=json", "--dry-run=server", "-p", json.dumps(image_patch),
        )
        if sha(canonical(preview["spec"])) != self.plan["candidate_spec_sha256"]:
            raise Refused("image_only_server_preview_differs")
        pvc = self.kdoc("get", "pvc", "coulomb-social-content", "-n", "coulomb-social")
        if (
            pvc["metadata"]["uid"] != self.plan["pvc_uid"]
            or pvc["status"]["phase"] != "Bound"
        ):
            raise Refused("pvc_changed")
        apps = self.kdoc("get", "applications.argoproj.io", "-A")["items"]
        if any(
            a.get("spec", {}).get("destination", {}).get("namespace")
            == "coulomb-social"
            for a in apps
        ):
            raise Refused("gitops_adoption_requires_owner_path")
        rendered = self.command(
            [
                "helm",
                "template",
                "coulomb-social",
                str(self.root / "charts/coulomb-social"),
                "-f",
                str(self.root / self.artifact_name("values_file")),
                "-n",
                "coulomb-social",
            ]
        ).stdout
        if sha(rendered) != self.plan["candidate_render_sha256"]:
            raise Refused("candidate_render_changed")
        self.command(["kubectl", "diff", "-f", "-"], payload=rendered, allowed=(0, 1))
        self.command(
            ["kubectl", "apply", "--dry-run=server", "-f", "-"], payload=rendered
        )
        self.command(
            [
                "helm",
                "upgrade",
                "coulomb-social",
                str(self.root / "charts/coulomb-social"),
                "-f",
                str(self.root / self.artifact_name("values_file")),
                "-n",
                "coulomb-social",
                "--dry-run=server",
            ]
        )
        before = self.probe()
        self.same_data(before)
        self.receipt["preflight"] = before
        self.save()

    def verify_installed_helpers(self):
        for path, expected in self.plan["installed_helper_hashes"].items():
            metadata = Path(path).lstat()
            if (
                not stat.S_ISREG(metadata.st_mode)
                or metadata.st_uid != 0
                or metadata.st_mode & 0o022
                or sha(Path(path).read_bytes()) != expected
            ):
                raise Refused("installed_backup_helper_changed")
    def capture(self):
        self.verify_installed_helpers()
        environment = dict(self.env)
        for name in ("primary.env", "recipient.env"):
            path = Path("/etc/coulomb-social-backup") / name
            metadata = path.lstat()
            if (
                not stat.S_ISREG(metadata.st_mode)
                or metadata.st_uid != 0
                or metadata.st_mode & 0o077
            ):
                raise Refused("private_own_backup_configuration_required")
            environment.update(
                dict(
                    line.split("=", 1)
                    for line in path.read_text().splitlines()
                    if line and not line.startswith("#")
                )
            )
        output = self.command(
            ["python3", "/opt/coulomb-social-backup/paired-backup.py", "--execute"],
            env=environment,
            timeout=1200,
        )
        point = json.loads(output.stdout)
        if (
            point.get("status") != "offsite_fetched_verified"
            or not point.get("remote_retrieval_verified")
            or not point.get("version_id")
            or point.get("version_id") == "null"
            or point.get("image") != self.plan["old_image"]
            or point.get("age_recipient") != self.plan["public_recipient"]
            or not point.get("immutable_create")
            or not point.get("destination", "").startswith(
                "s3://railiance-platform-pg-backup/platform-pg/application-archives/coulomb-social/"
            )
        ):
            raise Refused("fresh_paired_primary_readback_required")
        self.receipt["fresh_paired_point"] = point
        self.save()
        self.same_data(self.probe())

    def restore_pin(self, old):
        path = SNAPSHOT
        if (
            sha(path.read_bytes())
            == self.plan["files"][self.artifact_name("snapshot_file")]
        ):
            self.replace_pin(path, old)
        elif sha(path.read_bytes()) != sha(old):
            raise Refused("snapshot_pin_changed_during_rollback")

    def replace_pin(self, path, value):
        fd, name = tempfile.mkstemp(prefix=".reviewed-snapshot-", dir=path.parent)
        try:
            with os.fdopen(fd, "wb") as stream:
                stream.write(value)
                stream.flush()
                os.fsync(stream.fileno())
            os.replace(name, 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(name):
                os.unlink(name)

    def execute(self):
        began = time.monotonic()
        self.preflight()
        self.capture()
        old_pin = SNAPSHOT.read_bytes()
        self.receipt["status"] = "upgrade_started"
        self.save()
        try:
            self.command(
                [
                    "helm",
                    "upgrade",
                    "coulomb-social",
                    str(self.root / "charts/coulomb-social"),
                    "-f",
                    str(self.root / self.artifact_name("values_file")),
                    "-n",
                    "coulomb-social",
                    "--atomic",
                    "--wait",
                    "--timeout",
                    "5m",
                ],
                timeout=660,
            )
            after = self.deployment()
            if (
                sha(canonical(after["spec"])) != self.plan["candidate_spec_sha256"]
                or after["status"].get("readyReplicas") != 1
            ):
                raise Refused("candidate_spec_or_readiness_differs")
            probe = self.probe("views")
            self.same_data(probe)
            for suffix in ("/healthz", "/", "/static/spaces/copy-content.js"):
                with urllib.request.urlopen(
                    "https://app.coulomb.social" + suffix, timeout=20
                ) as response:
                    body = response.read(1024**2)
                    if response.status != 200:
                        raise Refused("public_http_postflight_refused")
                    if (
                        suffix.endswith("copy-content.js")
                        and sha(body) != self.plan["clipboard_static_sha256"]
                    ):
                        raise Refused("served_static_differs")
            path = SNAPSHOT
            if (
                sha(path.read_bytes())
                != self.plan["installed_helper_hashes"][str(path)]
            ):
                raise Refused("snapshot_pin_drifted")
            self.replace_pin(
                path, (self.root / self.artifact_name("snapshot_file")).read_bytes()
            )
            self.command(
                [
                    "python3",
                    str(path),
                    "--destination",
                    "/tmp/csoc-post-image-server-review",
                ]
            )
            self.receipt.update(
                status="image_candidate_deployed_verified",
                postflight=probe,
                helm_revision=self.history()[-1]["revision"],
                snapshot_pin_refreshed=True,
                public_health_landing_static_verified=True,
                existing_global_constraint_preserved=True,
            )
        except Exception:
            self.receipt["rollback_requested"] = True
            try:
                self.command(
                    [
                        "helm",
                        "rollback",
                        "coulomb-social",
                        str(self.plan["baseline_revision"]),
                        "-n",
                        "coulomb-social",
                        "--wait",
                        "--timeout",
                        "5m",
                    ],
                    timeout=360,
                )
                self.restore_pin(old_pin)
                restored = self.deployment()
                if (sha(canonical(restored["spec"])) != self.plan["baseline_spec_sha256"]
                        or restored["status"].get("readyReplicas") != 1):
                    raise Refused("rollback_spec_or_readiness_differs")
                self.receipt["image_rollback_completed"] = True
            except Exception:  # noqa: BLE001 -- retain explicit unconfirmed rollback outcome
                self.receipt["image_rollback_completed"] = False
            raise
        self.receipt.update(
            elapsed_seconds=round(time.monotonic() - began, 3),
            completed_at=datetime.now(UTC).isoformat(),
            private_values_printed=False,
        )
        self.save()
        return self.receipt


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--capsule", type=Path, required=True)
    parser.add_argument("--expect-digest", required=True)
    parser.add_argument("--execute", action="store_true")
    parser.add_argument("--preflight", action="store_true")
    args = parser.parse_args()
    if not args.execute and not args.preflight:
        print(json.dumps({"status": "prepared", "live_changes": False}))
        return
    if args.execute and args.preflight:
        raise SystemExit("choose read-only preflight or attended execution")
    release = None
    try:
        release = Release(args.capsule, args.expect_digest, read_only=args.preflight)
        if args.preflight:
            release.preflight()
            print(
                json.dumps(
                    dict(
                        release.receipt,
                        status="live_release_preflight_verified",
                        live_changes=False,
                    )
                )
            )
            return
        print(json.dumps(release.execute()))
    except Exception as error:  # noqa: BLE001 -- private child diagnostics are suppressed
        receipt = release.receipt if release is not None else {
            "proposal_digest": args.expect_digest, "execution_started": False,
        }
        receipt.update(status="release_refused_or_failed", failure_type=type(error).__name__)
        if isinstance(error, Refused):
            receipt["failure_reason"] = str(error)
        if release is not None:
            release.save()
        print(json.dumps(receipt))
        raise SystemExit(1) from None


if __name__ == "__main__":
    main()