Restrict MSGraph deferrable pagination to the configured host

[FrankYang0529]

Why

• When MSGraphAsyncOperator fetches the next page, the url comes from the previous response. The Kiota adapter attaches the connection's access token to every request it sends. Nothing checks that url before the request goes out.

How

• Extract the netloc comparison that paginated_run() already does (added in Restrict MSGraph pagination nextLink to the configured host #69742) into KiotaRequestAdapterHook.assert_allowed_host(). The allowed host comes from the request adapter's base_url.
• Add pagination_link: bool = False to MSGraphTrigger, carry it through serialize(), and call the helper at the top of run() when it is set. MSGraphAsyncOperator.trigger_next_link() passes pagination_link=True, so the check applies only to urls that came back from a response.
• Relative urls still pass the check. Airflow's own $top/$skip branch and a custom pagination_function can both produce one, so assert_allowed_host() returns early for any url that is not absolute.

Verification

• uv run --project providers/microsoft/azure pytest providers/microsoft/azure/tests/unit/microsoft/azure -k msgraph

---

Was generative AI tooling used to co-author this PR?

• Yes - Claude Code

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[dabla]

This PR addresses a genuine CWE-918 (SSRF) risk: when MSGraphAsyncOperator.trigger_next_link() fires a new MSGraphTrigger for the next page, the url comes directly from the previous API response's @odata.nextLink. Without a host check the Kiota adapter would attach the bearer token to whatever host the response contained.

The approach — adding pagination_link: bool to the trigger, propagating it through serialize(), and calling a centralised assert_allowed_host() before the request — is clean and correct for the trigger path.

Two points worth addressing before merge:

1. Configurability gap (inline comment on hooks/msgraph.py): the single-endpoint check is correct for normal use (SharePoint/Power BI callers configure host to match, so pagination links pass). The gap is when a caller passes an absolute url that differs from the connection’s host, or uses a custom pagination_function returning absolute URLs for a different target. Making allowed_netlocs an explicit opt-in set fixes this at no security cost. The reviewer has working local code; a ready-to-use snippet is in the inline comment.

2. Test isolation (inline comment on test_msgraph.py): test_pagination_refuses_cross_host_next_link validates the full end-to-end path but hides the exception-wrapping mechanics. A direct unit test of assert_allowed_host() on the hook would make the contract explicit and survive any future changes to the trigger’s exception handler.

The nit on hooks/msgraph.py is a minor refactoring suggestion (making the check synchronous) that follows naturally if point 1 is adopted.

---

Drafted-by: Claude Sonnet 4.6 (claude-sonnet-4.6); reviewed by @dabla before posting

[dabla]

I would prefer having an implementation looking like this. Instead of adding allowed_netlocs, I would keep the existing allowed_netloc + dynamically derive the allowed_hosts from the RequestAdapter. Advantage of re-using this is that everything is already setup in the connection and we don't have to do duplicate configuration.

So I was thinking something like this.

A method which resolves the allowed_hosts from the RequestAdapter:

    async def get_allowed_hosts(self) -> set[str]:
        request_adapter = await self.get_async_conn()
        return set(request_adapter._authentication_provider.access_token_provider.allowed_hosts)

Some casts will probably be needed by mypy for this one.

Then we adapt the extracted assert_allowed_host method (I really like this refactoring you did):

    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_hosts = await self.get_allowed_hosts() | self.allowed_netloc

        if urlparse(url).netloc.lower() not in allowed_hosts:
            raise ValueError(
                f"Refusing to follow pagination link {url!r}: its host is not among the allowed "
                f"Microsoft Graph endpoints {sorted(allowed_hosts)}."
            )

Something like this, let me know what you think about it.

[dabla]

I also opened an issue on the Microsoft Github repository for kiota regarding the allowed_hosts to add support for wildcards in the AllowedHostsValidator with corresponding PR which addresses it. Let's see what comes out of it.