# Copyright (C) 2026 Kingsrook, LLC. # SPDX-License-Identifier: Apache-2.0 """Static rendering and fail-closed snapshot tests; no containers or clusters.""" import copy import hashlib import json from pathlib import Path import re import subprocess import sys import tempfile from typing import Any import unittest import render_qualification as renderer import snapshot_execution as snapshot def fixture() -> dict[str, Any]: """Supply synthetic identifiers for rendering tests only.""" # Synthetic non-secret identifiers; fixtures are never applied. return { "enabled": True, "namespace": "qqq-qual-unit", "image": "invalid.example/unit@sha256:" + "a" * 64, "sandboxRuntimeClass": "unit-sandbox", "podSecurityVersion": "v1.34", "tlsSecret": "unit-server", "trustSecret": "unit-clients", "passwordSecret": "unit-passwords", "certificateNamespaces": {"b" * 64: "unit-client"}, "clientNamespace": "unit-client", "clientPodLabels": {"app": "unit-client"}, "nodeSelector": {"qualification-pool": "unit"}, "profile": "unit-profile", "toolchain": "unit-toolchain", "evidence": "unit-evidence", "requiredEvidence": "unit-evidence", "cpuMillis": 1000, "memoryMiB": 1024, "pidLimit": 256, "pidPolicyEvidence": "unit-node-policy", "tmpMiB": 16, "startupSeconds": 120, "slots": 1, "connections": 2, "coldOperations": 1, "entries": 8, "codeBytes": 8388608, "guestMemoryMiB": 64, "idleMillis": 30000, "fuel": 100000000, } class DeploymentTest(unittest.TestCase): """Verify generated boundaries without a deployment or native runtime.""" def test_disabled_input_and_every_missing_field_refuse(self) -> None: """An incomplete descriptor never yields runnable manifests.""" disabled = json.loads((Path(__file__).parent / "qualification.disabled.json").read_text()) with self.assertRaises(ValueError): renderer.render(disabled) for field in renderer.REQUIRED: value = fixture() value.pop(field) with self.subTest(field=field), self.assertRaises(ValueError): renderer.render(value) def test_fixed_security_network_and_exact_supervisor_config(self) -> None: """Keep secrets referenced and network authority narrowly selected.""" manifest = renderer.render(fixture()) by_kind = {entry["kind"]: entry for entry in manifest["items"] if entry["kind"] != "NetworkPolicy"} pod = by_kind["Deployment"]["spec"]["template"]["spec"] container = pod["containers"][0] self.assertFalse(pod["automountServiceAccountToken"]) self.assertFalse(pod["enableServiceLinks"]) self.assertEqual(65532, pod["securityContext"]["runAsUser"]) self.assertEqual({"type": "RuntimeDefault"}, pod["securityContext"]["seccompProfile"]) self.assertEqual(["ALL"], container["securityContext"]["capabilities"]["drop"]) self.assertTrue(container["securityContext"]["readOnlyRootFilesystem"]) self.assertFalse(container["securityContext"]["allowPrivilegeEscalation"]) self.assertNotIn("livenessProbe", container) self.assertNotIn("pidsLimit", pod) self.assertTrue(all("valueFrom" in item and "value" not in item for item in container["env"])) settings = json.loads(by_kind["ConfigMap"]["data"]["config.json"]) self.assertEqual(snapshot.WASM_SHA, settings["runtimeSha256"]) self.assertEqual(snapshot.EXTISM_SHA, settings["nativeSha256"]) self.assertEqual("0.0.0.0", settings["bind"]) self.assertFalse(any("password" in key.lower() for key in settings)) policies = [item["spec"] for item in manifest["items"] if item["kind"] == "NetworkPolicy"] self.assertEqual([], policies[0]["egress"]) self.assertEqual([], policies[0]["ingress"]) peer = policies[1]["ingress"][0]["from"] self.assertEqual(1, len(peer)) self.assertEqual({"namespaceSelector", "podSelector"}, set(peer[0])) self.assertEqual([{"protocol": "TCP", "port": 18443}], policies[1]["ingress"][0]["ports"]) self.assertNotIn("Secret", by_kind) self.assertNotIn("RuntimeClass", by_kind) def test_malformed_policy_cannot_fall_back(self) -> None: """Reject permissive substitutes and mixed JSON types.""" changes = {"image": "example/unit:latest", "namespace": "production", "sandboxRuntimeClass": "runc", "certificateNamespaces": {"*": "unit"}, "clientPodLabels": {}, "nodeSelector": {}, "pidLimit": 0, "cpuMillis": True, "memoryMiB": 32, "slots": 65, "requiredEvidence": "different", "enabled": "true", "profile": "REPLACE_ME"} for field, value in changes.items(): config = fixture() config[field] = value with self.subTest(field=field), self.assertRaises(ValueError): renderer.render(config) config = fixture() config["secretValue"] = "not-accepted" with self.assertRaises(ValueError): renderer.render(config) def test_input_is_not_mutated_and_duplicate_json_refused(self) -> None: """Preserve caller intent and reject ambiguous repeated keys.""" config = fixture() before = copy.deepcopy(config) renderer.render(config) self.assertEqual(before, config) for hook in (renderer.no_duplicates, snapshot.no_duplicates): with self.assertRaises(ValueError): json.loads('{"enabled":true,"enabled":false}', object_pairs_hook=hook) def test_wire_identity_and_kubernetes_label_grammars_refuse_invalid_values(self) -> None: """Refuse values accepted previously but rejected by the real consumers.""" for value in ("wrong/profile", "x" * 129): for key in ("profile", "toolchain", "evidence", "requiredEvidence"): config = fixture() config[key] = value if key in {"evidence", "requiredEvidence"}: config["evidence"] = config["requiredEvidence"] = value with self.subTest(key=key, value=value), self.assertRaises(ValueError): renderer.render(config) config = fixture() config["certificateNamespaces"] = {"b" * 64: value} with self.assertRaises(ValueError): renderer.render(config) for selectors in ({"app-": "client"}, {"app": "client."}, {"bad..domain/app": "client"}, {"-domain/app": "client"}, {"a" * 64 + "/app": "client"}): for key in ("clientPodLabels", "nodeSelector"): config = fixture() config[key] = selectors with self.subTest(key=key, selectors=selectors), self.assertRaises(ValueError): renderer.render(config) config = fixture() config["profile"] = "x" * 128 config["clientPodLabels"] = {"qualified.example/app_name": "client.v1"} renderer.render(config) def test_complete_positive_fixture_cross_object_contracts(self) -> None: """Review the entire rendered fixture and its actual mount/selector linkage.""" directory = Path(__file__).parent / "fixtures" source = json.loads((directory / "synthetic-input.json").read_text(encoding="utf-8")) expected = json.loads((directory / "synthetic-manifest.json").read_text(encoding="utf-8")) actual = renderer.render(source) self.assertEqual(expected, actual) self.assertEqual(6, len(actual["items"])) identities = [(item["kind"], item["metadata"].get("namespace"), item["metadata"]["name"]) for item in actual["items"]] self.assertEqual(len(identities), len(set(identities))) by_kind = {item["kind"]: item for item in actual["items"] if item["kind"] != "NetworkPolicy"} namespace = by_kind["Namespace"] self.assertNotIn("namespace", namespace["metadata"]) self.assertEqual("restricted", namespace["metadata"]["labels"]["pod-security.kubernetes.io/enforce"]) for item in actual["items"]: if item["kind"] != "Namespace": self.assertEqual(namespace["metadata"]["name"], item["metadata"]["namespace"]) deployment = by_kind["Deployment"]["spec"] template = deployment["template"] pod = template["spec"] container = pod["containers"][0] self.assertEqual(deployment["selector"]["matchLabels"], template["metadata"]["labels"]) self.assertEqual(template["metadata"]["labels"], by_kind["Service"]["spec"]["selector"]) self.assertEqual("false", template["metadata"]["annotations"]["qqq.kingsrook.com/productionQualified"]) self.assertEqual("false", template["metadata"]["annotations"]["qqq.kingsrook.com/redistributionCleared"]) self.assertEqual(source["cpuMillis"], int(container["resources"]["limits"]["cpu"][:-1])) self.assertEqual(container["resources"]["requests"]["memory"], container["resources"]["limits"]["memory"]) self.assertEqual("ClusterIP", by_kind["Service"]["spec"]["type"]) self.assertNotIn("externalIPs", by_kind["Service"]["spec"]) self.assertEqual("mtls", by_kind["Service"]["spec"]["ports"][0]["targetPort"]) self.assertEqual(18443, container["ports"][0]["containerPort"]) config_map = by_kind["ConfigMap"] settings = json.loads(config_map["data"]["config.json"]) self.assertTrue(config_map["immutable"]) volumes = {volume["name"]: volume for volume in pod["volumes"]} mounts = {mount["name"]: mount for mount in container["volumeMounts"]} self.assertEqual(set(volumes), set(mounts)) self.assertEqual(config_map["metadata"]["name"], volumes["config"]["configMap"]["name"]) for name, setting, secret_field in (("tls", "keyStore", "tlsSecret"), ("trust", "trustStore", "trustSecret")): secret = volumes[name]["secret"] self.assertEqual(source[secret_field], secret["secretName"]) self.assertEqual(0o440, secret["defaultMode"]) self.assertTrue(mounts[name]["readOnly"]) self.assertEqual(mounts[name]["mountPath"] + "/" + secret["items"][0]["path"], settings[setting]) self.assertEqual(0o444, volumes["config"]["configMap"]["defaultMode"]) self.assertEqual({"medium": "Memory", "sizeLimit": str(source["tmpMiB"]) + "Mi"}, volumes["tmp"]["emptyDir"]) for forbidden in ("hostPID", "hostIPC", "hostNetwork", "hostUsers", "initContainers", "ephemeralContainers"): self.assertNotIn(forbidden, pod) self.assertFalse(any("hostPath" in value or "persistentVolumeClaim" in value for value in volumes.values())) self.assertEqual(30, pod["terminationGracePeriodSeconds"]) self.assertNotIn("livenessProbe", container) def test_actual_cli_renders_fixture_once_and_refuses_disabled_without_output(self) -> None: """Exercise file parsing and exclusive output creation without cluster tools.""" root = Path(__file__).parent with tempfile.TemporaryDirectory() as temporary: output = Path(temporary) / "manifest.json" command = [sys.executable, str(root / "render_qualification.py"), str(root / "fixtures/synthetic-input.json"), str(output)] first = subprocess.run(command, capture_output=True, timeout=10, check=False) self.assertEqual(0, first.returncode) self.assertEqual(json.loads((root / "fixtures/synthetic-manifest.json").read_text(encoding="utf-8")), json.loads(output.read_text(encoding="utf-8"))) before = output.read_bytes() second = subprocess.run(command, capture_output=True, timeout=10, check=False) self.assertNotEqual(0, second.returncode) self.assertEqual(before, output.read_bytes()) disabled_output = Path(temporary) / "disabled.json" refused = subprocess.run([sys.executable, str(root / "render_qualification.py"), str(root / "qualification.disabled.json"), str(disabled_output)], capture_output=True, timeout=10, check=False) self.assertNotEqual(0, refused.returncode) self.assertFalse(disabled_output.exists()) def test_rendered_config_matches_actual_closed_java_entrypoint(self) -> None: """Catch drift between packaging and the real SupervisorMain schema.""" java = (Path(__file__).parent.parent / "supervisor/src/main/java/com/kingsrook/scripting/supervisor/SupervisorMain.java").read_text(encoding="utf-8") matched = re.search(r"Protocol\.keys\(config, (.*?)\);", java, re.DOTALL) self.assertIsNotNone(matched) assert matched is not None accepted = set(re.findall(r'"([A-Za-z0-9]+)"', matched.group(1))) config_map = next(item for item in renderer.render(fixture())["items"] if item["kind"] == "ConfigMap") self.assertEqual(accepted, set(json.loads(config_map["data"]["config.json"]))) def test_context_verifier_refuses_unlisted_file(self) -> None: """Even a harmless extra context file cannot silently enter the image.""" with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) rows = [] files = {relative for relative, _ in snapshot.FILES.values()} | { "payload/native/libjnidispatch.so", "Containerfile", "metadata/NOTICE-GAPS.md", ".dockerignore", } for relative in files: target = root / relative target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(b"bounded test bytes") rows.append({"path": relative, "sha256": snapshot.digest(target)}) manifest = {"productionQualified": False, "redistributionCleared": False, "architecture": "linux/amd64", "runtimeSha256": snapshot.WASM_SHA, "nativeSha256": snapshot.EXTISM_SHA, "files": rows} (root / "metadata/candidate.json").write_text(json.dumps(manifest), encoding="utf-8") (root / "unlisted.txt").write_text("must refuse", encoding="utf-8") with self.assertRaisesRegex(ValueError, "Unexpected or missing build-context file"): snapshot.verify_snapshot(root) def test_snapshot_missing_unknown_and_hash_mismatch_refuse_before_output(self) -> None: """Do not create a context after any input-integrity failure.""" with tempfile.TemporaryDirectory() as temporary: output = Path(temporary) / "output" for descriptor in ({}, {"certificate": {}}): with self.assertRaises(ValueError): snapshot.snapshot(descriptor, output) source = Path(temporary) / "source" source.write_bytes(b"fixed harmless fixture") descriptor = {key: {"path": str(source), "sha256": "0" * 64} for key in snapshot.FILES} with self.assertRaises(ValueError): snapshot.snapshot(descriptor, output) self.assertFalse(output.exists()) class PinnedWasmTest(unittest.TestCase): """The fixed Wasm pin must always correspond to the recorded source inputs.""" RUNTIME = Path(__file__).resolve().parent.parent / "wasm-runtime" def pin(self) -> dict[str, Any]: """Load the recorded pin.""" return json.loads((self.RUNTIME / "PINNED-WASM.json").read_text(encoding="utf-8")) def test_deploy_constant_and_inventories_equal_the_recorded_pin(self) -> None: """Every place that names the fixed Wasm hash must agree with PINNED-WASM.json.""" pin = self.pin() self.assertEqual(pin["sha256"], snapshot.WASM_SHA) inventory = json.loads((self.RUNTIME / "docs" / "license-inventory.json").read_text(encoding="utf-8")) imports = json.loads((self.RUNTIME / "docs" / "wasm-imports.json").read_text(encoding="utf-8")) self.assertEqual(pin["sha256"], inventory["artifactSha256"]) self.assertEqual(pin["sha256"], imports["sha256"]) self.assertEqual(pin["bytes"], imports["bytes"]) review = (self.RUNTIME / "docs" / "release-license-review.md").read_text(encoding="utf-8") self.assertIn(pin["sha256"], review) def test_pin_matches_the_current_source_inputs_and_lockfile(self) -> None: """A source or Cargo.lock edit without a reviewed re-pin must fail here, not silently drift.""" pin = self.pin() combined = "" for name in pin["sourceInputs"]: digest = hashlib.sha256((self.RUNTIME / name).read_bytes()).hexdigest() combined += name + "\n" + digest + "\n" self.assertEqual(pin["sourceInputsSha256"], hashlib.sha256(combined.encode("utf-8")).hexdigest()) self.assertEqual(pin["cargoLockSha256"], hashlib.sha256((self.RUNTIME / "Cargo.lock").read_bytes()).hexdigest()) def test_pin_names_locked_toolchain_and_previous_artifact(self) -> None: """The recipe inputs are explicit: no unpinned tool, and the superseded hash stays recorded.""" pin = self.pin() self.assertRegex(pin["sha256"], r"^[0-9a-f]{64}$") self.assertNotEqual(pin["sha256"], pin["previous"]["sha256"]) self.assertEqual("1.97.1", pin["toolchain"]["rust"]) for key in ("archiveSha256", "clangExeSha256", "libclangDllSha256", "wasiLibcSha256"): self.assertRegex(pin["toolchain"]["wasiSdk"][key], r"^[0-9a-f]{64}$") self.assertFalse(pin["qualification"].count("productionQualified=true")) script = (self.RUNTIME / "Build-Wasm.ps1").read_text(encoding="utf-8") for fragment in ("--offline", "--locked", "OutputDir must not already exist", "does not match the pin"): self.assertIn(fragment, script) if __name__ == "__main__": unittest.main()