X-Orbit-Signature: t=1715357600,v1=4f9c2e6b8a1d3f5e7c9b1a3d5f7e9c1b3a5d7f9e1c3b5a7d9f1e3c5b7a9d1f3e

X-Orbit-Signature-Next: t=1715357600,v1=<sig_with_prev_secret>

X-Devotel-Signature: t=1715357600,v1=4f9c2e6b8a1d3f5e7c9b1a3d5f7e9c1b3a5d7f9e1c3b5a7d9f1e3c5b7a9d1f3e

X-Devotel-Signature: t=1715357600,v1=<sig_with_new_secret>,v1=<sig_with_prev_secret>

import crypto from 'node:crypto';

function parseSignatureHeader(header) {
  const parts = header.split(',').map((p) => p.trim());
  let t = null;
  const v1s = [];
  for (const part of parts) {
    const eq = part.indexOf('=');
    if (eq < 0) continue;
    const key = part.slice(0, eq);
    const value = part.slice(eq + 1);
    if (key === 't') t = value;
    else if (key === 'v1') v1s.push(value);
  }
  return { t, v1s };
}

function verifyWebhookSignature(rawBody, header, secret) {
  const { t, v1s } = parseSignatureHeader(header);
  if (!t || v1s.length === 0) return false;

  // Replay protection: reject signatures older than 5 minutes.
  const ageSec = Math.floor(Date.now() / 1000) - Number(t);
  if (!Number.isFinite(ageSec) || ageSec < 0 || ageSec > 5 * 60) return false;

  const expected = crypto
    .createHmac('sha256', secret)
    .update(`${t}.${rawBody}`, 'utf-8')
    .digest('hex');
  const expectedBuf = Buffer.from(expected, 'hex');

  // Accept the request if ANY v1 matches (covers rotation grace).
  return v1s.some((candidate) => {
    const candidateBuf = Buffer.from(candidate, 'hex');
    return (
      candidateBuf.length === expectedBuf.length &&
      crypto.timingSafeEqual(candidateBuf, expectedBuf)
    );
  });
}

// In your webhook handler:
app.post('/webhooks/orbit', (req, res) => {
  // Prefer the canonical header; fall back to the legacy one for older deliveries.
  const header = req.headers['x-orbit-signature'] ?? req.headers['x-devotel-signature'];
  const isValid = verifyWebhookSignature(req.rawBody, header, 'whsec_your_secret');

  if (!isValid) {
    return res.status(401).json({ error: 'Invalid signature' });
  }

  const event = JSON.parse(req.rawBody);
  // Process the event...

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

import hmac
import hashlib
import time


def parse_signature_header(header: str):
    t = None
    v1s = []
    for part in header.split(","):
        part = part.strip()
        if "=" not in part:
            continue
        key, _, value = part.partition("=")
        if key == "t":
            t = value
        elif key == "v1":
            v1s.append(value)
    return t, v1s


def verify_webhook_signature(raw_body: bytes, header: str, secret: str) -> bool:
    t, v1s = parse_signature_header(header)
    if not t or not v1s:
        return False

    # Replay protection: reject signatures older than 5 minutes.
    age = int(time.time()) - int(t)
    if age < 0 or age > 5 * 60:
        return False

    signed = f"{t}.".encode("utf-8") + raw_body
    expected = hmac.new(secret.encode("utf-8"), signed, hashlib.sha256).hexdigest()

    # Accept the request if ANY v1 matches (covers rotation grace).
    return any(hmac.compare_digest(candidate, expected) for candidate in v1s)


# In your webhook handler (Flask example):
@app.route("/webhooks/orbit", methods=["POST"])
def handle_webhook():
    # Prefer the canonical header; fall back to the legacy one for older deliveries.
    header = request.headers.get("X-Orbit-Signature") or request.headers.get(
        "X-Devotel-Signature", ""
    )
    if not verify_webhook_signature(request.data, header, "whsec_your_secret"):
        return jsonify({"error": "Invalid signature"}), 401

    event = request.get_json()
    # Process the event...

    return jsonify({"received": True}), 200

package main

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

func parseSignatureHeader(header string) (string, []string) {
	var t string
	var v1s []string
	for _, part := range strings.Split(header, ",") {
		part = strings.TrimSpace(part)
		eq := strings.IndexByte(part, '=')
		if eq < 0 {
			continue
		}
		key, value := part[:eq], part[eq+1:]
		switch key {
		case "t":
			t = value
		case "v1":
			v1s = append(v1s, value)
		}
	}
	return t, v1s
}

func verifyWebhookSignature(rawBody []byte, header, secret string) bool {
	t, v1s := parseSignatureHeader(header)
	if t == "" || len(v1s) == 0 {
		return false
	}

	// Replay protection: reject signatures older than 5 minutes.
	ts, err := strconv.ParseInt(t, 10, 64)
	if err != nil {
		return false
	}
	age := time.Now().Unix() - ts
	if age < 0 || age > 5*60 {
		return false
	}

	mac := hmac.New(sha256.New, []byte(secret))
	mac.Write([]byte(t + "."))
	mac.Write(rawBody)
	expected := hex.EncodeToString(mac.Sum(nil))

	// Accept the request if ANY v1 matches (covers rotation grace).
	for _, candidate := range v1s {
		if hmac.Equal([]byte(candidate), []byte(expected)) {
			return true
		}
	}
	return false
}

# The source IP(s) every webhook delivery is sent from — allowlist these on
# a firewall that restricts inbound HTTPS to your endpoint.
curl https://api.orbit.devotel.io/api/v1/webhooks/egress-ips \
  -H "X-API-Key: dv_live_sk_..."

# Omit `secret` to have Orbit auto-generate one. The response includes the
# full `whsec_...` secret either way — copy it now.
curl -X POST https://api.orbit.devotel.io/api/v1/webhooks \
  -H "X-API-Key: dv_live_sk_..." \
  -H "Content-Type: application/json" \
  -d '{"url": "https://example.com/webhooks", "events": ["message.delivered"]}'

# Or supply your own secret (at least 16 characters) in the `secret` field.
curl -X POST https://api.orbit.devotel.io/api/v1/webhooks \
  -H "X-API-Key: dv_live_sk_..." \
  -H "Content-Type: application/json" \
  -d '{"url": "https://example.com/webhooks", "events": ["message.delivered"], "secret": "whsec_your_own_signing_secret"}'

# Returns a new cleartext `whsec_...` secret. The previous secret is honored
# during a 7-day grace window (604800 s) so in-flight deliveries keep verifying.
curl -X POST https://api.orbit.devotel.io/api/v1/webhooks/wh_abc123/rotate-secret \
  -H "X-API-Key: dv_live_sk_..."