home-network-k3s-impl/catalog/render.py

#!/usr/bin/env python3
"""ServiceDefinition renderer (task T-0.3) — the platform's core GitOps contract.

Consumes ONE ServiceDefinition (schemas/servicedefinition.schema.json, catalog/*.yaml)
and emits the concrete manifest set the renderer contract specifies
(schemas/renderer-contract.md): Namespace + default-deny baseline, Deployment/Service or
KubeVirt VirtualMachine + CDI DataVolume, PVCs, per-dependency CiliumNetworkPolicy allows,
egress-class steering label, HTTPRoute(s), ServiceMonitor/alerts, Flux image-automation,
and the auth-mode / backup wiring hints.

Requirements: R-GITOPS-02 (single reviewable object wires every requirement), R-NET-01/02
(default-deny + explicit allows), R-NET-05 (egress class), R-STATE-01/02/04 (storage),
R-TLS/R-DNS (ingressRoutes/exposure), R-OBS-01 (observability), R-UPD-01/02/03 (update),
R-IDP-02/06 (auth), R-PLAT-02/03 (kind/source).

Design (documented per the task's "if the contract is ambiguous, choose the simplest
reproducible option"): a deterministic Python renderer. The contract does not mandate a
mechanism (script / kustomize generator / templates); a render-then-consume script is the
simplest reproducible option the repo can run in CI and that Flux consumes from a rendered
output path. CI invokes it via `render.py <sd>` (or `--all`); the offline L1 render
conformance verifier diffs its deterministic output against committed goldens.

Determinism: fixed document order + fixed key order + PyYAML with sort_keys=False and a
wide line width, so identical input always yields byte-identical output (golden-diffable).

Dependency: PyYAML (input parse + output emit). This is the ONE renderer dependency; the
conformance harness stays stdlib-only and only shells out to this script + diffs text.
"""
from __future__ import annotations

import argparse
import sys
from pathlib import Path

try:
    import yaml
except ImportError:  # pragma: no cover - explicit, actionable failure
    sys.stderr.write(
        "render.py requires PyYAML (the renderer's single dependency). "
        "Install with: pip install pyyaml\n"
    )
    raise SystemExit(2)

# --- Label / naming conventions (match platform/kyverno/policies + platform/*/) --------
LBL = "homelab.keithjackson.dev"
ZONE = "home.keithjackson.dev"
GATEWAY_NAME = "shared"
GATEWAY_NS = "gateway"
KP_RELEASE = "kube-prometheus-stack"  # ServiceMonitor/PrometheusRule discovery label

# egress classes and auth modes fixed by docs/03 / docs/13 + the Kyverno enums.
EGRESS_CLASSES = ("internet", "vpn")
AUTH_MODES = ("oidc", "forward-auth", "native", "none")


# =============================================================================
# Defaults (renderer-contract.md "Default posture") — materialize BEFORE mapping.
# =============================================================================
def apply_defaults(sd: dict) -> dict:
    """Return a copy of the ServiceDefinition with the contract's default posture filled.

    A field absent from a ServiceDefinition renders from its default posture, not "not at
    all" (renderer-contract.md).
    """
    s = dict(sd)
    s.setdefault("kind", "container")
    s.setdefault("exposure", "internal")
    s.setdefault("egress", "internet")
    s.setdefault("storage", [])
    s.setdefault("dependencies", [])
    s.setdefault("ingressRoutes", [])

    # auth.mode default posture: forward-auth for any exposed UI, else none (internal, no
    # UI). An explicit auth block always wins (R-IDP-02).
    auth = dict(s.get("auth") or {})
    if "mode" not in auth:
        auth["mode"] = "forward-auth" if s["ingressRoutes"] else "none"
    auth.setdefault("groups", [])
    s["auth"] = auth

    update = dict(s.get("update") or {})
    update.setdefault("channel", "stable")
    update.setdefault("autoRollbackOnFailedHealth", True)
    s["update"] = update

    obs = dict(s.get("observability") or {})
    obs.setdefault("metrics", True)
    obs.setdefault("slis", [])
    s["observability"] = obs

    # storage entry class default is longhorn (replicated tier).
    s["storage"] = [{**{"class": "longhorn"}, **v} for v in s["storage"]]
    # dependency protocol default is tcp.
    s["dependencies"] = [{**{"protocol": "tcp"}, **d} for d in s["dependencies"]]
    return s


# =============================================================================
# Field -> resource mapping (renderer-contract.md "Field -> resource mapping")
# =============================================================================
def r_namespace(s: dict) -> dict:
    """Namespace = the service trust zone / isolation boundary (R-NET, docs/03).

    Carries app-namespace=true so Kyverno generates the default-deny baseline and the
    egress-class / auth-mode admission policies apply. gateway-access=shared lets its
    HTTPRoutes attach to the shared Gateway (only when it exposes routes).
    """
    labels = {f"{LBL}/app-namespace": "true"}
    if s["ingressRoutes"]:
        labels["gateway-access"] = "shared"
    ann = {}
    backup = (s.get("backup") or {}).get("policy")
    if backup:
        ann[f"{LBL}/backup-policy"] = backup  # R-STATE-03 off-cluster schedule ref (hint)
    meta = {"name": s["name"], "labels": labels}
    if ann:
        meta["annotations"] = ann
    return {"apiVersion": "v1", "kind": "Namespace", "metadata": meta}


def r_default_deny(s: dict) -> dict:
    """The default-deny baseline: deny all ingress+egress, allow ONLY DNS egress (R-NET-01).

    Rendered explicitly (GitOps at namespace creation, renderer-contract §dependencies) —
    it matches the Kyverno-generated baseline shape (platform/kyverno/.../generated-default
    -deny.yaml) so the render carries the guarantee even before admission generates it.
    """
    return {
        "apiVersion": "networking.k8s.io/v1",
        "kind": "NetworkPolicy",
        "metadata": {
            "name": "default-deny",
            "namespace": s["name"],
            "labels": {f"{LBL}/baseline": "default-deny"},
        },
        "spec": {
            "podSelector": {},
            "policyTypes": ["Ingress", "Egress"],
            "ingress": [],
            "egress": [
                {
                    "to": [
                        {
                            "namespaceSelector": {
                                "matchLabels": {
                                    "kubernetes.io/metadata.name": "kube-system"
                                }
                            }
                        }
                    ],
                    "ports": [
                        {"protocol": "UDP", "port": 53},
                        {"protocol": "TCP", "port": 53},
                    ],
                }
            ],
        },
    }


