# Webhook Agent Gate: the safe webhook pattern for agents and a ten-check handler scan (free sample)

Version 1.0, 2026-10-31. By PaperChaseWebb. For anyone who lets a Stripe, GitHub, form or store webhook start an agent, a Zap, an n8n flow or a refund. One page and one Python 3 script, standard library only, read-only: it reads handler files and never runs them, sends nothing and reads no secrets. Tested with 3 unit tests and 41 planted faults (each of the ten checks fires) on Linux with Python 3.13, 3.12, 3.11, 3.9 and 3.8, and run on the eight Python and Node webhook servers in Stripe's public accept-a-payment samples.

## The pattern

Providers send the same event more than once, out of order, and keep retrying when your handler is slow or answers with an error. That is normal and documented. It turns into a problem the day the handler starts an agent or moves money, because each copy can run the action again. The safe shape:

1. Verify the signature on the raw body, before parsing. A failed check answers 400 and stops. A missing secret stops the server from starting; it never means "skip the check".
2. Store the event once, under a unique key on the event id (INSERT OR IGNORE or ON CONFLICT DO NOTHING). Answer 200 quickly, including to a duplicate.
3. Do the work later, in a worker. Each action gets a key built from what it acts on (refund:ch_123), not from the event or the run. Two copies of one event, or two events about the same charge, land on one row.
4. Money and anything a customer would notice wait for the owner, at a terminal the agent cannot reach. The agent may only propose.
5. If the worker died mid-action, the outcome is unknown. Look it up at the provider. Never re-run it by itself.

## Rules to paste into AGENTS.md, CLAUDE.md or a Grok Bot's standing instructions

1. A webhook handler verifies, stores and answers. It never calls a payments, email or agent API inside the request.
2. Every outside action has a key built from what it acts on. Never a random id, the time or a run id.
3. Refunds, charges and messages to customers wait for the owner. You may propose them; you never approve them.
4. Webhook secrets live in env vars the owner sets. Never in code, logs or chat.
5. If an action may or may not have happened, say so with its key and stop.
6. After changing a handler, run `python3 hook_mini.py <file>` and paste the output. Ask the owner before anything that touches money.

## What the script checks

| Code | Finding |
|---|---|
| H01 | no signature check |
| H02 | signature compared with ==, which leaks timing |
| H03 | body parsed before the signature check, verified against a re-serialized body, or express.json() on every route |
| H05 | no duplicate protection |
| H06 | refund, email or agent call inside the webhook request |
| H07 | Stripe write with no idempotency key |
| H08 | secret written in the code |
| H09 | idempotency key that is new on every try (random id, time or run id) |
| H10 | failed signature check still answers 2xx, or is caught and the handler carries on |
| H12 | duplicate check that reads then writes, so two copies arriving together both pass |

```
python3 hook_mini.py app/webhooks.py src/routes/stripe.ts
```

Exit 0 with no high findings, 1 with any high finding. Python and JavaScript/TypeScript only; other files are skipped.

On a made-up "billing starter" Express route of the kind sold as boilerplate, the script finds 8 problems, 7 of them high: express.json() on every route, a signature checked against JSON.stringify(req.body), a failed check that is logged and then ignored, a read-then-write duplicate check, a refund inside the request with no idempotency key, an email inside the request, and a live-looking key in the code. The full kit finds 12 on the same file, adding the signature check that is skipped when the secret is unset, the signature written to the log, the agent that can trigger the refund, and duplicates answered with 409.

## Install in one line

Save this page as FREE-SAMPLE.md, then:

```
python3 -c "import re;t=open('FREE-SAMPLE.md').read();open('hook_mini.py','w').write(re.search(r'~~~python\n(.*?)~~~',t,re.S).group(1))"
```

## Checklist before a webhook triggers an agent

- [ ] Signature verified on the raw bytes, with a timestamp tolerance where the provider signs one (Stripe and Standard Webhooks do).
- [ ] The server refuses to start without the secret.
- [ ] Event id stored under a unique key; duplicates answered 200.
- [ ] No refund, email or agent call inside the request.
- [ ] Each action keyed by what it acts on.
- [ ] Money waits for the owner.
- [ ] You sent the same event 20 times at once at a local copy and counted one side effect.

