Files
home-network-k3s-impl/catalog/render.py
T
jakman85 b184a42729 merge: internal services stop advertising themselves in public DNS
Six exposure:internal services published A records into the public Cloudflare
zone answering 192.168.2.20, disclosing the internal subnet, the Gateway VIP
and a full service inventory, and forming a DNS-rebind primitive that is why
the SOHO router strips these answers in the first place.

The renderer now stamps external-dns.alpha.kubernetes.io/controller: none on
every HTTPRoute it renders for an internal service, derived from the exposure
field alone — so a service that omits exposure defaults to internal and is
likewise unpublishable. Established against the deployed v0.21.0's own source,
where the controller check runs before hostname resolution in both configured
sources, then proven live on a throwaway hostname including that adding the
annotation to an existing record triggers the DELETE.

Deletes exactly 12 records: the A and TXT pair for jackett, pihole,
qbittorrent, radarr, semaphore and sonarr. A 39-record recovery baseline is
committed alongside.

Safe to land before the DHCP change, contrary to my earlier caution: the
router returns NOTHING for all six names today, so these records serve nobody
on the LAN, while Pi-hole at 192.168.2.23 already answers for anyone pointed
at it.

The five exposure:public services are untouched. They remain unreachable
publicly, which is a separate problem: the legacy edge at 136.51.115.252
returns HTTP 403 for our hostnames and serves a placeholder certificate, so
nothing is reverse-proxying them today.
2026-09-18 20:50:43 -05:00

