Home-Network/home-network-k3s-impl
2
0 Fork 0
Code Issues 2 Pull Requests 5 Actions Packages Projects Releases Wiki Activity
Files
23ea72d01e8a565122f635d3f92986c9ac871857
home-network-k3s-impl/catalog/render.py
T
jakman85andClaude Opus 5 da8a0eac04 test(l0): R-STATE-04 gets a real check — bulk storage provably is not the replicated tier 
The gate BLOCKED this branch, correctly, and the block was the interesting part.

Attaching the media library selects R-STATE-04 as a touched requirement ("bulk media storage has
a capacity strategy distinct from the replicated tier"), and R-STATE-04 had NO scaffolded test, so
the coverage REQUIRE stage failed both l0 and l1. There were two ways out. One was to keep
avoiding the token — the existing discipline in catalog/plex.yaml and catalog/qbittorrent.yaml,
which say in as many words that naming the requirement would block the merge. That made sense while
the tier was a stub. It does not make sense in the change that DELIVERS the tier: it would mean
shipping the requirement and claiming nothing. So: the check.

WHAT IT ASSERTS, and why each one is a real failure mode rather than a shape
The `bulk` StorageClass sat as `server: REPLACE-ME` for six roadmap phases, and the only thing
keeping a multi-terabyte library off the 3-replica `longhorn` class was four hand-written comments.
That is the class of failure being closed: a bulk volume on the replicated tier does not fail
loudly — Longhorn accepts the PVC and starts replicating 3x of data three mini-PCs cannot hold.

  (1) DISTINCTNESS — no bulk PV/PVC on any class the repo declares with the Longhorn provisioner
      (derived from platform/longhorn/, so adding a Longhorn class cannot create a blind spot),
      and the class is explicitly EMPTY.
  (2) CAPACITY STRATEGY as arithmetic — a bulk volume at least an order of magnitude above the
      same service's replicated volume. The requirement is about two magnitudes of data; a "bulk"
      volume sized like a config PVC means the tier was named, not applied. The 10x floor is
      CALIBRATED against the real corpus, not guessed: the media library is ~460x its config
      volume and the Nextcloud archive ~13x its 20Gi, so a higher floor would fire on a legitimate
      volume and teach the next reader to weaken the check. My first draft used 100x and did
      exactly that — the corpus caught it.
  (3) PRE-BOUND FROM BOTH ENDS (claimRef + volumeName, agreeing). Guards the most deceptive
      failure available in this tier: a dynamic bulk class hands out a FRESH EMPTY directory, the
      pod mounts, the PVC Binds, everything is green, and the 8 TiB was never attached.
  (4) Retain, always. (5) mountOptions carry `hard` and exactly one nfsvers=4.x, and never `soft`
      — a soft mount hands a server hiccup to the app as a short read, which for a media library
      is silent corruption; and v4 is what keeps the firewall surface at one port.
  (6) ONE EXPORT, AT MOST ONE MOUNT PER WORKLOAD. This is the assertion I most wanted. Splitting
      one export into two subPath mounts looks TIGHTER and breaks link() with EXDEV; Sonarr and
      Radarr then degrade to copy-then-delete, rewriting every import over NFS and destroying the
      seeding copy, surfacing weeks later as "imports are slow and my ratio is gone". No reviewer
      sees that in a diff.
  (7) READ-ONLY COUPLING — nfs.readOnly, the volumeMount flag and ReadOnlyMany must agree, so
      dropping one still leaves the others; and every bulk mode is a `...Many`.
  (8) The server is a BARE cluster-vars token, that knob has a base default, and EVERY overlay
      supplies it — the seam that lets one authored definition serve two clusters that reach this
      host by different addresses.
  (9) LOAD-BEARING and NON-VACUOUS. Wired unconditionally into ci/l0.sh as Stage 3e for the same
      reason as 3c/3d: Stage 3b scopes to touched requirements, and the path map keys NEITHER
      apps/* NOR platform/longhorn/* to this requirement — so the change most likely to break
      this could attribute nothing and be gated by nothing. An EMPTY bulk corpus FAILS rather
      than passing trivially, and a floor on the assertion count keeps a pass from degenerating.

It reads the COMMITTED apps/ goldens rather than calling the renderer. Asking the renderer would
only prove it agrees with itself; the goldens are what Flux reconciles, and render==golden is
already its own gate stage.

PERTURBED, because a check that has never failed is not evidence. Each of these was applied to the
real tree and each fired with the intended message: bulk PVC moved onto `longhorn`; Retain -> Delete;
`hard` -> `soft`; the token replaced by the literal 192.168.1.9; sonarr's single mount split into
two subPath mounts (assertion 6); and the prod overlay's knob deleted.

Also hardens render.py: `size` is structurally required of a bulk entry (a PV with no capacity is
rejected by the API server) but the JSON Schema only enforces `export`, so omitting it raised a bare
KeyError three frames down. It now raises a sentence naming the service, the volume and the remedy.
Found by the same perturbation pass.

Coverage: 48/51 -> 49/51 (96.1%), 2 gaps, 0 safety-critical.
BASE_REF=main make -C ci gate -> MERGE GATE: PASS.

Co-Authored-By: Claude Opus 5 <[email protected]>
2026-09-08 00:03:34 -05:00

2217 lines
100 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 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"
KP_RELEASE = "kube-prometheus-stack" # ServiceMonitor/PrometheusRule discovery label
# egress classes and auth modes fixed by docs/03 / docs/13 + the Kyverno enums.
EGRESS_CLASSES = ("internet", "vpn")
AUTH_MODES = ("oidc", "forward-auth", "native", "none")
# --- 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", [])
# auth.mode default posture: forward-auth for any exposed UI, else none (internal, no
# UI). An explicit auth block always wins (R-IDP-02).
auth = dict(s.get("auth") or {})
if "mode" not in auth:
auth["mode"] = "forward-auth" if s["ingressRoutes"] else "none"
auth.setdefault("groups", [])
s["auth"] = auth
update = dict(s.get("update") or {})
update.setdefault("channel", "stable")
update.setdefault("autoRollbackOnFailedHealth", True)
s["update"] = update
obs = dict(s.get("observability") or {})
obs.setdefault("metrics", True)
obs.setdefault("slis", [])
s["observability"] = obs
# storage entry class default is longhorn (replicated tier); 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"]]
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)
if not ports:
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}
]
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 _route_meta(s: dict, route: dict) -> dict:
meta = {
"name": f"{s['name']}-{route['port']}",
"namespace": s["name"],
"labels": {"app.kubernetes.io/name": s["name"]},
}
ann = _auth_annotations(s)
if ann:
meta["annotations"] = ann
return meta
def r_internet_egress(s: dict) -> dict:
"""North-south egress allow: let the app reach the INTERNET (Cilium reserved `world`
entity = everything outside the cluster), which is the whole point of the `egress` class
(docs/03: internet-class → WAN; vpn-class → steered through the node tun0 PIA tunnel with
the nftables kill-switch, ADR-0015). The default-deny baseline (R-NET-01) allows only DNS
so EAST-WEST stays denied; this ADDITIVE policy opens ONLY `world` (non-cluster) egress —
cluster/east-west traffic stays denied except the declared `dependencies`. Every app
declares an egress class (EGRESS_CLASSES), so every app gets this. The steering + kill-
switch (a cluster-wide singleton keyed on the egress-class pod label) decide the PATH; this
policy decides that north-south egress is ALLOWED at all.
"""
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 r_servicemonitor(s: dict):
if not s["observability"]["metrics"]:
return None
return {
"apiVersion": "monitoring.coreos.com/v1",
"kind": "ServiceMonitor",
"metadata": {
"name": s["name"],
"namespace": s["name"],
"labels": {"release": KP_RELEASE, "app.kubernetes.io/name": s["name"]},
},
"spec": {
"selector": {"matchLabels": {"app.kubernetes.io/name": s["name"]}},
"namespaceSelector": {"matchNames": [s["name"]]},
"endpoints": [{"port": "metrics", "interval": "30s"}],
},
}
def r_prometheusrule(s: dict):
"""Each SLI becomes at least one alert rule (R-OBS-01)."""
slis = s["observability"]["slis"]
if not slis:
return None
rules = []
for sli in slis:
if sli == "up":
rules.append(
{
"alert": f"{s['name']}-down".replace("_", "-"),
"expr": f'up{{job="{s["name"]}"}} == 0',
"for": "5m",
"labels": {"severity": "critical", "service": s["name"]},
"annotations": {
"summary": f"{s['name']} is down",
"description": f"{s['name']} has been unreachable for 5m (SLI: up).",
},
}
)
else:
rules.append(
{
"alert": f"{s['name']}-{sli}".replace("_", "-"),
"expr": f'absent({sli}{{service="{s["name"]}"}})',
"for": "10m",
"labels": {"severity": "warning", "service": s["name"]},
"annotations": {"summary": f"{s['name']} SLI {sli} unhealthy"},
}
)
return {
"apiVersion": "monitoring.coreos.com/v1",
"kind": "PrometheusRule",
"metadata": {
"name": f"{s['name']}-slis",
"namespace": s["name"],
"labels": {"release": KP_RELEASE, "app.kubernetes.io/name": s["name"]},
},
"spec": {"groups": [{"name": f"{s['name']}.slis", "rules": rules}]},
}
def r_image_automation(s: dict) -> list:
"""container kind -> Flux ImageRepository + ImagePolicy encoding channel/range (R-UPD).
Flux image-automation advances the running version on its scan interval; no human/AI
edits a tag (R-UPD-02). `pin` freezes automation. vm kind uses in-guest updates instead.
"""
if s["kind"] != "container":
return []
image = s["source"]["image"].split(":", 1)[0]
upd = s["update"]
interval = "1h0m0s"
imagerepo = {
"apiVersion": "image.toolkit.fluxcd.io/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,
)
)
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",
],
},
}
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)",
)
# 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 `${...}`)",
)
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())
Reference in new issue View Git Blame Copy Permalink