Webhook Integrations

Introduction to Webhooks

Webhooks are event-triggered HTTPS requests sent from the Quiddly Platform to your external integrations or systems. Instead of continuously polling our API for updates, you can configure the Quiddly Platform to automatically notify your application the moment a specific event occurs, allowing you to build real-time, automated workflows.

A webhook listens for a dedicated event type—such as when an invoice has been successfully financed—and triggers instantly every time that event takes place.

To set this up, administrators can use the Operations Portal to define the HTTPS request specifications (the HTTP method and target URI) for the receiving resource. You will also configure a secure secret key, which is used to sign the outbound payload so your application can verify the authenticity of each request.

Managing Webhooks

Webhooks can be managed in the Operations Portal or via API.

Requires rights that can be manages by a system super users. Rights are CRUD:

Note

CREATE_WEBHOOK
READ_WEBHOOK
UPDATE_WEBHOOK
DELETE_WEBHOOK

Payload

Webhook body payloads are now serialized as Canonical JSON (sorted keys, no whitespace separators) and encoded strictly as UTF-8 (without ASCII escaping).

Please review your verification logic to ensure compatibility with consistent key ordering and raw UTF-8 byte matching.

Example Webhook receiver of HMAC v2

An example implementation of a webhook receiver

Production notes

Use these exact practices in production:

1. Always verify the RAW body bytes before json.loads().
2. Keep one secret per webhook endpoint/subscription.
3. Support secret rotation by trying current + previous secret.
4. Reject timestamps older than ~5 minutes.
5. Store webhook event IDs for idempotency / replay suppression.
6. Send only over HTTPS.
7. Log verification failures, but never log secrets or full sensitive payloads.
8. Use short request timeouts and retry safely on sender side.

Wire contract

Headers:

Note

Content-Type: application/json
X-Webhook-Id: <unique delivery or event id>
X-Webhook-Timestamp: <unix timestamp>
X-Webhook-Signature: v1=<hex_hmac_sha256>

Signed content:

Note

<timestamp>.<raw_request_body>

Full code example

This is a python FastAPI example implementation of a Webhook receiver.

Note

from __future__ import annotations
import hashlib
import hmac
import json
import secrets
import time
from dataclasses import dataclass
from typing import Iterable, Mapping, Sequence
class WebhookVerificationError(Exception):
    """Base class for webhook verification failures."""
class MissingHeaderError(WebhookVerificationError):
    """Required webhook header is missing."""
class InvalidSignatureError(WebhookVerificationError):
    """Webhook signature is invalid."""
class ReplayAttackError(WebhookVerificationError):
    """Webhook timestamp is outside the allowed tolerance."""
class InvalidTimestampError(WebhookVerificationError):
    """Webhook timestamp format is invalid."""
@dataclass(frozen=True)
class WebhookHeaders:
    """
    Standard headers used for webhook signing.
    Example:
        X-Webhook-Id: evt_123
        X-Webhook-Timestamp: 1710000000
        X-Webhook-Signature: v1=<hex_digest>
    """
    event_id: str
    timestamp: str
    signature: str
    HEADER_EVENT_ID = "X-Webhook-Id"
    HEADER_TIMESTAMP = "X-Webhook-Timestamp"
    HEADER_SIGNATURE = "X-Webhook-Signature"
    @classmethod
    def from_mapping(cls, headers: Mapping[str, str]) -> "WebhookHeaders":
        # Case-insensitive lookup support
        normalized = {k.lower(): v for k, v in headers.items()}
        try:
            event_id = normalized[cls.HEADER_EVENT_ID.lower()]
            timestamp = normalized[cls.HEADER_TIMESTAMP.lower()]
            signature = normalized[cls.HEADER_SIGNATURE.lower()]
        except KeyError as exc:
            raise MissingHeaderError(f"Missing required header: {exc}") from exc
        return cls(event_id=event_id, timestamp=timestamp, signature=signature)
def canonical_json(data: object) -> bytes:
    """
    Serialize JSON deterministically for sending.
    Use compact separators to avoid signature mismatches.
    """
    return json.dumps(data, separators=(",", ":"), sort_keys=True, ensure_ascii=False).encode("utf-8")
