Harden API requestor code against malicious URLs

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,

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"

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"],

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,19 +258,20 @@ 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",
                 {"baz": "5"},
             ),
             # 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)