coulomb-social/scripts/expire-backup-mirror.py

#!/usr/bin/env python3
"""Prepare, verify or perform an attended exact-version mirror expiry.

Preparation/preview is offline; verification uses only the independent reader.
Execution requires a separately admitted expiry pair and founder confirmation.
This tool does not grant authority, provision credentials or activate a schedule.
"""

from __future__ import annotations

import argparse
import fcntl
import hashlib
import importlib.util
import json
import os
import sys
import tempfile
from datetime import UTC, datetime
from pathlib import Path


def module(name, filename):
    spec = importlib.util.spec_from_file_location(name, Path(__file__).with_name(filename))
    loaded = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(loaded)
    return loaded


observer = module("expiry_observer", "observe-backup-retention.py")
reviewer = module("expiry_reviewer", "review-backup-retention.py")
health = observer.health
Refused = health.Refused
MIRROR = health.MIRROR_BUCKET
KIT = observer.KIT
SOURCE_FILES = (
    "expire-backup-mirror.py",
    "observe-backup-retention.py",
    "review-backup-retention.py",
    "mirror-paired-backup.py",
    "check-paired-backup-health.py",
    "backup_capacity.py",
)


def canonical(value):
    return observer.worker.canonical(value)


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


def source_hashes():
    return {name: sha(Path(__file__).with_name(name).read_bytes()) for name in SOURCE_FILES}


def key(item):
    return item["bucket"], item["object_key"], item["version_id"]


def version_map(observations):
    result = {}
    for bucket in observations:
        if bucket["bucket"] not in (MIRROR, KIT):
            continue
        if bucket.get("delete_markers"):
            raise Refused("delete_markers_require_separate_review")
        for item in bucket["versions"]:
            identity = bucket["bucket"], item["Key"], item["VersionId"]
            observer.version_id(item["VersionId"])
            if identity in result:
                raise Refused("duplicate_version")
            result[identity] = item["Size"]
    return result


def derive(inventory, recovery, observation, platform_scripts, created_at):
    owner = reviewer.load_owner(platform_scripts)
    selection = owner.plan("coulomb-social", inventory, recovery, created_at)
    review = reviewer.review(inventory, recovery, observation, selection, created_at)
    if not review["generation_selection_valid"] or review["blockers"]:
        raise Refused("valid_generation_selection_required")
    targets = []
    for candidate in review["expiration_candidates"]:
        targets.append(
            {
                "bucket": MIRROR,
                "object_key": candidate["object_key"],
                "version_id": candidate["version_id"],
                "sha256": candidate["ciphertext_sha256"],
                "bytes": candidate["ciphertext_bytes"],
                "role": "ciphertext",
            }
        )
        receipt = candidate["associated_receipt"]
        targets.append(
            {
                "bucket": MIRROR,
                "object_key": receipt["receipt_object_key"],
                "version_id": receipt["receipt_version_id"],
                "sha256": receipt["receipt_sha256"],
                "bytes": receipt["receipt_bytes"],
                "role": "receipt",
            }
        )
    ids = [key(t) for t in targets]
    if len(ids) != len(set(ids)):
        raise Refused("duplicate_expiry_target")
    keep = {key(t) for t in review["keep"]}
    if keep.intersection(ids) or any(t[0] != MIRROR for t in ids):
        raise Refused("expiry_overlaps_keep_or_other_bucket")
    baseline = version_map(observation["observations"])
    if any(baseline.get(key(t)) != t["bytes"] for t in targets):
        raise Refused("expiry_target_not_in_complete_inventory")
    return {"keep": review["keep"], "targets": targets}


def prepare(inventory, recovery, observation, platform_scripts, now, runtime_manifest):
    derived = derive(inventory, recovery, observation, platform_scripts, now)
    return {
        "schema": "coulomb.mirror-expiry-proposal.v1",
        "application": "coulomb-social",
        "project_id": health.PROJECT,
        "created_at": now.isoformat(),
        "sources": source_hashes(),
        "owner_sources": reviewer.OWNER_FILES,
        "runtime_manifest_sha256": sha(runtime_manifest.read_bytes()),
        "inventory": inventory,
        "recovery": recovery,
        "observation": observation,
        **derived,
        "requires_separate_expiry_admission": True,
        "requires_founder_attended_execution": True,
    }


