Webhooks

curl -X POST https://api.southpay.io/api/v2/webhook_endpoints \
  -H "Authorization: Bearer sp_sk_test_YOUR_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "webhook_endpoint": {
      "url": "https://yourapp.example/webhooks/southpay",
      "description": "Order fulfillment handler",
      "subscribed_events": [
        "payment_intent.completed",
        "payment_intent.failed",
        "payment_intent.expired",
        "payment_intent.refunded"
      ]
    }
  }'

{
  "id": "b7e4c210-5d83-4f16-9c02-6ea1f3d85b47",
  "url": "https://yourapp.example/webhooks/southpay",
  "description": "Order fulfillment handler",
  "status": "active",
  "subscribed_events": [
    "payment_intent.completed",
    "payment_intent.failed",
    "payment_intent.expired",
    "payment_intent.refunded"
  ],
  "signing_secret": "whsec_4f8b1c2d...",
  "delivery_success_rate": null,
  "created_at": "2026-08-07T14:01:00.000Z"
}

import express from "express";
import { createHmac, timingSafeEqual } from "node:crypto";

const TOLERANCE_SECONDS = 300;

export function verify(rawBody: Buffer, header: string, secret: string): boolean {
  const parts = Object.fromEntries(
    header.split(",").map(p => p.split("=", 2) as [string, string])
  );
  if (!parts.t || !parts.v1) return false;

  const age = Math.floor(Date.now() / 1000) - Number(parts.t);
  if (!Number.isFinite(age) || Math.abs(age) > TOLERANCE_SECONDS) return false;

  const computed = createHmac("sha256", secret)
    .update(`${parts.t}.${rawBody.toString("utf8")}`)
    .digest("hex");

  const a = Buffer.from(computed);
  const b = Buffer.from(parts.v1);
  return a.length === b.length && timingSafeEqual(a, b);
}

const app = express();

app.post(
  "/webhooks/southpay",
  express.raw({ type: "application/json" }),
  (req, res) => {
    const sig = req.headers["southpay-signature"] as string | undefined;
    if (!sig || !verify(req.body, sig, process.env.SOUTHPAY_WEBHOOK_SECRET!)) {
      return res.status(401).end();
    }

    const event = JSON.parse(req.body.toString());
    res.json({ received: true });
    queue.enqueue("southpay_event", event);
  }
);

import hashlib
import hmac
import os
import time

from flask import Flask, jsonify, request

app = Flask(__name__)
TOLERANCE_SECONDS = 300


def verify(raw_body: bytes, header: str, secret: str) -> bool:
    parts = dict(p.split("=", 1) for p in header.split(","))
    timestamp, v1 = parts.get("t"), parts.get("v1")
    if not timestamp or not v1:
        return False

    if abs(int(time.time()) - int(timestamp)) > TOLERANCE_SECONDS:
        return False

    computed = hmac.new(
        secret.encode(),
        f"{timestamp}.{raw_body.decode()}".encode(),
        hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(computed, v1)


@app.post("/webhooks/southpay")
def webhook():
    sig = request.headers.get("Southpay-Signature", "")
    if not verify(request.data, sig, os.environ["SOUTHPAY_WEBHOOK_SECRET"]):
        return jsonify(error="Invalid signature"), 401

    enqueue("southpay_event", request.get_json())
    return jsonify(received=True)

package webhook

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"fmt"
	"strconv"
	"strings"
	"time"
)

const toleranceSeconds = 300

func Verify(rawBody []byte, header, secret string) bool {
	parts := map[string]string{}
	for _, p := range strings.Split(header, ",") {
		if kv := strings.SplitN(p, "=", 2); len(kv) == 2 {
			parts[kv[0]] = kv[1]
		}
	}

	ts, v1 := parts["t"], parts["v1"]
	if ts == "" || v1 == "" {
		return false
	}

	sent, err := strconv.ParseInt(ts, 10, 64)
	if err != nil || abs(time.Now().Unix()-sent) > toleranceSeconds {
		return false
	}

	mac := hmac.New(sha256.New, []byte(secret))
	fmt.Fprintf(mac, "%s.%s", ts, rawBody)
	return hmac.Equal([]byte(hex.EncodeToString(mac.Sum(nil))), []byte(v1))
}

func abs(n int64) int64 {
	if n < 0 {
		return -n
	}
	return n
}

require "openssl"

module SouthPayWebhook
  TOLERANCE_SECONDS = 300

  def self.verify(raw_body, header, secret)
    parts = header.to_s.split(",").map { |p| p.split("=", 2) }.to_h
    timestamp = parts["t"]
    v1 = parts["v1"]
    return false unless timestamp && v1
    return false if (Time.now.to_i - timestamp.to_i).abs > TOLERANCE_SECONDS

    computed = OpenSSL::HMAC.hexdigest("sha256", secret, "#{timestamp}.#{raw_body}")
    ActiveSupport::SecurityUtils.secure_compare(computed, v1)
  end
end