From 2693d68fbb843b2c3886a8f0fcdad2fb1a98a973 Mon Sep 17 00:00:00 2001 From: David Brownman Date: Mon, 27 Jul 2026 17:47:48 -0700 Subject: [PATCH 1/4] Add cloud provider event parsing methods to StripeClient --- stripe/_stripe_client.py | 49 +++++++--- stripe/_webhook.py | 67 ++++++++++--- tests/test_cloud_provider.py | 183 +++++++++++++++++++++++++++++++++++ tests/test_webhook.py | 2 +- 4 files changed, 274 insertions(+), 27 deletions(-) create mode 100644 tests/test_cloud_provider.py diff --git a/stripe/_stripe_client.py b/stripe/_stripe_client.py index 8f6c23ebc..79de8d63a 100644 --- a/stripe/_stripe_client.py +++ b/stripe/_stripe_client.py @@ -1,7 +1,6 @@ # -*- coding: utf-8 -*- import json -from collections import OrderedDict from stripe import ( DEFAULT_API_BASE, @@ -23,7 +22,12 @@ 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, + build_v1_event, + extract_from_cloud_provider_envelope, +) from stripe._event import Event from stripe.v2.core._event import EventNotification @@ -221,7 +225,7 @@ def parse_event_notification( 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. + This should be your main method for interacting with Thin Event Notifications. 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. """ @@ -245,23 +249,38 @@ def construct_event( secret: str, tolerance: int = Webhook.DEFAULT_TOLERANCE, ) -> Event: - if hasattr(payload, "decode"): - payload = cast(bytes, payload).decode("utf-8") + return Webhook.construct_event( + payload, + sig_header, + secret, + tolerance, + api_requestor=self._requestor, + ) - WebhookSignature.verify_header(payload, sig_header, secret, tolerance) + def construct_event_from_cloud_provider( + self, + payload: Union[bytes, str], + ) -> Event: + """Constructs an Event from an [AWS EventBridge](https://docs.stripe.com/event-destinations/eventbridge) or [Azure Event Grid](https://docs.stripe.com/event-destinations/eventgrid) payload.""" + inner = extract_from_cloud_provider_envelope(payload) + return build_v1_event(inner, self._requestor) - data = json.loads(payload, object_pairs_hook=OrderedDict) - event = Event._construct_from( - values=data, - requestor=self._requestor, - api_mode="V1", - ) - if event.object == "v2.core.event": # type: ignore + def parse_event_notification_from_cloud_provider( + self, + payload: Union[bytes, str], + ) -> "ALL_EVENT_NOTIFICATIONS": + """Parses a Thin Event Notification from an [AWS EventBridge](https://docs.stripe.com/event-destinations/eventbridge) or [Azure Event Grid](https://docs.stripe.com/event-destinations/eventgrid) payload.""" + inner = extract_from_cloud_provider_envelope(payload) + + if inner.get("object") == "event": raise ValueError( - "You passed a thin event notification to StripeClient.construct_event, which expects a webhook payload. Use StripeClient.parse_event_notification instead." + "It looks like this cloud event contains a v1 Event. Use construct_event_from_cloud_provider instead." ) - return event + return cast( + "ALL_EVENT_NOTIFICATIONS", + EventNotification.from_json(json.dumps(inner), self), + ) def raw_request(self, method_: str, url_: str, **params): params = params.copy() diff --git a/stripe/_webhook.py b/stripe/_webhook.py index fcead1260..c67175e50 100644 --- a/stripe/_webhook.py +++ b/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,12 +13,62 @@ 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 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 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 + inner = data["detail"] + elif "specversion" in data: + # Azure + # https://docs.stripe.com/event-destinations/eventgrid#event-structure + inner = data["data"] + elif isinstance(data.get("id"), str) and data["id"].startswith("evt_"): + raise ValueError( + "It looks like you passed a Stripe Event directly. Use construct_event instead to parse a webhook payload with signature verification." + ) + else: + raise ValueError( + "Unrecognized cloud event format. The payload must be an AWS EventBridge or Azure Event Grid event envelope." + ) + + return inner + + class Webhook(object): DEFAULT_TOLERANCE = 300 @staticmethod def construct_event( - payload, sig_header, secret, tolerance=DEFAULT_TOLERANCE, api_key=None + payload, + sig_header, + secret, + tolerance=DEFAULT_TOLERANCE, + api_key=None, + api_requestor: Optional[_APIRequestor] = None, ): if hasattr(payload, "decode"): payload = payload.decode("utf-8") @@ -25,20 +76,14 @@ def construct_event( 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( + data, + 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 - class WebhookSignature(object): EXPECTED_SCHEME = "v1" diff --git a/tests/test_cloud_provider.py b/tests/test_cloud_provider.py new file mode 100644 index 000000000..fa4d98469 --- /dev/null +++ b/tests/test_cloud_provider.py @@ -0,0 +1,183 @@ +import json + +import pytest + +import stripe + + +@pytest.fixture +def client(): + return stripe.StripeClient("sk_test_fake") + + +@pytest.fixture +def eventbridge_payload(): + return json.dumps( + { + "version": "0", + "id": "17e8dff5-d6cd-3770-ace9-aeac02b6ac3f", + "detail-type": "customer.created", + "source": "aws.partner/stripe.com/ed_123", + "account": "506417113029", + "time": "2024-03-07T18:27:56Z", + "region": "us-west-2", + "resources": [], + "detail": { + "id": "evt_test_123", + "object": "event", + "api_version": "2023-10-16", + "created": 1709836076, + "data": {"object": {"id": "cus_123", "object": "customer"}}, + "livemode": True, + "pending_webhooks": 0, + "request": {"id": "req_123", "idempotency_key": None}, + "type": "customer.created", + }, + } + ) + + +@pytest.fixture +def eventgrid_payload(): + return json.dumps( + { + "specversion": "1.0", + "type": "customer.created", + "source": "/providers/stripe/ed_test_123", + "id": "9aeb0fdf-c01e-0131-0922-9eb54906e209", + "time": "2025-07-11T14:30:00Z", + "subject": None, + "dataContentType": "application/cloudevents+json", + "data": { + "id": "evt_test_456", + "object": "event", + "api_version": "2023-10-16", + "created": 1709836076, + "data": {"object": {"id": "cus_456", "object": "customer"}}, + "livemode": False, + "pending_webhooks": 0, + "request": {"id": "req_456", "idempotency_key": None}, + "type": "customer.created", + }, + } + ) + + +@pytest.fixture +def eventbridge_notification_payload(): + return json.dumps( + { + "version": "0", + "id": "17e8dff5-d6cd-3770-ace9-aeac02b6ac3f", + "detail-type": "v2.core.event_destination.ping", + "source": "aws.partner/stripe.com/ed_123", + "account": "506417113029", + "time": "2024-03-07T18:27:56Z", + "region": "us-west-2", + "resources": [], + "detail": { + "id": "evt_test_789", + "object": "v2.core.event", + "type": "v2.core.event_destination.ping", + "created": "2024-03-07T18:27:56.000Z", + "context": "acct_123", + "livemode": True, + "related_object": { + "id": "ed_123", + "type": "v2.core.event_destination", + "url": "/v2/core/event_destinations/ed_123", + }, + }, + } + ) + + +@pytest.fixture +def eventgrid_notification_payload(): + return json.dumps( + { + "specversion": "1.0", + "type": "v2.core.event_destination.ping", + "source": "/providers/stripe/ed_test_123", + "id": "9aeb0fdf-c01e-0131-0922-9eb54906e209", + "time": "2025-07-11T14:30:00Z", + "data": { + "id": "evt_test_790", + "object": "v2.core.event", + "type": "v2.core.event_destination.ping", + "created": "2024-03-07T18:27:56.000Z", + "context": "acct_123", + "livemode": True, + "related_object": { + "id": "ed_test_123", + "type": "v2.core.event_destination", + "url": "/v2/core/event_destinations/ed_test_123", + }, + }, + } + ) + + +class TestConstructEventFromCloudProvider: + def test_eventbridge(self, client, eventbridge_payload): + result = client.construct_event_from_cloud_provider( + eventbridge_payload + ) + assert isinstance(result, stripe.Event) + assert result.id == "evt_test_123" + assert result.type == "customer.created" + + def test_eventgrid(self, client, eventgrid_payload): + result = client.construct_event_from_cloud_provider(eventgrid_payload) + assert isinstance(result, stripe.Event) + assert result.id == "evt_test_456" + assert result.type == "customer.created" + + def test_invalid_json(self, client): + with pytest.raises(json.JSONDecodeError): + client.construct_event_from_cloud_provider("not valid json") + + def test_raw_event_suggests_construct_event(self, client): + raw_event = json.dumps( + { + "id": "evt_test_123", + "object": "event", + "type": "customer.created", + } + ) + with pytest.raises(ValueError, match="construct_event"): + client.construct_event_from_cloud_provider(raw_event) + + def test_unrecognized_format(self, client): + with pytest.raises( + ValueError, match="Unrecognized cloud event format" + ): + client.construct_event_from_cloud_provider( + json.dumps({"foo": "bar"}) + ) + + +class TestParseEventNotificationFromCloudProvider: + def test_eventbridge(self, client, eventbridge_notification_payload): + result = client.parse_event_notification_from_cloud_provider( + eventbridge_notification_payload + ) + assert result.id == "evt_test_789" + assert result.type == "v2.core.event_destination.ping" + + def test_eventgrid(self, client, eventgrid_notification_payload): + result = client.parse_event_notification_from_cloud_provider( + eventgrid_notification_payload + ) + assert result.id == "evt_test_790" + assert result.type == "v2.core.event_destination.ping" + + def test_v1_event_suggests_construct_event_from_cloud_provider( + self, client, eventbridge_payload + ): + with pytest.raises( + ValueError, match="construct_event_from_cloud_provider" + ): + client.parse_event_notification_from_cloud_provider( + eventbridge_payload + ) diff --git a/tests/test_webhook.py b/tests/test_webhook.py index 5d3856e9b..e66e11ea8 100644 --- a/tests/test_webhook.py +++ b/tests/test_webhook.py @@ -83,7 +83,7 @@ def test_raise_on_v2_payload(self): stripe.Webhook.construct_event( DUMMY_V2_WEBHOOK_PAYLOAD, header, DUMMY_WEBHOOK_SECRET ) - assert "StripeClient.parse_event_notification" in str(e.value) + assert "parse_event_notification" in str(e.value) class TestWebhookSignature(object): From 59ba12ec1750919177b0587be1bc7eca44c78299 Mon Sep 17 00:00:00 2001 From: David Brownman Date: Thu, 30 Jul 2026 17:54:15 -0700 Subject: [PATCH 2/4] add missing methods & tests --- .claude/CLAUDE.md | 3 ++ stripe/_stripe_client.py | 73 +++++++++++++---------------- stripe/_webhook.py | 77 ++++++++++++++++++++---------- stripe/v2/core/_event.py | 8 +++- tests/test_cloud_provider.py | 91 +++++++++++++++++++++++++++++------- tests/test_v2_event.py | 13 ++++-- tests/test_webhook.py | 67 ++++++++++++++++++++------ 7 files changed, 229 insertions(+), 103 deletions(-) diff --git a/.claude/CLAUDE.md b/.claude/CLAUDE.md index 1d4331827..b65b38505 100644 --- a/.claude/CLAUDE.md +++ b/.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 diff --git a/stripe/_stripe_client.py b/stripe/_stripe_client.py index 79de8d63a..c086ba3d5 100644 --- a/stripe/_stripe_client.py +++ b/stripe/_stripe_client.py @@ -1,7 +1,5 @@ # -*- coding: utf-8 -*- -import json - from stripe import ( DEFAULT_API_BASE, DEFAULT_CONNECT_API_BASE, @@ -25,8 +23,7 @@ from stripe._webhook import ( Webhook, WebhookSignature, - build_v1_event, - extract_from_cloud_provider_envelope, + maybe_extract_from_cloud_provider_envelope, ) from stripe._event import Event from stripe.v2.core._event import EventNotification @@ -217,6 +214,31 @@ 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], @@ -224,11 +246,7 @@ def parse_event_notification( secret: str, tolerance: int = Webhook.DEFAULT_TOLERANCE, ) -> "ALL_EVENT_NOTIFICATIONS": - """ - This should be your main method for interacting with Thin Event Notifications. 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") @@ -242,44 +260,17 @@ def parse_event_notification( EventNotification.from_json(payload, self), ) - def construct_event( - self, - payload: Union[bytes, str], - sig_header: str, - secret: str, - tolerance: int = Webhook.DEFAULT_TOLERANCE, - ) -> Event: - return Webhook.construct_event( - payload, - sig_header, - secret, - tolerance, - api_requestor=self._requestor, - ) - - def construct_event_from_cloud_provider( - self, - payload: Union[bytes, str], - ) -> Event: - """Constructs an Event from an [AWS EventBridge](https://docs.stripe.com/event-destinations/eventbridge) or [Azure Event Grid](https://docs.stripe.com/event-destinations/eventgrid) payload.""" - inner = extract_from_cloud_provider_envelope(payload) - return build_v1_event(inner, self._requestor) - - def parse_event_notification_from_cloud_provider( + def parse_event_notification_without_verification( self, payload: Union[bytes, str], ) -> "ALL_EVENT_NOTIFICATIONS": - """Parses a Thin Event Notification from an [AWS EventBridge](https://docs.stripe.com/event-destinations/eventbridge) or [Azure Event Grid](https://docs.stripe.com/event-destinations/eventgrid) payload.""" - inner = extract_from_cloud_provider_envelope(payload) - - if inner.get("object") == "event": - raise ValueError( - "It looks like this cloud event contains a v1 Event. Use construct_event_from_cloud_provider instead." - ) + """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.""" return cast( "ALL_EVENT_NOTIFICATIONS", - EventNotification.from_json(json.dumps(inner), self), + EventNotification.from_json( + maybe_extract_from_cloud_provider_envelope(payload), self + ), ) def raw_request(self, method_: str, url_: str, **params): diff --git a/stripe/_webhook.py b/stripe/_webhook.py index c67175e50..586eea986 100644 --- a/stripe/_webhook.py +++ b/stripe/_webhook.py @@ -26,11 +26,12 @@ def build_v1_event(values: Dict[str, Any], requestor: _APIRequestor) -> Event: ) -def extract_from_cloud_provider_envelope( +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) + 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") @@ -41,21 +42,18 @@ def extract_from_cloud_provider_envelope( if "detail" in data: # AWS # https://docs.stripe.com/event-destinations/eventbridge#event-structure - inner = data["detail"] + return data["detail"] elif "specversion" in data: # Azure # https://docs.stripe.com/event-destinations/eventgrid#event-structure - inner = data["data"] - elif isinstance(data.get("id"), str) and data["id"].startswith("evt_"): - raise ValueError( - "It looks like you passed a Stripe Event directly. Use construct_event instead to parse a webhook payload with signature verification." - ) - else: - raise ValueError( - "Unrecognized cloud event format. The payload must be an AWS EventBridge or Azure Event Grid event envelope." - ) + return data["data"] + elif data.get("object") in ("event", "v2.core.event"): + # Raw Stripe event passed directly: pass through as-is + return data - return inner + raise ValueError( + "Unrecognized cloud event format. The payload must be an AWS EventBridge or Azure Event Grid event envelope." + ) class Webhook(object): @@ -63,33 +61,45 @@ class Webhook(object): @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) return build_v1_event( - data, + json.loads(payload, object_pairs_hook=OrderedDict), api_requestor or _APIRequestor._global_with_options( api_key=api_key or stripe.api_key ), ) + @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"), @@ -98,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 = "%d.%s" % (timestamp, payload) + signature = cls._compute_signature(signed_payload, secret) + return "t=%d,%s=%s" % (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 diff --git a/stripe/v2/core/_event.py b/stripe/v2/core/_event.py index 6b84908dd..d109a0522 100644 --- a/stripe/v2/core/_event.py +++ b/stripe/v2/core/_event.py @@ -167,13 +167,17 @@ def __init__( self._client = client @staticmethod - def from_json(payload: str, client: "StripeClient") -> "EventNotification": + def from_json( + payload: 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." diff --git a/tests/test_cloud_provider.py b/tests/test_cloud_provider.py index fa4d98469..d3f2a04f1 100644 --- a/tests/test_cloud_provider.py +++ b/tests/test_cloud_provider.py @@ -3,6 +3,8 @@ import pytest import stripe +from stripe._webhook import Webhook +from stripe.v2.core._event import EventNotification @pytest.fixture @@ -118,9 +120,9 @@ def eventgrid_notification_payload(): ) -class TestConstructEventFromCloudProvider: +class TestConstructEventWithoutVerification: def test_eventbridge(self, client, eventbridge_payload): - result = client.construct_event_from_cloud_provider( + result = client.construct_event_without_verification( eventbridge_payload ) assert isinstance(result, stripe.Event) @@ -128,16 +130,12 @@ def test_eventbridge(self, client, eventbridge_payload): assert result.type == "customer.created" def test_eventgrid(self, client, eventgrid_payload): - result = client.construct_event_from_cloud_provider(eventgrid_payload) + result = client.construct_event_without_verification(eventgrid_payload) assert isinstance(result, stripe.Event) assert result.id == "evt_test_456" assert result.type == "customer.created" - def test_invalid_json(self, client): - with pytest.raises(json.JSONDecodeError): - client.construct_event_from_cloud_provider("not valid json") - - def test_raw_event_suggests_construct_event(self, client): + def test_raw_event_passthrough(self, client): raw_event = json.dumps( { "id": "evt_test_123", @@ -145,39 +143,96 @@ def test_raw_event_suggests_construct_event(self, client): "type": "customer.created", } ) - with pytest.raises(ValueError, match="construct_event"): - client.construct_event_from_cloud_provider(raw_event) + result = client.construct_event_without_verification(raw_event) + assert isinstance(result, stripe.Event) + assert result.id == "evt_test_123" + assert result.type == "customer.created" + + def test_invalid_json(self, client): + with pytest.raises(json.JSONDecodeError): + client.construct_event_without_verification("not valid json") + + def test_thin_event_suggests_parse_event_notification_without_verification( + self, client, eventbridge_notification_payload + ): + with pytest.raises(ValueError, match="parse_event_notification"): + client.construct_event_without_verification( + eventbridge_notification_payload + ) def test_unrecognized_format(self, client): with pytest.raises( ValueError, match="Unrecognized cloud event format" ): - client.construct_event_from_cloud_provider( + client.construct_event_without_verification( json.dumps({"foo": "bar"}) ) + def test_webhook_static_method_eventbridge(self, eventbridge_payload): + result = Webhook.construct_event_without_verification( + eventbridge_payload + ) + assert isinstance(result, stripe.Event) + assert result.id == "evt_test_123" + assert result.type == "customer.created" -class TestParseEventNotificationFromCloudProvider: + +class TestParseEventNotificationWithoutVerification: def test_eventbridge(self, client, eventbridge_notification_payload): - result = client.parse_event_notification_from_cloud_provider( + result = client.parse_event_notification_without_verification( eventbridge_notification_payload ) assert result.id == "evt_test_789" assert result.type == "v2.core.event_destination.ping" def test_eventgrid(self, client, eventgrid_notification_payload): - result = client.parse_event_notification_from_cloud_provider( + result = client.parse_event_notification_without_verification( eventgrid_notification_payload ) assert result.id == "evt_test_790" assert result.type == "v2.core.event_destination.ping" - def test_v1_event_suggests_construct_event_from_cloud_provider( + def test_v1_event_suggests_construct_event_without_verification( self, client, eventbridge_payload ): + with pytest.raises(ValueError, match="construct_event"): + client.parse_event_notification_without_verification( + eventbridge_payload + ) + + def test_invalid_json(self, client): + with pytest.raises(json.JSONDecodeError): + client.parse_event_notification_without_verification( + "not valid json" + ) + + def test_unrecognized_format(self, client): with pytest.raises( - ValueError, match="construct_event_from_cloud_provider" + ValueError, match="Unrecognized cloud event format" ): - client.parse_event_notification_from_cloud_provider( - eventbridge_payload + client.parse_event_notification_without_verification( + json.dumps({"foo": "bar"}) ) + + def test_raw_event_notification_passthrough(self, client): + raw_notification = json.dumps( + { + "id": "evt_234", + "object": "v2.core.event", + "type": "v2.core.event_destination.ping", + "created": "2022-02-15T00:27:45.330Z", + "livemode": True, + "context": "acct_123", + "related_object": { + "id": "ed_123", + "type": "v2.core.event_destination", + "url": "/v2/core/event_destinations/ed_123", + }, + } + ) + result = client.parse_event_notification_without_verification( + raw_notification + ) + assert isinstance(result, EventNotification) + assert result.id == "evt_234" + assert result.type == "v2.core.event_destination.ping" diff --git a/tests/test_v2_event.py b/tests/test_v2_event.py index 9af9b5501..4ab30d5c4 100644 --- a/tests/test_v2_event.py +++ b/tests/test_v2_event.py @@ -16,7 +16,8 @@ ) from stripe.v2.core._event import UnknownEventNotification from stripe.events._event_classes import ALL_EVENT_NOTIFICATIONS -from tests.test_webhook import DUMMY_WEBHOOK_SECRET, generate_header +from stripe._webhook import WebhookSignature +from tests.test_webhook import DUMMY_WEBHOOK_SECRET EventParser = Callable[[str], ALL_EVENT_NOTIFICATIONS] @@ -83,7 +84,11 @@ def parse_event_notif(self, stripe_client: StripeClient) -> EventParser: def _parse_event_notif(payload: str): return stripe_client.parse_event_notification( - payload, generate_header(payload=payload), DUMMY_WEBHOOK_SECRET + payload, + WebhookSignature.generate_signature_header( + payload, DUMMY_WEBHOOK_SECRET + ), + DUMMY_WEBHOOK_SECRET, ) return _parse_event_notif @@ -232,7 +237,9 @@ def test_v2_events_integration( event_notif = stripe_client.parse_event_notification( v2_payload_no_data, - generate_header(payload=v2_payload_no_data), + WebhookSignature.generate_signature_header( + v2_payload_no_data, DUMMY_WEBHOOK_SECRET + ), DUMMY_WEBHOOK_SECRET, ) assert event_notif.type == "v1.billing.meter.error_report_triggered" diff --git a/tests/test_webhook.py b/tests/test_webhook.py index e66e11ea8..fade53a01 100644 --- a/tests/test_webhook.py +++ b/tests/test_webhook.py @@ -4,6 +4,7 @@ import stripe from stripe._error import SignatureVerificationError +from stripe._webhook import WebhookSignature DUMMY_WEBHOOK_PAYLOAD = """{ @@ -23,19 +24,40 @@ DUMMY_WEBHOOK_SECRET = "whsec_test_secret" -def generate_header(**kwargs): - timestamp = kwargs.get("timestamp", int(time.time())) - payload = kwargs.get("payload", DUMMY_WEBHOOK_PAYLOAD) - secret = kwargs.get("secret", DUMMY_WEBHOOK_SECRET) - scheme = kwargs.get("scheme", stripe.WebhookSignature.EXPECTED_SCHEME) - signature = kwargs.get("signature", None) - if signature is None: - payload_to_sign = "%d.%s" % (timestamp, payload) - signature = stripe.WebhookSignature._compute_signature( - payload_to_sign, secret - ) - header = "t=%d,%s=%s" % (timestamp, scheme, signature) - return header +def generate_header( + payload=DUMMY_WEBHOOK_PAYLOAD, secret=DUMMY_WEBHOOK_SECRET, timestamp=None +): + """Thin wrapper around WebhookSignature.generate_signature_header for tests.""" + return WebhookSignature.generate_signature_header( + payload, secret, timestamp + ) + + +def _build_header_with_scheme( + scheme, + payload=DUMMY_WEBHOOK_PAYLOAD, + secret=DUMMY_WEBHOOK_SECRET, + timestamp=None, +): + """Build a header with a custom scheme, for testing scheme-mismatch error paths.""" + if timestamp is None: + timestamp = int(time.time()) + payload_to_sign = "%d.%s" % (timestamp, payload) + signature = WebhookSignature._compute_signature(payload_to_sign, secret) + return "t=%d,%s=%s" % (timestamp, scheme, signature) + + +def _build_header_with_signature( + signature, payload=DUMMY_WEBHOOK_PAYLOAD, timestamp=None +): + """Build a header with a pre-computed (possibly bad) signature, for testing signature-mismatch error paths.""" + if timestamp is None: + timestamp = int(time.time()) + return "t=%d,%s=%s" % ( + timestamp, + WebhookSignature.EXPECTED_SCHEME, + signature, + ) class TestWebhook(object): @@ -98,7 +120,7 @@ def test_raise_on_malformed_header(self): ) def test_raise_on_no_signatures_with_expected_scheme(self): - header = generate_header(scheme="v0") + header = _build_header_with_scheme("v0") with pytest.raises( SignatureVerificationError, match="No signatures found with expected scheme v1", @@ -108,7 +130,7 @@ def test_raise_on_no_signatures_with_expected_scheme(self): ) def test_raise_on_no_valid_signatures_for_payload(self): - header = generate_header(signature="bad_signature") + header = _build_header_with_signature("bad_signature") with pytest.raises( SignatureVerificationError, match="No signatures found matching the expected signature for payload", @@ -142,6 +164,21 @@ def test_header_contains_valid_signature(self): DUMMY_WEBHOOK_PAYLOAD, header, DUMMY_WEBHOOK_SECRET, tolerance=10 ) + def test_generate_signature_header(self): + timestamp = 1234567890 + header = WebhookSignature.generate_signature_header( + DUMMY_WEBHOOK_PAYLOAD, DUMMY_WEBHOOK_SECRET, timestamp + ) + # Header must follow the format t=,v1= + assert header.startswith("t=%d,v1=" % timestamp) + parts = dict(part.split("=", 1) for part in header.split(",")) + assert parts["t"] == str(timestamp) + assert len(parts["v1"]) == 64 # SHA-256 hex digest is 64 chars + # The generated header must pass verification (no tolerance since timestamp is old) + assert WebhookSignature.verify_header( + DUMMY_WEBHOOK_PAYLOAD, header, DUMMY_WEBHOOK_SECRET + ) + def test_timestamp_off_but_no_tolerance(self): header = generate_header(timestamp=12345) assert stripe.WebhookSignature.verify_header( From 92bee478687d75a3b8afa56f51e79c1c4ea695dd Mon Sep 17 00:00:00 2001 From: David Brownman Date: Fri, 31 Jul 2026 14:54:03 -0700 Subject: [PATCH 3/4] fix lint --- stripe/v2/core/_event.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/stripe/v2/core/_event.py b/stripe/v2/core/_event.py index d109a0522..8bf3e7827 100644 --- a/stripe/v2/core/_event.py +++ b/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 @@ -168,7 +168,7 @@ def __init__( @staticmethod def from_json( - payload: str | Dict[str, Any], client: "StripeClient" + payload: Union[str, Dict[str, Any]], client: "StripeClient" ) -> "EventNotification": """ Helper for constructing an Event Notification. Doesn't perform signature validation, so you From 1ca60dfe36061cdc5d9a4ac7122c8f8926f47e33 Mon Sep 17 00:00:00 2001 From: David Brownman Date: Fri, 31 Jul 2026 15:59:15 -0700 Subject: [PATCH 4/4] use f-strings --- stripe/_webhook.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/stripe/_webhook.py b/stripe/_webhook.py index 586eea986..a0c81ac36 100644 --- a/stripe/_webhook.py +++ b/stripe/_webhook.py @@ -122,9 +122,9 @@ def generate_signature_header( if timestamp is None: timestamp = int(time.time()) scheme = cls.EXPECTED_SCHEME - signed_payload = "%d.%s" % (timestamp, payload) + signed_payload = f"{timestamp}.{payload}" signature = cls._compute_signature(signed_payload, secret) - return "t=%d,%s=%s" % (timestamp, scheme, signature) + return f"t={timestamp},{scheme}={signature}" @classmethod def verify_header(