def _pod_labels(s: dict) -> dict:
    """Workload pod labels: identity + the egress-class steering hook + auth-mode.

    egress-class is what the shared Cilium Egress Gateway selects on to steer vpn-class
    pods (enforcement is OUTSIDE the workload — renderer-contract §egress; ADR-0015);
    auth-mode satisfies the Kyverno require-auth-mode-label policy (R-IDP-02).
    """
    return {
        "app.kubernetes.io/name": s["name"],
        f"{LBL}/egress-class": s["egress"],
        f"{LBL}/auth-mode": s["auth"]["mode"],
    }


def _service_ports(s: dict) -> list:
    ports = sorted({r["port"] for r in s["ingressRoutes"]})
    return [{"name": f"port-{p}", "port": p, "targetPort": p} for p in ports]


def r_pvcs(s: dict) -> list:
    """One PVC per storage entry (R-STATE-01/02/04).

    longhorn -> 3-replica replicated tier; bulk -> external media tier (not 3x replicated).
    """
    out = []
    for vol in s["storage"]:
        cls = vol.get("class", "longhorn")
        storage_class = "longhorn" if cls == "longhorn" else "bulk"
        spec = {
            "accessModes": ["ReadWriteOnce"],
            "storageClassName": storage_class,
        }
        if vol.get("size"):
            spec["resources"] = {"requests": {"storage": vol["size"]}}
        out.append(
            {
                "apiVersion": "v1",
                "kind": "PersistentVolumeClaim",
                "metadata": {
                    "name": f"{s['name']}-{vol['name']}",
                    "namespace": s["name"],
                    "labels": {"app.kubernetes.io/name": s["name"]},
                },
                "spec": spec,
            }
        )
    return out


def r_deployment(s: dict) -> dict:
    """Deployment for a container workload (R-PLAT-02)."""
    pod_labels = _pod_labels(s)
    container = {
        "name": s["name"],
        "image": s["source"]["image"],
    }
    ports = _service_ports(s)
    if ports:
        container["ports"] = [{"containerPort": p["port"]} for p in ports]

    # env / envFrom (schema-optional): the k8s EnvVar / EnvFromSource lists are emitted
    # VERBATIM onto the app container. This is what lets a headless-install/DB-backed app
    # (gitea, nextcloud, semaphore, unifi, minecraft) configure itself and pull a fixed
    # secret (e.g. gitea's SECRET_KEY via secretKeyRef) without a supplementary patch.
    # Absent => the key is never set, so apps that declare no env render byte-identical.
    if s.get("env"):
        container["env"] = s["env"]
    if s.get("envFrom"):
        container["envFrom"] = s["envFrom"]

    volume_mounts, volumes = [], []
    for vol in s["storage"]:
        # Default mount is /data/<name>; an explicit `mountPath` (schema-optional) overrides
        # it verbatim for apps that need a fixed on-disk layout (e.g. Gitea keeps all state
        # directly under /data). Omitting it preserves the /data/<name> default (back-compat).
        mount_path = vol.get("mountPath", f"/data/{vol['name']}")
        volume_mounts.append({"name": vol["name"], "mountPath": mount_path})
        volumes.append(
            {
                "name": vol["name"],
                "persistentVolumeClaim": {"claimName": f"{s['name']}-{vol['name']}"},
            }
        )
    if volume_mounts:
        container["volumeMounts"] = volume_mounts

    # lifecycle hooks (schema-optional): the k8s Lifecycle (postStart/preStop exec) is
    # emitted VERBATIM. Used for idempotent break-glass/bootstrap steps (e.g. gitea's
    # postStart exec that seeds a local admin). Absent => byte-identical for other apps.
    if s.get("lifecycle"):
        container["lifecycle"] = s["lifecycle"]

    pod_spec = {"containers": [container]}
    if volumes:
        pod_spec["volumes"] = volumes

    ann = _auth_annotations(s)
    tmpl_meta = {"labels": pod_labels}
    if ann:
        tmpl_meta["annotations"] = ann

    spec = {"replicas": 1}
    # An app with persistent storage mounts a ReadWriteOnce Longhorn PVC. The default
    # RollingUpdate strategy tries to start the NEW pod before terminating the old one, but
    # the old pod still holds the RWO volume -> the new pod is stuck ContainerCreating and
    # the rollout deadlocks (single-replica app can never hand the volume over). Recreate
    # terminates the old pod first, freeing the PVC for the new one (R-STATE-01/02).
    if s["storage"]:
        spec["strategy"] = {"type": "Recreate"}
    spec["selector"] = {"matchLabels": {"app.kubernetes.io/name": s["name"]}}
    spec["template"] = {"metadata": tmpl_meta, "spec": pod_spec}

    return {
        "apiVersion": "apps/v1",
        "kind": "Deployment",
        "metadata": {
            "name": s["name"],
            "namespace": s["name"],
            "labels": {"app.kubernetes.io/name": s["name"]},
        },
        "spec": spec,
    }


def r_service(s: dict):
    ports = _service_ports(s)
    if not ports:
        return None
    return {
        "apiVersion": "v1",
        "kind": "Service",
        "metadata": {
            "name": s["name"],
            "namespace": s["name"],
            "labels": {"app.kubernetes.io/name": s["name"]},
        },
        "spec": {
            "selector": {"app.kubernetes.io/name": s["name"]},
            "ports": ports,
        },
    }


