#!/usr/bin/env python3
"""Attended OpenBao Raft snapshot lane for the railiance01 reboot hold point.

RPF-WP-0015-T03, docs/railiance01-coordinated-reboot.md step 6.  Runs only as
the single, silent child of the contained ops-warden login::

    python3 scripts/openbao-attended-exec.py -- scripts/openbao-reboot-snapshot.py \
        --scenario-id <ID> --receipt-file /abs/new/file --private-dir /abs/0700/dir

Order: read-only ``bao status`` -> ``bao operator raft snapshot save`` into a
fresh 0700 temp dir -> ``bao status`` again (same cluster, index not behind) ->
age-encrypt to the declared offsite-lane recipient -> decrypt-stream verify
against the plaintext sha256 with the lane's escrow key (held only in memory)
-> upload ciphertext with ``If-None-Match: *`` to the governed Nextcloud lane
-> shred the temp dir (always) -> self-validate and write the receipt.

The process prints nothing: Warden fails closed on any child output.  The only
output is the 0600 receipt file (or a value-safe failure record) at
``--receipt-file``.  Only booleans, hashes, ids and status codes are recorded.
"""
from __future__ import annotations

import argparse
import hashlib
import hmac
import importlib.util
import json
import os
import re
import shutil
import signal
import stat
import subprocess
import sys
import tempfile
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[1]
SCRIPTS = ROOT / "scripts"
if str(SCRIPTS) not in sys.path:
    sys.path.insert(0, str(SCRIPTS))

RECIPIENT_SOURCE = ROOT / "lib" / "railiance-backup-common.sh"
LANE_PATH = "platform/workloads/railiance/backup/offsite-lane"
# Established offsite-lane folder used by every existing filesdrop upload.
UPLOAD_PREFIX = "openbao"  # founder-created railiance-backups/openbao/ (2026-10-14)
SOURCE = {"source_cluster": "railiance01", "source_namespace": "openbao", "source_pod": "openbao-0"}
OPERATOR = "railiance-platform"
SCENARIO_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9-]{2,95}$")
AGE_RECIPIENT_RE = re.compile(r"^age1[023456789acdefghjklmnpqrstuvwxyz]{58}$")
AGE_HEADER = b"age-encryption.org/v1\n"
GZIP_MAGIC = b"\x1f\x8b"
CHUNK = 1024 * 1024


def _load(name: str, filename: str) -> Any:
    spec = importlib.util.spec_from_file_location(name, SCRIPTS / filename)
    if not spec or not spec.loader:
        raise RuntimeError(f"cannot load {filename}")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


# Reuse, do not copy: contained login env + safe_run from the lease harness,
# the reviewed receipt validator, and the governed Nextcloud request helper.
LEASE = _load("audit_core_database_lease_recovery", "audit-core-database-lease-recovery.py")
PREFLIGHT = LEASE.PREFLIGHT
import migrate_nextcloud_backup_account as NEXTCLOUD  # noqa: E402


class SnapshotError(RuntimeError):
    pass


def fail(message: str) -> None:
    raise SnapshotError(message)


def stamp(value: datetime) -> str:
    return value.astimezone(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z")


def compact(value: datetime) -> str:
    return value.astimezone(UTC).strftime("%Y%m%dT%H%M%SZ")


def utc_now() -> datetime:
    return datetime.now(UTC)


def declared_recipient(path: Path = RECIPIENT_SOURCE) -> str:
    """The offsite lane's public recipient, as declared in the backup helper."""
    try:
        text = path.read_text(encoding="utf-8")
    except OSError:
        fail("offsite recipient declaration unavailable")
    found = re.findall(r'RAILIANCE_BACKUP_AGE_PUBLIC_KEY:=(age1[0-9a-z]+)\}', text)
    if len(found) != 1 or not AGE_RECIPIENT_RE.fullmatch(found[0]):
        fail("offsite recipient declaration is not exactly one age recipient")
    return found[0]


def inside_git_tree(path: Path) -> bool:
    for candidate in (path, *path.parents):
        if (candidate / ".git").exists():
            return True
    return False


def private_dir(value: str | None, environ: dict[str, str]) -> Path:
    raw = value or environ.get("HOME", "")
    path = Path(raw)
    if not raw or not path.is_absolute():
        fail("private dir must be an absolute path")
    try:
        info = path.lstat()
    except OSError:
        fail("private dir is missing")
    if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid() or info.st_mode & 0o077:
        fail("private dir must be an owner-only (0700) directory")
    if inside_git_tree(path.resolve()):
        fail("private dir must not be inside a Git tree")
    return path


def open_receipt(value: str) -> int:
    path = Path(value)
    if not path.is_absolute() or inside_git_tree(path.parent.resolve()):
        raise SnapshotError("receipt file must be an absolute path outside Git")
    return os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(CHUNK), b""):
            digest.update(chunk)
    return digest.hexdigest()


