39a544828f
Some checks failed
Tests / Declarative: Post Actions failed: 71, skipped: 28, passed: 3680
Re-applies the staged HUX topology pinned to the reviewed build-21 image (git-2f535d3a...-build-21-release@sha256:e5b9b2fa...), with the first-activation posture: HUX_FLAGS=hux.foundation only, HUX_TOOL_ENFORCEMENT=0, and a RollingUpdate partition of 3 so only hermes-chat-tenant-3 rolls. Adds the /healthz auth bypass on the chat proxy so HUX-12 health receipts can observe a real 200, points the evidence policy at it, and makes the delivery flag gate progressive (foundation first, cards enabled per lifecycle acceptance; unknown flags still never ship). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BvMSXH8VH2tMWXanb8SJdf
279 lines
12 KiB
Python
279 lines
12 KiB
Python
"""Flux delivery gates for the per-tenant HUX service boundary.
|
|
|
|
The gates are topology-adaptive: before the activation commit lands, the
|
|
manifests must contain no partial HUX wiring at all; once the ``hux``
|
|
sidecar exists, every boundary assertion below is enforced strictly.
|
|
"""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
SERVICE = ROOT / "services/hermes"
|
|
CHAT = SERVICE / "chat-statefulset.yaml"
|
|
CHAT_PVCS = SERVICE / "chat-pvcs.yaml"
|
|
KUSTOMIZATION = SERVICE / "kustomization.yaml"
|
|
WEBUI_IMAGE = "registry.bstein.dev/bstein/hermes-webui"
|
|
ALL_FLAGS = {
|
|
"hux.activity_timeline",
|
|
"hux.artifacts",
|
|
"hux.autonomy",
|
|
"hux.foundation",
|
|
"hux.friendly_modes",
|
|
"hux.memory_control",
|
|
"hux.multimodal",
|
|
"hux.onboarding",
|
|
"hux.privacy",
|
|
"hux.projects",
|
|
"hux.release_followthrough",
|
|
"hux.research",
|
|
}
|
|
|
|
|
|
def _statefulset() -> dict:
|
|
return yaml.safe_load(CHAT.read_text(encoding="utf-8"))
|
|
|
|
|
|
def _hux_active() -> bool:
|
|
pod = _statefulset()["spec"]["template"]["spec"]
|
|
return any(item["name"] == "hux" for item in pod["containers"])
|
|
|
|
|
|
def _require_activation() -> None:
|
|
if not _hux_active():
|
|
pytest.skip("HUX activation topology is not staged in this tree")
|
|
|
|
|
|
def test_hux_topology_is_all_or_nothing() -> None:
|
|
"""A partial HUX rollout (some wiring without the sidecar) never ships."""
|
|
stateful = _statefulset()
|
|
pod = stateful["spec"]["template"]["spec"]
|
|
if _hux_active():
|
|
return
|
|
assert not any(item["name"].startswith("hux") for item in pod["containers"])
|
|
assert not any(
|
|
item["name"] == "init-hux-runtime" for item in pod.get("initContainers", [])
|
|
)
|
|
assert not any(item["name"].startswith("hux-") for item in pod["volumes"])
|
|
claims = {
|
|
item["metadata"]["name"]
|
|
for item in yaml.safe_load_all(CHAT_PVCS.read_text(encoding="utf-8"))
|
|
}
|
|
assert "hermes-chat-hux-data" not in claims
|
|
for container in pod["containers"]:
|
|
assert not any(
|
|
entry["name"].startswith("HUX_") for entry in container.get("env", [])
|
|
)
|
|
|
|
|
|
def _named(items: list[dict], name: str) -> dict:
|
|
return next(item for item in items if item["name"] == name)
|
|
|
|
|
|
def _env(container: dict) -> dict[str, str]:
|
|
return {item["name"]: item.get("value", "") for item in container["env"]}
|
|
|
|
|
|
def _mounts(container: dict) -> dict[str, dict]:
|
|
return {item["name"]: item for item in container.get("volumeMounts", [])}
|
|
|
|
|
|
def test_hux_sidecar_is_loopback_only_and_uses_the_reviewed_webui_image() -> None:
|
|
"""The browser BFF and HUX backend ship as one immutable reviewed artifact."""
|
|
_require_activation()
|
|
pod = _statefulset()["spec"]["template"]["spec"]
|
|
webui = _named(pod["containers"], "webui")
|
|
hux = _named(pod["containers"], "hux")
|
|
values = _env(hux)
|
|
|
|
assert webui["image"] == hux["image"]
|
|
assert webui["image"].startswith(WEBUI_IMAGE + ":")
|
|
assert values["HUX_BIND"] == "127.0.0.1"
|
|
assert values["HUX_PORT"] == "8790"
|
|
assert values["HUX_DATA_ROOT"] == "/var/lib/hux/store"
|
|
# Progressive enablement: the canary starts foundation-only and cards are
|
|
# switched on one lifecycle gate at a time; unknown flags never ship.
|
|
flags = {flag for flag in values["HUX_FLAGS"].split(",") if flag}
|
|
assert "hux.foundation" in flags
|
|
assert flags <= ALL_FLAGS
|
|
assert set(values["HUX_SWITCHYARD_ROUTE_CATALOG"].split(",")) == {
|
|
"atlas/manual/codex/luna",
|
|
"atlas/manual/codex/terra",
|
|
"atlas/manual/codex/sol",
|
|
"atlas/manual/claude/haiku",
|
|
"atlas/manual/claude/fable",
|
|
"atlas/manual/claude/sonnet",
|
|
"atlas/manual/claude/opus",
|
|
"atlas/manual/local/qwen-14b",
|
|
}
|
|
assert values["PYTHONDONTWRITEBYTECODE"] == "1"
|
|
assert hux["securityContext"]["readOnlyRootFilesystem"] is True
|
|
assert hux["securityContext"]["capabilities"]["drop"] == ["ALL"]
|
|
for probe in ("readinessProbe", "livenessProbe"):
|
|
command = " ".join(hux[probe]["exec"]["command"])
|
|
assert "127.0.0.1" in command
|
|
assert "8790" in command
|
|
assert "/healthz" in command
|
|
assert "tcpSocket" not in hux[probe]
|
|
assert "8790" not in (SERVICE / "service.yaml").read_text(encoding="utf-8")
|
|
policy = (SERVICE / "networkpolicy.yaml").read_text(encoding="utf-8")
|
|
assert "port: 8790" not in policy
|
|
|
|
|
|
def test_hux_storage_and_keys_are_mounted_by_least_privilege() -> None:
|
|
"""Hermes receives only its key/context views, never the HUX ledger root."""
|
|
_require_activation()
|
|
stateful = _statefulset()
|
|
pod = stateful["spec"]["template"]["spec"]
|
|
hermes = _named(pod["containers"], "hermes")
|
|
webui = _named(pod["containers"], "webui")
|
|
hux = _named(pod["containers"], "hux")
|
|
media = _named(pod["containers"], "telegram-media")
|
|
hermes_mounts = hermes["volumeMounts"]
|
|
webui_mounts = _mounts(webui)
|
|
hux_mounts = _mounts(hux)
|
|
|
|
assert hux_mounts["hux-data"] == {
|
|
"name": "hux-data",
|
|
"mountPath": "/var/lib/hux",
|
|
"subPathExpr": "$(POD_NAME)",
|
|
}
|
|
assert {
|
|
item["subPathExpr"]
|
|
for item in hermes_mounts
|
|
if item["name"] == "hux-data"
|
|
} == {"$(POD_NAME)/binding", "$(POD_NAME)/context"}
|
|
assert webui_mounts["hux-data"]["subPathExpr"] == "$(POD_NAME)/context"
|
|
assert not any(item.get("mountPath") == "/var/lib/hux" for item in hermes["volumeMounts"])
|
|
assert not any(item.get("mountPath") == "/var/lib/hux" for item in webui["volumeMounts"])
|
|
assert not any(item["name"].startswith("hux-") for item in media["volumeMounts"])
|
|
assert not any(item["name"] == "hux-relay-key" for item in hermes_mounts)
|
|
assert "hux-worker-key" not in webui_mounts
|
|
|
|
assert {item["metadata"]["name"] for item in stateful["spec"]["volumeClaimTemplates"]} == {
|
|
"home",
|
|
"workspace",
|
|
}
|
|
claims = {
|
|
item["metadata"]["name"]: item
|
|
for item in yaml.safe_load_all(CHAT_PVCS.read_text(encoding="utf-8"))
|
|
}
|
|
claim = claims["hermes-chat-hux-data"]["spec"]
|
|
assert claim["accessModes"] == ["ReadWriteMany"]
|
|
assert claim["storageClassName"] == "astreae"
|
|
assert claim["resources"]["requests"]["storage"] == "10Gi"
|
|
|
|
|
|
def test_hux_init_preserves_context_identity_and_rotates_transport_keys() -> None:
|
|
"""Context identity is durable while relay/worker credentials are pod-local."""
|
|
_require_activation()
|
|
pod = _statefulset()["spec"]["template"]["spec"]
|
|
init = _named(pod["initContainers"], "init-hux-runtime")
|
|
script = init["args"][0]
|
|
volumes = {item["name"]: item for item in pod["volumes"]}
|
|
|
|
assert 'tenant_root="/hux-data/${HOSTNAME}"' in script
|
|
assert "chown -R" not in script
|
|
assert 'if [ ! -e "${tenant_root}/context/context-key" ]' in script
|
|
assert "bs=32 count=1" in script
|
|
assert 'chmod 0600 "${tenant_root}/context/context-key"' in script
|
|
assert "/hux-relay/relay-key /hux-worker/worker-key" in script
|
|
assert "chmod 0400" in script
|
|
assert 'ordinal="${HOSTNAME##*-}"' in script
|
|
assert 'b"hux.subject.id.v1\\0" + slot.encode("ascii")' in script
|
|
assert 'target = root / "binding/subject"' in script
|
|
assert 'or stat.S_IMODE(info.st_mode) != 0o440' in script
|
|
assert volumes["hux-relay-key"]["emptyDir"]["medium"] == "Memory"
|
|
assert volumes["hux-worker-key"]["emptyDir"]["medium"] == "Memory"
|
|
assert init["securityContext"]["capabilities"] == {
|
|
"drop": ["ALL"],
|
|
"add": ["CHOWN", "DAC_OVERRIDE", "FOWNER"],
|
|
}
|
|
|
|
|
|
def test_hux_identity_and_authentication_inputs_are_file_backed() -> None:
|
|
"""No HUX shared key or subject is placed directly in an environment value."""
|
|
_require_activation()
|
|
pod = _statefulset()["spec"]["template"]["spec"]
|
|
hermes = _env(_named(pod["containers"], "hermes"))
|
|
webui = _env(_named(pod["containers"], "webui"))
|
|
hux = _env(_named(pod["containers"], "hux"))
|
|
|
|
assert hermes["HUX_WORKER_KEY_FILE"] == "/run/hermes-hux-worker/worker-key"
|
|
assert hermes["HUX_SUBJECT_FILE"] == "/run/hermes-hux-subject/subject"
|
|
assert hermes["HUX_CONTEXT_KEY_FILE"] == "/run/hermes-hux-context/context-key"
|
|
assert hermes["HUX_TOOL_ENFORCEMENT"] == "0"
|
|
assert webui["HUX_CONTEXT_KEY_FILE"] == "/run/hermes-hux-context/context-key"
|
|
assert hux["HUX_RELAY_KEY_FILE"] == "/run/hermes-webui-hux/relay-key"
|
|
assert hux["HUX_WORKER_KEY_FILE"] == "/run/hermes-hux-worker/worker-key"
|
|
assert hux["HUX_SUBJECT_BINDING_FILE"] == "/var/lib/hux/binding/subject"
|
|
assert hux["HUX_CONTEXT_KEY_FILE"] == "/var/lib/hux/context/context-key"
|
|
assert hux["HUX_IMAGE_TAG"].startswith("git-")
|
|
assert hux["HUX_IMAGE_TAG"].endswith("-release")
|
|
assert hux["HUX_IMAGE_DIGEST"].startswith("sha256:")
|
|
assert not {"HUX_RELAY_KEY", "HUX_WORKER_KEY", "HUX_ROUTER_KEY"} & set(hux)
|
|
|
|
|
|
def test_hux_evidence_producer_is_isolated_and_fail_closed() -> None:
|
|
"""The HUX-12 evidence key reaches only the service and the producer."""
|
|
_require_activation()
|
|
stateful = _statefulset()
|
|
pod = stateful["spec"]["template"]["spec"]
|
|
producer = _named(pod["containers"], "hux-evidence-producer")
|
|
hux = _named(pod["containers"], "hux")
|
|
hermes = _named(pod["containers"], "hermes")
|
|
webui = _named(pod["containers"], "webui")
|
|
media = _named(pod["containers"], "telegram-media")
|
|
|
|
assert producer["image"] == hux["image"]
|
|
values = _env(producer)
|
|
assert values["HUX_RELEASE_EVIDENCE_KEY_FILE"] == "/run/hermes-hux-evidence/evidence-key"
|
|
assert values["HUX_RELEASE_EVIDENCE_POLICY_FILE"] == "/etc/hux-evidence/policy.json"
|
|
assert values["HUX_PRODUCER_WORKLOAD"] == "hermes-webui"
|
|
assert _env(hux)["HUX_RELEASE_EVIDENCE_KEY_FILE"] == "/run/hermes-hux-evidence/evidence-key"
|
|
assert producer["securityContext"]["readOnlyRootFilesystem"] is True
|
|
assert producer["securityContext"]["capabilities"]["drop"] == ["ALL"]
|
|
for denied in (hermes, webui, media):
|
|
assert not any(
|
|
item["name"] == "hux-evidence-key" for item in denied.get("volumeMounts", [])
|
|
)
|
|
staging = _named(pod["initContainers"], "stage-hux-evidence")
|
|
script = staging["args"][0]
|
|
assert "if [ -s /vault/secrets/hux-evidence-key ]" in script
|
|
assert "chmod 0400" in script
|
|
volumes = {item["name"]: item for item in pod["volumes"]}
|
|
assert volumes["hux-evidence-key"]["emptyDir"]["medium"] == "Memory"
|
|
assert volumes["hux-evidence-policy"]["configMap"]["name"] == "hermes-hux-evidence-policy"
|
|
policy = json.loads((SERVICE / "hux-evidence-policy.json").read_text(encoding="utf-8"))
|
|
assert policy["schema"] == "hux.release_evidence_policy.v1"
|
|
assert set(policy["workloads"]) == {"hermes-webui"}
|
|
rbac = (SERVICE / "hux-evidence-rbac.yaml").read_text(encoding="utf-8")
|
|
assert "verbs: [get, list]" in rbac and "secrets" not in rbac
|
|
rendered = KUSTOMIZATION.read_text(encoding="utf-8")
|
|
assert "hux-evidence-rbac.yaml" in rendered
|
|
assert "policy.json=hux-evidence-policy.json" in rendered
|
|
|
|
|
|
def test_hux_runtime_plugin_renders_its_vendored_hook_package() -> None:
|
|
"""The mounted plugin must contain its Kustomize-local Python package tree."""
|
|
_require_activation()
|
|
stateful = _statefulset()
|
|
pod = stateful["spec"]["template"]["spec"]
|
|
plugin = _named(pod["volumes"], "hux-runtime-plugin")["configMap"]
|
|
items = {item["key"]: item["path"] for item in plugin["items"]}
|
|
expected = {
|
|
"hux-hook-init.py": "hux_hook/__init__.py",
|
|
"hux-hook-client.py": "hux_hook/client.py",
|
|
"hux-hook-hooks.py": "hux_hook/hooks.py",
|
|
}
|
|
|
|
assert expected.items() <= items.items()
|
|
rendered = KUSTOMIZATION.read_text(encoding="utf-8")
|
|
assert "hux-hook-init.py=plugins/hux-runtime/hux_hook/__init__.py" in rendered
|
|
assert "hux-hook-client.py=plugins/hux-runtime/hux_hook/client.py" in rendered
|
|
assert "hux-hook-hooks.py=plugins/hux-runtime/hux_hook/hooks.py" in rendered
|