Receiving webhooks

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

const app = express();
const SIGNING_KEY = process.env.OMS_WEBHOOK_SIGNING_KEY!;
const TOLERANCE_SECONDS = 300;

function verify(rawBody: Buffer, header: string | undefined): boolean {
  if (!header) return false;

  const parts = new Map(
    header.split(",").map((kv) => {
      const [k, v] = kv.split("=");
      return [k.trim(), v?.trim() ?? ""];
    }),
  );
  const timestamp = parts.get("t");
  const signature = parts.get("v1");
  if (!timestamp || !signature) return false;

  const age = Math.abs(Math.floor(Date.now() / 1000) - Number(timestamp));
  if (!Number.isFinite(age) || age > TOLERANCE_SECONDS) return false;

  const expected = crypto
    .createHmac("sha256", SIGNING_KEY)
    .update(`${timestamp}.`)
    .update(rawBody)
    .digest();
  const received = Buffer.from(signature, "hex");

  return (
    received.length === expected.length &&
    crypto.timingSafeEqual(received, expected)
  );
}

// express.raw keeps the exact bytes OMS signed.
app.post(
  "/webhooks/oms",
  express.raw({ type: "application/json" }),
  (req, res) => {
    if (!verify(req.body, req.header("Webhook-Signature"))) {
      return res.sendStatus(401);
    }

    const delivery = JSON.parse(req.body.toString("utf8"));

    // Acknowledge first, process outside the request.
    res.sendStatus(200);
    void enqueue(delivery);
  },
);

import hashlib
import hmac
import json
import os
import time

from flask import Flask, request

app = Flask(__name__)
SIGNING_KEY = os.environ["OMS_WEBHOOK_SIGNING_KEY"].encode()
TOLERANCE_SECONDS = 300


def verify(raw_body: bytes, header: str | None) -> bool:
    if not header:
        return False

    parts = dict(
        part.split("=", 1) for part in header.split(",") if "=" in part
    )
    timestamp, signature = parts.get("t"), parts.get("v1")
    if not timestamp or not signature:
        return False

    try:
        age = abs(int(time.time()) - int(timestamp))
    except ValueError:
        return False
    if age > TOLERANCE_SECONDS:
        return False

    expected = hmac.new(
        SIGNING_KEY, f"{timestamp}.".encode() + raw_body, hashlib.sha256
    ).hexdigest()

    return hmac.compare_digest(signature, expected)


@app.post("/webhooks/oms")
def receive():
    # request.get_data() keeps the exact bytes OMS signed.
    raw_body = request.get_data()
    if not verify(raw_body, request.headers.get("Webhook-Signature")):
        return "", 401

    delivery = json.loads(raw_body)

    # Acknowledge first, process outside the request.
    enqueue(delivery)
    return "", 200