#!/usr/bin/env python3
"""Value-safe preflight for RAILIANCE-WP-0024 recovery exercises.

This helper is deliberately read-only.  It never reads Kubernetes Secret data,
database usernames/passwords, OpenBao lease payloads, or unseal material.  Live
lease revocation and a host reboot remain separate, attended actions.
"""

from __future__ import annotations

import argparse
import json
import re
import shlex
import subprocess
import sys
from datetime import UTC, datetime
from pathlib import Path
from typing import Any


EXPECTED_IMAGE = (
    "forgejo.coulomb.social/coulomb/audit-core@"
    "sha256:c2fe39a0185b99be3fc0cb14d2de69772b8e66e20490097c9d11d90cc39719a6"
)
EXPECTED_DB_KEYS = {"username", "password", "host", "port", "dbname"}


class PreflightError(RuntimeError):
    pass


class Remote:
    def __init__(self, host: str) -> None:
        self.host = host

    def run(self, command: list[str], *, label: str) -> str:
        remote_command = " ".join(shlex.quote(part) for part in command)
        completed = subprocess.run(
            ["ssh", "-o", "BatchMode=yes", self.host, remote_command],
            text=True,
            capture_output=True,
            check=False,
        )
        if completed.returncode != 0:
            # Remote output can include provider or application detail that is
            # inappropriate for durable evidence.  Report only the failed step.
            raise PreflightError(f"{label} failed (exit {completed.returncode})")
        return completed.stdout.strip()

    def kubectl(self, args: list[str], *, label: str) -> str:
        return self.run(["kubectl", *args], label=label)

    def kubectl_json(self, args: list[str], *, label: str) -> dict[str, Any]:
        raw = self.kubectl([*args, "-o", "json"], label=label)
        try:
            value = json.loads(raw)
        except json.JSONDecodeError as exc:
            raise PreflightError(f"{label} returned invalid JSON") from exc
        if not isinstance(value, dict):
            raise PreflightError(f"{label} did not return an object")
        return value


def condition(resource: dict[str, Any], kind: str) -> str | None:
    for item in resource.get("status", {}).get("conditions", []):
        if item.get("type") == kind:
            return item.get("status")
    return None


def rfc3339(value: str) -> datetime:
    return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(UTC)


def latest_completed_backup(backups: dict[str, Any], now: datetime) -> dict[str, Any]:
    completed = [
        item
        for item in backups.get("items", [])
        if item.get("status", {}).get("phase") == "completed"
        and item.get("status", {}).get("stoppedAt")
    ]
    if not completed:
        raise PreflightError("no completed platform-pg backup is visible")
    latest = max(completed, key=lambda item: item["status"]["stoppedAt"])
    stopped = rfc3339(latest["status"]["stoppedAt"])
    return {
        "name": latest["metadata"]["name"],
        "stopped_at": stopped.isoformat().replace("+00:00", "Z"),
        "age_hours": round((now - stopped).total_seconds() / 3600, 2),
        "method": latest.get("spec", {}).get("method"),
    }


def resource_condition(resource: dict[str, Any]) -> dict[str, Any]:
    conditions = resource.get("status", {}).get("conditions", [])
    current = conditions[0] if conditions else {}
    return {
        "ready": current.get("status") == "True",
        "reason": current.get("reason"),
    }


def secret_key_names(description: str) -> set[str]:
    """Extract names from ``kubectl describe secret`` without reading data."""
    keys: set[str] = set()
    for line in description.splitlines():
        match = re.match(r"^([^\s:]+):\s+\d+ bytes$", line.strip())
        if match:
            keys.add(match.group(1))
    return keys


