Verifying Bastion webhook signatures

{
  "signatures": [
    { "protected": "<base64url-encoded protected header>", "signature": "<base64url-encoded HMAC-SHA256>" }
  ]
}

{
  "alg": "HS256",
  "typ": "webhook+jws",
  "kid": "2cFp7yShQ8vN3kRtZ1wL9bGmJxY",
  "iat": 1714501234,
  "jti": "msg_2cFp9aBdE4gH6jK8mN0pQ2rS",
  "b64": false,
  "crit": ["b64"]
}

import (
    "encoding/base64"
    "encoding/json"
    "errors"
    "io"
    "net/http"
    "strings"
    "time"

    "github.com/go-jose/go-jose/v4"
)

const tolerance = 5 * time.Minute

// DecodeSecret turns the dashboard secret ("whsign_<base64url>") into the raw
// HMAC key bytes. Store the result keyed by kid.
func DecodeSecret(s string) ([]byte, error) {
    return base64.RawURLEncoding.DecodeString(strings.TrimPrefix(s, "whsign_"))
}

// keysByKid maps each kid to its DECODED secret bytes (see DecodeSecret).
func Verify(r *http.Request, keysByKid map[string][]byte) ([]byte, error) {
    headerValue := r.Header.Get("Webhook-Signature")
    if headerValue == "" {
        return nil, errors.New("missing Webhook-Signature header")
    }
    body, err := io.ReadAll(r.Body)
    if err != nil {
        return nil, err
    }
    // Pin alg=HS256 to reject alg:none and future-algorithm confusion.
    sig, err := jose.ParseSigned(headerValue, []jose.SignatureAlgorithm{jose.HS256})
    if err != nil {
        return nil, err
    }
    for _, entry := range sig.Signatures {
        // Enforce the iat freshness window before trusting the signature.
        var hdr struct {
            Iat int64  `json:"iat"`
            Kid string `json:"kid"`
        }
        if err := json.Unmarshal(entry.Protected, &hdr); err != nil {
            continue
        }
        if delta := time.Since(time.Unix(hdr.Iat, 0)); delta > tolerance || delta < -tolerance {
            continue
        }
        secret, ok := keysByKid[hdr.Kid]
        if !ok {
            continue
        }
        // DetachedVerify recomputes the HMAC over base64url(protected) + "." + body.
        if _, err := sig.DetachedVerify(body, secret); err == nil {
            return body, nil
        }
    }
    return nil, errors.New("no valid signature")
}

import { generalVerify } from 'jose';

const TOLERANCE_MS = 5 * 60 * 1000;

// Turn the dashboard secret ("whsign_<base64url>") into the raw HMAC key bytes.
function decodeSecret(s: string): Uint8Array {
  return new Uint8Array(Buffer.from(s.replace(/^whsign_/, ''), 'base64url'));
}

// keysByKid maps each kid to its DECODED secret bytes (see decodeSecret).
export async function verify(
  headers: Record<string, string>,
  body: string,
  keysByKid: Record<string, Uint8Array>,
): Promise<boolean> {
  const headerValue = headers['webhook-signature'];
  if (!headerValue) return false;
  let env;
  try {
    env = JSON.parse(headerValue);
  } catch {
    return false;
  }
  if (!Array.isArray(env.signatures)) return false;

  for (const entry of env.signatures) {
    let hdr;
    try {
      hdr = JSON.parse(Buffer.from(entry.protected, 'base64url').toString('utf8'));
    } catch {
      continue;
    }
    // iat freshness — check before calling the library verify.
    if (Math.abs(Date.now() - hdr.iat * 1000) > TOLERANCE_MS) continue;
    const secret = keysByKid[hdr.kid];
    if (!secret) continue;
    try {
      await generalVerify({ ...env, payload: new TextEncoder().encode(body), signatures: [entry] }, secret, {
        algorithms: ['HS256'],
      });
      return true;
    } catch {
      // Try the next entry (e.g. during a rotation overlap).
    }
  }
  return false;
}

import base64
import json
import time

from authlib.jose import JsonWebSignature

TOLERANCE = 5 * 60  # seconds

def decode_secret(s: str) -> bytes:
    """Dashboard secret ('whsign_<base64url>') → raw HMAC key bytes."""
    raw = s.removeprefix("whsign_")
    return base64.urlsafe_b64decode(raw + "=" * (-len(raw) % 4))

def verify(headers, body, keys_by_kid):
    """keys_by_kid maps kid -> DECODED secret bytes (see decode_secret)."""
    header_value = headers.get("Webhook-Signature")
    if not header_value:
        return False
    try:
        env = json.loads(header_value)
    except (ValueError, TypeError):
        return False
    if not isinstance(env.get("signatures"), list):
        return False
    jws = JsonWebSignature(algorithms=["HS256"])  # algorithm pinning
    for entry in env["signatures"]:
        try:
            protected = json.loads(
                base64.urlsafe_b64decode(entry["protected"] + "=" * (-len(entry["protected"]) % 4))
            )
        except Exception:
            continue
        if abs(time.time() - protected["iat"]) > TOLERANCE:
            continue
        secret = keys_by_kid.get(protected["kid"])
        if not secret:
            continue
        try:
            jws.deserialize_json(
                {"payload": "", "signatures": [entry]}, secret, payload=body
            )
            return True
        except Exception:
            continue
    return False

import base64
import hmac
import json
import time
from hashlib import sha256

TOLERANCE = 5 * 60  # seconds

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

def decode_secret(s: str) -> bytes:
    return b64url_decode(s.removeprefix("whsign_"))

def verify(raw_body: bytes, header_value: str, secrets_by_kid: dict) -> bool:
    env = json.loads(header_value)
    for entry in env["signatures"]:
        protected = json.loads(b64url_decode(entry["protected"]))

        if protected["alg"] != "HS256":                 # algorithm pinning
            continue
        if abs(time.time() - protected["iat"]) > TOLERANCE:  # iat freshness window
            continue
        secret = secrets_by_kid.get(protected["kid"])    # raw bytes via decode_secret
        if secret is None:
            continue

        # signing input = base64url(protected) + "." + raw body bytes
        signing_input = entry["protected"].encode("ascii") + b"." + raw_body
        expected = hmac.new(secret, signing_input, sha256).digest()
        got = b64url_decode(entry["signature"])

        if hmac.compare_digest(got, expected):           # constant-time compare
            return True
    return False  # caller responds 4xx, NOT 200

import { createHmac, timingSafeEqual } from 'crypto';

function decodeSecret(s: string): Buffer {
  return Buffer.from(s.replace(/^whsign_/, ''), 'base64url');
}

function verify(rawBody: Buffer, headerValue: string, secretsByKid: Record<string, Buffer>): boolean {
  const env = JSON.parse(headerValue);
  for (const entry of env.signatures) {
    const hdr = JSON.parse(Buffer.from(entry.protected, 'base64url').toString('utf8'));
    if (hdr.alg !== 'HS256') continue;                          // algorithm pinning
    if (Math.abs(Date.now() / 1000 - hdr.iat) > 300) continue;  // iat freshness window
    const secret = secretsByKid[hdr.kid];
    if (!secret) continue;

    const signingInput = Buffer.concat([Buffer.from(entry.protected, 'ascii'), Buffer.from('.'), rawBody]);
    const expected = createHmac('sha256', secret).update(signingInput).digest();
    const got = Buffer.from(entry.signature, 'base64url');

    if (got.length === expected.length && timingSafeEqual(got, expected)) return true;
  }
  return false; // caller responds 4xx, NOT 200
}