Webhooks

# Given the raw body in body.json and the header value in $SIG_HEADER:
T=$(printf '%s' "$SIG_HEADER" | sed -E 's/^t=([0-9]+),.*/\1/')
V1=$(printf '%s' "$SIG_HEADER" | sed -E 's/.*,v1=//')

# Recompute: HMAC-SHA256( secret, "<t>." + rawBody )
EXPECTED=$( { printf '%s.' "$T"; cat body.json; } \
  | openssl dgst -sha256 -hmac "$PHOSRA_WEBHOOK_SECRET" -r \
  | cut -d' ' -f1 )

[ "$V1" = "$EXPECTED" ] && echo "signature OK" || echo "signature INVALID"

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

const MAX_AGE_SECONDS = 300 // reject deliveries older than 5 minutes

export function verifyPhosraWebhook(
  rawBody: string,
  sigHeader: string | null,
  webhookSecret: string,
): boolean {
  if (!sigHeader) return false

  // Parse "t=<timestamp>,v1=<signature>"
  const parts = Object.fromEntries(
    sigHeader.split(",").map((p) => p.split("=") as [string, string]),
  )
  const t = parts.t
  const v1 = parts.v1
  if (!t || !v1) return false

  const timestamp = Number.parseInt(t, 10)
  if (Number.isNaN(timestamp)) return false
  if (Math.floor(Date.now() / 1000) - timestamp > MAX_AGE_SECONDS) return false

  const expected = createHmac("sha256", webhookSecret)
    .update(`${t}.${rawBody}`)
    .digest("hex")

  if (v1.length !== expected.length) return false
  return timingSafeEqual(Buffer.from(v1), Buffer.from(expected))
}

// Express — capture the RAW body so the HMAC matches.
app.post(
  "/webhooks/phosra",
  express.raw({ type: "application/json" }),
  (req, res) => {
    const raw = req.body.toString("utf8")
    const sig = req.headers["x-phosra-signature"] as string | undefined
    if (!verifyPhosraWebhook(raw, sig ?? null, process.env.PHOSRA_WEBHOOK_SECRET!)) {
      return res.status(401).json({ error: "invalid signature" })
    }
    const event = JSON.parse(raw)
    // … handle event, then ack fast …
    res.sendStatus(200)
  },
)

import hashlib
import hmac
import time


def verify_phosra_webhook(
    raw_body: bytes,
    sig_header: str | None,
    webhook_secret: str,
    max_age_seconds: int = 300,
) -> bool:
    if not sig_header:
        return False

    parts = dict(
        p.split("=", 1) for p in sig_header.split(",") if "=" in p
    )
    t, v1 = parts.get("t"), parts.get("v1")
    if not t or not v1:
        return False

    try:
        timestamp = int(t)
    except ValueError:
        return False
    if int(time.time()) - timestamp > max_age_seconds:
        return False

    preimage = t.encode() + b"." + raw_body
    expected = hmac.new(
        webhook_secret.encode(), preimage, hashlib.sha256
    ).hexdigest()

    return hmac.compare_digest(v1, expected)


# Flask — request.get_data() returns the RAW body (do not read request.json first).
@app.post("/webhooks/phosra")
def phosra_webhook():
    raw = request.get_data()
    sig = request.headers.get("X-Phosra-Signature")
    if not verify_phosra_webhook(raw, sig, os.environ["PHOSRA_WEBHOOK_SECRET"]):
        return {"error": "invalid signature"}, 401
    event = json.loads(raw)
    # … handle event, then ack fast …
    return "", 200

package webhook

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

// Verify reports whether sigHeader is a valid Phosra signature for rawBody.
func Verify(rawBody []byte, sigHeader, secret string, maxAge time.Duration) bool {
	var t, v1 string
	for _, p := range strings.Split(sigHeader, ",") {
		kv := strings.SplitN(p, "=", 2)
		if len(kv) != 2 {
			continue
		}
		switch kv[0] {
		case "t":
			t = kv[1]
		case "v1":
			v1 = kv[1]
		}
	}
	if t == "" || v1 == "" {
		return false
	}

	ts, err := strconv.ParseInt(t, 10, 64)
	if err != nil {
		return false
	}
	if time.Since(time.Unix(ts, 0)) > maxAge {
		return false // reject replays
	}

	mac := hmac.New(sha256.New, []byte(secret))
	fmt.Fprintf(mac, "%s.", t) // preimage: "<t>." + body
	mac.Write(rawBody)
	expected := hex.EncodeToString(mac.Sum(nil))

	return hmac.Equal([]byte(v1), []byte(expected))
}

