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 4952531dcc6fd..881a89fed07ee 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 @@ -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 diff --git a/providers/microsoft/azure/tests/unit/microsoft/azure/hooks/test_msgraph.py b/providers/microsoft/azure/tests/unit/microsoft/azure/hooks/test_msgraph.py index 8eb60e91eed69..d41407d4c0359 100644 --- a/providers/microsoft/azure/tests/unit/microsoft/azure/hooks/test_msgraph.py +++ b/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)