2912 lines
134 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/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 and per-in-cluster-client
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 re
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"
# The platform hostname suffix, emitted as the Flux post-build TOKEN rather than a literal.
# Every rendered manifest lands under a Flux path whose Kustomization declares
# postBuild.substituteFrom cluster-vars, so `${BASE_DOMAIN}` is resolved per environment at
# reconcile time: dev renders home.keithjackson.dev, prod renders prod.home.keithjackson.dev.
# It is a DECLARED cluster-vars knob (clusters/base/cluster-vars.yaml), so it is written BARE —
# the escape contract below reserves `$${...}` for expansions meant for a container's shell.
# Hardcoding the literal here is what let prod be authored to serve dev's public hostnames,
# and prod's external-dns then deleted dev's records (infra/PROD-PROVISIONING.md §10).
ZONE = "${BASE_DOMAIN}"
# The bulk (external NFS) tier's SERVER, emitted as a Flux post-build TOKEN for exactly the same
# reason as ZONE: dev and prod reach the same Proxmox ZFS host by DIFFERENT addresses (dev's nodes
# are on the host's own LAN and dial it directly; prod's nodes are on the far side of pfSense and
# must be given whichever address the still-pending routing decision produces). Hardcoding either
# one authors a manifest that is silently wrong in the other environment. Declared in
# clusters/base/cluster-vars.yaml and patched by BOTH overlays; written BARE (it is a real knob,
# not a shell expansion), so StrictPostBuildSubstitutions fails the tier if an overlay forgets it.
BULK_NFS_SERVER = "${BULK_NFS_SERVER}"
# Mount options for every bulk NFS volume. These are a PLATFORM decision, not a per-service one,
# so they live here rather than on the ServiceDefinition surface:
# nfsvers=4.1 NFSv4 carries mount+lock IN the protocol, so it needs ONE port (tcp/2049) and no
# rpcbind/111, mountd or statd. That is what makes the prod path a single pfSense
# rule instead of four, and it is why v4 is chosen over v3. The server advertises
# +3 +4 +4.1 +4.2; 4.1 is the conservative floor every node kernel already speaks.
# hard A `soft` mount turns a server hiccup into a short read / EIO handed to the
# application, which for a media library and an archive is silent corruption.
# `hard` blocks instead. The cost is explicit: an NFS outage HANGS I/O rather than
# failing it, so an outage shows up as stuck pods, not as damaged files. Chosen
# deliberately -- hung is recoverable, truncated is not.
# noatime The datasets are relatime, so every read would otherwise generate a metadata
# WRITE back to the pool. Plex walking an 8 T library must not do that.
# timeo=600/retrans=2 the TCP defaults, stated explicitly so a future reader knows the retry
# budget was considered rather than inherited.
BULK_NFS_MOUNT_OPTIONS = ["nfsvers=4.1", "hard", "noatime", "timeo=600", "retrans=2"]
GATEWAY_NAME = "shared"
GATEWAY_NS = "gateway"
# ExternalDNS per-resource scoping (R-DNS-03; see _externaldns_route_annotations for the full WHY
# and for the v0.21.0 source citation). The KEY is external-dns's own `ControllerKey`
# (annotationPrefix + "controller"); external-dns skips any source object whose value is PRESENT
# and != `ControllerValue` ("dns-controller"), so the VALUE only has to be some other string.
# `none` is chosen because it reads as intent in `kubectl get httproute -o yaml` rather than as a
# typo — "no DNS controller owns this route" — and because the one value that must never be
# emitted here is external-dns's own, which a literal `none` cannot drift into.
EXTERNALDNS_CONTROLLER_KEY = "external-dns.alpha.kubernetes.io/controller"
EXTERNALDNS_CONTROLLER_DISOWN = "none"
# The value external-dns claims for itself. Held here ONLY so the renderer and the L0 guard can
# assert the disown value is not it; never emitted.
EXTERNALDNS_CONTROLLER_CLAIMED = "dns-controller"
KP_RELEASE = "kube-prometheus-stack" # ServiceMonitor/PrometheusRule discovery label
MONITORING_NS = "monitoring" # where Prometheus runs; the only admitted scrape source
# The kube-state-metrics series a workload's readiness IS (r_prometheusrule). kube-state-metrics
# is scraped by the bundled kube-prometheus-stack ServiceMonitor, i.e. these exist for EVERY
# workload in the cluster with no per-service exporter — which is the whole point of using them
# for `slis: [up]`. Each entry is (metric, label naming the object). VERIFIED on prod 2026-09-13:
# kube_deployment_status_replicas_available returned a series for all ten deployed app
# Deployments, and kube_statefulset_status_replicas_ready{namespace="keycloak",
# statefulset="keycloak"} returned 2.
LIVENESS_METRICS = {
"Deployment": ("kube_deployment_status_replicas_available", "deployment"),
"StatefulSet": ("kube_statefulset_status_replicas_ready", "statefulset"),
"DaemonSet": ("kube_daemonset_status_number_available", "daemonset"),
}
# A KubeVirt VMI is not a replica count: kubevirt_vmi_info carries the phase as a LABEL, so
# liveness is the EXISTENCE of a running-phase series. The value is lowercase `running` — a
# capital-R matcher selects nothing and is just as inert (the same trap that made
# kubevirt_vmi_phase_count{phase="Failed"} a no-op; verified on prod, where the one live VMI
# reports phase="running").
VMI_METRIC = "kubevirt_vmi_info"
VMI_RUNNING_PHASE = "running"
# 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")
# --- forward-auth (Option B) per-app oauth2-proxy sidecar ------------------------------
# A `forward-auth` app has a bare web UI with no usable native OIDC (PiHole, *arr, Jackett,
# qBittorrent). Cilium 1.16.x cannot inject ext_authz into the shared Gateway's Envoy config
# (docs/plans/cilium-native-externalauth-evaluation.md; platform/oauth2-proxy/README.md), so
# the platform's forward-auth CEC is an UNWIRED template. Instead each forward-auth app is
# gated by a PER-POD oauth2-proxy REVERSE-PROXY sidecar (Option B) — EXACTLY the pattern
# platform/crafty proves working: the shared Gateway routes <app>.home -> the sidecar (:4180),
# the sidecar authenticates the browser against Keycloak (homelab realm, the app's auth.groups)
# and only THEN proxies to the co-located app UI at http://127.0.0.1:<uiport>. The gate is IN
# the data path (fail-closed: no valid session in an authorized group => no upstream; R-IDP-02).
OAUTH2_PROXY_IMAGE = "quay.io/oauth2-proxy/oauth2-proxy:v7.8.1"
OAUTH2_PROXY_PORT = 4180
KEYCLOAK_ISSUER = f"https://keycloak.{ZONE}/realms/homelab"
# =============================================================================
# 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", [])
s.setdefault("inClusterClients", [])
# 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", [])
# metricsEndpoint is NEVER defaulted (schema 1.6.0). `metrics: true` says the platform
# collects signals FOR this service; it cannot say the workload SERVES a Prometheus
# endpoint, because for most of this catalog it does not (see r_servicemonitor). When the
# endpoint IS declared, its name/path/interval carry defaults so the common case is
# `metricsEndpoint: {port: 9100}`.
ep = obs.get("metricsEndpoint")
if ep:
obs["metricsEndpoint"] = {
**{"name": "metrics", "path": "/metrics", "interval": "30s"},
**ep,
}
# livenessWorkload: WHICH Kubernetes object's readiness IS this service being up. Defaulted
# from `kind` — a container renders a Deployment named after the service, a vm renders a
# VirtualMachine whose VMI carries the same name — and overridable for the case where the
# object that actually runs on the cluster is not the one this SD renders (keycloak: the
# Keycloak operator runs a StatefulSet, so a Deployment-shaped rule would `absent()` forever).
lw = dict(obs.get("livenessWorkload") or {})
lw.setdefault("kind", "VirtualMachineInstance" if s["kind"] == "vm" else "Deployment")
lw.setdefault("name", s["name"])
obs["livenessWorkload"] = lw
s["observability"] = obs
# storage entry class default is longhorn (replicated tier); readOnly default is False
# (grant-nothing-extra is expressed by SETTING it, so the default stays back-compatible and
# every pre-existing golden renders byte-identical).
s["storage"] = [{**{"class": "longhorn", "readOnly": False}, **v} for v in s["storage"]]
# dependency protocol default is tcp.
s["dependencies"] = [{**{"protocol": "tcp"}, **d} for d in s["dependencies"]]
# in-cluster client protocol default is tcp (as for a dependency); its `name` — which only
# ever appears in the rendered POLICY name — defaults to the namespace it admits, so the
# common case declares just `namespace` + `port` and reads like a dependency edge.
s["inClusterClients"] = [
{**{"protocol": "tcp", "name": c.get("namespace", "")}, **c}
for c in s["inClusterClients"]
]
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 _bulk_access_mode(vol: dict) -> str:
"""A bulk volume's access mode follows its read/write posture.
Every bulk volume is `...Many`: the SAME export is consumed by several namespaces at once
(plex, sonarr and radarr all sit on /mnt/Storage/plex) and by several nodes at once, which is
precisely what ReadWriteOnce cannot express. WHICH `Many` it is says whether this consumer
writes -- so `kubectl get pv` reports the intent, not merely the topology.
"""
return "ReadOnlyMany" if vol.get("readOnly") else "ReadWriteMany"
def r_bulk_pvs(s: dict) -> list:
"""One STATIC PersistentVolume per `class: bulk` storage entry (docs/04 bulk tier).
WHY A PV AT ALL, RATHER THAN AN INLINE `nfs:` POD VOLUME
-------------------------------------------------------
An inline volume is fewer objects and it would work. It is rejected for four reasons, in
descending order of what they would cost to discover later:
1. RECLAIM POLICY. `persistentVolumeReclaimPolicy: Retain` is a declared, reviewable
property of a PV. There is no equivalent for an inline volume -- and the data behind
these mounts is 8.33 TiB of media and a 267 GiB archive that ALSO has live non-k3s
consumers. A statement in git that nothing in this cluster may reclaim it is worth an
object.
2. READ-ONLY LIVES IN ONE PLACE. `spec.nfs.readOnly: true` becomes a real `ro` mount option
for every pod that ever binds this volume. The same intent expressed on an inline volume
is one Deployment edit away from being dropped silently, with no diff anywhere near the
storage. (It is still a CLIENT-side guarantee -- the export cannot distinguish plex from
sonarr, since both may run on any node -- but it is one that lives in one reviewable
place.)
3. VISIBILITY. `kubectl get pv,pvc` is where an operator looks for "what external storage is
attached, and who holds it". An inline volume is invisible there, and invisible to
anything that reasons over PVCs.
4. IT IS THE CONTRACT THE REPO ALREADY HAS. Every other storage entry is a PVC; the
renderer, the Kyverno ServiceDefinition policy and the backup wiring all key on that. An
inline volume would be a second, parallel storage model inside the same renderer.
CROSS-NAMESPACE. A PV binds 1:1 to a PVC, so "plex, sonarr and radarr share one export" is NOT
one shared PV -- it is three PVs, one per namespace, each naming the same server+path. That is
the correct model rather than a workaround: the sharing happens on the NFS server, which is
built for concurrent clients, and each namespace keeps its own independently reviewable access
mode (plex ReadOnlyMany, the arrs ReadWriteMany). One shared PV would have forced a single
posture on all three.
BINDING IS PINNED FROM BOTH ENDS -- `claimRef` here and `volumeName` on the PVC. Either alone
leaves a race in which some other PVC of matching size and mode grabs the volume first;
together the binding is deterministic. `storageClassName: ""` (explicitly empty, NOT the `bulk`
StorageClass) is what tells the binder this volume is statically provisioned: it suppresses the
default-StorageClass annotation and guarantees no provisioner is ever asked to create anything.
It also sidesteps a trap -- a StorageClass's `provisioner` is IMMUTABLE, so re-pointing the
existing `bulk` class (whose stub names an NFS CSI driver that is not installed) at this data
would make Flux fail forever on that one object.
"""
out = []
for vol in s["storage"]:
if vol.get("class") != "bulk":
continue
name = f"{s['name']}-{vol['name']}"
# `export` and `size` are both structurally required of a bulk entry, and the JSON
# Schema only enforces the first. A PersistentVolume with no `capacity` is rejected by
# the API server, so a missing size must fail HERE with a sentence a human can act on
# rather than as a bare KeyError three frames down.
for field, why in (
("export", "the absolute NFS export path this volume attaches"),
("size", "the PV capacity (advisory, but the API requires the field)"),
):
if not vol.get(field):
raise ValueError(
f"{s['name']}: storage entry {vol['name']!r} has class 'bulk' but no "
f"'{field}' — {why}. A bulk volume attaches storage that already exists "
f"outside the cluster, so it cannot be rendered without it."
)
nfs = {"server": BULK_NFS_SERVER, "path": vol["export"]}
if vol.get("readOnly"):
nfs["readOnly"] = True
spec = {
"capacity": {"storage": vol["size"]},
"accessModes": [_bulk_access_mode(vol)],
# Retain: deleting the PVC must NEVER be able to reach the data. There is nothing to
# "reclaim" here in any case -- the volume was not provisioned by the cluster.
"persistentVolumeReclaimPolicy": "Retain",
"storageClassName": "",
"mountOptions": list(BULK_NFS_MOUNT_OPTIONS),
"nfs": nfs,
"claimRef": {"namespace": s["name"], "name": name},
}
out.append(
{
"apiVersion": "v1",
"kind": "PersistentVolume",
"metadata": {
"name": name,
"labels": {"app.kubernetes.io/name": s["name"]},
},
"spec": spec,
}
)
return out
def r_pvcs(s: dict) -> list:
"""One PVC per storage entry (R-STATE-01/02/04).
longhorn -> 3-replica replicated tier, dynamically provisioned, ReadWriteOnce.
bulk -> the STATIC external NFS PV rendered by r_bulk_pvs, pre-bound by volumeName.
"""
out = []
for vol in s["storage"]:
cls = vol.get("class", "longhorn")
if cls == "bulk":
# Mirror image of the PV: the same explicitly-empty class (so nothing dynamically
# provisions), the same access mode, and volumeName completing the two-sided binding.
out.append(
{
"apiVersion": "v1",
"kind": "PersistentVolumeClaim",
"metadata": {
"name": f"{s['name']}-{vol['name']}",
"namespace": s["name"],
"labels": {"app.kubernetes.io/name": s["name"]},
},
"spec": {
"accessModes": [_bulk_access_mode(vol)],
"storageClassName": "",
"volumeName": f"{s['name']}-{vol['name']}",
"resources": {"requests": {"storage": vol["size"]}},
},
}
)
continue
spec = {
"accessModes": ["ReadWriteOnce"],
"storageClassName": "longhorn",
}
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']}")
mount = {"name": vol["name"], "mountPath": mount_path}
# subPath mounts a subtree of the volume instead of its root. Emitted only when declared,
# so every pre-existing golden stays byte-identical.
if vol.get("subPath"):
mount["subPath"] = vol["subPath"]
# readOnly on the mount as well as on the PV's nfs source. Belt and braces on purpose: the
# PV field is what makes the kernel mount `ro`; this one is what a reviewer reading the
# Deployment sees. An edit that drops one still leaves the other.
if vol.get("readOnly"):
mount["readOnly"] = True
volume_mounts.append(mount)
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]}
# forward-auth (Option B): append the per-pod oauth2-proxy reverse-proxy sidecar AFTER the app
# container (app stays containers[0]). The shared Gateway routes to the sidecar (:4180); it gates
# the co-located UI against Keycloak and only then proxies to http://127.0.0.1:<uiport>. `oidc`
# apps (native client) and `native`/`none` render byte-identical (no sidecar).
if _needs_oauth2_sidecar(s):
pod_spec["containers"].append(_oauth2_sidecar_container(s))
# podSecurityContext (schema-optional): the k8s PodSecurityContext is emitted VERBATIM onto the
# pod spec. Its primary use is `fsGroup`, which makes a mounted Longhorn PVC group-writable by a
# non-root image (e.g. semaphore runs as UID 1001 and cannot otherwise write its config PVC).
# Absent => the key is never set, so apps that declare none render byte-identical.
if s.get("podSecurityContext"):
pod_spec["securityContext"] = s["podSecurityContext"]
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)
# A declared metrics endpoint becomes a NAMED Service port, because that name is the only
# thing a ServiceMonitor endpoint can select on. Appended BELOW rather than inside
# _service_ports so it never leaks into r_gateway_ingress_policy's north-south allow list —
# the metrics port is for the monitoring namespace, not for the Gateway.
ep = s["observability"].get("metricsEndpoint")
if not ports and not ep:
return None
# forward-auth (Option B): the shared Gateway must reach the oauth2-proxy sidecar, so front it
# with a `authproxy` :4180 Service port (the HTTPRoute backendRef targets this). The app UI/API
# port(s) stay in the Service for EAST-WEST callers (e.g. arr -> qbittorrent:8080 / jackett:9117);
# north-south to those app ports is denied at the Gateway (see r_gateway_ingress_policy).
if _needs_oauth2_sidecar(s):
ports = ports + [
{"name": "authproxy", "port": OAUTH2_PROXY_PORT, "targetPort": OAUTH2_PROXY_PORT}
]
if ep:
ports = ports + [
{"name": ep["name"], "port": ep["port"], "targetPort": ep["port"]}
]
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.20.1) 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
# forward-auth (Option B): the ONLY north-south path is the oauth2-proxy sidecar. Open the
# Gateway to :4180 ONLY — never the app UI port — so the un-gated UI is unreachable from the
# Gateway (fail-closed; R-IDP-02). Every other app opens its exposed service port(s) as before.
if _needs_oauth2_sidecar(s):
allow_ports = [{"port": str(OAUTH2_PROXY_PORT), "protocol": "TCP"}]
else:
allow_ports = [
{"port": str(p["port"]), "protocol": "TCP"} for p in _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": allow_ports}],
}
],
},
}
def r_in_cluster_client_policies(s: dict) -> list:
"""East-west allow: let a declared IN-CLUSTER client reach this service DIRECTLY (R-NET-02).
`ingressRoutes` opens the NORTH-SOUTH path only — r_gateway_ingress_policy admits the
reserved `ingress` identity (the Gateway proxy) and nothing else — and `dependencies`
compiles edges only BETWEEN CATALOG SERVICES (both halves are rendered from a catalog
entry's own declaration). A PLATFORM CONTROLLER that dials this service's ClusterIP
directly therefore had no authorable form at all, and under default-deny (R-NET-01) it is
DROPPED.
That gap is not theoretical: prod's `flux-system` source-controller clones the platform's
own repo from the in-cluster Gitea Service (clusters/prod/kustomization.yaml pins
`url: http://gitea.gitea.svc.cluster.local:3000/...` so prod's GitOps depends on neither
public DNS, Cloudflare, its own Gateway nor ${BASE_DOMAIN}). gitea-allow-gateway-ingress
admits `fromEntities: [ingress]` only, so the clone was denied, the source could not
reconcile — and the HTTPRoute that would have served a public name instead lives in a
DOWNSTREAM tier that cannot apply while the source is down. Source down -> routes never
rename -> source stays down. It had to be broken with a live `kubectl apply`, i.e. with a
cluster object outside git, which one `kubectl delete` or one rebuild would take away.
Each entry renders ONE least-privilege ingress CiliumNetworkPolicy, shaped exactly like the
reciprocal-dependency half (r_reverse_dependency_policies) but selecting the client by
NAMESPACE alone, because a platform controller's pod labels are the upstream chart's to
change and are not part of this platform's contract: `fromEndpoints` carrying ONLY the
reserved `k8s:io.kubernetes.pod.namespace` label, `toPorts` carrying ONLY the declared
port/protocol. Never `fromEntities: all`, and never the whole service. Additive: the
namespace default-deny baseline keeps denying every undeclared source and port.
The policy NAME is `<service>-allow-<name>-ingress`, `name` defaulting to the namespace —
it is operator-facing (`kubectl get cnp`, a Hubble drop), and letting an author pin it is
what allows an allowance that already had to be applied live to be authored under its
EXISTING name, so Flux adopts that object in place rather than creating a near-duplicate
beside it and leaving the live one unmanaged.
"""
out = []
for client in s["inClusterClients"]:
proto = client.get("protocol", "tcp").upper()
out.append(
{
"apiVersion": "cilium.io/v2",
"kind": "CiliumNetworkPolicy",
"metadata": {
"name": f"{s['name']}-allow-{client['name']}-ingress",
"namespace": s["name"],
"labels": {"app.kubernetes.io/name": s["name"]},
},
"spec": {
"endpointSelector": {
"matchLabels": {"app.kubernetes.io/name": s["name"]}
},
"ingress": [
{
"fromEndpoints": [
{
"matchLabels": {
"k8s:io.kubernetes.pod.namespace": client[
"namespace"
]
}
}
],
"toPorts": [
{
"ports": [
{"port": str(client["port"]), "protocol": proto}
]
}
],
}
],
},
}
)
return out
def _route_meta(s: dict, route: dict) -> dict:
"""HTTPRoute metadata: auth wiring hints PLUS the ExternalDNS scoping derived from
`exposure` (_externaldns_route_annotations — R-DNS-03: an `internal` service renders no
public DNS record, and now cannot)."""
meta = {
"name": f"{s['name']}-{route['port']}",
"namespace": s["name"],
"labels": {"app.kubernetes.io/name": s["name"]},
}
ann = {**_auth_annotations(s), **_externaldns_route_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.
"""
egress = [{"toEntities": ["world"]}]
# SSO apps (oidc/forward-auth) reach Keycloak at keycloak.home, which coredns split-horizon
# resolves to the shared Gateway VIP (an L2-announced LoadBalancer IP Cilium classifies as
# `world`), so the pod hairpins THROUGH the Gateway. Under Cilium >=1.16.8/1.17.2 (the
# CVE-2025-30162 fix, present in 1.20), the Gateway's L7 proxy re-evaluates THIS pod's egress
# policy against Keycloak's REAL backend cluster identity (enforce_policy_on_l7lb) — `world`
# matches the VIP at L3/L4 but NOT the resolved keycloak identity at L7, so OIDC discovery/
# token 403s ("Access denied") without an explicit allow to the keycloak backend. Scope it to
# the apps that actually talk to Keycloak (auth mode oidc or forward-auth).
if s.get("auth", {}).get("mode") in ("oidc", "forward-auth"):
egress.append(
{
"toEndpoints": [
{
"matchLabels": {
"io.kubernetes.pod.namespace": "keycloak",
"app": "keycloak",
}
}
]
}
)
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": egress,
},
}
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"]:
# forward-auth (Option B): the Gateway backendRef targets the oauth2-proxy sidecar (:4180),
# NOT the app UI port — so the gate is in the data path and the bare UI is never served
# un-authenticated. Other apps route straight to their exposed app port.
backend_port = OAUTH2_PROXY_PORT if _needs_oauth2_sidecar(s) else route["port"]
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": backend_port}
],
}
],
},
}
)
return out
def _needs_oauth2_sidecar(s: dict) -> bool:
"""True iff this service gets a per-pod oauth2-proxy reverse-proxy sidecar (Option B).
A `forward-auth` app with an exposed UI has no usable native OIDC, so its Gateway route is
gated IN the data path by a co-located oauth2-proxy sidecar (crafty's proven pattern). `oidc`
apps authenticate themselves (native Keycloak client) and `native`/`none` are unchanged."""
return s["auth"]["mode"] == "forward-auth" and bool(s["ingressRoutes"])
def _oauth2_sidecar_container(s: dict) -> dict:
"""The oauth2-proxy REVERSE-PROXY sidecar (Option B) that gates a forward-auth app's UI.
Mirrors platform/crafty/deployment.yaml's `oauth2-proxy` container EXACTLY, adapted per app:
the upstream is the co-located app UI over plain HTTP (http://127.0.0.1:<uiport>) — the *arr /
PiHole / Jackett / qBittorrent UIs are HTTP, so (unlike crafty's self-signed HTTPS :8443) there
is NO --ssl-upstream-insecure-skip-verify. --allowed-group is emitted once per declared
auth.group (R-IDP-06: a valid session is not enough — the user must be in an authorized group,
else /oauth2 returns 403). client-id/secret + cookie-secret come by name from the SOPS-encrypted
`<app>-oauth2-proxy` Secret (R-IDP-04). Single provider => --skip-provider-button 302s straight
to Keycloak. The sidecar->app hop is pod-local (127.0.0.1) and not subject to network policy."""
route = s["ingressRoutes"][0]
host = route["host"]
upstream_port = route["port"]
args = [
f"--http-address=0.0.0.0:{OAUTH2_PROXY_PORT}",
"--reverse-proxy=true",
"--provider=oidc",
f"--oidc-issuer-url={KEYCLOAK_ISSUER}",
f"--upstream=http://127.0.0.1:{upstream_port}",
"--email-domain=*",
"--scope=openid email profile",
"--oidc-groups-claim=groups",
]
# R-IDP-06: authorize by group. One flag per declared group (fail-closed to 403 otherwise).
for group in s["auth"].get("groups", []):
args.append(f"--allowed-group={group}")
args += [
"--skip-provider-button=true",
f"--redirect-url=https://{host}/oauth2/callback",
f"--cookie-domain=.{ZONE}",
f"--whitelist-domain=.{ZONE}",
"--cookie-secure=true",
"--pass-access-token=true",
"--set-xauthrequest=true",
]
secret = f"{s['name']}-oauth2-proxy"
return {
"name": "oauth2-proxy",
"image": OAUTH2_PROXY_IMAGE,
"imagePullPolicy": "IfNotPresent",
"args": args,
"env": [
{
"name": "OAUTH2_PROXY_CLIENT_ID",
"valueFrom": {"secretKeyRef": {"name": secret, "key": "client-id"}},
},
{
"name": "OAUTH2_PROXY_CLIENT_SECRET",
"valueFrom": {"secretKeyRef": {"name": secret, "key": "client-secret"}},
},
{
"name": "OAUTH2_PROXY_COOKIE_SECRET",
"valueFrom": {"secretKeyRef": {"name": secret, "key": "cookie-secret"}},
},
],
"ports": [{"name": "http", "containerPort": OAUTH2_PROXY_PORT, "protocol": "TCP"}],
"resources": {
"requests": {"cpu": "25m", "memory": "64Mi"},
"limits": {"memory": "128Mi"},
},
"securityContext": {
"allowPrivilegeEscalation": False,
"readOnlyRootFilesystem": True,
"capabilities": {"drop": ["ALL"]},
},
"livenessProbe": {
"httpGet": {"path": "/ping", "port": "http"},
"initialDelaySeconds": 5,
"periodSeconds": 15,
},
"readinessProbe": {
"httpGet": {"path": "/ping", "port": "http"},
"initialDelaySeconds": 5,
"periodSeconds": 15,
},
}
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 _externaldns_route_annotations(s: dict) -> dict:
"""ExternalDNS scoping, rendered FROM `exposure` (renderer-contract §exposure; R-DNS-03).
`internal` renders NO public DNS record. That sentence was in the contract from the start and
nothing enforced it: ExternalDNS watched EVERY HTTPRoute the shared Gateway carried, so an
`internal` service's route published `<name>.${BASE_DOMAIN}` into the PUBLIC authoritative
Cloudflare zone, answering with the Gateway VIP — an RFC1918 address. Six of them did
(jackett, pihole, qbittorrent, radarr, semaphore, sonarr), which published the internal
subnet, the Gateway VIP and a full service inventory to anyone who asked, and made every one
of those names a DNS-rebind primitive that the SOHO router strips on the way back in.
THE MECHANISM, verified against the DEPLOYED version (chart >=1.21.0 <1.22.0, app v0.21.0 —
`registry.k8s.io/external-dns/external-dns:v0.21.0`, read off the live prod Deployment), not
from memory. external-dns v0.21.0 `source/annotations/processors.go`:
func IsControllerMismatch(entity metav1.ObjectMetaAccessor, rType string) bool {
value, ok := entity.GetObjectMeta().GetAnnotations()[ControllerKey]
if ok && value != ControllerValue { ...; return true }
return false
}
with `ControllerKey = "external-dns.alpha.kubernetes.io/controller"` and
`ControllerValue = "dns-controller"`. `source/gateway.go` calls it per route inside
`Endpoints()`, BEFORE any hostname is resolved:
if annotations.IsControllerMismatch(meta, src.rtKind) { continue }
so an HTTPRoute carrying any other value is not merely unpublished — it never becomes an
endpoint at all, for any hostname it names. `source/service.go` applies the identical check,
so the same annotation scopes the `service` source too if a rendered Service ever grows an
external-dns hostname.
WHY THIS IS STRUCTURAL AND NOT INCIDENTAL. The annotation is derived HERE, from `exposure`,
with no per-service list anywhere: every HTTPRoute this renderer emits for an `internal`
service carries it, and ci/render-golden.sh proves `apps/` reproduces byte-for-byte from
`catalog/`, so an `internal` service cannot EXIST in the applied tree without it. Flipping
`exposure: internal` to `public` is the only edit that can publish a name, which is exactly
the operator-facing declaration R-DNS-03 is written in terms of. The L0 check
conformance/tests/l0-static/internal-services-publish-no-public-dns re-proves all of that on
every gate run, including that this function reads `exposure` rather than naming services.
WHY NOT A CLUSTER-WIDE OPT-IN FILTER (`--label-filter`, chart `labelFilter`). It exists, and
v0.21.0's own `--help` names both sources in use as supporting it ("currently supported by
source types crd, gateway-httproute, ... service and ambassador-host"), so a "publish only
what is labelled" posture is available and is STRONGER by default. It is not taken here
because its blast radius is not this defect's: seven published names come from hand-authored
platform/apps manifests that carry no `exposure` field and that this renderer never sees
(auth/oauth2-proxy, crafty + crafty-mc, grafana, headlamp, the homepage apex, unifi). An
opt-in filter DELETES all seven the moment it is set, and whether each should stay published
is an operator decision, not a renderer one. Per-resource exclusion has exactly the blast
radius of the reported defect. The filter remains the available hardening once those seven
are adjudicated — see platform/externaldns/helmrelease.yaml.
"""
if s["exposure"] == "internal":
return {EXTERNALDNS_CONTROLLER_KEY: EXTERNALDNS_CONTROLLER_DISOWN}
return {}
def r_servicemonitor(s: dict):
"""ServiceMonitor — rendered ONLY when the SD says WHERE metrics are served.
THIS FUNCTION USED TO RENDER A MONITOR UNCONDITIONALLY FROM `metrics: true`, hardcoding
`endpoints: [{port: metrics}]`. No Service this renderer emits had ever defined a port of
that name, so all twelve monitors discovered ZERO targets: no `up{job="<app>"}` series
existed anywhere, `up{job="<app>"} == 0` could not fire, and `PrometheusTargetsMissing`
could not see the hole either, because a target that was never discovered has no `up` series
to be 0. Verified on prod 2026-09-13: `count by (job) (up)` listed 24 jobs and not one of
the twelve appeared; `up{job="plex"}` and `up{job="gitea"}` both returned nothing.
THE ROOT CAUSE WAS NOT A TYPO — it was that `metrics: true` cannot state a fact the workload
does not have. Plex, Pi-hole, Jackett, qBittorrent, Sonarr, Radarr, Seerr, Semaphore and
Nextcloud serve no Prometheus endpoint at all without a separate exporter; Gitea can serve
one but only once its app.ini enables it. So the port name was never the missing piece. A
monitor is now rendered only from `observability.metricsEndpoint`, which names the port the
workload actually serves — and that same declaration renders the Service port (r_service) and
the monitoring-namespace ingress allow (r_monitoring_ingress_policy), so the three halves of
a working scrape cannot be authored apart.
`metrics: true` still means something without it: logs ship to Loki via the node agent, and
liveness comes from kube-state-metrics (r_prometheusrule) — neither needs a per-service CR.
"""
obs = s["observability"]
if not obs["metrics"]:
return None
ep = obs.get("metricsEndpoint")
if not ep:
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": ep["name"], "path": ep["path"], "interval": ep["interval"]}
],
},
}
def r_monitoring_ingress_policy(s: dict):
"""Let the monitoring namespace reach a declared metrics port (R-NET-02, R-OBS-01).
renderer-contract.md has always said the default-deny baseline carries "a standing allow ...
for observability scraping from the monitoring namespace" — and no renderer output ever did.
Every app namespace ships `ingress: []`, so a scrape of an app metrics port would be DROPPED
by policy even with the Service port and the monitor endpoint in agreement. That is the same
trap Longhorn's own NetworkPolicies set (platform/longhorn/networkpolicy-metrics.yaml exists
for exactly this reason), and it is a THIRD independent way for a metrics SLI to be inert.
Rendered only alongside a declared metricsEndpoint, and it opens ONLY that port to ONLY the
monitoring namespace: an allow for a port nothing serves would be a standing hole for no
benefit.
"""
ep = s["observability"].get("metricsEndpoint")
if not ep or not s["observability"]["metrics"]:
return None
return {
"apiVersion": "cilium.io/v2",
"kind": "CiliumNetworkPolicy",
"metadata": {
"name": f"{s['name']}-allow-monitoring-scrape",
"namespace": s["name"],
"labels": {"app.kubernetes.io/name": s["name"]},
},
"spec": {
"endpointSelector": {"matchLabels": {"app.kubernetes.io/name": s["name"]}},
"ingress": [
{
"fromEndpoints": [
{"matchLabels": {"k8s:io.kubernetes.pod.namespace": MONITORING_NS}}
],
"toPorts": [
{"ports": [{"port": str(ep["port"]), "protocol": "TCP"}]}
],
}
],
},
}
def _up_sli_rule(s: dict) -> dict:
"""`slis: [up]` -> a `<name>-down` alert against a signal THAT EXISTS on this cluster.
IT USED TO READ `up{job="<name>"} == 0`. That is the textbook expression and it was inert
for all twelve services in the catalog, because `up` only exists for a target Prometheus
actually DISCOVERED and the rendered ServiceMonitor discovered none (see r_servicemonitor).
Worse than a wrong threshold: an inert rule and a correct rule are indistinguishable on a
healthy cluster, and the generic safety net cannot see the difference either — a target that
was never discovered has no `up` series for PrometheusTargetsMissing to find at 0.
Rewriting the SLI against the app's own /metrics was NOT an option for most of this catalog:
Plex, Pi-hole, Jackett, qBittorrent, Sonarr, Radarr, Seerr, Semaphore and Nextcloud export no
Prometheus metrics without a separate exporter sidecar per app. So the `up` SLI is expressed
against kube-state-metrics, which IS scraped and DOES carry a series for every workload — the
same idiom KeycloakDown in platform/monitoring/alerts.yaml has used since it was fixed.
The `or absent(...)` half is deliberate and load-bearing: it makes the DISAPPEARANCE of the
signal fire the alert instead of silencing it. A rule that can only fire on `== 0` is a rule
that goes quiet when the workload object (or kube-state-metrics) vanishes, which is precisely
how the old expression failed. It is also why this rule cannot be permanently firing where the
workload exists: the series is present and non-zero on a healthy cluster (verified on prod).
"""
lw = s["observability"]["livenessWorkload"]
kind, obj = lw["kind"], lw["name"]
ns = s["name"]
if kind == "VirtualMachineInstance":
sel = f'{VMI_METRIC}{{namespace="{ns}", name="{obj}", phase="{VMI_RUNNING_PHASE}"}}'
expr = f"absent({sel})"
desc = (
f"{s['name']}: no KubeVirt VMI in phase {VMI_RUNNING_PHASE} for 5m "
f"(SLI: up, via {VMI_METRIC})."
)
else:
try:
metric, label = LIVENESS_METRICS[kind]
except KeyError:
raise ValueError(
f"{s['name']}: observability.livenessWorkload.kind {kind!r} has no liveness "
f"metric; known kinds: {sorted(LIVENESS_METRICS)} + VirtualMachineInstance"
) from None
sel = f'{metric}{{namespace="{ns}", {label}="{obj}"}}'
expr = f"{sel} == 0 or absent({sel})"
desc = (
f"{s['name']}: its {kind} has no available/ready replica for 5m, or the signal is "
f"gone entirely (SLI: up, via {metric})."
)
return {
"alert": f"{s['name']}-down".replace("_", "-"),
"expr": expr,
"for": "5m",
"labels": {"severity": "critical", "service": s["name"]},
"annotations": {
"summary": f"{s['name']} is down",
"description": desc,
},
}
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(_up_sli_rule(s))
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/v1",
"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/v1",
"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},
},
],
},
},
},
}
# =============================================================================
# Substitution safety (R-ENV-02 / R-GITOPS-02) — refuse to emit a manifest that
# Flux's envsubst would silently eat before the workload ever runs.
# =============================================================================
# Every manifest this renderer writes lands under apps/wave-<N>/, and EVERY apps-wave-<N>
# Flux Kustomization declares `postBuild.substituteFrom: cluster-vars` (see
# flux_wave_kustomization_text below). Flux therefore runs envsubst over the WHOLE rendered
# document — including any shell script carried verbatim in a `lifecycle` hook or an env
# value — at reconcile time, long before a container sees it. The escape contract is stated
# in clusters/base/apps-vms.yaml and proven with `flux envsubst`:
#
# `${VAR}` -> SUBSTITUTED (empty when unset) => a shell brace must be written `$${VAR}`,
# which envsubst emits back as `${VAR}`.
# `$VAR` -> left alone => bare shell vars need NO escaping.
# `$(cmd)` -> left alone => command substitution needs NO escaping.
# `${A[0]}` -> HARD ERROR ("missing closing brace"), failing the whole Kustomization.
#
# This bit us for real (fixed 2026-09-01): gitea/nextcloud guarded their whole Keycloak OIDC
# registration with `if [ -z "${..._OIDC_CLIENT_SECRET:-}" ]`. Neither name is a cluster-var,
# so envsubst rendered the guard as `if [ -z "" ]` — ALWAYS TRUE — and every boot skipped the
# registration entirely. Flux's STRICT substitution mode cannot catch this class: it
# deliberately EXEMPTS the default-operator forms (`${V:-}`, `${V:=x}`, `${V:?}`), which is
# exactly the form a defensive shell script uses. So the renderer checks it itself.
CLUSTER_VARS_PATH = (
Path(__file__).resolve().parent.parent / "clusters" / "base" / "cluster-vars.yaml"
)
# A run of `$` followed by `{...}`. envsubst consumes `$$` as a literal `$`, so a run of N
# dollars leaves the brace SUBSTITUTED when N is odd (`${X}`, `$$${X}`) and LITERAL when N is
# even (`$${X}` — the escape). Deliberately greedy about the body so malformed/indexed forms
# (`${A[0]}`) are caught too rather than skipped.
_BRACED_EXPANSION_RE = re.compile(r"(\$+)\{([^{}]*)\}")
_PLAIN_IDENT_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*\Z")
_CLUSTER_VARS_CACHE: dict = {}
def cluster_vars_keys(path: Path | None = None) -> set:
"""The declared cluster-vars keys — the ONLY names envsubst may legitimately expand.
Read from clusters/base/cluster-vars.yaml (the single knob surface, R-ENV-02) so the
renderer never carries a stale hardcoded copy. Unreadable/absent => empty set, i.e.
FAIL-CLOSED: with no known knobs, every braced expansion must be escaped for the shell.
"""
key = str(path or CLUSTER_VARS_PATH)
if key not in _CLUSTER_VARS_CACHE:
try:
doc = yaml.safe_load(Path(key).read_text())
data = doc.get("data") if isinstance(doc, dict) else None
_CLUSTER_VARS_CACHE[key] = set(data) if isinstance(data, dict) else set()
except Exception: # pragma: no cover - absent/malformed => fail closed
_CLUSTER_VARS_CACHE[key] = set()
return _CLUSTER_VARS_CACHE[key]
def substitution_violations(text: str, known: set | None = None) -> list:
"""Every unescaped `${...}` in `text` that is NOT a bare, declared cluster-vars key.
Returns `(line_number, token, reason)` triples in line order. An expansion is ALLOWED only
when it is exactly `${KNOWN_KEY}` — a deliberate R-ENV-02 knob. Anything else is a bug:
a shell expansion that envsubst will eat (`${V:-}`), a typo'd/undeclared knob that
envsubst will blank, or an indexed form that hard-fails the whole Kustomization.
"""
keys = cluster_vars_keys() if known is None else known
out = []
for lineno, line in enumerate(text.splitlines(), start=1):
for m in _BRACED_EXPANSION_RE.finditer(line):
if len(m.group(1)) % 2 == 0:
continue # even run of `$` => escaped; envsubst emits it back literally
expr = m.group(2)
if _PLAIN_IDENT_RE.match(expr):
if expr in keys:
continue # a declared cluster-vars knob, intentionally substituted
reason = (
f"'{expr}' is not a declared cluster-vars key — envsubst will replace "
"it with the EMPTY STRING"
)
elif "[" in expr:
reason = (
"indexed expansion — envsubst hard-fails with 'missing closing brace' "
"and the whole Kustomization stops reconciling"
)
else:
reason = (
"shell expansion (default-operator/parameter form) — envsubst eats it "
"at reconcile time and STRICT mode exempts this form, so nothing else "
"will catch it"
)
out.append((lineno, m.group(0), reason))
return out
def assert_substitution_safe(text: str, origin: str = "<render>") -> None:
"""Refuse to emit a manifest carrying an unescaped braced expansion (see above).
Raises ValueError listing every offending token. The fix is almost always to escape the
shell brace as `$${...}` (envsubst emits a literal `${...}` for the shell), or — if the
value really is an environment knob — to declare it in clusters/base/cluster-vars.yaml
and write it as a bare `${KNOB}`.
"""
bad = substitution_violations(text)
if not bad:
return
lines = [
f"{origin}: refusing to emit — unescaped `${{...}}` in a manifest that Flux's "
"postBuild.substituteFrom (cluster-vars) will substitute at reconcile time:"
]
for lineno, token, reason in bad:
lines.append(f" line {lineno}: {token} -> {reason}")
lines.append(
" FIX: a SHELL expansion must be escaped `$${...}` (envsubst emits a literal "
"`${...}`); bare `$VAR` / `$(cmd)` need no escaping. A genuine environment knob "
"must be declared in clusters/base/cluster-vars.yaml and written bare `${KNOB}`. "
"Contract: clusters/base/apps-vms.yaml."
)
raise ValueError("\n".join(lines))
# =============================================================================
# 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.)
# The static bulk PV is rendered BEFORE the PVC that names it. Document order inside one
# kustomize build is not an apply-ordering guarantee, but a PVC whose PV does not exist
# yet merely stays Pending and binds on the next reconcile, so this is a readability
# convention rather than a correctness requirement.
for pv in r_bulk_pvs(s):
docs.append(
(f"PersistentVolume {pv['metadata']['name']} - static external NFS (bulk tier)", pv)
)
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,
)
)
for cnp in r_in_cluster_client_policies(s):
docs.append(
(
"CiliumNetworkPolicy in-cluster client ingress allow (R-NET-02): "
f"{cnp['metadata']['name']}",
cnp,
)
)
mon_allow = r_monitoring_ingress_policy(s)
if mon_allow:
docs.append(
(
"CiliumNetworkPolicy monitoring scrape ingress allow (R-NET-02/R-OBS-01): "
f"{mon_allow['metadata']['name']}",
mon_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, origin: str = "<render>") -> str:
"""Deterministic multi-document YAML text (golden-diffable).
Every emitted document is checked against the substitution contract before it is handed
back (assert_substitution_safe): the rendered manifest lands on a Flux path that runs
envsubst over it, so an unescaped `${...}` meant for a SHELL is silently blanked. Raises
ValueError rather than emitting such a manifest.
"""
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")
text = "".join(parts)
assert_substitution_safe(text, origin)
return text
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), origin=str(path)
)
# =============================================================================
# Wave partitioning — dependency-ordered app-tier delivery (WS-APPS wave delivery)
# =============================================================================
# A single monolithic `apps` Flux Kustomization health-gates (wait:true) the ENTIRE app
# set, so one crashlooping workload blocks ALL apps and every new revision. Instead we split
# the tier into dependency-ordered WAVES: each wave is its own Flux Kustomization
# `apps-wave-<N>` that dependsOn wave N-1 (wave 0 dependsOn platform-config), so a failure in
# a later wave can no longer stall an earlier one, while everything WITHIN a wave still
# reconciles in parallel. This module computes the wave assignment from the declared catalog
# dependency graph and emits the per-wave kustomize + Flux Kustomization CRs. It changes only
# WHERE a rendered manifest is placed — never its bytes (the L1 goldens stay green).
CO_DB_SUFFIX = "-db" # a co-located CNPG database for `<svc>` is the node `<svc>-db`.
# Non-catalog app-tier units (they carry NO ServiceDefinition, so compute_waves can't see
# them). Each is anchored to a catalog service; `above=True` forces it into the wave STRICTLY
# ABOVE its anchor because it is a hard runtime prerequisite it cannot start without (e.g. the
# Gitea Actions act_runner registers against + clones from Gitea, so Gitea must be healthy
# first); `above=False` would co-locate it in the anchor's own wave.
EXTRA_UNITS: dict = {
"gitea-actions": {
"anchor": "gitea",
"above": True,
"files": [
"gitea-actions.yaml",
"gitea-actions-runner-secret.sops.yaml",
],
},
# PiHole's DNS exposure: the udp/53 + tcp/53 LoadBalancer Services on a pinned VIP, plus the
# narrow LAN ingress CiliumNetworkPolicy that admits them (apps/wave-0/pihole-dns.yaml).
# `above: False` co-locates it in pihole's OWN wave: these are the pihole pod's own Services and
# its own policy, applied in the same wave as the Deployment they select, exactly as the
# rendered Service and CNPs in pihole.yaml are. It is NOT a runtime prerequisite of anything, so
# there is nothing for a higher wave to gate on.
#
# It is a non-catalog unit because the ServiceDefinition schema cannot express a LoadBalancer
# Service, a pinned lbipam VIP, a sharing key or UDP exposure at all (additionalProperties is
# false, and `ingressRoutes` only renders HTTPRoutes on the L7 Gateway) — the same reason
# apps/vms/unifi/50-service-l4.yaml and platform/crafty/service-l4.yaml are hand-written. The
# split-horizon ANSWERS are in the catalog, as FTLCONF env on catalog/pihole.yaml.
"pihole-dns": {
"anchor": "pihole",
"above": False,
"files": [
"pihole-dns.yaml",
],
},
}
def compute_waves(sds: list) -> dict:
"""Longest-dependency-path wave per node over the catalog dependency DAG.
Nodes = every catalog service ∪ every dependency target. A dependency target that has NO
ServiceDefinition (a co-located `<svc>-db` CNPG database) is a leaf. wave(n) = 0 if n
depends on nothing, else 1 + max(wave(dep)) over its declared dependencies. Because a
co-located DB is a leaf and its consumer declares a dependency ON it, the consumer always
lands in a strictly higher wave — the "DB strictly below its consumer" guarantee falls
straight out of the longest-path rule (no special case). Raises ValueError on a cycle.
"""
edges: dict = {} # node -> set(targets it depends on)
nodes: set = set()
for sd in sds:
if not isinstance(sd, dict) or "name" not in sd:
continue # tolerate non-ServiceDefinition siblings (e.g. render.py itself)
s = apply_defaults(sd)
name = s["name"]
nodes.add(name)
edges.setdefault(name, set())
for dep in s["dependencies"]:
edges[name].add(dep["to"])
nodes.add(dep["to"])
edges.setdefault(dep["to"], set())
wave: dict = {}
visiting: set = set()
def depth(n: str) -> int:
if n in wave:
return wave[n]
if n in visiting:
raise ValueError(f"dependency cycle through {n!r}")
visiting.add(n)
d = 0
for t in sorted(edges.get(n, ())):
d = max(d, depth(t) + 1)
visiting.discard(n)
wave[n] = d
return d
for n in sorted(nodes):
depth(n)
return wave
def _catalog_by_name(sds: list) -> dict:
out = {}
for sd in sds:
if isinstance(sd, dict) and "name" in sd:
s = apply_defaults(sd)
out[s["name"]] = s
return out
def service_wave(sds: list, name: str) -> int:
"""The wave a single service's rendered manifest belongs in (for wave-aware `-o`)."""
return compute_waves(sds).get(name, 0)
def plan_apps_waves(sds: list, present_files: set) -> dict:
"""Map the ONBOARDED app-tier file set into waves -> sorted resource basenames.
`present_files` is the set of manifest basenames currently committed under apps/ (flat or
already in wave-<N>/ subdirs), e.g. {"pihole.yaml", "nextcloud-db.yaml",
"nextcloud-db-app.sops.yaml", ...}. A catalog service is ONBOARDED iff `<svc>.yaml` is
present (the catalog also holds not-yet-onboarded SDs like keycloak/vm-demo, which must
NOT be placed). Each onboarded unit's sidecar files inherit the correct wave by naming
convention:
- `<svc>.yaml` -> wave(svc)
- `<svc>-bootstrap-secret.sops.yaml` -> wave(svc) (consumed by svc's env, same apply)
- `<svc>-oauth2-proxy-credentials.sops.yaml` -> wave(svc) (forward-auth sidecar OIDC creds)
- `<svc>-db.yaml` -> wave(svc-db) (a strictly-lower wave; CNPG DB)
- `<svc>-db-app.sops.yaml` -> wave(svc-db) (CNPG initdb owner secret)
- `<svc>-db-backup-credentials.sops.yaml` -> wave(svc-db) (CNPG Barman S3 backup secret)
- `<svc>-db-backup-credentials-prod.sops.yaml` -> wave(svc-db) (the PROD half of that
secret: the two environments carry DIFFERENT object-store credentials, selected in
the Cluster CR by `${BACKUP_CREDENTIALS_SUFFIX}`. Both objects ship to both clusters,
so both must be placed in the DB wave — the Secret has to exist before the Cluster
that names it.)
Non-catalog units (EXTRA_UNITS, e.g. gitea-actions) place by their declared anchor+offset.
Only files that actually exist in `present_files` are placed, so the plan is robust to a
service having no bootstrap secret or no DB. Deterministic (sorted).
"""
waves = compute_waves(sds)
by_name = _catalog_by_name(sds)
plan: dict = {}
def add(w: int, f: str) -> None:
if f in present_files:
plan.setdefault(w, set()).add(f)
onboarded = {n for n in by_name if f"{n}.yaml" in present_files}
for name in sorted(onboarded):
w = waves[name]
add(w, f"{name}.yaml")
add(w, f"{name}-bootstrap-secret.sops.yaml")
# forward-auth (Option B): the per-app oauth2-proxy sidecar's OIDC client + cookie secret.
# Same wave as the app (the sidecar reads it at pod start); placed only when the file exists,
# so non-forward-auth apps are unaffected. Mirrors the bootstrap-secret naming convention.
add(w, f"{name}-oauth2-proxy-credentials.sops.yaml")
db = f"{name}{CO_DB_SUFFIX}"
if any(dep["to"] == db for dep in by_name[name]["dependencies"]):
wdb = waves[db]
add(wdb, f"{db}.yaml")
add(wdb, f"{db}-app.sops.yaml")
add(wdb, f"{db}-backup-credentials.sops.yaml") # CNPG Barman S3 creds (R-STATE-03)
# PROD half of the same credential (2026-09-07 per-env split). Placed only when
# the file exists, so a DB that has not been split is unaffected.
add(wdb, f"{db}-backup-credentials-prod.sops.yaml")
for unit, cfg in EXTRA_UNITS.items():
anchor = cfg["anchor"]
if anchor not in onboarded:
continue
w = waves[anchor] + (1 if cfg.get("above") else 0)
for f in cfg["files"]:
add(w, f)
return {w: sorted(fs) for w, fs in plan.items()}
def wave_kustomization_text(wave: int, files: list) -> str:
"""The kustomize Kustomization for one wave directory (apps/wave-<N>/kustomization.yaml)."""
dep = "platform-config" if wave == 0 else f"apps-wave-{wave - 1}"
header = (
f"# Wave {wave} of the app tier — GENERATED by `catalog/render.py --emit-waves`.\n"
f"# Reconciled by the Flux Kustomization `apps-wave-{wave}` "
f"(clusters/base/apps-wave-{wave}.yaml), which dependsOn `{dep}` and health-gates\n"
f"# (wait:true) ONLY this wave — a crashloop here cannot stall an earlier wave.\n"
f"# DO NOT hand-edit: re-run `render.py --emit-waves` after onboarding a service."
)
body = {
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": list(files),
}
dumped = yaml.dump(body, sort_keys=False, default_flow_style=False, width=4096)
return f"---\n{header}\n{dumped}"
def flux_wave_kustomization_text(wave: int) -> str:
"""The Flux Kustomization CR for one wave (clusters/base/apps-wave-<N>.yaml).
dependsOn wave N-1 (wave 0 dependsOn platform-config); prune+wait match the retired
monolith. SOPS decryption is added by the environment overlay patch, IDENTICALLY to the
former single `apps` Kustomization (clusters/dev patches each apps-wave-<N> by name) — the
base stays env-agnostic and references the out-of-band `sops-age` Secret by name only.
"""
dep = "platform-config" if wave == 0 else f"apps-wave-{wave - 1}"
header = (
f"# Flux Kustomization: app-tier WAVE {wave}. Reconciles ./apps/wave-{wave} AFTER "
f"`{dep}` is healthy.\n"
f"# GENERATED by `catalog/render.py --emit-waves`. Waves isolate a crashloop to its "
f"own wave + downstream:\n"
f"# wait:true health-gates ONLY this wave's resources, so a later wave can no longer "
f"stall an earlier one.\n"
f"# SOPS decryption is layered on per-environment (clusters/dev patches this "
f"Kustomization by name), identical to the retired `apps` Kustomization."
)
cr = {
"apiVersion": "kustomize.toolkit.fluxcd.io/v1",
"kind": "Kustomization",
"metadata": {"name": f"apps-wave-{wave}", "namespace": "flux-system"},
"spec": {
"interval": "10m",
"retryInterval": "2m",
"timeout": "5m",
"prune": True,
"wait": True,
"dependsOn": [{"name": dep}],
"sourceRef": {"kind": "GitRepository", "name": "flux-system"},
"path": f"./apps/wave-{wave}",
"postBuild": {
"substituteFrom": [{"kind": "ConfigMap", "name": "cluster-vars"}]
},
},
}
dumped = yaml.dump(cr, sort_keys=False, default_flow_style=False, width=4096)
return f"---\n{header}\n{dumped}"
def _discover_app_files(apps_dir: Path) -> set:
"""Basenames of every committed app-tier manifest under apps/ (flat OR in wave-<N>/).
Excludes the generated per-wave kustomization.yaml and any README. Deterministic.
"""
present = set()
for p in apps_dir.rglob("*.yaml"):
if p.name == "kustomization.yaml":
continue
present.add(p.name)
return present
def emit_waves(catalog_dir: Path, apps_dir: Path, clusters_dir: Path) -> dict:
"""Write apps/wave-<N>/kustomization.yaml + clusters/base/apps-wave-<N>.yaml.
Computes the wave plan from the catalog DAG and the app files present under apps/, then
(re)generates the per-wave kustomize files and the Flux Kustomization CRs. Idempotent:
it never moves or rewrites a service manifest (only its placement changes, done in git),
so the rendered bytes — and the L1 goldens — are untouched. Returns the wave plan.
"""
sds = _load_catalog(catalog_dir)
present = _discover_app_files(apps_dir)
plan = plan_apps_waves(sds, present)
for wave in sorted(plan):
wdir = apps_dir / f"wave-{wave}"
wdir.mkdir(parents=True, exist_ok=True)
(wdir / "kustomization.yaml").write_text(
wave_kustomization_text(wave, plan[wave])
)
(clusters_dir / f"apps-wave-{wave}.yaml").write_text(
flux_wave_kustomization_text(wave)
)
return plan
# =============================================================================
# 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). Uses an explicit
# non-forward-auth mode so this exercises the direct app-port gateway path (a
# forward-auth app instead routes the Gateway to the :4180 sidecar — see 6c).
exp_docs = render(
{
"name": "z",
"source": {"image": "a:1"},
"ingressRoutes": [{"host": "z.home.keithjackson.dev", "port": 80}],
"auth": {"mode": "none"},
}
)
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)",
)
# 4b-dns) exposure -> ExternalDNS SCOPING on the HTTPRoute (R-DNS-03). `internal` renders no
# public DNS record and must be INCAPABLE of rendering one: its route carries
# external-dns's own controller annotation set to something other than `dns-controller`,
# which v0.21.0's gateway-httproute source (source/gateway.go ->
# annotations.IsControllerMismatch) uses to skip the route before any hostname is
# resolved. `public` must NOT carry it, or the five public names would vanish instead.
# Both directions are asserted here because only asserting the internal half would pass
# just as happily if the renderer stamped EVERY route and silently unpublished the zone.
dns_int_docs = render(
{
"name": "di",
"source": {"image": "a:1"},
"exposure": "internal",
"ingressRoutes": [{"host": "di.home.keithjackson.dev", "port": 80}],
"auth": {"mode": "none"},
}
)
dns_int_routes = [d[1] for d in dns_int_docs if d[1]["kind"] == "HTTPRoute"]
check(
len(dns_int_routes) == 1,
"an internal exposed service must still render its HTTPRoute (LAN reachability is "
"unchanged — only the PUBLIC record is withdrawn)",
)
if dns_int_routes:
iann = dns_int_routes[0]["metadata"].get("annotations", {})
check(
iann.get(EXTERNALDNS_CONTROLLER_KEY) == EXTERNALDNS_CONTROLLER_DISOWN,
"an internal service's HTTPRoute must carry the ExternalDNS disown annotation",
)
check(
iann.get(EXTERNALDNS_CONTROLLER_KEY) != EXTERNALDNS_CONTROLLER_CLAIMED,
"the disown value must never be external-dns's own `dns-controller` (that CLAIMS "
"the route rather than releasing it — the one value that inverts the guarantee)",
)
dns_pub_docs = render(
{
"name": "dp",
"source": {"image": "a:1"},
"exposure": "public",
"ingressRoutes": [{"host": "dp.home.keithjackson.dev", "port": 80}],
"auth": {"mode": "native"},
}
)
dns_pub_routes = [d[1] for d in dns_pub_docs if d[1]["kind"] == "HTTPRoute"]
check(len(dns_pub_routes) == 1, "a public service must render its HTTPRoute")
if dns_pub_routes:
check(
EXTERNALDNS_CONTROLLER_KEY
not in dns_pub_routes[0]["metadata"].get("annotations", {}),
"a public service's HTTPRoute must NOT carry the ExternalDNS disown annotation",
)
# The DEFAULT posture is `internal` (renderer-contract "Default posture"), so a
# ServiceDefinition that simply omits `exposure` must also be unpublishable. Omitting the
# field is the likeliest way a new service arrives, and it is the case a per-service
# allow-list would miss.
dns_dflt_docs = render(
{
"name": "dd",
"source": {"image": "a:1"},
"ingressRoutes": [{"host": "dd.home.keithjackson.dev", "port": 80}],
"auth": {"mode": "none"},
}
)
dns_dflt_route = next(d[1] for d in dns_dflt_docs if d[1]["kind"] == "HTTPRoute")
check(
dns_dflt_route["metadata"]["annotations"].get(EXTERNALDNS_CONTROLLER_KEY)
== EXTERNALDNS_CONTROLLER_DISOWN,
"a ServiceDefinition that OMITS `exposure` defaults to internal and must also be "
"unpublishable (default posture, renderer-contract)",
)
# The scoping annotation must COEXIST with the auth hints rather than replace them: both are
# merged into one `annotations` map, and a forward-auth internal app is the overlap case.
dns_fa_docs = render(
{
"name": "df",
"source": {"image": "a:1"},
"exposure": "internal",
"ingressRoutes": [{"host": "df.home.keithjackson.dev", "port": 80}],
"auth": {"mode": "forward-auth", "groups": ["platform-admins"]},
}
)
dns_fa_ann = next(
d[1] for d in dns_fa_docs if d[1]["kind"] == "HTTPRoute"
)["metadata"]["annotations"]
check(
dns_fa_ann.get(EXTERNALDNS_CONTROLLER_KEY) == EXTERNALDNS_CONTROLLER_DISOWN
and dns_fa_ann.get(f"{LBL}/auth-mode") == "forward-auth",
"DNS scoping and auth-mode hints must COEXIST on one internal forward-auth route",
)
# 4c) inClusterClients -> one ADDITIVE, least-privilege ingress allow per entry that admits a
# PLATFORM controller's NAMESPACE (not the reserved `ingress` identity, not `all`) to ONE
# declared port. This is what keeps a cluster whose GitOps source is an in-cluster Service
# reconcilable: fromEntities[ingress] does not cover a controller dialling the ClusterIP.
icc_docs = render(
{
"name": "g",
"source": {"image": "a:1"},
"ingressRoutes": [{"host": "g.home.keithjackson.dev", "port": 3000}],
"auth": {"mode": "none"},
"inClusterClients": [
{"name": "flux", "namespace": "flux-system", "port": 3000}
],
}
)
icc = [
d[1]
for d in icc_docs
if d[1]["kind"] == "CiliumNetworkPolicy"
and d[1]["metadata"]["name"] == "g-allow-flux-ingress"
]
check(
len(icc) == 1,
"an inClusterClients entry must render one ingress CNP named "
"<service>-allow-<name>-ingress",
)
if icc:
p = icc[0]
check(
p["metadata"]["namespace"] == "g"
and p["spec"]["endpointSelector"]["matchLabels"]
== {"app.kubernetes.io/name": "g"},
"in-cluster client allow must live in the service's namespace and select the service",
)
ing = p["spec"]["ingress"]
check(
ing[0]["fromEndpoints"]
== [{"matchLabels": {"k8s:io.kubernetes.pod.namespace": "flux-system"}}],
"in-cluster client allow must match the client NAMESPACE only (reserved k8s: label)",
)
check(
"fromEntities" not in ing[0],
"in-cluster client allow must not widen to an entity (never fromEntities: all)",
)
icc_ports = {
(pt["port"], pt["protocol"])
for r in ing
for tp in r["toPorts"]
for pt in tp["ports"]
}
check(
icc_ports == {("3000", "TCP")},
"in-cluster client allow must be least-privilege (only the declared port/protocol)",
)
# ADDITIVE: default-deny ingress stays [] and the gateway allow is untouched.
icc_dd = next(d[1] for d in icc_docs if d[1]["kind"] == "NetworkPolicy")
check(
icc_dd["spec"]["ingress"] == [],
"in-cluster client allow must be ADDITIVE — default-deny ingress stays [] (R-NET-01)",
)
check(
any(
d[1]["metadata"].get("name") == "g-allow-gateway-ingress"
for d in icc_docs
if d[1]["kind"] == "CiliumNetworkPolicy"
),
"in-cluster client allow must not displace the north-south gateway allow",
)
# `name` is OPTIONAL and defaults to the namespace, so the common case is namespace + port.
icc_default = render(
{
"name": "g",
"source": {"image": "a:1"},
"inClusterClients": [{"namespace": "monitoring", "port": 9090}],
}
)
check(
any(
d[1]["metadata"].get("name") == "g-allow-monitoring-ingress"
for d in icc_default
if d[1]["kind"] == "CiliumNetworkPolicy"
),
"inClusterClients.name must default to the admitted namespace",
)
# A service that declares none must render NO such policy (grant-nothing-extra).
check(
not any(
d[1]["kind"] == "CiliumNetworkPolicy"
and d[1]["metadata"]["name"].endswith("-ingress")
and "fromEndpoints" in d[1]["spec"].get("ingress", [{}])[0]
for d in render({"name": "q", "source": {"image": "a:1"}})
),
"a service declaring no inClusterClients must render no in-cluster client allow",
)
# 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"]}}
},
"podSecurityContext": {"fsGroup": 1001, "runAsUser": 1001},
}
)
envd_pod = next(
d[1] for d in envd_docs if d[1]["kind"] == "Deployment"
)["spec"]["template"]["spec"]
check(
envd_pod.get("securityContext") == {"fsGroup": 1001, "runAsUser": 1001},
"podSecurityContext must render verbatim onto the pod spec",
)
ec = envd_pod["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",
)
plain_pod = next(
d[1] for d in render({"name": "plain2", "source": {"image": "a:1"}})
if d[1]["kind"] == "Deployment"
)["spec"]["template"]["spec"]
check(
"securityContext" not in plain_pod,
"an SD without podSecurityContext must not gain a pod securityContext",
)
# 6c) forward-auth (Option B): an exposed forward-auth app gets an oauth2-proxy sidecar that
# fronts the Gateway path (fail-closed), while oidc/native/none apps are unchanged.
fa_docs = render(
{
"name": "fa",
"source": {"image": "a:1"},
"ingressRoutes": [{"host": "fa.home.keithjackson.dev", "port": 9117}],
"auth": {"mode": "forward-auth", "groups": ["media-users"]},
}
)
fa_dep = next(d[1] for d in fa_docs if d[1]["kind"] == "Deployment")
fa_containers = fa_dep["spec"]["template"]["spec"]["containers"]
check(
len(fa_containers) == 2
and fa_containers[0]["name"] == "fa"
and fa_containers[1]["name"] == "oauth2-proxy",
"forward-auth app must render the app container FIRST + an oauth2-proxy sidecar SECOND",
)
side = fa_containers[1]
check(
"--upstream=http://127.0.0.1:9117" in side["args"],
"sidecar upstream must be the co-located app UI over http on 127.0.0.1:<uiport>",
)
check(
"--allowed-group=media-users" in side["args"],
"sidecar must enforce the declared auth.group (R-IDP-06)",
)
check(
"--redirect-url=https://fa.home.keithjackson.dev/oauth2/callback" in side["args"],
"sidecar redirect-url must be the app host's /oauth2/callback",
)
check(
not any(a.startswith("--ssl-upstream-insecure-skip-verify") for a in side["args"]),
"http upstream must NOT skip TLS verify (that flag is only for crafty's self-signed https)",
)
check(
side["env"][0]["valueFrom"]["secretKeyRef"]["name"] == "fa-oauth2-proxy",
"sidecar creds must come by name from the per-app <app>-oauth2-proxy Secret (R-IDP-04)",
)
fa_svc = next(d[1] for d in fa_docs if d[1]["kind"] == "Service")
fa_svc_ports = {p["port"] for p in fa_svc["spec"]["ports"]}
check(
fa_svc_ports == {9117, 4180},
"forward-auth Service must expose the app port (east-west) AND :4180 (gateway->sidecar)",
)
fa_route = next(d[1] for d in fa_docs if d[1]["kind"] == "HTTPRoute")
check(
fa_route["spec"]["rules"][0]["backendRefs"][0]["port"] == 4180,
"forward-auth HTTPRoute backendRef must target the sidecar :4180, not the app UI port",
)
fa_gw = next(
d[1] for d in fa_docs
if d[1]["kind"] == "CiliumNetworkPolicy"
and d[1]["metadata"]["name"] == "fa-allow-gateway-ingress"
)
fa_gw_ports = {
p["port"] for r in fa_gw["spec"]["ingress"] for tp in r["toPorts"] for p in tp["ports"]
}
check(
fa_gw_ports == {"4180"},
"fail-closed: the Gateway may reach ONLY the sidecar :4180, never the un-gated app port",
)
# an oidc app (native client) must NOT gain a sidecar or a :4180 path.
oidc_docs = render(
{
"name": "oi",
"source": {"image": "a:1"},
"ingressRoutes": [{"host": "oi.home.keithjackson.dev", "port": 3000}],
"auth": {"mode": "oidc", "groups": ["platform-admins"]},
}
)
oi_containers = next(
d[1] for d in oidc_docs if d[1]["kind"] == "Deployment"
)["spec"]["template"]["spec"]["containers"]
check(
len(oi_containers) == 1
and not any("oauth2-proxy" in c["name"] for c in oi_containers),
"an oidc app must NOT gain an oauth2-proxy sidecar (native client authenticates itself)",
)
oi_route = next(d[1] for d in oidc_docs if d[1]["kind"] == "HTTPRoute")
check(
oi_route["spec"]["rules"][0]["backendRefs"][0]["port"] == 3000,
"an oidc app's HTTPRoute must still target its own app port (unchanged)",
)
# 7) wave partitioning: longest-path over the dependency DAG, CNPG-DB strictly below.
wsds = [
{"name": "a", "source": {"image": "x:1"}}, # leaf
{"name": "b", "source": {"image": "x:1"}}, # leaf
{"name": "c", "source": {"image": "x:1"},
"dependencies": [{"to": "a", "port": 1}, {"to": "b", "port": 1}]}, # wave 1
{"name": "d", "source": {"image": "x:1"},
"dependencies": [{"to": "c", "port": 1}]}, # wave 2
{"name": "svc", "source": {"image": "x:1"},
"dependencies": [{"to": "svc-db", "port": 5432}]}, # DB consumer
]
w = compute_waves(wsds)
check(w["a"] == 0 and w["b"] == 0, "leaf services must be wave 0")
check(w["c"] == 1, "a service depending on wave-0 nodes must be wave 1")
check(w["d"] == 2, "longest-path depth must be transitive (d -> c -> {a,b})")
check(
w["svc-db"] == 0 and w["svc"] == 1 and w["svc-db"] < w["svc"],
"a co-located <svc>-db must be a leaf STRICTLY BELOW its consumer (CNPG-DB rule)",
)
# cycle detection.
try:
compute_waves([
{"name": "p", "source": {"image": "x:1"}, "dependencies": [{"to": "q", "port": 1}]},
{"name": "q", "source": {"image": "x:1"}, "dependencies": [{"to": "p", "port": 1}]},
])
check(False, "compute_waves must raise on a dependency cycle")
except ValueError:
pass
# 7a) plan_apps_waves: sidecars inherit their unit's wave; not-onboarded SDs are skipped;
# EXTRA_UNITS anchor strictly above; a file absent from the tree is never placed.
present = {
"a.yaml", "svc.yaml", "svc-bootstrap-secret.sops.yaml",
"svc-db.yaml", "svc-db-app.sops.yaml", "svc-db-backup-credentials.sops.yaml",
"svc-db-backup-credentials-prod.sops.yaml",
"gitea.yaml",
"gitea-actions.yaml", "gitea-actions-runner-secret.sops.yaml",
}
plan_sds = wsds + [{"name": "gitea", "source": {"image": "x:1"}},
{"name": "notonboarded", "source": {"image": "x:1"}}]
plan = plan_apps_waves(plan_sds, present)
flat = {f: wv for wv, fs in plan.items() for f in fs}
check(flat.get("a.yaml") == 0, "leaf app must land in wave 0")
check(flat.get("svc-db.yaml") == 0 and flat.get("svc-db-app.sops.yaml") == 0
and flat.get("svc-db-backup-credentials.sops.yaml") == 0
and flat.get("svc-db-backup-credentials-prod.sops.yaml") == 0,
"CNPG DB manifest + its owner secret + BOTH halves of its per-env Barman backup secret "
"must be in the DB's (lower) wave")
check(flat.get("svc.yaml") == 1 and flat.get("svc-bootstrap-secret.sops.yaml") == 1,
"a DB-backed app + its bootstrap secret must be in the consumer wave, above the DB")
check(flat.get("gitea.yaml") == 0, "gitea (leaf) must be wave 0")
check(flat.get("gitea-actions.yaml") == 1
and flat.get("gitea-actions-runner-secret.sops.yaml") == 1,
"gitea-actions (EXTRA_UNITS, above=True) must be one wave above its anchor gitea")
check("notonboarded.yaml" not in flat,
"a catalog SD with no committed manifest must NOT be placed (not onboarded)")
check("b.yaml" not in flat, "a file absent from the tree must never be placed")
# 7b) generated Flux Kustomization CR wiring (dependsOn chain + prune/wait/path).
import yaml as _yaml # already imported at module top; local alias for clarity
fw0 = _yaml.safe_load(flux_wave_kustomization_text(0))
fw2 = _yaml.safe_load(flux_wave_kustomization_text(2))
check(fw0["metadata"]["name"] == "apps-wave-0"
and fw0["spec"]["dependsOn"] == [{"name": "platform-config"}]
and fw0["spec"]["path"] == "./apps/wave-0",
"wave 0 Flux Kustomization must dependsOn platform-config and point at ./apps/wave-0")
check(fw2["spec"]["dependsOn"] == [{"name": "apps-wave-1"}]
and fw2["spec"]["prune"] is True and fw2["spec"]["wait"] is True,
"wave N Flux Kustomization must dependsOn wave N-1 with prune+wait (health-gated)")
kt = _yaml.safe_load(wave_kustomization_text(1, ["svc.yaml", "sonarr.yaml"]))
check(kt["kind"] == "Kustomization" and kt["resources"] == ["svc.yaml", "sonarr.yaml"],
"wave kustomization must list exactly the wave's resources")
# 8) substitution safety (R-ENV-02): the renderer must REFUSE to emit a manifest whose
# shell script carries an unescaped `${...}`. Every rendered manifest lands under a
# Flux path with postBuild.substituteFrom cluster-vars, so envsubst rewrites it before
# the container runs — this is the gitea/nextcloud OIDC-guard bug (`[ -z "${X:-}" ]`
# -> `[ -z "" ]`, always true, whole registration block skipped on every boot).
known = {"BASE_DOMAIN", "GATEWAY_VIP"}
# allowed: a bare, DECLARED cluster-vars knob is a deliberate R-ENV-02 substitution.
check(
substitution_violations("host: app.${BASE_DOMAIN}", known) == [],
"a bare declared cluster-vars key must be allowed (that is what substitution is FOR)",
)
# allowed: escaped shell braces, and the forms envsubst leaves alone.
check(
substitution_violations('if [ -z "$${X:-}" ]; then', known) == [],
"an ESCAPED `$${...}` must be allowed (envsubst emits it back as a literal `${...}`)",
)
check(
substitution_violations('t="$(mktemp)"; echo "$HOME $1"', known) == [],
"bare `$VAR` / `$(cmd)` are left alone by envsubst and must NOT be flagged",
)
# `$$` is a literal `$` to envsubst, so an ODD run of dollars still leaves the brace live.
check(
len(substitution_violations("$$${X:-}", known)) == 1,
"an ODD run of `$` still substitutes ($$ -> literal $, then `${X:-}` is eaten)",
)
# flagged: the exact regression — a default-operator form STRICT mode deliberately exempts.
guard_bug = substitution_violations('if [ -z "${GITEA_OIDC_CLIENT_SECRET:-}" ]; then', known)
check(
len(guard_bug) == 1 and guard_bug[0][1] == "${GITEA_OIDC_CLIENT_SECRET:-}",
"an unescaped `${VAR:-}` shell guard must be flagged (strict mode exempts this form)",
)
# flagged: an undeclared bare knob (envsubst blanks it), and an indexed form (hard error).
check(
len(substitution_violations("value: ${NOT_A_KNOB}", known)) == 1,
"an undeclared bare `${VAR}` must be flagged (envsubst blanks it silently)",
)
idx = substitution_violations("echo ${PIPESTATUS[0]}", known)
check(
len(idx) == 1 and "missing closing brace" in idx[0][2],
"an indexed `${A[0]}` must be flagged (envsubst hard-fails the whole Kustomization)",
)
# the checker must be wired into emit(): rendering an SD whose lifecycle hook carries the
# bug RAISES rather than writing a manifest whose guard is dead on arrival.
bad_sd = {
"name": "subbad",
"source": {"image": "a:1"},
"lifecycle": {
"postStart": {
"exec": {
"command": [
"/bin/sh",
"-c",
'if [ -z "${SUBBAD_SECRET:-}" ]; then exit 0; fi',
]
}
}
},
}
try:
emit(render(bad_sd), origin="<selftest>")
check(False, "emit must REFUSE a manifest with an unescaped `${...}` shell expansion")
except ValueError as exc:
check(
"SUBBAD_SECRET" in str(exc) and "$${" in str(exc),
"the refusal must name the offending token AND state the `$${...}` fix",
)
# the ESCAPED form of the same hook renders normally (the fix must be emittable).
good_sd = dict(bad_sd, name="subgood")
good_sd["lifecycle"] = {
"postStart": {
"exec": {
"command": [
"/bin/sh",
"-c",
'if [ -z "$${SUBBAD_SECRET:-}" ]; then exit 0; fi',
]
}
}
}
check(
"$${SUBBAD_SECRET:-}" in emit(render(good_sd), origin="<selftest>"),
"the escaped `$${...}` form must render (and reach the shell as `${...}`)",
)
# 9) observability (R-OBS-01): `metrics: true` alone renders NO ServiceMonitor, and the `up`
# SLI reads a signal that exists. This is the regression test for the defect that made all
# twelve `<app>-down` rules structurally inert on prod (see r_servicemonitor).
obs_sd = {
"name": "obsdemo",
"source": {"image": "a:1"},
"ingressRoutes": [{"host": "obsdemo.${BASE_DOMAIN}", "port": 8080}],
"auth": {"mode": "none"},
"observability": {"metrics": True, "slis": ["up"]},
}
obs_docs = render(obs_sd)
check(
not [d for d in obs_docs if d[1]["kind"] == "ServiceMonitor"],
"metrics: true with NO metricsEndpoint must render no ServiceMonitor — a monitor naming a "
"port the Service does not define discovers zero targets and makes the SLI inert",
)
pr = next(d[1] for d in obs_docs if d[1]["kind"] == "PrometheusRule")
up_rule = pr["spec"]["groups"][0]["rules"][0]
check(up_rule["alert"] == "obsdemo-down", "slis: [up] must render <name>-down")
check(
"up{job=" not in up_rule["expr"],
"the up SLI must NOT read up{job=...}: no repo-rendered scrape target produces that job, "
"so the rule can never fire (verified inert on prod for all twelve apps)",
)
check(
up_rule["expr"]
== 'kube_deployment_status_replicas_available{namespace="obsdemo", deployment="obsdemo"} '
'== 0 or absent(kube_deployment_status_replicas_available{namespace="obsdemo", '
'deployment="obsdemo"})',
"container up SLI must read kube_deployment_status_replicas_available with an absent() "
f"half; got: {up_rule['expr']}",
)
# 9a) livenessWorkload override: the object that RUNS is not always the one this SD renders.
sts_docs = render(
dict(
obs_sd,
name="obssts",
observability={
"metrics": True,
"slis": ["up"],
"livenessWorkload": {"kind": "StatefulSet", "name": "obssts"},
},
)
)
sts_expr = next(
d[1] for d in sts_docs if d[1]["kind"] == "PrometheusRule"
)["spec"]["groups"][0]["rules"][0]["expr"]
check(
"kube_statefulset_status_replicas_ready" in sts_expr,
f"livenessWorkload kind StatefulSet must read the statefulset series; got: {sts_expr}",
)
# 9b) vm kind: a VMI has no replica count, so liveness is a running-phase series EXISTING.
vm_pr = next(
d[1]
for d in render(
{
"name": "obsvm",
"kind": "vm",
"source": {"vmTemplate": "quay.io/x/y:1"},
"storage": [{"name": "root", "size": "6Gi"}],
"observability": {"metrics": True, "slis": ["up"]},
}
)
if d[1]["kind"] == "PrometheusRule"
)
check(
vm_pr["spec"]["groups"][0]["rules"][0]["expr"]
== 'absent(kubevirt_vmi_info{namespace="obsvm", name="obsvm", phase="running"})',
"vm up SLI must read kubevirt_vmi_info with the LOWERCASE running phase: "
f"{vm_pr['spec']['groups'][0]['rules'][0]['expr']}",
)
# 9c) metricsEndpoint renders ALL THREE halves of a working scrape, and they AGREE. Any one of
# them missing is an independent way for the scrape to silently discover/deliver nothing.
ep_docs = render(
dict(
obs_sd,
name="obsep",
observability={
"metrics": True,
"slis": ["up"],
"metricsEndpoint": {"port": 9100},
},
)
)
ep_svc = next(d[1] for d in ep_docs if d[1]["kind"] == "Service")
ep_sm = next(d[1] for d in ep_docs if d[1]["kind"] == "ServiceMonitor")
svc_port_names = {p["name"] for p in ep_svc["spec"]["ports"]}
ep_ports = {e["port"] for e in ep_sm["spec"]["endpoints"]}
check(
ep_ports <= svc_port_names,
f"ServiceMonitor endpoint port(s) {sorted(ep_ports)} must all exist on the Service "
f"{sorted(svc_port_names)} — this exact disagreement is what made twelve monitors inert",
)
check(
{"name": "metrics", "port": 9100, "targetPort": 9100} in ep_svc["spec"]["ports"],
"metricsEndpoint must render the named Service port it declares",
)
mon_cnp = [
d[1]
for d in ep_docs
if d[1]["kind"] == "CiliumNetworkPolicy"
and d[1]["metadata"]["name"] == "obsep-allow-monitoring-scrape"
]
check(
len(mon_cnp) == 1,
"a declared metricsEndpoint must render the monitoring-namespace ingress allow — the "
"namespace default-deny has `ingress: []` and would DROP the scrape without it",
)
if mon_cnp:
ing = mon_cnp[0]["spec"]["ingress"][0]
check(
ing["fromEndpoints"][0]["matchLabels"]["k8s:io.kubernetes.pod.namespace"]
== MONITORING_NS
and ing["toPorts"][0]["ports"] == [{"port": "9100", "protocol": "TCP"}],
"the scrape allow must admit ONLY the monitoring namespace to ONLY the metrics port",
)
gw_cnp = next(
d[1]
for d in ep_docs
if d[1]["kind"] == "CiliumNetworkPolicy"
and d[1]["metadata"]["name"] == "obsep-allow-gateway-ingress"
)
gw_ports = {p["port"] for p in gw_cnp["spec"]["ingress"][0]["toPorts"][0]["ports"]}
check(
"9100" not in gw_ports,
"the metrics port must NOT be opened to the Gateway (north-south); it is for the "
f"monitoring namespace only. Gateway allows: {sorted(gw_ports)}",
)
# A metricsEndpoint the renderer cannot express must be REFUSED, never rendered inert.
try:
render(
dict(
obs_sd,
name="obsbad",
observability={
"metrics": True,
"slis": ["up"],
"livenessWorkload": {"kind": "CronJob", "name": "obsbad"},
},
)
)
check(False, "an unknown livenessWorkload.kind must raise, not render a bogus expression")
except ValueError as exc:
check("CronJob" in str(exc), "the refusal must name the unsupported kind")
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)")
ap.add_argument("--wave-layout", action="store_true",
help="With -o: place each rendered file under <out>/wave-<N>/<name>.yaml, "
"N = its dependency-ordered wave (compute_waves over the catalog)")
ap.add_argument("--emit-waves", action="store_true",
help="Generate the per-wave kustomize + Flux Kustomization CRs from the "
"catalog DAG and the committed apps tree (no manifest is rewritten)")
ap.add_argument("--print-waves", action="store_true",
help="Print the computed wave table (node -> wave) for the catalog and exit")
ap.add_argument("--catalog", default="catalog",
help="Catalog dir for --emit-waves/--print-waves/--wave-layout (default: catalog)")
ap.add_argument("--apps-dir", default="apps",
help="App-tier dir for --emit-waves (default: apps)")
ap.add_argument("--clusters-dir", default="clusters/base",
help="Dir to write clusters/base/apps-wave-<N>.yaml (default: clusters/base)")
args = ap.parse_args(argv)
if args.selftest:
return _selftest()
if args.print_waves:
waves = compute_waves(_load_catalog(Path(args.catalog)))
for node in sorted(waves, key=lambda n: (waves[n], n)):
sys.stdout.write(f"wave {waves[node]} {node}\n")
return 0
if args.emit_waves:
plan = emit_waves(Path(args.catalog), Path(args.apps_dir), Path(args.clusters_dir))
for wave in sorted(plan):
sys.stderr.write(f"[emit-waves] wave {wave}: {', '.join(plan[wave])}\n")
sys.stderr.write(
f"[emit-waves] wrote {len(plan)} wave kustomization(s) under {args.apps_dir}/ "
f"and clusters CRs under {args.clusters_dir}/\n"
)
return 0
image_automation = not args.no_image_automation
def _out_path(outdir: Path, name: str, sds: list) -> Path:
if args.wave_layout:
wdir = outdir / f"wave-{service_wave(sds, name)}"
wdir.mkdir(parents=True, exist_ok=True)
return wdir / f"{name}.yaml"
outdir.mkdir(parents=True, exist_ok=True)
return outdir / f"{name}.yaml"
if args.all:
src = Path(args.all)
files = sorted(p for p in src.glob("*.yaml") if p.name != "render.py")
sds = _load_catalog(src) if args.wave_layout else []
for f in files:
text = render_file(f, image_automation=image_automation)
if args.out_dir:
_out_path(Path(args.out_dir), load_sd(f)["name"], sds).write_text(text)
else:
sys.stdout.write(text)
return 0
if not args.input:
ap.error("provide a ServiceDefinition file, --all DIR, --emit-waves, or --selftest")
inp = Path(args.input)
text = render_file(inp, image_automation=image_automation)
if args.out_dir:
sds = _load_catalog(Path(args.catalog)) if args.wave_layout else []
_out_path(Path(args.out_dir), load_sd(inp)["name"], sds).write_text(text)
else:
sys.stdout.write(text)
return 0
if __name__ == "__main__":
raise SystemExit(main())