Quick Start

import hashlib, hmac, json, os
from datetime import datetime, timezone
from http.server import HTTPServer, BaseHTTPRequestHandler

SECRET = os.environ["WEBHOOK_SECRET"].encode()   # pxlwh_...
TOLERANCE = 300                                   # 5-minute replay window

class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        length = int(self.headers.get("Content-Length", 0))
        raw_body = self.rfile.read(length).decode()

        sig_header = self.headers.get("X-PAXOS-LABS-SIGNATURE", "")
        ts_header  = self.headers.get("X-PAXOS-LABS-TIMESTAMP", "")

        try:
            self._verify(raw_body, sig_header, ts_header)
        except Exception as e:
            self.send_response(401)
            self.end_headers()
            self.wfile.write(str(e).encode())
            return

        event = json.loads(raw_body)
        print(f"Received {event['type']} — {event['id']}")
        self.send_response(200)
        self.end_headers()
        self.wfile.write(b"ok")

    def _verify(self, raw_body, sig_header, ts_header):
        ts = datetime.fromisoformat(ts_header)
        now = datetime.now(timezone.utc)

        if abs((now - ts).total_seconds()) > TOLERANCE:
            raise ValueError("Timestamp outside tolerance")

        expected = hmac.new(
            SECRET, f"{ts_header}.{raw_body}".encode(), hashlib.sha256
        ).hexdigest()

        if not hmac.compare_digest(sig_header, expected):
            raise ValueError("Invalid signature")

HTTPServer(("", 4000), Handler).serve_forever()

package main

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

const tolerance = 300 // seconds

func verify(body []byte, sigHeader, tsHeader, secret string) error {
	ts, err := time.Parse(time.RFC3339, tsHeader)
	if err != nil {
		return fmt.Errorf("bad timestamp: %w", err)
	}
	if math.Abs(time.Since(ts).Seconds()) > tolerance {
		return fmt.Errorf("timestamp outside tolerance")
	}

	mac := hmac.New(sha256.New, []byte(secret))
	fmt.Fprintf(mac, "%s.%s", tsHeader, body)
	expected := hex.EncodeToString(mac.Sum(nil))

	if !hmac.Equal([]byte(sigHeader), []byte(expected)) {
		return fmt.Errorf("invalid signature")
	}

	return nil
}

func main() {
	secret := os.Getenv("WEBHOOK_SECRET")

	http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
		body, _ := io.ReadAll(r.Body)
		if err := verify(
			body,
			r.Header.Get("X-Paxos-Labs-Signature"),
			r.Header.Get("X-Paxos-Labs-Timestamp"),
			secret,
		); err != nil {
			http.Error(w, err.Error(), 401)
			return
		}
		var evt map[string]any
		json.Unmarshal(body, &evt)
		fmt.Printf("Received %s — %s\n", evt["type"], evt["id"])
		w.WriteHeader(200)
	})

	fmt.Println("Listening on :4000")
	http.ListenAndServe(":4000", nil)
}