# Copyright (C) 2026 Kingsrook, LLC. # SPDX-License-Identifier: Apache-2.0 """Render inert qualification JSON into a bounded Kubernetes List; never apply it.""" import argparse import json import logging from pathlib import Path import re from typing import Any from snapshot_execution import EXTISM_SHA, WASM_SHA NAME = r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?" IDENTITY = r"[A-Za-z0-9._:-]{1,128}" LABEL = r"[A-Za-z0-9](?:[A-Za-z0-9_.-]{0,61}[A-Za-z0-9])?" SHA = r"[a-f0-9]{64}" REQUIRED = { "enabled", "namespace", "image", "sandboxRuntimeClass", "podSecurityVersion", "tlsSecret", "trustSecret", "passwordSecret", "certificateNamespaces", "clientNamespace", "clientPodLabels", "nodeSelector", "profile", "toolchain", "evidence", "requiredEvidence", "cpuMillis", "memoryMiB", "pidLimit", "pidPolicyEvidence", "tmpMiB", "startupSeconds", "slots", "connections", "coldOperations", "entries", "codeBytes", "guestMemoryMiB", "idleMillis", "fuel", } def require(condition: bool, message: str) -> None: """Refuse malformed deployment intent without a fallback.""" if not condition: raise ValueError(message) def text(value: Any, pattern: str) -> bool: """Validate exact string types and the whole bounded value.""" return isinstance(value, str) and re.fullmatch(pattern, value) is not None def number(config: dict[str, Any], name: str, minimum: int, maximum: int) -> int: """Reject booleans, coercion and values outside the candidate limits.""" value = config[name] require(type(value) is int and minimum <= value <= maximum, "Invalid bound: " + name) return value def labels(value: Any) -> dict[str, str]: """Require explicit, nonempty selectors for client and node isolation.""" require(isinstance(value, dict) and 1 <= len(value) <= 8, "Explicit nonempty selectors required") for key, item in value.items(): require(isinstance(key, str) and key.count("/") <= 1, "Invalid selector key") parts = key.split("/") if len(parts) == 2: require(len(parts[0]) <= 253 and all(text(part, NAME) for part in parts[0].split(".")), "Invalid selector DNS prefix") require(text(parts[-1], LABEL) and text(item, LABEL), "Invalid selector") return dict(value) def render(config: dict[str, Any]) -> dict[str, Any]: """Build namespace-scoped qualification objects from explicit operator intent.""" require(isinstance(config, dict) and set(config) == REQUIRED, "Exact qualification fields required") require(config["enabled"] is True, "Qualification is disabled") for key in ("namespace", "sandboxRuntimeClass", "tlsSecret", "trustSecret", "passwordSecret", "clientNamespace"): require(text(config[key], NAME), "Invalid name: " + key) require(config["namespace"].startswith("qqq-qual-") and config["namespace"] != config["clientNamespace"], "Use a separate explicit qqq-qual-* namespace") require(config["sandboxRuntimeClass"] not in {"runc", "default"}, "An explicit sandbox RuntimeClass is required") require(text(config["podSecurityVersion"], r"v1\.[0-9]{2}"), "Explicit Pod Security version required") require(text(config["image"], r"[a-z0-9][a-z0-9./:_-]{1,240}@sha256:" + SHA), "Immutable image@sha256 required") for key in ("profile", "toolchain", "evidence", "requiredEvidence", "pidPolicyEvidence"): require(text(config[key], IDENTITY), "Explicit identity required: " + key) require(not any(word in config[key].lower() for word in ("replace", "placeholder", "unqualified")), "Placeholder identity refused") require(config["evidence"] == config["requiredEvidence"], "Evidence identity mismatch") mappings = config["certificateNamespaces"] require(isinstance(mappings, dict) and 1 <= len(mappings) <= 64, "Exact client certificate mappings required") require(all(text(key, SHA) and text(value, IDENTITY) for key, value in mappings.items()), "Invalid certificate mapping") client_labels = labels(config["clientPodLabels"]) node_labels = labels(config["nodeSelector"]) cpu = number(config, "cpuMillis", 100, 8000) memory = number(config, "memoryMiB", 512, 16384) pids = number(config, "pidLimit", 64, 2048) temporary = number(config, "tmpMiB", 1, 256) startup = number(config, "startupSeconds", 30, 600) slots = number(config, "slots", 1, 64) connections = number(config, "connections", slots, 128) cold = number(config, "coldOperations", 1, 8) entries = number(config, "entries", 1, 256) code = number(config, "codeBytes", 1, 64 * 1024 * 1024) guest = number(config, "guestMemoryMiB", 64, 256) require(guest % 64 == 0, "Guest profile must use whole 64MiB steps") require(memory >= slots * guest + 256 + temporary, "Insufficient minimum candidate memory headroom") idle = number(config, "idleMillis", 100, 60000) fuel = number(config, "fuel", 1, 1000000000) namespace = config["namespace"] name = "execution-supervisor" selector = {"app.kubernetes.io/name": name} annotations = { "qqq.kingsrook.com/productionQualified": "false", "qqq.kingsrook.com/redistributionCleared": "false", "qqq.kingsrook.com/expected-pod-pids-limit": str(pids), "qqq.kingsrook.com/pids-policy-evidence": config["pidPolicyEvidence"], "qqq.kingsrook.com/readiness-scope": "cold-listener-only-not-required-bindings", } settings = { "bind": "0.0.0.0", "port": 18443, "keyStore": "/run/qqq-supervisor/tls/server.p12", "trustStore": "/run/qqq-supervisor/trust/clients.p12", "certificateNamespaces": mappings, "profile": config["profile"], "toolchain": config["toolchain"], "evidence": config["evidence"], "requiredEvidence": config["requiredEvidence"], "runtimePath": "/opt/qqq-supervisor/runtime/qqq_javy_runtime.wasm", "runtimeSha256": WASM_SHA, "nativePath": "/opt/qqq-supervisor/native/libextism.so", "nativeSha256": EXTISM_SHA, "slots": slots, "connections": connections, "coldOperations": cold, "entries": entries, "codeBytes": code, "memoryBytes": guest * 1024 * 1024, "idleMillis": idle, "fuel": fuel, } def resource(kind: str, spec: dict[str, Any], api: str = "v1", extra: dict[str, Any] | None = None) -> dict[str, Any]: """Give each object an explicit qualification namespace.""" result = {"apiVersion": api, "kind": kind, "metadata": {"name": name, "namespace": namespace}, **spec} if extra: result["metadata"].update(extra) return result probe = {"tcpSocket": {"port": "mtls"}, "timeoutSeconds": 1, "periodSeconds": 5} container = { "name": name, "image": config["image"], "imagePullPolicy": "IfNotPresent", "ports": [{"name": "mtls", "containerPort": 18443}], "securityContext": {"allowPrivilegeEscalation": False, "readOnlyRootFilesystem": True, "capabilities": {"drop": ["ALL"]}}, "resources": {"requests": {"cpu": str(cpu) + "m", "memory": str(memory) + "Mi", "ephemeral-storage": "64Mi"}, "limits": {"cpu": str(cpu) + "m", "memory": str(memory) + "Mi", "ephemeral-storage": "128Mi"}}, "startupProbe": {**probe, "failureThreshold": (startup + 4) // 5}, "readinessProbe": {**probe, "failureThreshold": 1}, "env": [{"name": "QQQ_SUPERVISOR_" + key + "_PASSWORD", "valueFrom": {"secretKeyRef": {"name": config["passwordSecret"], "key": field}}} for key, field in (("KEYSTORE", "keyStorePassword"), ("TRUSTSTORE", "trustStorePassword"))], "volumeMounts": [{"name": "config", "mountPath": "/run/qqq-supervisor/config", "readOnly": True}, {"name": "tls", "mountPath": "/run/qqq-supervisor/tls", "readOnly": True}, {"name": "trust", "mountPath": "/run/qqq-supervisor/trust", "readOnly": True}, {"name": "tmp", "mountPath": "/tmp"}], } pod = {"automountServiceAccountToken": False, "enableServiceLinks": False, "runtimeClassName": config["sandboxRuntimeClass"], "nodeSelector": node_labels, "terminationGracePeriodSeconds": 30, "securityContext": {"runAsNonRoot": True, "runAsUser": 65532, "runAsGroup": 65532, "fsGroup": 65532, "seccompProfile": {"type": "RuntimeDefault"}}, "containers": [container], "volumes": [{"name": "config", "configMap": {"name": name, "defaultMode": 292}}, {"name": "tls", "secret": {"secretName": config["tlsSecret"], "defaultMode": 288, "items": [{"key": "server.p12", "path": "server.p12"}]}}, {"name": "trust", "secret": {"secretName": config["trustSecret"], "defaultMode": 288, "items": [{"key": "clients.p12", "path": "clients.p12"}]}}, {"name": "tmp", "emptyDir": {"medium": "Memory", "sizeLimit": str(temporary) + "Mi"}}]} items = [ {"apiVersion": "v1", "kind": "Namespace", "metadata": {"name": namespace, "labels": { "pod-security.kubernetes.io/enforce": "restricted", "pod-security.kubernetes.io/enforce-version": config["podSecurityVersion"], "pod-security.kubernetes.io/audit": "restricted", "pod-security.kubernetes.io/warn": "restricted"}}}, resource("ConfigMap", {"immutable": True, "data": {"config.json": json.dumps(settings, separators=(",", ":"))}}), resource("Deployment", {"spec": {"replicas": 1, "strategy": {"type": "Recreate"}, "selector": {"matchLabels": selector}, "template": {"metadata": {"labels": selector, "annotations": annotations}, "spec": pod}}}, "apps/v1"), resource("Service", {"spec": {"type": "ClusterIP", "selector": selector, "ports": [{"name": "mtls", "port": 18443, "targetPort": "mtls"}]}}), resource("NetworkPolicy", {"spec": {"podSelector": {}, "policyTypes": ["Ingress", "Egress"], "ingress": [], "egress": []}}, "networking.k8s.io/v1", {"name": "default-deny"}), resource("NetworkPolicy", {"spec": {"podSelector": {"matchLabels": selector}, "policyTypes": ["Ingress"], "ingress": [{"from": [{"namespaceSelector": {"matchLabels": {"kubernetes.io/metadata.name": config["clientNamespace"]}}, "podSelector": {"matchLabels": client_labels}}], "ports": [{"protocol": "TCP", "port": 18443}]}]}}, "networking.k8s.io/v1"), ] return {"apiVersion": "v1", "kind": "List", "items": items} def no_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]: """Refuse ambiguous policy documents.""" result = {} for key, value in pairs: require(key not in result, "Duplicate JSON key") result[key] = value return result def main() -> None: """Write a new local artifact; never contact Kubernetes.""" parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("input", type=Path) parser.add_argument("output", type=Path) args = parser.parse_args() require(args.input.stat().st_size <= 32768, "Input too large") manifest = render(json.loads(args.input.read_text(encoding="utf-8"), object_pairs_hook=no_duplicates)) with args.output.open("x", encoding="utf-8") as stream: stream.write(json.dumps(manifest, indent=2) + "\n") logging.warning("Qualification manifest rendered; not deployed or qualified") if __name__ == "__main__": main()