Ir para o conteúdo

webhook_handler

module wittgenstein_crm_client.sync.webhook_handler

Classes

  • WebhookPayload — Pydantic model representing standard CRM webhook payloads.

  • WebhookHandler — Webhook signature-verification and payload-parsing helper.

wittgenstein_crm_client.sync.webhook_handler.WebhookPayload

class WebhookPayload()

Bases : BaseModel

Pydantic model representing standard CRM webhook payloads.

wittgenstein_crm_client.sync.webhook_handler.WebhookHandler

class WebhookHandler(secret: Optional[str] = None, provider: str = 'hubspot')

Webhook signature-verification and payload-parsing helper.

Real HMAC-based signature verification is implemented only for HubSpot, the only adapter in this package backed by a production CRM API (see adapters/hubspot.py). Calling verify_signature for any other provider raises NotImplementedError rather than silently reporting the webhook as authentic -- a blind True would let anyone who discovers the webhook URL inject fake CRM events (fake deals, contacts, lead updates).

Methods

  • verify_signature — Verify the signature HubSpot attaches to an inbound webhook request.

  • process_events — Parse raw incoming webhook payload dictionaries to WebhookPayload.

wittgenstein_crm_client.sync.webhook_handler.WebhookHandler.verify_signature

method WebhookHandler.verify_signature(signature: str, request_body: bytes, *, request_method: str, request_uri: str, timestamp: str) → bool

Verify the signature HubSpot attaches to an inbound webhook request.

Implements HubSpot's webhook v3 scheme: the X-HubSpot-Signature-v3 header is HMAC-SHA256(client_secret, request_method + request_uri + request_body + timestamp), base64-encoded. timestamp is the value of the X-HubSpot-Request-Timestamp header; requests older than HUBSPOT_SIGNATURE_MAX_AGE_SECONDS are rejected as stale (replay protection).

NOTE FOR PRODUCTION: this follows HubSpot's publicly documented v3 scheme as of this writing (verified against https://developers.hubspot.com/docs/apps/legacy-apps/authentication/validating-requests and https://developers.hubspot.com/changelog/introducing-version-3-of-webhook-signatures). Re-check header names and the exact concatenation order against HubSpot's current docs before depending on this in production, in case the scheme has since changed.

Parameters

  • signature : str — Value of the X-HubSpot-Signature-v3 request header.

  • request_body : bytes — Raw (unparsed) request body bytes as received.

  • request_method : str — HTTP method of the inbound request (e.g. "POST").

  • request_uri : str — Full request URI HubSpot signed (path + query string).

  • timestamp : str — Value of the X-HubSpot-Request-Timestamp header.

Returns

  • bool — True if the signature matches and the request is not stale. False if the signature doesn't match or the timestamp is stale.

Raises

  • NotImplementedError — if this handler's provider isn't "hubspot".

  • ValueError — if no webhook secret was configured for this handler.

wittgenstein_crm_client.sync.webhook_handler.WebhookHandler.process_events

method WebhookHandler.process_events(events: List[Dict[str, Any]]) → List[WebhookPayload]

Parse raw incoming webhook payload dictionaries to WebhookPayload.