Skip to content
AI Inbx
Esc
navigateopen⌘Jpreview
On this page

Verifying

The signature scheme, how to verify it in each SDK, and how to do it by hand in any language.

Your webhook URL is a public endpoint. Anyone who learns it can post to it. The signature is what separates an event AI Inbx sent from one somebody made up — verify it on every request, before you parse the body.

The scheme

Each delivery carries:

AIInbx-Signature: t=1756721464,v1=5f2b8c...

t is the Unix timestamp of the signature. v1 is HMAC-SHA256(secret, "<t>.<raw body>"), hex-encoded.

Two properties matter:

  • The signed string includes the timestamp, so an old capture can’t be replayed indefinitely. Anything outside a five-minute window is rejected by default.
  • It’s computed over the raw bytes, so re-serializing the JSON before verifying will produce a different string and fail. Key order, whitespace, and Unicode escaping all change the bytes.

TypeScript

verifyWebhookRequest takes the Request and does everything — raw body, header, timing-safe comparison, replay window:

import { verifyWebhookRequest } from "aiinbx/webhooks"

export async function POST(request: Request) {
  let event

  try {
    event = await verifyWebhookRequest(
      request,
      process.env.AI_INBX_WEBHOOK_SECRET!
    )
  } catch {
    return new Response("invalid signature", { status: 400 })
  }

  // `type` is a discriminant — TypeScript narrows `data` to this event's shape.
  if (event.type === "email.received") {
    console.log(event.data.thread_id, event.data.category)
  }

  return new Response(null, { status: 204 })
}

When your framework hands you the body and header separately, use verifyWebhook:

import { verifyWebhook } from "aiinbx/webhooks"

const event = await verifyWebhook(
  rawBody, // string or Uint8Array — exactly as received
  request.headers["aiinbx-signature"],
  process.env.AI_INBX_WEBHOOK_SECRET!
)

Both throw WebhookSignatureError on a bad signature, a malformed header, or a timestamp outside the window. Both work anywhere Web Crypto does — Node 20+, Bun, Deno, Cloudflare Workers, Vercel Edge.

Python

from aiinbx import AIInbx

client = AIInbx()

event = client.webhooks.verify(
    request_body,                          # bytes or str, unmodified
    request.headers["AIInbx-Signature"],
    webhook_secret,
)

if event["type"] == "email.bounced":
    for recipient in event["data"]["recipients"]:
        print(recipient)

event is a discriminated WebhookEvent union, so a type checker narrows data from type.

If your framework gives you a bare digest and the timestamp separately, pass the timestamp explicitly:

event = client.webhooks.verify(body, digest, secret, timestamp=header_timestamp)

Any language

Six steps, no library required:

Read the raw body

As bytes or a string. Do not parse it yet.

Parse the header

Split AIInbx-Signature on ,, then each part on the first =, giving t and v1.

Check the timestamp

Reject if |now − t| exceeds 300 seconds.

Build the signed string

"<t>.<raw body>" — a literal dot between them.

Compute the HMAC

HMAC-SHA256 with your endpoint secret, hex-encoded.

Compare in constant time

Use a timing-safe comparison, never ==.

func verify(body []byte, header, secret string) (bool, error) {
    var timestamp, provided string
    for _, part := range strings.Split(header, ",") {
        key, value, _ := strings.Cut(strings.TrimSpace(part), "=")
        switch key {
        case "t":
            timestamp = value
        case "v1":
            provided = value
        }
    }

    seconds, err := strconv.ParseInt(timestamp, 10, 64)
    if err != nil {
        return false, err
    }
    if math.Abs(float64(time.Now().Unix()-seconds)) > 300 {
        return false, errors.New("outside the replay window")
    }

    mac := hmac.New(sha256.New, []byte(secret))
    fmt.Fprintf(mac, "%s.%s", timestamp, body)
    expected := hex.EncodeToString(mac.Sum(nil))

    return hmac.Equal([]byte(expected), []byte(provided)), nil
}

Getting the raw body

The usual failure is a framework parsing JSON before you see it. Per framework:

Request gives you the raw body directly:

export async function POST(request: Request) {
  const event = await verifyWebhookRequest(request, secret)
  return new Response(null, { status: 204 })
}

Mount a raw body parser on the webhook route only:

app.post(
  "/webhooks/aiinbx",
  express.raw({ type: "application/json" }),
  async (req, res) => {
    const event = await verifyWebhook(
      req.body, // a Buffer, thanks to express.raw
      req.header("AIInbx-Signature")!,
      process.env.AI_INBX_WEBHOOK_SECRET!
    )
    res.status(204).end()
  }
)

A global express.json() mounted before this route will have consumed the stream — order matters.

@app.post("/webhooks/aiinbx")
async def webhook(request: Request):
    body = await request.body()          # raw bytes
    event = client.webhooks.verify(
        body,
        request.headers["aiinbx-signature"],
        WEBHOOK_SECRET,
    )
    return Response(status_code=204)

Take Request rather than a Pydantic model — a declared body model parses the request before your code runs.

@csrf_exempt
def aiinbx_webhook(request):
    event = client.webhooks.verify(
        request.body,                    # raw bytes
        request.headers["AIInbx-Signature"],
        settings.AIINBX_WEBHOOK_SECRET,
    )
    return HttpResponse(status=204)

Clock skew

The five-minute window assumes your server’s clock is roughly right. A host drifting by more than that rejects every event, which looks exactly like a wrong secret. If verification fails across the board and the secret is definitely correct, check NTP before anything else.

Both SDKs let you widen the window, but treat that as a diagnostic rather than a fix:

await verifyWebhookRequest(request, secret, { tolerance: 600 })
client.webhooks.verify(body, signature, secret, tolerance=600)

After verifying

A valid signature proves the event came from AI Inbx. It doesn’t promise you haven’t seen it before — retries and manual replays deliver the same event again. Deduplicate on event.id:

const event = await verifyWebhookRequest(request, secret)

// Application-defined: atomically insert into durable storage by event.id.
// An existing event is a successful no-op; a storage failure must throw.
await enqueueOnce(event.id, event)
return new Response(null, { status: 204 })

Do not mark an event complete before its work succeeds. A separate “check then insert” can race under concurrent delivery; use a unique constraint or an equivalent atomic queue operation.

The pairing to remember: event.id for “have I seen this event”, and an idempotency key derived from it for “have I already sent the reply”.

Last updated on September 9, 2026

Was this page helpful?