def r_dependency_policies(s: dict) -> list:
    """One CiliumNetworkPolicy egress allow per dependency, on top of default-deny (R-NET-02).

    Everything undeclared stays denied; empty dependencies => reaches only DNS.

    Cross-namespace note: a CiliumNetworkPolicy's `toEndpoints` selector is scoped to the
    POLICY's own namespace unless it carries the reserved `k8s:io.kubernetes.pod.namespace`
    label. Every service lives in its own namespace named after it (r_namespace), and a
    dependency always targets a DIFFERENT service/namespace, so the destination's namespace
    (== dep["to"]) MUST be pinned here — otherwise Cilium looks for the destination's app
    label inside the SOURCE namespace, finds nothing, and silently denies the egress (proven
    in-cluster via Hubble: `EGRESS DENIED`). This pins it so the declared dependency's egress
    half actually reaches the destination (verified: EGRESS ALLOWED / FORWARDED).
    """
    out = []
    for dep in s["dependencies"]:
        proto = dep.get("protocol", "tcp").upper()
        out.append(
            {
                "apiVersion": "cilium.io/v2",
                "kind": "CiliumNetworkPolicy",
                "metadata": {
                    "name": f"{s['name']}-allow-{dep['to']}-{dep['port']}",
                    "namespace": s["name"],
                    "labels": {"app.kubernetes.io/name": s["name"]},
                },
                "spec": {
                    "endpointSelector": {
                        "matchLabels": {"app.kubernetes.io/name": s["name"]}
                    },
                    "egress": [
                        {
                            "toEndpoints": [
                                {
                                    "matchLabels": {
                                        "app.kubernetes.io/name": dep["to"],
                                        "k8s:io.kubernetes.pod.namespace": dep["to"],
                                    }
                                }
                            ],
                            "toPorts": [
                                {
                                    "ports": [
                                        {"port": str(dep["port"]), "protocol": proto}
                                    ]
                                }
                            ],
                        }
                    ],
                },
            }
        )
    return out


def build_reverse_map(sds: list) -> dict:
    """Build the cross-service reverse-dependency map (T-2.3 reciprocal ingress).

    A `dependencies` edge is declared ONCE, in the SOURCE's catalog entry, as
    `{to: X, port: P}`. With the default-deny-BOTH-directions baseline (R-NET-01), that one
    declaration must compile into TWO matching halves (docs/03 "Explicit allow rules"; the
    conformance fixtures spell it out: "egress permit on the source, ingress permit on the
    destination"): (a) source egress-allow to X:P — rendered by r_dependency_policies on the
    source — AND (b) X ingress-allow from source:P — the reciprocal half rendered here.

    render.py otherwise processes each ServiceDefinition independently, so the destination X
    never sees the SOURCE's declaration. This pass scans ALL ServiceDefinitions once and
    returns `{destination_name: [{source, port, protocol}, ...]}` so that, when rendering X,
    r_reverse_dependency_policies can emit the ingress half for every source that declares a
    dependency ON X. Deterministic: each destination's inbound list is sorted by
    (source, port, protocol) so the render stays byte-reproducible (golden-diffable).
    """
    rev: dict = {}
    for sd in sds:
        if not isinstance(sd, dict) or "name" not in sd:
            continue  # tolerate non-ServiceDefinition siblings (e.g. a stray yaml)
        s = apply_defaults(sd)
        src = s["name"]
        for dep in s["dependencies"]:
            rev.setdefault(dep["to"], []).append(
                {
                    "source": src,
                    "port": dep["port"],
                    "protocol": dep.get("protocol", "tcp"),
                }
            )
    for dest in rev:
        rev[dest].sort(key=lambda e: (e["source"], int(e["port"]), e["protocol"]))
    return rev


def r_reverse_dependency_policies(s: dict, inbound: list) -> list:
    """Reciprocal ingress-allow CiliumNetworkPolicies on THIS service (R-NET-02).

    `inbound` is this service's slice of the cross-service reverse map (build_reverse_map):
    the list of sources that declared a dependency ON this service. Each becomes an ADDITIVE
    ingress allow on top of default-deny, mirroring the source-side egress allow
    (r_dependency_policies) exactly but inverted: `fromEndpoints` matching the SOURCE's
    identity on the declared port/protocol. As with the egress half, the source lives in a
    DIFFERENT namespace, and a CiliumNetworkPolicy's `fromEndpoints` selector is scoped to the
    POLICY's own namespace unless it carries the reserved `k8s:io.kubernetes.pod.namespace`
    label — so the source's namespace (== edge["source"]) is pinned alongside its app label.
    Without both halves so pinned the destination's default-deny-ingress drops the source's
    traffic and the declared dependency does not actually work (docs/03 R-NET-01/02; proven
    in-cluster via Hubble: pinned => FORWARDED, unpinned => Policy denied DROPPED).
    """
    out = []
    for edge in inbound or []:
        proto = edge.get("protocol", "tcp").upper()
        out.append(
            {
                "apiVersion": "cilium.io/v2",
                "kind": "CiliumNetworkPolicy",
                "metadata": {
                    "name": f"{s['name']}-allow-{edge['source']}-{edge['port']}",
                    "namespace": s["name"],
                    "labels": {"app.kubernetes.io/name": s["name"]},
                },
                "spec": {
                    "endpointSelector": {
                        "matchLabels": {"app.kubernetes.io/name": s["name"]}
                    },
                    "ingress": [
                        {
                            "fromEndpoints": [
                                {
                                    "matchLabels": {
                                        "app.kubernetes.io/name": edge["source"],
                                        "k8s:io.kubernetes.pod.namespace": edge["source"],
                                    }
                                }
                            ],
                            "toPorts": [
                                {
                                    "ports": [
                                        {"port": str(edge["port"]), "protocol": proto}
                                    ]
                                }
                            ],
                        }
                    ],
                },
            }
        )
    return out


