Verify a webhook signature

Goal: accept only deliveries that really came from Octane AI and were not replayed, and never process the same event twice.

Uses: Webhooks. The endpoint is created with a key that carries webhooks:manage (or from Developer > Webhooks in the dashboard); the secret is shown when the endpoint is created and again by POST /v1/webhooks/{endpoint_id}/reveal-secret. Put it in the OCTANEAI_WEBHOOK_SECRET environment variable of the receiver.

Deliveries follow the Standard Webhooks specification: the webhook-id, webhook-timestamp and webhook-signature headers, signed with your endpoint’s whsec_... secret over the raw request body. Each receiver below (Flask, pip install flask; Express, npm install express; Go’s net/http) repeats the verify() from the Webhooks page verbatim so the block runs on its own, then calls one function of yours, accept(delivery_id, event): in a single transaction it enqueues the work and records the verified webhook-id header (an INSERT ... ON CONFLICT DO NOTHING on the id plus the job row), so the id is only ever recorded once the job is durably queued; it returns false when the id was already there. Dedupe on the header, not on a body field: a ping has no id inside data.object. The receiver acknowledges only after accept returns; the work happens off the request.

Python

import base64
import hashlib
import hmac
import os
import time
from flask import Flask, request

app = Flask(__name__)
SECRET = os.environ["OCTANEAI_WEBHOOK_SECRET"]  # whsec_...

def accept(delivery_id: str, event) -> bool:
    """Your function here: enqueue the job and insert delivery_id in one transaction; False when already seen."""
    raise NotImplementedError

def verify(secret: str, headers, body: bytes) -> bool:
    try:
        key = base64.b64decode(secret[len("whsec_"):], validate=True)
        msg_id, ts, sigs = headers["webhook-id"], headers["webhook-timestamp"], headers["webhook-signature"]
        if abs(time.time() - int(ts)) > 300:
            return False
        expected = hmac.new(key, f"{msg_id}.{ts}.".encode() + body, hashlib.sha256).digest()
        for sig in sigs.split(" "):
            version, _, value = sig.partition(",")
            if version == "v1" and hmac.compare_digest(base64.b64decode(value, validate=True), expected):
                return True
    except (KeyError, ValueError):  # a missing header, a bad timestamp, invalid base64
        pass
    return False

@app.post("/octane")
def octane():
    if not verify(SECRET, request.headers, request.get_data()):
        return "bad signature", 401
    accept(request.headers["webhook-id"], request.get_json())  # False on a retry or a redelivery
    return "", 200  # acknowledged only once the job is durably queued

if __name__ == "__main__":
    app.run(port=8080)

TypeScript

import express from "express";
import { createHmac, timingSafeEqual } from "node:crypto";

const SECRET = process.env.OCTANEAI_WEBHOOK_SECRET!; // whsec_...

// Your function here: enqueue the job and insert deliveryId in one transaction; resolves false when already seen.
async function accept(deliveryId: string, event: { type: string; data: unknown }): Promise<boolean> {
  throw new Error("not implemented");
}

function verify(secret: string, headers: Record<string, string | undefined>, body: Buffer): boolean {
  const id = headers["webhook-id"], ts = headers["webhook-timestamp"], sigs = headers["webhook-signature"];
  if (!id || !ts || !sigs || !/^\d+$/.test(ts)) return false;
  if (Math.abs(Date.now() / 1000 - Number(ts)) > 300) return false;
  const key = Buffer.from(secret.slice("whsec_".length), "base64");
  const expected = createHmac("sha256", key).update(`${id}.${ts}.`).update(body).digest();
  for (const sig of sigs.split(" ")) {
    const [version, value] = sig.split(",");
    if (version !== "v1" || !value) continue;
    const given = Buffer.from(value, "base64");
    if (given.length === expected.length && timingSafeEqual(given, expected)) return true;
  }
  return false;
}

const app = express();
app.post("/octane", express.raw({ type: "application/json" }), async (req, res) => {
  const headers = {
    "webhook-id": req.get("webhook-id"),
    "webhook-timestamp": req.get("webhook-timestamp"),
    "webhook-signature": req.get("webhook-signature"),
  };
  if (!verify(SECRET, headers, req.body)) return res.status(401).send("bad signature");
  await accept(headers["webhook-id"]!, JSON.parse(req.body.toString("utf8"))); // false on a retry or a redelivery
  res.sendStatus(200); // acknowledged only once the job is durably queued
});
app.listen(8080);

