diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6fcb321..c6bd130 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -47,6 +47,8 @@ jobs: - uses: actions/setup-go@v7 with: go-version-file: go.mod + - name: Acceptance runner fail-closed behavior (mocked; no cluster) + run: python3 scripts/test_cluster_acceptance.py - run: pip install pyyaml - name: RBAC must not have drifted from config/rbac/role.yaml run: python3 scripts/sync-rbac.py --check diff --git a/README.md b/README.md index 6697944..606c906 100644 --- a/README.md +++ b/README.md @@ -927,6 +927,10 @@ See [SECURITY.md](SECURITY.md) to report a vulnerability. --- +## Cluster acceptance + +Use the [explicit-context acceptance runner](docs/CLUSTER_ACCEPTANCE.md) against an existing disposable test installation. It records ownership, workload readiness, exact image IDs, policy selectors, stability, and opt-in token/Temporal/network observations. Runner unit tests are mocked; real cluster acceptance is separate and remains unproven until those checks execute. + ## Contributing Pull requests are welcome. For substantial changes, open an issue first to discuss the approach. diff --git a/docs/CLUSTER_ACCEPTANCE.md b/docs/CLUSTER_ACCEPTANCE.md new file mode 100644 index 0000000..3eb0cf4 --- /dev/null +++ b/docs/CLUSTER_ACCEPTANCE.md @@ -0,0 +1,71 @@ +# Operator cluster acceptance observations + +This runner inspects an **existing** operator-managed Pulse installation on a user-selected OpenShift cluster. It does not create a cluster, install Pulse, change kubeconfig context, log in, or execute provider-backed agent prompts. Nothing in this guide has been run against a real cluster as part of the runner implementation. + +## Plan and read-only observations + +Requirements: Python 3.9+, `oc`, access to the explicit context/namespace/CR, and permission to read namespaces, OpenShiftPulse, Deployments, StatefulSets, ReplicaSets, Pods, NetworkPolicies, and relevant Secrets. Secret content stays in memory and is not written to the report. The cluster and CR must already exist. + +```bash +# No cluster calls: inspect the planned scope. +python3 scripts/cluster-acceptance.py \ + --context YOUR_CONTEXT --namespace YOUR_NAMESPACE --pulse YOUR_CR + +# Explicit opt-in to real reads. No mutations without additional flags. +python3 scripts/cluster-acceptance.py \ + --context YOUR_CONTEXT --namespace YOUR_NAMESPACE --pulse YOUR_CR \ + --execute --report /tmp/pulse-operator-observations.json +``` + +Every `oc` command carries the supplied context and namespace and a bounded request timeout. The runner verifies controller ownership, observed rollout generations/readiness, enabled optional workloads, exact pod image IDs (including UI sidecars), and the PostgreSQL NetworkPolicy's exact TCP 5432 peer selectors. Pod ownership is traced through ReplicaSets to Deployments, rather than trusting an `app` label alone. + +A 65-second observation checks that the nginx Secret resourceVersion and UI template generation/hash stay stable over more than two normal 30-second reconciliation intervals. This detects observable churn. It does **not** prove the operator sent no redundant API UPDATE requests: unchanged writes may be suppressed by the API server. Use API audit evidence if that distinction matters. + +Missing components, prerequisites, ownership, image IDs, expected policy, or readiness cause failure and a nonzero exit. JSON records context, CR name/UID/generation, checks, pod image IDs, and explicit unverified acceptance areas. `status: passed` refers only to requested operator observations; `acceptance: incomplete` remains explicit because these observations cannot establish full release readiness. + +## Optional mutation checks in a disposable namespace + +Mutation checks require `--allow-mutations` **and** an existing namespace labeled `pulse.ai/acceptance=disposable`. Do not label a production namespace to bypass this prerequisite. Stop active experiments/workflows before token rotation or Temporal changes. These checks briefly interrupt service access or Temporal's database access. + +For a disposable test namespace, set the label through your normal reviewed cluster workflow: + +```bash +oc --context YOUR_CONTEXT label namespace YOUR_NAMESPACE pulse.ai/acceptance=disposable +``` + +Then choose the checks explicitly: + +```bash +python3 scripts/cluster-acceptance.py \ + --context YOUR_CONTEXT --namespace YOUR_NAMESPACE --pulse YOUR_CR \ + --execute --allow-mutations \ + --rotate-token --exercise-temporal-toggle \ + --network-probe-image YOUR_PYTHON_IMAGE@sha256:YOUR_64_HEX_DIGEST \ + --report /tmp/pulse-operator-mutation-observations.json +``` + +The image must contain `python3`, support an arbitrary non-root UID, and be available to the cluster. Use a tested digest; do not copy the placeholder literally. OpenShift SCC allocates the UID; the pod drops capabilities, disables privilege escalation, mounts no service-account token, and requests a read-only root filesystem. Image pull or scheduling failure is a failed prerequisite, never a successful denial test. + +- **Packet probes:** creates short-lived probe pods with agent, Temporal, and unrelated label classes. Each opens a fresh TCP connection to PostgreSQL. Agent succeeds; Temporal succeeds only when enabled; unrelated labels must fail. This tests observed network behavior for label classes, not database authentication or actual worker workflow execution. Probe cleanup uses an atomic UID precondition in raw DeleteOptions so a replacement pod cannot be deleted by name. +- **Token rotation:** changes only the WS token key using UID/resourceVersion checks, observes a changed nginx hash and ready rollout, then restores the original token and waits for the original hash/rollout. Patch bodies travel over stdin, not process arguments. No token is logged or included in the report. This verifies propagation, not an authenticated browser session. +- **Temporal toggle:** requires Temporal initially enabled. Disables it, checks that its PostgreSQL policy peer is revoked, then restores enabled state and readiness. With the probe image it checks fresh Temporal-label connections in both states. It does not provision Temporal on an initially disabled installation or infer workflow durability from a ready Deployment. + +Restoration runs in `finally`, but process termination, credential expiry, network failure, or concurrent configuration changes can prevent it. A restoration failure is a failed check. Inspect the supplied CR/token Secret and restore intended configuration through your normal operational procedure; the runner deliberately does not overwrite concurrent token or CR changes. Reports do not contain backup credentials. + +## Runner verification and release evidence + +```bash +python3 scripts/test_cluster_acceptance.py +``` + +These tests mock every `oc` operation. They verify fail-closed prerequisites, explicit targeting, policy matrix, exact-image prerequisites, safe reporting, patch stdin, UID cleanup, and restoration/error behavior. They prove runner behavior only; they are not cluster health evidence. + +Before accepting a release, retain: + +1. Exact matched agent/UI versions and pod `imageID`/digest evidence, operator version, cluster context and CR UID/generation. +2. UI isolated acceptance output from `pnpm run test:acceptance` in pulse-ui. This validates controlled frontend behavior; it does not replace browser/proxy tests on OpenShift. +3. Agent outcome/behavior gate output using the current pulse-agent acceptance instructions, including failed or unavailable provider runs. The provider-backed suite and runtime outcome checks are distinct from fixture dry-runs. +4. Actual operator runner reports, including packet probes in enabled and disabled Temporal states when testing that feature. Missing optional mutation checks remain missing evidence. +5. Authenticated UI/API, user RBAC/approval denial, verified writes/rollback in a disposable scope, Temporal approvals/persistence, and database recovery evidence. Those flows require separate end-to-end checks; this runner lists them as unverified. + +Do not turn mocked tests, a policy manifest, pod readiness, or an `acceptance: incomplete` report into a claim that the full Pulse release is healthy. diff --git a/scripts/cluster-acceptance.py b/scripts/cluster-acceptance.py new file mode 100755 index 0000000..0be9f0b --- /dev/null +++ b/scripts/cluster-acceptance.py @@ -0,0 +1,680 @@ +#!/usr/bin/env python3 +"""Opt-in OpenShift acceptance observations; never installs Pulse or changes context.""" + +from __future__ import annotations + +import argparse +import base64 +import json +import math +import re +import secrets +import shutil +import subprocess +import sys +import time +from pathlib import Path + + +class AcceptanceError(RuntimeError): + pass + + +class Runner: + def __init__( + self, args, command=subprocess.run, sleep=time.sleep, clock=time.monotonic + ): + self.args, self.command, self.sleep, self.clock = args, command, sleep, clock + self.checks = [] + self.cr = None + self.image_evidence = [] + + def oc(self, *argv, payload=None): + command = [ + "oc", + "--context", + self.args.context, + "--namespace", + self.args.namespace, + "--request-timeout=30s", + *argv, + ] + result = self.command( + command, + input=json.dumps(payload) if payload is not None else None, + capture_output=True, + text=True, + timeout=45, + ) + if result.returncode: + # oc may echo request payloads or Secrets. Never copy raw output into reports. + raise AcceptanceError( + f"oc {argv[0]} failed (exit {result.returncode}); inspect cluster separately" + ) + try: + return json.loads(result.stdout) if result.stdout.strip() else {} + except json.JSONDecodeError as exc: + raise AcceptanceError("oc returned invalid JSON") from exc + + def get(self, kind, name): + return self.oc("get", kind, name, "-o", "json") + + def check(self, name, fn): + try: + detail = fn() + except Exception as exc: + self.checks.append( + { + "name": name, + "status": "failed", + "detail": str(exc) + if isinstance(exc, AcceptanceError) + else f"{type(exc).__name__}; inspect prerequisite/cluster separately", + } + ) + raise + self.checks.append({"name": name, "status": "passed", "detail": detail}) + + def wait(self, predicate, description): + deadline = self.clock() + self.args.timeout + while True: + result = predicate() + if result: + return result + if self.clock() >= deadline: + raise AcceptanceError(f"timed out: {description}") + self.sleep(self.args.poll) + + def owned(self, obj): + if not any( + o.get("uid") == self.cr["metadata"]["uid"] and o.get("controller") is True + for o in obj.get("metadata", {}).get("ownerReferences", []) + ): + raise AcceptanceError( + "observed resource lacks expected Pulse controller owner" + ) + + def prerequisites(self): + ns = self.get("namespace", self.args.namespace) + self.cr = self.get("openshiftpulse.pulse.ai", self.args.pulse) + if self.cr["metadata"].get("namespace") != self.args.namespace: + raise AcceptanceError("CR namespace mismatch") + if ( + self.args.rotate_token + or self.args.exercise_temporal_toggle + or self.args.network_probe_image + ): + if not self.args.allow_mutations: + raise AcceptanceError("mutation checks require --allow-mutations") + if ( + ns["metadata"].get("labels", {}).get("pulse.ai/acceptance") + != "disposable" + ): + raise AcceptanceError( + "mutation checks require namespace label pulse.ai/acceptance=disposable" + ) + return "explicit context, namespace and existing CR accessible; no installation performed" + + def readiness(self): + names = [ + ("deployment", self.args.pulse + "-openshift-sre-agent"), + ("deployment", self.args.pulse + "-openshiftpulse"), + ("statefulset", self.args.pulse + "-openshift-sre-agent-postgresql"), + ] + if self.cr.get("spec", {}).get("agent", {}).get("mcp", {}).get("enabled"): + names.append(("deployment", self.args.pulse + "-mcp-server")) + if self.cr.get("spec", {}).get("temporal", {}).get("enabled", False): + names.append(("deployment", self.args.pulse + "-temporal")) + if self.cr["spec"]["temporal"].get("ui"): + names.append(("deployment", self.args.pulse + "-temporal-ui")) + workload_uids = {} + for kind, name in names: + + def ready(kind=kind, name=name): + obj = self.get(kind, name) + self.owned(obj) + workload_uids[name] = obj["metadata"]["uid"] + desired, status = ( + obj.get("spec", {}).get("replicas", 1), + obj.get("status", {}), + ) + return ( + desired > 0 + and status.get("observedGeneration", 0) + >= obj["metadata"]["generation"] + and status.get("readyReplicas", 0) == desired + and status.get("updatedReplicas", 0) == desired + ) + + self.wait(ready, f"{kind}/{name} rollout ready") + pods = self.oc("get", "pods", "-o", "json").get("items", []) + evidence = [] + for kind, name in names: + selected = [ + pod + for pod in pods + if pod.get("metadata", {}).get("labels", {}).get("app") == name + and not pod.get("metadata", {}).get("deletionTimestamp") + ] + if not selected: + raise AcceptanceError(f"no pods found for workload {name}") + for pod in selected: + refs = [ + o + for o in pod["metadata"].get("ownerReferences", []) + if o.get("controller") is True + ] + if len(refs) != 1: + raise AcceptanceError( + f"pod lacks controller owner chain for {name}" + ) + owner = refs[0] + if kind == "deployment": + if owner.get("kind") != "ReplicaSet": + raise AcceptanceError( + f"pod controller is not expected ReplicaSet for {name}" + ) + rs = self.get("replicaset", owner["name"]) + if rs["metadata"]["uid"] != owner["uid"] or not any( + o.get("uid") == workload_uids[name] + and o.get("controller") is True + for o in rs["metadata"].get("ownerReferences", []) + ): + raise AcceptanceError( + f"pod ReplicaSet does not belong to workload {name}" + ) + elif ( + owner.get("uid") != workload_uids[name] + or owner.get("kind") != "StatefulSet" + ): + raise AcceptanceError( + f"pod controller does not belong to workload {name}" + ) + statuses = pod.get("status", {}).get("containerStatuses", []) + expected_containers = {c["name"] for c in pod["spec"]["containers"]} + if {c["name"] for c in statuses} != expected_containers or any( + not c.get("imageID") for c in statuses + ): + raise AcceptanceError( + f"missing exact imageID evidence for workload {name}" + ) + evidence.append( + { + "workload": name, + "pod": pod["metadata"]["name"], + "uid": pod["metadata"]["uid"], + "containers": [ + { + "name": c["name"], + "image": c["image"], + "imageID": c["imageID"], + } + for c in statuses + ], + } + ) + self.image_evidence = evidence + return "owned workload rollouts ready; exact pod/container imageIDs captured (including sidecars)" + + def policy(self, enabled=None): + if enabled is None: + enabled = self.cr.get("spec", {}).get("temporal", {}).get("enabled", False) + np = self.get("networkpolicy", self.args.pulse + "-pg-access") + self.owned(np) + spec = np.get("spec", {}) + expected = {self.args.pulse + "-openshift-sre-agent"} + if enabled: + expected.add(self.args.pulse + "-temporal") + rules = spec.get("ingress", []) + valid = ( + spec.get("podSelector") + == { + "matchLabels": { + "app": self.args.pulse + "-openshift-sre-agent-postgresql" + } + } + and spec.get("policyTypes") == ["Ingress"] + and len(rules) == 1 + ) + if valid: + rule = rules[0] + peers = rule.get("from", []) + valid = ( + rule.get("ports") == [{"port": 5432, "protocol": "TCP"}] + and len(peers) == len(expected) + and { + p.get("podSelector", {}).get("matchLabels", {}).get("app") + for p in peers + } + == expected + and all( + p + == { + "podSelector": { + "matchLabels": { + "app": p["podSelector"]["matchLabels"]["app"] + } + } + } + for p in peers + ) + ) + if not valid: + raise AcceptanceError( + f"PostgreSQL policy is not scoped to expected TCP5432 peers (Temporal {enabled})" + ) + return f"actual NetworkPolicy manifest admits agent and Temporal={enabled}; packet enforcement not tested" + + def idempotency(self): + def snapshot(): + secret = self.get("secret", self.args.pulse + "-nginx") + deploy = self.get("deployment", self.args.pulse + "-openshiftpulse") + self.owned(secret) + self.owned(deploy) + return ( + secret["metadata"]["resourceVersion"], + deploy["metadata"]["generation"], + deploy["spec"]["template"] + .get("metadata", {}) + .get("annotations", {}) + .get("pulse.ai/nginx-config-hash"), + ) + + before = snapshot() + if not before[2]: + raise AcceptanceError("UI template lacks nginx configuration hash") + self.sleep(self.args.observe_seconds) + if snapshot() != before: + raise AcceptanceError( + "nginx Secret or UI template changed during stability observation" + ) + return "Secret resourceVersion and UI generation/hash stable; does not prove absence of no-op API UPDATE requests" + + def probe(self, label, should_connect): + name = "pulse-acceptance-" + secrets.token_hex(5) + host = self.args.pulse + "-openshift-sre-agent-postgresql" + code = ( + "import socket,json; s=socket.socket(); s.settimeout(5); ok=s.connect_ex((" + + repr(host) + + ",5432))==0; print(json.dumps({'connected':ok})); s.close()" + ) + pod = { + "apiVersion": "v1", + "kind": "Pod", + "metadata": {"name": name, "labels": {"app": label}}, + "spec": { + "restartPolicy": "Never", + "automountServiceAccountToken": False, + "securityContext": { + "runAsNonRoot": True, + "seccompProfile": {"type": "RuntimeDefault"}, + }, + "containers": [ + { + "name": "probe", + "image": self.args.network_probe_image, + "command": ["python3", "-c", code], + "securityContext": { + "allowPrivilegeEscalation": False, + "readOnlyRootFilesystem": True, + "capabilities": {"drop": ["ALL"]}, + }, + "resources": { + "requests": {"cpu": "10m", "memory": "32Mi"}, + "limits": {"cpu": "100m", "memory": "64Mi"}, + }, + } + ], + }, + } + created = False + try: + obj = self.oc("create", "-f", "-", "-o", "json", payload=pod) + created = True + uid = obj["metadata"]["uid"] + + def finished(): + current = self.get("pod", name) + if current["metadata"]["uid"] != uid: + raise AcceptanceError("probe pod identity changed") + phase = current.get("status", {}).get("phase") + if phase == "Failed": + raise AcceptanceError( + "probe pod failed; no network conclusion available" + ) + return phase == "Succeeded" + + self.wait(finished, "network probe pod completion") + result = self.oc("logs", name, "-c", "probe") + if ( + type(result.get("connected")) is not bool + or result["connected"] != should_connect + ): + raise AcceptanceError( + f"TCP5432 probe result mismatched for label class {label}" + ) + finally: + if created: + self.oc( + "delete", + "--raw", + f"/api/v1/namespaces/{self.args.namespace}/pods/{name}", + "-f", + "-", + payload={ + "apiVersion": "v1", + "kind": "DeleteOptions", + "preconditions": {"uid": uid}, + }, + ) + self.wait( + lambda: ( + self.oc("get", "pod", name, "--ignore-not-found", "-o", "json") + .get("metadata", {}) + .get("uid") + != uid + ), + "original probe pod removed", + ) + return "new TCP connection observed; probe pod removed" + + def network(self, enabled=None): + if enabled is None: + enabled = self.cr.get("spec", {}).get("temporal", {}).get("enabled", False) + self.probe(self.args.pulse + "-openshift-sre-agent", True) + self.probe(self.args.pulse + "-temporal", enabled) + self.probe("pulse-acceptance-untrusted", False) + return f"fresh agent-label TCP connection allowed; Temporal-label allowed={enabled}; unrelated label denied" + + def patch(self, kind, name, operations): + return self.oc( + "patch", + kind, + name, + "--type=json", + "--patch-file", + "/dev/stdin", + "-o", + "json", + payload=operations, + ) + + def rotate(self): + name = self.args.pulse + "-ws-token" + old = self.get("secret", name) + self.owned(old) + original = old.get("data", {}).get("token") + if not original: + raise AcceptanceError("WS token Secret missing token data") + before = self.get("deployment", self.args.pulse + "-openshiftpulse")["spec"][ + "template" + ]["metadata"]["annotations"]["pulse.ai/nginx-config-hash"] + token = base64.b64encode(secrets.token_urlsafe(32).encode()).decode() + changed = False + try: + self.patch( + "secret", + name, + [ + { + "op": "test", + "path": "/metadata/uid", + "value": old["metadata"]["uid"], + }, + { + "op": "test", + "path": "/metadata/resourceVersion", + "value": old["metadata"]["resourceVersion"], + }, + {"op": "replace", "path": "/data/token", "value": token}, + ], + ) + changed = True + self.wait( + lambda: ( + self.get("deployment", self.args.pulse + "-openshiftpulse")["spec"][ + "template" + ]["metadata"]["annotations"]["pulse.ai/nginx-config-hash"] + != before + ), + "token rotation propagates nginx hash", + ) + self.readiness() + finally: + if changed: + self.patch( + "secret", + name, + [ + { + "op": "test", + "path": "/metadata/uid", + "value": old["metadata"]["uid"], + }, + {"op": "test", "path": "/data/token", "value": token}, + {"op": "replace", "path": "/data/token", "value": original}, + ], + ) + self.wait( + lambda: ( + self.get("deployment", self.args.pulse + "-openshiftpulse")[ + "spec" + ]["template"]["metadata"]["annotations"][ + "pulse.ai/nginx-config-hash" + ] + == before + ), + "original token hash restored", + ) + self.readiness() + return "rotated token, observed new UI hash/ready rollout, restored original token/hash; authenticated traffic not tested" + + def toggle(self): + if not self.cr.get("spec", {}).get("temporal", {}).get("enabled", False): + raise AcceptanceError( + "toggle check requires initially enabled Temporal; does not provision it" + ) + changed = False + try: + current = self.get("openshiftpulse.pulse.ai", self.args.pulse) + self.patch( + "openshiftpulse.pulse.ai", + self.args.pulse, + [ + { + "op": "test", + "path": "/metadata/resourceVersion", + "value": current["metadata"]["resourceVersion"], + }, + {"op": "test", "path": "/spec/temporal/enabled", "value": True}, + {"op": "replace", "path": "/spec/temporal/enabled", "value": False}, + ], + ) + changed = True + + def revoked(): + try: + self.policy(False) + return True + except AcceptanceError: + return False + + self.wait(revoked, "Temporal peer revoked") + if self.args.network_probe_image: + self.network(False) + finally: + if changed: + self.patch( + "openshiftpulse.pulse.ai", + self.args.pulse, + [ + { + "op": "test", + "path": "/metadata/uid", + "value": self.cr["metadata"]["uid"], + }, + { + "op": "test", + "path": "/spec/temporal/enabled", + "value": False, + }, + { + "op": "replace", + "path": "/spec/temporal/enabled", + "value": True, + }, + ], + ) + + def restored(): + try: + self.policy(True) + return True + except AcceptanceError: + return False + + self.wait(restored, "Temporal peer restored") + if self.args.network_probe_image: + self.network(True) + self.readiness() + return "Temporal policy peer revoked while disabled and restored when enabled; workflow/packet behavior not tested" + + def run(self): + for name, fn in [ + ("prerequisites", self.prerequisites), + ("workload-readiness", self.readiness), + ("postgresql-policy", self.policy), + ("nginx-stability-observation", self.idempotency), + ]: + self.check(name, fn) + if self.args.network_probe_image: + self.check("postgresql-packet-probes", self.network) + if self.args.rotate_token: + self.check("token-rotation-and-restoration", self.rotate) + if self.args.exercise_temporal_toggle: + self.check("temporal-policy-toggle-and-restoration", self.toggle) + + +def parser(): + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("--context", required=True) + p.add_argument("--namespace", required=True) + p.add_argument("--pulse", required=True) + p.add_argument( + "--execute", action="store_true", help="opt in to real cluster reads/checks" + ) + p.add_argument("--allow-mutations", action="store_true") + p.add_argument("--rotate-token", action="store_true") + p.add_argument("--exercise-temporal-toggle", action="store_true") + p.add_argument( + "--network-probe-image", + help="digest-pinned image containing python3; creates/deletes probe pods", + ) + p.add_argument( + "--report", type=Path, default=Path("cluster-acceptance-report.json") + ) + p.add_argument("--timeout", type=float, default=600) + p.add_argument("--poll", type=float, default=5) + p.add_argument( + "--observe-seconds", + type=float, + default=65, + help="at least two 30s controller reconciliation intervals", + ) + return p + + +def main(argv=None): + args = parser().parse_args(argv) + if any( + not re.fullmatch(r"[a-z0-9]([-a-z0-9]*[a-z0-9])?", v) + for v in [args.namespace, args.pulse] + ): + raise SystemExit("namespace/pulse must be DNS labels") + if args.network_probe_image and not re.fullmatch( + r"[^\s]+@sha256:[0-9a-f]{64}", args.network_probe_image + ): + raise SystemExit("network probe image must be pinned by sha256 digest") + if ( + not all( + math.isfinite(v) for v in [args.timeout, args.poll, args.observe_seconds] + ) + or min(args.timeout, args.poll) <= 0 + or args.observe_seconds < 65 + ): + raise SystemExit("timeout/poll must be positive and observe-seconds >=65") + if not args.execute: + print( + "Plan only. No cluster contacted. Add --execute for explicit cluster checks; mutation checks also require disposable label and --allow-mutations." + ) + return 0 + runner = Runner(args) + status = "failed" + try: + if not shutil.which("oc"): + raise AcceptanceError("oc executable not found") + runner.run() + status = "passed" + except Exception as exc: # noqa: BLE001 -- every runner failure must produce a redacted failed report + if not runner.checks or runner.checks[-1]["status"] != "failed": + runner.checks.append( + { + "name": "runner", + "status": "failed", + "detail": str(exc) + if isinstance(exc, AcceptanceError) + else type(exc).__name__, + } + ) + report = { + "status": status, + "acceptance": "incomplete" if status == "passed" else "failed", + "cr_identity": ( + { + k: runner.cr["metadata"].get(k) + for k in ["name", "namespace", "uid", "generation"] + } + if runner.cr + else None + ), + "images": runner.image_evidence, + "context": args.context, + "namespace": args.namespace, + "pulse": args.pulse, + "checks": runner.checks, + "optional_checks": { + "token_rotation": args.rotate_token, + "temporal_toggle": args.exercise_temporal_toggle, + "packet_probes": bool(args.network_probe_image), + }, + "unverified": [ + *( + ["token rotation propagation/restoration"] + if not args.rotate_token + else [] + ), + *( + ["Temporal enabled/disabled policy transition"] + if not args.exercise_temporal_toggle + else [] + ), + *( + ["actual NetworkPolicy packet enforcement"] + if not args.network_probe_image + else [] + ), + "authenticated UI/API requests", + "provider-backed agent behavior", + "Temporal workflow persistence and approvals", + "API audit proof of no-op reconciliation", + ], + } + args.report.write_text(json.dumps(report, indent=2) + "\n") + print( + f"Operator checks {status}; full release acceptance remains unproven. Report: {args.report}" + ) + return 0 if status == "passed" else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/test_cluster_acceptance.py b/scripts/test_cluster_acceptance.py new file mode 100644 index 0000000..d7f7ce2 --- /dev/null +++ b/scripts/test_cluster_acceptance.py @@ -0,0 +1,384 @@ +"""Runner behavior tests; all oc commands mocked, no cluster contacted.""" + +import importlib.util +import json +import tempfile +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock, patch + +spec = importlib.util.spec_from_file_location( + "cluster_acceptance", Path(__file__).with_name("cluster-acceptance.py") +) +m = importlib.util.module_from_spec(spec) +spec.loader.exec_module(m) + + +def args(**changes): + values = { + "context": "test-context", + "namespace": "pulse-test", + "pulse": "pulse", + "allow_mutations": False, + "rotate_token": False, + "exercise_temporal_toggle": False, + "network_probe_image": None, + "timeout": 10, + "poll": 1, + "observe_seconds": 65, + } + values.update(changes) + return SimpleNamespace(**values) + + +def owned(**fields): + return dict( + metadata={ + "uid": "resource", + "ownerReferences": [{"uid": "cr-uid", "controller": True}], + }, + **fields, + ) + + +class RunnerTests(unittest.TestCase): + def runner(self, **changes): + runner = m.Runner(args(**changes), sleep=Mock()) + runner.cr = {"metadata": {"uid": "cr-uid"}, "spec": {}} + return runner + + def test_every_command_pins_context_namespace_and_bounded_request(self): + command = Mock(return_value=SimpleNamespace(returncode=0, stdout="{}")) + runner = m.Runner(args(), command=command) + runner.get("secret", "pulse-nginx") + self.assertEqual( + command.call_args.args[0][:7], + [ + "oc", + "--context", + "test-context", + "--namespace", + "pulse-test", + "--request-timeout=30s", + "get", + ], + ) + self.assertEqual(command.call_args.kwargs["timeout"], 45) + + def test_command_failure_does_not_leak_secret_response(self): + command = Mock( + return_value=SimpleNamespace( + returncode=1, stdout="private-token", stderr="secret-password" + ) + ) + with self.assertRaises(m.AcceptanceError) as error: + m.Runner(args(), command=command).get("secret", "pulse-ws-token") + self.assertNotIn("private-token", str(error.exception)) + self.assertNotIn("secret-password", str(error.exception)) + + def test_mutations_require_both_opt_in_and_disposable_label(self): + for allow, label in [(False, "disposable"), (True, None)]: + runner = self.runner(rotate_token=True, allow_mutations=allow) + runner.get = Mock( + side_effect=[ + {"metadata": {"labels": {"pulse.ai/acceptance": label}}}, + {"metadata": {"namespace": "pulse-test", "uid": "cr-uid"}}, + ] + ) + with self.assertRaises(m.AcceptanceError): + runner.prerequisites() + + def test_namespace_mismatch_fails(self): + runner = self.runner() + runner.get = Mock( + side_effect=[{"metadata": {}}, {"metadata": {"namespace": "other"}}] + ) + with self.assertRaises(m.AcceptanceError): + runner.prerequisites() + + def test_unscoped_policy_or_other_port_fails(self): + runner = self.runner() + for peers, port in [ + ([{}], 5432), + ( + [ + { + "podSelector": { + "matchLabels": {"app": "pulse-openshift-sre-agent"} + } + } + ], + 443, + ), + ]: + runner.get = Mock( + return_value=owned( + spec={ + "podSelector": { + "matchLabels": { + "app": "pulse-openshift-sre-agent-postgresql" + } + }, + "policyTypes": ["Ingress"], + "ingress": [ + { + "from": peers, + "ports": [{"protocol": "TCP", "port": port}], + } + ], + } + ) + ) + with self.assertRaises(m.AcceptanceError): + runner.policy(False) + + def test_temporal_enabled_disabled_policy_matrix(self): + runner = self.runner() + for enabled in [False, True]: + peers = [ + {"podSelector": {"matchLabels": {"app": "pulse-openshift-sre-agent"}}} + ] + if enabled: + peers.append( + {"podSelector": {"matchLabels": {"app": "pulse-temporal"}}} + ) + runner.get = Mock( + return_value=owned( + spec={ + "podSelector": { + "matchLabels": { + "app": "pulse-openshift-sre-agent-postgresql" + } + }, + "policyTypes": ["Ingress"], + "ingress": [ + { + "from": peers, + "ports": [{"protocol": "TCP", "port": 5432}], + } + ], + } + ) + ) + runner.policy(enabled) + with self.assertRaises(m.AcceptanceError): + runner.policy(not enabled) + + def test_noop_update_observation_detects_secret_resourceversion_churn(self): + runner = self.runner() + + def secret(version): + obj = owned() + obj["metadata"]["resourceVersion"] = version + return obj + + deployment = owned( + spec={ + "template": { + "metadata": {"annotations": {"pulse.ai/nginx-config-hash": "hash"}} + } + } + ) + deployment["metadata"]["generation"] = 3 + runner.get = Mock( + side_effect=[secret("1"), deployment, secret("2"), deployment] + ) + with self.assertRaises(m.AcceptanceError): + runner.idempotency() + + def test_missing_image_evidence_fails_instead_of_claiming_readiness(self): + runner = self.runner() + obj = owned( + spec={"replicas": 1}, + status={"observedGeneration": 1, "readyReplicas": 1, "updatedReplicas": 1}, + ) + obj["metadata"]["generation"] = 1 + runner.get = Mock(return_value=obj) + runner.oc = Mock(return_value={"items": []}) + with self.assertRaises(m.AcceptanceError): + runner.readiness() + + def test_token_restored_when_rotation_observation_fails(self): + runner = self.runner() + secret = owned(data={"token": "old-token"}) + secret["metadata"].update(resourceVersion="10") + deploy = { + "spec": { + "template": { + "metadata": {"annotations": {"pulse.ai/nginx-config-hash": "hash"}} + } + } + } + runner.get = Mock(side_effect=[secret, deploy]) + runner.patch = Mock() + runner.wait = Mock(side_effect=[m.AcceptanceError("rollout failed"), True]) + runner.readiness = Mock() + with self.assertRaises(m.AcceptanceError): + runner.rotate() + restoration = runner.patch.call_args_list[-1].args[2] + self.assertEqual( + restoration[-1], + {"op": "replace", "path": "/data/token", "value": "old-token"}, + ) + self.assertEqual(restoration[0]["value"], "resource") + + def test_probe_failure_still_removes_temporary_pod(self): + runner = self.runner(network_probe_image="python@sha256:" + "a" * 64) + runner.oc = Mock(return_value={"metadata": {"uid": "probe-uid"}}) + runner.wait = Mock(side_effect=[m.AcceptanceError("probe failed"), True]) + with self.assertRaises(m.AcceptanceError): + runner.probe("pulse-temporal", True) + self.assertEqual(runner.oc.call_args.args[0:2], ("delete", "--raw")) + self.assertEqual( + runner.oc.call_args.kwargs["payload"]["preconditions"]["uid"], "probe-uid" + ) + + def test_patch_payload_is_stdin_not_process_arguments(self): + runner = self.runner() + runner.oc = Mock() + operations = [ + {"op": "replace", "path": "/data/token", "value": "private-token"} + ] + runner.patch("secret", "pulse-ws-token", operations) + self.assertNotIn("private-token", str(runner.oc.call_args.args)) + self.assertEqual(runner.oc.call_args.kwargs["payload"], operations) + self.assertIn("/dev/stdin", runner.oc.call_args.args) + + def test_nonfinite_time_values_rejected_before_execution(self): + for flag in ["--timeout", "--poll", "--observe-seconds"]: + for value in ["nan", "inf"]: + with self.assertRaises(SystemExit): + m.main( + [ + "--context", + "test", + "--namespace", + "pulse-test", + "--pulse", + "pulse", + flag, + value, + ] + ) + + def test_exact_pod_image_ids_and_owner_chains_recorded(self): + runner = self.runner() + names = [ + "pulse-openshift-sre-agent", + "pulse-openshiftpulse", + "pulse-openshift-sre-agent-postgresql", + ] + pods = [] + for index, name in enumerate(names): + owner = { + "kind": "ReplicaSet" if index < 2 else "StatefulSet", + "name": name + "-rs", + "uid": name + "-rs" if index < 2 else name, + "controller": True, + } + pods.append( + { + "metadata": { + "name": name + "-pod", + "uid": name + "-pod", + "labels": {"app": name}, + "ownerReferences": [owner], + }, + "spec": {"containers": [{"name": "app"}]}, + "status": { + "containerStatuses": [ + { + "name": "app", + "image": "repo:v1", + "imageID": "repo@sha256:digest", + } + ] + }, + } + ) + + def get(kind, name): + if kind == "replicaset": + return { + "metadata": { + "uid": name, + "ownerReferences": [{"uid": name[:-3], "controller": True}], + } + } + obj = owned( + spec={"replicas": 1}, + status={ + "observedGeneration": 1, + "readyReplicas": 1, + "updatedReplicas": 1, + }, + ) + obj["metadata"].update(uid=name, generation=1) + return obj + + runner.get = Mock(side_effect=get) + runner.oc = Mock(return_value={"items": pods}) + runner.readiness() + self.assertEqual(len(runner.image_evidence), 3) + self.assertEqual( + runner.image_evidence[0]["containers"][0]["imageID"], "repo@sha256:digest" + ) + pods[0]["metadata"]["ownerReferences"][0]["uid"] = "wrong-rs" + with self.assertRaises(m.AcceptanceError): + runner.readiness() + + def test_plan_only_does_not_contact_cluster(self): + with patch.object(m.Runner, "run") as run: + self.assertEqual( + m.main( + [ + "--context", + "test", + "--namespace", + "pulse-test", + "--pulse", + "pulse", + ] + ), + 0, + ) + run.assert_not_called() + + def test_missing_oc_writes_failed_report(self): + with ( + tempfile.TemporaryDirectory() as folder, + patch.object(m.shutil, "which", return_value=None), + ): + report = Path(folder) / "report.json" + rc = m.main( + [ + "--context", + "test", + "--namespace", + "pulse-test", + "--pulse", + "pulse", + "--execute", + "--report", + str(report), + ] + ) + self.assertEqual(rc, 1) + evidence = json.loads(report.read_text()) + self.assertEqual(evidence["status"], "failed") + self.assertIn("oc executable not found", evidence["checks"][0]["detail"]) + self.assertEqual(evidence["acceptance"], "failed") + + def test_failure_aborts_later_checks(self): + runner = self.runner() + runner.prerequisites = Mock(side_effect=m.AcceptanceError("forbidden")) + runner.readiness = Mock() + with self.assertRaises(m.AcceptanceError): + runner.run() + runner.readiness.assert_not_called() + self.assertEqual(runner.checks[0]["status"], "failed") + + +if __name__ == "__main__": + unittest.main()