From 9e37e8c484cc65f1895a7e435ce8ce53933f841f Mon Sep 17 00:00:00 2001 From: David Brownman Date: Fri, 28 Aug 2026 12:27:45 -0700 Subject: [PATCH] validate that incoming urls don't redirect requests --- stripe/_api_requestor.py | 2 + stripe/_util.py | 40 +++++++++- stripe/v2/core/_event.py | 6 +- tests/test_api_requestor.py | 146 ++++++++++++++++++++++++++++++++++-- 4 files changed, 180 insertions(+), 14 deletions(-) diff --git a/stripe/_api_requestor.py b/stripe/_api_requestor.py index 96f5e3c9b..e0f22186d 100644 --- a/stripe/_api_requestor.py +++ b/stripe/_api_requestor.py @@ -29,6 +29,7 @@ log_debug, log_info, dashboard_link, + validate_path, _convert_to_stripe_object, get_api_mode, ) @@ -618,6 +619,7 @@ def _args_for_request_with_retries( "questions." ) + validate_path(url) abs_url = "%s%s" % ( self._options.base_addresses.get(base_address), url, diff --git a/stripe/_util.py b/stripe/_util.py index 386894080..3b87f2918 100644 --- a/stripe/_util.py +++ b/stripe/_util.py @@ -7,7 +7,7 @@ from stripe._api_mode import ApiMode -from urllib.parse import quote_plus +from urllib.parse import quote_plus, urlsplit from typing_extensions import Type, TYPE_CHECKING from typing import ( @@ -260,8 +260,12 @@ def _convert_to_stripe_object( klass = get_object_class(api_mode, klass_name) # TODO: this is a horrible hack. The API needs # to return something for `object` here. - - elif "data" in resp and "next_page_url" in resp: + # + # Gated on V2: this runs recursively over every nested value, so without + # the mode check any nested map in a v1 payload carrying `data` and + # `next_page_url` becomes an auto-paginating v2 collection. A malicious webhook + # could potentially choose the host of a subsequent authenticated request. + elif api_mode == "V2" and "data" in resp and "next_page_url" in resp: klass = stripe.v2.ListObject elif klass_ is not None: klass = klass_ @@ -346,6 +350,36 @@ def sanitize_id(id): return quotedId +def validate_path(path: str) -> None: + """ + Assert that a request path is origin-relative: that it begins with a single + "/" and carries no scheme, authority or userinfo. + + + + The absolute URL is built by concatenating a base address onto this path, and + no base address ends in a slash. A path like "@evil.example/v1/x" or + ".evil.example/v1/x" would therefore land inside the authority component and + send the request (Authorization header included) -- to a host of the path's + choosing. Some request paths originate in remote data (a webhook body's + related_object.url, a response's next_page_url), so the path cannot be + assumed to be well-formed. + + Stripe only ever issues plain paths, so anything else is tampering and is + rejected rather than sanitized. + """ + if not path.startswith("/") or path.startswith("//"): + raise ValueError( + f'Request path must be a string beginning with a single "/", got: {path!r}' + ) + + parts = urlsplit(path) + if parts.scheme or parts.netloc: + raise ValueError( + f"Request path may not contain a scheme or authority, got: {path!r}" + ) + + def get_api_mode(url: str) -> ApiMode: if url.startswith("/v2"): return "V2" diff --git a/stripe/v2/core/_event.py b/stripe/v2/core/_event.py index 9f054f2fc..fe014db67 100644 --- a/stripe/v2/core/_event.py +++ b/stripe/v2/core/_event.py @@ -8,7 +8,7 @@ from typing_extensions import Literal, TYPE_CHECKING from stripe._stripe_object import StripeObject, UntypedStripeObject -from stripe._util import get_api_mode +from stripe._util import get_api_mode, sanitize_id from stripe._stripe_context import StripeContext from stripe._webhook import WebhookPayload @@ -219,7 +219,7 @@ def __repr__(self) -> str: def fetch_event(self) -> Event: response = self._client.raw_request( "get", - f"/v2/core/events/{self.id}", + f"/v2/core/events/{sanitize_id(self.id)}", stripe_context=self.context, headers={"Stripe-Request-Trigger": f"event={self.id}"}, usage=["pushed_event_pull"], @@ -229,7 +229,7 @@ def fetch_event(self) -> Event: async def fetch_event_async(self) -> Event: response = await self._client.raw_request_async( "get", - f"/v2/core/events/{self.id}", + f"/v2/core/events/{sanitize_id(self.id)}", stripe_context=self.context, headers={"Stripe-Request-Trigger": f"event={self.id}"}, usage=["pushed_event_pull", "pushed_event_pull_async"], diff --git a/tests/test_api_requestor.py b/tests/test_api_requestor.py index 84deff6b9..c49803c46 100644 --- a/tests/test_api_requestor.py +++ b/tests/test_api_requestor.py @@ -23,8 +23,13 @@ StripeStreamResponse, StripeStreamResponseAsync, ) +from stripe._util import ( + validate_path, + _convert_to_stripe_object, +) from stripe.v2._deleted_object import DeletedObject from tests.http_client_mock import HTTPClientMock +from tests.test_webhook import generate_header VALID_API_METHODS = ("get", "post", "delete") @@ -175,10 +180,16 @@ def test_param_encoding(self, requestor, http_client_mock): urlencode(expectation).replace("%5B", "[").replace("%5D", "]") ) http_client_mock.stub_request( - "get", query_string=query_string, rbody="{}", rcode=200 + "get", + path=self.v1_path, + query_string=query_string, + rbody="{}", + rcode=200, ) - requestor.request("get", "", self.ENCODE_INPUTS, base_address="api") + requestor.request( + "get", self.v1_path, self.ENCODE_INPUTS, base_address="api" + ) http_client_mock.assert_requested("get", query_string=query_string) @@ -247,10 +258,11 @@ def test_ordereddict_encoding(self): assert encoded[4][0] == "ordered[nested][b]" def test_url_construction(self, requestor, http_client_mock): + # Paths must be origin-relative -- see validate_path. CASES = ( - (f"{stripe.api_base}?foo=bar", "", {"foo": "bar"}), - (f"{stripe.api_base}?foo=bar", "?", {"foo": "bar"}), - (stripe.api_base, "", {}), + (f"{stripe.api_base}/v1/foo?foo=bar", "/v1/foo", {"foo": "bar"}), + (f"{stripe.api_base}/v1/foo?foo=bar", "/v1/foo?", {"foo": "bar"}), + (f"{stripe.api_base}/v1/foo", "/v1/foo", {}), ( f"{stripe.api_base}/%20spaced?baz=5&foo=bar%24", "/%20spaced?foo=bar%24", @@ -258,8 +270,8 @@ def test_url_construction(self, requestor, http_client_mock): ), # duplicate query params keys should be deduped ( - f"{stripe.api_base}?foo=bar", - "?foo=bar", + f"{stripe.api_base}/v1/foo?foo=bar", + "/v1/foo?foo=bar", {"foo": "bar"}, ), ) @@ -982,7 +994,7 @@ def test_invalid_json(self, requestor, http_client_mock): def test_invalid_method(self, requestor): with pytest.raises(stripe.APIConnectionError): - requestor.request("foo", "bar", base_address="api") + requestor.request("foo", self.v1_path, base_address="api") def test_oauth_invalid_requestor_error(self, requestor, http_client_mock): http_client_mock.stub_request( @@ -1135,6 +1147,124 @@ def test_raw_request_with_file_param(self, requestor, http_client_mock): ) assert supplied_headers["Content-Type"] == "multipart/form-data" + ORIGIN_RELATIVE_PATHS = [ + "/v1/customers/cus_123", + "/v1/customers", + "/v2/core/accounts?page=page_123&limit=2", + # "@" is legal inside a path or query string -- it only opens an + # authority when it precedes the first "/". + "/v1/customers?email=user%40example.com", + "/v1/invoices/in_123@456", + # A backslash does not open an authority: the "/" already closed it. + "/v1/\\evil.example", + ] + + HOSTILE_PATHS = [ + # Concatenated onto a base address with no trailing slash, each of these + # moves the request's authority off api.stripe.com. + "@evil.example/v1/leak", + ":pw@evil.example/v1/leak", + ":80@evil.example/v1/leak", + # Extends the host into an attacker-owned subdomain + # (api.stripe.com.evil.example), which has a valid certificate. + ".evil.example/v1/leak", + "-evil.example/v1/leak", + "https://evil.example/v1/leak", + "//evil.example/v1/leak", + "", + "v1/customers", + ] + + @pytest.mark.parametrize("path", ORIGIN_RELATIVE_PATHS) + def test_accepts_origin_relative_path(self, path): + validate_path(path) + + @pytest.mark.parametrize("path", HOSTILE_PATHS) + def test_rejects_hostile_path(self, path): + with pytest.raises(ValueError): + validate_path(path) + + @pytest.mark.parametrize("path", HOSTILE_PATHS) + def test_request_rejects_hostile_path_without_issuing_request( + self, path, requestor, http_client_mock + ): + with pytest.raises(ValueError): + requestor.request("get", path, base_address="api") + + http_client_mock.assert_no_request() + + def test_raw_request_rejects_hostile_path_without_issuing_request( + self, http_client_mock + ): + client = stripe.StripeClient( + "sk_test_123", http_client=http_client_mock.get_mock_http_client() + ) + + with pytest.raises(ValueError): + client.raw_request("get", "@evil.example/v1/leak") + + http_client_mock.assert_no_request() + + def test_fetch_related_object_rejects_hostile_url_without_issuing_request( + self, http_client_mock + ): + client = stripe.StripeClient( + "sk_test_123", http_client=http_client_mock.get_mock_http_client() + ) + payload = json.dumps( + { + "id": "evt_123", + "object": "v2.core.event", + "type": "v2.core.account.created", + "created": "2026-01-01T00:00:00Z", + "related_object": { + "id": "acct_123", + "type": "account", + "url": "@evil.example/v1/leak", + }, + } + ) + secret = "whsec_test_secret" + header = generate_header(payload=payload, secret=secret) + + notification = client.parse_event_notification(payload, header, secret) + + with pytest.raises(ValueError): + notification.fetch_related_object() + + http_client_mock.assert_no_request() + + def test_v1_payload_does_not_produce_v2_list_object(self, requestor): + # A signature-verified v1 webhook body is attacker-shaped. Without the + # api_mode gate, `lines` here became an auto-paginating v2 collection + # whose next_page_url chose the host of the next authenticated request. + obj = _convert_to_stripe_object( + resp={ + "id": "in_123", + "object": "invoice", + "lines": { + "data": [{"id": "il_123"}], + "next_page_url": "@evil.example/v1/leak", + }, + }, + requestor=requestor, + api_mode="V1", + ) + + assert not isinstance(obj["lines"], stripe.v2.ListObject) + + def test_v2_response_still_produces_v2_list_object(self, requestor): + obj = _convert_to_stripe_object( + resp={ + "data": [{"id": "acct_123"}], + "next_page_url": "/v2/core/accounts?page=page_123", + }, + requestor=requestor, + api_mode="V2", + ) + + assert isinstance(obj, stripe.v2.ListObject) + class TestDefaultClient(object): @pytest.fixture(autouse=True)