Go

package main

import (
  "crypto/hmac"
  "crypto/sha256"
  "encoding/base64"
  "encoding/json"
  "io"
  "math"
  "net/http"
  "os"
  "strconv"
  "strings"
  "time"
)

var secret = os.Getenv("OCTANEAI_WEBHOOK_SECRET") // whsec_...

// accept is your function: enqueue the job and insert deliveryID in one transaction; false when already seen.
func accept(deliveryID, topic string, data json.RawMessage) (bool, error) {
  panic("not implemented")
}

func verify(secret string, h http.Header, body []byte) bool {
  key, err := base64.StdEncoding.DecodeString(strings.TrimPrefix(secret, "whsec_"))
  if err != nil {
    return false
  }
  id, ts, sigs := h.Get("webhook-id"), h.Get("webhook-timestamp"), h.Get("webhook-signature")
  sent, err := strconv.ParseInt(ts, 10, 64)
  if id == "" || err != nil || math.Abs(float64(time.Now().Unix()-sent)) > 300 {
    return false
  }
  mac := hmac.New(sha256.New, key)
  mac.Write([]byte(id + "." + ts + "."))
  mac.Write(body)
  expected := mac.Sum(nil)
  for _, sig := range strings.Split(sigs, " ") {
    version, value, ok := strings.Cut(sig, ",")
    if !ok || version != "v1" {
      continue
    }
    if given, err := base64.StdEncoding.DecodeString(value); err == nil && hmac.Equal(given, expected) {
      return true
    }
  }
  return false
}

func octane(w http.ResponseWriter, r *http.Request) {
  body, _ := io.ReadAll(r.Body)
  if !verify(secret, r.Header, body) {
    http.Error(w, "bad signature", http.StatusUnauthorized)
    return
  }
  var event struct {
    Type string          `json:"type"`
    Data json.RawMessage `json:"data"`
  }
  json.Unmarshal(body, &event)
  if _, err := accept(r.Header.Get("webhook-id"), event.Type, event.Data); err != nil {
    http.Error(w, "try again", http.StatusServiceUnavailable) // Octane AI retries a 5xx
    return
  }
  w.WriteHeader(http.StatusOK) // acknowledged only once the job is durably queued
}

func main() {
  http.HandleFunc("/octane", octane)
  http.ListenAndServe(":8080", nil)
}

Why each line is there

• Raw body. The signature covers {webhook-id}.{webhook-timestamp}.{body} byte for byte. express.raw, request.get_data() and io.ReadAll hand you those bytes untouched; a re-serialized body will not match.
• Timestamp window. webhook-timestamp is fresh on every attempt; rejecting one more than five minutes from your clock defeats a replay of an old delivery.
• Malformed input answers 401, never 500. A missing header, a non-numeric timestamp or invalid base64 makes verify return false instead of throwing, so a bad delivery does not become a retried server error.
• Two signatures during a rotation. After POST .../rotate-secret, deliveries carry two space-separated v1,... signatures for 24 hours. The loop accepts the delivery when any of them matches the secret you hold, so update OCTANEAI_WEBHOOK_SECRET any time inside the window.
• Constant-time compare. hmac.compare_digest, timingSafeEqual and hmac.Equal do not leak how many bytes matched.
• Dedupe and enqueue in one step, on the header. Delivery is at least once: a retry after a slow answer, or a redelivery you asked for, sends the same webhook-id again. accept writes the job and the id in one transaction, so a crash between “queued” and “seen” cannot lose an event and a retry cannot run it twice; the id comes from the verified header because a ping carries none in its object.
• Answer fast. Each attempt has a 10-second deadline; answer a 5xx when accept fails and the delivery is retried. A 2xx marks it delivered; a 410 disables the endpoint; other 4xx answers are final; 408, 429, 5xx and timeouts are retried over about a day.
• The first delivery is a ping. A new endpoint receives {"type": "ping", "data": {"object": {"endpoint_id": "whe_..."}}} and becomes active once you answer 2xx. Handle it like any other event: verify, then ignore.
• Route on the header. X-Octane-Topic repeats the event’s type, so you can dispatch before parsing the body.

You can also match the sender in a firewall rule: every attempt carries User-Agent: OctaneAI-CarrierPigeon/1.0 (+https://developers.octaneai.com/webhooks).