Webhook Handling skill for Claude Code

#!/usr/bin/env python3
"""Sign, verify, and send test webhooks for common signature schemes.

Schemes:
  standard  Standard Webhooks (webhook-id / webhook-timestamp / webhook-signature, v1,<b64>)
  stripe    Stripe-Signature: t=<ts>,v1=<hex HMAC-SHA256 of "<ts>.<body>">
  github    X-Hub-Signature-256: sha256=<hex HMAC-SHA256 of body>
  slack     X-Slack-Signature: v0=<hex HMAC-SHA256 of "v0:<ts>:<body>">, X-Slack-Request-Timestamp
  shopify   X-Shopify-Hmac-Sha256: <base64 HMAC-SHA256 of body>

Usage:
  python3 webhook_sig.py sign   --scheme stripe --secret whsec_x --body-file event.json
  python3 webhook_sig.py verify --scheme github --secret s3cr3t --body-file body.bin \
        --header "X-Hub-Signature-256=sha256=abc..."
  python3 webhook_sig.py send   --scheme standard --secret whsec_BASE64 --body-file event.json \
        --url http://localhost:3000/webhooks/test

Exit codes: 0 success/valid, 1 invalid signature or HTTP error, 2 usage error.
Standard library only.
"""
from __future__ import annotations

import argparse
import base64
import hashlib
import hmac
import sys
import time
import urllib.error
import urllib.request
import uuid


def _hmac(key: bytes, msg: bytes) -> bytes:
    return hmac.new(key, msg, hashlib.sha256).digest()


def _standard_key(secret: str) -> bytes:
    raw = secret[len("whsec_"):] if secret.startswith("whsec_") else secret
    try:
        return base64.b64decode(raw, validate=True)
    except Exception:
        sys.exit("standard scheme expects a base64 secret, optionally prefixed with whsec_")


def sign(scheme: str, secret: str, body: bytes, ts: int, msg_id: str) -> dict[str, str]:
    if scheme == "standard":
        content = f"{msg_id}.{ts}.".encode() + body
        sig = base64.b64encode(_hmac(_standard_key(secret), content)).decode()
        return {"webhook-id": msg_id, "webhook-timestamp": str(ts), "webhook-signature": f"v1,{sig}"}
    if scheme == "stripe":
        sig = _hmac(secret.encode(), f"{ts}.".encode() + body).hex()
        return {"Stripe-Signature": f"t={ts},v1={sig}"}
    if scheme == "github":
        return {"X-Hub-Signature-256": "sha256=" + _hmac(secret.encode(), body).hex()}
    if scheme == "slack":
        sig = _hmac(secret.encode(), f"v0:{ts}:".encode() + body).hex()
        return {"X-Slack-Request-Timestamp": str(ts), "X-Slack-Signature": f"v0={sig}"}
    if scheme == "shopify":
        return {"X-Shopify-Hmac-Sha256": base64.b64encode(_hmac(secret.encode(), body)).decode()}
    sys.exit(f"unknown scheme {scheme}")