def validate(proposal, expected, platform_scripts, now, runtime_manifest):
    if sha(canonical(proposal)) != expected:
        raise Refused("proposal_digest_differs")
    if (
        proposal.get("schema") != "coulomb.mirror-expiry-proposal.v1"
        or proposal.get("application") != "coulomb-social"
        or proposal.get("project_id") != health.PROJECT
        or proposal.get("sources") != source_hashes()
        or proposal.get("owner_sources") != reviewer.OWNER_FILES
        or proposal.get("runtime_manifest_sha256") != sha(runtime_manifest.read_bytes())
        or proposal.get("requires_separate_expiry_admission") is not True
        or proposal.get("requires_founder_attended_execution") is not True
    ):
        raise Refused("reviewed_expiry_sources_and_scope_required")
    created = health.timestamp(proposal["created_at"])
    if created > now:
        raise Refused("future_proposal")
    # Validate original selection at proposal time. Every run, including resume,
    # independently rereads current custody and the entire frozen version set.
    derived = derive(
        proposal["inventory"],
        proposal["recovery"],
        proposal["observation"],
        platform_scripts,
        created,
    )
    if any(proposal[k] != v for k, v in derived.items()):
        raise Refused("proposal_selection_differs_from_owner")


def new_journal(digest, count):
    return {
        "schema": "coulomb.mirror-expiry-attempt.v1",
        "proposal_digest": digest,
        "steps": ["pending"] * count,
        "status": "prepared",
    }


def validate_journal(journal, digest, count):
    if (
        journal.get("schema") != "coulomb.mirror-expiry-attempt.v1"
        or journal.get("proposal_digest") != digest
        or len(journal.get("steps", [])) != count
        or journal.get("status") not in {"prepared", "running", "completed"}
    ):
        raise Refused("journal_binding_differs")
    tail = False
    for state in journal["steps"]:
        if state not in {"pending", "intent", "absent_verified"}:
            raise Refused("invalid_journal_step")
        if state == "absent_verified":
            if tail:
                raise Refused("out_of_order_expiry_journal")
        elif state == "intent":
            if tail:
                raise Refused("out_of_order_expiry_journal")
            tail = True
        else:
            tail = True
    if journal["status"] == "completed" and any(s != "absent_verified" for s in journal["steps"]):
        raise Refused("incomplete_terminal_journal")


def exact_head(reader, target):
    value = reader.head_object(
        Bucket=target["bucket"], Key=target["object_key"], VersionId=target["version_id"]
    )
    if (
        value.get("VersionId") != target["version_id"]
        or value.get("ContentLength") != target["bytes"]
        or value.get("Metadata", {}).get("sha256") != target["sha256"]
        or value.get("DeleteMarker")
    ):
        raise Refused("retained_version_head_differs")


def confirm_absent(reader, target):
    try:
        reader.head_object(
            Bucket=target["bucket"], Key=target["object_key"], VersionId=target["version_id"]
        )
    except Exception as error:
        if observer.status(error) == 404:
            return
        raise
    raise Refused("expired_version_still_present")


