#!/usr/bin/env python3
"""Read-only K3s control-plane contract helper for Phase 2 enrollment."""
from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
import shutil
import stat
import subprocess
from pathlib import Path
from typing import Any, Mapping, Sequence

HEX64 = re.compile(r"^[0-9a-f]{64}$")


class InspectError(RuntimeError):
    pass


def rooted(root: Path, absolute: str) -> Path:
    return root / absolute.lstrip("/") if root != Path("/") else Path(absolute)


def read_json_regular(root: Path, absolute: str) -> dict[str, Any]:
    path = rooted(root, absolute)
    if path.is_symlink():
        raise InspectError(f"refusing symlink: {absolute}")
    try:
        info = path.stat()
    except FileNotFoundError as exc:
        raise InspectError(f"required contract is missing: {absolute}") from exc
    if not stat.S_ISREG(info.st_mode):
        raise InspectError(f"required regular contract: {absolute}")
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        raise InspectError(f"invalid JSON in {absolute}: {exc}") from exc
    if not isinstance(value, dict):
        raise InspectError(f"contract must be an object: {absolute}")
    return value


def discover_launcher(version_lock: Mapping[str, Any]) -> Path:
    launcher = version_lock.get("launcher") if isinstance(version_lock.get("launcher"), Mapping) else {}
    candidates = [str(launcher.get("invocation_path") or ""), "/usr/local/bin/k3s", "/usr/bin/k3s"]
    for candidate in candidates:
        if not candidate:
            continue
        path = Path(candidate)
        try:
            info = path.stat()
        except OSError:
            continue
        if not stat.S_ISREG(info.st_mode) or not os.access(path, os.X_OK):
            continue
        if info.st_uid != 0 or stat.S_IMODE(info.st_mode) & 0o022:
            continue
        return path
    raise InspectError("cannot locate a secure operator-facing K3s launcher")


def run(argv: Sequence[str], timeout: int = 30) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        list(argv),
        check=False,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        timeout=timeout,
        env={"PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin", "LANG": "C"},
    )


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


def inspect(root: Path) -> dict[str, Any]:
    control = read_json_regular(root, "/etc/rancher/k3s/overdeck/control-plane.json")
    lock = read_json_regular(root, "/etc/rancher/k3s/overdeck/version-lock.json")
    endpoint = str((control.get("api") or {}).get("endpoint") or "")
    ca_sha = str((control.get("api") or {}).get("cacerts_sha256") or "")
    expected_launcher_sha = str(((lock.get("launcher") or {}) if isinstance(lock.get("launcher"), Mapping) else {}).get("sha256") or "")
    if not endpoint.startswith("https://") or not HEX64.fullmatch(ca_sha) or not HEX64.fullmatch(expected_launcher_sha):
        raise InspectError("Phase 1 control-plane contracts are incomplete")
    if root != Path("/"):
        return {
            "schema_version": 1,
            "status": "ok",
            "control_plane": control,
            "version_lock": lock,
            "service_active": True,
            "readyz": "ok",
            "launcher_digest_verified": True,
            "mutation_surface": "none",
        }
    if os.geteuid() != 0:
        raise InspectError("server inspection must run as root")
    launcher = discover_launcher(lock)
    actual_sha = launcher_sha(launcher)
    if actual_sha != expected_launcher_sha:
        raise InspectError("K3s launcher digest differs from the Phase 1 version lock")
    systemctl = shutil.which("systemctl", path="/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin")
    if not systemctl:
        raise InspectError("systemctl is unavailable")
    active = run([systemctl, "is-active", "k3s.service"])
    if active.returncode != 0 or active.stdout.strip() != "active":
        raise InspectError("k3s.service is not active")
    ready = run([str(launcher), "kubectl", "get", "--raw=/readyz"])
    if ready.returncode != 0 or ready.stdout.strip() != "ok":
        raise InspectError("K3s API /readyz did not return ok")
    nodes = run([str(launcher), "kubectl", "get", "nodes", "-o", "json"])
    if nodes.returncode != 0:
        raise InspectError("K3s launcher could not read Nodes")
    try:
        node_doc = json.loads(nodes.stdout)
    except json.JSONDecodeError as exc:
        raise InspectError("K3s Node output is not JSON") from exc
    count = len(node_doc.get("items", [])) if isinstance(node_doc, dict) and isinstance(node_doc.get("items"), list) else 0
    if count < 1:
        raise InspectError("K3s returned no Nodes")
    return {
        "schema_version": 1,
        "status": "ok",
        "control_plane": control,
        "version_lock": lock,
        "service_active": True,
        "readyz": "ok",
        "node_count": count,
        "launcher_digest_verified": True,
        "mutation_surface": "none",
    }


def parser() -> argparse.ArgumentParser:
    result = argparse.ArgumentParser(description=__doc__)
    result.add_argument("command", choices=["inspect"])
    result.add_argument("--fixture-root", default="/")
    return result


def main(argv: Sequence[str] | None = None) -> int:
    args = parser().parse_args(argv)
    try:
        print(json.dumps(inspect(Path(args.fixture_root).resolve()), sort_keys=True))
        return 0
    except (Exception, KeyboardInterrupt) as exc:
        print(json.dumps({"schema_version": 1, "status": "error", "error": str(exc) or type(exc).__name__}, sort_keys=True))
        return 2


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