How to Test Plaid Webhooks

{
  "webhook_type": "TRANSACTIONS",
  "webhook_code": "DEFAULT_UPDATE",
  "item_id": "item_4NW9bQK2m9Tr8A",
  "new_transactions": 3,
  "removed_transactions": [],
  "environment": "sandbox"
}

// npm i jsonwebtoken jwk-to-pem
import jwt from 'jsonwebtoken';
import jwkToPem from 'jwk-to-pem';
import crypto from 'node:crypto';
import express from 'express';
import { PlaidApi } from 'plaid';

const plaidClient = new PlaidApi(/* ... */);
const app = express();

app.post(
  '/webhooks/plaid',
  express.raw({ type: 'application/json' }),
  async (req, res) => {
    const token = req.headers['plaid-verification'] as string | undefined;
    if (!token) return res.status(401).send('missing token');

    // Inspect JWT header to get kid.
    const [headerB64] = token.split('.');
    const header = JSON.parse(Buffer.from(headerB64, 'base64').toString());
    if (header.alg !== 'ES256') return res.status(401).send('wrong alg');

    // Fetch JWK from Plaid using the kid.
    const { data } = await plaidClient.webhookVerificationKeyGet({ key_id: header.kid });
    const pem = jwkToPem(data.key as jwkToPem.JWK);

    // Verify the JWT signature + iat freshness (5 min tolerance).
    const decoded = jwt.verify(token, pem, {
      algorithms: ['ES256'],
      maxAge: '5m',
    }) as { request_body_sha256: string };

    // Check the body hash matches what was signed.
    const bodyHash = crypto.createHash('sha256').update(req.body).digest('hex');
    if (
      bodyHash.length !== decoded.request_body_sha256.length ||
      !crypto.timingSafeEqual(
        Buffer.from(bodyHash),
        Buffer.from(decoded.request_body_sha256),
      )
    ) {
      return res.status(403).send('body hash mismatch');
    }
    res.json({ ok: true });
  },
);

# pip install python-jose plaid-python
import hashlib, os, time
from jose import jwt as jose_jwt
from plaid.api import plaid_api
from fastapi import FastAPI, Request, HTTPException

app = FastAPI()
plaid_client = plaid_api.PlaidApi(...)  # configured per Plaid SDK docs

@app.post("/webhooks/plaid")
async def plaid_webhook(request: Request):
    body = await request.body()
    token = request.headers.get('plaid-verification')
    if not token:
        raise HTTPException(status_code=401, detail='missing token')

    # 1. Inspect header to get kid + confirm ES256.
    header = jose_jwt.get_unverified_header(token)
    if header.get('alg') != 'ES256':
        raise HTTPException(status_code=401, detail='wrong alg')

    # 2. Fetch JWK from Plaid.
    resp = plaid_client.webhook_verification_key_get({'key_id': header['kid']})
    jwk = resp['key']

    # 3. Verify JWT (ES256) — python-jose accepts JWK dicts directly.
    claims = jose_jwt.decode(token, jwk, algorithms=['ES256'])

    # 4. Replay window + body hash check.
    if abs(int(time.time()) - claims.get('iat', 0)) > 60 * 5:
        raise HTTPException(status_code=401, detail='stale request')
    body_hash = hashlib.sha256(body).hexdigest()
    if body_hash != claims.get('request_body_sha256'):
        raise HTTPException(status_code=403, detail='body hash mismatch')
    return {'ok': True}