Verify PagerDuty webhook signature headers, replay defense, rotation

// Minimal PagerDuty verification (Express raw body)
// npm i express
import crypto from "crypto";
import express from "express";

const app = express();

app.post("/webhook", express.raw({ type: "*/*", limit: "2mb" }), (req, res) => {
  const secret = process.env.WEBHOOK_SECRET ?? "replace_me";
  const signatureHeader = req.header("x-pagerduty-signature") ?? "";
  const rawBody = Buffer.isBuffer(req.body) ? req.body : Buffer.from(req.body ?? "");
  const payload = rawBody.toString("utf8");
  if (!signatureHeader) return res.status(401).send("missing x-pagerduty-signature");

  const expected = crypto.createHmac("sha256", secret).update(`v1:${payload}`).digest("hex");

  // Header can contain multiple signatures: v1=xxxx,v1=yyyy
  const signatures = signatureHeader.split(",");
  for (const sig of signatures) {
    const [version, value] = sig.trim().split("=");
    if (version !== "v1") continue;
    if (timingSafeEqualString(value, expected)) return res.sendStatus(202);
  }

  return res.status(401).send("bad signature");
});

app.listen(3000, () => console.log("listening on :3000"));

function timingSafeEqualString(a, b) {
  const aBuf = Buffer.from(a, "utf8");
  const bBuf = Buffer.from(b, "utf8");
  if (aBuf.length !== bBuf.length) return false;
  return crypto.timingSafeEqual(aBuf, bBuf);
}

# Minimal PagerDuty verification (FastAPI raw body)
# pip install fastapi uvicorn
import hashlib
import hmac
import os
from fastapi import FastAPI, Request, Response

app = FastAPI()

@app.post("/webhook")
async def webhook(request: Request) -> Response:
    secret = (os.getenv("WEBHOOK_SECRET") or "replace_me").encode("utf-8")
    signature_header = request.headers.get("x-pagerduty-signature", "")
    raw_body = await request.body()
    payload = raw_body.decode("utf-8")
    if not signature_header:
        return Response("missing x-pagerduty-signature", status_code=401)

    expected = hmac.new(secret, f"v1:{payload}".encode("utf-8"), hashlib.sha256).hexdigest()
    for part in signature_header.split(","):
        version, _, value = part.strip().partition("=")
        if version != "v1":
            continue
        if hmac.compare_digest(value, expected):
            return Response(status_code=202)

    return Response("bad signature", status_code=401)

// Replay protection: timestamp max-age (when available) + dedupe window (in-memory example)
import crypto from "crypto";

const seen = new Map(); // key -> expiresAtMs

function purgeSeen() {
  const now = Date.now();
  for (const [key, expiresAt] of seen.entries()) if (expiresAt <= now) seen.delete(key);
}

// Inside your verified webhook handler:
purgeSeen();
const sig = req.header("x-pagerduty-signature") ?? "";

// Prefer a provider delivery/event ID when available. Otherwise hash signature + raw body.
const rawBody = Buffer.isBuffer(req.body) ? req.body : Buffer.from(req.body ?? "");
const key = crypto.createHash("sha256").update(sig).update(".").update(rawBody).digest("hex");
const ttlMs = 10 * 60 * 1000;
if (seen.has(key)) return res.status(409).send("replay detected");
seen.set(key, Date.now() + ttlMs);

# Replay protection: timestamp max-age (when available) + dedupe window (in-memory example)
import hashlib
import time

seen: dict[str, float] = {}  # key -> expires_at_epoch_seconds

def purge_seen() -> None:
    now = time.time()
    for k, exp in list(seen.items()):
        if exp <= now:
            del seen[k]

# Inside your verified webhook handler:
purge_seen()
sig = request.headers.get("x-pagerduty-signature", "")

raw_body = await request.body()
key = hashlib.sha256((sig + ".").encode("utf-8") + raw_body).hexdigest()
ttl_seconds = 10 * 60
if key in seen:
    return Response("replay detected", status_code=409)
seen[key] = time.time() + ttl_seconds

// PagerDuty secret rotation: try {current, previous} during overlap
import crypto from "crypto";

function timingSafeEqualString(a, b) {
  const aBuf = Buffer.from(a, "utf8");
  const bBuf = Buffer.from(b, "utf8");
  if (aBuf.length !== bBuf.length) return false;
  return crypto.timingSafeEqual(aBuf, bBuf);
}

function verifyPagerDutyWithSecrets(payload, signatureHeader, secrets) {
  const signatures = signatureHeader.split(",");
  for (const secret of secrets) {
    const expected = crypto.createHmac("sha256", secret).update(`v1:${payload}`).digest("hex");
    for (const sig of signatures) {
      const [version, value] = sig.trim().split("=");
      if (version !== "v1") continue;
      if (timingSafeEqualString(value, expected)) return true;
    }
  }
  return false;
}

const secrets = [process.env.WEBHOOK_SECRET_CURRENT, process.env.WEBHOOK_SECRET_PREVIOUS].filter(Boolean);

# PagerDuty secret rotation: try {current, previous} during overlap
import hashlib
import hmac
import os

def verify_pagerduty_with_secrets(payload: str, signature_header: str, secrets: list[bytes]) -> bool:
    signatures = [p.strip() for p in signature_header.split(",") if p.strip()]
    for secret in secrets:
        expected = hmac.new(secret, f"v1:{payload}".encode("utf-8"), hashlib.sha256).hexdigest()
        for part in signatures:
            version, _, value = part.partition("=")
            if version == "v1" and hmac.compare_digest(value, expected):
                return True
    return False

secrets = [os.getenv("WEBHOOK_SECRET_CURRENT"), os.getenv("WEBHOOK_SECRET_PREVIOUS")]
secrets = [s.encode("utf-8") for s in secrets if s]