Verify Webhook Signatures

message = timestamp + "." + raw_request_body

import { createVerify, createPublicKey } from "crypto";

function verifyWebhookSignature(opts: {
  rawBody: string | Buffer;
  signatureBase64: string;
  timestamp: string;
  publicKeyPem: string;
}): boolean {
  const message = `${opts.timestamp}.${opts.rawBody.toString()}`;
  const verifier = createVerify("RSA-SHA256");
  verifier.update(message);
  verifier.end();
  return verifier.verify(
    createPublicKey(opts.publicKeyPem),
    opts.signatureBase64,
    "base64",
  );
}

// Express-style: use express.raw() or capture raw body before JSON middleware
export function webhookHandler(req: {
  body: Buffer;
  headers: Record<string, string | string[] | undefined>;
}, res: { status: (n: number) => { send: (s: string) => void } }) {
  const signature = String(req.headers["x-boomfi-signature"] ?? "");
  const timestamp = String(req.headers["x-boomfi-timestamp"] ?? "");
  const publicKeyPem = process.env.WEBHOOK_PUBLIC_KEY!.replace(/\\n/g, "\n");

  const ts = Number(timestamp);
  if (!Number.isFinite(ts) || Math.abs(Date.now() / 1000 - ts) > 300) {
    return res.status(401).send("stale timestamp");
  }

  const ok = verifyWebhookSignature({
    rawBody: req.body,
    signatureBase64: signature,
    timestamp,
    publicKeyPem,
  });
  if (!ok) return res.status(401).send("invalid signature");

  const payload = JSON.parse(req.body.toString("utf8"));
  // Assert org_id matches your organisation; handle payload.event idempotently
  return res.status(200).send("ok");
}

import base64
import hashlib
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.exceptions import InvalidSignature


def verify_signature(public_key_pem: str, message: bytes, signature_base64: str) -> None:
    public_key = serialization.load_pem_public_key(public_key_pem.encode("utf-8"))
    signature = base64.b64decode(signature_base64)
    public_key.verify(
        signature,
        message,
        padding.PKCS1v15(),
        hashes.SHA256(),
    )


def verify_webhook(raw_body: bytes, timestamp: str, signature_b64: str, public_key_pem: str) -> None:
    message = f"{timestamp}.".encode("utf-8") + raw_body
    try:
        verify_signature(public_key_pem, message, signature_b64)
    except InvalidSignature as exc:
        raise ValueError("invalid signature") from exc

package webhooks

import (
	"crypto"
	"crypto/rsa"
	"crypto/sha256"
	"crypto/x509"
	"encoding/base64"
	"encoding/pem"
	"errors"
	"fmt"
)

func VerifySignature(publicKeyPEM string, message []byte, signatureBase64 string) error {
	block, _ := pem.Decode([]byte(publicKeyPEM))
	if block == nil || block.Type != "PUBLIC KEY" {
		return errors.New("failed to decode PEM block containing public key")
	}
	publicKey, err := x509.ParsePKIXPublicKey(block.Bytes)
	if err != nil {
		return fmt.Errorf("failed to parse public key: %w", err)
	}
	rsaPublicKey, ok := publicKey.(*rsa.PublicKey)
	if !ok {
		return errors.New("not an RSA public key")
	}
	signature, err := base64.StdEncoding.DecodeString(signatureBase64)
	if err != nil {
		return fmt.Errorf("failed to decode signature: %w", err)
	}
	h := sha256.New()
	h.Write(message)
	return rsa.VerifyPKCS1v15(rsaPublicKey, crypto.SHA256, h.Sum(nil), signature)
}

func VerifyWebhook(rawBody []byte, timestamp, signatureB64, publicKeyPEM string) error {
	message := append([]byte(timestamp+"."), rawBody...)
	return VerifySignature(publicKeyPEM, message, signatureB64)
}