## What the full kit adds

The Webhook Agent Gate kit ($29) is the working version of this page: a small receiver that verifies Stripe, GitHub, Standard Webhooks and hex-signed deliveries and stores each once; an action ledger with keys built from the action, atomic claims and an unknown state that is never re-run on its own; a propose script for the agent and an approve script that only a person at a terminal can use; replay.py, which fires duplicates, bursts, forged and stale deliveries at your own handler; the full checker with handler, events-map, ledger and agent-rules checks; rules and skills for Grok Bot, Claude Code, Codex and Hermes; and 99 planted faults plus 97 unit tests you can run yourself. paperchasewebb.ai, or write to mgmt@paperchasewebb.com.

## Not tested

Live Stripe or GitHub deliveries, macOS and Windows. The script reads patterns; it does not follow helper functions defined in other files, so read the handler yourself too.

## The script

~~~python
#!/usr/bin/env python3
"""hook_mini.py: the free ten-check version of hook_check.py.

    python3 hook_mini.py path/to/webhook_handler.py [more files]

Read-only. Checks one or more webhook handler files (Python or JS/TS) for:
  H01 no signature check           H02 signature compared with ==
  H03 parsed or re-serialized body H05 no duplicate protection
  H06 refund, email or agent call inside the request
  H07 Stripe write with no idempotency key
  H08 secret written in the code   H09 idempotency key new on every try
  H10 failed signature check still answers 2xx
  H12 duplicate check that reads then writes (races)
The paid kit adds timestamp, logging, agent and approval checks, the
events, ledger and rules checks, the gate itself and replay.py.
Standard library only. Python 3.8 or newer. Exit 1 when a HIGH shows up.
"""
import re
import sys

KEEP = ['H01', 'H02', 'H03', 'H05', 'H06', 'H07', 'H08', 'H09', 'H10', 'H12']


class Finding(object):
    def __init__(self, code, sev, where, msg, fix):
        self.code, self.sev, self.where, self.msg, self.fix = code, sev, where, msg, fix


VERIFY = re.compile(r"compare_digest|timingSafeEqual|construct_?[Ee]vent|constructEventAsync|"
                    r"verify_?[Ss]ignature|verify_?[Hh]eader|\.verify\(|verify_(stripe|github|standard|hex)\(|"
                    r"verifyWebhook|validateEvent|Webhook\(|(?<!function )(?<!def )\bverify\w*\(")
HMAC = re.compile(r"hmac\.new|createHmac|crypto\.subtle\.sign|HMAC")
PARSE_PY = re.compile(r"request\.get_json\(|request\.json\b|await request\.json\(\)|json\.loads\(")
PARSE_JS = re.compile(r"await (req|request)\.json\(\)|JSON\.parse\(")
GLOBAL_JSON_JS = re.compile(r"app\.use\(\s*(express\.json|bodyParser\.json)\(")
RAW_JS = re.compile(r"express\.raw\(|bodyParser\.raw\(|\.text\(\)|rawBody|raw_body|arrayBuffer\(\)|get_data\(|request\.body\b|rfile\.read")
RESTRINGIFY = re.compile(r"JSON\.stringify\(\s*req(uest)?\.body\s*\)|json\.dumps\(\s*(request\.json|payload|body|event)")
EQ_SIG = re.compile(r"(sig|signature|digest|hmac|expected)[\w.()\"']*\s*(===|!==|==|!=)\s*[\w\"']|"
                    r"[\w)\]]\s*(===|!==|==|!=)\s*[\w.]*(sig|signature|digest|hmac|expected|\.v1\b)", re.I)
MANUAL_TS = re.compile(r"t=|parts\.t\b|\[['\"]t['\"]\]|\$\{t\}\.|\$\{timestamp\}\.|webhook-timestamp|webhook_timestamp|"
                       r"x-slack-request-timestamp|svix-timestamp", re.I)