// net/http handler — read the RAW body before decoding.
func Handler(secret string) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		raw, _ := io.ReadAll(r.Body)
		if !Verify(raw, r.Header.Get("X-Phosra-Signature"), secret, 5*time.Minute) {
			http.Error(w, "invalid signature", http.StatusUnauthorized)
			return
		}
		// … json.Unmarshal(raw, &event), handle, then ack fast …
		w.WriteHeader(http.StatusOK)
	}
}

# Extract the raw payload bytes of the first failed delivery into body.json.
curl -s "$PHOSRA_API/api/v1/webhooks/$WEBHOOK_ID/deliveries?limit=50" \
  -H "Authorization: Bearer $PHOSRA_API_KEY" \
  | jq -c '[.[] | select(.success == false)][0].payload' > body.json

# Sign "<t>." + rawBody with the same HMAC the receiver verifies.
T=$(date +%s)
SIG=$( { printf '%s.' "$T"; cat body.json; } \
  | openssl dgst -sha256 -hmac "$PHOSRA_WEBHOOK_SECRET" -r | cut -d' ' -f1 )
EVENT=$(jq -r '.event' body.json)
EVENT_ID=$(jq -r '.event_id' body.json)

# Replay it straight at your local handler (through the tunnel, or localhost directly).
curl -sX POST "http://localhost:4242/webhooks/phosra" \
  -H "Content-Type: application/json" \
  -H "X-Phosra-Signature: t=$T,v1=$SIG" \
  -H "X-Phosra-Event: $EVENT" \
  -H "X-Phosra-Event-ID: $EVENT_ID" \
  --data-binary @body.json

import { createHmac } from "node:crypto"

const API = process.env.PHOSRA_API!
const KEY = process.env.PHOSRA_API_KEY!
const SECRET = process.env.PHOSRA_WEBHOOK_SECRET!
const WEBHOOK_ID = process.env.WEBHOOK_ID!
const TARGET = "http://localhost:4242/webhooks/phosra"

// 1. Read the delivery history and pick the first failed delivery.
const deliveries = await fetch(
  `${API}/api/v1/webhooks/${WEBHOOK_ID}/deliveries?limit=50`,
  { headers: { Authorization: `Bearer ${KEY}` } },
).then((r) => r.json())

const failed = deliveries.find((d: any) => d.success === false)
if (!failed) throw new Error("no failed delivery to replay")

// 2. Re-serialize the stored envelope to the exact bytes we will sign.
const raw = JSON.stringify(failed.payload)
const t = Math.floor(Date.now() / 1000).toString()
const v1 = createHmac("sha256", SECRET).update(`${t}.${raw}`).digest("hex")

// 3. POST it back at your handler with a freshly minted signature.
const res = await fetch(TARGET, {
  method: "POST",
  headers: {
    "Content-Type": "application/json",
    "X-Phosra-Signature": `t=${t},v1=${v1}`,
    "X-Phosra-Event": failed.event,
    "X-Phosra-Event-ID": failed.payload.event_id,
  },
  body: raw,
})
console.log("replayed →", res.status)

import hashlib
import hmac
import json
import os
import time

import requests

API = os.environ["PHOSRA_API"]
KEY = os.environ["PHOSRA_API_KEY"]
SECRET = os.environ["PHOSRA_WEBHOOK_SECRET"].encode()
WEBHOOK_ID = os.environ["WEBHOOK_ID"]
TARGET = "http://localhost:4242/webhooks/phosra"

# 1. Read the delivery history and pick the first failed delivery.
deliveries = requests.get(
    f"{API}/api/v1/webhooks/{WEBHOOK_ID}/deliveries",
    params={"limit": 50},
    headers={"Authorization": f"Bearer {KEY}"},
).json()
failed = next((d for d in deliveries if d["success"] is False), None)
if failed is None:
    raise SystemExit("no failed delivery to replay")

# 2. Re-serialize the stored envelope to the exact bytes we will sign.
raw = json.dumps(failed["payload"], separators=(",", ":")).encode()
t = str(int(time.time()))
v1 = hmac.new(SECRET, t.encode() + b"." + raw, hashlib.sha256).hexdigest()

# 3. POST it back at your handler with a freshly minted signature.
res = requests.post(
    TARGET,
    data=raw,
    headers={
        "Content-Type": "application/json",
        "X-Phosra-Signature": f"t={t},v1={v1}",
        "X-Phosra-Event": failed["event"],
        "X-Phosra-Event-ID": failed["payload"]["event_id"],
    },
)
print("replayed →", res.status_code)