Securing your webhook endpoint

GET https://api.sandbox.ibanq.com/.well-known/jwks.json

GET https://api.ibanq.com/.well-known/jwks.json

{
  "keys": [
    {
      "kty": "RSA",
      "use": "sig",
      "alg": "PS256",
      "kid": "70WOedgzv6hJ63CiLF9TDh6L5XBoL7c71svrGglv65o",
      "n": "sN1Yq...",
      "e": "AQAB"
    }
  ]
}

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

const KEY_SET_URL = 'https://api.sandbox.ibanq.com/.well-known/jwks.json';
const TOLERANCE_SECONDS = 300;

let keys = new Map();
let lastAttemptAt = 0;
let etag = null;

async function loadKeys() {
  if (Date.now() - lastAttemptAt < 60_000) return; // at most one fetch a minute, success or failure
  lastAttemptAt = Date.now();                    // stamp before the call: a failing key set must not hot-loop
  const response = await fetch(KEY_SET_URL, etag ? { headers: { 'if-none-match': etag } } : undefined);
  if (response.status === 304 || !response.ok) return; // 304 means the cache is still good
  etag = response.headers.get('etag');
  const document = await response.json();
  keys = new Map(document.keys.map(jwk => [jwk.kid, crypto.createPublicKey({ key: jwk, format: 'jwk' })]));
}

function parseSignatureInput(header) {
  const params = header.slice('sig1='.length); // ("@method" ...);created=...;keyid="...";alg="..."
  const close = params.indexOf(')');
  const components = params.slice(1, close).split(' ').map(c => c.replaceAll('"', ''));
  const values = Object.fromEntries(
    params.slice(close + 2).split(';').map(pair => {
      const eq = pair.indexOf('=');              // split on the first = only: a value can contain =
      return [pair.slice(0, eq), pair.slice(eq + 1).replaceAll('"', '')];
    }));
  return { params, components, ...values };
}

function digestMatches(body, header) {
  const expected = `sha-256=:${crypto.createHash('sha256').update(body).digest('base64')}:`;
  const a = Buffer.from(header ?? '');
  const b = Buffer.from(expected);
  return a.length === b.length && crypto.timingSafeEqual(a, b);
}

async function verify(req) {
  const body = req.body;                                    // a Buffer: the raw bytes, unparsed
  if (!digestMatches(body, req.get('content-digest'))) return 'digest mismatch';

  const input = req.get('signature-input');
  const signature = req.get('signature');
  if (!input?.startsWith('sig1=') || !signature?.startsWith('sig1=:') || !signature.endsWith(':')) {
    return 'missing sig1';
  }

  const { params, components, created, keyid, alg } = parseSignatureInput(input);
  if (alg !== 'rsa-pss-sha256') return 'unexpected alg';
  if (Math.abs(Math.floor(Date.now() / 1000) - Number(created)) > TOLERANCE_SECONDS) return 'created outside tolerance';

  if (!keys.has(keyid)) await loadKeys();                   // a rotation publishes before it signs
  const key = keys.get(keyid);
  if (!key) return `unknown keyid ${keyid}`;

  let base = '';
  for (const component of components) {
    const value = component === '@method' ? req.method.toUpperCase()
      : component === '@authority' ? req.get('host')
      : component === '@request-target' ? req.originalUrl
      : req.get(component);
    if (value === undefined) return `covered header ${component} missing`;
    base += `"${component}": ${value}\n`;
  }
  base += `"@signature-params": ${params}`;

  const ok = crypto.verify(
    'sha256',
    Buffer.from(base, 'utf8'),
    { key, padding: crypto.constants.RSA_PKCS1_PSS_PADDING, saltLength: crypto.constants.RSA_PSS_SALTLEN_DIGEST },
    Buffer.from(signature.slice('sig1=:'.length, -1), 'base64'));

  return ok ? null : 'signature invalid';
}

const app = express();
await loadKeys();

app.post('/your/webhook/path', express.raw({ type: 'application/json' }), async (req, res) => {
  const failure = await verify(req);
  if (failure) {
    console.warn('rejected webhook', failure);
    return res.sendStatus(401);
  }

  const event = JSON.parse(req.body.toString('utf8'));
  await enqueue(event);            // persist by event.id, process out of band
  res.sendStatus(200);             // answer fast; any 2xx counts
});

