fix(webhooks): unwrap SNS notification envelope in decode_sns_payload

decode_sns_payload now JSON-parses the SNS HTTP notification envelope
({"Type":"Notification","Message":"..."}) and extracts the inner
Message field before running the SQS pipeline. Falls through to the
pre-extracted Message string when the input is not a JSON envelope so
existing call sites keep working.

Test adds a realistic SNS HTTP notification body fixture and exercises
both the new envelope path and the existing pre-extracted Message path.
Docs updated to show the typical "pass the raw HTTP body" call site.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: nijeesh-stream

docs/webhooks/webhooks_overview/webhooks_overview.md

@@ -141,17 +141,27 @@ The original `client.verify_webhook(request.body, request.headers["X-Signature"]
 
 #### SQS / SNS firehose
 
-For events delivered through SQS or SNS, call the matching helper on the message body. It base64-decodes the envelope, gzip-decompresses when the magic bytes are present, verifies the HMAC, and returns the parsed event.
+For events delivered through SQS or SNS, call the matching helper. It base64-decodes the envelope, gzip-decompresses when the magic bytes are present, verifies the HMAC, and returns the parsed event.
+
+For SQS, pass the message `Body` (already the payload):
 
 ```python
 event = client.verify_and_parse_sqs(
     sqs_message["Body"],
     sqs_message["MessageAttributes"]["X-Signature"]["StringValue"],
 )
+```
+
+For SNS, pass the **raw notification body** (the full `{"Type":"Notification", ...}` JSON envelope Amazon delivers). The SDK extracts the inner `Message` field for you, so the call site mirrors what HTTP frameworks already hand you in `request.body`:
+
+```python
+import json
 
+# Django SNS HTTP delivery
+attrs = json.loads(request.body)["MessageAttributes"]
 event = client.verify_and_parse_sns(
-    sns_notification["Message"],
-    sns_notification["MessageAttributes"]["X-Signature"]["Value"],
+    request.body,                                # raw envelope (bytes/str)
+    attrs["X-Signature"]["Value"],
 )
 ```
 
@@ -164,7 +174,7 @@ from stream_chat import webhook
 
 event = webhook.verify_and_parse_webhook(body, signature, secret)
 event = webhook.verify_and_parse_sqs(message_body, signature, secret)
-event = webhook.verify_and_parse_sns(message, signature, secret)
+event = webhook.verify_and_parse_sns(notification_body, signature, secret)
 ```
 
 The module also exposes the primitives the composites are built from — `ungzip_payload`, `decode_sqs_payload`, `decode_sns_payload`, `verify_signature` (constant-time HMAC-SHA256), and `parse_event` — for callers that need to run the steps individually.

stream_chat/tests/test_webhook_compression.py

@@ -115,14 +115,40 @@ def test_invalid_base64_raises(self):
         assert "base64" in str(exc_info.value).lower()
 
 
+def _sns_envelope(inner_message: str) -> str:
+    return json.dumps(
+        {
+            "Type": "Notification",
+            "MessageId": "22b80b92-fdea-4c2c-8f9d-bdfb0c7bf324",
+            "TopicArn": "arn:aws:sns:us-east-1:123456789012:stream-webhooks",
+            "Message": inner_message,
+            "Timestamp": "2026-05-11T10:00:00.000Z",
+            "SignatureVersion": "1",
+            "MessageAttributes": {
+                "X-Signature": {"Type": "String", "Value": "<signature placeholder>"},
+            },
+        }
+    )
+
+
 class TestDecodeSnsPayload:
-    def test_aliases_decode_sqs_payload(self):
+    def test_pre_extracted_message_matches_decode_sqs_payload(self):
         wrapped = _b64(_gzip(JSON_BODY))
         assert decode_sns_payload(wrapped) == decode_sqs_payload(wrapped)
 
-    def test_round_trip(self):
+    def test_pre_extracted_message_round_trip(self):
         assert decode_sns_payload(_b64(_gzip(JSON_BODY))) == JSON_BODY
 
+    def test_unwraps_full_sns_envelope(self):
+        wrapped = _b64(_gzip(JSON_BODY))
+        envelope = _sns_envelope(wrapped)
+        assert decode_sns_payload(envelope) == JSON_BODY
+
+    def test_handles_envelope_with_leading_whitespace(self):
+        wrapped = _b64(_gzip(JSON_BODY))
+        envelope = "\n  " + _sns_envelope(wrapped)
+        assert decode_sns_payload(envelope) == JSON_BODY
+
 
 class TestVerifySignature:
     def test_matching(self):
@@ -181,7 +207,9 @@ def test_plain_body(self):
 
     def test_gzip_body(self):
         sig = _sign(JSON_BODY)
