import json import hmac import hashlib import time from typing import Any, Dict from .errors import BadRequestError class WebhooksClient: """ A client to handle ElevenLabs webhook-related functionality """ def construct_event(self, rawBody: str, sig_header: str, secret: str) -> Dict: """ Constructs a webhook event object from a payload and signature. Verifies the webhook signature to ensure the event came from ElevenLabs. Args: rawBody: The webhook request body. Must be the raw body, not a JSON object sig_header: The signature header from the request secret: Your webhook secret Returns: The verified webhook event Raises: BadRequestError: If the signature is invalid or missing """ if not sig_header: raise BadRequestError(body="Missing signature header") if not secret: raise BadRequestError(body="Webhook secret not configured") headers = sig_header.split(',') timestamp = None signature = None for header in headers: if header.startswith('t='): timestamp = header[2:] elif header.startswith('v0='): signature = header if not timestamp or not signature: raise BadRequestError(body="No signature hash found with expected scheme v0") # Validate timestamp req_timestamp = int(timestamp) * 1000 tolerance = int(time.time() * 1000) - 30 * 60 * 1000 if req_timestamp < tolerance: raise BadRequestError(body="Timestamp outside the tolerance zone") # Validate hash message = f"{timestamp}.{rawBody}" digest = "v0=" + hmac.new( secret.encode('utf-8'), message.encode('utf-8'), hashlib.sha256 ).hexdigest() if signature != digest: raise BadRequestError( body="Signature hash does not match the expected signature hash for payload" ) return json.loads(rawBody)