"""Run a fail-closed, read-only acceptance pass against a deployed runtime. This runner is intended for a customer-owned staging URL. It never creates a tenant, changes a Control Pack, sends email or prints credentials. Mutating pilot steps remain explicit operator actions because they require customer approval and a customer-controlled signing key. """ from __future__ import annotations import argparse import json import os import statistics import ssl import sys import time from pathlib import Path from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen def _percentile(values, fraction): ordered = sorted(values) return ordered[min(len(ordered) - 1, int((len(ordered) - 1) * fraction))] def _request(base_url, path, admin_key=None, timeout=15): headers = {"Accept": "application/json"} if admin_key: headers["X-Admin-Key"] = admin_key started = time.perf_counter_ns() try: request = Request(base_url.rstrip("/") + path, headers=headers) with urlopen(request, timeout=timeout, context=ssl.create_default_context()) as response: body = response.read() try: payload = json.loads(body.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError): payload = {"raw_length": len(body)} return {"status": response.status, "body": payload, "latency_ms": (time.perf_counter_ns() - started) / 1_000_000} except HTTPError as exc: try: raw = exc.read() body = json.loads(raw.decode("utf-8")) except (OSError, UnicodeDecodeError, json.JSONDecodeError): body = {} return {"status": exc.code, "body": body, "error": f"HTTP {exc.code}", "latency_ms": (time.perf_counter_ns() - started) / 1_000_000} except (URLError, TimeoutError, ValueError, OSError) as exc: return {"status": None, "error": type(exc).__name__, "latency_ms": (time.perf_counter_ns() - started) / 1_000_000} def _check(name, status, expected, detail, metrics=None, evidence=None): return {"name": name, "status": status, "expected": expected, "detail": detail, "metrics": metrics or {}, "evidence": evidence or []} def _latency_probe(base_url, path, iterations): samples, errors = [], 0 for _ in range(iterations): result = _request(base_url, path) samples.append(result["latency_ms"]) if result.get("status") != 200: errors += 1 return { "iterations": iterations, "completed": iterations - errors, "errors": errors, "p50_ms": round(statistics.median(samples), 4), "p95_ms": round(_percentile(samples, .95), 4), "p99_ms": round(_percentile(samples, .99), 4), "throughput_per_sec": round(1000 / max(statistics.fmean(samples), .0001), 2), } def run(base_url, iterations=20, admin_key=None): checks = [] health = _request(base_url, "/healthz") checks.append(_check("process_liveness", "PASS" if health.get("status") == 200 else "FAIL", "HTTP 200", f"healthz returned HTTP {health.get('status')}", {"single_request_ms": round(health["latency_ms"], 4)}, ["health response"])) ready = _request(base_url, "/readyz") ready_body = ready.get("body") or {} ready_ok = ready.get("status") == 200 and ready_body.get("ok") is True if ready_ok: ready_state = "PASS" elif ready.get("status") == 503 and isinstance(ready_body.get("checks"), dict): ready_state = "PENDING_ENVIRONMENT_CONFIG" else: ready_state = "FAIL" checks.append(_check("deployment_readiness", ready_state, "HTTP 200 with ok=true", f"readyz returned HTTP {ready.get('status')}; checks are boolean only", {"single_request_ms": round(ready["latency_ms"], 4)}, ["readiness response", "boolean configuration checks"])) evidence = _request(base_url, "/evidence-pack/enterprise-validation-suite.json") ev_body = evidence.get("body") or {} ev = ev_body.get("evidence") or {} ev_ok = (evidence.get("status") == 200 and ev.get("scenario_count") == 20 and ev.get("failed") == 0 and not ev.get("critical_failures")) checks.append(_check("published_validation_evidence", "PASS" if ev_ok else "FAIL", "20 scenarios, zero failures and zero critical failures", f"EvidencePack returned HTTP {evidence.get('status')}; report status={ev.get('status', 'unavailable')}", {"single_request_ms": round(evidence["latency_ms"], 4)}, ["enterprise validation EvidencePack", "scenario matrix"])) openapi = _request(base_url, "/integration/openapi.yaml") openapi_ok = openapi.get("status") == 200 checks.append(_check("integration_contract_available", "PASS" if openapi_ok else "FAIL", "HTTP 200", f"OpenAPI contract returned HTTP {openapi.get('status')}", {"single_request_ms": round(openapi["latency_ms"], 4)}, ["OpenAPI contract"])) headers = _request(base_url, "/healthz") # urllib does not retain headers in the compact request helper; keep this # check explicit rather than implying that a header was measured. checks.append(_check("security_headers", "PENDING_OPERATOR_HEADER_SCAN", "CSP, HSTS (HTTPS), frame, referrer and content-type protections", "Run the browser/security-header scan in the customer environment; this read-only runner does not infer headers.", evidence=["header scan required"])) latency = _latency_probe(base_url, "/healthz", iterations) checks.append(_check("liveness_latency", "PASS" if latency["errors"] == 0 else "FAIL", f"{iterations}/{iterations} successful responses", "Repeated liveness timing; not an application SLO", latency, ["latency sample set"])) for name, path in (("operator_audit_access", "/api/admin/audit-events?limit=1"), ("tenant_catalog_access", "/api/admin/integrations")): if not admin_key: checks.append(_check(name, "PENDING_OPERATOR_CREDENTIAL", "Authenticated operator response", "No admin credential supplied; no request was attempted.", evidence=["operator credential required"])) else: result = _request(base_url, path, admin_key=admin_key) passed = result.get("status") == 200 checks.append(_check(name, "PASS" if passed else "FAIL", "HTTP 200 with bounded response", f"authenticated endpoint returned HTTP {result.get('status')}", {"single_request_ms": round(result["latency_ms"], 4)}, [path])) checks.extend([ _check("customer_tenant_isolation", "PENDING_MUTATING_PILOT", "Cross-tenant read/write denied in customer database", "Requires two customer tenants and approved test identities; not run by this read-only pass.", evidence=["cross-tenant negative test", "database isolation result"]), _check("runtime_kms_storage_smoke", "PENDING_RUNTIME_IDENTITY", "Customer runtime role can put/get/verify an EvidencePack", "Requires the deployed application identity and customer KMS/object store; operator identity must not be substituted.", evidence=["runtime-role storage smoke", "KMS verification"]), _check("backup_restore_rto_rpo", "PENDING_CUSTOMER_DRILL", "Restore meets customer-approved RTO/RPO", "Requires customer backup policy, restore rehearsal and written objectives.", evidence=["restore log", "RTO/RPO measurement"]), _check("bounded_shadow_pilot", "PENDING_CUSTOMER_PILOT", "One named workflow completes shadow → change → replay/review → signed EvidencePack", "Requires customer workflow owner, controls, reviewers, data boundary and signing key.", evidence=["pilot brief", "change event", "signed EvidencePack", "go/no-go"]), ]) failures = [item["name"] for item in checks if item["status"] == "FAIL"] pending = [item["name"] for item in checks if item["status"].startswith("PENDING")] return { "report_type": "EcoKure customer staging acceptance report", "verification_state": "READ_ONLY_DEPLOYED_RUNTIME_PROBE", "base_url": base_url, "generated_at_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "status": "PASS" if not failures and not pending else ("FAIL" if failures else "REVIEW_REQUIRED"), "checks": checks, "passed": sum(item["status"] == "PASS" for item in checks), "failed": len(failures), "pending": len(pending), "failures": failures, "pending_gates": pending, "claim_boundary": [ "Read-only acceptance probe; it does not create tenants or mutate customer state.", "A green read-only probe is necessary, not sufficient, for production approval.", "Performance values are diagnostic and are not a customer SLO.", "Customer tenant isolation, runtime KMS storage, restore objectives and pilot evidence remain explicit gates.", ], } def main(): parser = argparse.ArgumentParser(description="Run a read-only customer staging acceptance probe") parser.add_argument("--base-url", required=True) parser.add_argument("--iterations", type=int, default=20) parser.add_argument("--output", type=Path, required=True) parser.add_argument("--admin-key-env", default="ECOKURE_ADMIN_KEY", help="Environment variable containing an operator key; value is never printed") parser.add_argument("--strict", action="store_true", help="Exit non-zero for failures or pending customer gates") args = parser.parse_args() if not 1 <= args.iterations <= 200: raise SystemExit("--iterations must be between 1 and 200") report = run(args.base_url, args.iterations, os.environ.get(args.admin_key_env, "").strip() or None) args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8") print(json.dumps({"status": report["status"], "passed": report["passed"], "failed": report["failed"], "pending": report["pending"], "output": str(args.output)}, indent=2)) if args.strict and report["status"] != "PASS": raise SystemExit(1) if __name__ == "__main__": main()