def build_signed_payload(timestamp: str, body: bytes) -> bytes:
    """
    Signed content format:
        <timestamp>.<raw_body_bytes>
    """
    return timestamp.encode("utf-8") + b"." + body
def compute_signature(secret: str, timestamp: str, body: bytes) -> str:
    """
    Compute a v1 HMAC-SHA256 signature as a hex digest.
    """
    mac = hmac.new(
        key=secret.encode("utf-8"),
        msg=build_signed_payload(timestamp, body),
        digestmod=hashlib.sha256,
    )
    return f"v1={mac.hexdigest()}"
def parse_signature_header(signature_header: str) -> Sequence[str]:
    """
    Support one or more signatures in a comma-separated header.
    Example:
        "v1=abc,v1=def"
    """
    parts = [part.strip() for part in signature_header.split(",") if part.strip()]
    if not parts:
        raise InvalidSignatureError("Signature header is empty.")
    return parts
def secure_compare(a: str, b: str) -> bool:
    """
    Constant-time string comparison.
    """
    return hmac.compare_digest(a.encode("utf-8"), b.encode("utf-8"))
def verify_webhook(
    *,
    headers: Mapping[str, str],
    body: bytes,
    secrets_to_try: Iterable[str],
    tolerance_seconds: int = 300,
    now: int | None = None,
) -> WebhookHeaders:
    """
    Verify inbound webhook request.
    Args:
        headers:
            HTTP request headers.
        body:
            Raw request body bytes. Do not parse or re-serialize before verifying.
        secrets_to_try:
            One or more secrets. Provide multiple during secret rotation.
        tolerance_seconds:
            Maximum age/skew allowed for replay protection.
        now:
            Override current time for testing.
    Returns:
        Parsed webhook headers if valid.
    Raises:
        MissingHeaderError
        InvalidTimestampError
        ReplayAttackError
        InvalidSignatureError
    """
    parsed = WebhookHeaders.from_mapping(headers)
    try:
        timestamp = int(parsed.timestamp)
    except ValueError as exc:
        raise InvalidTimestampError("Timestamp header must be a Unix timestamp.") from exc
    current_time = now if now is not None else int(time.time())
    age = abs(current_time - timestamp)
    if age > tolerance_seconds:
        raise ReplayAttackError(
            f"Webhook timestamp is outside tolerance: age={age}s tolerance={tolerance_seconds}s"
        )
    provided_signatures = parse_signature_header(parsed.signature)
    valid = False
    for secret in secrets_to_try:
        expected = compute_signature(secret=secret, timestamp=parsed.timestamp, body=body)
        for provided in provided_signatures:
            if secure_compare(expected, provided):
                valid = True
                break
        if valid:
            break
    if not valid:
        raise InvalidSignatureError("No matching webhook signature found.")
    return parsed
###
from fastapi import FastAPI, HTTPException, Request, status
app = FastAPI()
# During secret rotation, include both active and previous secrets.
WEBHOOK_SECRETS = [
    "current_secret_here",
    "previous_secret_here",
]
@app.post("/webhooks/payments")
async def receive_webhook(request: Request):
    raw_body = await request.body()
    try:
        verified = verify_webhook(
            headers=request.headers,
            body=raw_body,
            secrets_to_try=WEBHOOK_SECRETS,
            tolerance_seconds=300,
        )
    except MissingHeaderError as exc:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=str(exc),
        ) from exc
    except InvalidTimestampError as exc:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=str(exc),
        ) from exc
    except ReplayAttackError as exc:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=str(exc),
        ) from exc
    except InvalidSignatureError as exc:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=str(exc),
        ) from exc
    # Verify first, then parse.
    payload = json.loads(raw_body)
    # Strongly recommended: enforce idempotency using verified.event_id.
    # Example:
    # if event_id_already_seen(verified.event_id):
    #     return {"status": "duplicate_ignored"}
    # mark_event_id_seen(verified.event_id)
    # Process payload here.
    return {
        "status": "ok",
        "event_id": verified.event_id,
        "type": payload.get("type"),
    }