qbit-scripting/deploy/render_qualification.py

# 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()