Webhook Receivers skill for Codex

Gives the Codex app a display name, a one-line description and a starter prompt for this skill.

interface:
  display_name: "Webhook Receivers"
  short_description: "Verify, store, ack fast, process exactly once"
  default_prompt: "$webhook-receivers Add a Stripe webhook endpoint that verifies signatures and handles retries"

#!/usr/bin/env python3
"""Verify (or produce) webhook signatures offline, using only the standard library.

Supported schemes:
  stripe    Stripe-Signature: t=<ts>,v1=<hex>        signs "<ts>.<body>"
  github    X-Hub-Signature-256: sha256=<hex>       signs "<body>"
  slack     X-Slack-Signature: v0=<hex>             signs "v0:<ts>:<body>" (needs --timestamp)
  shopify   X-Shopify-Hmac-Sha256: <base64>         signs "<body>"
  standard  webhook-signature: v1,<base64> ...       signs "<id>.<ts>.<body>" (needs --id, --timestamp)
            (Standard Webhooks / Svix style; secret "whsec_<base64>")

The secret is read from an environment variable (default WEBHOOK_SECRET) or a file, never
from the command line, so it does not land in shell history.

Exit codes: 0 = signature valid, 1 = invalid or stale, 2 = usage error.
"""
from __future__ import annotations

import argparse
import base64
import hashlib
import hmac
import os
import sys
import time
from typing import NoReturn


def usage_error(msg: str) -> NoReturn:
    print(f"error: {msg}", file=sys.stderr)
    sys.exit(2)


def hmac_sha256(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:
        usage_error("standard-webhooks secret must be base64 (optionally prefixed whsec_)")


def compute(scheme: str, secret: str, body: bytes, ts: str | None, msg_id: str | None) -> str:
    """Return the header value a sender would produce for this body."""
    key = secret.encode()
    if scheme == "stripe":
        assert ts is not None
        return f"t={ts},v1=" + hmac_sha256(key, f"{ts}.".encode() + body).hex()
    if scheme == "github":
        return "sha256=" + hmac_sha256(key, body).hex()
    if scheme == "slack":
        assert ts is not None
        return "v0=" + hmac_sha256(key, f"v0:{ts}:".encode() + body).hex()
    if scheme == "shopify":
        return base64.b64encode(hmac_sha256(key, body)).decode()
    if scheme == "standard":
        assert ts is not None and msg_id is not None
        sig = hmac_sha256(standard_key(secret), f"{msg_id}.{ts}.".encode() + body)
        return "v1," + base64.b64encode(sig).decode()
    usage_error(f"unknown scheme {scheme}")


def candidates(scheme: str, header: str) -> tuple[str | None, list[str]]:
    """Split a received header into (timestamp if embedded, list of signature strings)."""
    header = header.strip()
    if scheme == "stripe":
        parts = dict()
        sigs = []
        for item in header.split(","):
            k, _, v = item.strip().partition("=")
            if k == "v1":
                sigs.append(v)
            else:
                parts[k] = v
        return parts.get("t"), sigs
    if scheme == "standard":
        return None, [s.split(",", 1)[1] for s in header.split() if s.startswith("v1,")]
    if scheme == "github":
        return None, [header.removeprefix("sha256=")]
    if scheme == "slack":
        return None, [header.removeprefix("v0=")]
    return None, [header]


def expected_value(scheme: str, full: str) -> str:
    """Strip the prefix from a computed header so it compares with candidates()."""
    if scheme == "stripe":
        return full.split(",v1=", 1)[1]
    if scheme == "standard":
        return full.split(",", 1)[1]
    if scheme == "github":
        return full.removeprefix("sha256=")
    if scheme == "slack":
        return full.removeprefix("v0=")
    return full


def read_secret(args: argparse.Namespace) -> str:
    if args.secret_file:
        with open(args.secret_file, encoding="utf-8") as fh:
            return fh.read().strip()
    value = os.environ.get(args.secret_env)
    if not value:
        usage_error(f"set ${args.secret_env} or pass --secret-file")
    return value


def main() -> int:
    p = argparse.ArgumentParser(description="Verify or produce webhook signatures offline.")
    p.add_argument("--scheme", required=True, choices=["stripe", "github", "slack", "shopify", "standard"])
    p.add_argument("--body-file", required=True, help="exact raw request body as received (bytes)")
    p.add_argument("--signature", help="received signature header value (omit with --sign)")
    p.add_argument("--timestamp", help="timestamp header (slack, standard; stripe when signing)")
    p.add_argument("--id", dest="msg_id", help="webhook-id header (standard)")
    p.add_argument("--secret-env", default="WEBHOOK_SECRET", help="env var holding the secret")
    p.add_argument("--secret-file", help="file holding the secret")
    p.add_argument("--tolerance", type=int, default=300, help="max age in seconds (0 disables)")
    p.add_argument("--now", type=int, help="override current unix time (for fixtures)")
    p.add_argument("--sign", action="store_true", help="print the header a sender would send")
    args = p.parse_args()

    try:
        with open(args.body_file, "rb") as fh:
            body = fh.read()
    except OSError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2
    secret = read_secret(args)
    now = args.now if args.now is not None else int(time.time())

    if args.sign:
        ts = args.timestamp or str(now)
        if args.scheme == "standard" and not args.msg_id:
            print("error: --id is required for the standard scheme", file=sys.stderr)
            return 2
        print(compute(args.scheme, secret, body, ts, args.msg_id))
        return 0

    if not args.signature:
        print("error: --signature is required unless --sign is used", file=sys.stderr)
        return 2

    embedded_ts, received = candidates(args.scheme, args.signature)
    ts = embedded_ts or args.timestamp
    if args.scheme in ("stripe", "slack", "standard") and not ts:
        print("error: this scheme needs a timestamp (--timestamp or t= in the header)", file=sys.stderr)
        return 2
    if args.scheme == "standard" and not args.msg_id:
        print("error: --id is required for the standard scheme", file=sys.stderr)
        return 2
    if not received:
        print("INVALID: no signature of the expected version found in header")
        return 1

    expected = expected_value(args.scheme, compute(args.scheme, secret, body, ts, args.msg_id))
    match = any(hmac.compare_digest(expected.encode(), r.encode()) for r in received)

    stale = False
    if ts and args.tolerance > 0:
        try:
            age = now - int(ts)
        except ValueError:
            print(f"INVALID: timestamp {ts!r} is not an integer")
            return 1
        stale = abs(age) > args.tolerance

    if match and not stale:
        print(f"VALID: {args.scheme} signature matches ({len(body)} body bytes)")
        return 0

    if not match:
        print(f"INVALID: {args.scheme} signature does not match")
        print(f"  expected: {expected}")
        print(f"  received: {', '.join(received)}")
        hints = []
        if body.endswith(b"\n"):
            hints.append("body ends with a newline; was it added when saving the file?")
        if b"\r\n" in body:
            hints.append("body contains CRLF line endings; check the capture preserved raw bytes")
        if body[:1] in (b"{", b"[") and b": " in body and b'":"' not in body:
            hints.append("body looks pretty-printed; the sender signed the compact raw bytes")
        hints.append("confirm the secret belongs to this endpoint and environment (test vs live)")
        for h in hints:
            print(f"  hint: {h}")
    if stale:
        print(f"INVALID: timestamp is {abs(now - int(ts))}s from now (tolerance {args.tolerance}s)")
    return 1


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