Restrict MSGraph deferrable pagination to the configured host

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

@@ -491,11 +491,6 @@ 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
 
@@ -620,6 +615,27 @@ async def run(
 
         return response
 
+    async def assert_allowed_host(self, url: str | None) -> None:
+        """
+        Refuse an absolute ``url`` whose host differs from the configured Microsoft Graph endpoint.
+
+        A pagination link (e.g. ``@odata.nextLink``) is echoed from the API response and is re-fetched
+        with the connection's bearer token attached. That token is withheld only from hosts outside
+        ``allowed_hosts``, which defaults to empty (any host) unless configured, so a tampered response
+        could send it to an arbitrary host (CWE-918).
+        """
+        if not url or not url.startswith("http"):
+            return
+
+        if self.allowed_netloc is None:
+            await self.get_async_conn()
+
+        if urlparse(url).netloc != self.allowed_netloc:
+            raise ValueError(
+                f"Refusing to follow pagination link {url!r}: its host differs "
+                f"from the configured Microsoft Graph endpoint {self.allowed_netloc!r}."
+            )
+
     async def paginated_run(
         self,
         url: str = "",
@@ -667,15 +683,7 @@ 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}."
-                            )
+                        await self.assert_allowed_host(next_url)
                         url = next_url
                 else:
                     break

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

@@ -351,6 +351,7 @@ def trigger_next_link(self, response, method_name: str, context: Context) -> Non
                         scopes=self.scopes,
                         api_version=self.api_version,
                         serializer=type(self.serializer),
+                        pagination_link=True,
                     ),
                     method_name=method_name,
                 )

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

@@ -95,6 +95,9 @@ class MSGraphTrigger(BaseTrigger):
         or you can pass a string as `v1.0` or `beta`.
     :param serializer: Class which handles response serialization (default is ResponseSerializer).
         Bytes will be base64 encoded into a string, so it can be stored as an XCom.
+    :param pagination_link: Whether `url` was taken from a pagination link of a previous response
+        (default is False). When True, its host is verified against the configured Microsoft Graph
+        endpoint before the request is made.
     """
 
     def __init__(
@@ -113,6 +116,7 @@ def __init__(
         scopes: str | list[str] | None = None,
         api_version: APIVersion | str | None = None,
         serializer: type[ResponseSerializer] = ResponseSerializer,
+        pagination_link: bool = False,
     ):
         super().__init__()
         self.conn_id = conn_id
@@ -129,6 +133,7 @@ def __init__(
         self.headers = headers
         self.data = data
         self.serializer: ResponseSerializer = self.resolve_type(serializer, default=ResponseSerializer)()
+        self.pagination_link = pagination_link
 
     @classmethod
     def resolve_type(cls, value: str | type, default) -> type:
@@ -157,6 +162,7 @@ def serialize(self) -> tuple[str, dict[str, Any]]:
                 "headers": self.headers,
                 "data": self.data,
                 "response_type": self.response_type,
+                "pagination_link": self.pagination_link,
             },
         )
 
@@ -182,6 +188,9 @@ def hook(self) -> KiotaRequestAdapterHook:
     async def run(self) -> AsyncIterator[TriggerEvent]:
         """Make a series of asynchronous HTTP calls via a KiotaRequestAdapterHook."""
         try:
+            if self.pagination_link:
+                await self.hook.assert_allowed_host(self.url)
+
             response = await self.hook.run(
                 url=self.url,
                 response_type=self.response_type,

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

@@ -369,6 +369,47 @@ def test_pagination_issues_every_page_with_the_configured_request(self):
             assert request.headers.try_get("ConsistencyLevel") == {"eventual"}
             assert request.content == json.dumps(data).encode("utf-8")
 
+    def test_pagination_refuses_cross_host_next_link(self):
+        first_page = {
+            "@odata.nextLink": "https://attacker.example/v1.0/users?$skiptoken=steal",
+            "value": [{"id": "1"}],
+        }
+        second_page = {"value": [{"id": "2"}]}
+        response = mock_json_response(200, first_page, second_page)
+
+        with patch_hook_and_request_adapter(response) as (*_, mock_get_http_response):
+            operator = MSGraphAsyncOperator(
+                task_id="users_delta",
+                conn_id="msgraph_api",
+                url="users",
+            )
+
+            with pytest.raises(AirflowException, match="attacker.example"):
+                execute_operator(operator)
+
+        # assert_allowed_host rejects the link before the request goes out, so the second page is never
+        # fetched and the bearer token does not reach attacker.example.
+        assert mock_get_http_response.call_count == 1
+
+    def test_relative_pagination_link_is_not_treated_as_cross_host(self):
+        pages = [{"next": "users?$skip=1", "value": [{"id": "1"}]}, {"value": [{"id": "2"}]}]
+        response = mock_json_response(200, *pages)
+
+        with patch_hook_and_request_adapter(response) as (*_, mock_get_http_response):
+            operator = MSGraphAsyncOperator(
+                task_id="users",
+                conn_id="msgraph_api",
+                url="users",
+                pagination_function=lambda operator, response, **context: (response.get("next"), None),
+            )
+
+            results, _ = execute_operator(operator)
+
+        # A pagination function may return a relative url, whose netloc is empty and never matches the
+        # configured endpoint. The startswith("http") check in assert_allowed_host lets it pass.
+        assert mock_get_http_response.call_count == 2
+        assert results == pages
+
     def test_execute_callable(self):
         with pytest.warns(
             AirflowProviderDeprecationWarning,

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

@@ -133,6 +133,7 @@ def test_serialize(self):
                 "scopes": [KiotaRequestAdapterHook.DEFAULT_SCOPE],
                 "api_version": APIVersion.v1.value,
                 "serializer": f"{ResponseSerializer.__module__}.{ResponseSerializer.__name__}",
+                "pagination_link": False,
             }
 
     def test_get_conn(self):