diff --git a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/msgraph.py b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/msgraph.py index 881a89fed07ee..230432caa51e9 100644 --- a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/msgraph.py +++ b/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 diff --git a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/msgraph.py b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/msgraph.py index 7cd68b6470691..bc8d8c7540730 100644 --- a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/msgraph.py +++ b/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, ) diff --git a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/triggers/msgraph.py b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/triggers/msgraph.py index 2ff0761b1a235..b6c32bb3f1d0a 100644 --- a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/triggers/msgraph.py +++ b/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, diff --git a/providers/microsoft/azure/tests/unit/microsoft/azure/operators/test_msgraph.py b/providers/microsoft/azure/tests/unit/microsoft/azure/operators/test_msgraph.py index b722bc4a617c7..f6f9c05c0d0c7 100644 --- a/providers/microsoft/azure/tests/unit/microsoft/azure/operators/test_msgraph.py +++ b/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, diff --git a/providers/microsoft/azure/tests/unit/microsoft/azure/triggers/test_msgraph.py b/providers/microsoft/azure/tests/unit/microsoft/azure/triggers/test_msgraph.py index e225a147de9f5..cea2d4f5b0b3d 100644 --- a/providers/microsoft/azure/tests/unit/microsoft/azure/triggers/test_msgraph.py +++ b/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):