Class WebhookHandler

packages/liveblocks-python/liveblocks/webhooks.py:406–525

Verify incoming Liveblocks webhook requests. Usage:: handler = WebhookHandler("whsec_...") event = handler.verify_request(headers=request.headers, raw_body=request.body)

class WebhookHandler:
    """Verify incoming Liveblocks webhook requests.

    Usage::

        handler = WebhookHandler("whsec_...")
        event = handler.verify_request(headers=request.headers, raw_body=request.body)
    """

    def __init__(self, secret: str) -> None:
        if not secret or not isinstance(secret, str):
            raise ValueError("Secret is required and must be a non-empty string")

        if not secret.startswith(_SECRET_PREFIX):
            raise ValueError("Invalid secret, must start with whsec_")

        secret_key = secret[len(_SECRET_PREFIX) :]
        try:
            self._secret_bytes = base64.b64decode(secret_key)
        except Exception:
            raise ValueError(
                "Webhook secret contains invalid base64 after the 'whsec_' prefix. "
                "Please copy the full secret from your Liveblocks dashboard."
            ) from None

    def verify_request(self, *, headers: dict[str, str], raw_body: str) -> WebhookEvent:
        """Verify a webhook request and return the parsed event.

        Args:
            headers: The HTTP headers as a string-to-string mapping.
            raw_body: The raw request body as a string (do **not** parse it first).

        Returns:
            The parsed webhook event dictionary.

        Raises:
            ValueError: If the request cannot be verified.
        """
        webhook_id, timestamp, raw_signatures = self._verify_headers(headers)

        if not isinstance(raw_body, str):
            raise ValueError(
                f"Invalid raw_body, must be a string, got {type(raw_body).__name__!r} instead. "
                "Make sure you pass the raw request body string, not a parsed object."
            )

        self._verify_timestamp(timestamp)

        signature = self._sign(f"{webhook_id}.{timestamp}.{raw_body}")

        expected_signatures = [
            parts[1] for raw_sig in raw_signatures.split(" ") if len(parts := raw_sig.split(",")) > 1
        ]

        if not any(hmac.compare_digest(signature, s) for s in expected_signatures):
            raise ValueError(
                f"Invalid signature for webhook {webhook_id}. "
                "Make sure you are using the correct webhook secret "