def verify(reader, proposal, journal):
    """Fresh native reads; permit disappearance only after durable own intent."""
    digest = sha(canonical(proposal))
    validate_journal(journal, digest, len(proposal["targets"]))
    current = [observer.observe_bucket(reader, bucket) for bucket in (MIRROR, KIT)]
    for bucket in current:
        rules = bucket["lifecycle"]["Rules"]
        if (bucket["bucket"] == KIT and rules) or any(r.get("Status") == "Enabled" for r in rules):
            raise Refused("generation_or_protected_custody_lifecycle_changed")
    baseline = version_map(proposal["observation"]["observations"])
    present = version_map(current)
    allowed_absent = {
        key(t)
        for t, state in zip(proposal["targets"], journal["steps"])
        if state in {"intent", "absent_verified"}
    }
    missing = baseline.keys() - present.keys()
    if (
        present.keys() - baseline.keys()
        or not missing <= allowed_absent
        or any(baseline[k] != size for k, size in present.items())
    ):
        raise Refused("provider_inventory_changed_outside_own_expiry")
    for target, state in zip(proposal["targets"], journal["steps"]):
        if key(target) in missing:
            confirm_absent(reader, target)
        elif state == "absent_verified":
            raise Refused("expired_version_reappeared")
    heads = [h for h in proposal["observation"]["version_heads"] if h["bucket"] in (MIRROR, KIT)]
    for target in heads:
        if key(target) in present:
            exact_head(reader, target)
    # Independently stream/hash every retained known mirror object, including
    # the recovered anchor. Kit identity is HEAD-checked against its prior full
    # provider-only recovery proof; the large kit is not restored on each expiry.
    kept = {key(t) for t in proposal["keep"]}
    verified_bytes = 0
    for target in heads:
        if key(target) not in present or (target["bucket"] != MIRROR):
            continue
        # Also validate candidates/sidecars before touching them. No plaintext.
        value = reader.get_object(
            Bucket=MIRROR, Key=target["object_key"], VersionId=target["version_id"]
        )
        total = 0
        hasher = hashlib.sha256()
        with value["Body"] as stream:
            while chunk := stream.read(min(1024 * 1024, target["bytes"] + 1 - total)):
                total += len(chunk)
                if total > target["bytes"]:
                    raise Refused("retained_version_body_oversized")
                hasher.update(chunk)
        if (
            value.get("VersionId") != target["version_id"]
            or value.get("ContentLength") != target["bytes"]
            or total != target["bytes"]
            or hasher.hexdigest() != target["sha256"]
        ):
            raise Refused("retained_version_body_differs")
        verified_bytes += total
    if not kept <= present.keys():
        raise Refused("kept_generation_missing")
    return {
        "absent": missing,
        "mirror_bytes_verified": verified_bytes,
        "protected_versions_head_verified": True,
    }


def execute(reader, expiry, proposal, journal, save):
    """Exact single-version deletes; caller is never used for reads or fallback."""
    digest = sha(canonical(proposal))
    validate_journal(journal, digest, len(proposal["targets"]))
    state = verify(reader, proposal, journal)
    for index, target in enumerate(proposal["targets"]):
        if key(target) in state["absent"]:
            journal["steps"][index] = "absent_verified"
            save()
            continue
        # Recheck custody and the complete keep set immediately before each action.
        state = verify(reader, proposal, journal)
        if key(target) in state["absent"]:
            journal["steps"][index] = "absent_verified"
            save()
            continue
        journal["steps"][index] = "intent"
        journal["status"] = "running"
        save()  # Must be durable before any destructive request, including retries.
        value = expiry.delete_object(
            Bucket=MIRROR, Key=target["object_key"], VersionId=target["version_id"]
        )
        if (
            value.get("ResponseMetadata", {}).get("HTTPStatusCode") != 204
            or value.get("DeleteMarker") is True
        ):
            raise Refused("uncertain_exact_version_delete")
        state = verify(reader, proposal, journal)
        if key(target) not in state["absent"]:
            raise Refused("deletion_not_independently_verified")
        journal["steps"][index] = "absent_verified"
        save()
    verify(reader, proposal, journal)
    journal.update(status="completed", completed_at=datetime.now(UTC).isoformat())
    save()
    return {
        "status": "exact_mirror_versions_expired_verified",
        "proposal_digest": digest,
        "versions_absent_verified": len(proposal["targets"]),
        "kept_generations": len(proposal["keep"]),
        "protected_custody_verified": True,
        "primary_or_kit_deleted": False,
        "unknown_objects_preserved": True,
    }