def r_gateway_ingress_policy(s: dict):
    """North-south allow: let the shared Cilium Gateway reach an exposed service's port(s).

    Any service that renders an HTTPRoute (non-empty ingressRoutes) is reachable through the
    shared Gateway, but the namespace default-deny has `ingress: []` (deny all east-west and
    north-south). This CiliumNetworkPolicy is the ADDITIVE, least-privilege allow that opens
    ONLY the north-south path (docs/09 "Ingress opens north-south only — it does not relax
    east-west default-deny"): ingress from Cilium's reserved `ingress` identity (the Gateway/
    Ingress Envoy proxy, `reserved:ingress`, numeric identity 8 in Cilium 1.16.5) to ONLY the
    exposed service port(s). Everything else stays denied; undeclared ports stay denied.

    Verified empirically against the cluster: with default-deny in place, Hubble shows the
    Gateway proxy reaching the backend as source identity `(ingress)` and the flow
    `Policy denied DROPPED` on the service port — so `fromEntities: [ingress]` is exactly the
    source the Cilium Gateway uses, and the port is the service targetPort (R-NET-01/02).
    """
    if not s["ingressRoutes"]:
        return None
    ports = _service_ports(s)
    return {
        "apiVersion": "cilium.io/v2",
        "kind": "CiliumNetworkPolicy",
        "metadata": {
            "name": f"{s['name']}-allow-gateway-ingress",
            "namespace": s["name"],
            "labels": {"app.kubernetes.io/name": s["name"]},
        },
        "spec": {
            "endpointSelector": {
                "matchLabels": {"app.kubernetes.io/name": s["name"]}
            },
            "ingress": [
                {
                    "fromEntities": ["ingress"],
                    "toPorts": [
                        {
                            "ports": [
                                {"port": str(p["port"]), "protocol": "TCP"}
                                for p in ports
                            ]
                        }
                    ],
                }
            ],
        },
    }


def _route_meta(s: dict, route: dict) -> dict:
    meta = {
        "name": f"{s['name']}-{route['port']}",
        "namespace": s["name"],
        "labels": {"app.kubernetes.io/name": s["name"]},
    }
    ann = _auth_annotations(s)
    if ann:
        meta["annotations"] = ann
    return meta


def r_internet_egress(s: dict) -> dict:
    """North-south egress allow: let the app reach the INTERNET (Cilium reserved `world`
    entity = everything outside the cluster), which is the whole point of the `egress` class
    (docs/03: internet-class → WAN; vpn-class → steered through the node tun0 PIA tunnel with
    the nftables kill-switch, ADR-0015). The default-deny baseline (R-NET-01) allows only DNS
    so EAST-WEST stays denied; this ADDITIVE policy opens ONLY `world` (non-cluster) egress —
    cluster/east-west traffic stays denied except the declared `dependencies`. Every app
    declares an egress class (EGRESS_CLASSES), so every app gets this. The steering + kill-
    switch (a cluster-wide singleton keyed on the egress-class pod label) decide the PATH; this
    policy decides that north-south egress is ALLOWED at all.
    """
    return {
        "apiVersion": "cilium.io/v2",
        "kind": "CiliumNetworkPolicy",
        "metadata": {
            "name": f"{s['name']}-allow-internet-egress",
            "namespace": s["name"],
            "labels": {"app.kubernetes.io/name": s["name"]},
        },
        "spec": {
            "endpointSelector": {"matchLabels": {"app.kubernetes.io/name": s["name"]}},
            "egress": [{"toEntities": ["world"]}],
        },
    }


def r_httproutes(s: dict) -> list:
    """One HTTPRoute per ingressRoute, attaching host -> service port on the shared Gateway.

    TLS is the Gateway's wildcard *.home.keithjackson.dev cert (no per-service cert step);
    ExternalDNS manages the record from the HTTPRoute (R-TLS/R-DNS, renderer-contract).
    """
    out = []
    for route in s["ingressRoutes"]:
        out.append(
            {
                "apiVersion": "gateway.networking.k8s.io/v1",
                "kind": "HTTPRoute",
                "metadata": _route_meta(s, route),
                "spec": {
                    "parentRefs": [
                        {
                            "name": GATEWAY_NAME,
                            "namespace": GATEWAY_NS,
                            "sectionName": "https",
                        }
                    ],
                    "hostnames": [route["host"]],
                    "rules": [
                        {
                            "matches": [
                                {"path": {"type": "PathPrefix", "value": "/"}}
                            ],
                            "backendRefs": [
                                {"name": s["name"], "port": route["port"]}
                            ],
                        }
                    ],
                },
            }
        )
    return out


def _auth_annotations(s: dict) -> dict:
    """Auth wiring hints (renderer-contract §auth): the platform (oauth2-proxy/Keycloak)
    consumes these to protect ingressRoutes. oidc/forward-auth carry the authorized groups
    (R-IDP-06: access is 'in an authorized group', not merely 'logged in')."""
    mode = s["auth"]["mode"]
    if mode in ("oidc", "forward-auth"):
        groups = ",".join(s["auth"].get("groups", []))
        return {
            f"{LBL}/auth-mode": mode,
            f"{LBL}/auth-groups": groups,
        }
    return {}


def r_servicemonitor(s: dict):
    if not s["observability"]["metrics"]:
        return None
    return {
        "apiVersion": "monitoring.coreos.com/v1",
        "kind": "ServiceMonitor",
        "metadata": {
            "name": s["name"],
            "namespace": s["name"],
            "labels": {"release": KP_RELEASE, "app.kubernetes.io/name": s["name"]},
        },
        "spec": {
            "selector": {"matchLabels": {"app.kubernetes.io/name": s["name"]}},
            "namespaceSelector": {"matchNames": [s["name"]]},
            "endpoints": [{"port": "metrics", "interval": "30s"}],
        },
    }


def r_prometheusrule(s: dict):
    """Each SLI becomes at least one alert rule (R-OBS-01)."""
    slis = s["observability"]["slis"]
    if not slis:
        return None
    rules = []
    for sli in slis:
        if sli == "up":
            rules.append(
                {
                    "alert": f"{s['name']}-down".replace("_", "-"),
                    "expr": f'up{{job="{s["name"]}"}} == 0',
                    "for": "5m",
                    "labels": {"severity": "critical", "service": s["name"]},
                    "annotations": {
                        "summary": f"{s['name']} is down",
                        "description": f"{s['name']} has been unreachable for 5m (SLI: up).",
                    },
                }
            )
        else:
            rules.append(
                {
                    "alert": f"{s['name']}-{sli}".replace("_", "-"),
                    "expr": f'absent({sli}{{service="{s["name"]}"}})',
                    "for": "10m",
                    "labels": {"severity": "warning", "service": s["name"]},
                    "annotations": {"summary": f"{s['name']} SLI {sli} unhealthy"},
                }
            )
    return {
        "apiVersion": "monitoring.coreos.com/v1",
        "kind": "PrometheusRule",
        "metadata": {
            "name": f"{s['name']}-slis",
            "namespace": s["name"],
            "labels": {"release": KP_RELEASE, "app.kubernetes.io/name": s["name"]},
        },
        "spec": {"groups": [{"name": f"{s['name']}.slis", "rules": rules}]},
    }


