Webhook signature verification

// Express example
const crypto = require("crypto");
const express = require("express");

function verifyGroundSignature(rawBody, signatureHeader, signingSecret, toleranceSeconds = 300) {
  if (!signatureHeader) throw new Error("Missing Ground-Signature header");

  const parts = signatureHeader.split(",");
  const tPart = parts.find((p) => p.startsWith("t="));
  const v1Part = parts.find((p) => p.startsWith("v1="));
  if (!tPart || !v1Part) throw new Error("Invalid Ground-Signature header format");

  const timestamp = Number(tPart.split("=")[1]);
  const signature = v1Part.split("=")[1];

  // Optional replay protection
  const now = Math.floor(Date.now() / 1000);
  if (Math.abs(now - timestamp) > toleranceSeconds) {
    throw new Error("Ground-Signature timestamp outside allowed window");
  }

  const payloadToSign = `${timestamp}.${rawBody}`;
  const computed = crypto.createHmac("sha256", signingSecret).update(payloadToSign).digest("hex");

  const expected = Buffer.from(computed, "utf8");
  const actual = Buffer.from(signature, "utf8");
  if (expected.length !== actual.length || !crypto.timingSafeEqual(expected, actual)) {
    throw new Error("Invalid Ground-Signature");
  }
}

const app = express();

app.post("/ground/webhook", express.raw({ type: "application/json" }), (req, res) => {
  const rawBody = req.body.toString("utf8");
  const signatureHeader = req.get("Ground-Signature");
  const signingSecret = process.env.GROUND_WEBHOOK_SECRET;

  try {
    verifyGroundSignature(rawBody, signatureHeader, signingSecret);
  } catch (err) {
    return res.status(400).json({ error: "Invalid signature" });
  }

  const event = JSON.parse(rawBody);
  // Use event.event to route handling:
  // - "portfolio_wallet.withdrawal.status_changed"
  // - "portfolio_wallet.deposit.status_changed"

  return res.status(200).json({ received: true });
});

# Flask example
import hmac
import hashlib
import time
from flask import Flask, request, jsonify


def verify_ground_signature(raw_body: bytes, signature_header: str, signing_secret: str, tolerance_seconds: int = 300):
    if not signature_header:
        raise ValueError("Missing Ground-Signature header")

    parts = [p.strip() for p in signature_header.split(",")]
    t_part = next((p for p in parts if p.startswith("t=")), None)
    v1_part = next((p for p in parts if p.startswith("v1=")), None)
    if not t_part or not v1_part:
        raise ValueError("Invalid Ground-Signature header format")

    timestamp = int(t_part.split("=", 1)[1])
    signature = v1_part.split("=", 1)[1]

    now = int(time.time())
    if abs(now - timestamp) > tolerance_seconds:
        raise ValueError("Ground-Signature timestamp outside allowed window")

    payload_to_sign = f"{timestamp}.".encode("utf-8") + raw_body
    computed = hmac.new(signing_secret.encode("utf-8"), payload_to_sign, hashlib.sha256).hexdigest()

    if not hmac.compare_digest(computed, signature):
        raise ValueError("Invalid Ground-Signature")


app = Flask(__name__)


@app.post("/ground/webhook")
def ground_webhook():
    raw_body = request.get_data(cache=False)  # raw bytes
    signature_header = request.headers.get("Ground-Signature")
    signing_secret = "YOUR_GROUND_WEBHOOK_SECRET"

    try:
        verify_ground_signature(raw_body, signature_header, signing_secret)
    except Exception:
        return jsonify({"error": "Invalid signature"}), 400

    event = request.get_json()
    # Use event["event"] to route handling:
    # - "portfolio_wallet.withdrawal.status_changed"
    # - "portfolio_wallet.deposit.status_changed"
    return jsonify({"received": True}), 200

package main

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"os"
	"strconv"
	"strings"
	"time"
)

func verifyGroundSignature(rawBody []byte, signatureHeader string, signingSecret string, toleranceSeconds int64) error {
	if signatureHeader == "" {
		return errors.New("missing Ground-Signature header")
	}

	var tStr, v1 string
	for _, part := range strings.Split(signatureHeader, ",") {
		part = strings.TrimSpace(part)
		if strings.HasPrefix(part, "t=") {
			tStr = strings.TrimPrefix(part, "t=")
		} else if strings.HasPrefix(part, "v1=") {
			v1 = strings.TrimPrefix(part, "v1=")
		}
	}
	if tStr == "" || v1 == "" {
		return errors.New("invalid Ground-Signature header format")
	}

	timestamp, err := strconv.ParseInt(tStr, 10, 64)
	if err != nil {
		return errors.New("invalid Ground-Signature timestamp")
	}

	now := time.Now().Unix()
	if now-timestamp > toleranceSeconds || timestamp-now > toleranceSeconds {
		return errors.New("Ground-Signature timestamp outside allowed window")
	}

	sigBytes, err := hex.DecodeString(v1)
	if err != nil {
		return errors.New("invalid Ground-Signature v1 hex")
	}

	mac := hmac.New(sha256.New, []byte(signingSecret))
	mac.Write([]byte(fmt.Sprintf("%d.", timestamp)))
	mac.Write(rawBody)
	expected := mac.Sum(nil)

	if !hmac.Equal(sigBytes, expected) {
		return errors.New("invalid signature")
	}

	return nil
}

func groundWebhook(w http.ResponseWriter, r *http.Request) {
	rawBody, _ := io.ReadAll(r.Body)
	signatureHeader := r.Header.Get("Ground-Signature")
	signingSecret := os.Getenv("GROUND_WEBHOOK_SECRET")

	if err := verifyGroundSignature(rawBody, signatureHeader, signingSecret, 300); err != nil {
		w.WriteHeader(http.StatusBadRequest)
		_ = json.NewEncoder(w).Encode(map[string]string{"error": "Invalid signature"})
		return
	}

	var event map[string]any
	_ = json.Unmarshal(rawBody, &event)
	_ = json.NewEncoder(w).Encode(map[string]bool{"received": true})
}

func main() {
	// net/http example
	http.HandleFunc("/ground/webhook", groundWebhook)
	_ = http.ListenAndServe(":3000", nil)
}