TOLERANCE = re.compile(r"toleran|abs\(|Math\.abs|max_age|MAX_AGE|> *300|>= *300|300 *[*<>]|five_minutes|FIVE_MIN|stale", re.I)
DEDUPE = re.compile(r"INSERT OR IGNORE|ON CONFLICT|setnx|['\"]NX['\"]|nx\s*[:=]\s*(True|true)|"
                    r"createIfNotExists|\bupsert\b|ignoreDuplicates|idempotency_?[Kk]ey|idempotencyKey", re.I)
CHECK_THEN_SET = re.compile(r"\bin (processed|seen|handled)\w*|\.(get|exists|has)\(\s*\w*(event|evt|delivery)\w*\.?(id|\[)|"
                            r"SELECT .* FROM \w*(event|processed)", re.I)
MONEY = re.compile(r"(refunds?|Refund)\.create\(|(charges?|Charge)\.create\(|paymentIntents?\.(create|confirm|capture)\(|"
                   r"PaymentIntent\.(create|confirm|capture)\(|(payouts?|Payout|transfers?|Transfer)\.create\(|"
                   r"subscriptions?\.(cancel|del|update)\(|Subscription\.(delete|cancel|modify)\(|issue_refund\(|refund\(", re.I)
OTHER_EFFECT = re.compile(r"send_?[Mm]ail\(|sendEmail\(|send_email\(|emails\.send\(|messages\.create\(|"
                          r"chat\.completions\.create\(|responses\.create\(|agent\.(run|invoke)\(|run_agent\(|"
                          r"chat_postMessage\(|chat\.postMessage\(|requests\.post\(|httpx\.post\(|method:\s*['\"]POST")
STRIPE_WRITE = re.compile(r"(stripe\.|\.)(refunds|Refund|charges|Charge|paymentIntents|PaymentIntent|payouts|Payout|"
                          r"transfers|Transfer|invoices|Invoice|customers|Customer|subscriptions|Subscription)\."
                          r"(create|confirm|capture|cancel|update|pay|modify|del)\(")
IDEM_ARG = re.compile(r"idempotency_?[Kk]ey|Idempotency-Key|idempotencyKey", re.I)
RANDOM_KEY = re.compile(r"(idempotency_?[Kk]ey|Idempotency-Key|idempotencyKey|dedupe_?key|action_?key)\W{1,6}[^\n]{0,80}?"
                        r"(uuid4\(|uuid\.v4\(|randomUUID\(|uuidv4\(|Date\.now\(|time\.time\(|datetime\.now\(|"
                        r"run_?[Ii]d|execution_?[Ii]d|\$execution|attempt|Math\.random|nanoid\()", re.I)
SECRET = re.compile(r"(whsec_[A-Za-z0-9+/=]{8,}|sk_live_[A-Za-z0-9]{8,}|sk_test_[A-Za-z0-9]{10,}|rk_live_[A-Za-z0-9]{8,}|"
                    r"ghp_[A-Za-z0-9]{20,}|xox[bpa]-[A-Za-z0-9-]{10,})")
LEGACY_GH = re.compile(r"x-hub-signature(?!-256)|X_HUB_SIGNATURE(?!_256)|HTTP_X_HUB_SIGNATURE(?!_256)|sha1", re.I)
LOG_SECRET = re.compile(r"(print|console\.(log|error|info|warn)|logger\.\w+|logging\.\w+)\([^\n]*(secret|whsec|signature|sig_header|\bsig\b)", re.I)
LOG_HEADER = re.compile(r"(print|console\.(log|error|info|warn)|logger\.\w+|logging\.\w+)\([^\n]*headers?(\.get\(|\[)\s*['\"][^'\"]*signature", re.I)
APPROVAL = re.compile(r"approv|owner|needs_review|human", re.I)
AGENT = re.compile(r"messages\.create\(|chat\.completions\.create\(|responses\.create\(|agent\.(run|invoke)\(|run_agent\(|"
                   r"generateText\(|streamText\(|query\(\s*\{|ClaudeSDKClient|Agent\(")
