Receive and verify PanoFlow webhooks

const crypto = require("node:crypto");
const express = require("express");

const TOLERANCE_SECONDS = 300;

function verifyPanoFlowSignature(rawBody, header, secret, now = Math.floor(Date.now() / 1000)) {
  let timestamp;
  const signatures = [];
  for (const part of header.split(",")) {
    const i = part.indexOf("=");
    if (i === -1) continue;
    const key = part.slice(0, i);
    const value = part.slice(i + 1);
    if (key === "t") timestamp = value;
    else if (key === "v1") signatures.push(value);
    // Ignore other schemes, such as a future v2.
  }
  if (!timestamp || !/^\d+$/.test(timestamp) || signatures.length === 0) return false;
  if (Math.abs(now - Number(timestamp)) > TOLERANCE_SECONDS) return false;

  const expected = crypto
    .createHmac("sha256", Buffer.from(secret, "utf8"))
    .update(`${timestamp}.`)
    .update(rawBody) // the raw body bytes, not re-serialised JSON
    .digest();

  return signatures.some(
    (sig) => /^[0-9a-f]{64}$/.test(sig) && crypto.timingSafeEqual(Buffer.from(sig, "hex"), expected)
  );
}

const app = express();

// express.raw keeps the body as a Buffer, exactly as received.
app.post("/webhooks/panoflow", express.raw({ type: "application/json" }), (req, res) => {
  const header = req.get("PanoFlow-Signature") ?? "";
  if (!verifyPanoFlowSignature(req.body, header, process.env.PANOFLOW_WEBHOOK_SECRET)) {
    return res.status(400).end();
  }
  const event = JSON.parse(req.body.toString("utf8"));
  // Acknowledge first, then process; skip event.id values you've already handled.
  res.status(200).end();
  handleEvent(event);
});

import hashlib
import hmac
import os
import time

from flask import Flask, request

TOLERANCE_SECONDS = 300


def verify_panoflow_signature(raw_body, header, secret, now=None):
    timestamp = None
    signatures = []
    for part in header.split(","):
        key, sep, value = part.partition("=")
        if not sep:
            continue
        if key == "t":
            timestamp = value
        elif key == "v1":
            signatures.append(value)
        # Ignore other schemes, such as a future v2.
    if timestamp is None or not timestamp.isdigit() or not signatures:
        return False
    if now is None:
        now = int(time.time())
    if abs(now - int(timestamp)) > TOLERANCE_SECONDS:
        return False

    expected = hmac.new(
        secret.encode("utf-8"),
        timestamp.encode("ascii") + b"." + raw_body,  # the raw body bytes
        hashlib.sha256,
    ).hexdigest()
    return any(hmac.compare_digest(expected.encode(), sig.encode()) for sig in signatures)


app = Flask(__name__)


@app.post("/webhooks/panoflow")
def panoflow_webhook():
    raw_body = request.get_data()  # bytes, exactly as received
    header = request.headers.get("PanoFlow-Signature", "")
    if not verify_panoflow_signature(raw_body, header, os.environ["PANOFLOW_WEBHOOK_SECRET"]):
        return "", 400
    event = request.get_json()
    # Skip event["id"] values you've already handled, and keep slow work out of the request.
    enqueue(event)
    return "", 200