add/adjust event parsing helpers

.claude/CLAUDE.md

@@ -33,6 +33,9 @@
 - Virtual env managed in `.venv/`; `just` recipes handle setup automatically
 - Work is not complete until `just test`, `just lint` and `just typecheck` complete successfully.
 - All code must run on all supported Python versions (full list in the test section of @.github/workflows/ci.yml)
+- In test files, prefer pytest fixtures over module-level declarations
+- Prefer f-strings over `.format` in new code.
+    - When editing a method, update any `.format` calls to use f-strings.
 
 ### Comments
 

stripe/_stripe_client.py

@@ -1,8 +1,5 @@
 # -*- coding: utf-8 -*-
 
-import json
-from collections import OrderedDict
-
 from stripe import (
     DEFAULT_API_BASE,
     DEFAULT_CONNECT_API_BASE,
@@ -23,7 +20,11 @@
 from stripe._stripe_object import StripeObject
 from stripe._stripe_response import StripeResponse
 from stripe._util import _convert_to_stripe_object, get_api_mode
-from stripe._webhook import Webhook, WebhookSignature
+from stripe._webhook import (
+    Webhook,
+    WebhookSignature,
+    maybe_extract_from_cloud_provider_envelope,
+)
 from stripe._event import Event
 from stripe.v2.core._event import EventNotification
 
@@ -213,18 +214,39 @@ def __init__(
         self.v2 = V2Services(self._requestor)
         # top-level services: The end of the section generated from our OpenAPI spec
 
+    def construct_event(
+        self,
+        payload: Union[bytes, str],
+        sig_header: str,
+        secret: str,
+        tolerance: int = Webhook.DEFAULT_TOLERANCE,
+    ) -> Event:
+        """Constructs a [snapshot event](https://docs.stripe.com/event-destinations#snapshot-payload) from an incoming webhook after verifying its authenticity. To work with a webhook that has already been verified (i.e. one from a cloud provider, an asynchronous queue, or during testing), see `construct_event_without_verification`."""
+        return Webhook.construct_event(
+            payload,
+            sig_header,
+            secret,
+            tolerance,
+            api_requestor=self._requestor,
+        )
+
+    def construct_event_without_verification(
+        self,
+        payload: Union[bytes, str],
+    ) -> Event:
+        """Constructs a [snapshot event](https://docs.stripe.com/event-destinations#snapshot-payload) from an incoming webhook without first verifying its authenticity. Should be used after calling `Webhook.verify_header(...)` or with input from a trusted source (such as [AWS EventBridge](https://docs.stripe.com/event-destinations/eventbridge), or [Azure Event Grid](https://docs.stripe.com/event-destinations/eventgrid) payload). Or, to verify & construct in a single call, use `Webhook.construct_event(...)` instead."""
+        return Webhook.construct_event_without_verification(
+            payload, api_requestor=self._requestor
+        )
+
     def parse_event_notification(
         self,
         raw: Union[bytes, str, bytearray],
         sig_header: str,
         secret: str,
         tolerance: int = Webhook.DEFAULT_TOLERANCE,
     ) -> "ALL_EVENT_NOTIFICATIONS":
-        """
-        This should be your main method for interacting with `EventNotifications`. It's the V2 equivalent of `construct_event()`, but with better typing support.
-
-        It returns a union representing all known `EventNotification` classes. They have a `type` property that can be used for narrowing, which will get you very specific type support. If parsing an event the SDK isn't familiar with, it'll instead return `UnknownEventNotification`. That's not reflected in the return type of the function (because it messes up type narrowing) but is otherwise intended.
-        """
+        """Constructs a [thin event notification](https://docs.stripe.com/event-destinations#thin-payload) from an incoming webhook after verifying its authenticity. To work with a webhook that has already been verified (i.e. one from a cloud provider, an asynchronous queue, or during testing), see `parse_event_notification_without_verification`."""
         payload = (
             cast(Union[bytes, bytearray], raw).decode("utf-8")
             if hasattr(raw, "decode")
@@ -238,30 +260,18 @@ def parse_event_notification(
             EventNotification.from_json(payload, self),
         )
 
-    def construct_event(
+    def parse_event_notification_without_verification(
         self,
         payload: Union[bytes, str],
-        sig_header: str,
-        secret: str,
-        tolerance: int = Webhook.DEFAULT_TOLERANCE,
-    ) -> Event:
-        if hasattr(payload, "decode"):
-            payload = cast(bytes, payload).decode("utf-8")
-
-        WebhookSignature.verify_header(payload, sig_header, secret, tolerance)
+    ) -> "ALL_EVENT_NOTIFICATIONS":
+        """Constructs a [thin event notification](https://docs.stripe.com/event-destinations#thin-payload) from an incoming webhook without first verifying its authenticity. Should be used after calling `Webhook.verify_header(...)` or with input from a trusted source (such as [AWS EventBridge](https://docs.stripe.com/event-destinations/eventbridge), or [Azure Event Grid](https://docs.stripe.com/event-destinations/eventgrid) payload). Or, to verify & parse in a single call, use `parse_event_notification(...)` instead."""
 
-        data = json.loads(payload, object_pairs_hook=OrderedDict)
-        event = Event._construct_from(
-            values=data,
-            requestor=self._requestor,
-            api_mode="V1",
+        return cast(
+            "ALL_EVENT_NOTIFICATIONS",
+            EventNotification.from_json(
+                maybe_extract_from_cloud_provider_envelope(payload), self
+            ),
         )
-        if event.object == "v2.core.event":  # type: ignore
-            raise ValueError(
-                "You passed a thin event notification to StripeClient.construct_event, which expects a webhook payload. Use StripeClient.parse_event_notification instead."
-            )
-
-        return event
 
     def raw_request(self, method_: str, url_: str, **params):
         params = params.copy()

stripe/_webhook.py

@@ -3,6 +3,7 @@
 import time
 from collections import OrderedDict
 from hashlib import sha256
+from typing import Any, Dict, Optional, Union
 
 # Used for global variables
 import stripe  # noqa: IMP101
@@ -12,39 +13,93 @@
 from stripe._api_requestor import _APIRequestor
 
 
+def build_v1_event(values: Dict[str, Any], requestor: _APIRequestor) -> Event:
+    """
+    Internal helper for centralizing v1 event creation
+    """
+    if values.get("object") == "v2.core.event":
+        raise ValueError(
+            "You passed a thin event notification to a method that expects a webhook body. Use the corresponding parse_event_notification* method instead."
+        )
+    return Event._construct_from(
+        values=values, requestor=requestor, api_mode="V1"
+    )
+
+
+def maybe_extract_from_cloud_provider_envelope(
+    payload: Union[bytes, str],
+):
+    """
+    Internal helper to extract the inner type from a cloud provider envelope (regardless of what's in there).
+    If the payload is already a raw Stripe event (object is 'event' or 'v2.core.event'), returns the parsed dict as-is.
+    """
+    if isinstance(payload, bytes):
+        payload = payload.decode("utf-8")
+
+    data = json.loads(payload, object_pairs_hook=OrderedDict)
+
+    # could add as many checks as we want here, but we'll start simple
+    if "detail" in data:
+        # AWS
+        # https://docs.stripe.com/event-destinations/eventbridge#event-structure
+        return data["detail"]
+    elif "specversion" in data:
+        # Azure
+        # https://docs.stripe.com/event-destinations/eventgrid#event-structure
+        return data["data"]
+    elif data.get("object") in ("event", "v2.core.event"):
+        # Raw Stripe event passed directly: pass through as-is
+        return data
+
+    raise ValueError(
+        "Unrecognized cloud event format. The payload must be an AWS EventBridge or Azure Event Grid event envelope."
+    )
+
+
 class Webhook(object):
     DEFAULT_TOLERANCE = 300
 
     @staticmethod
     def construct_event(
-        payload, sig_header, secret, tolerance=DEFAULT_TOLERANCE, api_key=None
+        payload: Union[bytes, str],
+        sig_header: str,
+        secret: str,
+        tolerance: int = DEFAULT_TOLERANCE,
+        api_key: Optional[str] = None,
+        api_requestor: Optional[_APIRequestor] = None,
     ):
-        if hasattr(payload, "decode"):
+        """Constructs a [snapshot event](https://docs.stripe.com/event-destinations#snapshot-payload) from an incoming webhook after verifying its authenticity. To work with a webhook that has already been verified (i.e. one from a cloud provider, an asynchronous queue, or during testing), see `construct_event_without_verification`."""
+        if isinstance(payload, (bytes, bytearray)):
             payload = payload.decode("utf-8")
 
         WebhookSignature.verify_header(payload, sig_header, secret, tolerance)
 
-        data = json.loads(payload, object_pairs_hook=OrderedDict)
-        event = Event._construct_from(
-            values=data,
-            requestor=_APIRequestor._global_with_options(
+        return build_v1_event(
+            json.loads(payload, object_pairs_hook=OrderedDict),
+            api_requestor
+            or _APIRequestor._global_with_options(
                 api_key=api_key or stripe.api_key
             ),
-            api_mode="V1",
         )
 
-        if event.object == "v2.core.event":  # type: ignore
-            raise ValueError(
-                "You passed a thin event notification to Webhook.construct_event, which expects a webhook payload. Use StripeClient.parse_event_notification instead."
-            )
-        return event
+    @staticmethod
+    def construct_event_without_verification(
+        payload: Union[bytes, str],
+        api_requestor: Optional[_APIRequestor] = None,
+    ):
+        """Constructs a [snapshot event](https://docs.stripe.com/event-destinations#snapshot-payload) from an incoming webhook without first verifying its authenticity. Should be used after calling `WebhookSignature.verify_header(...)` or with input from a trusted source (such as [AWS EventBridge](https://docs.stripe.com/event-destinations/eventbridge), or [Azure Event Grid](https://docs.stripe.com/event-destinations/eventgrid) payload). Or, to verify & construct in a single call, use `Webhook.construct_event(...)` instead."""
+        return build_v1_event(
+            maybe_extract_from_cloud_provider_envelope(payload),
+            api_requestor
+            or _APIRequestor._global_with_options(api_key=stripe.api_key),
+        )
 
 
 class WebhookSignature(object):
     EXPECTED_SCHEME = "v1"
 
     @staticmethod
-    def _compute_signature(payload, secret):
+    def _compute_signature(payload: str, secret: str) -> str:
         mac = hmac.new(
             secret.encode("utf-8"),
             msg=payload.encode("utf-8"),
@@ -53,14 +108,33 @@ def _compute_signature(payload, secret):
         return mac.hexdigest()
 
     @staticmethod
-    def _get_timestamp_and_signatures(header, scheme):
+    def _get_timestamp_and_signatures(header: str, scheme: str):
         list_items = [i.split("=", 2) for i in header.split(",")]
         timestamp = int([i[1] for i in list_items if i[0] == "t"][0])
         signatures = [i[1] for i in list_items if i[0] == scheme]
         return timestamp, signatures
 
     @classmethod
-    def verify_header(cls, payload, header, secret, tolerance=None):
+    def generate_signature_header(
+        cls, payload: str, secret: str, timestamp=None
+    ):
+        """Compute the `Stripe-Signature` header for a given webhook body & secret. Useful for signing payloads in unit tests."""
+        if timestamp is None:
+            timestamp = int(time.time())
+        scheme = cls.EXPECTED_SCHEME
+        signed_payload = f"{timestamp}.{payload}"
+        signature = cls._compute_signature(signed_payload, secret)
+        return f"t={timestamp},{scheme}={signature}"
+
+    @classmethod
+    def verify_header(
+        cls,
+        payload: Union[bytes, str],
+        header: str,
+        secret: str,
+        tolerance=None,
+    ):
+        """Verifies the authenticity (and recency) of a webhook, throwing a `SignatureVerificationError` if there's a mismatch. Useful for quickly validating incoming webhooks before storing them for later processing (at which time you can use the `*_without_verification` methods for parsing)."""
         try:
             timestamp, signatures = cls._get_timestamp_and_signatures(
                 header, cls.EXPECTED_SCHEME

stripe/v2/core/_event.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 
 import json
-from typing import Any, ClassVar, Dict, Optional, cast
+from typing import Any, ClassVar, Dict, Optional, Union, cast
 
 from typing_extensions import Literal, TYPE_CHECKING
 
@@ -167,13 +167,17 @@ def __init__(
         self._client = client
 
     @staticmethod
-    def from_json(payload: str, client: "StripeClient") -> "EventNotification":
+    def from_json(
+        payload: Union[str, Dict[str, Any]], client: "StripeClient"
+    ) -> "EventNotification":
         """
         Helper for constructing an Event Notification. Doesn't perform signature validation, so you
         should use StripeClient.parse_event_notification() instead for initial handling.
         This is useful in unit tests and working with EventNotifications that you've already validated the authenticity of.
         """
-        parsed_body = json.loads(payload)
+        parsed_body = (
+            json.loads(payload) if isinstance(payload, str) else payload
+        )
         if parsed_body.get("object") == "event":
             raise ValueError(
                 "You passed a webhook payload to StripeClient.parse_event_notification, which expects a thin event notification. Use StripeClient.construct_event instead."