#!/usr/bin/env python3 """Collect sanitized Kaiju Coder 7 paid API launch evidence. This script helps fill release/paid-api-launch-evidence.json without storing API keys, secret values, full prompts, or model responses. It is preview-only by default; pass --write to update the evidence file. """ from __future__ import annotations import argparse import json import os import re import statistics import sys import time import urllib.error import urllib.request import uuid from datetime import datetime, timezone from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] DEFAULT_OUT = ROOT / "release/paid-api-launch-evidence.json" MODEL_ID = "kaiju-coder-7" DEFAULT_ROUTE = "/v1/chat/completions" DEFAULT_USER_AGENT = "KaijuCoder7LaunchEvidence/1.0" SECRET_PATTERNS = [ ("openai_api_key", re.compile(r"\bsk-[A-Za-z0-9][A-Za-z0-9_-]{20,}\b")), ("anthropic_api_key", re.compile(r"\bsk-ant-[A-Za-z0-9_-]{20,}\b")), ("stripe_secret_key", re.compile(r"\b[rs]k_(?:live|test)_[A-Za-z0-9]{16,}\b")), ("stripe_webhook_secret", re.compile(r"\bwhsec_[A-Za-z0-9]{16,}\b")), ("huggingface_token", re.compile(r"\bhf_[A-Za-z0-9]{20,}\b")), ("github_token", re.compile(r"\b(?:ghp_[A-Za-z0-9]{20,}|github_pat_[A-Za-z0-9_]{22,})\b")), ("google_api_key", re.compile(r"\bAIza[0-9A-Za-z_-]{20,}\b")), ("bearer_token", re.compile(r"\bBearer\s+[A-Za-z0-9._~+/-]{24,}={0,2}\b", re.IGNORECASE)), ("private_key_block", re.compile(r"-----BEGIN (?:RSA |OPENSSH |EC |DSA )?PRIVATE KEY-----")), ] def utc_now() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def load_existing(path: Path) -> dict[str, Any]: if not path.is_file(): return {} return json.loads(path.read_text(encoding="utf-8")) def secret_findings(text: str) -> list[str]: return sorted({label for label, pattern in SECRET_PATTERNS if pattern.search(text)}) def assert_sanitized(payload: dict[str, Any]) -> None: rendered = json.dumps(payload, sort_keys=True) findings = secret_findings(rendered) if findings: raise SystemExit("Refusing to write secret-looking evidence: " + ", ".join(findings)) def api_url(base_url: str, path: str) -> str: return base_url.rstrip("/") + path def request_json(url: str, payload: dict[str, Any], api_key: str, request_id: str, timeout: int) -> tuple[int, str, float]: body = json.dumps(payload).encode("utf-8") request = urllib.request.Request( url, data=body, method="POST", headers={ "authorization": f"Bearer {api_key}", "content-type": "application/json", "user-agent": DEFAULT_USER_AGENT, "x-request-id": request_id, }, ) start = time.perf_counter() try: with urllib.request.urlopen(request, timeout=timeout) as response: content_type = response.headers.get("content-type", "") response.read() return response.status, content_type, (time.perf_counter() - start) * 1000 except urllib.error.HTTPError as exc: exc.read() return exc.code, exc.headers.get("content-type", ""), (time.perf_counter() - start) * 1000 def probe_health(base_url: str, timeout: int) -> tuple[int, float] | None: start = time.perf_counter() try: request = urllib.request.Request( api_url(base_url, "/health"), headers={"user-agent": DEFAULT_USER_AGENT}, ) with urllib.request.urlopen(request, timeout=timeout) as response: response.read() return response.status, (time.perf_counter() - start) * 1000 except Exception: return None def percentile_95(values: list[float]) -> float: if len(values) == 1: return values[0] try: return statistics.quantiles(values, n=20, method="inclusive")[18] except statistics.StatisticsError: return max(values) def run_staging_samples(args: argparse.Namespace) -> tuple[dict[str, Any] | None, dict[str, Any] | None]: if args.skip_live_request: return None, None if not args.api_base_url: raise SystemExit("--api-base-url is required unless --skip-live-request is set") api_key = os.environ.get(args.api_key_env) if not api_key: raise SystemExit(f"{args.api_key_env} is not set; refusing to read API keys from arguments") latencies: list[float] = [] first_request_id = "" first_status = 0 first_streamed = False url = api_url(args.api_base_url, DEFAULT_ROUTE) sample_count = max(args.live_samples, 1) for index in range(sample_count): request_id = f"kaiju-paid-staging-{uuid.uuid4()}" payload = { "model": MODEL_ID, "stream": True, "max_tokens": 48, "messages": [ { "role": "user", "content": "Return a short Kaiju Coder 7 paid API staging smoke response.", } ], } status, content_type, latency_ms = request_json(url, payload, api_key, request_id, args.timeout) if index == 0: first_request_id = request_id first_status = status first_streamed = "event-stream" in content_type.lower() if status == 200: latencies.append(latency_ms) request_evidence = { "status": "pass" if first_status == 200 and first_streamed else "pending", "checked_at": utc_now(), "route": DEFAULT_ROUTE, "model": MODEL_ID, "http_status": first_status, "streamed": first_streamed, "request_id": first_request_id, } latency_evidence = { "status": "pass" if len(latencies) >= 5 else "pending", "checked_at": utc_now(), "route": DEFAULT_ROUTE, "sample_count": len(latencies), "p95_ms": round(percentile_95(latencies), 2) if latencies else 0, "max_acceptable_ms": args.max_acceptable_ms, } return request_evidence, latency_evidence def add_optional_manual_evidence(evidence: dict[str, Any], args: argparse.Namespace) -> None: checked_at = args.checked_at or utc_now() if args.public_route_ok: health = probe_health(args.api_base_url, args.timeout) if args.api_base_url else None exposure_mode = args.public_route_mode evidence["public_route_mode"] = { "status": "pass", "checked_at": checked_at, "exposure_mode": exposure_mode, "route": args.api_base_url, "result": f"{exposure_mode} route resolves to the intended Kaiju Worker" + (f"; /health={health[0]} in {health[1]:.0f}ms" if health else ""), } if args.wrangler_secret_name: evidence["wrangler_secrets_verified"] = { "status": "pass", "checked_at": checked_at, "command": "wrangler secret list", "observed_names": sorted(set(args.wrangler_secret_name)), } if args.d1_migration_result: evidence["d1_migration_applied"] = { "status": "pass", "checked_at": checked_at, "command": args.d1_migration_command, "migration": "0001_paid_api.sql", "result": args.d1_migration_result, } if args.stripe_checkout_topup_pass: evidence["stripe_checkout_topup_staging"] = { "status": "pass", "checked_at": checked_at, "mode": args.stripe_mode, "webhook_event": "checkout.session.completed", "credited_api_key_id": args.credited_api_key_id, "idempotency_checked": args.stripe_idempotency_checked, } if args.rollback_result: evidence["rollback_exercised"] = { "status": "pass", "checked_at": checked_at, "command": args.rollback_command, "result": args.rollback_result, } if args.staging_request_id: evidence["worker_to_gojira_staging_request"] = { "status": "pass", "checked_at": checked_at, "route": DEFAULT_ROUTE, "model": MODEL_ID, "http_status": args.staging_http_status, "streamed": args.staging_streamed, "request_id": args.staging_request_id, } if args.paid_route_p95_ms is not None: evidence["paid_route_latency"] = { "status": "pass", "checked_at": checked_at, "route": DEFAULT_ROUTE, "sample_count": args.paid_route_sample_count, "p95_ms": args.paid_route_p95_ms, "max_acceptable_ms": args.max_acceptable_ms, } def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--out", type=Path, default=DEFAULT_OUT) parser.add_argument("--write", action="store_true", help="Write the evidence file. Default is preview only.") parser.add_argument("--merge-existing", action="store_true", help="Merge with existing evidence at --out.") parser.add_argument("--checked-at", help="Override checked_at timestamp for manual evidence.") parser.add_argument("--api-base-url", default="", help="Public paid API base URL, for example https://api.example.com.") parser.add_argument("--api-key-env", default="KAIJU_PAID_API_KEY", help="Environment variable containing the staging API key.") parser.add_argument("--timeout", type=int, default=120) parser.add_argument("--skip-live-request", action="store_true", help="Do not call the paid API.") parser.add_argument("--live-samples", type=int, default=5) parser.add_argument("--max-acceptable-ms", type=float, default=120_000) parser.add_argument("--public-route-ok", action="store_true", help="Record public custom-domain route evidence.") parser.add_argument( "--public-route-mode", choices=["workers_dev", "custom_domain"], default="workers_dev", help="Public route type used for the launch evidence.", ) parser.add_argument("--wrangler-secret-name", action="append", default=[], help="Observed Wrangler secret name. Repeatable.") parser.add_argument("--d1-migration-result", choices=["success", "already_applied"]) parser.add_argument( "--d1-migration-command", default="wrangler d1 migrations apply KAIJU_BILLING_DB --remote", ) parser.add_argument("--stripe-checkout-topup-pass", action="store_true") parser.add_argument("--stripe-mode", default="test") parser.add_argument("--credited-api-key-id", default="key_staging_001") parser.add_argument("--stripe-idempotency-checked", action="store_true") parser.add_argument("--rollback-result", choices=["success"]) parser.add_argument("--rollback-command", default="wrangler rollback") parser.add_argument("--staging-request-id", help="Sanitized request id from a separate staging request.") parser.add_argument("--staging-http-status", type=int, default=200) parser.add_argument("--staging-streamed", action="store_true") parser.add_argument("--paid-route-p95-ms", type=float) parser.add_argument("--paid-route-sample-count", type=int, default=5) return parser.parse_args() def main() -> int: args = parse_args() evidence = load_existing(args.out) if args.merge_existing else {} add_optional_manual_evidence(evidence, args) request_evidence, latency_evidence = run_staging_samples(args) if request_evidence: evidence["worker_to_gojira_staging_request"] = request_evidence if latency_evidence: evidence["paid_route_latency"] = latency_evidence assert_sanitized(evidence) rendered = json.dumps(evidence, indent=2, sort_keys=True) + "\n" if args.write: args.out.parent.mkdir(parents=True, exist_ok=True) args.out.write_text(rendered, encoding="utf-8") print(f"Wrote sanitized paid API evidence to {args.out}") else: print(rendered, end="") print("Preview only. Pass --write to update the evidence file.", file=sys.stderr) return 0 if __name__ == "__main__": raise SystemExit(main())