#!/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 ` (or `--all`); the offline L1 render conformance verifier diffs its deterministic output against committed goldens. Determinism: fixed document order + fixed key order + PyYAML with sort_keys=False and a wide line width, so identical input always yields byte-identical output (golden-diffable). Dependency: PyYAML (input parse + output emit). This is the ONE renderer dependency; the conformance harness stays stdlib-only and only shells out to this script + diffs text. """ from __future__ import annotations import argparse import sys from pathlib import Path try: import yaml except ImportError: # pragma: no cover - explicit, actionable failure sys.stderr.write( "render.py requires PyYAML (the renderer's single dependency). " "Install with: pip install pyyaml\n" ) raise SystemExit(2) # --- Label / naming conventions (match platform/kyverno/policies + platform/*/) -------- LBL = "homelab.keithjackson.dev" ZONE = "home.keithjackson.dev" GATEWAY_NAME = "shared" GATEWAY_NS = "gateway" KP_RELEASE = "kube-prometheus-stack" # ServiceMonitor/PrometheusRule discovery label # egress classes and auth modes fixed by docs/03 / docs/13 + the Kyverno enums. EGRESS_CLASSES = ("internet", "vpn") AUTH_MODES = ("oidc", "forward-auth", "native", "none") # ============================================================================= # Defaults (renderer-contract.md "Default posture") — materialize BEFORE mapping. # ============================================================================= def apply_defaults(sd: dict) -> dict: """Return a copy of the ServiceDefinition with the contract's default posture filled. A field absent from a ServiceDefinition renders from its default posture, not "not at all" (renderer-contract.md). """ s = dict(sd) s.setdefault("kind", "container") s.setdefault("exposure", "internal") s.setdefault("egress", "internet") s.setdefault("storage", []) s.setdefault("dependencies", []) s.setdefault("ingressRoutes", []) # auth.mode default posture: forward-auth for any exposed UI, else none (internal, no # UI). An explicit auth block always wins (R-IDP-02). auth = dict(s.get("auth") or {}) if "mode" not in auth: auth["mode"] = "forward-auth" if s["ingressRoutes"] else "none" auth.setdefault("groups", []) s["auth"] = auth update = dict(s.get("update") or {}) update.setdefault("channel", "stable") update.setdefault("autoRollbackOnFailedHealth", True) s["update"] = update obs = dict(s.get("observability") or {}) obs.setdefault("metrics", True) obs.setdefault("slis", []) s["observability"] = obs # storage entry class default is longhorn (replicated tier). s["storage"] = [{**{"class": "longhorn"}, **v} for v in s["storage"]] # dependency protocol default is tcp. s["dependencies"] = [{**{"protocol": "tcp"}, **d} for d in s["dependencies"]] return s # ============================================================================= # Field -> resource mapping (renderer-contract.md "Field -> resource mapping") # ============================================================================= def r_namespace(s: dict) -> dict: """Namespace = the service trust zone / isolation boundary (R-NET, docs/03). Carries app-namespace=true so Kyverno generates the default-deny baseline and the egress-class / auth-mode admission policies apply. gateway-access=shared lets its HTTPRoutes attach to the shared Gateway (only when it exposes routes). """ labels = {f"{LBL}/app-namespace": "true"} if s["ingressRoutes"]: labels["gateway-access"] = "shared" ann = {} backup = (s.get("backup") or {}).get("policy") if backup: ann[f"{LBL}/backup-policy"] = backup # R-STATE-03 off-cluster schedule ref (hint) meta = {"name": s["name"], "labels": labels} if ann: meta["annotations"] = ann return {"apiVersion": "v1", "kind": "Namespace", "metadata": meta} def r_default_deny(s: dict) -> dict: """The default-deny baseline: deny all ingress+egress, allow ONLY DNS egress (R-NET-01). Rendered explicitly (GitOps at namespace creation, renderer-contract §dependencies) — it matches the Kyverno-generated baseline shape (platform/kyverno/.../generated-default -deny.yaml) so the render carries the guarantee even before admission generates it. """ return { "apiVersion": "networking.k8s.io/v1", "kind": "NetworkPolicy", "metadata": { "name": "default-deny", "namespace": s["name"], "labels": {f"{LBL}/baseline": "default-deny"}, }, "spec": { "podSelector": {}, "policyTypes": ["Ingress", "Egress"], "ingress": [], "egress": [ { "to": [ { "namespaceSelector": { "matchLabels": { "kubernetes.io/metadata.name": "kube-system" } } } ], "ports": [ {"protocol": "UDP", "port": 53}, {"protocol": "TCP", "port": 53}, ], } ], }, } def _pod_labels(s: dict) -> dict: """Workload pod labels: identity + the egress-class steering hook + auth-mode. egress-class is what the shared Cilium Egress Gateway selects on to steer vpn-class pods (enforcement is OUTSIDE the workload — renderer-contract §egress; ADR-0015); auth-mode satisfies the Kyverno require-auth-mode-label policy (R-IDP-02). """ return { "app.kubernetes.io/name": s["name"], f"{LBL}/egress-class": s["egress"], f"{LBL}/auth-mode": s["auth"]["mode"], } def _service_ports(s: dict) -> list: ports = sorted({r["port"] for r in s["ingressRoutes"]}) return [{"name": f"port-{p}", "port": p, "targetPort": p} for p in ports] def r_pvcs(s: dict) -> list: """One PVC per storage entry (R-STATE-01/02/04). longhorn -> 3-replica replicated tier; bulk -> external media tier (not 3x replicated). """ out = [] for vol in s["storage"]: cls = vol.get("class", "longhorn") storage_class = "longhorn" if cls == "longhorn" else "bulk" spec = { "accessModes": ["ReadWriteOnce"], "storageClassName": storage_class, } if vol.get("size"): spec["resources"] = {"requests": {"storage": vol["size"]}} out.append( { "apiVersion": "v1", "kind": "PersistentVolumeClaim", "metadata": { "name": f"{s['name']}-{vol['name']}", "namespace": s["name"], "labels": {"app.kubernetes.io/name": s["name"]}, }, "spec": spec, } ) return out def r_deployment(s: dict) -> dict: """Deployment for a container workload (R-PLAT-02).""" pod_labels = _pod_labels(s) container = { "name": s["name"], "image": s["source"]["image"], } ports = _service_ports(s) if ports: container["ports"] = [{"containerPort": p["port"]} for p in ports] volume_mounts, volumes = [], [] for vol in s["storage"]: # Default mount is /data/; 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/ default (back-compat). mount_path = vol.get("mountPath", f"/data/{vol['name']}") volume_mounts.append({"name": vol["name"], "mountPath": mount_path}) volumes.append( { "name": vol["name"], "persistentVolumeClaim": {"claimName": f"{s['name']}-{vol['name']}"}, } ) if volume_mounts: container["volumeMounts"] = volume_mounts pod_spec = {"containers": [container]} if volumes: pod_spec["volumes"] = volumes ann = _auth_annotations(s) tmpl_meta = {"labels": pod_labels} if ann: tmpl_meta["annotations"] = ann spec = {"replicas": 1} # An app with persistent storage mounts a ReadWriteOnce Longhorn PVC. The default # RollingUpdate strategy tries to start the NEW pod before terminating the old one, but # the old pod still holds the RWO volume -> the new pod is stuck ContainerCreating and # the rollout deadlocks (single-replica app can never hand the volume over). Recreate # terminates the old pod first, freeing the PVC for the new one (R-STATE-01/02). if s["storage"]: spec["strategy"] = {"type": "Recreate"} spec["selector"] = {"matchLabels": {"app.kubernetes.io/name": s["name"]}} spec["template"] = {"metadata": tmpl_meta, "spec": pod_spec} return { "apiVersion": "apps/v1", "kind": "Deployment", "metadata": { "name": s["name"], "namespace": s["name"], "labels": {"app.kubernetes.io/name": s["name"]}, }, "spec": spec, } def r_service(s: dict): ports = _service_ports(s) if not ports: return None return { "apiVersion": "v1", "kind": "Service", "metadata": { "name": s["name"], "namespace": s["name"], "labels": {"app.kubernetes.io/name": s["name"]}, }, "spec": { "selector": {"app.kubernetes.io/name": s["name"]}, "ports": ports, }, } def r_dependency_policies(s: dict) -> list: """One CiliumNetworkPolicy egress allow per dependency, on top of default-deny (R-NET-02). Everything undeclared stays denied; empty dependencies => reaches only DNS. """ 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"]}} ], "toPorts": [ { "ports": [ {"port": str(dep["port"]), "protocol": proto} ] } ], } ], }, } ) return out def r_gateway_ingress_policy(s: dict): """North-south allow: let the shared Cilium Gateway reach an exposed service's port(s). Any service that renders an HTTPRoute (non-empty ingressRoutes) is reachable through the shared Gateway, but the namespace default-deny has `ingress: []` (deny all east-west and north-south). This CiliumNetworkPolicy is the ADDITIVE, least-privilege allow that opens ONLY the north-south path (docs/09 "Ingress opens north-south only — it does not relax east-west default-deny"): ingress from Cilium's reserved `ingress` identity (the Gateway/ Ingress Envoy proxy, `reserved:ingress`, numeric identity 8 in Cilium 1.16.5) to ONLY the exposed service port(s). Everything else stays denied; undeclared ports stay denied. Verified empirically against the cluster: with default-deny in place, Hubble shows the Gateway proxy reaching the backend as source identity `(ingress)` and the flow `Policy denied DROPPED` on the service port — so `fromEntities: [ingress]` is exactly the source the Cilium Gateway uses, and the port is the service targetPort (R-NET-01/02). """ if not s["ingressRoutes"]: return None ports = _service_ports(s) return { "apiVersion": "cilium.io/v2", "kind": "CiliumNetworkPolicy", "metadata": { "name": f"{s['name']}-allow-gateway-ingress", "namespace": s["name"], "labels": {"app.kubernetes.io/name": s["name"]}, }, "spec": { "endpointSelector": { "matchLabels": {"app.kubernetes.io/name": s["name"]} }, "ingress": [ { "fromEntities": ["ingress"], "toPorts": [ { "ports": [ {"port": str(p["port"]), "protocol": "TCP"} for p in ports ] } ], } ], }, } def _route_meta(s: dict, route: dict) -> dict: meta = { "name": f"{s['name']}-{route['port']}", "namespace": s["name"], "labels": {"app.kubernetes.io/name": s["name"]}, } ann = _auth_annotations(s) if ann: meta["annotations"] = ann return meta def r_internet_egress(s: dict) -> dict: """North-south egress allow: let the app reach the INTERNET (Cilium reserved `world` entity = everything outside the cluster), which is the whole point of the `egress` class (docs/03: internet-class → WAN; vpn-class → steered through the node tun0 PIA tunnel with the nftables kill-switch, ADR-0015). The default-deny baseline (R-NET-01) allows only DNS so EAST-WEST stays denied; this ADDITIVE policy opens ONLY `world` (non-cluster) egress — cluster/east-west traffic stays denied except the declared `dependencies`. Every app declares an egress class (EGRESS_CLASSES), so every app gets this. The steering + kill- switch (a cluster-wide singleton keyed on the egress-class pod label) decide the PATH; this policy decides that north-south egress is ALLOWED at all. """ return { "apiVersion": "cilium.io/v2", "kind": "CiliumNetworkPolicy", "metadata": { "name": f"{s['name']}-allow-internet-egress", "namespace": s["name"], "labels": {"app.kubernetes.io/name": s["name"]}, }, "spec": { "endpointSelector": {"matchLabels": {"app.kubernetes.io/name": s["name"]}}, "egress": [{"toEntities": ["world"]}], }, } def r_httproutes(s: dict) -> list: """One HTTPRoute per ingressRoute, attaching host -> service port on the shared Gateway. TLS is the Gateway's wildcard *.home.keithjackson.dev cert (no per-service cert step); ExternalDNS manages the record from the HTTPRoute (R-TLS/R-DNS, renderer-contract). """ out = [] for route in s["ingressRoutes"]: out.append( { "apiVersion": "gateway.networking.k8s.io/v1", "kind": "HTTPRoute", "metadata": _route_meta(s, route), "spec": { "parentRefs": [ { "name": GATEWAY_NAME, "namespace": GATEWAY_NS, "sectionName": "https", } ], "hostnames": [route["host"]], "rules": [ { "matches": [ {"path": {"type": "PathPrefix", "value": "/"}} ], "backendRefs": [ {"name": s["name"], "port": route["port"]} ], } ], }, } ) return out def _auth_annotations(s: dict) -> dict: """Auth wiring hints (renderer-contract §auth): the platform (oauth2-proxy/Keycloak) consumes these to protect ingressRoutes. oidc/forward-auth carry the authorized groups (R-IDP-06: access is 'in an authorized group', not merely 'logged in').""" mode = s["auth"]["mode"] if mode in ("oidc", "forward-auth"): groups = ",".join(s["auth"].get("groups", [])) return { f"{LBL}/auth-mode": mode, f"{LBL}/auth-groups": groups, } return {} def r_servicemonitor(s: dict): if not s["observability"]["metrics"]: return None return { "apiVersion": "monitoring.coreos.com/v1", "kind": "ServiceMonitor", "metadata": { "name": s["name"], "namespace": s["name"], "labels": {"release": KP_RELEASE, "app.kubernetes.io/name": s["name"]}, }, "spec": { "selector": {"matchLabels": {"app.kubernetes.io/name": s["name"]}}, "namespaceSelector": {"matchNames": [s["name"]]}, "endpoints": [{"port": "metrics", "interval": "30s"}], }, } def r_prometheusrule(s: dict): """Each SLI becomes at least one alert rule (R-OBS-01).""" slis = s["observability"]["slis"] if not slis: return None rules = [] for sli in slis: if sli == "up": rules.append( { "alert": f"{s['name']}-down".replace("_", "-"), "expr": f'up{{job="{s["name"]}"}} == 0', "for": "5m", "labels": {"severity": "critical", "service": s["name"]}, "annotations": { "summary": f"{s['name']} is down", "description": f"{s['name']} has been unreachable for 5m (SLI: up).", }, } ) else: rules.append( { "alert": f"{s['name']}-{sli}".replace("_", "-"), "expr": f'absent({sli}{{service="{s["name"]}"}})', "for": "10m", "labels": {"severity": "warning", "service": s["name"]}, "annotations": {"summary": f"{s['name']} SLI {sli} unhealthy"}, } ) return { "apiVersion": "monitoring.coreos.com/v1", "kind": "PrometheusRule", "metadata": { "name": f"{s['name']}-slis", "namespace": s["name"], "labels": {"release": KP_RELEASE, "app.kubernetes.io/name": s["name"]}, }, "spec": {"groups": [{"name": f"{s['name']}.slis", "rules": rules}]}, } def r_image_automation(s: dict) -> list: """container kind -> Flux ImageRepository + ImagePolicy encoding channel/range (R-UPD). Flux image-automation advances the running version on its scan interval; no human/AI edits a tag (R-UPD-02). `pin` freezes automation. vm kind uses in-guest updates instead. """ if s["kind"] != "container": return [] image = s["source"]["image"].split(":", 1)[0] upd = s["update"] interval = "1h0m0s" imagerepo = { "apiVersion": "image.toolkit.fluxcd.io/v1beta2", "kind": "ImageRepository", "metadata": {"name": s["name"], "namespace": s["name"]}, "spec": {"image": image, "interval": interval}, } policy_spec: dict = {"imageRepositoryRef": {"name": s["name"]}} if upd.get("pin"): # Operator freeze escape hatch: pin to a fixed tag, stopping automation. policy_spec["filterTags"] = {"pattern": f"^{upd['pin']}$"} policy_spec["policy"] = {"alphabetical": {"order": "asc"}} elif upd.get("range"): policy_spec["policy"] = {"semver": {"range": upd["range"]}} else: policy_spec["policy"] = {"semver": {"range": ">=0.0.0"}} imagepolicy = { "apiVersion": "image.toolkit.fluxcd.io/v1beta2", "kind": "ImagePolicy", "metadata": { "name": s["name"], "namespace": s["name"], "annotations": { f"{LBL}/update-channel": upd.get("channel", "stable"), f"{LBL}/auto-rollback": str( upd.get("autoRollbackOnFailedHealth", True) ).lower(), }, }, "spec": policy_spec, } return [imagerepo, imagepolicy] # --- VM (kind: vm) path: KubeVirt VirtualMachine + CDI DataVolume ---------------------- def r_datavolume(s: dict) -> dict: """Standalone CDI DataVolume (NOT a dataVolumeTemplate) so the disk OUTLIVES the VM and survives destroy/rebuild (R-STATE-01) — matches the vm-lifecycle fixture shape.""" root = s["storage"][0] if s["storage"] else {"name": "rootdisk", "size": "10Gi"} return { "apiVersion": "cdi.kubevirt.io/v1beta1", "kind": "DataVolume", "metadata": { "name": f"{s['name']}-{root['name']}", "namespace": s["name"], }, "spec": { "source": {"registry": {"url": f"docker://{s['source']['vmTemplate']}"}}, "storage": { "accessModes": ["ReadWriteOnce"], "storageClassName": "longhorn", "resources": {"requests": {"storage": root.get("size", "10Gi")}}, }, }, } VM_CLOUD_INIT = ( "#cloud-config\n" "package_update: true\n" "packages:\n" " - dnf-automatic\n" "write_files:\n" " - path: /etc/dnf/automatic.conf\n" " content: |\n" " [commands]\n" " upgrade_type = default\n" " random_sleep = 0\n" " download_updates = yes\n" " apply_updates = yes\n" "runcmd:\n" " - [ systemctl, enable, --now, dnf-automatic.timer ]\n" ) def r_virtualmachine(s: dict) -> dict: """KubeVirt VirtualMachine; egress-class propagates to the virt-launcher pod so the shared Cilium egress steering + Kyverno policy see the class (R-PLAT-03; ADR-0015). In-guest dnf-automatic delivers auto-updates with no human step (R-UPD-01, docs/11).""" root = s["storage"][0] if s["storage"] else {"name": "rootdisk"} pod_labels = { "kubevirt.io/vm": s["name"], f"{LBL}/egress-class": s["egress"], f"{LBL}/auth-mode": s["auth"]["mode"], } return { "apiVersion": "kubevirt.io/v1", "kind": "VirtualMachine", "metadata": { "name": s["name"], "namespace": s["name"], "labels": {"app.kubernetes.io/name": s["name"]}, }, "spec": { "running": False, "template": { "metadata": {"labels": pod_labels}, "spec": { "domain": { "cpu": {"cores": 1}, "memory": {"guest": "1536Mi"}, "devices": { "disks": [ {"name": "rootdisk", "disk": {"bus": "virtio"}}, {"name": "cloudinit", "disk": {"bus": "virtio"}}, ], "interfaces": [{"name": "default", "masquerade": {}}], }, "resources": {"requests": {"memory": "1536Mi"}}, }, "networks": [{"name": "default", "pod": {}}], "volumes": [ { "name": "rootdisk", "dataVolume": {"name": f"{s['name']}-{root['name']}"}, }, { "name": "cloudinit", "cloudInitNoCloud": {"userData": VM_CLOUD_INIT}, }, ], }, }, }, } # ============================================================================= # Top-level render + emit # ============================================================================= def render(sd: dict, image_automation: bool = True) -> 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. """ s = apply_defaults(sd) if s["kind"] not in ("container", "vm"): raise ValueError(f"unknown kind: {s['kind']!r}") if s["egress"] not in EGRESS_CLASSES: raise ValueError(f"invalid egress class: {s['egress']!r}") if s["auth"]["mode"] not in AUTH_MODES: raise ValueError(f"invalid auth.mode: {s['auth']['mode']!r}") docs: list = [] docs.append(("Namespace (trust zone; default-deny + policies apply)", r_namespace(s))) docs.append(("Default-deny baseline (R-NET-01): deny all, allow only DNS", r_default_deny(s))) if s["kind"] == "container": # Container PVCs. (For a vm, the CDI DataVolume owns/creates the disk PVC, so we # never render a plain PVC for the VM root disk — it would collide with CDI.) for pvc in r_pvcs(s): docs.append((f"PVC {pvc['metadata']['name']} (R-STATE)", pvc)) docs.append(("Deployment (R-PLAT-02)", r_deployment(s))) svc = r_service(s) if svc: docs.append(("Service", svc)) else: docs.append(("CDI DataVolume (standalone; survives rebuild, R-STATE-01)", r_datavolume(s))) docs.append(("KubeVirt VirtualMachine (R-PLAT-03)", r_virtualmachine(s))) for cnp in r_dependency_policies(s): docs.append((f"CiliumNetworkPolicy egress allow (R-NET-02): {cnp['metadata']['name']}", cnp)) gw_allow = r_gateway_ingress_policy(s) if gw_allow: docs.append( ( "CiliumNetworkPolicy gateway-ingress allow (R-NET-01/02): " f"{gw_allow['metadata']['name']}", gw_allow, ) ) ie = r_internet_egress(s) docs.append( ( "CiliumNetworkPolicy world egress allow (egress class governs the path; R-NET-05): " f"{ie['metadata']['name']}", ie, ) ) for hr in r_httproutes(s): docs.append((f"HTTPRoute {hr['metadata']['name']} (R-TLS/R-DNS)", hr)) sm = r_servicemonitor(s) if sm: docs.append(("ServiceMonitor (R-OBS-01)", sm)) pr = r_prometheusrule(s) if pr: docs.append(("PrometheusRule SLI alerts (R-OBS-01)", pr)) if image_automation: for res in r_image_automation(s): docs.append((f"Flux {res['kind']} (R-UPD)", res)) return docs def emit(docs: list) -> str: """Deterministic multi-document YAML text (golden-diffable).""" parts = [] for comment, resource in docs: body = yaml.dump( resource, sort_keys=False, default_flow_style=False, width=4096, ) parts.append(f"---\n# {comment}\n{body}".rstrip() + "\n") return "".join(parts) def load_sd(path: Path) -> dict: with path.open() as fh: return yaml.safe_load(fh) def render_file(path: Path, image_automation: bool = True) -> str: return emit(render(load_sd(path), image_automation=image_automation)) # ============================================================================= # 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") check( any(c["metadata"]["name"] == "y-allow-internet-egress" for c in cnps), "every app (egress class) must render a world/internet egress allow", ) # 4b) exposure (ingressRoutes) -> an ADDITIVE gateway-ingress CiliumNetworkPolicy that # allows ONLY the reserved `ingress` (Gateway proxy) identity to the exposed port, # while default-deny still denies all other ingress (R-NET-01/02). exp_docs = render( { "name": "z", "source": {"image": "a:1"}, "ingressRoutes": [{"host": "z.home.keithjackson.dev", "port": 80}], } ) gw = [ d[1] for d in exp_docs if d[1]["kind"] == "CiliumNetworkPolicy" and d[1]["metadata"]["name"].endswith("-allow-gateway-ingress") ] check(len(gw) == 1, "an exposed service must render one gateway-ingress allow (R-NET-01)") if gw: ing = gw[0]["spec"]["ingress"] check( ing[0]["fromEntities"] == ["ingress"], "gateway-ingress allow must be from the reserved `ingress` (Gateway) identity", ) allowed_ports = {p["port"] for r in ing for tp in r["toPorts"] for p in tp["ports"]} check( allowed_ports == {"80"}, "gateway-ingress allow must be least-privilege (only the exposed port)", ) # default-deny stays deny-all-ingress even when an exposed service adds the allow. exp_dd = next(d[1] for d in exp_docs if d[1]["kind"] == "NetworkPolicy") check( exp_dd["spec"]["ingress"] == [], "gateway-ingress allow must be ADDITIVE — default-deny ingress stays [] (R-NET-01)", ) # an internal (no ingressRoutes) service must NOT get a gateway-ingress allow. internal_docs = render({"name": "int", "source": {"image": "a:1"}}) check( not any( d[1]["kind"] == "CiliumNetworkPolicy" and d[1]["metadata"]["name"].endswith("-allow-gateway-ingress") for d in internal_docs ), "a non-exposed service must not render a gateway-ingress allow (least privilege)", ) # 5) vm kind -> VirtualMachine + standalone DataVolume. vm_docs = render( { "name": "vmx", "kind": "vm", "source": {"vmTemplate": "quay.io/containerdisks/fedora:41"}, "storage": [{"name": "rootdisk", "size": "6Gi", "class": "longhorn"}], } ) vm_kinds = [d[1]["kind"] for d in vm_docs] check("VirtualMachine" in vm_kinds, "vm: missing VirtualMachine") check("DataVolume" in vm_kinds, "vm: missing CDI DataVolume") 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 .yaml files into this dir") ap.add_argument("--selftest", action="store_true", help="Run the built-in self-test") ap.add_argument("--no-image-automation", action="store_true", help="Omit Flux ImageRepository/ImagePolicy CRs (for onboarding before the " "image-reflector/-automation controllers are installed)") args = ap.parse_args(argv) if args.selftest: return _selftest() image_automation = not args.no_image_automation if args.all: src = Path(args.all) files = sorted(p for p in src.glob("*.yaml") if p.name != "render.py") for f in files: text = render_file(f, image_automation=image_automation) if args.out_dir: outdir = Path(args.out_dir) outdir.mkdir(parents=True, exist_ok=True) name = load_sd(f)["name"] (outdir / f"{name}.yaml").write_text(text) else: sys.stdout.write(text) return 0 if not args.input: ap.error("provide a ServiceDefinition file, --all DIR, or --selftest") sys.stdout.write(render_file(Path(args.input), image_automation=image_automation)) return 0 if __name__ == "__main__": raise SystemExit(main())