service: guard against cross-origin __next URL redirection

CHANGELOG.md

@@ -6,10 +6,16 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ## [Unreleased]
 
-- service: `response_hook` parameter, enables inspection or rejection of raw responses (e.g. header-encoded SAP domain errors) without leaking HTTP transport objects through the OData API boundary.
-- vendor/SAP: `sap_header_error_hook(response)` — a stateless hook that detects SAP domain errors encoded in the `sap-message` response header and raises `BusinessGatewayError` before pyodata's domain handler runs.
+### Added
+
+- service: `response_hook` parameter, enables inspection or rejection of raw responses (e.g. header-encoded SAP domain errors) without leaking HTTP transport objects through the OData API boundary. - Petr Hanak
+- vendor/SAP: `sap_header_error_hook(response)` — a stateless hook that detects SAP domain errors encoded in the `sap-message` response header and raises `BusinessGatewayError` before pyodata's domain handler runs. - Petr Hanak
 - service: let FunctionRequests return a list of EntityProxies instead of the raw json, when the `ReturnType` is a Collection. - Emil B.
+
+### Fixed
+
 - model: replace regexp-based ISO datetime parsing with `datetime.fromisoformat` for `Edm.DateTime` and `Edm.DateTimeOffset` - Petr Hanak
+- service: guard against cross-origin __next URL redirection - Petr Hanak
 
 ### Removed
 - Python 3.9 is no longer supported by pyodata. Python 3.10 is now the minimal supported version.

pyodata/v2/service.py

@@ -13,7 +13,7 @@
 from email.parser import Parser
 from http.client import HTTPResponse
 from io import BytesIO
-from urllib.parse import urlencode, quote
+from urllib.parse import urlencode, quote, urlparse
 
 
 from pyodata.exceptions import HttpError, PyODataException, ExpressionError, ProgramError
@@ -296,6 +296,13 @@ def add_headers(self, value):
 
     def _build_request(self):
         if self._next_url:
+            parsed_next = urlparse(self._next_url)
+            parsed_base = urlparse(self._url)
+            if (parsed_next.scheme, parsed_next.netloc) != (parsed_base.scheme, parsed_base.netloc):
+                raise PyODataException(
+                    f'cross-origin __next URL rejected: {self._next_url!r} differs from '
+                    f'service root {self._url!r}'
+                )
             url = self._next_url
         else:
             url = urljoin(self._url, self.get_path())

tests/test_service_v2.py

@@ -2429,6 +2429,40 @@ def test_partial_listing(service):
     assert result.next_url is None
 
 
+@responses.activate
+def test_next_url_cross_origin_raises(service):
+    """__next URL pointing to a different origin must be refused before any request is dispatched."""
+    # pylint: disable=redefined-outer-name
+    cross_origin_next = "http://attacker.example.com/collect?$skiptoken=opaque"
+
+    request = service.entity_sets.Employees.get_entities().next_url(cross_origin_next)
+    with pytest.raises(PyODataException, match="cross-origin"):
+        request.execute()
+    assert len(responses.calls) == 0
+
+
+@responses.activate
+def test_next_url_same_origin_allowed(service):
+    """__next URL on the same origin (scheme + host + port) must be followed normally."""
+    # pylint: disable=redefined-outer-name
+    same_origin_next = f"{service.url}/Employees?$skiptoken=safe"
+
+    responses.add(
+        responses.GET,
+        same_origin_next,
+        json={'d': {
+            'results': [
+                {'ID': 23, 'NameFirst': 'Rob', 'NameLast': 'Ickes'}
+            ]
+        }},
+        status=200)
+
+    request = service.entity_sets.Employees.get_entities().next_url(same_origin_next)
+    result = request.execute()
+    assert len(result) == 1
+    assert result[0].ID == 23
+
+
 @responses.activate
 def test_count_with_chainable_filter_lt_operator(service):
     """Check getting $count with $filter with new filter syntax using multiple filters"""