From 469e52fd20fff61997eedb248143cf7d075c29e1 Mon Sep 17 00:00:00 2001 From: jenkins Date: Mon, 24 Aug 2026 04:35:12 -0300 Subject: [PATCH] hermes(hux): add the HUX-12 release evidence producer A companion package (outside the network-free hux/ service package) that independently verifies and binds the whole release chain before any transition: reviewed proposal URL, Jenkins job/build/result and revision, immutable Harbor tag/digest equality, Flux kustomization and applied revision with pin containment, desired workload image, every Ready pod imageID, bounded-age health receipt, and rollback target. Pure injectable verifier core, HTTPS-only collectors (SA token for the Kubernetes API), and an evidence-trust driver that posts exactly one If-Match transition with deterministic idempotency. Rejects stale, replayed, downgraded, incomplete, cross-workload, mismatched, and self-asserted evidence. 100% line and branch coverage (71 tests). Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01BvMSXH8VH2tMWXanb8SJdf --- .../hux_producer/__init__.py | 478 ++++++++++++++++++ testing/quality_contract.json | 4 + .../tests/test_hermes_hux_release_producer.py | 478 ++++++++++++++++++ 3 files changed, 960 insertions(+) create mode 100644 dockerfiles/hermes-hux-foundation/hux_producer/__init__.py create mode 100644 testing/tests/test_hermes_hux_release_producer.py diff --git a/dockerfiles/hermes-hux-foundation/hux_producer/__init__.py b/dockerfiles/hermes-hux-foundation/hux_producer/__init__.py new file mode 100644 index 00000000..d7c683c3 --- /dev/null +++ b/dockerfiles/hermes-hux-foundation/hux_producer/__init__.py @@ -0,0 +1,478 @@ +"""HUX-12 release follow-through evidence producer (companion process). + +This package lives OUTSIDE the network-free ``hux`` service package (SO-29): +it is the only component allowed to open outbound connections. It may import +``hux.errors`` and ``hux.release_security``; ``hux`` never imports it. + +A pure observation -> verdict core (``evidence_for``) binds every observation +to the release being advanced and the workload policy, returning the exact +evidence payload for the next legal transition or raising ``Invalid``. A thin +``Collector`` gathers observations from Jenkins, Harbor, the Kubernetes API +and the health URL over HTTPS with an injectable transport. The driver +(``advance``/``run_once``) posts at most one verified transition per pass with +If-Match and a deterministic Idempotency-Key; verdicts derive only from passed +observations and the injected clock, and nothing here logs the evidence key. +""" + +from __future__ import annotations + +import json +import re +import ssl +import urllib.error +import urllib.request +from collections.abc import Callable, Mapping +from datetime import datetime, timezone +from pathlib import Path +from typing import Any +from urllib.parse import quote, urlsplit + +from hux import release_security +from hux.errors import Invalid + +ORDER = ("reviewed", "merged", "built", "verified", "deployed", "converged", "live_verified") +TERMINAL = frozenset({"live_verified", "rolled_back"}) +ROLLBACK_STATES = frozenset({"built", "verified", "deployed", "converged", "live_verified"}) +COMMIT = re.compile(r"^[0-9a-f]{40}$") +SHA = re.compile(r"^sha256:[0-9a-f]{64}$") +SHA_ANY = re.compile(r"sha256:[0-9a-f]{64}") +TIME = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$") +FLUX_REV = re.compile(r"^main@sha1:[0-9a-f]{40}$") +NAMESPACE = re.compile(r"^[a-z0-9-]{1,63}$") +DEFAULT_BASE_URL = "http://127.0.0.1:8790" +DEFAULT_KUBE_API = "https://kubernetes.default.svc" +DEFAULT_KUBE_TOKEN = "/var/run/secrets/kubernetes.io/serviceaccount/token" # noqa: S105 +MAX_RESPONSE_BYTES = 1024 * 1024 +Opener = Callable[[urllib.request.Request, float, ssl.SSLContext | None], Any] + +def _check(condition: Any, message: str) -> None: + """Fail closed with a precise reason unless ``condition`` holds.""" + if not condition: + raise Invalid(message) + +def parse_time(value: Any, label: str) -> datetime: + """Parse one strict RFC 3339 UTC second-precision timestamp.""" + _check(isinstance(value, str) and TIME.fullmatch(value), f"{label} must be an RFC 3339 UTC timestamp") + return datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc) + +def _fresh(value: Any, now: datetime, max_age: int, label: str) -> str: + age = (now - parse_time(value, label)).total_seconds() + _check(0 <= age <= max_age, f"stale evidence: {label} is outside the freshness window") + return str(value) + +def _observation(observations: Mapping[str, Any], name: str, workload: str, now: datetime, max_age: int) -> dict[str, Any]: + obs = observations.get(name) + _check(isinstance(obs, dict), f"incomplete observations: {name} is missing") + _check(obs.get("workload") == workload, f"{name} observation is bound to a different workload") + _fresh(obs.get("observed_at"), now, max_age, f"{name}.observed_at") + return obs + +def _need(obs: Mapping[str, Any], name: str, field: str, kind: type = str) -> Any: + value = obs.get(field) + _check(not isinstance(value, bool) and isinstance(value, kind), f"incomplete observations: {name}.{field} is missing or malformed") + return value + +def _verify_merged(release, observations, wpolicy, max_age, now): + git = _observation(observations, "git", release["workload"], now, max_age) + commit = _need(git, "git", "commit") + _check(COMMIT.fullmatch(commit), "git commit must be an exact 40-hex SHA") + _check(parse_time(_need(git, "git", "committed_at"), "git.committed_at") <= now, "git commit timestamp is in the future") + review_url = _need(git, "git", "review_url") + _check(review_url == release.get("evidence", {}).get("review_url"), "git observation is for a different review than this release") + _check(review_url.startswith(wpolicy["review_url_prefix"]), "review URL is outside the policy review prefix") + return {"merge_commit": commit} + +def _verify_built(release, observations, wpolicy, max_age, now): + jenkins = _observation(observations, "jenkins", release["workload"], now, max_age) + harbor = _observation(observations, "harbor", release["workload"], now, max_age) + sha = str(release.get("evidence", {}).get("merge_commit", "")) + build_url, job = _need(jenkins, "jenkins", "build_url"), wpolicy["jenkins_job_url"] + _check(build_url == job or build_url.startswith(job + "/"), "Jenkins build URL is outside the policy job") + _check(_need(jenkins, "jenkins", "result") == "SUCCESS", "Jenkins build result is not SUCCESS") + _check(COMMIT.fullmatch(sha) and _need(jenkins, "jenkins", "revision") == sha, "Jenkins build revision does not equal the merged commit") + number = _need(jenkins, "jenkins", "build_number", int) + _check(number > 0, "Jenkins build number is invalid") + tag = f"git-{sha}-build-{number}-release" + _check(_need(harbor, "harbor", "tag") == tag, "Harbor tag does not bind the merged commit and Jenkins build") + _check(_need(harbor, "harbor", "repository") == wpolicy["image_repository"], "Harbor repository does not match the policy image repository") + digest = _need(harbor, "harbor", "digest") + _check(SHA.fullmatch(digest), "Harbor digest is malformed") + _check(_need(jenkins, "jenkins", "image_digest") == digest, "Harbor digest does not equal the advertised image digest") + ref = f"{wpolicy['image_repository']}:{tag}@{digest}" + return {"ci_build_url": build_url, "image_ref": ref, "image_digest": digest, "harbor_digest": digest} + +def _verify_verified(release, observations, wpolicy, max_age, now): + harbor = _observation(observations, "harbor", release["workload"], now, max_age) + evidence = release.get("evidence", {}) + tag = str(evidence.get("image_ref", "")).rpartition("@")[0].rpartition(":")[2] + same = (_need(harbor, "harbor", "repository") == wpolicy["image_repository"] + and tag and _need(harbor, "harbor", "tag") == tag + and _need(harbor, "harbor", "digest") == evidence.get("image_digest")) + _check(same, "Harbor no longer serves the built tag at the recorded digest") + return {} + +def _verify_deployed(release, observations, wpolicy, max_age, now): + flux = _observation(observations, "flux", release["workload"], now, max_age) + _check(_need(flux, "flux", "name") == wpolicy["flux_kustomization"], "Flux kustomization name does not match policy") + revision = _need(flux, "flux", "applied_revision") + _check(FLUX_REV.fullmatch(revision), "Flux applied revision must look like main@sha1:<40-hex>") + _check(flux.get("pin_in_revision") is True, "manifest pin is not proven to be contained in the applied revision") + return {"flux_revision": revision} + +def _verify_converged(release, observations, wpolicy, max_age, now): + spec = _observation(observations, "workload_spec", release["workload"], now, max_age) + pods = _observation(observations, "pods", release["workload"], now, max_age) + evidence = release.get("evidence", {}) + ref, digest = str(evidence.get("image_ref", "")), str(evidence.get("image_digest", "")) + _check(ref and _need(spec, "workload_spec", "image") == ref, "desired workload image does not equal the built release reference") + entries = _need(pods, "pods", "pods", list) + _check(entries, "no Ready pods were observed for the workload") + for pod in entries: + _check(isinstance(pod, dict) and pod.get("ready") is True, "a workload pod is not Ready") + image_id = pod.get("image_id") + _check(isinstance(image_id, str) and image_id.endswith("@" + digest), "a Ready pod is not running the release digest") + return {"pod_digest": digest} + +def _verify_live(release, observations, wpolicy, max_age, now): + health = _observation(observations, "health", release["workload"], now, max_age) + _check(_need(health, "health", "url") == wpolicy["health_url"], "health receipt is not for the policy health URL") + status = _need(health, "health", "status", int) + _check(status == 200, f"health check returned HTTP {status}, not 200") + checked = _fresh(health.get("checked_at"), now, max_age, "health.checked_at") + return {"health_check": {"url": health["url"], "status": "pass", "at": checked}} + +def _verify_rollback(release, observations, wpolicy, max_age, now): + rollback = _observation(observations, "rollback", release["workload"], now, max_age) + digest = _need(rollback, "rollback", "digest") + _check(SHA.fullmatch(digest), "rollback target digest is malformed") + return {"rollback_target": digest} + +_VERIFIERS = { + "merged": _verify_merged, "built": _verify_built, "verified": _verify_verified, "deployed": _verify_deployed, + "converged": _verify_converged, "live_verified": _verify_live, "rolled_back": _verify_rollback, +} + +def next_target(state: str) -> str | None: + """The only legal forward transition from ``state``; None when terminal.""" + if state in TERMINAL: + return None + _check(state in ORDER, "release is in an unknown state") + return ORDER[ORDER.index(state) + 1] + +def evidence_for(release: Mapping[str, Any], target: str, observations: Mapping[str, Any], wpolicy: Mapping[str, Any], max_age: int, now: datetime) -> dict[str, Any]: + """Verify observations for exactly one legal transition or raise Invalid.""" + _check(now.tzinfo is not None, "now must be a timezone-aware UTC datetime") + state = str(release.get("state", "")) + if target == "rolled_back": + _check(state in ROLLBACK_STATES, "there is no deployed image to roll back") + else: + _check(target == next_target(state), f"transition to {target} would skip, repeat or downgrade from {state}") + return _VERIFIERS[target](dict(release), observations, wpolicy, int(max_age), now) + +def _default_opener(request: urllib.request.Request, timeout: float, context: ssl.SSLContext | None): + return urllib.request.urlopen(request, timeout=timeout, context=context) # noqa: S310 + +def _context(ca_file: str) -> ssl.SSLContext: + """A verifying TLS context, trusting the given CA bundle when one is mounted.""" + try: + return ssl.create_default_context(cafile=ca_file or None) + except (OSError, ssl.SSLError) as error: + raise Invalid("collector CA bundle is unavailable") from error + +def _bounded_timeout(environ: Mapping[str, str]) -> float: + try: + value = float(environ.get("HUX_PRODUCER_TIMEOUT_SECONDS", 10.0)) + except (TypeError, ValueError): + return 10.0 + return value if 0.1 <= value <= 60.0 else 10.0 + +class Collector: + """Observation gatherer; every field is observed via the transport, never asserted.""" + + def __init__(self, environ: Mapping[str, str], workload: str, wpolicy: Mapping[str, Any], opener: Opener | None = None, clock: Callable[[], datetime] | None = None) -> None: + self.environ = dict(environ) + self.workload = workload + self.wpolicy = dict(wpolicy) + self.opener = opener or _default_opener + self.clock = clock or (lambda: datetime.now(timezone.utc)) + self.timeout = _bounded_timeout(environ) + + def _stamp(self, fields: dict[str, Any]) -> dict[str, Any]: + observed = self.clock().strftime("%Y-%m-%dT%H:%M:%SZ") + return {"workload": self.workload, "observed_at": observed, **fields} + + def _fetch(self, url: str, label: str, headers: Mapping[str, str] | None = None, context: ssl.SSLContext | None = None) -> tuple[int, bytes]: + _check(urlsplit(url).scheme == "https", f"{label} observation URL must be HTTPS") + request = urllib.request.Request(url, headers=dict(headers or {}), method="GET") # noqa: S310 + try: + with self.opener(request, self.timeout, context or _context("")) as response: + status, data = int(response.status), response.read(MAX_RESPONSE_BYTES + 1) + except urllib.error.HTTPError as error: + return int(error.code), b"" + except (OSError, ValueError) as error: + raise Invalid(f"{label} observation fetch failed") from error + _check(len(data) <= MAX_RESPONSE_BYTES, f"{label} observation response is too large") + return status, data + + def _json(self, url: str, label: str, headers: Mapping[str, str] | None = None, context: ssl.SSLContext | None = None) -> dict[str, Any]: + status, data = self._fetch(url, label, headers, context) + _check(status == 200, f"{label} observation returned HTTP {status}") + try: + parsed = json.loads(data) + except ValueError as error: + raise Invalid(f"{label} observation is not JSON") from error + _check(isinstance(parsed, dict), f"{label} observation is not a JSON object") + return parsed + + def git(self, review_url: str) -> dict[str, Any]: + """Observe the merged proposal named by the release's recorded review URL.""" + prefix = self.wpolicy["review_url_prefix"] + number = review_url[len(prefix):] if review_url.startswith(prefix) else "" + _check(number.isdigit(), "release review URL is outside the policy review prefix") + parsed = urlsplit(prefix) + parts = parsed.path.strip("/").split("/") + _check(len(parts) == 3 and parts[2] == "pulls", "policy review prefix is not a proposal URL") + body = self._json(f"https://{parsed.netloc}/api/v1/repos/{parts[0]}/{parts[1]}/pulls/{number}", "git") + fields: dict[str, Any] = {"review_url": review_url} + commit, merged_at = body.get("merge_commit_sha"), body.get("merged_at") + if body.get("merged") is True and isinstance(commit, str): + fields["commit"] = commit.lower() + if isinstance(merged_at, str) and TIME.fullmatch(merged_at): + fields["committed_at"] = merged_at + return self._stamp(fields) + + def jenkins(self) -> dict[str, Any]: + """Observe the last successful build of the policy job.""" + body = self._json(self.wpolicy["jenkins_job_url"] + "/lastSuccessfulBuild/api/json", "jenkins") + fields: dict[str, Any] = {} + url, result, number = body.get("url"), body.get("result"), body.get("number") + if isinstance(url, str): + fields["build_url"] = url.rstrip("/") + if isinstance(result, str): + fields["result"] = result + if isinstance(number, int) and not isinstance(number, bool): + fields["build_number"] = number + for action in body.get("actions") or []: + revision = action.get("lastBuiltRevision") if isinstance(action, dict) else None + sha = revision.get("SHA1") if isinstance(revision, dict) else None + if isinstance(sha, str) and COMMIT.fullmatch(sha.lower()): + fields["revision"] = sha.lower() + break + advertised = SHA_ANY.search(str(body.get("description") or "")) + if advertised: + fields["image_digest"] = advertised.group(0) + return self._stamp(fields) + + def harbor(self, tag: str) -> dict[str, Any]: + """Observe the immutable Harbor artifact behind one release tag.""" + repository = self.wpolicy["image_repository"] + host, _, path = repository.partition("/") + project, _, name = path.partition("/") + _check(host and project and name, "policy image repository is not host/project/name") + body = self._json(f"https://{host}/api/v2.0/projects/{project}/repositories/{quote(name, safe='')}/artifacts/{quote(tag, safe='')}", "harbor") + fields: dict[str, Any] = {"repository": repository} + digest = body.get("digest") + if isinstance(digest, str) and SHA.fullmatch(digest): + fields["digest"] = digest + names = [entry.get("name") for entry in body.get("tags") or [] if isinstance(entry, dict)] + if tag in names: + fields["tag"] = tag + return self._stamp(fields) + + def _namespace(self) -> str: + namespace = self.environ.get("HUX_PRODUCER_NAMESPACE", "") + _check(NAMESPACE.fullmatch(namespace), "HUX_PRODUCER_NAMESPACE is missing or malformed") + return namespace + + def _kube(self, path: str) -> dict[str, Any]: + base = self.environ.get("HUX_PRODUCER_KUBE_API", DEFAULT_KUBE_API).rstrip("/") + token_file = self.environ.get("HUX_PRODUCER_KUBE_TOKEN_FILE", DEFAULT_KUBE_TOKEN) + try: + token = Path(token_file).read_text(encoding="utf-8").strip() + except (OSError, UnicodeDecodeError) as error: + raise Invalid("kubernetes credentials are unavailable") from error + _check(token, "kubernetes credentials are unavailable") + context = _context(self.environ.get("HUX_PRODUCER_KUBE_CA_FILE", "")) + return self._json(base + path, "kubernetes", {"Authorization": "Bearer " + token}, context) + + def _workload_image(self) -> str | None: + kind = self.environ.get("HUX_PRODUCER_WORKLOAD_KIND", "statefulset").lower() + plural = {"deployment": "deployments", "statefulset": "statefulsets"}.get(kind) + _check(plural, "HUX_PRODUCER_WORKLOAD_KIND must be deployment or statefulset") + name = self.environ.get("HUX_PRODUCER_WORKLOAD_NAME", self.workload) + body = self._kube(f"/apis/apps/v1/namespaces/{self._namespace()}/{plural}/{name}") + containers = (((body.get("spec") or {}).get("template") or {}).get("spec") or {}).get("containers") or [] + prefix = self.wpolicy["image_repository"] + ":" + for container in containers: + image = container.get("image") if isinstance(container, dict) else None + if isinstance(image, str) and image.startswith(prefix): + return image + return None + + def workload_spec(self) -> dict[str, Any]: + """Observe the live desired image of the policy workload.""" + image = self._workload_image() + return self._stamp({"image": image} if image is not None else {}) + + def flux(self, expected_image: str) -> dict[str, Any]: + """Observe the policy Kustomization; pin containment is proven, not asserted.""" + name = self.wpolicy["flux_kustomization"] + body = self._kube(f"/apis/kustomize.toolkit.fluxcd.io/v1/namespaces/{self._namespace()}/kustomizations/{name}") + status = body.get("status") or {} + applied = status.get("lastAppliedRevision") + fields: dict[str, Any] = {"name": (body.get("metadata") or {}).get("name")} + if isinstance(applied, str): + fields["applied_revision"] = applied + ready = any(isinstance(c, dict) and c.get("type") == "Ready" and c.get("status") == "True" for c in status.get("conditions") or []) + settled = isinstance(applied, str) and status.get("lastAttemptedRevision") in (None, applied) + pinned = bool(expected_image) and self._workload_image() == expected_image + fields["pin_in_revision"] = bool(ready and settled and pinned) + return self._stamp(fields) + + def pods(self) -> dict[str, Any]: + """Observe readiness and running imageID for every selected pod.""" + selector = self.environ.get("HUX_PRODUCER_POD_SELECTOR", "") + _check(selector, "HUX_PRODUCER_POD_SELECTOR is required to observe pods") + body = self._kube(f"/api/v1/namespaces/{self._namespace()}/pods?labelSelector={quote(selector, safe='=,')}") + prefix = self.wpolicy["image_repository"] + ":" + entries = [] + for item in body.get("items") or []: + if not isinstance(item, dict): + continue + status = item.get("status") or {} + ready = any(isinstance(c, dict) and c.get("type") == "Ready" and c.get("status") == "True" for c in status.get("conditions") or []) + image_id = None + for container in status.get("containerStatuses") or []: + if isinstance(container, dict) and str(container.get("image", "")).startswith(prefix): + image_id = container.get("imageID") + break + entries.append({"name": (item.get("metadata") or {}).get("name"), "ready": ready, "image_id": image_id}) + return self._stamp({"pods": entries}) + + def health(self) -> dict[str, Any]: + """Observe the policy health URL; the receipt carries the real HTTP status.""" + url = self.wpolicy["health_url"] + checked = self.clock().strftime("%Y-%m-%dT%H:%M:%SZ") + status, _ = self._fetch(url, "health") + return self._stamp({"url": url, "status": status, "checked_at": checked}) + + def rollback_target(self) -> dict[str, Any]: + """Observe the digest currently pinned in the live workload spec.""" + image = self._workload_image() + fields: dict[str, Any] = {} + digest = image.rpartition("@")[2] if isinstance(image, str) else "" + if SHA.fullmatch(digest): + fields["digest"] = digest + return self._stamp(fields) + +def collect_for(collector: Collector, target: str, release: Mapping[str, Any]) -> dict[str, Any]: + """Gather exactly the observations the verifier needs for one target state.""" + evidence = release.get("evidence", {}) + if target == "merged": + return {"git": collector.git(str(evidence.get("review_url", "")))} + if target == "built": + jenkins = collector.jenkins() + observations = {"jenkins": jenkins} + sha, number = jenkins.get("revision"), jenkins.get("build_number") + if sha is not None and number is not None: + observations["harbor"] = collector.harbor(f"git-{sha}-build-{number}-release") + return observations + if target == "verified": + tag = str(evidence.get("image_ref", "")).rpartition("@")[0].rpartition(":")[2] + return {"harbor": collector.harbor(tag)} + if target == "deployed": + return {"flux": collector.flux(str(evidence.get("image_ref", "")))} + if target == "converged": + return {"workload_spec": collector.workload_spec(), "pods": collector.pods()} + if target == "live_verified": + return {"health": collector.health()} + _check(target == "rolled_back", "no collector plan exists for the target state") + return {"rollback": collector.rollback_target()} + +class ProducerClient: + """Loopback HUX API client for the evidence identity; the key never leaves it.""" + + def __init__(self, base_url: str, slot: str, subject: str, key: str, opener: Opener | None = None, timeout: float = 10.0) -> None: + base = str(base_url or DEFAULT_BASE_URL).rstrip("/") + parts = urlsplit(base) + loopback = parts.scheme == "http" and parts.hostname in {"127.0.0.1", "::1", "localhost"} + _check(parts.scheme == "https" or loopback, "HUX base URL must be loopback HTTP or HTTPS") + self.base = base + self.opener = opener or _default_opener + self.timeout = timeout + self._headers = { + "X-Hermes-Tenant-Identity": slot, "X-Hux-Subject": subject, "X-Hux-Surface": "api", + "X-Hux-Trust": "evidence", "X-Hux-Relay-Key": key, "Content-Type": "application/json", + } + + def request(self, method: str, path: str, body: Any = None, headers: Mapping[str, str] | None = None) -> tuple[int, Any, bool]: + data = None if body is None else json.dumps(body).encode() + request = urllib.request.Request( # noqa: S310 + self.base + path, data=data, headers={**self._headers, **(headers or {})}, method=method) + try: + response = self.opener(request, self.timeout, None) + except urllib.error.HTTPError as error: + response = error + except OSError as error: + raise Invalid("HUX service is unreachable") from error + with response: + status = int(response.status) + raw = response.read(MAX_RESPONSE_BYTES) + replayed = str(response.headers.get("HUX-Replayed") or "") == "true" + try: + parsed = json.loads(raw) if raw else None + except ValueError: + parsed = None + return status, parsed, replayed + +def advance(client: ProducerClient, collector: Collector, wpolicy: Mapping[str, Any], max_age: int, scope_path: str, release_id: str, now: datetime) -> dict[str, Any]: + """Read the head, verify observations, and post at most one transition.""" + status, view, _ = client.request("GET", f"{scope_path}/{release_id}") + _check(status == 200 and isinstance(view, dict), f"release read failed with HTTP {status}") + release, revision = view.get("release") or {}, view.get("revision") + _check(not isinstance(revision, bool) and isinstance(revision, int), "release view carries no revision") + target = next_target(str(release.get("state", ""))) + if target is None: + return {"changed": False, "release_id": release_id, "state": release.get("state"), "reason": "release is terminal"} + observations = collect_for(collector, target, release) + evidence = evidence_for(release, target, observations, wpolicy, max_age, now) + status, view, replayed = client.request( + "POST", f"{scope_path}/{release_id}/transitions", {"to": target, "evidence": evidence}, + {"If-Match": str(revision), "Idempotency-Key": f"producer-{release_id}-{target}-{revision}"[:120]}) + _check(status == 200 and isinstance(view, dict), f"release transition was rejected with HTTP {status}") + head = view.get("release") or {} + return {"changed": not replayed, "release_id": release_id, "state": head.get("state"), + "revision": view.get("revision"), "replayed": replayed} + +def _require(environ: Mapping[str, str], name: str) -> str: + value = str(environ.get(name, "")).strip() + _check(value, f"{name} is required") + return value + +def _pick_release(client: ProducerClient, scope_path: str, workload: str) -> str | None: + status, listing, _ = client.request("GET", scope_path) + _check(status == 200 and isinstance(listing, dict), f"release list failed with HTTP {status}") + for item in listing.get("items") or []: + release = (item.get("release") or {}) if isinstance(item, dict) else {} + if release.get("workload") == workload and release.get("state") not in TERMINAL: + return str(release.get("id")) + return None + +def run_once(environ: Mapping[str, str], opener: Opener | None = None, clock: Callable[[], datetime] | None = None) -> dict[str, Any]: + """One idempotent producer pass wired entirely from the environment.""" + environ = dict(environ) + workload = _require(environ, "HUX_PRODUCER_WORKLOAD") + policy = release_security.load_policy(environ) + wpolicy = policy["workloads"].get(workload) + _check(wpolicy is not None, "workload is not enabled for release evidence") + key = release_security.evidence_key(environ) + client = ProducerClient(environ.get("HUX_BASE_URL", DEFAULT_BASE_URL), _require(environ, "HUX_TENANT_SLOT"), + _require(environ, "HUX_PRODUCER_SUBJECT"), key, opener, _bounded_timeout(environ)) + project = _require(environ, "HUX_PRODUCER_PROJECT_ID") + conversation = _require(environ, "HUX_PRODUCER_CONVERSATION_ID") + scope_path = f"/hux/v1/projects/{project}/conversations/{conversation}/releases" + collector = Collector(environ, workload, wpolicy, opener, clock) + release_id = str(environ.get("HUX_PRODUCER_RELEASE_ID", "")).strip() or _pick_release(client, scope_path, workload) + if release_id is None: + return {"changed": False, "reason": "no advanceable release"} + return advance(client, collector, wpolicy, policy["max_evidence_age_seconds"], scope_path, release_id, collector.clock()) diff --git a/testing/quality_contract.json b/testing/quality_contract.json index dd21ec2e..d4ca1ec4 100644 --- a/testing/quality_contract.json +++ b/testing/quality_contract.json @@ -131,6 +131,7 @@ "dockerfiles/hermes-hux-foundation/hux/server.py", "dockerfiles/hermes-hux-foundation/hux/store.py", "dockerfiles/hermes-hux-foundation/hux/suggestions.py", + "dockerfiles/hermes-hux-foundation/hux_producer/__init__.py", "dockerfiles/hermes-worker-hux/hux_hook/__init__.py", "dockerfiles/hermes-worker-hux/hux_hook/client.py", "dockerfiles/hermes-worker-hux/hux_hook/hooks.py", @@ -241,6 +242,7 @@ "dockerfiles/hermes-hux-foundation/hux/server.py", "dockerfiles/hermes-hux-foundation/hux/store.py", "dockerfiles/hermes-hux-foundation/hux/suggestions.py", + "dockerfiles/hermes-hux-foundation/hux_producer/__init__.py", "dockerfiles/hermes-worker-hux/hux_hook/__init__.py", "dockerfiles/hermes-worker-hux/hux_hook/client.py", "dockerfiles/hermes-worker-hux/hux_hook/hooks.py", @@ -442,6 +444,7 @@ "dockerfiles/hermes-hux-foundation/hux/server.py", "dockerfiles/hermes-hux-foundation/hux/store.py", "dockerfiles/hermes-hux-foundation/hux/suggestions.py", + "dockerfiles/hermes-hux-foundation/hux_producer/__init__.py", "dockerfiles/hermes-worker-hux/hux_hook/__init__.py", "dockerfiles/hermes-worker-hux/hux_hook/client.py", "dockerfiles/hermes-worker-hux/hux_hook/hooks.py", @@ -562,6 +565,7 @@ "dockerfiles/hermes-hux-foundation/hux/server.py", "dockerfiles/hermes-hux-foundation/hux/store.py", "dockerfiles/hermes-hux-foundation/hux/suggestions.py", + "dockerfiles/hermes-hux-foundation/hux_producer/__init__.py", "dockerfiles/hermes-worker-hux/hux_hook/__init__.py", "dockerfiles/hermes-worker-hux/hux_hook/client.py", "dockerfiles/hermes-worker-hux/hux_hook/hooks.py", diff --git a/testing/tests/test_hermes_hux_release_producer.py b/testing/tests/test_hermes_hux_release_producer.py new file mode 100644 index 00000000..97fccfe2 --- /dev/null +++ b/testing/tests/test_hermes_hux_release_producer.py @@ -0,0 +1,478 @@ +"""HUX-12 release evidence producer: verifier core, collectors and driver.""" + +from __future__ import annotations + +import io +import json +import sys +import threading +import urllib.error +import urllib.request +from datetime import datetime, timedelta, timezone +from email import message_from_string +from pathlib import Path +from types import SimpleNamespace + +import pytest + +ROOT = Path(__file__).resolve().parents[2] +FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation" +if str(FOUNDATION) not in sys.path: + sys.path.insert(0, str(FOUNDATION)) + +import hux_producer as release_producer # noqa: E402 +from hux import contracts # noqa: E402 +from hux.errors import Invalid # noqa: E402 +from hux.http import serve # noqa: E402 +from hux.server import build_router # noqa: E402 + +ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"]) +KEY = "evidence-key-value-that-is-at-least-32-bytes" +SUBJECT = "usr_0123456789abcdef" +SHA_M = "a" * 40 +DIGEST = "sha256:" + "d" * 64 +OTHER_DIGEST = "sha256:" + "e" * 64 +REPO = "registry.bstein.dev/bstein/hermes-webui" +JOB = "https://jenkins.bstein.dev/job/hermes-webui-image" +PREFIX = "https://scm.bstein.dev/atlas/titan-iac/pulls/" +REVIEW = PREFIX + "55" +HEALTH = "https://chat.bstein.dev/healthz" +TAG = f"git-{SHA_M}-build-20-release" +REF = f"{REPO}:{TAG}@{DIGEST}" +FLUX_REV = "main@sha1:" + "b" * 40 +WP = {"review_url_prefix": PREFIX, "jenkins_job_url": JOB, "image_repository": REPO, "flux_kustomization": "hermes", "health_url": HEALTH} +POLICY = {"schema": "hux.release_evidence_policy.v1", "max_evidence_age_seconds": 900, "workloads": {"hermes-webui": WP}} +FIXED = datetime(2026, 8, 24, 12, 0, 0, tzinfo=timezone.utc) +NOW_S = "2026-08-24T12:00:00Z" +OLD_S = "2026-08-24T10:00:00Z" +GIT_API = "https://scm.bstein.dev/api/v1/repos/atlas/titan-iac/pulls/55" +JENKINS_API = JOB + "/lastSuccessfulBuild/api/json" +HARBOR_API = f"https://registry.bstein.dev/api/v2.0/projects/bstein/repositories/hermes-webui/artifacts/{TAG}" +KUSTOMIZATION_API = "https://kube.test/apis/kustomize.toolkit.fluxcd.io/v1/namespaces/hermes/kustomizations/hermes" +STS_API = "https://kube.test/apis/apps/v1/namespaces/hermes/statefulsets/hermes-webui" +PODS_API = "https://kube.test/api/v1/namespaces/hermes/pods?labelSelector=app=hermes-webui" +ROUTER_HEADERS = {"X-Hermes-Tenant-Identity": "slot-3", "X-Hux-Subject": SUBJECT, "X-Hux-Surface": "chat", "X-Hux-Relay-Key": "rk"} +CA_PEM = """-----BEGIN CERTIFICATE----- +MIIBgTCCASegAwIBAgIUNjZXQ5S3lIfyo4QEuwB6CmxthPwwCgYIKoZIzj0EAwIw +FjEUMBIGA1UEAwwLaHV4LXRlc3QtY2EwHhcNMjYwODI0MDcyMTQzWhcNMzYwODIx +MDcyMTQzWjAWMRQwEgYDVQQDDAtodXgtdGVzdC1jYTBZMBMGByqGSM49AgEGCCqG +SM49AwEHA0IABKUXVC/hmAAhDxIT8HrMvXH/N67OB6zEFKnaW24bg907WXzKqQO6 +eSJzPPznQvzuivQtIG6D+39amDg8utWGERijUzBRMB0GA1UdDgQWBBQcnGVz8rJ+ +MJrVZ3geD32daLpJBzAfBgNVHSMEGDAWgBQcnGVz8rJ+MJrVZ3geD32daLpJBzAP +BgNVHRMBAf8EBTADAQH/MAoGCCqGSM49BAMCA0gAMEUCIQDVLn3BP3aUfBVdmzck +xydrFRXApxVn0NpQ5v7DBZvZXwIgKibREmX5TKcB9YZyrHaddGTjfPSfREOhxaQQ +nMKqK1Q= +-----END CERTIFICATE----- +""" + +def obs(**fields): + return {"workload": "hermes-webui", "observed_at": NOW_S, **fields} + +def rel(state, **evidence): + return {"workload": "hermes-webui", "state": state, "evidence": evidence} + +GIT_OBS = obs(commit=SHA_M, committed_at="2026-08-24T11:00:00Z", review_url=REVIEW) +JENKINS_OBS = obs(build_url=JOB + "/20", result="SUCCESS", build_number=20, revision=SHA_M, image_digest=DIGEST) +HARBOR_OBS = obs(repository=REPO, tag=TAG, digest=DIGEST) +FLUX_OBS = obs(name="hermes", applied_revision=FLUX_REV, pin_in_revision=True) +SPEC_OBS = obs(image=REF) +PODS_OBS = obs(pods=[{"name": "p0", "ready": True, "image_id": REPO + "@" + DIGEST}]) +HEALTH_OBS = obs(url=HEALTH, status=200, checked_at=NOW_S) +BUILT_EV = {"review_url": REVIEW, "merge_commit": SHA_M, "ci_build_url": JOB + "/20", "image_ref": REF, "image_digest": DIGEST, "harbor_digest": DIGEST} + +class FakeResponse: + def __init__(self, status, payload, headers=None): + self.status = status + self._body = payload if isinstance(payload, bytes) else json.dumps(payload).encode() + self.headers = headers or {} + + def read(self, limit=None): + return self._body + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + +class FakeOpener: + def __init__(self, routes): + self.routes = dict(routes) + self.calls = [] + + def __call__(self, request, timeout, context): + url = request.full_url + self.calls.append((request.get_method(), url)) + found = self.routes.get(url) + if found is None: + raise urllib.error.URLError(f"no fake route: {url}") + if isinstance(found, Exception): + raise found + status, payload = found + if status >= 400: + raise urllib.error.HTTPError(url, status, "error", message_from_string(""), io.BytesIO(b"{}")) + return FakeResponse(status, payload) + +def good_routes(): + return { + GIT_API: (200, {"merged": True, "merge_commit_sha": SHA_M.upper(), "merged_at": "2026-08-24T11:00:00Z"}), + JENKINS_API: (200, {"url": JOB + "/20/", "result": "SUCCESS", "number": 20, "description": f"pushed {DIGEST}", + "actions": [None, {"other": 1}, {"lastBuiltRevision": {"SHA1": SHA_M.upper()}}]}), + HARBOR_API: (200, {"digest": DIGEST, "tags": ["junk", {"name": TAG}]}), + KUSTOMIZATION_API: (200, {"metadata": {"name": "hermes"}, "status": { + "lastAppliedRevision": FLUX_REV, "lastAttemptedRevision": FLUX_REV, "conditions": [{"type": "Ready", "status": "True"}]}}), + STS_API: (200, {"spec": {"template": {"spec": {"containers": ["junk", {"image": "other:1"}, {"image": REF}]}}}}), + PODS_API: (200, {"items": ["junk", {"metadata": {"name": "p0"}, "status": { + "conditions": [{"type": "Ready", "status": "True"}], + "containerStatuses": [{"image": "other:1", "imageID": "x"}, {"image": REF, "imageID": REPO + "@" + DIGEST}]}}]}), + HEALTH: (200, b""), + } + +def hybrid(routes): + fake = FakeOpener(routes) + + def opener(request, timeout, context): + if request.full_url.startswith("http://127.0.0.1"): + return urllib.request.urlopen(request, timeout=timeout) + return fake(request, timeout, context) + return opener + +def secrets(tmp_path): + key = tmp_path / "evidence.key" + key.write_text(KEY + "\n") + key.chmod(0o400) + policy = tmp_path / "policy.json" + policy.write_text(json.dumps(POLICY)) + policy.chmod(0o444) + return {"HUX_RELEASE_EVIDENCE_KEY_FILE": str(key), "HUX_RELEASE_EVIDENCE_POLICY_FILE": str(policy)} + +def dispatch(router, method, path, body=None, headers=None): + raw = b"" if body is None else json.dumps(body).encode() + response = router.dispatch(method, path, {**ROUTER_HEADERS, **(headers or {})}, raw) + return response.status, response.body + +def make_collector(routes, extra=None): + env = {"HUX_PRODUCER_NAMESPACE": "hermes", "HUX_PRODUCER_KUBE_API": "https://kube.test", + "HUX_PRODUCER_POD_SELECTOR": "app=hermes-webui", **(extra or {})} + return release_producer.Collector(env, "hermes-webui", WP, FakeOpener(routes), lambda: FIXED) + +@pytest.fixture +def stack(tmp_path): + env = secrets(tmp_path) + router = build_router(tmp_path, {"HUX_FLAGS": ALL_ON, "HUX_ROUTER_KEY": "rk", **env}) + server = serve(router, "127.0.0.1", 0) + threading.Thread(target=server.serve_forever, daemon=True).start() + base = f"http://127.0.0.1:{server.server_address[1]}" + _, project = dispatch(router, "POST", "/hux/v1/projects", {"name": "P"}) + _, conversation = dispatch(router, "POST", "/hux/v1/conversations", {"title": "C", "project_id": project["id"]}) + scope = f"/hux/v1/projects/{project['id']}/conversations/{conversation['id']}/releases" + token = tmp_path / "kube.token" + token.write_text("sa-token\n") + status, created = dispatch(router, "POST", scope, {"workload": "hermes-webui", "commit": "1" * 40, "evidence": {"review_url": REVIEW}}, + {"If-Match": "0", "Idempotency-Key": "producer-suite-create"}) + assert status == 201 + env.update({"HUX_BASE_URL": base, "HUX_TENANT_SLOT": "slot-3", "HUX_PRODUCER_SUBJECT": SUBJECT, + "HUX_PRODUCER_WORKLOAD": "hermes-webui", "HUX_PRODUCER_PROJECT_ID": project["id"], + "HUX_PRODUCER_CONVERSATION_ID": conversation["id"], "HUX_PRODUCER_NAMESPACE": "hermes", + "HUX_PRODUCER_KUBE_API": "https://kube.test", "HUX_PRODUCER_KUBE_TOKEN_FILE": str(token), + "HUX_PRODUCER_POD_SELECTOR": "app=hermes-webui"}) + yield {"router": router, "base": base, "scope": scope, "env": env, "release_id": created["release"]["id"]} + server.shutdown() + +def producer_client(stack, opener=None): + return release_producer.ProducerClient(stack["base"], "slot-3", SUBJECT, KEY, opener) + +def test_full_chain_replay_and_terminal_noop(stack): + routes = good_routes() + client = producer_client(stack) + release_id = stack["release_id"] + _, first_view, _ = client.request("GET", f"{stack['scope']}/{release_id}") + states = [] + for _ in range(6): + outcome = release_producer.run_once(stack["env"], opener=hybrid(routes), clock=lambda: FIXED) + assert outcome["changed"] is True and outcome["replayed"] is False + states.append(outcome["state"]) + assert states == ["merged", "built", "verified", "deployed", "converged", "live_verified"] + status, view, _ = client.request("GET", f"{stack['scope']}/{release_id}") + assert status == 200 and view["revision"] == 7 + evidence = view["release"]["evidence"] + assert evidence["image_ref"] == REF and evidence["pod_digest"] == DIGEST and evidence["health_check"]["status"] == "pass" + assert release_producer.run_once(stack["env"], opener=hybrid(routes), clock=lambda: FIXED) == {"changed": False, "reason": "no advanceable release"} + by_id = release_producer.run_once({**stack["env"], "HUX_PRODUCER_RELEASE_ID": release_id}, opener=hybrid(routes), clock=lambda: FIXED) + assert by_id == {"changed": False, "release_id": release_id, "state": "live_verified", "reason": "release is terminal"} + # A duplicate POST with the producer's deterministic key replays the same entry. + status, replay, replayed = client.request("POST", f"{stack['scope']}/{release_id}/transitions", + {"to": "merged", "evidence": {"merge_commit": SHA_M}}, + {"If-Match": "1", "Idempotency-Key": f"producer-{release_id}-merged-1"}) + assert status == 200 and replayed is True and replay["release"]["state"] == "merged" and replay["revision"] == 2 + # The driver reports a replayed transition as unchanged. + stale_get = lambda request, timeout, context: ( # noqa: E731 + FakeResponse(200, first_view) if request.get_method() == "GET" else urllib.request.urlopen(request, timeout=timeout)) + stub = SimpleNamespace(git=lambda url: dict(GIT_OBS)) + replay_client = release_producer.ProducerClient(stack["base"], "slot-3", SUBJECT, KEY, stale_get) + outcome = release_producer.advance(replay_client, stub, WP, 900, stack["scope"], release_id, FIXED) + assert outcome["changed"] is False and outcome["replayed"] is True and outcome["state"] == "merged" + healthz = release_producer._default_opener(urllib.request.Request(stack["base"] + "/healthz"), 5.0, None) + assert healthz.status == 200 + +def test_driver_never_posts_when_verification_or_collection_fails(stack): + routes = good_routes() + env, scope, release_id = stack["env"], stack["scope"], stack["release_id"] + client = producer_client(stack) + assert release_producer.run_once(env, opener=hybrid(routes), clock=lambda: FIXED)["state"] == "merged" + bad = dict(routes) + bad[JENKINS_API] = (200, {**routes[JENKINS_API][1], "result": "FAILURE"}) + with pytest.raises(Invalid, match="not SUCCESS"): + release_producer.run_once(env, opener=hybrid(bad), clock=lambda: FIXED) + gone = dict(routes) + gone[JENKINS_API] = (503, {}) + with pytest.raises(Invalid, match="returned HTTP 503"): + release_producer.run_once(env, opener=hybrid(gone), clock=lambda: FIXED) + collector = release_producer.Collector(env, "hermes-webui", WP, hybrid(routes), lambda: FIXED - timedelta(hours=1)) + with pytest.raises(Invalid, match="stale evidence"): + release_producer.advance(client, collector, WP, 900, scope, release_id, FIXED) + status, view, _ = client.request("GET", f"{scope}/{release_id}") + assert status == 200 and view["revision"] == 2 and view["release"]["state"] == "merged" + +HAPPY = [ + (rel("reviewed", review_url=REVIEW), "merged", {"git": GIT_OBS}, {"merge_commit": SHA_M}), + (rel("merged", merge_commit=SHA_M), "built", {"jenkins": JENKINS_OBS, "harbor": HARBOR_OBS}, + {"ci_build_url": JOB + "/20", "image_ref": REF, "image_digest": DIGEST, "harbor_digest": DIGEST}), + (rel("built", **BUILT_EV), "verified", {"harbor": HARBOR_OBS}, {}), + (rel("verified", **BUILT_EV), "deployed", {"flux": FLUX_OBS}, {"flux_revision": FLUX_REV}), + (rel("deployed", **BUILT_EV), "converged", {"workload_spec": SPEC_OBS, "pods": PODS_OBS}, {"pod_digest": DIGEST}), + (rel("converged", **BUILT_EV), "live_verified", {"health": HEALTH_OBS}, + {"health_check": {"url": HEALTH, "status": "pass", "at": NOW_S}}), + (rel("deployed", **BUILT_EV), "rolled_back", {"rollback": obs(digest=OTHER_DIGEST)}, {"rollback_target": OTHER_DIGEST}), +] + +@pytest.mark.parametrize(("release", "target", "observations", "payload"), HAPPY) +def test_verifier_produces_exact_payloads(release, target, observations, payload): + assert release_producer.evidence_for(release, target, observations, WP, 900, FIXED) == payload + +REJECTS = [ + (rel("banana"), "merged", {"git": GIT_OBS}, "unknown state"), + (rel("reviewed", review_url=REVIEW), "built", {"jenkins": JENKINS_OBS, "harbor": HARBOR_OBS}, "skip, repeat or downgrade"), + (rel("merged", merge_commit=SHA_M), "merged", {"git": GIT_OBS}, "skip, repeat or downgrade"), + (rel("live_verified", **BUILT_EV), "live_verified", {"health": HEALTH_OBS}, "skip, repeat or downgrade"), + (rel("reviewed", review_url=REVIEW), "rolled_back", {"rollback": obs(digest=OTHER_DIGEST)}, "no deployed image"), + (rel("reviewed", review_url=REVIEW), "merged", {}, "git is missing"), + (rel("reviewed", review_url=REVIEW), "merged", {"git": {**GIT_OBS, "workload": "hermes-agent"}}, "different workload"), + (rel("reviewed", review_url=REVIEW), "merged", {"git": {**GIT_OBS, "observed_at": OLD_S}}, "stale evidence"), + (rel("reviewed", review_url=REVIEW), "merged", {"git": {**GIT_OBS, "observed_at": "2026-08-24T13:00:00Z"}}, "stale evidence"), + (rel("reviewed", review_url=REVIEW), "merged", {"git": {**GIT_OBS, "observed_at": "not-a-time"}}, "RFC 3339"), + (rel("reviewed", review_url=REVIEW), "merged", {"git": obs(commit="short", committed_at=NOW_S, review_url=REVIEW)}, "40-hex"), + (rel("reviewed", review_url=REVIEW), "merged", {"git": {**GIT_OBS, "committed_at": "2026-08-24T13:00:00Z"}}, "in the future"), + (rel("reviewed", review_url=PREFIX + "56"), "merged", {"git": GIT_OBS}, "different review"), + (rel("reviewed", review_url="https://evil.example/pulls/55"), "merged", + {"git": {**GIT_OBS, "review_url": "https://evil.example/pulls/55"}}, "outside the policy review prefix"), + (rel("merged", merge_commit=SHA_M), "built", {"jenkins": JENKINS_OBS}, "harbor is missing"), + (rel("merged", merge_commit=SHA_M), "built", {"jenkins": {**JENKINS_OBS, "build_url": "https://jenkins.bstein.dev/job/other/20"}, "harbor": HARBOR_OBS}, "outside the policy job"), + (rel("merged", merge_commit=SHA_M), "built", {"jenkins": {**JENKINS_OBS, "result": "FAILURE"}, "harbor": HARBOR_OBS}, "not SUCCESS"), + (rel("merged", merge_commit=SHA_M), "built", {"jenkins": {**JENKINS_OBS, "revision": "f" * 40}, "harbor": HARBOR_OBS}, "does not equal the merged commit"), + (rel("merged", merge_commit="zz"), "built", {"jenkins": JENKINS_OBS, "harbor": HARBOR_OBS}, "does not equal the merged commit"), + (rel("merged", merge_commit=SHA_M), "built", {"jenkins": {**JENKINS_OBS, "build_number": True}, "harbor": HARBOR_OBS}, "build_number"), + (rel("merged", merge_commit=SHA_M), "built", {"jenkins": {**JENKINS_OBS, "build_number": 0}, "harbor": HARBOR_OBS}, "build number is invalid"), + (rel("merged", merge_commit=SHA_M), "built", {"jenkins": {**JENKINS_OBS, "build_number": 21}, "harbor": HARBOR_OBS}, "does not bind"), + (rel("merged", merge_commit=SHA_M), "built", {"jenkins": JENKINS_OBS, "harbor": {**HARBOR_OBS, "repository": "registry.bstein.dev/evil/x"}}, "policy image repository"), + (rel("merged", merge_commit=SHA_M), "built", {"jenkins": JENKINS_OBS, "harbor": {**HARBOR_OBS, "digest": "sha256:short"}}, "digest is malformed"), + (rel("merged", merge_commit=SHA_M), "built", {"jenkins": {**JENKINS_OBS, "image_digest": OTHER_DIGEST}, "harbor": HARBOR_OBS}, "advertised image digest"), + (rel("built", **BUILT_EV), "verified", {"harbor": {**HARBOR_OBS, "digest": OTHER_DIGEST}}, "no longer serves"), + (rel("built", review_url=REVIEW, merge_commit=SHA_M), "verified", {"harbor": HARBOR_OBS}, "no longer serves"), + (rel("verified", **BUILT_EV), "deployed", {"flux": {**FLUX_OBS, "name": "other"}}, "name does not match"), + (rel("verified", **BUILT_EV), "deployed", {"flux": {**FLUX_OBS, "applied_revision": "feature@sha1:" + "b" * 40}}, "main@sha1"), + (rel("verified", **BUILT_EV), "deployed", {"flux": {**FLUX_OBS, "pin_in_revision": False}}, "not proven"), + (rel("verified", **BUILT_EV), "deployed", {"flux": {**FLUX_OBS, "pin_in_revision": "true"}}, "not proven"), + (rel("deployed", **BUILT_EV), "converged", {"workload_spec": obs(image=f"{REPO}:latest@{DIGEST}"), "pods": PODS_OBS}, "desired workload image"), + (rel("deployed", review_url=REVIEW), "converged", {"workload_spec": SPEC_OBS, "pods": PODS_OBS}, "desired workload image"), + (rel("deployed", **BUILT_EV), "converged", {"workload_spec": SPEC_OBS, "pods": obs(pods=[])}, "no Ready pods"), + (rel("deployed", **BUILT_EV), "converged", {"workload_spec": SPEC_OBS, "pods": obs(pods=[{"name": "p0", "ready": False, "image_id": REPO + "@" + DIGEST}])}, "not Ready"), + (rel("deployed", **BUILT_EV), "converged", {"workload_spec": SPEC_OBS, "pods": obs(pods=["junk"])}, "not Ready"), + (rel("deployed", **BUILT_EV), "converged", {"workload_spec": SPEC_OBS, "pods": obs(pods=[{"name": "p0", "ready": True, "image_id": REPO + "@" + OTHER_DIGEST}])}, "not running the release digest"), + (rel("deployed", **BUILT_EV), "converged", {"workload_spec": SPEC_OBS, "pods": obs(pods=[{"name": "p0", "ready": True, "image_id": None}])}, "not running the release digest"), + (rel("converged", **BUILT_EV), "live_verified", {"health": {**HEALTH_OBS, "url": "https://evil.example/healthz"}}, "policy health URL"), + (rel("converged", **BUILT_EV), "live_verified", {"health": {**HEALTH_OBS, "status": 503}}, "HTTP 503, not 200"), + (rel("converged", **BUILT_EV), "live_verified", {"health": {**HEALTH_OBS, "status": True}}, "health.status"), + (rel("converged", **BUILT_EV), "live_verified", {"health": {**HEALTH_OBS, "checked_at": OLD_S}}, "stale evidence"), + (rel("converged", **BUILT_EV), "live_verified", {"health": {**HEALTH_OBS, "checked_at": None}}, "RFC 3339"), + (rel("deployed", **BUILT_EV), "rolled_back", {"rollback": obs(digest="bad")}, "rollback target digest"), + (rel("deployed", **BUILT_EV), "rolled_back", {"rollback": obs()}, "rollback.digest"), +] + +@pytest.mark.parametrize(("release", "target", "observations", "match"), REJECTS) +def test_verifier_rejects_every_bad_observation_set(release, target, observations, match): + with pytest.raises(Invalid, match=match): + release_producer.evidence_for(release, target, observations, WP, 900, FIXED) + +def test_verifier_requires_aware_clock_and_next_target_is_total(): + with pytest.raises(Invalid, match="timezone-aware"): + release_producer.evidence_for(rel("reviewed"), "merged", {}, WP, 900, datetime(2026, 8, 24)) + assert release_producer.next_target("live_verified") is None + assert release_producer.next_target("reviewed") == "merged" + +def test_collectors_extract_only_observed_fields(): + collector = make_collector(good_routes()) + git = collector.git(REVIEW) + assert git == obs(review_url=REVIEW, commit=SHA_M, committed_at="2026-08-24T11:00:00Z") + jenkins = collector.jenkins() + assert jenkins == obs(build_url=JOB + "/20", result="SUCCESS", build_number=20, revision=SHA_M, image_digest=DIGEST) + assert collector.harbor(TAG) == obs(repository=REPO, tag=TAG, digest=DIGEST) + with pytest.raises(Invalid, match="outside the policy review prefix"): + collector.git("https://evil.example/pulls/55") + with pytest.raises(Invalid, match="outside the policy review prefix"): + collector.git(PREFIX + "55x") + +def test_collectors_omit_unproven_fields(): + weak = make_collector({ + GIT_API: (200, {"merged": False, "merge_commit_sha": SHA_M, "merged_at": "yesterday"}), + JENKINS_API: (200, {"url": 5, "result": None, "number": True, "actions": [{"lastBuiltRevision": {"SHA1": "zz"}}], "description": None}), + HARBOR_API: (200, {"digest": "sha256:short", "tags": []}), + }) + assert weak.git(REVIEW) == obs(review_url=REVIEW) + assert weak.jenkins() == obs() + assert weak.harbor(TAG) == obs(repository=REPO) + bad_prefix = release_producer.Collector({}, "hermes-webui", {**WP, "review_url_prefix": "https://scm.bstein.dev/x/"}, FakeOpener({}), lambda: FIXED) + with pytest.raises(Invalid, match="not a proposal URL"): + bad_prefix.git("https://scm.bstein.dev/x/55") + bad_repo = release_producer.Collector({}, "hermes-webui", {**WP, "image_repository": "hermes"}, FakeOpener({}), lambda: FIXED) + with pytest.raises(Invalid, match="host/project/name"): + bad_repo.harbor(TAG) + +def test_fetch_is_https_only_bounded_and_fail_closed(): + collector = make_collector({HEALTH: (200, b"x" * (release_producer.MAX_RESPONSE_BYTES + 1)), JENKINS_API: (200, b"nonsense")}) + with pytest.raises(Invalid, match="must be HTTPS"): + collector._fetch("http://chat.bstein.dev/healthz", "health") + with pytest.raises(Invalid, match="fetch failed"): + collector._fetch("https://unrouted.example/x", "health") + with pytest.raises(Invalid, match="too large"): + collector._fetch(HEALTH, "health") + with pytest.raises(Invalid, match="not JSON"): + collector._json(JENKINS_API, "jenkins") + arrays = make_collector({JENKINS_API: (200, [1, 2])}) + with pytest.raises(Invalid, match="not a JSON object"): + arrays._json(JENKINS_API, "jenkins") + down = make_collector({HEALTH: (503, {})}) + assert down.health() == obs(url=HEALTH, status=503, checked_at=NOW_S) + +def test_tls_context_and_timeout_bounds(tmp_path): + ca = tmp_path / "ca.pem" + ca.write_text(CA_PEM) + assert release_producer._context(str(ca)).check_hostname is True + assert release_producer._context("").check_hostname is True + bad = tmp_path / "bad.pem" + bad.write_text("not a certificate") + with pytest.raises(Invalid, match="CA bundle"): + release_producer._context(str(bad)) + assert release_producer._bounded_timeout({}) == 10.0 + assert release_producer._bounded_timeout({"HUX_PRODUCER_TIMEOUT_SECONDS": "2.5"}) == 2.5 + assert release_producer._bounded_timeout({"HUX_PRODUCER_TIMEOUT_SECONDS": "nan-ish"}) == 10.0 + assert release_producer._bounded_timeout({"HUX_PRODUCER_TIMEOUT_SECONDS": "900"}) == 10.0 + plain = release_producer.Collector({}, "w", WP) + assert plain.opener is release_producer._default_opener and plain.clock().tzinfo is not None + +def test_kubernetes_collectors_fail_closed_on_credentials_and_config(tmp_path): + token = tmp_path / "token" + token.write_text("sa-token\n") + with_token = {"HUX_PRODUCER_KUBE_TOKEN_FILE": str(token)} + with pytest.raises(Invalid, match="credentials are unavailable"): + make_collector(good_routes(), {"HUX_PRODUCER_KUBE_TOKEN_FILE": str(tmp_path / "absent")}).pods() + empty = tmp_path / "empty" + empty.write_text(" \n") + with pytest.raises(Invalid, match="credentials are unavailable"): + make_collector(good_routes(), {"HUX_PRODUCER_KUBE_TOKEN_FILE": str(empty)}).pods() + bad_ns = make_collector(good_routes(), {**with_token, "HUX_PRODUCER_NAMESPACE": "Bad_NS"}) + with pytest.raises(Invalid, match="HUX_PRODUCER_NAMESPACE"): + bad_ns.pods() + with pytest.raises(Invalid, match="WORKLOAD_KIND"): + make_collector(good_routes(), {**with_token, "HUX_PRODUCER_WORKLOAD_KIND": "daemonset"}).workload_spec() + no_selector = make_collector(good_routes(), {**with_token, "HUX_PRODUCER_POD_SELECTOR": ""}) + with pytest.raises(Invalid, match="POD_SELECTOR"): + no_selector.pods() + +def test_kubernetes_collectors_observe_flux_workload_pods_and_rollback(tmp_path): + token = tmp_path / "token" + token.write_text("sa-token\n") + extra = {"HUX_PRODUCER_KUBE_TOKEN_FILE": str(token)} + collector = make_collector(good_routes(), extra) + assert collector.flux(REF) == obs(name="hermes", applied_revision=FLUX_REV, pin_in_revision=True) + assert collector.flux(f"{REPO}:other@{OTHER_DIGEST}")["pin_in_revision"] is False + assert collector.flux("")["pin_in_revision"] is False + assert collector.workload_spec() == obs(image=REF) + assert collector.pods() == obs(pods=[{"name": "p0", "ready": True, "image_id": REPO + "@" + DIGEST}]) + assert collector.rollback_target() == obs(digest=DIGEST) + routes = good_routes() + routes[KUSTOMIZATION_API] = (200, {"metadata": {}, "status": {"lastAppliedRevision": None, "conditions": [{"type": "Ready", "status": "False"}]}}) + routes[STS_API] = (200, {"spec": {"template": {"spec": {"containers": [{"image": "other:1"}]}}}}) + routes[PODS_API] = (200, {"items": [{"metadata": {}, "status": {"conditions": [{"type": "Other"}], "containerStatuses": [{"image": "other:1", "imageID": "x"}]}}]}) + drifted = make_collector(routes, extra) + flux = drifted.flux(REF) + assert "applied_revision" not in flux and flux["pin_in_revision"] is False and flux["name"] is None + assert drifted.workload_spec() == obs() + assert drifted.pods() == obs(pods=[{"name": None, "ready": False, "image_id": None}]) + assert drifted.rollback_target() == obs() + routes[STS_API] = (200, {"spec": {"template": {"spec": {"containers": [{"image": REPO + ":tag-without-digest"}]}}}}) + assert make_collector(routes, extra).rollback_target() == obs() + +def test_collect_for_covers_every_target_and_fails_closed(): + stub = SimpleNamespace(jenkins=lambda: obs(result="SUCCESS"), rollback_target=lambda: obs(digest=DIGEST)) + gathered = release_producer.collect_for(stub, "built", rel("merged", merge_commit=SHA_M)) + assert "harbor" not in gathered and gathered["jenkins"]["result"] == "SUCCESS" + assert release_producer.collect_for(stub, "rolled_back", rel("deployed"))["rollback"]["digest"] == DIGEST + with pytest.raises(Invalid, match="no collector plan"): + release_producer.collect_for(stub, "reviewed", rel("reviewed")) + +def test_client_requires_loopback_or_https_and_fails_closed(): + with pytest.raises(Invalid, match="loopback HTTP or HTTPS"): + release_producer.ProducerClient("http://10.0.0.5:8790", "slot-3", SUBJECT, KEY) + default = release_producer.ProducerClient("", "slot-3", SUBJECT, KEY) + assert default.base == release_producer.DEFAULT_BASE_URL and default.opener is release_producer._default_opener + unreachable = release_producer.ProducerClient("https://hux.test", "slot-3", SUBJECT, KEY, FakeOpener({})) + with pytest.raises(Invalid, match="unreachable"): + unreachable.request("GET", "/x") + odd = release_producer.ProducerClient("https://hux.test", "slot-3", SUBJECT, KEY, + FakeOpener({"https://hux.test/raw": (200, b"nonsense"), "https://hux.test/empty": (200, b"")})) + assert odd.request("GET", "/raw") == (200, None, False) + assert odd.request("GET", "/empty") == (200, None, False) + +def test_driver_read_and_transition_failures_are_invalid(): + scope = "/hux/v1/projects/p/conversations/c/releases" + stub = SimpleNamespace(git=lambda url: dict(GIT_OBS)) + def client_for(routes): + return release_producer.ProducerClient("https://hux.test", "slot-3", SUBJECT, KEY, FakeOpener(routes)) + with pytest.raises(Invalid, match="read failed with HTTP 500"): + release_producer.advance(client_for({f"https://hux.test{scope}/rel_1": (500, {})}), stub, WP, 900, scope, "rel_1", FIXED) + with pytest.raises(Invalid, match="carries no revision"): + release_producer.advance(client_for({f"https://hux.test{scope}/rel_1": (200, {"release": rel("reviewed")})}), stub, WP, 900, scope, "rel_1", FIXED) + routes = {f"https://hux.test{scope}/rel_1": (200, {"release": rel("reviewed", review_url=REVIEW), "revision": 1}), + f"https://hux.test{scope}/rel_1/transitions": (409, {})} + with pytest.raises(Invalid, match="rejected with HTTP 409"): + release_producer.advance(client_for(routes), stub, WP, 900, scope, "rel_1", FIXED) + listing = {f"https://hux.test{scope}": (200, {"items": ["junk", {"release": {"id": "rel_x", "workload": "hermes-agent", "state": "reviewed"}}, + {"release": {"id": "rel_y", "workload": "hermes-webui", "state": "rolled_back"}}, + {"release": {"id": "rel_z", "workload": "hermes-webui", "state": "merged"}}]})} + assert release_producer._pick_release(client_for(listing), scope, "hermes-webui") == "rel_z" + with pytest.raises(Invalid, match="list failed with HTTP 500"): + release_producer._pick_release(client_for({f"https://hux.test{scope}": (500, {})}), scope, "hermes-webui") + +@pytest.mark.parametrize("missing", ["HUX_PRODUCER_WORKLOAD", "HUX_TENANT_SLOT", "HUX_PRODUCER_SUBJECT", "HUX_PRODUCER_PROJECT_ID", "HUX_PRODUCER_CONVERSATION_ID"]) +def test_run_once_requires_complete_environment(tmp_path, missing): + env = {**secrets(tmp_path), "HUX_PRODUCER_WORKLOAD": "hermes-webui", "HUX_TENANT_SLOT": "slot-3", + "HUX_PRODUCER_SUBJECT": SUBJECT, "HUX_PRODUCER_PROJECT_ID": "prj_1", "HUX_PRODUCER_CONVERSATION_ID": "cnv_1"} + env.pop(missing) + with pytest.raises(Invalid, match=missing): + release_producer.run_once(env, opener=FakeOpener({})) + +def test_run_once_rejects_unknown_workload_and_reports_empty_scope(tmp_path): + env = {**secrets(tmp_path), "HUX_PRODUCER_WORKLOAD": "hermes-agent", "HUX_TENANT_SLOT": "slot-3", + "HUX_PRODUCER_SUBJECT": SUBJECT, "HUX_PRODUCER_PROJECT_ID": "prj_1", "HUX_PRODUCER_CONVERSATION_ID": "cnv_1"} + with pytest.raises(Invalid, match="not enabled"): + release_producer.run_once(env, opener=FakeOpener({})) + env["HUX_PRODUCER_WORKLOAD"] = "hermes-webui" + env["HUX_BASE_URL"] = "https://hux.test" + empty = FakeOpener({"https://hux.test/hux/v1/projects/prj_1/conversations/cnv_1/releases": (200, {"items": [], "next": None})}) + assert release_producer.run_once(env, opener=empty) == {"changed": False, "reason": "no advanceable release"} + assert empty.calls[0][0] == "GET" + +def test_module_and_suite_stay_bounded(): + for path in (FOUNDATION / "hux_producer" / "__init__.py", Path(__file__)): + assert len(path.read_text().splitlines()) <= 500