ROUTE = re.compile(r"webhook|/hooks?/|stripe-signature|x-hub-signature|webhook-signature|x-signature", re.I)
DUP_4XX = re.compile(r"(duplicate|already|seen\b|\.has\(|exists\()[^\n]{0,80}(409|400|422|status\(4)|(409|422)[^\n]{0,80}(duplicate|already)", re.I)
EXCEPT = re.compile(r"^\s*(except\b|}?\s*catch\b)")
OPTIONAL_VERIFY = re.compile(r"^\s*if\s*\(?\s*(process\.env\.\w*(SECRET|WHSEC|SIGNING)\w*|os\.(getenv|environ\.get)\([^)]*(SECRET|WHSEC|SIGNING)[^)]*\)|"
                             r"\w*(secret|SECRET|Secret)\w*)\s*\)?\s*(:|\{|$)")
FAIL_BRANCH = re.compile(r"if\s*\(?\s*(!|not )\s*\w*(verif|valid|check_sig|checkSig)\w*\(", re.I)


ROUTE_DECL = re.compile(r"@(app|router|bp|blueprint|api)\.(route|post|get|put|patch|delete|api_route)\(|"
                        r"\b(app|router|server|fastify)\.(post|get|put|patch|delete|all)\(\s*['\"`]|"
                        r"export\s+(async\s+)?function\s+(POST|GET|PUT)\b")
HOOK_WORD = re.compile(r"webhook|/hooks?\b|/events?\b|stripe|github|lemon|gumroad", re.I)


def _hook_regions(live):
    """Line ranges of webhook routes when a file has several routes. None means
    one route (or none found), so the whole file counts."""
    decl = [(i, n, l) for i, (n, l) in enumerate(live) if ROUTE_DECL.search(l)]
    if len(decl) < 2:
        return None
    out = []
    for j, (i, n, l) in enumerate(decl):
        if HOOK_WORD.search(l):
            end = decl[j + 1][1] if j + 1 < len(decl) else 10 ** 9
            out.append((n, end))
    return out


def _in(regions, n):
    return regions is None or any(a <= n < b for a, b in regions)


def _code_only(line):
    """Drop string literal text, keep ${...} parts of template strings."""
    keep = " ".join(re.findall(r"\$\{([^}]*)\}", line))
    return re.sub(r"`[^`]*`|'[^']*'|\"[^\"]*\"", "''", line) + " " + keep


def _comment(line, js):
    s = line.strip()
    return s.startswith("#") or (js and (s.startswith("//") or s.startswith("*") or s.startswith("/*")))


def _block(live, idx, js):
    """Text of the except/catch block that starts at live[idx]."""
    n0, head = live[idx]
    tail = head.split("catch", 1)[1] if js and "catch" in head else head.split(":", 1)[-1]
    if js and "}" in tail.split("{", 1)[-1]:
        return tail
    if not js and tail.strip():
        return tail
    ind = len(head) - len(head.lstrip())
    out = []
    for _, l in live[idx + 1:idx + 12]:
        if not l.strip():
            continue
        if js and l.strip().startswith("}") and len(l) - len(l.lstrip()) <= ind:
            break
        if not js and len(l) - len(l.lstrip()) <= ind:
            break
        out.append(l)
    return " ".join(out)


def _call_text(text, start):
    """Text of a call from its opening paren to the matching close."""
    i = text.find("(", start)
    depth = 0
    for j in range(i, min(len(text), i + 2000)):
        if text[j] == "(":
            depth += 1
        elif text[j] == ")":
            depth -= 1
            if depth == 0:
                return text[start:j + 1]
    return text[start:start + 2000]


CODE_EXT = (".py", ".js", ".ts", ".mjs", ".cjs", ".tsx", ".jsx")


