stripe-webhook-verify-and-parse

copy

import sys
import os
import subprocess
import time
import urllib.request
import json

# ─────────────────────────────────────────
# PRE_EXECUTION
# ─────────────────────────────────────────

for attempt in range(2):
    try:
        req = urllib.request.Request(
            "https://checklist.day/api/registry/stripe",
            headers={"User-Agent": "checklist-agent/1.0"}
        )
        with urllib.request.urlopen(req, timeout=10) as resp:
            registry = json.loads(resp.read())
            break
    except Exception as e:
        if attempt == 1:
            print(f"ABORT: registry unreachable — {e}")
            sys.exit(1)
        time.sleep(2)

warnings = registry.get("warnings", [])
if warnings:
    print("[stripe] WARNINGS:")
    for w in warnings if isinstance(warnings, list) else [warnings]:
        print(f"  ⚠ {w}")

# ─────────────────────────────────────────
# EXECUTION
# ─────────────────────────────────────────

subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", "stripe>=7.0.0"])

import stripe
SignatureVerificationError = stripe.SignatureVerificationError

STRIPE_SECRET_KEY     = os.environ.get("STRIPE_SECRET_KEY")
STRIPE_WEBHOOK_SECRET = os.environ.get("STRIPE_WEBHOOK_SECRET")

if not STRIPE_SECRET_KEY:
    print("ABORT: STRIPE_SECRET_KEY not set"); sys.exit(1)
if not STRIPE_SECRET_KEY.startswith("sk_test_"):
    print("ABORT: use test mode key"); sys.exit(1)
if not STRIPE_WEBHOOK_SECRET:
    print("ABORT: STRIPE_WEBHOOK_SECRET not set (whsec_...)"); sys.exit(1)

stripe.api_key = STRIPE_SECRET_KEY

# Build a realistic webhook payload
ts        = int(time.time())
payload   = json.dumps({
    "id":      f"evt_test_{ts}",
    "type":    "payment_intent.succeeded",
    "object":  "event",
    "data":    {"object": {"id": "pi_test", "amount": 2000, "currency": "usd"}},
    "created": ts,
}).encode("utf-8")

# Generate valid signature
# FOOTGUN: Stripe signs the raw bytes, not the parsed JSON — never re-serialize before verifying
signed_header = stripe.WebhookSignature._compute_signature(
    f"{ts}.{payload.decode()}", STRIPE_WEBHOOK_SECRET
)
valid_header = f"t={ts},v1={signed_header}"

# 1. Verify valid signature
valid_sig_ok = False
event_type   = None
try:
    event        = stripe.Webhook.construct_event(payload, valid_header, STRIPE_WEBHOOK_SECRET)
    valid_sig_ok = True
    event_type   = event["type"]
    print(f"  valid signature: OK — event type={event_type}")
except SignatureVerificationError as e:
    print(f"  FAIL: valid sig rejected — {e}")

# 2. Verify tampered payload is rejected
# FOOTGUN: never trust payload without verifying signature first
invalid_sig_blocked = False
tampered_payload    = payload + b" tampered"
try:
    stripe.Webhook.construct_event(tampered_payload, valid_header, STRIPE_WEBHOOK_SECRET)
    print(f"  FAIL: tampered payload was accepted (should have been rejected)")
except SignatureVerificationError:
    invalid_sig_blocked = True
    print(f"  tampered payload rejected: OK")

# ─────────────────────────────────────────
# POST_EXECUTION
# ─────────────────────────────────────────

assert valid_sig_ok, "FAIL: valid signature was rejected"
assert invalid_sig_blocked, "FAIL: tampered payload was not rejected"
assert event_type == "payment_intent.succeeded", f"FAIL: unexpected event type {event_type}"

result = {
    "valid_sig_ok":        valid_sig_ok,
    "invalid_sig_blocked": invalid_sig_blocked,
    "event_type":          event_type,
}
print(json.dumps(result, indent=2))
print("PASS")