coulomb-social/scripts/observe-source-date-target.py

#!/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",
            "production_access": True,
            "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)