Verify signatures

Check each message really came from O2M before you act on it.

Overview

Every webhook carries a signature made with your signing secret. Your server repeats the calculation to confirm the message is genuine.

Why check

Your webhook address is on the public internet, so anyone could send it a message. The signature proves a message came from O2M and was not changed on the way.

Important Check the signature on every message, and use the raw body exactly as it arrived. Parsing the JSON and turning it back into text changes it, and the signature will not match.

How it works

HTTP

X-O2M-Signature: t=1790000000,v1=5f2c9a…

1Split the header into t, the time in Unix seconds, and v1, the signature.
2Join t, a full stop and the raw body: t.body.
3Compute the HMAC-SHA256 of that text with your signing secret, in hexadecimal.
4Compare it with v1, using a constant-time comparison.
5Refuse a message whose t is more than 5 minutes old, which stops an old message being sent again.

Node.js

import crypto from "node:crypto";
import express from "express";

const app = express();

function isFromO2M(header, rawBody, secret) {
const parts = Object.fromEntries(header.split(",").map((part) => part.split("=")));
const expected = crypto.createHmac("sha256", secret).update(`${parts.t}.${rawBody}`).digest("hex");
const a = Buffer.from(parts.v1 ?? "");
const b = Buffer.from(expected);
const fresh = Math.abs(Date.now() / 1000 - Number(parts.t)) < 300;
return fresh && a.length === b.length && crypto.timingSafeEqual(a, b);
}

// express.raw keeps the body exactly as it arrived.
app.post("/o2m-webhook", express.raw({ type: "application/json" }), (req, res) => {
if (!isFromO2M(req.get("X-O2M-Signature") ?? "", req.body, process.env.O2M_SIGNING_SECRET)) {
  return res.status(401).end();
}
const event = JSON.parse(req.body);
res.status(200).end(); // Answer first, then do the work.
handle(event);
});

Python

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

app = Flask(__name__)

def is_from_o2m(header, raw_body, secret):
  parts = dict(part.split("=", 1) for part in header.split(",") if "=" in part)
  signed = parts.get("t", "").encode() + b"." + raw_body
  expected = hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest()
  fresh = abs(time.time() - float(parts.get("t") or 0)) < 300
  return fresh and hmac.compare_digest(parts.get("v1", ""), expected)

@app.post("/o2m-webhook")
def receive():
  raw_body = request.get_data()
  header = request.headers.get("X-O2M-Signature", "")
  if not is_from_o2m(header, raw_body, os.environ["O2M_SIGNING_SECRET"]):
      abort(401)
  event = request.get_json()
  handle(event)  # Keep this quick: pass slow work to a queue.
  return "", 200