-        assert verify_and_parse_webhook(_gzip(JSON_BODY), sig, API_SECRET) == EVENT_DICT
+        assert (
+            verify_and_parse_webhook(_gzip(JSON_BODY), sig, API_SECRET) == EVENT_DICT
+        )
 
     def test_returns_dict(self):
         sig = _sign(JSON_BODY)
@@ -239,23 +267,38 @@ def test_signature_over_compressed_or_wrapped_bytes_rejected(self):
 
 
 class TestVerifyAndParseSns:
-    def test_round_trip(self):
+    def test_pre_extracted_message_round_trip(self):
         wrapped = _b64(_gzip(JSON_BODY))
         sig = _sign(JSON_BODY)
         assert verify_and_parse_sns(wrapped, sig, API_SECRET) == EVENT_DICT
 
-    def test_matches_sqs_behaviour(self):
+    def test_matches_sqs_behaviour_for_pre_extracted_message(self):
         wrapped = _b64(_gzip(JSON_BODY))
         sig = _sign(JSON_BODY)
         assert verify_and_parse_sns(wrapped, sig, API_SECRET) == verify_and_parse_sqs(
             wrapped, sig, API_SECRET
         )
 
+    def test_full_sns_envelope(self):
+        wrapped = _b64(_gzip(JSON_BODY))
+        envelope = _sns_envelope(wrapped)
+        sig = _sign(JSON_BODY)
+        assert verify_and_parse_sns(envelope, sig, API_SECRET) == EVENT_DICT
+
+    def test_rejects_signature_over_envelope(self):
+        wrapped = _b64(_gzip(JSON_BODY))
+        envelope = _sns_envelope(wrapped)
+        sig_over_envelope = _sign(envelope.encode("utf-8"))
+        with pytest.raises(WebhookSignatureError):
+            verify_and_parse_sns(envelope, sig_over_envelope, API_SECRET)
+
 
 class TestSyncClientMethods:
     def test_verify_and_parse_webhook(self, sync_client: StreamChat):
         sig = _sign(JSON_BODY)
-        assert sync_client.verify_and_parse_webhook(_gzip(JSON_BODY), sig) == EVENT_DICT
+        assert (
+            sync_client.verify_and_parse_webhook(_gzip(JSON_BODY), sig) == EVENT_DICT
+        )
 
     def test_verify_and_parse_sqs(self, sync_client: StreamChat):
         wrapped = _b64(_gzip(JSON_BODY))
@@ -286,7 +329,9 @@ class TestAsyncClientMethods:
     async def test_verify_and_parse_webhook(self):
         sig = _sign(JSON_BODY)
         async with StreamChatAsync(api_key=API_KEY, api_secret=API_SECRET) as client:
-            assert client.verify_and_parse_webhook(_gzip(JSON_BODY), sig) == EVENT_DICT
+            assert (
+                client.verify_and_parse_webhook(_gzip(JSON_BODY), sig) == EVENT_DICT
+            )
 
     async def test_verify_and_parse_sqs(self):
         wrapped = _b64(_gzip(JSON_BODY))

stream_chat/webhook.py

@@ -19,7 +19,7 @@
 import hashlib
 import hmac
 import json
-from typing import Any, Dict, Union
+from typing import Any, Dict, Optional, Union
 
 from stream_chat.base.exceptions import WebhookSignatureError
 
@@ -70,11 +70,33 @@ def decode_sqs_payload(body: _BytesLike) -> bytes:
     return ungzip_payload(decoded)
 
 
-def decode_sns_payload(message: _BytesLike) -> bytes:
-    """Reverse the SNS firehose envelope. Byte-for-byte identical to
-    :func:`decode_sqs_payload`; exposed under both names so call sites
-    read intent."""
-    return decode_sqs_payload(message)
+def decode_sns_payload(notification_body: _BytesLike) -> bytes:
+    """Reverse an SNS HTTP notification envelope.
+
+    When ``notification_body`` is a JSON envelope
+    (``{"Type":"Notification","Message":"..."}``), the inner
+    ``Message`` field is extracted and run through
+    :func:`decode_sqs_payload` (base64-decode, then gzip-if-magic). When
+    the input is not a JSON envelope it is treated as the already-extracted
+    ``Message`` string, so call sites that pre-unwrap continue to work.
+    """
+    raw = _to_bytes(notification_body)
+    inner = _extract_sns_message(raw)
+    return decode_sqs_payload(inner if inner is not None else raw)
+
+
+def _extract_sns_message(notification_body: bytes) -> Optional[str]:
+    trimmed = notification_body.lstrip()
+    if not trimmed or trimmed[:1] != b"{":
+        return None
+    try:
+        envelope = json.loads(trimmed)
+    except (json.JSONDecodeError, ValueError):
+        return None
+    if not isinstance(envelope, dict):
+        return None
+    message = envelope.get("Message")
+    return message if isinstance(message, str) else None
 
 
 def verify_signature(