def common_state(remote: Remote, now: datetime) -> dict[str, Any]:
    node_list = remote.kubectl_json(["get", "nodes"], label="read node state")
    nodes = node_list.get("items", [])
    if len(nodes) != 1:
        raise PreflightError(f"expected one railiance01 node, observed {len(nodes)}")
    node = nodes[0]

    cluster = remote.kubectl_json(
        ["-n", "databases", "get", "cluster", "platform-pg"],
        label="read platform-pg state",
    )
    deployment = remote.kubectl_json(
        ["-n", "audit-core", "get", "deployment", "audit-core"],
        label="read audit-core deployment",
    )
    pod_list = remote.kubectl_json(
        ["-n", "audit-core", "get", "pods", "-l", "app.kubernetes.io/name=audit-core"],
        label="read audit-core pod metadata",
    )
    pods = pod_list.get("items", [])
    if len(pods) != 1:
        raise PreflightError(f"expected one audit-core pod, observed {len(pods)}")
    pod = pods[0]
    container_statuses = pod.get("status", {}).get("containerStatuses", [])
    restart_count = sum(int(item.get("restartCount", 0)) for item in container_statuses)

    stores: dict[str, Any] = {}
    for name in ("openbao-audit-core", "openbao-audit-core-database"):
        resource = remote.kubectl_json(
            ["get", "clustersecretstore", name], label=f"read {name} state"
        )
        stores[name] = resource_condition(resource)

    external_secrets: dict[str, Any] = {}
    for name in ("audit-core-database", "audit-core-senders"):
        resource = remote.kubectl_json(
            ["-n", "audit-core", "get", "externalsecret", name],
            label=f"read {name} state",
        )
        external_secrets[name] = resource_condition(resource)

    bao_raw = remote.kubectl(
        ["-n", "openbao", "exec", "openbao-0", "--", "bao", "status", "-format=json"],
        label="read OpenBao seal state",
    )
    try:
        bao_status = json.loads(bao_raw)
    except json.JSONDecodeError as exc:
        raise PreflightError("OpenBao seal state returned invalid JSON") from exc
    image = deployment["spec"]["template"]["spec"]["containers"][0]["image"]
    state = {
        "captured_at": now.replace(microsecond=0).isoformat().replace("+00:00", "Z"),
        "node": {
            "name": node["metadata"]["name"],
            "ready": condition(node, "Ready") == "True",
            "k3s_active": remote.run(
                ["systemctl", "is-active", "k3s"], label="read k3s service state"
            ) == "active",
            "uptime_seconds": float(
                remote.run(["cat", "/proc/uptime"], label="read host uptime").split()[0]
            ),
        },
        "platform_pg": {
            "ready_instances": cluster.get("status", {}).get("readyInstances", 0),
            "instances": cluster.get("spec", {}).get("instances", 0),
            "phase": cluster.get("status", {}).get("phase"),
            "continuous_archiving": condition(cluster, "ContinuousArchiving") == "True",
            "last_backup_succeeded": condition(cluster, "LastBackupSucceeded") == "True",
        },
        "openbao": {
            "initialized": bool(bao_status.get("initialized")),
            "sealed": bool(bao_status.get("sealed")),
            "seal_type": bao_status.get("type"),
            "threshold": bao_status.get("t"),
            "shares": bao_status.get("n"),
        },
        "external_secret_stores": stores,
        "external_secrets": external_secrets,
        "audit_core": {
            "image": image,
            "image_matches_reviewed_digest": image == EXPECTED_IMAGE,
            "ready_replicas": deployment.get("status", {}).get("readyReplicas", 0),
            "replicas": deployment.get("spec", {}).get("replicas", 0),
            "pod_uid": pod["metadata"]["uid"],
            "restart_count": restart_count,
        },
        "secret_values_observed": False,
    }
    return state


def automated_common_pass(state: dict[str, Any]) -> bool:
    return all(
        (
            state["node"]["ready"],
            state["node"]["k3s_active"],
            state["platform_pg"]["ready_instances"] == state["platform_pg"]["instances"] == 1,
            state["platform_pg"]["continuous_archiving"],
            state["platform_pg"]["last_backup_succeeded"],
            state["openbao"]["initialized"],
            not state["openbao"]["sealed"],
            all(item["ready"] for item in state["external_secret_stores"].values()),
            all(item["ready"] for item in state["external_secrets"].values()),
            state["audit_core"]["image_matches_reviewed_digest"],
            state["audit_core"]["ready_replicas"] == state["audit_core"]["replicas"] == 1,
        )
    )