def check_code(path, text=None):
    if not path.endswith(CODE_EXT):
        return [Finding("H00", "low", path, "not Python or JavaScript/TypeScript, so not checked",
                        "read it by hand against the checklist in section 5")]
    text = open(path, encoding="utf-8", errors="replace").read() if text is None else text
    js = path.endswith((".js", ".ts", ".mjs", ".cjs", ".tsx", ".jsx"))
    lines = text.splitlines()
    live = [(i + 1, l) for i, l in enumerate(lines) if not _comment(l, js)]
    body = "\n".join(l for _, l in live)
    out = []

    def f(code, sev, ln, msg, fix):
        out.append(Finding(code, sev, "%s:%s" % (path, ln), msg, fix))

    if not ROUTE.search(body):
        f("H00", "low", 1, "no webhook route or signature header found in this file",
          "point hook_check at the file that receives the webhook")
    first_verify = next((n for n, l in live if VERIFY.search(l)), None)
    if first_verify is None and not (HMAC.search(body) and re.search(r"compare_digest|timingSafeEqual", body)):
        f("H01", "high", 1, "no signature check found: anyone who finds the URL can trigger the agent",
          "verify the provider signature on the raw body before anything else (see section 3)")
    for n, l in live:
        if EQ_SIG.search(l) and not re.search(r"compare_digest|timingSafeEqual|len\(|\.length|is None|== *None|null|undefined|startswith|typeof", l):
            f("H02", "high", n, "signature compared with ==, which leaks timing",
              "use hmac.compare_digest (Python) or crypto.timingSafeEqual (Node)")
    # H03 parse before verify / re-serialized body
    parse = PARSE_JS if js else PARSE_PY
    regions = _hook_regions(live)
    first_parse = next((n for n, l in live if parse.search(l) and _in(regions, n)), None)
    if first_verify and first_parse and first_parse < first_verify:
        f("H03", "high", first_parse, "body is parsed before the signature is checked",
          "verify the raw bytes first, then parse")
    for n, l in live:
        if RESTRINGIFY.search(l):
            f("H03", "high", n, "signature checked against a re-serialized body, not the raw bytes",
              "keep the raw body (express.raw, request.get_data(), await req.text()) for verification")
    if js and GLOBAL_JSON_JS.search(body) and not RAW_JS.search(body):
        n = next(n for n, l in live if GLOBAL_JSON_JS.search(l))
        f("H03", "high", n, "express.json() runs on every route, so the webhook never sees the raw body",
          "mount express.raw({type: 'application/json'}) on the webhook route before express.json()")
    # H04 timestamp tolerance on manual verification
    sdk = re.search(r"construct_?[Ee]vent|constructEventAsync|Webhook\(|verify_header|\.verify\(", body)
    if HMAC.search(body) and MANUAL_TS.search(body) and not sdk and not TOLERANCE.search(body):
        n = next(n for n, l in live if MANUAL_TS.search(l))
        f("H04", "med", n, "timestamp is signed but never checked, so an old delivery can be replayed",
          "reject deliveries more than 300 seconds from now")
    # H05 / H12 dedupe
    if not DEDUPE.search(body):
        if CHECK_THEN_SET.search(body):
            n = next(n for n, l in live if CHECK_THEN_SET.search(l))
            f("H12", "high", n, "duplicate check reads then writes, so two copies arriving together both pass",
              "let the database decide: INSERT OR IGNORE / ON CONFLICT DO NOTHING on a unique event id")
        else:
            f("H05", "high", 1, "no duplicate protection found: providers send the same event more than once",
              "store each event id under a unique key and skip repeats (section 4)")
    # H06 side effects inside the request
    regions = _hook_regions(live)
    for n, l in live:
        if not _in(regions, n):
            continue
        if MONEY.search(l):
            f("H06", "high", n, "money moves inside the webhook request",
              "store the event, answer 200, and let a worker run the refund through the ledger")
        elif OTHER_EFFECT.search(l):
            f("H06", "med", n, "outside action or agent call inside the webhook request",
              "answer 200 first; do the work in a worker that checks the ledger")
    # H07 stripe writes with no idempotency key
    for m in STRIPE_WRITE.finditer(body):
        call = _call_text(body, m.start())
        if not IDEM_ARG.search(call):
            n = body[:m.start()].count("\n") + 1
            ln = live[n - 1][0] if n - 1 < len(live) else n
            if _in(regions, ln):
                f("H07", "high", ln, "Stripe write with no idempotency key",
                  "pass idempotency_key built from the action, like refund:ch_123")
            else:
                f("H07", "low", ln, "Stripe write outside the webhook route has no idempotency key; a retried request repeats it",
                  "pass an idempotency key built from the order or cart id")
    for n, l in live:
        if RANDOM_KEY.search(l):
            f("H09", "high", n, "idempotency key is new on every try (random id, time or run id), so retries repeat the action",
              "build the key from the action and its target: kind:object_id")
    for n, l in enumerate(lines, 1):
        if SECRET.search(l):
            f("H08", "high", n, "secret written in the code", "read it from an env var; rotate this one")
    # H10 verification error swallowed
    for idx, (n, l) in enumerate(live):
        if EXCEPT.search(l):
            before = " ".join(x for _, x in live[max(0, idx - 10):idx])
            after = " ".join(x for _, x in live[idx + 1:idx + 4])
            if VERIFY.search(before) and re.search(r"\b200\b|\bpass\b|sendStatus\(2|status\(2|ok\(\)|continue", after) \
                    and not re.search(r"\b40[0-9]\b|raise|throw|abort", after):
                f("H10", "high", n, "a failed signature check still answers 2xx or carries on",
                  "answer 400 and stop when verification fails")
            elif VERIFY.search(before) and not re.search(r"\b40[0-9]\b|raise|throw|abort|return|exit\(|next\(", _block(live, idx, js)):
                f("H10", "high", n, "a failed signature check is caught and the handler carries on",
                  "answer 400 and return inside the except/catch block")
    for idx, (n, l) in enumerate(live):
        if FAIL_BRANCH.search(l):
            blk = " ".join(x for _, x in live[idx:idx + 3])
            m = re.search(r"\b(200|201|202|204)\b|sendStatus\(2|status\(2", blk)
            if m and not re.search(r"\b40[0-9]\b|raise|throw|abort", blk[:m.start()] if m else blk):
                f("H10", "high", n, "a failed signature check still answers 2xx or carries on",
                  "answer 400 and stop when verification fails")
    for idx, (n, l) in enumerate(live):
        if OPTIONAL_VERIFY.search(l) and "not" not in l.split("if", 1)[1][:6] and "!" not in l.split("if", 1)[1][:4]:
            nxt = " ".join(x for _, x in live[idx + 1:idx + 7])
            if VERIFY.search(nxt):
                f("H16", "high", n, "signature check only runs when the secret is set; a deploy without it accepts anything",
                  "fail closed: refuse to start (or answer 500) when the secret is missing")
    for n, l in live:
        if LEGACY_GH.search(l) and "256" not in l:
            f("H11", "med", n, "uses the legacy SHA-1 X-Hub-Signature header",
              "use X-Hub-Signature-256 with HMAC-SHA256")
    for n, l in live:
        if LOG_SECRET.search(_code_only(l)) or LOG_HEADER.search(l):
            f("H13", "med", n, "secret or signature written to logs", "log the event id and type only")
    hook_lines = [(n, l) for n, l in live if _in(regions, n)]
    if any(AGENT.search(l) for _, l in hook_lines) and any(MONEY.search(l) for _, l in hook_lines) \
            and not APPROVAL.search(body):
        n = next(n for n, l in hook_lines if AGENT.search(l))
        f("H14", "high", n, "an agent runs in the same path that can move money, with no owner step",
          "the agent may propose; money waits for approve.py")
    for n, l in live:
        if DUP_4XX.search(l):
            f("H15", "low", n, "duplicates get a 4xx, so the provider keeps retrying",
              "answer 200 to a duplicate you already have")
    return out


def main(argv):
    if not argv:
        print(__doc__)
        return 2
    fs = []
    for p in argv:
        fs += [f for f in check_code(p) if f.code in KEEP]
    for f in fs:
        print("%-4s %s %s\n     %s\n     fix: %s" % (f.sev.upper(), f.code, f.where, f.msg, f.fix))
    print("\n%d findings (%d high). Ten checks only; see the full kit for the rest." %
          (len(fs), len([f for f in fs if f.sev == "high"])))
    return 1 if any(f.sev == "high" for f in fs) else 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))
~~~

License: free to use and share unchanged. PaperChaseWebb Inc. Stripe, GitHub and the agents named belong to their owners; this page is independent.