This is an automated email from the ASF dual-hosted git repository.
dabla pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 449e589da21 Restrict MSGraph deferrable pagination to the configured
host (#71842)
449e589da21 is described below
commit 449e589da21f776edf3f6d0fa00c6db0ee1ddd70
Author: PoAn Yang <[email protected]>
AuthorDate: Sat Aug 29 04:12:40 2026 +0900
Restrict MSGraph deferrable pagination to the configured host (#71842)
Signed-off-by: PoAn Yang <[email protected]>
---
.../providers/microsoft/azure/hooks/msgraph.py | 50 +++++++++++++++-------
.../providers/microsoft/azure/operators/msgraph.py | 1 +
.../providers/microsoft/azure/triggers/msgraph.py | 9 ++++
.../unit/microsoft/azure/hooks/test_msgraph.py | 27 ++++++++++++
.../unit/microsoft/azure/operators/test_msgraph.py | 41 ++++++++++++++++++
.../azure/tests/unit/microsoft/azure/test_utils.py | 8 +++-
.../unit/microsoft/azure/triggers/test_msgraph.py | 16 +++++++
7 files changed, 135 insertions(+), 17 deletions(-)
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 881a89fed07..db736169bc3 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,12 +491,7 @@ class KiotaRequestAdapterHook(BaseHook):
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
+ self.allowed_netloc = urlparse(request_adapter.base_url).netloc.lower()
return request_adapter
def get_proxies(self, config: dict) -> dict | None:
@@ -620,6 +615,39 @@ class KiotaRequestAdapterHook(BaseHook):
return response
+ async def get_allowed_netlocs(self) -> set[str]:
+ """Return the endpoint's host and the connection's allowed hosts, for
checking pagination links."""
+ request_adapter = await self.get_async_conn()
+ adapter = cast("HttpxRequestAdapter", request_adapter)
+ provider = cast("BaseBearerTokenAuthenticationProvider",
adapter._authentication_provider)
+ access_token_provider = cast("AzureIdentityAccessTokenProvider",
provider.access_token_provider)
+ allowed_hosts =
access_token_provider.get_allowed_hosts_validator().get_allowed_hosts()
+ netlocs = {host.lower() for host in allowed_hosts}
+
+ if self.allowed_netloc:
+ netlocs.add(self.allowed_netloc)
+ return netlocs
+
+ async def assert_allowed_host(self, url: str | None) -> None:
+ """
+ Refuse an absolute ``url`` whose host the connection does not allow.
+
+ 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
+
+ allowed_netlocs = await self.get_allowed_netlocs()
+
+ if urlparse(url).netloc.lower() not in allowed_netlocs:
+ raise ValueError(
+ f"Refusing to follow pagination link {url!r}: its host is not
among the allowed "
+ f"Microsoft Graph endpoints {sorted(allowed_netlocs)}."
+ )
+
async def paginated_run(
self,
url: str = "",
@@ -667,15 +695,7 @@ class KiotaRequestAdapterHook(BaseHook):
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 40782fdba5d..049c7004477 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
@@ -352,6 +352,7 @@ class MSGraphAsyncOperator(BaseOperator):
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 2ff0761b1a2..b6c32bb3f1d 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 @@ class MSGraphTrigger(BaseTrigger):
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 @@ class MSGraphTrigger(BaseTrigger):
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 @@ class MSGraphTrigger(BaseTrigger):
"headers": self.headers,
"data": self.data,
"response_type": self.response_type,
+ "pagination_link": self.pagination_link,
},
)
@@ -182,6 +188,9 @@ class MSGraphTrigger(BaseTrigger):
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/hooks/test_msgraph.py
b/providers/microsoft/azure/tests/unit/microsoft/azure/hooks/test_msgraph.py
index d41407d4c03..b505055ace4 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
@@ -461,6 +461,33 @@ class TestKiotaRequestAdapterHook:
# token is never sent to the attacker host.
assert mock_get_http_response.call_count == 1
+ @pytest.mark.asyncio
+ async def test_assert_allowed_host_refuses_another_host(self):
+ with patch_hook_and_request_adapter(mock_json_response(200, {})):
+ hook = KiotaRequestAdapterHook(conn_id="msgraph_api")
+
+ with pytest.raises(ValueError, match="attacker.example"):
+ await
hook.assert_allowed_host("https://attacker.example/v1.0/users";)
+
+ @pytest.mark.asyncio
+ async def test_assert_allowed_host_accepts_a_relative_url(self):
+ with patch_hook_and_request_adapter(mock_json_response(200, {})):
+ hook = KiotaRequestAdapterHook(conn_id="msgraph_api")
+
+ await hook.assert_allowed_host("users?$skip=100")
+
+ @pytest.mark.asyncio
+ async def
test_assert_allowed_host_accepts_a_host_listed_in_the_connection(self):
+ with patch_hook_and_request_adapter(
+ mock_json_response(200, {}),
+ side_effect=lambda conn_id: get_airflow_connection(
+ conn_id, allowed_hosts="graph.microsoft.com,other.example"
+ ),
+ ):
+ hook = KiotaRequestAdapterHook(conn_id="msgraph_api")
+
+ await hook.assert_allowed_host("https://other.example/v1.0/users";)
+
@pytest.mark.asyncio
async def test_build_request_adapter_masks_secrets(self):
"""Test that sensitive data is masked when building request adapter."""
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 cbb2c28f19d..cdb1715b7e1 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
@@ -407,6 +407,47 @@ class TestMSGraphAsyncOperator:
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/test_utils.py
b/providers/microsoft/azure/tests/unit/microsoft/azure/test_utils.py
index bc8577ba9aa..7b8cf56e912 100644
--- a/providers/microsoft/azure/tests/unit/microsoft/azure/test_utils.py
+++ b/providers/microsoft/azure/tests/unit/microsoft/azure/test_utils.py
@@ -191,6 +191,7 @@ def get_airflow_connection(
api_version: APIVersion | str | None = APIVersion.v1.value,
authority: str | None = None,
disable_instance_discovery: bool = False,
+ allowed_hosts: str | None = None,
):
from airflow.models import Connection
@@ -203,6 +204,9 @@ def get_airflow_connection(
"disable_instance_discovery": disable_instance_discovery,
}
+ if allowed_hosts:
+ extra["allowed_hosts"] = allowed_hosts
+
if azure_tenant_id:
extra["tenantId"] = azure_tenant_id
else:
@@ -266,8 +270,8 @@ def patch_hook(side_effect: Callable =
get_airflow_connection):
@contextmanager
-def patch_hook_and_request_adapter(response):
- with patch_hook() as hook_mocks:
+def patch_hook_and_request_adapter(response, side_effect: Callable =
get_airflow_connection):
+ with patch_hook(side_effect=side_effect) as hook_mocks:
with patch.object(HttpxRequestAdapter, "get_http_response_message") as
mock_get_http_response:
if isinstance(response, Exception):
mock_get_http_response.side_effect = response
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 e225a147de9..7ea42c7df49 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
@@ -61,6 +61,21 @@ class TestMSGraphTrigger:
assert actual[0].payload["type"] == "builtins.dict"
assert actual[0].payload["response"] == json.dumps(users)
+ def test_run_refuses_a_cross_host_pagination_link(self):
+ with patch_hook_and_request_adapter(mock_json_response(200, {})) as
(*_, mock_get_http_response):
+ trigger = MSGraphTrigger(
+ "https://attacker.example/v1.0/users";,
+ conn_id="msgraph_api",
+ pagination_link=True,
+ )
+ actual = run_trigger(trigger)
+
+ assert mock_get_http_response.call_count == 0
+ assert len(actual) == 1
+ assert isinstance(actual[0], TriggerEvent)
+ assert actual[0].payload["status"] == "failure"
+ assert "attacker.example" in actual[0].payload["message"]
+
def test_run_when_response_is_none(self):
response = mock_json_response(200)
@@ -133,6 +148,7 @@ class TestMSGraphTrigger:
"scopes": [KiotaRequestAdapterHook.DEFAULT_SCOPE],
"api_version": APIVersion.v1.value,
"serializer":
f"{ResponseSerializer.__module__}.{ResponseSerializer.__name__}",
+ "pagination_link": False,
}
def test_get_conn(self):
