Verify webhook signatures

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

const SECRET = process.env.PANTECH_WEBHOOK_SECRET; // whsec_…, the whole string
const TOLERANCE_SECONDS = 5 * 60;

// True if `header` holds a valid signature of `rawBody` (a Buffer) made with
// `secret` within the last five minutes.
function verifyPantechSignature(rawBody, header, secret, now = Math.floor(Date.now() / 1000)) {
  if (typeof header !== "string") return false;
  let timestamp = null;
  const signatures = [];
  for (const part of header.split(",")) {
    const eq = part.indexOf("=");
    if (eq === -1) continue;
    const key = part.slice(0, eq).trim();
    const value = part.slice(eq + 1).trim();
    if (key === "t") timestamp = value;
    else if (key === "v1") signatures.push(value);
  }
  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", secret).update(`${timestamp}.`).update(rawBody).digest();
  return signatures.some((signature) => {
    const given = Buffer.from(signature, "hex");
    return given.length === expected.length && crypto.timingSafeEqual(given, expected);
  });
}

const app = express();

app.post("/hooks/pantech", express.raw({ type: "application/json" }), (req, res) => {
  if (!verifyPantechSignature(req.body, req.get("Pantech-Signature"), SECRET)) {
    return res.status(400).send("invalid signature");
  }
  const event = JSON.parse(req.body.toString("utf8"));
  res.sendStatus(204); // answer first, then work
  handle(event).catch((err) => console.error("webhook", event.id, err));
});

async function handle(event) {
  // Skip an event.id you have already handled: delivery is at least once.
  console.log(event.id, event.type, event.data);
}

app.listen(3000);

import hashlib
import hmac
import json
import os
import time
from typing import Optional

from flask import Flask, abort, request

SECRET = os.environ["PANTECH_WEBHOOK_SECRET"]  # whsec_…, the whole string
TOLERANCE_SECONDS = 5 * 60


def verify_pantech_signature(raw_body: bytes, header: Optional[str], secret: str, now: Optional[float] = None) -> bool:
    """True if `header` holds a valid signature of `raw_body` made with `secret` in the last five minutes."""
    if not header:
        return False
    timestamp, signatures = None, []
    for part in header.split(","):
        key, sep, value = part.strip().partition("=")
        if not sep:
            continue
        if key == "t":
            timestamp = value
        elif key == "v1":
            signatures.append(value)
    if timestamp is None or not (timestamp.isascii() and timestamp.isdigit()) or not signatures:
        return False
    if abs((time.time() if now is None else now) - int(timestamp)) > TOLERANCE_SECONDS:
        return False

    expected = hmac.new(secret.encode(), timestamp.encode() + b"." + raw_body, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(expected.encode(), s.encode()) for s in signatures)


app = Flask(__name__)


@app.post("/hooks/pantech")
def pantech_webhook():
    raw_body = request.get_data()
    if not verify_pantech_signature(raw_body, request.headers.get("Pantech-Signature"), SECRET):
        abort(400)
    event = json.loads(raw_body)
    # Queue the event for a worker, and skip an event["id"] you have already handled.
    print(event["id"], event["type"], event["data"])
    return "", 204

import hashlib
import hmac
import json
import os
import time
from typing import Optional

from fastapi import BackgroundTasks, FastAPI, HTTPException, Request, Response

SECRET = os.environ["PANTECH_WEBHOOK_SECRET"]  # whsec_…, the whole string
TOLERANCE_SECONDS = 5 * 60


def verify_pantech_signature(raw_body: bytes, header: Optional[str], secret: str, now: Optional[float] = None) -> bool:
    """True if `header` holds a valid signature of `raw_body` made with `secret` in the last five minutes."""
    if not header:
        return False
    timestamp, signatures = None, []
    for part in header.split(","):
        key, sep, value = part.strip().partition("=")
        if not sep:
            continue
        if key == "t":
            timestamp = value
        elif key == "v1":
            signatures.append(value)
    if timestamp is None or not (timestamp.isascii() and timestamp.isdigit()) or not signatures:
        return False
    if abs((time.time() if now is None else now) - int(timestamp)) > TOLERANCE_SECONDS:
        return False

    expected = hmac.new(secret.encode(), timestamp.encode() + b"." + raw_body, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(expected.encode(), s.encode()) for s in signatures)


app = FastAPI()


def handle(event: dict) -> None:
    # Skip an event["id"] you have already handled: delivery is at least once.
    print(event["id"], event["type"], event["data"])


