Python SDK - Webhook Handling Example

import os

from flask import Flask, request, abort
from corsa_sdk import verify_webhook_signature

app = Flask(__name__)

WEBHOOK_SECRET = os.environ["CORSA_WEBHOOK_SECRET"]


@app.route("/webhook", methods=["POST"])
def handle_webhook():
    signature = request.headers.get("x-hub-signature-256")
    if not signature:
        abort(400, "Missing signature header")

    raw_body = request.get_data(as_text=True)

    if not verify_webhook_signature(WEBHOOK_SECRET, raw_body, signature):
        abort(403, "Invalid signature")

    event = request.get_json()
    event_type = event.get("type", "unknown")

    print(f"Received event: {event_type}")
    print(f"Timestamp: {event.get('timestamp')}")

    match event_type:
        case "individual_client.created" | "individual_client.updated":
            print(f"Client: {event['data']['id']}")
        case "alert.created" | "alert.updated":
            print(f"Alert: {event['data']['id']}")
        case "case.created" | "case.updated":
            print(f"Case: {event['data']['id']}")
        case _:
            print(f"Unhandled event type: {event_type}")

    return "OK", 200


if __name__ == "__main__":
    app.run(port=3000)

from corsa_sdk import sign_webhook_payload

signature = sign_webhook_payload(
    "your-webhook-secret",
    '{"type":"alert.created","data":{"id":"abc-123"}}',
)
print(signature)