def shred(path: Path) -> None:
    """Overwrite then unlink every regular file below ``path``; remove the tree."""
    if not path.exists():
        return
    for item in sorted(path.rglob("*"), reverse=True):
        try:
            if item.is_file() and not item.is_symlink():
                size = item.stat().st_size
                with item.open("r+b") as handle:
                    remaining = size
                    while remaining > 0:
                        step = min(CHUNK, remaining)
                        handle.write(b"\0" * step)
                        remaining -= step
                    handle.flush()
                    os.fsync(handle.fileno())
                item.unlink()
            elif item.is_dir() and not item.is_symlink():
                item.rmdir()
            else:
                item.unlink()
        except OSError:
            pass
    shutil.rmtree(path, ignore_errors=True)


class Lane:
    """All side effects of one attended run; tests replace the boundaries."""

    def __init__(self, env: dict[str, str]) -> None:
        self.env = env
        self.tool_env = {"PATH": env.get("PATH", "/usr/bin:/bin"), "HOME": env.get("HOME", "/")}

    def bao(self, args: list[str], *, label: str) -> str:
        return LEASE.safe_run(["bao", *args], label=label, env=self.env).stdout

    def status(self) -> dict[str, Any]:
        try:
            status = json.loads(self.bao(["status", "-format=json"], label="read OpenBao status"))
            return {
                "initialized": status["initialized"] is True,
                "unsealed": status["sealed"] is False,
                "storage_type": status.get("storage_type"),
                "cluster_id": str(status["cluster_id"]),
                "raft_applied_index": int(status["raft_applied_index"]),
            }
        except (json.JSONDecodeError, KeyError, TypeError, ValueError):
            fail("OpenBao status is incomplete")

    def snapshot(self, target: Path) -> None:
        self.bao(["operator", "raft", "snapshot", "save", str(target)], label="save OpenBao Raft snapshot")

    def lane_identity(self) -> str:
        """The escrow key may be stored as a standard age identity file
        (``# created``/``# public key`` comments plus one key line); keep only that line."""
        raw = self.bao(["kv", "get", "-field=AGE_PRIVATE_KEY", LANE_PATH], label="read offsite lane AGE_PRIVATE_KEY")
        keys = []
        for line in raw.replace("\r\n", "\n").split("\n"):
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            keys.append(line)
        del raw
        if len(keys) != 1 or not keys[0].startswith("AGE-SECRET-KEY-1"):
            fail("offsite lane AGE_PRIVATE_KEY is unusable")
        return keys[0]

    def lane_field(self, field: str) -> str:
        value = self.bao(["kv", "get", f"-field={field}", LANE_PATH], label=f"read offsite lane {field}")
        value = value.strip()
        if not value or "\n" in value or "\r" in value:
            fail(f"offsite lane {field} is unusable")
        return value

    def tool(self, argv: list[str], *, label: str, payload: bytes = b"") -> bytes:
        try:
            result = subprocess.run(argv, input=payload, capture_output=True, env=self.tool_env, timeout=1800)
        except (OSError, subprocess.TimeoutExpired):
            fail(f"{label} unavailable")
        if result.returncode != 0:
            fail(f"{label} failed (exit {result.returncode})")
        return result.stdout

    def encrypt(self, recipients: list[str], source: Path, target: Path) -> None:
        argv = ["age"]
        for recipient in recipients:
            argv += ["-r", recipient]
        self.tool([*argv, "-o", str(target), str(source)], label="age encrypt")

    def recipient_of(self, identity: bytes) -> str:
        return self.tool(["age-keygen", "-y", "/dev/stdin"], label="derive escrow recipient", payload=identity).decode().strip()

    def decrypt_digest(self, identity: bytes, ciphertext: Path) -> str:
        """Stream-decrypt into a hash; plaintext never touches disk again."""
        digest = hashlib.sha256()
        try:
            proc = subprocess.Popen(
                ["age", "-d", "-i", "/dev/stdin", str(ciphertext)],
                stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, env=self.tool_env,
            )
        except OSError:
            fail("age decrypt unavailable")
        try:
            assert proc.stdin and proc.stdout
            proc.stdin.write(identity)
            proc.stdin.close()
            for chunk in iter(lambda: proc.stdout.read(CHUNK), b""):
                digest.update(chunk)
            code = proc.wait(timeout=1800)
        except (OSError, subprocess.TimeoutExpired):
            proc.kill()
            proc.wait()
            fail("age decrypt failed")
        finally:
            if proc.stdout:
                proc.stdout.close()
        if code != 0:
            fail(f"age decrypt failed (exit {code})")
        return digest.hexdigest()

    def upload(self, url: str, token: str, body: bytes) -> int:
        status, _ = NEXTCLOUD.request(
            url, "PUT", body, (token, ""),
            {"Content-Type": "application/octet-stream", "If-None-Match": "*",
             "Content-Length": str(len(body))},
        )
        return status


