""" Healthie Webhook Verification Service. Handles HMAC-SHA256 signature verification for incoming webhooks. Verification algorithm matches Healthie's documented JavaScript reference: https://docs.gethealthie.com/guides/webhooks/ Data string construction (matches Healthie's JS code exactly): - contentDigest = Content-Digest header value, split on "=" → hash portion only - contentType = "application/json" (hardcoded) - contentLength = length of JSON.stringify(body) — i.e., compact serialized form - dataToSign = f"{method} {path} {query} {contentDigest} {contentType} {contentLength}" HMAC-SHA256 with signing key (format: whsec_) """ import hashlib import hmac import json import logging from django.conf import settings from django.http import HttpRequest logger = logging.getLogger(__name__) class WebhookSignatureError(Exception): """Raised when webhook signature verification fails.""" class WebhookPayload: """Parsed and verified webhook payload.""" def __init__( self, resource_id: str, resource_id_type: str, event_type: str, changed_fields: list | None = None, raw_body: dict | None = None, ): self.resource_id = resource_id self.resource_id_type = resource_id_type self.event_type = event_type self.changed_fields = changed_fields or [] self.raw_body = raw_body or {} @property def is_create_event(self) -> bool: return self.event_type.endswith(".created") @property def is_update_event(self) -> bool: return self.event_type.endswith(".updated") @property def is_delete_event(self) -> bool: return self.event_type.endswith(".deleted") def verify_webhook_signature(request: HttpRequest) -> bool: """ Verify Healthie webhook HMAC signature. Matches Healthie's JavaScript reference implementation exactly. See: https://docs.gethealthie.com/guides/webhooks/ Returns True if valid. Raises WebhookSignatureError if invalid or missing. """ secret = getattr(settings, "HEALTHIE_WEBHOOK_SECRET", "") verify = getattr(settings, "HEALTHIE_WEBHOOK_VERIFY_SIGNATURE", True) if not verify: logger.warning( "Webhook signature verification disabled — accepting all payloads" ) return True # Healthie signing key format is whsec_ — the full string (including prefix) # is used as the HMAC key. Auto-prepend if user provided just the raw secret. if secret and not secret.startswith("whsec_"): secret = f"whsec_{secret}" if not secret: raise WebhookSignatureError("HEALTHIE_WEBHOOK_SECRET not configured") # Extract headers content_digest_header = request.headers.get("Content-Digest", "") signature_header = request.headers.get("Signature", "") if not content_digest_header or not signature_header: raise WebhookSignatureError("Missing Content-Digest or Signature header") # Parse content digest — extract just the hash value (after "SHA-256=") # Healthie's JS: headers['content-digest'].split('=')[1]. A malformed # header (no "=" at all, or an empty hash) is a signature failure, not a # server error. parts = content_digest_header.split("=", 1) if len(parts) != 2 or not parts[1]: raise WebhookSignatureError( f"Malformed Content-Digest header: {content_digest_header!r}" ) content_digest_hash = parts[1] # Parse signature (format: "sig1=hex_signature") signature_value = signature_header.split("=", 1)[-1] # Build data string to sign — matches Healthie's JS code exactly: # const contentDigest = headers['content-digest'].split('=')[1]; # const contentType = 'application/json'; # const contentLength = JSON.stringify(body).length; # return `${method} ${path} ${query} ${contentDigest} ${contentType} ${contentLength}`; path = request.path query = request.META.get("QUERY_STRING", "") # Parse body and re-serialize as compact JSON to match Healthie's # `JSON.stringify(body).length` behavior (which produces compact form and # preserves non-ASCII characters — hence ensure_ascii=False so the length # matches JS string length instead of counting \uXXXX escapes) try: body_obj = json.loads(request.body) content_length = str( len(json.dumps(body_obj, separators=(",", ":"), ensure_ascii=False)) ) except (json.JSONDecodeError, ValueError): # Fallback to raw body length if JSON parsing fails content_length = str(len(request.body)) data_to_sign = f"{request.method.lower()} {path} {query} {content_digest_hash} application/json {content_length}" # Compute HMAC-SHA256 key_bytes = secret.encode("utf-8") data_bytes = data_to_sign.encode("utf-8") computed = hmac.new(key_bytes, data_bytes, hashlib.sha256).hexdigest() if not hmac.compare_digest(computed, signature_value): raise WebhookSignatureError( f"Signature mismatch: computed={computed[:16]}..., received={signature_value[:16]}..." ) return True def parse_webhook(request: HttpRequest) -> WebhookPayload: """Verify and parse incoming webhook request into a WebhookPayload object.""" verify_webhook_signature(request) body = json.loads(request.body) return WebhookPayload( resource_id=str(body.get("resource_id", "")), resource_id_type=body.get("resource_id_type", ""), event_type=body.get("event_type", ""), changed_fields=body.get("changed_fields", []), raw_body=body, )