#!/usr/bin/env python3 from __future__ import annotations import argparse import json import sys from pathlib import Path from typing import Any DEFAULT_MAX_ADMISSION_LATENCY_MS = 1000 DEFAULT_MAX_ENQUEUE_P95_MS = 1000 DEFAULT_MAX_DRAIN_SECONDS = 600 FORBIDDEN_KEY_FRAGMENTS = ("secret", "token", "password", "payload", "body", "authorization", "cookie") class GateError(Exception): pass def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Validate sanitized HubSpot connector pressure regression evidence for #3337." ) parser.add_argument("--report", required=True, help="Path to the sanitized pressure-gate JSON report") parser.add_argument("--max-admission-latency-ms", type=int, default=DEFAULT_MAX_ADMISSION_LATENCY_MS) parser.add_argument("--max-enqueue-p95-ms", type=int, default=DEFAULT_MAX_ENQUEUE_P95_MS) parser.add_argument("--max-drain-seconds", type=int, default=DEFAULT_MAX_DRAIN_SECONDS) return parser.parse_args() def load_report(path: Path) -> dict[str, Any]: try: data = json.loads(path.read_text()) except json.JSONDecodeError as exc: raise GateError(f"{path} is not valid JSON: {exc}") from exc if not isinstance(data, dict): raise GateError("pressure gate report must be a JSON object") return data def reject_sensitive_keys(value: Any, *, path: str = "$") -> None: if isinstance(value, dict): for key, child in value.items(): key_lower = str(key).lower() if any(fragment in key_lower for fragment in FORBIDDEN_KEY_FRAGMENTS): raise GateError(f"report contains forbidden sensitive/payload-like key at {path}.{key}") reject_sensitive_keys(child, path=f"{path}.{key}") elif isinstance(value, list): for index, child in enumerate(value): reject_sensitive_keys(child, path=f"{path}[{index}]") def as_number(value: Any, *, field: str) -> float: if isinstance(value, bool) or not isinstance(value, (int, float)): raise GateError(f"{field} must be numeric") return float(value) def require_bool(value: Any, *, field: str) -> bool: if not isinstance(value, bool): raise GateError(f"{field} must be boolean") return value def check_admission(report: dict[str, Any], *, max_latency_ms: int) -> list[str]: admission = report.get("webhook_admission") if not isinstance(admission, dict): raise GateError("missing webhook_admission object") samples = admission.get("samples") if not isinstance(samples, list) or not samples: raise GateError("webhook_admission.samples must be a non-empty list") durable_admitted = 0 for index, sample in enumerate(samples): if not isinstance(sample, dict): raise GateError(f"webhook_admission.samples[{index}] must be an object") status = sample.get("status") contract = sample.get("contract") retry_after_present = require_bool( sample.get("retry_after_present"), field=f"webhook_admission.samples[{index}].retry_after_present", ) if retry_after_present or "retry_after_seconds" in sample: raise GateError( f"webhook_admission.samples[{index}] must not expose Retry-After" ) latency_ms = as_number(sample.get("latency_ms"), field=f"webhook_admission.samples[{index}].latency_ms") if latency_ms > max_latency_ms: raise GateError( f"webhook_admission.samples[{index}] latency {latency_ms:g}ms exceeds {max_latency_ms}ms" ) if status == 202 and contract == "durable_admit": durable_admitted += 1 else: raise GateError( f"webhook_admission.samples[{index}] must be durable 202 admission, " f"got status={status!r} contract={contract!r}" ) return [ f"webhook_admission bounded: samples={len(samples)} " f"durable_admit_202={durable_admitted} " f"max_latency_ms={max_latency_ms}" ] def check_workflow_action_burst( report: dict[str, Any], *, max_enqueue_p95_ms: int, max_drain_seconds: int ) -> list[str]: burst = report.get("workflow_action_burst") if not isinstance(burst, dict): raise GateError("missing workflow_action_burst object") expected = int(as_number(burst.get("expected_callbacks"), field="workflow_action_burst.expected_callbacks")) if expected <= 0: raise GateError("workflow_action_burst.expected_callbacks must be positive") enqueue = burst.get("enqueue_latency") drain = burst.get("worker_drain") campaign_member = burst.get("campaign_member") rollup = burst.get("rollup") for field, value in ( ("enqueue_latency", enqueue), ("worker_drain", drain), ("campaign_member", campaign_member), ("rollup", rollup), ): if not isinstance(value, dict): raise GateError(f"workflow_action_burst.{field} must be an object") p95_ms = as_number(enqueue.get("p95_ms"), field="workflow_action_burst.enqueue_latency.p95_ms") if p95_ms > max_enqueue_p95_ms: raise GateError(f"workflow_action_burst enqueue p95 {p95_ms:g}ms exceeds {max_enqueue_p95_ms}ms") completed = int(as_number(drain.get("completed"), field="workflow_action_burst.worker_drain.completed")) drain_seconds = as_number(drain.get("drain_seconds"), field="workflow_action_burst.worker_drain.drain_seconds") if completed != expected: raise GateError(f"workflow_action_burst drained {completed}/{expected} callbacks") if drain_seconds > max_drain_seconds: raise GateError(f"workflow_action_burst drain {drain_seconds:g}s exceeds {max_drain_seconds}s") member_count = int(as_number(campaign_member.get("actual_members"), field="workflow_action_burst.campaign_member.actual_members")) duplicates = int(as_number(campaign_member.get("duplicates"), field="workflow_action_burst.campaign_member.duplicates")) wrong_campaign = int(as_number(campaign_member.get("wrong_campaign"), field="workflow_action_burst.campaign_member.wrong_campaign")) if member_count != expected or duplicates != 0 or wrong_campaign != 0: raise GateError( "workflow_action_burst CampaignMember correctness failed: " f"actual_members={member_count} expected={expected} duplicates={duplicates} wrong_campaign={wrong_campaign}" ) if not require_bool(rollup.get("correct"), field="workflow_action_burst.rollup.correct"): raise GateError("workflow_action_burst rollup correctness failed") return [ f"workflow_action_burst enqueue bounded: expected={expected} p95_ms={p95_ms:g} max_p95_ms={max_enqueue_p95_ms}", f"workflow_action_burst drain complete: completed={completed} drain_seconds={drain_seconds:g}", f"workflow_action_burst CampaignMember correct: members={member_count} duplicates=0 wrong_campaign=0", "workflow_action_burst rollup correct", ] def evaluate(report: dict[str, Any], *, max_admission_latency_ms: int, max_enqueue_p95_ms: int, max_drain_seconds: int) -> list[str]: reject_sensitive_keys(report) messages: list[str] = [] messages.extend(check_admission(report, max_latency_ms=max_admission_latency_ms)) messages.extend( check_workflow_action_burst( report, max_enqueue_p95_ms=max_enqueue_p95_ms, max_drain_seconds=max_drain_seconds, ) ) return messages def main() -> int: args = parse_args() try: messages = evaluate( load_report(Path(args.report)), max_admission_latency_ms=args.max_admission_latency_ms, max_enqueue_p95_ms=args.max_enqueue_p95_ms, max_drain_seconds=args.max_drain_seconds, ) except GateError as exc: print(f"BLOCK: {exc}", file=sys.stderr) return 1 for message in messages: print(f"PASS: {message}") return 0 if __name__ == "__main__": raise SystemExit(main())