def run(
    args: argparse.Namespace, lane: Lane, receipt: dict[str, Any], now=utc_now
) -> tuple[dict[str, Any], dict[str, Any]]:
    recipient = declared_recipient()
    base = private_dir(args.private_dir, lane.env)
    stage = receipt.setdefault("stages", [])

    stage.append("status_before")
    before = lane.status()
    if not (before["initialized"] and before["unsealed"] and before["storage_type"] == "raft"):
        fail("OpenBao source is not an initialized, unsealed Raft node")
    if before["raft_applied_index"] <= 0:
        fail("OpenBao Raft applied index is not positive")

    created = now()
    work = Path(tempfile.mkdtemp(prefix="openbao-reboot-snapshot-", dir=base))
    try:
        os.chmod(work, 0o700)
        plain = work / "openbao.snap"
        cipher = work / "openbao.snap.age"
        stage.append("snapshot_save")
        lane.snapshot(plain)
        if not plain.is_file() or plain.stat().st_size == 0:
            fail("snapshot file was not written")
        os.chmod(plain, 0o600)
        with plain.open("rb") as handle:
            if handle.read(2) != GZIP_MAGIC:
                fail("snapshot is not a Raft snapshot archive")

        stage.append("status_after")
        after = lane.status()
        if after["cluster_id"] != before["cluster_id"] or after["raft_applied_index"] < before["raft_applied_index"]:
            fail("OpenBao identity or Raft index changed incoherently during the snapshot")
        if not (after["initialized"] and after["unsealed"]):
            fail("OpenBao source sealed during the snapshot")
        plain_sha = sha256_file(plain)

        stage.append("encrypt")
        recipients = [recipient, *([args.custodian_recipient] if args.custodian_recipient else [])]
        lane.encrypt(recipients, plain, cipher)
        os.chmod(cipher, 0o600)
        with cipher.open("rb") as handle:
            if handle.read(len(AGE_HEADER)) != AGE_HEADER:
                fail("ciphertext is not an age file")
        cipher_sha = sha256_file(cipher)
        if cipher_sha == plain_sha:
            fail("ciphertext equals plaintext")

        stage.append("verify")
        url = lane.lane_field("NC_WEBDAV_URL")
        token = lane.lane_field("NC_WEBDAV_TOKEN")
        identity = bytearray((lane.lane_identity() + "\n").encode())
        try:
            escrow_recipient = lane.recipient_of(bytes(identity))
            if not hmac.compare_digest(escrow_recipient, recipient):
                fail("escrow key does not match the declared offsite recipient")
            decrypted_sha = lane.decrypt_digest(bytes(identity), cipher)
        finally:
            identity[:] = b"\0" * len(identity)
        if not hmac.compare_digest(decrypted_sha, plain_sha):
            fail("decrypted snapshot hash does not match the plaintext")

        stage.append("upload")
        if url not in (
            f"{NEXTCLOUD.HOST}/public.php/dav/filesdrop/{token}",
            f"{NEXTCLOUD.HOST}/public.php/dav/files/{token}",
        ):
            fail("offsite lane URL does not match the governed provider contract")
        uploaded = now()
        name = f"openbao-raft-snapshot-{args.scenario_id}-{compact(uploaded)}.snap.age"
        status = lane.upload(f"{url}/{UPLOAD_PREFIX}/{name}", token, cipher.read_bytes())
        if status != 201:
            fail("offsite upload did not create a new object")
        # Only a 201 on a no-overwrite PUT names a location; the drop cannot read back.
        receipt["upload_http_status"] = status
        receipt["encrypted_location_ref"] = f"offhost-custody:{args.scenario_id}:{compact(uploaded)}"
    finally:
        stage.append("shred")
        shred(work)
        receipt["temp_dir_removed"] = not work.exists()

    if not receipt["temp_dir_removed"]:
        fail("private temp dir could not be removed")
    return {
        "receipt_version": 1,
        "receipt_id": f"{args.scenario_id}-SNAPSHOT-{compact(created)}",
        "created_at": stamp(created),
        "operator": OPERATOR,
        **SOURCE,
        "cluster_id": before["cluster_id"],
        # Lower bound: the snapshot contains at least the pre-save index.
        "raft_applied_index": before["raft_applied_index"],
        "snapshot_created": True,
        "source_initialized": True,
        "source_unsealed": True,
        "snapshot_sha256": f"sha256:{plain_sha}",
        "snapshot_encrypted": True,
        "encrypted_snapshot_sha256": f"sha256:{cipher_sha}",
        "encrypted_location_ref": f"offhost-custody:{args.scenario_id}:{compact(uploaded)}",
        "encrypted_copy_off_host": True,
        "off_host_evidence": "upload acknowledged (HTTP 201, If-None-Match: *); remote hash not read back",
        "upload_folder": UPLOAD_PREFIX,
        "recipients": [recipient, *([args.custodian_recipient] if args.custodian_recipient else [])],
        "custodian_escrow_ref": args.custodian_escrow_ref,
        "escrow_key_read_decision": "reviews/RPF-WP-0015-T03-founder-go-20261014.json#escrow_key_read",
        "encryption_verified": True,
        "hash_verified": True,
        "no_secret_material_recorded": True,
    }, after