def r_image_automation(s: dict) -> list:
    """container kind -> Flux ImageRepository + ImagePolicy encoding channel/range (R-UPD).

    Flux image-automation advances the running version on its scan interval; no human/AI
    edits a tag (R-UPD-02). `pin` freezes automation. vm kind uses in-guest updates instead.
    """
    if s["kind"] != "container":
        return []
    image = s["source"]["image"].split(":", 1)[0]
    upd = s["update"]
    interval = "1h0m0s"
    imagerepo = {
        "apiVersion": "image.toolkit.fluxcd.io/v1beta2",
        "kind": "ImageRepository",
        "metadata": {"name": s["name"], "namespace": s["name"]},
        "spec": {"image": image, "interval": interval},
    }
    policy_spec: dict = {"imageRepositoryRef": {"name": s["name"]}}
    if upd.get("pin"):
        # Operator freeze escape hatch: pin to a fixed tag, stopping automation.
        policy_spec["filterTags"] = {"pattern": f"^{upd['pin']}$"}
        policy_spec["policy"] = {"alphabetical": {"order": "asc"}}
    elif upd.get("range"):
        policy_spec["policy"] = {"semver": {"range": upd["range"]}}
    else:
        policy_spec["policy"] = {"semver": {"range": ">=0.0.0"}}
    imagepolicy = {
        "apiVersion": "image.toolkit.fluxcd.io/v1beta2",
        "kind": "ImagePolicy",
        "metadata": {
            "name": s["name"],
            "namespace": s["name"],
            "annotations": {
                f"{LBL}/update-channel": upd.get("channel", "stable"),
                f"{LBL}/auto-rollback": str(
                    upd.get("autoRollbackOnFailedHealth", True)
                ).lower(),
            },
        },
        "spec": policy_spec,
    }
    return [imagerepo, imagepolicy]


# --- VM (kind: vm) path: KubeVirt VirtualMachine + CDI DataVolume ----------------------
def r_datavolume(s: dict) -> dict:
    """Standalone CDI DataVolume (NOT a dataVolumeTemplate) so the disk OUTLIVES the VM
    and survives destroy/rebuild (R-STATE-01) — matches the vm-lifecycle fixture shape."""
    root = s["storage"][0] if s["storage"] else {"name": "rootdisk", "size": "10Gi"}
    return {
        "apiVersion": "cdi.kubevirt.io/v1beta1",
        "kind": "DataVolume",
        "metadata": {
            "name": f"{s['name']}-{root['name']}",
            "namespace": s["name"],
        },
        "spec": {
            "source": {"registry": {"url": f"docker://{s['source']['vmTemplate']}"}},
            "storage": {
                "accessModes": ["ReadWriteOnce"],
                "storageClassName": "longhorn",
                "resources": {"requests": {"storage": root.get("size", "10Gi")}},
            },
        },
    }


VM_CLOUD_INIT = (
    "#cloud-config\n"
    "package_update: true\n"
    "packages:\n"
    "  - dnf-automatic\n"
    "write_files:\n"
    "  - path: /etc/dnf/automatic.conf\n"
    "    content: |\n"
    "      [commands]\n"
    "      upgrade_type = default\n"
    "      random_sleep = 0\n"
    "      download_updates = yes\n"
    "      apply_updates = yes\n"
    "runcmd:\n"
    "  - [ systemctl, enable, --now, dnf-automatic.timer ]\n"
)


def r_virtualmachine(s: dict) -> dict:
    """KubeVirt VirtualMachine; egress-class propagates to the virt-launcher pod so the
    shared Cilium egress steering + Kyverno policy see the class (R-PLAT-03; ADR-0015).
    In-guest dnf-automatic delivers auto-updates with no human step (R-UPD-01, docs/11)."""
    root = s["storage"][0] if s["storage"] else {"name": "rootdisk"}
    pod_labels = {
        "kubevirt.io/vm": s["name"],
        f"{LBL}/egress-class": s["egress"],
        f"{LBL}/auth-mode": s["auth"]["mode"],
    }
    return {
        "apiVersion": "kubevirt.io/v1",
        "kind": "VirtualMachine",
        "metadata": {
            "name": s["name"],
            "namespace": s["name"],
            "labels": {"app.kubernetes.io/name": s["name"]},
        },
        "spec": {
            "running": False,
            "template": {
                "metadata": {"labels": pod_labels},
                "spec": {
                    "domain": {
                        "cpu": {"cores": 1},
                        "memory": {"guest": "1536Mi"},
                        "devices": {
                            "disks": [
                                {"name": "rootdisk", "disk": {"bus": "virtio"}},
                                {"name": "cloudinit", "disk": {"bus": "virtio"}},
                            ],
                            "interfaces": [{"name": "default", "masquerade": {}}],
                        },
                        "resources": {"requests": {"memory": "1536Mi"}},
                    },
                    "networks": [{"name": "default", "pod": {}}],
                    "volumes": [
                        {
                            "name": "rootdisk",
                            "dataVolume": {"name": f"{s['name']}-{root['name']}"},
                        },
                        {
                            "name": "cloudinit",
                            "cloudInitNoCloud": {"userData": VM_CLOUD_INIT},
                        },
                    ],
                },
            },
        },
    }


