dabla commented on code in PR #71842:
URL: https://github.com/apache/airflow/pull/71842#discussion_r3838021349
##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/msgraph.py:
##########
@@ -620,6 +615,27 @@ async def run(
return response
+ async def assert_allowed_host(self, url: str | None) -> None:
Review Comment:
**[warning]** The allowlist is not configurable, which can cause failures
when an absolute URL passed to `paginated_run()` differs from the connection’s
configured `host`.
`assert_allowed_host` validates pagination links against
`self.allowed_netloc`, which is derived from the connection’s `base_url`. For
the common case — where SharePoint, Power BI, or Graph callers configure `host`
to the appropriate endpoint — pagination links will return the same host and
the check passes correctly.
However, if a caller passes an absolute `url` to `paginated_run()` or uses a
custom `pagination_function` that returns absolute URLs pointing to a host that
is not the connection’s `host`, those links will be rejected even if they are
legitimate. There is currently no way to opt in without changing the connection
configuration.
**Suggested improvement (no security regression):**
The reviewer has prototyped an `allowed_netlocs: list[str] | str | None`
parameter on `KiotaRequestAdapterHook.__init__` plus an `allowed_netloc`
connection-extra field that merge into a `set[str]`. The configured endpoint’s
host is always added automatically after `get_async_conn()`. This keeps the
default behaviour identical while letting callers explicitly opt in to
additional trusted hosts.
```python
# Hook constructor
def __init__(
self,
...,
allowed_netlocs: list[str] | str | None = None,
):
...
if isinstance(allowed_netlocs, str):
self.allowed_netlocs: set[str] = {n.strip() for n in
allowed_netlocs.split(",") if n.strip()}
else:
self.allowed_netlocs = set(allowed_netlocs) if allowed_netlocs else
set()
self.allowed_netloc: str | None = None # populated by get_async_conn()
# In get_async_conn(), after resolving the adapter:
self.allowed_netloc = urlparse(request_adapter.base_url).netloc
self.allowed_netlocs.add(self.allowed_netloc) # always includes configured
endpoint
# Synchronous check (no await needed once get_async_conn has run):
def assert_allowed_netloc(self, url: str | None, allowed_netlocs: set[str])
-> None:
if url and url.startswith("http") and urlparse(url).netloc not in
allowed_netlocs:
raise ValueError(
f"Refusing to follow pagination link {url!r}: its host differs "
f"from the configured Microsoft Graph endpoints
{','.join(sorted(allowed_netlocs))!r}."
)
```
The call in `paginated_run()` then passes the set explicitly:
```python
self.assert_allowed_netloc(next_url, self.allowed_netlocs)
```
And for the trigger path you’d carry `allowed_netlocs` through
`MSGraphTrigger.serialize()` the same way `pagination_link` is already carried.
Security note: the constructor parameter is developer-controlled code; the
connection-extra field requires Airflow admin access to edit — neither widens
the attack surface compared to the current single hardcoded-endpoint check.
---
Drafted-by: Claude Sonnet 4.6 (claude-sonnet-4.6); reviewed by @dabla before
posting
##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/msgraph.py:
##########
@@ -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
Review Comment:
**[nit]** `assert_allowed_host` needs to be `async` only because of the
`self.allowed_netloc is None` guard — which itself exists because the method
might be called before `get_async_conn()`.
In `paginated_run()` the connection is always established before
`assert_allowed_host` is reached (the very first `await self.run(...)` call
already calls `get_async_conn()`). In the trigger path,
`hook.assert_allowed_host(self.url)` is the first thing called in
`MSGraphTrigger.run()`, before any hook connection, so the guard fires there.
If you adopt the `allowed_netlocs` set approach from the other inline
comment, `self.allowed_netloc` is populated inside `get_async_conn()` and added
to `self.allowed_netlocs` atomically. The trigger can then call
`assert_allowed_netloc` synchronously **after** calling `await
self.hook.get_async_conn()` first, removing the need for the defensive guard
entirely:
```python
# MSGraphTrigger.run()
async def run(self) -> AsyncIterator[TriggerEvent]:
try:
if self.pagination_link:
await self.hook.get_async_conn() # ensures
allowed_netlocs is populated
self.hook.assert_allowed_netloc(self.url,
self.hook.allowed_netlocs)
response = await self.hook.run(...)
```
A synchronous check is also more straightforward to unit-test directly (no
`asyncio.run` boilerplate).
---
Drafted-by: Claude Sonnet 4.6 (claude-sonnet-4.6); reviewed by @dabla before
posting
##########
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)
Review Comment:
**[nit]** The test correctly expects `AirflowException`, but the reason
`ValueError` becomes `AirflowException` here is non-obvious:
`MSGraphTrigger.run()` catches all exceptions and wraps them in a
`TriggerEvent({"status": "failure", ...})`, and `execute_operator()` then
converts that failure event to `AirflowException`.
Consider adding a companion direct unit test on the hook to pin the method’s
own contract:
```python
@pytest.mark.asyncio
async def test_assert_allowed_host_rejects_cross_domain():
hook = KiotaRequestAdapterHook(conn_id="msgraph_api")
hook.allowed_netloc = "graph.microsoft.com"
with pytest.raises(ValueError, match="attacker.example"):
await hook.assert_allowed_host("https://attacker.example/steal";)
async def test_assert_allowed_host_passes_relative_url():
hook = KiotaRequestAdapterHook(conn_id="msgraph_api")
hook.allowed_netloc = "graph.microsoft.com"
await hook.assert_allowed_host("users?$skip=100") # must not raise
```
---
Drafted-by: Claude Sonnet 4.6 (claude-sonnet-4.6); reviewed by @dabla before
posting
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