class QuietParser(argparse.ArgumentParser):
    def error(self, message: str) -> None:  # never print usage
        raise SnapshotError("invalid arguments")


def parser() -> argparse.ArgumentParser:
    p = QuietParser(description=__doc__, add_help=False)
    p.add_argument("--scenario-id", required=True)
    p.add_argument("--receipt-file", required=True)
    p.add_argument("--private-dir")
    p.add_argument("--remote", default="railiance01", choices=["railiance01"])
    p.add_argument("--custodian-recipient", help="extra age recipient whose key is held outside OpenBao")
    p.add_argument("--custodian-escrow-ref", required=True,
                   help="value-safe reference to where a decryption key is held outside OpenBao")
    return p


def main(argv: list[str] | None = None, *, lane_factory=Lane, now=utc_now) -> int:
    try:
        args = parser().parse_args(argv)
        if not SCENARIO_RE.fullmatch(args.scenario_id):
            return 2
        # Always required: where a decryption key is held outside OpenBao (alone, when the
        # offline key is the offsite recipient; with --custodian-recipient otherwise).
        if not (args.custodian_escrow_ref or "").strip() or (
            args.custodian_recipient and not AGE_RECIPIENT_RE.fullmatch(args.custodian_recipient)
        ):
            return 2
        fd = open_receipt(args.receipt_file)
    except (SnapshotError, OSError):
        return 2
    record: dict[str, Any] = {"status": "failed", "scenario_id": args.scenario_id, "error": "internal_error"}
    result: dict[str, Any] | None = None
    def _abort(signum, _frame):  # warden timeout/abort: still run every finally/shred
        raise SnapshotError(f"interrupted by signal {signum}")

    signal.signal(signal.SIGTERM, _abort)
    signal.signal(signal.SIGHUP, _abort)
    try:
        os.umask(0o077)
        env = LEASE.contained_bao_env()
        receipt, live = run(args, lane_factory(env), record, now=now)
        # Self-check with the exact preflight validator before claiming success.
        check = Path(tempfile.mkdtemp(prefix="openbao-receipt-check-", dir=private_dir(args.private_dir, env)))
        try:
            probe = check / "receipt.json"
            probe.write_text(json.dumps(receipt), encoding="utf-8")
            PREFLIGHT.validate_snapshot_receipt(probe, live_openbao=live, now=now(), max_age_hours=1.0)
        finally:
            shutil.rmtree(check, ignore_errors=True)
        result = receipt
    except (SnapshotError, LEASE.ProcedureError, PREFLIGHT.PreflightError) as exc:
        record["error"] = str(exc)
    except Exception:  # noqa: BLE001 - value-safe: never record exception text
        record["error"] = "internal_error"
    finally:
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            json.dump(result if result is not None else record, handle, indent=2, sort_keys=False)
            handle.write("\n")
    return 0 if result is not None else 1


def silence() -> None:
    devnull = os.open(os.devnull, os.O_WRONLY)
    os.dup2(devnull, 1)
    os.dup2(devnull, 2)
    os.close(devnull)


if __name__ == "__main__":
    silence()
    raise SystemExit(main())