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/CHANGELOG.md b/CHANGELOG.md index b7064f90f..1af01cb1c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,58 @@ This release changes the pinned API version to 2026-07-29.preview. * ⚠️ Change `V2.MoneyManagement.ReceivedDebit.BankTransfer.us_bank_account` to be optional * Add support for error codes `us_bank_account_microdeposits_cannot_be_confirmed` and `us_bank_account_microdeposits_cannot_be_sent` on `ControlledByAlternateResourceError` +## 15.4.0 - 2026-07-29 +This release changes the pinned API version to 2026-07-29.dahlia. + +* [#1857](https://github.com/stripe/stripe-python/pull/1857) Update generated code + * Add support for new resource `financial_connections.Authorization` + * Add support for `unreject` method on resource `Account` + * Add support for `list` method on resource `PaymentRecord` + * Add support for new values `mass_transit_parking_tax` and `parking_tax` on enums `Tax.Calculation.ShippingCost.TaxBreakdown.TaxRateDetail.tax_type`, `Tax.Calculation.TaxBreakdown.TaxRateDetail.tax_type`, `Tax.CalculationLineItem.TaxBreakdown.TaxRateDetail.tax_type`, and `Tax.Transaction.ShippingCost.TaxBreakdown.TaxRateDetail.tax_type` + * Add support for new value `chaps` on enums `FundingInstructions.BankTransfer.FinancialAddress.supported_networks` and `PaymentIntent.NextAction.DisplayBankTransferInstruction.FinancialAddress.supported_networks` + * Add support for `smart_disputes_management` on `AccountSession.Component.DisputesList.Feature`, `AccountSession.Component.Payment.Feature`, `AccountSession.Component.PaymentDetail.Feature`, `AccountSession.Component.PaymentDispute.Feature`, `AccountSessionCreateParamsComponentDisputesListFeature`, `AccountSessionCreateParamsComponentPaymentDetailFeature`, `AccountSessionCreateParamsComponentPaymentDisputeFeature`, and `AccountSessionCreateParamsComponentPaymentFeature` + * Add support for `administrative_address` and `principal_place_of_business` on `Account.Company`, `AccountCreateParamsCompany`, `AccountModifyParamsCompany`, and `TokenCreateParamsAccountCompany` + * Add support for `sepa_debit_payments` on `AccountModifyParamsSetting` + * Remove support for `proof_of_registration` on `AccountCreateParamsDocument`. This field was limited-use and is being deprecated. + * Add support for `payouts_action` on `AccountRejectParams` + * Add support for new value `data_share_only` on enums `Charge.PaymentMethodDetail.Card.ThreeDSecure.result`, `PaymentAttemptRecord.PaymentMethodDetail.Card.ThreeDSecure.result`, `PaymentRecord.PaymentMethodDetail.Card.ThreeDSecure.result`, and `SetupAttempt.PaymentMethodDetail.Card.ThreeDSecure.result` + * Add support for new values `bnp_paribas`, `citibank`, and `mbsb_bank` on enums `Charge.PaymentMethodDetail.Fpx.bank`, `ConfirmationToken.PaymentMethodPreview.Fpx.bank`, `PaymentAttemptRecord.PaymentMethodDetail.Fpx.bank`, `PaymentMethod.Fpx.bank`, and `PaymentRecord.PaymentMethodDetail.Fpx.bank` + * Remove support for `dynamic_tax_rates` on `checkout.SessionCreateParamsLineItem`. This field was limited-use and is being deprecated. + * Add support for `setup_future_usage` on `Checkout.Session.PaymentMethodOption.Payco`, `Checkout.Session.PaymentMethodOption.SamsungPay`, `PaymentIntent.PaymentMethodOption.Payco`, `PaymentIntent.PaymentMethodOption.SamsungPay`, `PaymentIntentConfirmParamsPaymentMethodOptionPayco`, `PaymentIntentConfirmParamsPaymentMethodOptionSamsungPay`, `PaymentIntentCreateParamsPaymentMethodOptionPayco`, `PaymentIntentCreateParamsPaymentMethodOptionSamsungPay`, `PaymentIntentModifyParamsPaymentMethodOptionPayco`, `PaymentIntentModifyParamsPaymentMethodOptionSamsungPay`, `PaymentLinkModifyParamsPaymentIntentDatum`, `checkout.SessionCreateParamsPaymentMethodOptionPayco`, and `checkout.SessionCreateParamsPaymentMethodOptionSamsungPay` + * Add support for new value `ic_nif` on enums `Checkout.Session.CustomerDetail.TaxId.type`, `Invoice.CustomerTaxId.type`, `Tax.Calculation.CustomerDetail.TaxId.type`, `Tax.Transaction.CustomerDetail.TaxId.type`, and `TaxId.type` + * Add support for new values `bnp_paribas`, `citibank`, and `mbsb_bank` on enums `ConfirmationTokenCreateParamsPaymentMethodDatumFpx.bank`, `PaymentIntentConfirmParamsPaymentMethodDatumFpx.bank`, `PaymentIntentCreateParamsPaymentMethodDatumFpx.bank`, `PaymentIntentModifyParamsPaymentMethodDatumFpx.bank`, `PaymentMethodCreateParamsFpx.bank`, `SetupIntentConfirmParamsPaymentMethodDatumFpx.bank`, `SetupIntentCreateParamsPaymentMethodDatumFpx.bank`, and `SetupIntentModifyParamsPaymentMethodDatumFpx.bank` + * Add support for new value `ic_nif` on enums `CustomerCreateParamsTaxIdDatum.type`, `CustomerCreateTaxIdParams.type`, `InvoiceCreatePreviewParamsCustomerDetailTaxId.type`, `TaxIdCreateParams.type`, and `tax.CalculationCreateParamsCustomerDetailTaxId.type` + * Add support for `network` on `Dispute.PaymentMethodDetail.Card` + * Add support for new values `financial_connections.account.expected_deactivation_date_updated`, `financial_connections.account.supported_payment_method_types_updated`, `financial_connections.account.upcoming_deactivation`, `financial_connections.authorization.expected_deactivation_date_updated`, and `financial_connections.authorization.upcoming_deactivation` on enum `Event.type` + * Add support for `limits` and `manual_entry` on `FinancialConnections.Session` and `financial_connections.SessionCreateParams` + * Add support for `require_payment_method_support` on `FinancialConnections.Session.Filter` and `financial_connections.SessionCreateParamsFilter` + * Add support for `bank_account_token` on `FinancialConnections.Session` + * Add support for new values `alipay` and `mb_way` on enums `Invoice.PaymentSetting.payment_method_types`, `InvoiceCreateParamsPaymentSetting.payment_method_types`, `InvoiceModifyParamsPaymentSetting.payment_method_types`, `Subscription.PaymentSetting.payment_method_types`, `SubscriptionCreateParamsPaymentSetting.payment_method_types`, and `SubscriptionModifyParamsPaymentSetting.payment_method_types` + * Add support for new values `mass_transit_parking_tax` and `parking_tax` on enums `InvoiceAddLinesParamsLineTaxAmountTaxRateDatum.tax_type`, `InvoiceLineItemModifyParamsTaxAmountTaxRateDatum.tax_type`, `InvoiceUpdateLinesParamsLineTaxAmountTaxRateDatum.tax_type`, `TaxRate.tax_type`, `TaxRateCreateParams.tax_type`, and `TaxRateModifyParams.tax_type` + * Add support for `metadata` on `InvoiceCreatePreviewParamsSubscriptionDetail` + * Add support for new value `stripe_internal_error` on enum `Issuing.Authorization.RequestHistory.reason` + * Add support for `business_name` on `Issuing.Card.Shipping`, `issuing.CardCreateParamsShipping`, and `issuing.CardModifyParamsShipping` + * Add support for new value `correos` on enum `Issuing.Card.Shipping.carrier` + * Add support for `allowed_payment_method_types` on `PaymentIntentConfirmParams`, `PaymentIntentCreateParams`, `PaymentIntentModifyParams`, `PaymentIntent`, `SetupIntentConfirmParams`, `SetupIntentCreateParams`, `SetupIntentModifyParams`, and `SetupIntent` + * Add support for `referrer` on `PaymentIntentConfirmParamsRadarOption` and `PaymentIntentCreateParamsRadarOption` + * Add support for `consent_collection` and `shipping_options` on `PaymentLinkModifyParams` + * Add support for `custom_fields`, `description`, and `footer` on `Quote.InvoiceSetting`, `QuoteCreateParamsInvoiceSetting`, `QuoteModifyParamsInvoiceSetting`, `SubscriptionSchedule.DefaultSetting.InvoiceSetting`, `SubscriptionSchedule.Phase.InvoiceSetting`, `SubscriptionScheduleCreateParamsDefaultSettingInvoiceSetting`, `SubscriptionScheduleCreateParamsPhaseInvoiceSetting`, `SubscriptionScheduleModifyParamsDefaultSettingInvoiceSetting`, and `SubscriptionScheduleModifyParamsPhaseInvoiceSetting` + * Add support for `customer_account` and `customer` on `Refund` + * Add support for `payment_method` on `Refund` and `Topup` + * Add support for `trial` on `SubscriptionSchedule.Phase` + * Add support for `mass_transit_parking_tax` and `parking_tax` on `Tax.Registration.CountryOption.Me` and `tax.RegistrationCreateParamsCountryOptionMe` + * Add support for new values `mass_transit_parking_tax` and `parking_tax` on enum `tax.RegistrationCreateParamsCountryOptionMe.type` + * Add support for new values `mass_transit_parking_tax` and `parking_tax` on enum `Tax.Registration.CountryOption.Me.type` + * Add support for `initiated_by` and `payment_method_options` on `Topup` + * Add support for new values `financial_connections.account.expected_deactivation_date_updated`, `financial_connections.account.supported_payment_method_types_updated`, `financial_connections.account.upcoming_deactivation`, `financial_connections.authorization.expected_deactivation_date_updated`, and `financial_connections.authorization.upcoming_deactivation` on enums `WebhookEndpointCreateParams.enabled_events` and `WebhookEndpointModifyParams.enabled_events` + * Add support for new value `2026-07-29.dahlia` on enum `WebhookEndpointCreateParams.api_version` + * Add support for `additional_addresses` on `V2.Core.Account.Identity.BusinessDetail`, `v2.core.AccountCreateParamsIdentityBusinessDetail`, `v2.core.AccountModifyParamsIdentityBusinessDetail`, and `v2.core.AccountTokenCreateParamsIdentityBusinessDetail` + * Add support for snapshot events `financial_connections.account.expected_deactivation_date_updated`, `financial_connections.account.supported_payment_method_types_updated`, and `financial_connections.account.upcoming_deactivation` with resource `financial_connections.Account` + * Add support for snapshot events `financial_connections.authorization.expected_deactivation_date_updated` and `financial_connections.authorization.upcoming_deactivation` with resource `financial_connections.Authorization` + * Enum type annotations for non-exhaustive ("open") enums are now typed as `Union[Literal[...], str]` instead of plain `Literal[...]`. Many Stripe enums are open, meaning new values may appear in new versions of the API. This change ensures these fields have the correct type for both the values known at SDK release time and other values that may be added later. Refer to the [API Reference](https://docs.stripe.com) for the latest set of allowed values. +* [#1853](https://github.com/stripe/stripe-python/pull/1853) Fix intermittent Windows test flake in async httpx integration tests +* [#1849](https://github.com/stripe/stripe-python/pull/1849) Use ip address directly instead of localhost to speed up windows tests + ## 15.4.0b1 - 2026-06-24 This release changes the pinned API version to 2026-06-24.preview. diff --git a/stripe/__init__.py b/stripe/__init__.py index e1f0e7670..442bf87a6 100644 --- a/stripe/__init__.py +++ b/stripe/__init__.py @@ -30,6 +30,7 @@ upload_api_base: str = DEFAULT_UPLOAD_API_BASE meter_events_api_base: str = DEFAULT_METER_EVENTS_API_BASE api_version: str = _ApiVersion.CURRENT +major_api_version: str = _ApiVersion.CURRENT_MAJOR verify_ssl_certs: bool = True proxy: Optional[str] = None default_http_client: Optional["HTTPClient"] = None @@ -138,6 +139,10 @@ def add_beta_version( api_version = f"{api_version}; {beta_name}={beta_version}" +from stripe._util import _emit_claude_code_hint + +_emit_claude_code_hint() + # The beginning of the section generated from our OpenAPI spec from importlib import import_module diff --git a/stripe/_api_version.py b/stripe/_api_version.py index 5bc23ba39..5ca2c61e3 100644 --- a/stripe/_api_version.py +++ b/stripe/_api_version.py @@ -2,3 +2,4 @@ # File generated from our OpenAPI spec class _ApiVersion: CURRENT = "2026-07-29.preview" + CURRENT_MAJOR = "" diff --git a/stripe/_stripe_client.py b/stripe/_stripe_client.py index c96043a88..93f0349f0 100644 --- a/stripe/_stripe_client.py +++ b/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, @@ -27,7 +24,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 @@ -227,6 +228,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], @@ -234,11 +260,7 @@ def parse_event_notification( 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") @@ -252,30 +274,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() diff --git a/stripe/_util.py b/stripe/_util.py index f1ec0b639..c5dfbd950 100644 --- a/stripe/_util.py +++ b/stripe/_util.py @@ -388,3 +388,20 @@ def _wrapper(*args, **kwargs): return class_method(*args, **kwargs) return _wrapper + + +# If we detect environment variables that indicate we're running in a Claude Code +# session, emit a hint to the parent process so it can load the plugin. +# https://code.claude.com/docs/en/plugin-hints +def claude_code_hint_line() -> str: + return '\n' + + +def _emit_claude_code_hint() -> None: + try: + if os.environ.get("CLAUDECODE") or os.environ.get( + "CLAUDE_CODE_CHILD_SESSION" + ): + sys.stderr.write(claude_code_hint_line()) + except Exception: + pass diff --git a/stripe/_webhook.py b/stripe/_webhook.py index fcead1260..592ee512e 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,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 := data.get("detail"): + # AWS + # https://docs.stripe.com/event-destinations/eventbridge#event-structure + return detail + elif "specversion" in data and (data_ := data.get("data")): + # Azure + # https://docs.stripe.com/event-destinations/eventgrid#event-structure + return data_ + elif data.get("object") in ("event", "v2.core.event"): + # Raw Stripe event passed directly: pass through as-is + return data + + raise ValueError( + "Unrecognized event format. The payload must be an AWS EventBridge/Azure Event Grid event envelope or a Stripe webhook (thin event notification or snapshot)." + ) + + 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 diff --git a/stripe/v2/core/_event.py b/stripe/v2/core/_event.py index b46425608..3325dd946 100644 --- a/stripe/v2/core/_event.py +++ b/stripe/v2/core/_event.py @@ -3,7 +3,9 @@ # -*- coding: utf-8 -*- import json -from typing import Any, ClassVar, Dict, Optional, cast +from typing import Any, ClassVar, Dict, Optional, cast, Union +# v2-event-imports: The beginning of the section generated from our OpenAPI spec +# v2-event-imports: The end of the section generated from our OpenAPI spec from typing_extensions import Literal, TYPE_CHECKING @@ -143,6 +145,10 @@ class EventNotification: """ Livemode indicates if the event is from a production(true) or test(false) account. """ + object: str + """ + String representing the object's type. Objects of the same type share the same value. + """ context: Optional[StripeContext] = None """ [Optional] Authentication context needed to fetch the event or related object. @@ -156,6 +162,7 @@ def __init__( self, parsed_body: Dict[str, Any], client: "StripeClient" ) -> None: self.id = parsed_body["id"] + self.object = parsed_body["object"] self.type = parsed_body["type"] self.created = parsed_body["created"] self.livemode = bool(parsed_body.get("livemode")) @@ -169,17 +176,28 @@ 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. + This is useful in unit tests and working with EventNotifications whose authenticity you've already validated. """ - 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." ) + if ( + parsed_body.get("object") is not None + and parsed_body.get("object") != "v2.core.event" + ): + raise ValueError( + f"Unexpected object type '{parsed_body.get('object')}'. Expected 'v2.core.event' for an event notification." + ) # circular import busting from stripe.events._event_classes import ( @@ -193,7 +211,7 @@ def from_json(payload: str, client: "StripeClient") -> "EventNotification": def __repr__(self) -> str: return f"" - def fetch_event(self) -> "Event": + def fetch_event(self) -> Event: response = self._client.raw_request( "get", f"/v2/core/events/{self.id}", @@ -201,9 +219,9 @@ def fetch_event(self) -> "Event": headers={"Stripe-Request-Trigger": f"event={self.id}"}, usage=["pushed_event_pull"], ) - return cast("Event", self._client.deserialize(response, api_mode="V2")) + return cast(Event, self._client.deserialize(response, api_mode="V2")) - async def fetch_event_async(self) -> "Event": + async def fetch_event_async(self) -> Event: response = await self._client.raw_request_async( "get", f"/v2/core/events/{self.id}", @@ -211,7 +229,7 @@ async def fetch_event_async(self) -> "Event": headers={"Stripe-Request-Trigger": f"event={self.id}"}, usage=["pushed_event_pull", "pushed_event_pull_async"], ) - return cast("Event", self._client.deserialize(response, api_mode="V2")) + return cast(Event, self._client.deserialize(response, api_mode="V2")) class UnknownEventNotification(EventNotification): diff --git a/tests/test_cloud_provider.py b/tests/test_cloud_provider.py new file mode 100644 index 000000000..858cdd006 --- /dev/null +++ b/tests/test_cloud_provider.py @@ -0,0 +1,281 @@ +import json + +import pytest + +import stripe +from stripe._webhook import Webhook +from stripe.v2.core._event import EventNotification + + +@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 TestConstructEventWithoutVerification: + def test_eventbridge(self, client, eventbridge_payload): + result = client.construct_event_without_verification( + 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_without_verification(eventgrid_payload) + assert isinstance(result, stripe.Event) + assert result.id == "evt_test_456" + assert result.type == "customer.created" + + def test_raw_event_passthrough(self, client): + raw_event = json.dumps( + { + "id": "evt_test_123", + "object": "event", + "type": "customer.created", + } + ) + 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 event format"): + client.construct_event_without_verification( + json.dumps({"foo": "bar"}) + ) + + def test_azure_envelope_missing_data_field(self, client): + payload = json.dumps( + { + "specversion": "1.0", + "type": "customer.created", + "source": "/providers/stripe/ed_test_123", + "id": "test-missing-data", + } + ) + with pytest.raises(ValueError, match="Unrecognized event format"): + client.construct_event_without_verification(payload) + + 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 TestParseEventNotificationWithoutVerification: + def test_eventbridge(self, client, eventbridge_notification_payload): + 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_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_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="Unrecognized event format"): + client.parse_event_notification_without_verification( + json.dumps({"foo": "bar"}) + ) + + def test_azure_envelope_missing_data_field(self, client): + payload = json.dumps( + { + "specversion": "1.0", + "type": "v2.core.event_destination.ping", + "source": "/providers/stripe/ed_test_123", + "id": "test-missing-data", + } + ) + with pytest.raises(ValueError, match="Unrecognized event format"): + client.parse_event_notification_without_verification(payload) + + def test_unexpected_object_type_in_event_notification(self, client): + # Wrap in an EventBridge envelope so envelope extraction succeeds, + # letting the object-type guard inside from_json fire. + payload = 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": { + "object": "customer", + "type": "customer.created", + "id": "cus_123", + }, + } + ) + with pytest.raises(ValueError, match="Unexpected object type"): + client.parse_event_notification_without_verification(payload) + + 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_exports.py b/tests/test_exports.py index 44f11846d..3aff0a530 100644 --- a/tests/test_exports.py +++ b/tests/test_exports.py @@ -4,6 +4,8 @@ import subprocess import sys +from stripe._util import claude_code_hint_line + def assert_output(code: str, expected: str) -> None: process = subprocess.Popen( @@ -14,6 +16,7 @@ def assert_output(code: str, expected: str) -> None: stdout, stderr = process.communicate() + stderr = stderr.replace(claude_code_hint_line().encode(), b"") assert not stderr, f"Error: {stderr.decode()}" output = stdout.decode().strip() diff --git a/tests/test_integration.py b/tests/test_integration.py index bf80daf9f..9572ff127 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -41,7 +41,7 @@ def __init__(self, handler: BaseHTTPRequestHandler): class MyTestHandler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" - timeout = 30 + timeout = 0.5 num_requests = 0 diff --git a/tests/test_stripe_context.py b/tests/test_stripe_context.py index a06d16dd6..a0f40787f 100644 --- a/tests/test_stripe_context.py +++ b/tests/test_stripe_context.py @@ -156,6 +156,7 @@ def test_event_notification_parsing(self): mock_client = Mock() parsed_body = { "id": "evt_123", + "object": "v2.core.event", "type": "test.event", "created": "2023-01-01T00:00:00Z", "livemode": False, @@ -171,6 +172,7 @@ def test_event_notification_no_context(self): mock_client = Mock() parsed_body = { "id": "evt_123", + "object": "v2.core.event", "type": "test.event", "created": "2023-01-01T00:00:00Z", "livemode": False, @@ -184,6 +186,7 @@ def test_event_notification_empty_context(self): mock_client = Mock() parsed_body = { "id": "evt_123", + "object": "v2.core.event", "type": "test.event", "created": "2023-01-01T00:00:00Z", "livemode": False, diff --git a/tests/test_util.py b/tests/test_util.py index 1331dbe20..a8dc01c80 100644 --- a/tests/test_util.py +++ b/tests/test_util.py @@ -1,3 +1,5 @@ +import io +import os import sys from collections import namedtuple @@ -11,6 +13,8 @@ log_info, log_debug, sanitize_id, + claude_code_hint_line, + _emit_claude_code_hint, ) from stripe import Balance from stripe._api_mode import ApiMode @@ -178,3 +182,36 @@ def test_sanitize_id(self): ) def test_get_api_mode(self, url: str, expected: ApiMode): assert get_api_mode(url) == expected + + +class TestEmitClaudeCodeHint: + _HINT = claude_code_hint_line() + + def _capture(self, env_vars: dict) -> str: + buf = io.StringIO() + original = os.environ.copy() + try: + for k in ("CLAUDECODE", "CLAUDE_CODE_CHILD_SESSION"): + os.environ.pop(k, None) + os.environ.update(env_vars) + old_stderr, sys.stderr = sys.stderr, buf + try: + _emit_claude_code_hint() + finally: + sys.stderr = old_stderr + finally: + os.environ.clear() + os.environ.update(original) + return buf.getvalue() + + def test_emits_when_CLAUDECODE_set(self): + assert self._capture({"CLAUDECODE": "1"}) == self._HINT + + def test_emits_when_CLAUDE_CODE_CHILD_SESSION_set(self): + assert ( + self._capture({"CLAUDE_CODE_CHILD_SESSION": "session-id"}) + == self._HINT + ) + + def test_no_emit_without_env_vars(self): + assert self._capture({}) == "" diff --git a/tests/test_v2_event.py b/tests/test_v2_event.py index 9b8df8bf8..8ac40e40f 100644 --- a/tests/test_v2_event.py +++ b/tests/test_v2_event.py @@ -18,7 +18,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] @@ -85,7 +86,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 @@ -99,6 +104,7 @@ def test_parses_event_notif( notif, V1BillingMeterErrorReportTriggeredEventNotification ) assert notif.id == "evt_234" + assert notif.object == "v2.core.event" assert notif.related_object assert notif.related_object.id == "mtr_123" @@ -234,7 +240,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 5d3856e9b..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): @@ -83,7 +105,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): @@ -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(