Webhook Verification
Verify incoming webhook signatures to ensure they're from SpatialFlow.
A workspace webhook, created through the API, receives test deliveries only for now. To receive real geofence events at an endpoint, add a Webhook action to a workflow; its signing differs, see Webhook (HTTP Request). The signature, replay, and receiver guidance on this page applies to workspace webhook deliveries.
The secret argument is the webhook's signing secret. SpatialFlow shows it once, when you create the webhook, and again each time you rotate it. See Getting the signing secret.
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: Deprecated and ignored; the signature has no timestamp
Returns:
Parsed JSON payload
Raises:
WebhookSignatureError: If verification fails
"""
verify_webhook_signature returns the parsed body as it arrived. The default payload has id, event, timestamp and data. A custom_payload_template replaces the whole body, so those keys exist only if the template renders them. The examples on this page assume the default payload.
FastAPI Example
The receivers on this page record each delivery in a processed_webhooks table, as described under Replay Protection:
CREATE TABLE processed_webhooks (id text PRIMARY KEY);
CREATE TABLE geofence_visits (user_id text, geofence_id text);
import os
import psycopg
from fastapi import FastAPI, Request, HTTPException
from spatialflow import verify_webhook_signature, WebhookSignatureError
app = FastAPI()
WEBHOOK_SECRET = "your-webhook-secret"
DATABASE_URL = os.environ["DATABASE_URL"]
MAX_BODY_BYTES = 1024 * 1024
async def process_event(conn, event):
if event["event"] == "geofence.enter":
data = event["data"]
await conn.execute(
"INSERT INTO geofence_visits (user_id, geofence_id) VALUES (%s, %s)",
(data["user_id"], data["geofence_id"]),
)
@app.post("/webhook")
async def handle_webhook(request: Request):
# Stop reading before the signature check if the body is too large
payload = b""
async for chunk in request.stream():
payload += chunk
if len(payload) > MAX_BODY_BYTES:
raise HTTPException(status_code=413, detail="Payload too large")
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,
)
except WebhookSignatureError as e:
raise HTTPException(status_code=400, detail=str(e))
delivery_id = event.get("id")
if not delivery_id:
# A custom_payload_template without an id leaves nothing to deduplicate on
raise HTTPException(status_code=400, detail="Missing delivery id")
# Use a connection pool in production
async with await psycopg.AsyncConnection.connect(DATABASE_URL) as conn:
async with conn.transaction():
cur = await conn.execute(
"INSERT INTO processed_webhooks (id) VALUES (%s) "
"ON CONFLICT (id) DO NOTHING RETURNING id",
(delivery_id,),
)
if await cur.fetchone() is None:
return {"status": "already_processed"}
await process_event(conn, event)
# An exception rolls back the id and the work, and surfaces as a 500
return {"status": "ok"}
Flask Example
import os
import psycopg
from flask import Flask, request, jsonify
from spatialflow import verify_webhook_signature, WebhookSignatureError
app = Flask(__name__)
app.config["MAX_CONTENT_LENGTH"] = 1024 * 1024 # Flask answers 413 for larger bodies
WEBHOOK_SECRET = "your-webhook-secret"
DATABASE_URL = os.environ["DATABASE_URL"]
def process_event(conn, event):
if event["event"] == "geofence.enter":
data = event["data"]
conn.execute(
"INSERT INTO geofence_visits (user_id, geofence_id) VALUES (%s, %s)",
(data["user_id"], data["geofence_id"]),
)
@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,
)
except WebhookSignatureError as e:
return jsonify({"error": str(e)}), 400
delivery_id = event.get("id")
if not delivery_id:
# A custom_payload_template without an id leaves nothing to deduplicate on
return jsonify({"error": "Missing delivery id"}), 400
# Use a connection pool in production
with psycopg.connect(DATABASE_URL) as conn:
with conn.transaction():
row = conn.execute(
"INSERT INTO processed_webhooks (id) VALUES (%s) "
"ON CONFLICT (id) DO NOTHING RETURNING id",
(delivery_id,),
).fetchone()
if row is None:
return jsonify({"status": "already_processed"})
process_event(conn, event)
# An exception rolls back the id and the work, and surfaces as a 500
return jsonify({"status": "ok"})
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, and someone holding a valid delivery can resend the same body and signature. The X-Idempotency-Key and X-SF-Event-ID headers are not covered by the signature, so deduplicating on them does not stop a replay: the sender can change the header and keep the signed body.
After verify_webhook_signature succeeds, deduplicate on the delivery id in the signed body (event.get("id")). A retry of the same delivery carries the same id. The default payload carries a signed id, the delivery UUID. A custom_payload_template replaces the whole body with the rendered template, so the template must include a stable id for replay protection: "id": "{{ event.id }}". In a template, event is the default payload, and data, timestamp, and event_type are also available. If event.get("id") is missing, reject the delivery with a 400 instead of processing it without replay protection. The FastAPI and Flask examples above do both.
Acknowledge a delivery only after its work is committed. SpatialFlow treats any 2xx response as success and does not resend it. Insert the id in the same database transaction as the work, with INSERT INTO processed_webhooks (id) VALUES (%s) ON CONFLICT (id) DO NOTHING RETURNING id. If no row comes back, the delivery is a duplicate, so return 200 without processing. Otherwise do the work in that transaction, and let any exception roll back both and return a non-2xx response so the retry processes the event. Do not catch a broad integrity error around the work, because a constraint failure inside it would look like a duplicate. A queue counts as accepting the work only if it has persisted the message before you respond. Redis does that only with appendfsync always. See the Webhooks concept page.
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
Workspace webhooks carry these event types. For now they are sent only by the test endpoints.
| Event | Description |
|---|---|
geofence.enter | Device entered a geofence |
geofence.exit | Device exited a geofence |
webhook.test | Test event from webhook configuration |