# =============================================================================
# Top-level render + emit
# =============================================================================
def render(sd: dict, image_automation: bool = True, inbound: list | None = None) -> list:
    """Return an ordered list of (header_comment, resource_dict) for a ServiceDefinition.

    image_automation=False omits the Flux ImageRepository/ImagePolicy CRs — used when
    onboarding apps before the Flux image-reflector/-automation controllers are installed
    (their CRDs would otherwise be missing and the apps Kustomization would fail). The L1
    render goldens use the default (True), so their expected output is unchanged.

    inbound is this service's slice of the cross-service reverse-dependency map
    (build_reverse_map): the sources that declared a dependency ON this service. It renders
    the reciprocal ingress-allow half of each declared dependency (R-NET-02). None/[] (the
    default, and every single-service render with no cross-catalog context) renders no
    reciprocal ingress — so a lone render is unchanged and stays golden-diffable.
    """
    s = apply_defaults(sd)
    if s["kind"] not in ("container", "vm"):
        raise ValueError(f"unknown kind: {s['kind']!r}")
    if s["egress"] not in EGRESS_CLASSES:
        raise ValueError(f"invalid egress class: {s['egress']!r}")
    if s["auth"]["mode"] not in AUTH_MODES:
        raise ValueError(f"invalid auth.mode: {s['auth']['mode']!r}")

    docs: list = []
    docs.append(("Namespace (trust zone; default-deny + policies apply)", r_namespace(s)))
    docs.append(("Default-deny baseline (R-NET-01): deny all, allow only DNS", r_default_deny(s)))

    if s["kind"] == "container":
        # Container PVCs. (For a vm, the CDI DataVolume owns/creates the disk PVC, so we
        # never render a plain PVC for the VM root disk — it would collide with CDI.)
        for pvc in r_pvcs(s):
            docs.append((f"PVC {pvc['metadata']['name']} (R-STATE)", pvc))
        docs.append(("Deployment (R-PLAT-02)", r_deployment(s)))
        svc = r_service(s)
        if svc:
            docs.append(("Service", svc))
    else:
        docs.append(("CDI DataVolume (standalone; survives rebuild, R-STATE-01)", r_datavolume(s)))
        docs.append(("KubeVirt VirtualMachine (R-PLAT-03)", r_virtualmachine(s)))

    for cnp in r_dependency_policies(s):
        docs.append((f"CiliumNetworkPolicy egress allow (R-NET-02): {cnp['metadata']['name']}", cnp))

    for cnp in r_reverse_dependency_policies(s, inbound or []):
        docs.append(
            (
                "CiliumNetworkPolicy reciprocal ingress allow (R-NET-02): "
                f"{cnp['metadata']['name']}",
                cnp,
            )
        )

    gw_allow = r_gateway_ingress_policy(s)
    if gw_allow:
        docs.append(
            (
                "CiliumNetworkPolicy gateway-ingress allow (R-NET-01/02): "
                f"{gw_allow['metadata']['name']}",
                gw_allow,
            )
        )

    ie = r_internet_egress(s)
    docs.append(
        (
            "CiliumNetworkPolicy world egress allow (egress class governs the path; R-NET-05): "
            f"{ie['metadata']['name']}",
            ie,
        )
    )

    for hr in r_httproutes(s):
        docs.append((f"HTTPRoute {hr['metadata']['name']} (R-TLS/R-DNS)", hr))

    sm = r_servicemonitor(s)
    if sm:
        docs.append(("ServiceMonitor (R-OBS-01)", sm))
    pr = r_prometheusrule(s)
    if pr:
        docs.append(("PrometheusRule SLI alerts (R-OBS-01)", pr))

    if image_automation:
        for res in r_image_automation(s):
            docs.append((f"Flux {res['kind']} (R-UPD)", res))

    return docs


def emit(docs: list) -> str:
    """Deterministic multi-document YAML text (golden-diffable)."""
    parts = []
    for comment, resource in docs:
        body = yaml.dump(
            resource,
            sort_keys=False,
            default_flow_style=False,
            width=4096,
        )
        parts.append(f"---\n# {comment}\n{body}".rstrip() + "\n")
    return "".join(parts)


def load_sd(path: Path) -> dict:
    with path.open() as fh:
        return yaml.safe_load(fh)


def _load_catalog(dir_: Path) -> list:
    """Load every *.yaml ServiceDefinition in dir_ (tolerating unparseable siblings).

    Used to build the cross-service reverse-dependency map for a single-file render, so
    `render.py catalog/X.yaml` still emits X's reciprocal ingress for sources elsewhere in
    the same catalog. Deterministic (sorted); render.py itself is not a *.yaml so it is skipped.
    """
    sds = []
    for p in sorted(dir_.glob("*.yaml")):
        try:
            doc = load_sd(p)
        except Exception:  # pragma: no cover - a malformed sibling must not break the render
            continue
        if isinstance(doc, dict):
            sds.append(doc)
    return sds


def render_file(path: Path, image_automation: bool = True, catalog_dir: Path | None = None) -> str:
    """Render one ServiceDefinition file, wiring in its reciprocal ingress allows.

    The reverse-dependency map is built across the whole catalog (catalog_dir, defaulting to
    the file's own directory) so this service receives the ingress half of every dependency
    that OTHER services declare on it (R-NET-02). A directory with only this file (or with no
    inbound edges — e.g. the L1 render fixtures) yields an empty inbound list, leaving the
    render byte-identical to the pre-change output.
    """
    sd = load_sd(path)
    src_dir = catalog_dir if catalog_dir is not None else path.parent
    rev = build_reverse_map(_load_catalog(src_dir))
    inbound = rev.get(sd["name"], [])
    return emit(render(sd, image_automation=image_automation, inbound=inbound))


