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: bytes, signature: str, secret: str) -> dict:
    """
    Verify a webhook signature and return the parsed payload.

    Args:
        payload: Raw request body 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["type"]
        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['type']}")
        return jsonify({"status": "ok"})

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

Signature Format

The X-SF-Signature header contains:

t=1234567890,v1=abc123...

• t - Unix timestamp when the signature was generated
• v1 - HMAC-SHA256 signature

Custom Tolerance

By default, signatures older than 5 minutes (300 seconds) are rejected. You can customize this:

event = verify_webhook_signature(
    payload=payload,
    signature=signature,
    secret=WEBHOOK_SECRET,
    tolerance=600,  # Accept signatures up to 10 minutes old
)

Error Types

WebhookSignatureError is raised when:

• Signature header is missing or malformed
• Timestamp is too old (outside tolerance)
• 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