Webhook Verification

Verify incoming webhook signatures to ensure they're from SpatialFlow.

verify_webhook_signature

from spatialflow import verify_webhook_signature, WebhookSignatureError

def verify(
    payload: Union[str, bytes],
    signature: str,
    secret: str,
    tolerance: int = 300,
) -> dict:
    """
    Verify a webhook signature and return the parsed payload.

    Args:
        payload: Raw request body (str or bytes)
        signature: X-SF-Signature header value
        secret: Your webhook secret
        tolerance: Max age in seconds (default: 300)

    Returns:
        Parsed JSON payload

    Raises:
        WebhookSignatureError: If verification fails
    """

FastAPI Example

from fastapi import FastAPI, Request, HTTPException
from spatialflow import verify_webhook_signature, WebhookSignatureError

app = FastAPI()
WEBHOOK_SECRET = "your-webhook-secret"

@app.post("/webhook")
async def handle_webhook(request: Request):
    payload = await request.body()
    signature = request.headers.get("X-SF-Signature")

    if not signature:
        raise HTTPException(status_code=400, detail="Missing signature")

    try:
        event = verify_webhook_signature(
            payload=payload,
            signature=signature,
            secret=WEBHOOK_SECRET,
        )

        # Handle the event
        event_type = event["event"]
        if event_type == "geofence.enter":
            device_id = event["data"]["device_id"]
            geofence_id = event["data"]["geofence_id"]
            print(f"Device {device_id} entered geofence {geofence_id}")

        return {"status": "ok"}

    except WebhookSignatureError as e:
        raise HTTPException(status_code=400, detail=str(e))

Flask Example

from flask import Flask, request, jsonify
from spatialflow import verify_webhook_signature, WebhookSignatureError

app = Flask(__name__)
WEBHOOK_SECRET = "your-webhook-secret"

@app.route("/webhook", methods=["POST"])
def handle_webhook():
    payload = request.get_data()
    signature = request.headers.get("X-SF-Signature")

    if not signature:
        return jsonify({"error": "Missing signature"}), 400

    try:
        event = verify_webhook_signature(
            payload=payload,
            signature=signature,
            secret=WEBHOOK_SECRET,
        )

        # Handle the event
        print(f"Event type: {event['event']}")
        return jsonify({"status": "ok"})

    except WebhookSignatureError as e:
        return jsonify({"error": str(e)}), 400

Signature Format

The X-SF-Signature header is an HMAC-SHA256 of the raw request body, hex-encoded and prefixed with sha256=:

sha256=5257a869e7ec...9f2b

It is computed as HMAC_SHA256(secret, raw_request_body) and does not include a timestamp component.

Replay Protection

The SpatialFlow signature covers the request body only; it does not include a timestamp, so there is no time-based tolerance window. To guard against replays, deduplicate on the X-Idempotency-Key (or X-SF-Event-ID) header included with every delivery.

Error Types

WebhookSignatureError is raised when:

• Signature header is missing or malformed
• Signature doesn't match (wrong secret or tampered payload)
• Payload is not valid JSON

try:
    event = verify_webhook_signature(...)
except WebhookSignatureError as e:
    print(f"Verification failed: {e}")

Event Types

Common webhook event types:

Event	Description
geofence.enter	Device entered a geofence
geofence.exit	Device exited a geofence
geofence.dwell	Device dwelling in a geofence
workflow.triggered	Workflow was triggered
workflow.completed	Workflow execution completed
workflow.failed	Workflow execution failed