Restrict MSGraph pagination nextLink to the configured host

providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/msgraph.py

@@ -221,6 +221,7 @@ def __init__(
         else:
             self.scopes = scopes or [self.DEFAULT_SCOPE]
         self.api_version = self.resolve_api_version_from_value(api_version)
+        self.allowed_netloc: str | None = None
 
     def _ensure_protocol(self, host: str | None, schema: str = "https") -> str | None:
         """Ensure URL has http:// or https:// protocol prefix."""
@@ -490,6 +491,12 @@ async def get_async_conn(self) -> RequestAdapter:
             self.cached_request_adapters[self.conn_id] = (api_version, request_adapter)
 
         self.api_version = api_version
+        # The pagination link (e.g. ``@odata.nextLink``) is echoed from the API response and is
+        # re-fetched with the connection's bearer token attached. Kiota only scopes that token to
+        # ``allowed_hosts``, which defaults to empty (any host) unless configured, so a tampered
+        # response could redirect the token off-host. Pin follow-up requests to the configured
+        # endpoint's host (CWE-918).
+        self.allowed_netloc = urlparse(request_adapter.base_url).netloc
         return request_adapter
 
     def get_proxies(self, config: dict) -> dict | None:
@@ -648,7 +655,7 @@ async def run(
                     responses.append(response)
 
                     if pagination_function:
-                        url, query_parameters = execute_callable(
+                        next_url, query_parameters = execute_callable(
                             pagination_function,
                             response=response,
                             url=url,
@@ -660,6 +667,16 @@ async def run(
                             data=data,
                             responses=lambda: responses,
                         )
+                        if (
+                            next_url
+                            and next_url.startswith("http")
+                            and urlparse(next_url).netloc != self.allowed_netloc
+                        ):
+                            raise ValueError(
+                                f"Refusing to follow pagination link {next_url!r}: its host differs "
+                                f"from the configured Microsoft Graph endpoint {self.allowed_netloc!r}."
+                            )
+                        url = next_url
                 else:
                     break
 

providers/microsoft/azure/tests/unit/microsoft/azure/hooks/test_msgraph.py

@@ -442,6 +442,25 @@ async def test_paginated_run(self):
             assert isinstance(actual, list)
             assert actual == [users, next_users]
 
+    @pytest.mark.asyncio
+    async def test_paginated_run_refuses_cross_host_next_link(self):
+        first_page = {
+            "@odata.nextLink": "https://attacker.example/v1.0/users?$skiptoken=steal",
+            "value": [{"id": "1"}],
+        }
+        response = mock_json_response(200, first_page)
+
+        with patch_hook_and_request_adapter(response) as mocks:
+            mock_get_http_response = mocks[-1]
+            hook = KiotaRequestAdapterHook(conn_id="msgraph_api")
+
+            with pytest.raises(ValueError, match="attacker.example"):
+                await hook.paginated_run(url="users")
+
+            # The off-host pagination link is refused before it is fetched, so the bearer
+            # token is never sent to the attacker host.
+            assert mock_get_http_response.call_count == 1
+
     @pytest.mark.asyncio
     async def test_build_request_adapter_masks_secrets(self):
         """Test that sensitive data is masked when building request adapter."""
@@ -588,6 +607,7 @@ async def test_get_async_conn_rebuilds_adapter_when_http_client_is_closed(self):
 
             fresh_adapter = Mock(spec=HttpxRequestAdapter)
             fresh_adapter._http_client = Mock(is_closed=False)
+            fresh_adapter.base_url = "https://graph.microsoft.com/v1.0"
 
             with patch.object(hook, "_build_request_adapter", return_value=("v1.0", fresh_adapter)):
                 result = await hook.get_async_conn()
@@ -607,6 +627,7 @@ async def test_get_async_conn_rebuilds_adapter_when_credentials_session_is_close
 
             fresh_adapter = Mock(spec=HttpxRequestAdapter)
             fresh_adapter._http_client = Mock(is_closed=False)
+            fresh_adapter.base_url = "https://graph.microsoft.com/v1.0"
 
             with patch.object(hook, "_build_request_adapter", return_value=("v1.0", fresh_adapter)):
                 result = await hook.get_async_conn()
@@ -622,6 +643,7 @@ async def test_get_async_conn_does_not_rebuild_adapter_when_transport_never_open
             adapter = Mock(spec=HttpxRequestAdapter)
             adapter._http_client = Mock(spec=AsyncClient, is_closed=False)
             adapter._authentication_provider = mock_authentication_provider()
+            adapter.base_url = "https://graph.microsoft.com/v1.0"
             hook.cached_request_adapters[hook.conn_id] = (hook.api_version, adapter)
 
             result = await hook.get_async_conn()
@@ -637,6 +659,7 @@ async def test_send_request_invalidates_cache_and_raises_on_any_error(self):
             adapter = Mock(spec=HttpxRequestAdapter)
             adapter._http_client = Mock(spec=AsyncClient, is_closed=False)
             adapter._authentication_provider = mock_authentication_provider(closed=False)
+            adapter.base_url = "https://graph.microsoft.com/v1.0"
             adapter.send_no_response_content_async = AsyncMock(side_effect=RuntimeError("some error"))
             hook.cached_request_adapters[hook.conn_id] = (hook.api_version, adapter)