Record the first quiesced MAS evidence capture.

The facility stayed at one replica and the relay credential was not read.
Offsite publication still waits for a prefix-scoped publisher grant.

Assistant: grok
Assistant-Session: 01a198fa-fd27-7e41-a0a2-57355c3e4efd

scripts/mas_evidence_archive.py

@@ -0,0 +1,193 @@
+#!/usr/bin/env python3
+"""MAS evidence capture and restore rehearsal. No SMTP, no replica changes."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import re
+import sqlite3
+import stat
+import subprocess
+import tarfile
+import tempfile
+from contextlib import closing
+from pathlib import Path
+
+FILES = ("events.jsonl", "tickets.sqlite3", "tickets-outbox.sqlite3")
+PREFIX = "platform-pg/application-archives/mas/"
+MAX_BYTES = 64 * 1024**2
+
+
+class Refused(RuntimeError):
+    pass
+
+
+def digest(path):
+    with path.open("rb") as stream:
+        return hashlib.file_digest(stream, "sha256").hexdigest()
+
+
+def checkpoint(path, value):
+    fd, name = tempfile.mkstemp(prefix="receipt-", dir=path.parent)
+    with os.fdopen(fd, "w") as stream:
+        json.dump(value, stream, indent=2)
+        stream.flush()
+        os.fsync(stream.fileno())
+    os.chmod(name, 0o600)
+    Path(name).replace(path)
+
+
+def _regular(path):
+    info = path.lstat()
+    if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode):
+        raise Refused("evidence_file_required")
+    if info.st_size > MAX_BYTES or "mas-relay" in path.parts:
+        raise Refused("evidence_path_outside_contract")
+
+
+def _sqlite_fingerprint(path):
+    with closing(sqlite3.connect(path.absolute().as_uri() + "?mode=ro", uri=True)) as database:
+        database.execute("PRAGMA trusted_schema=OFF")
+        if database.execute("PRAGMA integrity_check").fetchall() != [("ok",)]:
+            raise Refused("sqlite_integrity_failed")
+        tables = sorted(
+            row[0]
+            for row in database.execute(
+                "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'"
+            )
+        )
+        result = {"tables": {}}
+        for name in tables:
+            if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", name):
+                raise Refused("unexpected_sqlite_table")
+            rows = [
+                json.dumps(row, separators=(",", ":"), default=lambda value: value.hex()).encode()
+                for row in database.execute(f'SELECT * FROM "{name}"')
+            ]
+            result["tables"][name] = {
+                "rows": len(rows),
+                "sha256": hashlib.sha256(b"\n".join(sorted(rows))).hexdigest(),
+            }
+        return result
+
+
+def _backup_sqlite(source, output):
+    _regular(source)
+    if output.exists() or output.is_symlink():
+        raise Refused("capture_destination_exists")
+    fd = os.open(output, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
+    os.close(fd)
+    with closing(sqlite3.connect(source.absolute().as_uri() + "?mode=ro", uri=True)) as src, closing(
+        sqlite3.connect(output)
+    ) as dst:
+        src.backup(dst, pages=64)
+    return {"sha256": digest(output), "bytes": output.stat().st_size, **_sqlite_fingerprint(output)}
+
+
+def capture(source, destination, *, captured_at, image, writers_quiesced, replicas):
+    """Copy the three evidence files. Caller must already have quiesced writers."""
+    if writers_quiesced is not True or replicas != {"mas-mx": 1, "mas-statute": 1}:
+        raise Refused("writers_not_quiesced")
+    destination.mkdir(mode=0o700, exist_ok=False)
+    files = {}
+    for name in FILES:
+        path = source / name
+        if name == "tickets-outbox.sqlite3" and not path.exists():
+            files[name] = {"present": False}
+            continue
+        if name.endswith(".sqlite3"):
+            files[name] = {"present": True, **_backup_sqlite(path, destination / name)}
+        else:
+            _regular(path)
+            target = destination / name
+            fd = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
+            with os.fdopen(fd, "wb") as out, path.open("rb") as src:
+                while block := src.read(1024**2):
+                    out.write(block)
+            files[name] = {"present": True, "sha256": digest(target), "bytes": target.stat().st_size}
+    receipt = {
+        "status": "captured",
+        "application": "mas",
+        "captured_at": captured_at,
+        "image": image,
+        "writers_quiesced": True,
+        "replicas": replicas,
+        "mail_sent": False,
+        "files": files,
+    }
+    checkpoint(destination / "receipt.json", receipt)
+    return receipt
+
+
+def seal(directory, stage, recipient):
+    stage.mkdir(mode=0o700, exist_ok=True)
+    encrypted = stage / "packet.tar.age"
+    if encrypted.exists() or encrypted.is_symlink():
+        raise Refused("partial_encryption_requires_review")
+    archive = stage / "packet.tar"
+    with archive.open("xb") as stream:
+        os.chmod(archive, 0o600)
+        with tarfile.open(fileobj=stream, mode="w") as packet:
+            for name in (*FILES, "receipt.json"):
+                path = directory / name
+                if path.exists():
+                    packet.add(path, arcname=name, recursive=False)
+    subprocess.run(["age", "-r", recipient, "-o", str(encrypted), str(archive)], check=True, timeout=120)
+    encrypted.chmod(0o600)
+    archive.unlink()
+    return {"ciphertext_sha256": digest(encrypted), "ciphertext_bytes": encrypted.stat().st_size}
+
+
+def rehearse(encrypted, identity, destination):
+    """Decrypt and compare hashes. Do not restore onto the facility or send mail."""
+    if destination.exists() or destination.is_symlink():
+        raise Refused("fresh_destination_required")
+    destination.mkdir(mode=0o700)
+    packet = destination / "packet.tar"
+    result = subprocess.run(
+        ["age", "-d", "-i", str(identity), "-o", str(packet), str(encrypted)],
+        capture_output=True,
+        timeout=120,
+    )
+    if result.returncode:
+        raise Refused("decryption_failed")
+    os.chmod(packet, 0o600)
+    with tarfile.open(packet, "r:") as archive:
+        entries = archive.getmembers()
+        names = {entry.name for entry in entries}
+        if (
+            "receipt.json" not in names
+            or any(not entry.isfile() or "/" in entry.name or entry.name.startswith(".") for entry in entries)
+        ):
+            raise Refused("invalid_packet_members")
+        for entry in entries:
+            target = destination / entry.name
+            with archive.extractfile(entry) as source, target.open("xb") as out:
+                os.chmod(target, 0o600)
+                out.write(source.read())
+    receipt = json.loads((destination / "receipt.json").read_text())
+    if receipt.get("application") != "mas" or receipt.get("mail_sent") is not False:
+        raise Refused("capture_receipt_rejected")
+    for name, meta in receipt["files"].items():
+        path = destination / name
+        if not meta.get("present"):
+            if path.exists():
+                raise Refused("unexpected_absent_file")
+            continue
+        if digest(path) != meta["sha256"] or path.stat().st_size != meta["bytes"]:
+            raise Refused("rehearsal_mismatch")
+        path.unlink()
+    packet.unlink()
+    (destination / "receipt.json").unlink()
+    proof = {
+        "status": "rehearsal_matched",
+        "application": "mas",
+        "ciphertext_sha256": digest(encrypted),
+        "mail_sent": False,
+        "production_restored": False,
+        "facility_scaled": False,
+    }
+    checkpoint(destination / "rehearsal.json", proof)
+    return proof

scripts/mas_evidence_capture.py

@@ -0,0 +1,166 @@
+#!/usr/bin/env python3
+"""Quiesce mas-mx in place, capture the evidence files, and rehearse locally.
+
+Does not change Deployment replicas, roll the image, delete the namespace,
+or read the TEM relay files. Offsite publication stays a separate grant.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import signal
+import subprocess
+from datetime import UTC, datetime
+from pathlib import Path
+
+import mas_evidence_archive as archive
+
+STATE = Path("/var/lib/mas-evidence-backup")
+
+
+def kubectl(*args):
+    return json.loads(
+        subprocess.check_output(["k3s", "kubectl", *args, "-o", "json"], text=True)
+    )
+
+
+def volume():
+    claim = kubectl("-n", "mas", "get", "pvc", "mas-evidence")
+    name = claim["spec"]["volumeName"]
+    body = kubectl("get", "pv", name)
+    local = body["spec"].get("local") or {}
+    host = body["spec"].get("hostPath") or {}
+    path = local.get("path") or host.get("path")
+    if not path or not Path(path).is_dir():
+        raise archive.Refused("evidence_volume_path_required")
+    return name, Path(path)
+
+
+def deployments():
+    found = {}
+    for name in ("mas-mx", "mas-statute"):
+        body = kubectl("-n", "mas", "get", "deploy", name)
+        image = body["spec"]["template"]["spec"]["containers"][0]["image"]
+        found[name] = {"replicas": body["spec"]["replicas"], "image": image}
+    return found
+
+
+def container_init_pid():
+    pods = kubectl("-n", "mas", "get", "pods", "-l", "app.kubernetes.io/name=mas-mx")
+    items = [item for item in pods["items"] if item["status"]["phase"] == "Running"]
+    if len(items) != 1:
+        raise archive.Refused("mas_mx_pod_required")
+    statuses = items[0]["status"]["containerStatuses"]
+    if len(statuses) != 1 or not statuses[0].get("containerID", "").startswith("containerd://"):
+        raise archive.Refused("mas_mx_container_required")
+    container = statuses[0]["containerID"].removeprefix("containerd://")
+    info = json.loads(
+        subprocess.check_output(["k3s", "crictl", "inspect", container], text=True)
+    )
+    return int(info["info"]["pid"])
+
+
+def namespace_pids(init_pid):
+    host = os.readlink("/proc/self/ns/pid")
+    target = os.readlink(f"/proc/{init_pid}/ns/pid")
+    if host == target:
+        raise archive.Refused("refusing_host_pid_namespace")
+    found = []
+    for entry in os.listdir("/proc"):
+        if not entry.isdigit():
+            continue
+        try:
+            if os.readlink(f"/proc/{entry}/ns/pid") == target:
+                found.append(int(entry))
+        except OSError:
+            continue
+    if init_pid not in found:
+        raise archive.Refused("container_process_missing")
+    return found
+
+
+def quiesced_copy(destination):
+    before = deployments()
+    volume_name, volume_path = volume()
+    if {name: item["replicas"] for name, item in before.items()} != {"mas-mx": 1, "mas-statute": 1}:
+        raise archive.Refused("facility_replica_drift")
+    if before["mas-mx"]["image"] != before["mas-statute"]["image"]:
+        raise archive.Refused("facility_image_drift")
+    pids = namespace_pids(container_init_pid())
+    stopped = []
+    try:
+        for pid in pids:
+            os.kill(pid, signal.SIGSTOP)
+            stopped.append(pid)
+        receipt = archive.capture(
+            volume_path,
+            destination,
+            captured_at=datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ"),
+            image=before["mas-mx"]["image"],
+            writers_quiesced=True,
+            replicas={"mas-mx": 1, "mas-statute": 1},
+        )
+    finally:
+        for pid in stopped:
+            try:
+                os.kill(pid, signal.SIGCONT)
+            except OSError:
+                pass
+    after = deployments()
+    if after != before:
+        raise archive.Refused("facility_changed_during_capture")
+    receipt["volume"] = volume_name
+    receipt["facility_scaled"] = False
+    receipt["relay_read"] = False
+    archive.checkpoint(destination / "receipt.json", receipt)
+    return receipt
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--execute", action="store_true")
+    parser.add_argument("--recipient")
+    args = parser.parse_args()
+    if not args.execute:
+        print(json.dumps({"status": "preview", "offsite": False, "facility_scaled": False}))
+        return
+    if os.geteuid() != 0:
+        raise archive.Refused("root_required")
+    if not isinstance(args.recipient, str) or not args.recipient.startswith("age1"):
+        raise archive.Refused("age_recipient_required")
+    STATE.mkdir(mode=0o700, exist_ok=True)
+    stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
+    destination = STATE / stamp
+    receipt = quiesced_copy(destination)
+    sealed = archive.seal(destination, destination, args.recipient)
+    for name in archive.FILES:
+        path = destination / name
+        if path.exists():
+            path.unlink()
+    public = {
+        "status": "sealed",
+        "application": "mas",
+        "captured_at": receipt["captured_at"],
+        "image": receipt["image"],
+        "volume": receipt["volume"],
+        "files": {
+            name: {"present": meta.get("present"), "bytes": meta.get("bytes")}
+            for name, meta in receipt["files"].items()
+        },
+        "ciphertext_sha256": sealed["ciphertext_sha256"],
+        "ciphertext_bytes": sealed["ciphertext_bytes"],
+        "mail_sent": False,
+        "production_restored": False,
+        "facility_scaled": False,
+        "relay_read": False,
+        "offsite_published": False,
+        "age_identity_on_node": False,
+    }
+    archive.checkpoint(STATE / "latest-public.json", public)
+    print(json.dumps(public))
+
+
+if __name__ == "__main__":
+    main()

scripts/primary_application_archive.py

@@ -23,6 +23,7 @@ BUCKET = "railiance-platform-pg-backup"
 PROFILES = {
     "coulomb-social": "platform-pg/application-archives/coulomb-social/",
     "email-connect": "platform-pg/application-archives/email-connect/",
+    "mas": "platform-pg/application-archives/mas/",
 }
 PART_SIZE = 64 * 1024**2
 MAX_BYTES = 20 * 1024**3