def verify(scheme: str, secret: str, body: bytes, headers: dict[str, str], tolerance: int) -> tuple[bool, str]:
    h = {k.lower(): v for k, v in headers.items()}
    now = int(time.time())

    def fresh(ts_str: str) -> bool:
        try:
            return abs(now - int(ts_str)) <= tolerance
        except ValueError:
            return False

    if scheme == "standard":
        mid, ts, sigs = h.get("webhook-id"), h.get("webhook-timestamp"), h.get("webhook-signature", "")
        if not (mid and ts):
            return False, "missing webhook-id or webhook-timestamp"
        if not fresh(ts):
            return False, f"timestamp outside tolerance of {tolerance}s"
        expected = sign(scheme, secret, body, int(ts), mid)["webhook-signature"].split(",", 1)[1]
        for part in sigs.split():                       # space-separated list, e.g. "v1,abc v1,def"
            version, _, value = part.partition(",")
            if version == "v1" and hmac.compare_digest(value, expected):
                return True, "valid"
        return False, "no matching v1 signature"
    if scheme == "stripe":
        items = [p.split("=", 1) for p in h.get("stripe-signature", "").split(",") if "=" in p]
        ts = next((v for k, v in items if k == "t"), None)
        if not ts:
            return False, "missing t= in Stripe-Signature"
        if not fresh(ts):
            return False, f"timestamp outside tolerance of {tolerance}s"
        expected = _hmac(secret.encode(), f"{ts}.".encode() + body).hex()
        ok = any(k == "v1" and hmac.compare_digest(v, expected) for k, v in items)
        return ok, "valid" if ok else "no matching v1 signature"
    if scheme == "github":
        got = h.get("x-hub-signature-256", "")
        ok = hmac.compare_digest(got, sign(scheme, secret, body, 0, "")["X-Hub-Signature-256"])
        return ok, "valid" if ok else "signature mismatch"
    if scheme == "slack":
        ts = h.get("x-slack-request-timestamp", "")
        if not fresh(ts):
            return False, f"timestamp outside tolerance of {tolerance}s"
        ok = hmac.compare_digest(h.get("x-slack-signature", ""), sign(scheme, secret, body, int(ts), "")["X-Slack-Signature"])
        return ok, "valid" if ok else "signature mismatch"
    if scheme == "shopify":
        ok = hmac.compare_digest(h.get("x-shopify-hmac-sha256", ""), sign(scheme, secret, body, 0, "")["X-Shopify-Hmac-Sha256"])
        return ok, "valid" if ok else "signature mismatch"
    return False, f"unknown scheme {scheme}"


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("action", choices=["sign", "verify", "send"])
    ap.add_argument("--scheme", required=True, choices=["standard", "stripe", "github", "slack", "shopify"])
    ap.add_argument("--secret", required=True)
    ap.add_argument("--body-file", required=True, help="exact raw bytes of the payload")
    ap.add_argument("--timestamp", type=int, default=None, help="unix seconds (default: now)")
    ap.add_argument("--id", default=None, help="message id for the standard scheme")
    ap.add_argument("--header", action="append", default=[], help="Name=Value, repeatable (verify)")
    ap.add_argument("--tolerance", type=int, default=300, help="max clock skew in seconds")
    ap.add_argument("--url", help="target URL (send)")
    args = ap.parse_args()

    body = open(args.body_file, "rb").read()
    ts = args.timestamp if args.timestamp is not None else int(time.time())
    msg_id = args.id or f"msg_{uuid.uuid4().hex}"

    if args.action == "sign":
        for k, v in sign(args.scheme, args.secret, body, ts, msg_id).items():
            print(f"{k}: {v}")
        return 0

    if args.action == "verify":
        headers = {}
        for item in args.header:
            if "=" not in item:
                print(f"bad --header {item!r}, expected Name=Value", file=sys.stderr)
                return 2
            k, v = item.split("=", 1)
            headers[k] = v
        ok, reason = verify(args.scheme, args.secret, body, headers, args.tolerance)
        print(("VALID: " if ok else "INVALID: ") + reason)
        return 0 if ok else 1

    if not args.url:
        print("--url is required for send", file=sys.stderr)
        return 2
    headers = {"Content-Type": "application/json", **sign(args.scheme, args.secret, body, ts, msg_id)}
    req = urllib.request.Request(args.url, data=body, headers=headers, method="POST")
    try:
        with urllib.request.urlopen(req, timeout=15) as resp:
            print(f"HTTP {resp.status}")
            print(resp.read(2000).decode(errors="replace"))
            return 0 if 200 <= resp.status < 300 else 1
    except urllib.error.HTTPError as e:
        print(f"HTTP {e.code}")
        print(e.read(2000).decode(errors="replace"))
        return 1
    except urllib.error.URLError as e:
        print(f"request failed: {e.reason}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    sys.exit(main())