Verifying Signatures

const crypto = require("crypto");

function verifyWebhook(rawBody, signatureHeader, secretHex, toleranceSeconds = 300) {
  // Parse "t=...,v1=..."
  const fields = {};
  for (const part of signatureHeader.split(",")) {
    const i = part.indexOf("=");
    fields[part.slice(0, i)] = part.slice(i + 1);
  }
  const timestamp = Number(fields.t);
  const provided = fields.v1;
  if (!timestamp || !provided) return false;

  // Reject stale timestamps to blunt replay attacks.
  if (Math.abs(Date.now() / 1000 - timestamp) > toleranceSeconds) return false;

  // The secret is hex-encoded; decode it to bytes for the HMAC key.
  const key = Buffer.from(secretHex, "hex");
  const expected = crypto
    .createHmac("sha256", key)
    .update(`${timestamp}.${rawBody}`)
    .digest("hex");

  const a = Buffer.from(expected);
  const b = Buffer.from(provided);
  return a.length === b.length && crypto.timingSafeEqual(a, b);
}

import hashlib
import hmac
import time


def verify_webhook(raw_body: bytes, signature_header: str, secret_hex: str, tolerance_seconds: int = 300) -> bool:
    # Parse "t=...,v1=..."
    try:
        fields = dict(part.split("=", 1) for part in signature_header.split(","))
        timestamp = int(fields["t"])
        provided = fields["v1"]
    except (KeyError, ValueError):
        return False

    # Reject stale timestamps to blunt replay attacks.
    if abs(time.time() - timestamp) > tolerance_seconds:
        return False

    # The secret is hex-encoded; decode it to bytes for the HMAC key.
    key = bytes.fromhex(secret_hex)
    signed = f"{timestamp}.".encode() + raw_body  # raw body, exactly as received
    expected = hmac.new(key, signed, hashlib.sha256).hexdigest()

    return hmac.compare_digest(expected, provided)

const express = require("express");
const app = express();

// Capture the raw body — required for signature verification.
app.use("/webhooks/stockinsights", express.raw({ type: "application/json" }));

app.post("/webhooks/stockinsights", (req, res) => {
  const signature = req.get("X-StockInsights-Signature");
  const rawBody = req.body.toString("utf8");

  // 503, not 400: a signature mismatch is usually a secret that has not rolled
  // out yet, and only a 5xx gets retried. A 4xx drops the event permanently.
  if (!verifyWebhook(rawBody, signature, process.env.STOCKINSIGHTS_WEBHOOK_SECRET)) {
    return res.status(503).send("invalid signature");
  }

  const event = JSON.parse(rawBody);
  // Acknowledge fast; process asynchronously.
  res.sendStatus(200);

  handleEvent(event); // your logic
});

from flask import Flask, request, abort

app = Flask(__name__)


@app.post("/webhooks/stockinsights")
def stockinsights_webhook():
    signature = request.headers.get("X-StockInsights-Signature", "")
    raw_body = request.get_data()  # bytes, exactly as received

    # 503, not 400: a signature mismatch is usually a secret that has not rolled
    # out yet, and only a 5xx gets retried. A 4xx drops the event permanently.
    if not verify_webhook(raw_body, signature, WEBHOOK_SECRET):
        abort(503, "invalid signature")

    event = request.get_json()
    # Acknowledge fast; process asynchronously.
    handle_event(event)  # your logic
    return "", 200