# =============================================================================
# Self-test (deterministic, no cluster) — `render.py --selftest`
# =============================================================================
def _selftest() -> int:
    failures = []

    def check(cond, msg):
        if not cond:
            failures.append(msg)

    # 1) A minimal container SD renders the mandatory contract invariants.
    sd = {"name": "demo", "source": {"image": "nginx:1"}}
    docs = render(sd)
    kinds = [d[1]["kind"] for d in docs]
    check("Namespace" in kinds, "container: missing Namespace")
    check("NetworkPolicy" in kinds, "container: missing default-deny NetworkPolicy")
    check("Deployment" in kinds, "container: missing Deployment")

    ns = next(d[1] for d in docs if d[1]["kind"] == "Namespace")
    check(
        ns["metadata"]["labels"].get(f"{LBL}/app-namespace") == "true",
        "namespace missing app-namespace=true (Kyverno hook)",
    )
    dep = next(d[1] for d in docs if d[1]["kind"] == "Deployment")
    plabels = dep["spec"]["template"]["metadata"]["labels"]
    check(
        plabels.get(f"{LBL}/egress-class") == "internet",
        "default egress-class must be internet (R-NET-05)",
    )
    check(
        plabels.get(f"{LBL}/auth-mode") in AUTH_MODES,
        "pod must carry a valid auth-mode label (R-IDP-02)",
    )
    ndp = next(d[1] for d in docs if d[1]["kind"] == "NetworkPolicy")
    check(
        ndp["spec"]["policyTypes"] == ["Ingress", "Egress"],
        "default-deny must cover both directions (R-NET-01)",
    )
    check(ndp["spec"]["ingress"] == [], "default-deny must deny all ingress (R-NET-01)")

    # 2) egress: vpn flips ONLY the egress-class label (R-NET-07).
    internet_docs = render({"name": "x", "source": {"image": "a:1"}, "egress": "internet"})
    vpn_docs = render({"name": "x", "source": {"image": "a:1"}, "egress": "vpn"})
    i_pod = next(d[1] for d in internet_docs if d[1]["kind"] == "Deployment")
    v_pod = next(d[1] for d in vpn_docs if d[1]["kind"] == "Deployment")
    check(
        i_pod["spec"]["template"]["metadata"]["labels"][f"{LBL}/egress-class"] == "internet"
        and v_pod["spec"]["template"]["metadata"]["labels"][f"{LBL}/egress-class"] == "vpn",
        "egress class must render onto the pod label",
    )
    # Everything except the egress-class label must be identical (R-NET-07).
    i_txt = emit(internet_docs).replace("egress-class: internet", "EGRESS")
    v_txt = emit(vpn_docs).replace("egress-class: vpn", "EGRESS")
    check(i_txt == v_txt, "flipping egress class changed more than the egress path (R-NET-07)")

    # 3) Determinism: same input twice -> byte-identical output.
    check(emit(render(sd)) == emit(render(sd)), "render is not deterministic")

    # 4) dependencies -> one CiliumNetworkPolicy allow each (R-NET-02).
    dep_docs = render(
        {
            "name": "y",
            "source": {"image": "a:1"},
            "dependencies": [{"to": "db", "port": 5432}],
        }
    )
    cnps = [d[1] for d in dep_docs if d[1]["kind"] == "CiliumNetworkPolicy"]
    dep_cnps = [c for c in cnps if c["metadata"]["name"] == "y-allow-db-5432"]
    check(len(dep_cnps) == 1, "one dependency must render one CiliumNetworkPolicy allow")
    if dep_cnps:
        to_ep = dep_cnps[0]["spec"]["egress"][0]["toEndpoints"][0]["matchLabels"]
        check(
            to_ep.get("k8s:io.kubernetes.pod.namespace") == "db"
            and to_ep.get("app.kubernetes.io/name") == "db",
            "egress allow must pin the destination namespace (cross-ns Cilium selector)",
        )
    check(
        any(c["metadata"]["name"] == "y-allow-internet-egress" for c in cnps),
        "every app (egress class) must render a world/internet egress allow",
    )

    # 4a) reciprocal ingress: a dependency declared ONCE on the SOURCE must render the
    #     matching ingress-allow on the DESTINATION (R-NET-02). Without it, default-deny-
    #     ingress drops the source and the declared dependency does not actually work.
    src_sd = {
        "name": "src",
        "source": {"image": "a:1"},
        "dependencies": [{"to": "dst", "port": 8080, "protocol": "tcp"}],
    }
    dst_sd = {"name": "dst", "source": {"image": "b:1"}}
    rev = build_reverse_map([src_sd, dst_sd])
    check(
        rev.get("dst") == [{"source": "src", "port": 8080, "protocol": "tcp"}],
        "reverse map must record src->dst:8080 under the destination",
    )
    check("src" not in rev, "a service nothing depends on must have no inbound edges")
    dst_docs = render(dst_sd, inbound=rev.get("dst", []))
    rev_cnps = [
        d[1]
        for d in dst_docs
        if d[1]["kind"] == "CiliumNetworkPolicy"
        and d[1]["metadata"]["name"] == "dst-allow-src-8080"
    ]
    check(len(rev_cnps) == 1, "destination must render one reciprocal ingress allow")
    if rev_cnps:
        ing = rev_cnps[0]["spec"]["ingress"]
        from_ep = ing[0]["fromEndpoints"][0]["matchLabels"]
        check(
            from_ep.get("app.kubernetes.io/name") == "src"
            and from_ep.get("k8s:io.kubernetes.pod.namespace") == "src",
            "reciprocal ingress must be from the SOURCE's identity, namespace-pinned (cross-ns)",
        )
        allowed = {p["port"] for r in ing for tp in r["toPorts"] for p in tp["ports"]}
        check(allowed == {"8080"}, "reciprocal ingress must open only the declared port")
    # additive: default-deny ingress stays [] even with the reciprocal allow present.
    dst_dd = next(d[1] for d in dst_docs if d[1]["kind"] == "NetworkPolicy")
    check(
        dst_dd["spec"]["ingress"] == [],
        "reciprocal ingress must be ADDITIVE — default-deny ingress stays [] (R-NET-01)",
    )
    # a lone render (no inbound) must NOT gain any reciprocal ingress (golden-stable).
    lone_docs = render(dst_sd)
    check(
        not any(
            d[1]["kind"] == "CiliumNetworkPolicy"
            and d[1]["metadata"]["name"].startswith("dst-allow-src")
            for d in lone_docs
        ),
        "a render with no inbound context must be unchanged (no reciprocal ingress)",
    )

    # 4b) exposure (ingressRoutes) -> an ADDITIVE gateway-ingress CiliumNetworkPolicy that
    #     allows ONLY the reserved `ingress` (Gateway proxy) identity to the exposed port,
    #     while default-deny still denies all other ingress (R-NET-01/02).
    exp_docs = render(
        {
            "name": "z",
            "source": {"image": "a:1"},
            "ingressRoutes": [{"host": "z.home.keithjackson.dev", "port": 80}],
        }
    )
    gw = [
        d[1]
        for d in exp_docs
        if d[1]["kind"] == "CiliumNetworkPolicy"
        and d[1]["metadata"]["name"].endswith("-allow-gateway-ingress")
    ]
    check(len(gw) == 1, "an exposed service must render one gateway-ingress allow (R-NET-01)")
    if gw:
        ing = gw[0]["spec"]["ingress"]
        check(
            ing[0]["fromEntities"] == ["ingress"],
            "gateway-ingress allow must be from the reserved `ingress` (Gateway) identity",
        )
        allowed_ports = {p["port"] for r in ing for tp in r["toPorts"] for p in tp["ports"]}
        check(
            allowed_ports == {"80"},
            "gateway-ingress allow must be least-privilege (only the exposed port)",
        )
    # default-deny stays deny-all-ingress even when an exposed service adds the allow.
    exp_dd = next(d[1] for d in exp_docs if d[1]["kind"] == "NetworkPolicy")
    check(
        exp_dd["spec"]["ingress"] == [],
        "gateway-ingress allow must be ADDITIVE — default-deny ingress stays [] (R-NET-01)",
    )
    # an internal (no ingressRoutes) service must NOT get a gateway-ingress allow.
    internal_docs = render({"name": "int", "source": {"image": "a:1"}})
    check(
        not any(
            d[1]["kind"] == "CiliumNetworkPolicy"
            and d[1]["metadata"]["name"].endswith("-allow-gateway-ingress")
            for d in internal_docs
        ),
        "a non-exposed service must not render a gateway-ingress allow (least privilege)",
    )

    # 5) vm kind -> VirtualMachine + standalone DataVolume.
    vm_docs = render(
        {
            "name": "vmx",
            "kind": "vm",
            "source": {"vmTemplate": "quay.io/containerdisks/fedora:41"},
            "storage": [{"name": "rootdisk", "size": "6Gi", "class": "longhorn"}],
        }
    )
    vm_kinds = [d[1]["kind"] for d in vm_docs]
    check("VirtualMachine" in vm_kinds, "vm: missing VirtualMachine")
    check("DataVolume" in vm_kinds, "vm: missing CDI DataVolume")

    # 6) env / envFrom / lifecycle -> emitted VERBATIM onto the app container (T-APPS).
    envd_docs = render(
        {
            "name": "envd",
            "source": {"image": "a:1"},
            "env": [
                {"name": "PLAIN", "value": "literal"},
                {
                    "name": "SECRET",
                    "valueFrom": {"secretKeyRef": {"name": "sec", "key": "k"}},
                },
            ],
            "envFrom": [{"secretRef": {"name": "bulk"}}],
            "lifecycle": {
                "postStart": {"exec": {"command": ["/bin/sh", "-c", "echo hi"]}}
            },
        }
    )
    ec = next(
        d[1] for d in envd_docs if d[1]["kind"] == "Deployment"
    )["spec"]["template"]["spec"]["containers"][0]
    check(
        ec.get("env") == [
            {"name": "PLAIN", "value": "literal"},
            {"name": "SECRET", "valueFrom": {"secretKeyRef": {"name": "sec", "key": "k"}}},
        ],
        "env must render verbatim onto the container (literal + secretKeyRef)",
    )
    check(
        ec.get("envFrom") == [{"secretRef": {"name": "bulk"}}],
        "envFrom must render verbatim onto the container",
    )
    check(
        ec.get("lifecycle")
        == {"postStart": {"exec": {"command": ["/bin/sh", "-c", "echo hi"]}}},
        "lifecycle must render verbatim onto the container",
    )
    # Absence => the keys are never set (existing apps stay byte-identical).
    plain_c = next(
        d[1] for d in render({"name": "plain", "source": {"image": "a:1"}})
        if d[1]["kind"] == "Deployment"
    )["spec"]["template"]["spec"]["containers"][0]
    check(
        "env" not in plain_c
        and "envFrom" not in plain_c
        and "lifecycle" not in plain_c,
        "an SD without env/envFrom/lifecycle must not gain those container keys",
    )

    if failures:
        for f in failures:
            sys.stderr.write(f"SELFTEST FAIL: {f}\n")
        return 1
    sys.stderr.write("render.py selftest: PASS\n")
    return 0


