How to Test HubSpot Webhooks

{
  "eventId": 18273645,
  "subscriptionType": "deal.propertyChange",
  "portalId": 1234567,
  "occurredAt": 1763582400000,
  "objectId": 987654321,
  "propertyName": "dealstage",
  "propertyValue": "closedwon"
}

import crypto from 'node:crypto';
import express from 'express';

const app = express();

app.post(
  '/webhooks/hubspot',
  express.raw({ type: 'application/json' }),
  (req, res) => {
    const signature = req.headers['x-hubspot-signature-v3'] as string | undefined;
    const timestamp = req.headers['x-hubspot-request-timestamp'] as string | undefined;
    if (!signature || !timestamp) return res.status(401).send('missing headers');

    // Replay protection — reject requests older than 5 minutes.
    const fiveMinutes = 1000 * 60 * 5;
    if (Math.abs(Date.now() - Number(timestamp)) > fiveMinutes) {
      return res.status(401).send('stale request');
    }

    // Reconstruct the URI HubSpot signed (decode percent-encoded chars).
    const uri = decodeURIComponent(req.originalUrl);
    const stringToSign =
      req.method + 'https://' + req.headers.host + uri + req.body.toString('utf8') + timestamp;

    const expected = crypto
      .createHmac('sha256', process.env.HUBSPOT_CLIENT_SECRET!)
      .update(stringToSign)
      .digest('base64');

    if (
      signature.length !== expected.length ||
      !crypto.timingSafeEqual(Buffer.from(signature), Buffer.from(expected))
    ) {
      return res.status(403).send('invalid signature');
    }
    res.json({ ok: true });
  },
);

import hmac, hashlib, base64, os, time
from urllib.parse import unquote
from fastapi import FastAPI, Request, HTTPException

app = FastAPI()

@app.post("/webhooks/hubspot")
async def hubspot_webhook(request: Request):
    body = await request.body()
    signature = request.headers.get('x-hubspot-signature-v3', '')
    timestamp = request.headers.get('x-hubspot-request-timestamp', '0')

    if abs(int(time.time() * 1000) - int(timestamp)) > 1000 * 60 * 5:
        raise HTTPException(status_code=401, detail='stale request')

    uri = unquote(str(request.url))
    string_to_sign = (request.method + uri + body.decode() + timestamp).encode()
    expected = base64.b64encode(
        hmac.new(
            os.environ['HUBSPOT_CLIENT_SECRET'].encode(),
            msg=string_to_sign,
            digestmod=hashlib.sha256,
        ).digest()
    ).decode()
    if not hmac.compare_digest(signature, expected):
        raise HTTPException(status_code=403, detail='invalid signature')
    return {'ok': True}