from __future__ import annotations

import base64, hashlib, hmac, json, time, urllib.error, urllib.request
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import padding, rsa

KEY_SET_URL = "https://api.sandbox.ibanq.com/.well-known/jwks.json"
TOLERANCE_SECONDS = 300

_keys: dict[str, rsa.RSAPublicKey] = {}
_attempted_at = 0.0
_etag: str | None = None


def _b64url(value: str) -> bytes:
    return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))


def load_keys() -> None:
    global _keys, _attempted_at, _etag
    if time.time() - _attempted_at < 60:  # at most one fetch a minute, success or failure
        return
    _attempted_at = time.time()  # stamp before the call: a failing key set must not hot-loop
    request = urllib.request.Request(KEY_SET_URL)
    if _etag:
        request.add_header("If-None-Match", _etag)
    try:
        with urllib.request.urlopen(request, timeout=5) as response:
            _etag = response.headers.get("ETag")
            document = json.load(response)
    except urllib.error.HTTPError as error:
        if error.code == 304:  # the cache is still good
            return
        raise
    _keys = {
        jwk["kid"]: rsa.RSAPublicNumbers(
            int.from_bytes(_b64url(jwk["e"]), "big"),
            int.from_bytes(_b64url(jwk["n"]), "big"),
        ).public_key()
        for jwk in document["keys"]
    }

def verify(method: str, authority: str, request_target: str, header, body: bytes) -> str | None:
    """Returns None when the delivery is genuine, otherwise the reason it is not."""
    expected = "sha-256=:" + base64.b64encode(hashlib.sha256(body).digest()).decode() + ":"
    if not hmac.compare_digest(header("content-digest") or "", expected):
        return "digest mismatch"

    signature_input = header("signature-input") or ""
    signature = header("signature") or ""
    if not signature_input.startswith("sig1=") or not signature.startswith("sig1=:") or not signature.endswith(":"):
        return "missing sig1"

    params = signature_input[len("sig1="):]
    close = params.index(")")
    components = [c.strip('"') for c in params[1:close].split()]
    values = dict(
        pair.split("=", 1) for pair in params[close + 2:].split(";")
    )
    values = {k: v.strip('"') for k, v in values.items()}

    if values.get("alg") != "rsa-pss-sha256":
        return "unexpected alg"
    if abs(time.time() - int(values.get("created", 0))) > TOLERANCE_SECONDS:
        return "created outside tolerance"

    key_id = values.get("keyid")
    if key_id not in _keys:
        try:
            load_keys()  # a rotation publishes before it signs
        except Exception:
            return "key set unavailable"  # fail closed: answer 5xx and let the retry ladder bring it back
    key = _keys.get(key_id)
    if key is None:
        return f"unknown keyid {key_id}"

    lines = []
    for component in components:
        if component == "@method":
            value = method.upper()
        elif component == "@authority":
            value = authority
        elif component == "@request-target":
            value = request_target
        else:
            value = header(component)
        if value is None:
            return f"covered header {component} missing"
        lines.append(f'"{component}": {value}')
    base = "\n".join(lines) + f'\n"@signature-params": {params}'

    try:
        key.verify(
            base64.b64decode(signature[len("sig1=:"):-1]),
            base.encode("utf-8"),
            padding.PSS(mgf=padding.MGF1(hashes.SHA256()), salt_length=32),
            hashes.SHA256(),
        )
    except Exception:
        return "signature invalid"
    return None

from flask import Flask, request

app = Flask(__name__)
load_keys()

@app.post("/your/webhook/path")
def receive():
    failure = verify(
        request.method,
        request.host,                   # the host we sent to; see "Mind the proxy" above
        request.full_path.rstrip("?"),  # path and query, exactly as registered
        request.headers.get,
        request.get_data(),             # raw bytes, before any JSON parsing
    )
    if failure:
        app.logger.warning("rejected webhook: %s", failure)
        return "", 401

    event = json.loads(request.get_data())
    enqueue(event)                      # persist by event["id"], process out of band
    return "", 200                      # answer fast; any 2xx counts