def main(argv=None) -> int:
    ap = argparse.ArgumentParser(description="Render a ServiceDefinition to manifests.")
    ap.add_argument("input", nargs="?", help="ServiceDefinition YAML file")
    ap.add_argument("--all", metavar="DIR", help="Render every *.yaml ServiceDefinition in DIR")
    ap.add_argument("-o", "--out-dir", help="With --all: write <name>.yaml files into this dir")
    ap.add_argument("--selftest", action="store_true", help="Run the built-in self-test")
    ap.add_argument("--no-image-automation", action="store_true",
                    help="Omit Flux ImageRepository/ImagePolicy CRs (for onboarding before the "
                         "image-reflector/-automation controllers are installed)")
    args = ap.parse_args(argv)

    if args.selftest:
        return _selftest()

    image_automation = not args.no_image_automation

    if args.all:
        src = Path(args.all)
        files = sorted(p for p in src.glob("*.yaml") if p.name != "render.py")
        for f in files:
            text = render_file(f, image_automation=image_automation)
            if args.out_dir:
                outdir = Path(args.out_dir)
                outdir.mkdir(parents=True, exist_ok=True)
                name = load_sd(f)["name"]
                (outdir / f"{name}.yaml").write_text(text)
            else:
                sys.stdout.write(text)
        return 0

    if not args.input:
        ap.error("provide a ServiceDefinition file, --all DIR, or --selftest")
    sys.stdout.write(render_file(Path(args.input), image_automation=image_automation))
    return 0


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