railiance-apps/tools/vergabe_pilot_inventory.py

#!/usr/bin/env python3
"""Read-only, value-free recovery inventory for the demo-company pilot.

Uses the caller's existing kubectl/Helm context for resource status and Helm
release summaries. No Secret output, values, logs, pod exec or resource changes.
Child output/errors stay private.
"""

from __future__ import annotations

import datetime
import json
import subprocess

CLUSTER_UID = "a553c742-0115-43d4-99a4-a5ca56fe0786"
NAMESPACE = "vergabe-demo-company"
RELEASE = "vergabe-teilnahme"
IMAGE = (
    "forgejo.coulomb.social/coulomb/vergabe-teilnahme@sha256:"
    "a26444f59c259698159c69ccb96f73dc648a261ece4c86bb2037a9d977870d91"
)
CLAIMS = {"media": "vergabe-teilnahme-media", "app-state": "vergabe-teilnahme-app-state"}


def read_json(argv):
    try:
        result = subprocess.run(argv, capture_output=True, timeout=20, check=False)
        if result.returncode:
            return None, "forbidden" if b"Forbidden" in result.stderr else "request_failed"
        return json.loads(result.stdout), "ok"
    except (OSError, subprocess.TimeoutExpired, ValueError, UnicodeError):
        return None, "unavailable_or_invalid_response"


def kube(*args):
    return read_json(["kubectl", "--request-timeout=12s", *args, "-o", "json"])


def collect():
    receipt = {
        "schema": "railiance.vergabe-pilot-native-inventory.v1",
        "observed_at": datetime.datetime.now(datetime.timezone.utc).isoformat(),
        "runtime_mutations": False,
        "native_inventory_complete": False,
    }
    cluster, status = kube("get", "namespace", "kube-system")
    receipt["cluster_read"] = status
    if cluster is None:
        return receipt
    receipt["cluster_matches_expected"] = cluster.get("metadata", {}).get("uid") == CLUSTER_UID
    if not receipt["cluster_matches_expected"]:
        return receipt

    nodes, status = kube("get", "nodes")
    receipt["nodes_read"] = status
    if nodes is not None:
        matching = [
            n for n in nodes.get("items", [])
            if any(a.get("address") == "92.205.62.239"
                   for a in n.get("status", {}).get("addresses", []))
        ]
        receipt["node_address_matches_expected"] = len(matching) == 1
        if len(matching) == 1:
            conditions = {c.get("type"): c.get("status")
                          for c in matching[0].get("status", {}).get("conditions", [])}
            receipt["node_ready"] = conditions.get("Ready") == "True"
            receipt["node_pressure_clear"] = all(
                conditions.get(c) == "False"
                for c in ("MemoryPressure", "DiskPressure", "PIDPressure")
            )

    deployment, status = kube("get", "deployment", RELEASE, "-n", NAMESPACE)
    receipt["deployment_read"] = status
    if deployment is not None:
        spec = deployment.get("spec", {})
        pod = spec.get("template", {}).get("spec", {})
        app = next((c for c in pod.get("containers", []) if c.get("name") == "app"), {})
        ready = deployment.get("status", {})
        receipt["deployment"] = {
            "image_matches_expected": app.get("image") == IMAGE,
            "replicas": spec.get("replicas"),
            "ready_replicas": ready.get("readyReplicas", 0),
            "updated_replicas": ready.get("updatedReplicas", 0),
            "generation_observed": ready.get("observedGeneration", 0)
            >= deployment.get("metadata", {}).get("generation", 1),
            "recreate": spec.get("strategy", {}).get("type") == "Recreate",
            "mounts_match_expected": all(
                any(v.get("name") == name
                    and v.get("persistentVolumeClaim", {}).get("claimName") == claim
                    for v in pod.get("volumes", []))
                for name, claim in CLAIMS.items()
            ) and all(
                any(m.get("name") == name and m.get("mountPath") == path
                    for m in app.get("volumeMounts", []))
                for name, path in (("media", "/app/media"), ("app-state", "/app/.issue-facade"))
            ),
        }

    receipt["claims"] = {}
    for name, claim in CLAIMS.items():
        data, status = kube("get", "pvc", claim, "-n", NAMESPACE)
        row = {"read": status}
        if data is not None:
            row["bound"] = data.get("status", {}).get("phase") == "Bound"
            row["identity_present"] = bool(data.get("metadata", {}).get("uid"))
        receipt["claims"][name] = row

    database, status = kube("get", "cluster", "apps-pg", "-n", "databases")
    receipt["database_cluster_read"] = status
    if database is not None:
        db_status = database.get("status", {})
        receipt["database_ready"] = (
            db_status.get("instances", 0) > 0
            and db_status.get("readyInstances", 0) == db_status.get("instances")
            and bool(db_status.get("currentPrimary"))
        )

    releases, status = read_json([
        "helm", "list", "--all", "--namespace", NAMESPACE,
        "--filter", "^vergabe-teilnahme$", "--output", "json",
    ])
    receipt["helm_read"] = status
    if isinstance(releases, list):
        rows = [r for r in releases if r.get("name") == RELEASE]
        if len(rows) == 1:
            receipt["helm"] = {
                "deployed": rows[0].get("status") == "deployed",
                "revision": int(rows[0]["revision"]),
                "chart_matches_expected": rows[0].get("chart") == "vergabe-teilnahme-0.2.1",
            }

    dep = receipt.get("deployment", {})
    release = receipt.get("helm", {})
    receipt["native_inventory_complete"] = (
        receipt.get("node_address_matches_expected") is True
        and receipt.get("node_ready") is True
        and receipt.get("node_pressure_clear") is True
        and dep.get("image_matches_expected") is True
        and dep.get("replicas") == dep.get("ready_replicas") == dep.get("updated_replicas") == 1
        and dep.get("generation_observed") is True
        and dep.get("recreate") is True and dep.get("mounts_match_expected") is True
        and all(c.get("bound") is True and c.get("identity_present") is True
                for c in receipt["claims"].values())
        and receipt.get("database_ready") is True
        and release.get("deployed") is True and release.get("chart_matches_expected") is True
    )
    return receipt


def main():
    try:
        receipt = collect()
    except Exception:
        receipt = {"schema": "railiance.vergabe-pilot-native-inventory.v1",
                   "native_inventory_complete": False, "failure": "invalid_inventory_response",
                   "runtime_mutations": False}
    print(json.dumps(receipt, indent=2))
    return 0 if receipt["native_inventory_complete"] else 1


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