@app.post("/hooks/pantech", status_code=204)
async def pantech_webhook(request: Request, background: BackgroundTasks) -> Response:
    raw_body = await request.body()
    if not verify_pantech_signature(raw_body, request.headers.get("pantech-signature"), SECRET):
        raise HTTPException(status_code=400, detail="invalid signature")
    background.add_task(handle, json.loads(raw_body))  # runs after the response is sent
    return Response(status_code=204)

package main

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"io"
	"log"
	"net/http"
	"os"
	"strconv"
	"strings"
	"time"
)

const tolerance = 5 * time.Minute

// verifyPantechSignature checks the Pantech-Signature header against the raw
// request body. It returns nil when one of the v1 signatures matches and the
// timestamp is within five minutes of now.
func verifyPantechSignature(body []byte, header, secret string, now time.Time) error {
	var timestamp string
	var signatures []string
	for _, part := range strings.Split(header, ",") {
		key, value, ok := strings.Cut(strings.TrimSpace(part), "=")
		if !ok {
			continue
		}
		switch key {
		case "t":
			timestamp = value
		case "v1":
			signatures = append(signatures, value)
		}
	}
	seconds, err := strconv.ParseInt(timestamp, 10, 64)
	if err != nil || len(signatures) == 0 {
		return errors.New("malformed Pantech-Signature header")
	}
	if age := now.Sub(time.Unix(seconds, 0)); age > tolerance || age < -tolerance {
		return errors.New("timestamp outside the tolerance")
	}

	mac := hmac.New(sha256.New, []byte(secret))
	mac.Write([]byte(timestamp + "."))
	mac.Write(body)
	expected := mac.Sum(nil)
	for _, signature := range signatures {
		given, err := hex.DecodeString(signature)
		if err == nil && hmac.Equal(given, expected) {
			return nil
		}
	}
	return errors.New("no matching signature")
}

type event struct {
	ID   string          `json:"id"`
	Type string          `json:"type"`
	Data json.RawMessage `json:"data"`
}

func main() {
	secret := os.Getenv("PANTECH_WEBHOOK_SECRET") // whsec_…, the whole string

	http.HandleFunc("POST /hooks/pantech", func(w http.ResponseWriter, r *http.Request) {
		body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, 1<<20))
		if err != nil {
			http.Error(w, "body too large", http.StatusRequestEntityTooLarge)
			return
		}
		if err := verifyPantechSignature(body, r.Header.Get("Pantech-Signature"), secret, time.Now()); err != nil {
			http.Error(w, "invalid signature", http.StatusBadRequest)
			return
		}
		var e event
		if err := json.Unmarshal(body, &e); err != nil {
			http.Error(w, "invalid JSON", http.StatusBadRequest)
			return
		}
		w.WriteHeader(http.StatusNoContent)
		// Hand e to a queue or worker here, and skip an e.ID you have already handled.
		log.Printf("event %s %s %s", e.ID, e.Type, e.Data)
	})

	log.Fatal(http.ListenAndServe(":8080", nil))
}

<?php

const PANTECH_TOLERANCE_SECONDS = 300;

/** True if $header holds a valid signature of $rawBody made with $secret in the last five minutes. */
function verify_pantech_signature(string $rawBody, ?string $header, string $secret, ?int $now = null): bool
{
    if ($header === null || $header === '') {
        return false;
    }
    $timestamp = null;
    $signatures = [];
    foreach (explode(',', $header) as $part) {
        $pair = explode('=', trim($part), 2);
        if (count($pair) !== 2) {
            continue;
        }
        [$key, $value] = $pair;
        if ($key === 't') {
            $timestamp = $value;
        } elseif ($key === 'v1') {
            $signatures[] = $value;
        }
    }
    if ($timestamp === null || !ctype_digit($timestamp) || count($signatures) === 0) {
        return false;
    }
    if (abs(($now ?? time()) - (int) $timestamp) > PANTECH_TOLERANCE_SECONDS) {
        return false;
    }

    $expected = hash_hmac('sha256', $timestamp . '.' . $rawBody, $secret);
    foreach ($signatures as $signature) {
        if (hash_equals($expected, $signature)) {
            return true;
        }
    }
    return false;
}

$secret = getenv('PANTECH_WEBHOOK_SECRET'); // whsec_…, the whole string
$rawBody = file_get_contents('php://input');
$header = $_SERVER['HTTP_PANTECH_SIGNATURE'] ?? null;

if ($secret === false || !verify_pantech_signature($rawBody, $header, $secret)) {
    http_response_code(400);
    exit('invalid signature');
}

$event = json_decode($rawBody, true);
// Queue $event for a worker, and skip an $event['id'] you have already handled.
error_log(sprintf('event %s %s', $event['id'], $event['type']));
http_response_code(204);