def database_lease_preflight(remote: Remote, args: argparse.Namespace) -> dict[str, Any]:
    now = datetime.now(UTC)
    state = common_state(remote, now)
    secret_rv = remote.kubectl(
        ["-n", "audit-core", "get", "secret", "audit-core-database", "-o", "jsonpath={.metadata.resourceVersion}"],
        label="read database Secret metadata",
    )
    description = remote.kubectl(
        ["-n", "audit-core", "describe", "secret", "audit-core-database"],
        label="read database Secret key names",
    )
    keys = secret_key_names(description)
    health_code = remote.kubectl(
        ["-n", "audit-core", "exec", "deploy/audit-core", "--", "python", "-c", 'import urllib.request; print(urllib.request.urlopen("http://127.0.0.1:8080/healthz", timeout=3).status)'],
        label="probe audit-core health",
    )
    ready_code = remote.kubectl(
        ["-n", "audit-core", "exec", "deploy/audit-core", "--", "python", "-c", 'import urllib.request; print(urllib.request.urlopen("http://127.0.0.1:8080/readyz", timeout=3).status)'],
        label="probe audit-core readiness",
    )
    gates = {
        "approved_window_id_recorded": bool(args.approved_window_id),
        "audit_core_owner_acknowledged": args.audit_core_owner_ack,
        "rapp_postgres_owner_acknowledged": args.rapp_postgres_owner_ack,
        "approved_synthetic_load_contract": bool(args.synthetic_load_id),
        "attended_abort_operator_named": bool(args.abort_operator),
    }
    automated = automated_common_pass(state) and keys == EXPECTED_DB_KEYS and health_code == "200" and ready_code == "200"
    return {
        "procedure": "audit-core-database-lease-recovery",
        "preflight_only": True,
        "automated_checks_passed": automated,
        "ready_for_live_execution": automated and all(gates.values()),
        "operator_gates": gates,
        "baseline": {
            "database_secret_resource_version": secret_rv,
            "database_secret_keys": sorted(keys),
            "health_status": int(health_code),
            "readiness_status": int(ready_code),
            "pod_uid": state["audit_core"]["pod_uid"],
            "restart_count": state["audit_core"]["restart_count"],
        },
        "state": state,
        "secret_values_observed": False,
    }


def reboot_preflight(remote: Remote, args: argparse.Namespace) -> dict[str, Any]:
    now = datetime.now(UTC)
    state = common_state(remote, now)
    backups = remote.kubectl_json(
        ["-n", "databases", "get", "backup", "-l", "cnpg.io/cluster=platform-pg"],
        label="read platform-pg backup metadata",
    )
    latest = latest_completed_backup(backups, now)
    snapshot_evidence = Path(args.openbao_snapshot_evidence).resolve() if args.openbao_snapshot_evidence else None
    gates = {
        "approved_window_id_recorded": bool(args.approved_window_id),
        "openbao_snapshot_evidence_present": bool(snapshot_evidence and snapshot_evidence.is_file()),
        "unseal_quorum_attested": args.unseal_quorum_attested,
        "provider_console_access_attested": args.provider_console_attested,
        "host_cluster_platform_database_audit_owners_acknowledged": args.all_owners_ack,
        "attended_abort_operator_named": bool(args.abort_operator),
    }
    automated = automated_common_pass(state) and latest["age_hours"] <= args.max_backup_age_hours
    return {
        "procedure": "railiance01-coordinated-reboot",
        "preflight_only": True,
        "automated_checks_passed": automated,
        "ready_for_live_execution": automated and all(gates.values()),
        "operator_gates": gates,
        "latest_platform_pg_backup": latest,
        "max_backup_age_hours": args.max_backup_age_hours,
        "state": state,
        "secret_values_observed": False,
    }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("procedure", choices=["database-lease", "node-reboot"])
    parser.add_argument("--remote", default="railiance01")
    parser.add_argument("--approved-window-id")
    parser.add_argument("--abort-operator")
    parser.add_argument("--audit-core-owner-ack", action="store_true")
    parser.add_argument("--rapp-postgres-owner-ack", action="store_true")
    parser.add_argument("--synthetic-load-id")
    parser.add_argument("--openbao-snapshot-evidence")
    parser.add_argument("--unseal-quorum-attested", action="store_true")
    parser.add_argument("--provider-console-attested", action="store_true")
    parser.add_argument("--all-owners-ack", action="store_true")
    parser.add_argument("--max-backup-age-hours", type=float, default=26.0)
    args = parser.parse_args()
    try:
        remote = Remote(args.remote)
        result = (
            database_lease_preflight(remote, args)
            if args.procedure == "database-lease"
            else reboot_preflight(remote, args)
        )
    except (OSError, ValueError, KeyError, PreflightError) as exc:
        print(f"recovery preflight failed: {exc}", file=sys.stderr)
        return 1
    print(json.dumps(result, indent=2, sort_keys=True))
    return 0


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