def checkpoint(path, value):
    if path.exists() or path.is_symlink():
        health.regular(path)
    fd, temporary = tempfile.mkstemp(prefix="expiry-journal-", 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)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    modes = parser.add_mutually_exclusive_group()
    modes.add_argument("--prepare", action="store_true")
    modes.add_argument("--verify", action="store_true")
    modes.add_argument("--execute", action="store_true")
    for name in (
        "inventory",
        "recovery",
        "observation",
        "platform-scripts",
        "runtime-manifest",
        "proposal",
        "output",
        "credentials-directory",
        "state-dir",
    ):
        parser.add_argument("--" + name, type=Path)
    parser.add_argument("--expect-digest")
    args = parser.parse_args()
    if not any((args.prepare, args.verify, args.execute)):
        print(
            json.dumps(
                {"status": "preview", "provider_contacted": False, "deletion_admitted": False}
            )
        )
        return
    if args.prepare:
        proposal = prepare(
            *[
                json.loads(getattr(args, name).read_bytes())
                for name in ("inventory", "recovery", "observation")
            ],
            args.platform_scripts,
            datetime.now(UTC),
            args.runtime_manifest,
        )
        observer.worker.private_directory(args.output.parent)
        if any((p / ".git").exists() for p in [args.output.parent, *args.output.parents]):
            raise Refused("private_proposal_outside_git_required")
        fd = os.open(args.output, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
        with os.fdopen(fd, "wb") as stream:
            stream.write(canonical(proposal))
        print(
            json.dumps(
                {
                    "status": "proposal_prepared",
                    "proposal_digest": sha(canonical(proposal)),
                    "candidate_points": len(proposal["targets"]) // 2,
                    "provider_contacted": False,
                    "deletion_admitted": False,
                }
            )
        )
        return
    proposal = health.read_json(args.proposal)
    validate(
        proposal,
        args.expect_digest,
        args.platform_scripts,
        datetime.now(UTC),
        args.runtime_manifest,
    )
    observer.worker.load_runtime(args.platform_scripts, args.runtime_manifest)
    from application_recovery_identities import private_directory, private_json
    from application_recovery_resources import s3

    private_directory(args.credentials_directory)
    reader_pair = private_json(args.credentials_directory / "reader-pair.json")
    digest = sha(canonical(proposal))
    if args.verify:
        reader = s3(reader_pair, health.PROJECT)
        try:
            result = verify(reader, proposal, new_journal(digest, len(proposal["targets"])))
            print(
                json.dumps(
                    {
                        "status": "read_only_expiry_preflight_verified",
                        "proposal_digest": digest,
                        "provider_mutations": False,
                        "mirror_bytes_verified": result["mirror_bytes_verified"],
                        "deletion_admitted": False,
                    }
                )
            )
        finally:
            reader.close()
        return
    if not sys.stdin.isatty():
        raise Refused("founder_attended_terminal_required")
    private_directory(args.state_dir)
    # This new pair must be provisioned/admitted separately. There is no fallback
    # to the publisher, reader or custodian when the dedicated file is absent.
    expiry_pair = private_json(args.credentials_directory / "expiry-pair.json")
    if expiry_pair["AWS_ACCESS_KEY_ID"] == reader_pair["AWS_ACCESS_KEY_ID"]:
        raise Refused("separate_expiry_caller_required")
    expected = "EXPIRE-MIRROR " + digest
    if input("Type " + expected + " only after separate expiry admission: ").strip() != expected:
        raise Refused("not_confirmed")
    fd = os.open(args.state_dir / "expiry.lock", os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
    with os.fdopen(fd, "w") as lock:
        health.regular(args.state_dir / "expiry.lock")
        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
        path = args.state_dir / (digest + ".json")
        journal = (
            health.read_json(path)
            if path.exists() or path.is_symlink()
            else new_journal(digest, len(proposal["targets"]))
        )
        validate_journal(journal, digest, len(proposal["targets"]))
        checkpoint(path, journal)
        reader = s3(reader_pair, health.PROJECT)
        expiry = None
        try:
            expiry = s3(expiry_pair, health.PROJECT)
            print(
                json.dumps(
                    execute(reader, expiry, proposal, journal, lambda: checkpoint(path, journal))
                )
            )
        finally:
            reader.close()
            if expiry is not None:
                expiry.close()


if __name__ == "__main__":
    try:
        main()
    except Exception as error:
        print(
            json.dumps(
                {
                    "status": "expiry_requires_review",
                    "failure_type": type(error).__name__,
                    "journal_preserved": True,
                }
            )
        )
        raise SystemExit(1)