Verifying webhook signatures

{
  "id": "evt_01JA3…",
  "type": "shipment.status_changed",
  "created_at": "2026-10-12T10:42:13+05:30",
  "org_id": "c0a8…",
  "data": {
    "shipment": { "…": "the shipment, as in the API" },
    "previous_status": "in_transit",
    "status": "out_for_delivery",
    "event": {"at": "2026-10-12T08:02:00+05:30", "location": "PUNE", "activity": "Out For Delivery"},
    "sequence": 7,
    "provisional": false
  }
}

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

const SECRET = process.env.TRACKPARCEL_WEBHOOK_SECRET; // the whole secret, whsec_…
const TOLERANCE_SECONDS = 300;

export function verifyTrackParcelSignature(rawBody, header, secret = SECRET) {
  if (!header) return false;
  let timestamp = null;
  const signatures = [];
  for (const part of header.split(",")) {
    const [key, value] = part.trim().split("=", 2);
    if (key === "t") timestamp = value;
    if (key === "v1" && value) signatures.push(value);
  }
  if (!timestamp || !/^\d+$/.test(timestamp) || signatures.length === 0) return false;
  if (Math.abs(Date.now() / 1000 - Number(timestamp)) > TOLERANCE_SECONDS) return false;

  const expected = crypto
    .createHmac("sha256", secret)
    .update(timestamp + ".")
    .update(rawBody) // the exact bytes received, before any JSON parsing
    .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("/webhooks/trackparcel", express.raw({ type: "application/json" }), (req, res) => {
  if (!verifyTrackParcelSignature(req.body, req.get("TrackParcel-Signature"))) {
    return res.status(400).send("Invalid signature");
  }
  const event = JSON.parse(req.body.toString("utf8"));
  // Skip events you have seen (event.id), queue the work, and answer quickly.
  res.sendStatus(200);
});

<?php
// The whole secret as shown, whsec_…
$secret = getenv('TRACKPARCEL_WEBHOOK_SECRET');
$payload = file_get_contents('php://input'); // the exact bytes received
$header = $_SERVER['HTTP_TRACKPARCEL_SIGNATURE'] ?? '';

function trackparcel_verify(string $payload, string $header, string $secret, int $tolerance = 300): bool
{
    $timestamp = null;
    $signatures = [];
    foreach (explode(',', $header) as $part) {
        $pair = explode('=', trim($part), 2);
        if (count($pair) !== 2) {
            continue;
        }
        if ($pair[0] === 't') {
            $timestamp = $pair[1];
        } elseif ($pair[0] === 'v1') {
            $signatures[] = $pair[1];
        }
    }
    if ($timestamp === null || !ctype_digit($timestamp) || count($signatures) === 0) {
        return false;
    }
    if (abs(time() - (int) $timestamp) > $tolerance) {
        return false;
    }
    $expected = hash_hmac('sha256', $timestamp . '.' . $payload, $secret);
    foreach ($signatures as $signature) {
        if (hash_equals($expected, $signature)) {
            return true;
        }
    }
    return false;
}

if (!trackparcel_verify($payload, $header, $secret)) {
    http_response_code(400);
    exit('Invalid signature');
}

$event = json_decode($payload, true);
// Skip events you have seen ($event['id']), queue the work, and answer quickly.
http_response_code(200);

import hashlib
import hmac
import os
import time

from flask import Flask, abort, request

app = Flask(__name__)
SECRET = os.environ["TRACKPARCEL_WEBHOOK_SECRET"]  # the whole secret, whsec_...
TOLERANCE_SECONDS = 300


def verify_trackparcel_signature(payload: bytes, header: str, secret: str = SECRET) -> bool:
    timestamp, signatures = None, []
    for part in header.split(","):
        key, _, value = part.strip().partition("=")
        if key == "t":
            timestamp = value
        elif key == "v1" and value:
            signatures.append(value)
    if not timestamp or not timestamp.isdigit() or not signatures:
        return False
    if abs(time.time() - int(timestamp)) > TOLERANCE_SECONDS:
        return False
    expected = hmac.new(secret.encode(), timestamp.encode() + b"." + payload, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(expected, signature) for signature in signatures)


@app.post("/webhooks/trackparcel")
def trackparcel_webhook():
    payload = request.get_data()  # the exact bytes received
    if not verify_trackparcel_signature(payload, request.headers.get("TrackParcel-Signature", "")):
        abort(400)
    event = request.get_json()
    # Skip events you have seen (event["id"]), queue the work, and answer quickly.
    return "", 200