This is an automated email from the ASF dual-hosted git repository.
hussein-awala 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 d4aa2093b80 Support Azure national clouds in FAB Azure AD id_token
validation (#72010)
d4aa2093b80 is described below
commit d4aa2093b80b5ce3007ab1a545ac687de86a65dc
Author: Hussein Awala <[email protected]>
AuthorDate: Mon Aug 24 23:26:19 2026 +0200
Support Azure national clouds in FAB Azure AD id_token validation (#72010)
---
docs/spelling_wordlist.txt | 1 +
providers/fab/docs/auth-manager/sso.rst | 13 ++
.../fab/auth_manager/security_manager/override.py | 183 +++++++++++++++++++--
.../auth_manager/security_manager/test_override.py | 181 ++++++++++++++++++++
4 files changed, 365 insertions(+), 13 deletions(-)
diff --git a/docs/spelling_wordlist.txt b/docs/spelling_wordlist.txt
index d67f839976b..c46acc096fb 100644
--- a/docs/spelling_wordlist.txt
+++ b/docs/spelling_wordlist.txt
@@ -1897,6 +1897,7 @@ Vertica
vertica
Vespa
vespa
+Vianet
videointelligence
views
virtualenv
diff --git a/providers/fab/docs/auth-manager/sso.rst
b/providers/fab/docs/auth-manager/sso.rst
index 05b0e3e5e77..61fda6e0b19 100644
--- a/providers/fab/docs/auth-manager/sso.rst
+++ b/providers/fab/docs/auth-manager/sso.rst
@@ -189,6 +189,19 @@ Provider Examples
authorities (``common``, ``organizations``, ``consumers``) are not
accepted; configure a
specific tenant GUID or domain instead.
+.. note::
+ National clouds are supported by pointing the endpoints at their authority
host:
+ ``login.microsoftonline.us`` (Azure Government) or
``login.partner.microsoftonline.cn``
+ (Azure operated by 21Vianet). The issuer and the signing key set are then
read from that
+ tenant's own OpenID metadata, so no extra configuration is needed. Azure AD
B2C
+ (``<tenant>.b2clogin.com``) is not supported, because its metadata is
addressed by policy
+ rather than by tenant alone.
+
+ Configuring a tenant domain, or any national-cloud tenant, makes an
outbound HTTPS request
+ to the authority's OpenID discovery endpoint the first time a user logs in.
The result is
+ cached for the lifetime of the process. Deployments with restricted egress
must allow that
+ host.
+
.. seealso::
For Azure app registration and OAuth setup, see
:doc:`apache-airflow-providers-microsoft-azure:connections/azure`
and the `Azure OAuth2 documentation
<https://learn.microsoft.com/en-us/azure/active-directory/develop/v2-oauth2-auth-code-flow>`_.
diff --git
a/providers/fab/src/airflow/providers/fab/auth_manager/security_manager/override.py
b/providers/fab/src/airflow/providers/fab/auth_manager/security_manager/override.py
index ed8d3f9c62f..7e9ced387cf 100644
---
a/providers/fab/src/airflow/providers/fab/auth_manager/security_manager/override.py
+++
b/providers/fab/src/airflow/providers/fab/auth_manager/security_manager/override.py
@@ -167,6 +167,21 @@ class AzureTenantResolutionError(FabException):
"""Raised when an Azure AD tenant identifier cannot be resolved to a
canonical GUID."""
+# Microsoft identity platform authority hosts. Each national cloud mints its
own issuer and
+# serves its own signing keys, so the host a deployment is configured against
decides which
+# issuer to pin and which key set to verify against. Azure AD B2C is
deliberately absent: its
+# authority is per-tenant (``<tenant>.b2clogin.com``) and its metadata is
addressed by policy,
+# which does not fit the tenant-only lookup below.
+AZURE_COMMERCIAL_AUTHORITY_HOST = "login.microsoftonline.com"
+AZURE_AUTHORITY_HOSTS = frozenset(
+ {
+ AZURE_COMMERCIAL_AUTHORITY_HOST,
+ "login.microsoftonline.us", # Azure Government
+ "login.partner.microsoftonline.cn", # Azure operated by 21Vianet
(China)
+ }
+)
+
+
class FabAirflowSecurityManagerOverride(AirflowSecurityManagerV2):
"""
This security manager overrides the default AirflowSecurityManager
security manager.
@@ -390,6 +405,7 @@ class
FabAirflowSecurityManagerOverride(AirflowSecurityManagerV2):
self.appbuilder = appbuilder
self._azure_tenant_guid_cache: dict[str, str] = {}
+ self._azure_tenant_metadata_cache: dict[tuple[str, str],
tuple[tuple[str, ...], str]] = {}
self._init_config()
self._init_auth()
self._init_data_model()
@@ -2426,15 +2442,46 @@ class
FabAirflowSecurityManagerOverride(AirflowSecurityManagerV2):
if conf.get("fab", "SESSION_BACKEND") == "database":
session.sid = str(uuid.uuid4()) # type: ignore
- def _get_microsoft_jwks(self) -> list[dict[str, Any]]:
- return requests.get(MICROSOFT_KEY_SET_URL, timeout=30).json()
+ def _get_microsoft_jwks(self, jwks_uri: str | None = None) ->
list[dict[str, Any]]:
+ """
+ Fetch the Microsoft signing key set.
+
+ ``jwks_uri`` is supplied for national clouds, which serve their own
keys; the
+ commercial cloud keeps using the key set URL from Flask-AppBuilder so
existing
+ deployments and overrides of this method are unaffected.
+ """
+ return requests.get(jwks_uri or MICROSOFT_KEY_SET_URL,
timeout=30).json()
+
+ def _get_azure_authority_host(self) -> str:
+ """
+ Return the Microsoft identity platform host the deployment is
configured against.
+
+ Falls back to the commercial cloud, which is what Flask-AppBuilder's
Azure defaults
+ point at when no endpoint is configured explicitly.
+ """
+ azure = self.oauth_remotes["azure"]
+ for url in (
+ getattr(azure, "api_base_url", None),
+ getattr(azure, "access_token_url", None),
+ getattr(azure, "authorize_url", None),
+ ):
+ if not isinstance(url, str):
+ continue
+ parsed = urllib.parse.urlsplit(url)
+ if parsed.scheme.lower() != "https":
+ continue
+ host = (parsed.hostname or "").lower()
+ if host in AZURE_AUTHORITY_HOSTS:
+ return host
+ return AZURE_COMMERCIAL_AUTHORITY_HOST
def _get_azure_tenant_identifier(self) -> str | None:
"""
Extract the configured Azure AD tenant identifier.
Prefers an explicit ``tenant_id`` in ``client_kwargs``; otherwise
derives it from
- the tenant path segment of configured Azure HTTPS endpoints on
``login.microsoftonline.com``.
+ the tenant path segment of configured Azure HTTPS endpoints on any
Microsoft identity
+ platform authority host (see ``AZURE_AUTHORITY_HOSTS``).
Returns ``None`` when an explicit ``tenant_id`` is tenant-agnostic or
when no
tenant-specific identifier can be derived from the configured
endpoints. Tenant-agnostic
@@ -2458,7 +2505,7 @@ class
FabAirflowSecurityManagerOverride(AirflowSecurityManagerV2):
parsed = urllib.parse.urlsplit(url)
if parsed.scheme.lower() != "https":
continue
- if (parsed.hostname or "").lower() !=
"login.microsoftonline.com":
+ if (parsed.hostname or "").lower() not in
AZURE_AUTHORITY_HOSTS:
continue
path_parts = [segment for segment in parsed.path.split("/") if
segment]
if path_parts:
@@ -2544,6 +2591,108 @@ class
FabAirflowSecurityManagerOverride(AirflowSecurityManagerV2):
self._azure_tenant_guid_cache[tenant_identifier] = canonical_guid
return canonical_guid
+ def _fetch_azure_openid_configuration(self, discovery_url: str,
tenant_identifier: str) -> dict[str, Any]:
+ """Fetch and sanity-check an OpenID discovery document, failing closed
on any problem."""
+ try:
+ resp = requests.get(discovery_url, timeout=5,
allow_redirects=False)
+ except requests.exceptions.RequestException as ex:
+ raise AzureTenantResolutionError(
+ f"Failed to resolve Azure tenant identifier
'{tenant_identifier}' via OpenID discovery."
+ ) from ex
+
+ if resp.status_code != 200:
+ raise AzureTenantResolutionError(
+ f"Failed to resolve Azure tenant identifier
'{tenant_identifier}': "
+ f"OpenID discovery endpoint returned HTTP {resp.status_code}."
+ )
+
+ try:
+ data = resp.json()
+ except ValueError as ex:
+ raise AzureTenantResolutionError(
+ f"Failed to resolve Azure tenant identifier
'{tenant_identifier}' via OpenID discovery."
+ ) from ex
+
+ if not isinstance(data, dict):
+ raise AzureTenantResolutionError(
+ f"Failed to resolve Azure tenant identifier
'{tenant_identifier}': "
+ "OpenID discovery response is not a JSON object."
+ )
+ return data
+
+ def _resolve_azure_tenant_metadata(
+ self, tenant_identifier: str, authority_host: str
+ ) -> tuple[tuple[str, ...], str]:
+ """
+ Resolve the acceptable issuers and the signing key set URL for a
national cloud tenant.
+
+ National clouds mint their own issuers and serve their own keys, and
the values are not
+ derivable from the commercial ones, so they are read from the tenant's
own OpenID
+ discovery metadata rather than assembled from a template. Both the
v2.0 and v1.0
+ documents are consulted because either issuer form may appear
depending on which
+ endpoints the deployment uses.
+
+ The issuer and ``jwks_uri`` are required to live on Microsoft-operated
hosts for the
+ configured cloud, so a tampered metadata document cannot redirect
verification
+ elsewhere. Results are cached per (authority host, tenant); failures
are not cached.
+ """
+ cache_key = (authority_host, tenant_identifier)
+ if cache_key in self._azure_tenant_metadata_cache:
+ return self._azure_tenant_metadata_cache[cache_key]
+
+ encoded_tenant = urllib.parse.quote(tenant_identifier, safe="")
+ base = f"https://{authority_host}/{encoded_tenant}";
+
+ v2 = self._fetch_azure_openid_configuration(
+ f"{base}/v2.0/.well-known/openid-configuration", tenant_identifier
+ )
+ issuer = v2.get("issuer")
+ jwks_uri = v2.get("jwks_uri")
+ if not isinstance(issuer, str) or not isinstance(jwks_uri, str):
+ raise AzureTenantResolutionError(
+ f"Failed to resolve Azure tenant identifier
'{tenant_identifier}': "
+ "OpenID discovery response missing 'issuer' or 'jwks_uri'."
+ )
+ if (urllib.parse.urlsplit(issuer).hostname or "").lower() !=
authority_host:
+ raise AzureTenantResolutionError(
+ f"Failed to resolve Azure tenant identifier
'{tenant_identifier}': "
+ f"OpenID discovery returned issuer '{issuer}' outside the
configured authority "
+ f"'{authority_host}'."
+ )
+ jwks_host = (urllib.parse.urlsplit(jwks_uri).hostname or "").lower()
+ if urllib.parse.urlsplit(jwks_uri).scheme.lower() != "https" or
jwks_host != authority_host:
+ raise AzureTenantResolutionError(
+ f"Failed to resolve Azure tenant identifier
'{tenant_identifier}': "
+ f"OpenID discovery returned jwks_uri '{jwks_uri}' outside the
configured authority "
+ f"'{authority_host}'."
+ )
+
+ issuers = [issuer]
+
+ # The v1.0 document carries the sts.* style issuer. It is fetched
separately because a
+ # deployment configured against v1.0 endpoints receives tokens with
that issuer. If it
+ # cannot be read, the v1.0 form is simply not accepted -- that denies
a login rather
+ # than widening what is trusted.
+ try:
+ v1 = self._fetch_azure_openid_configuration(
+ f"{base}/.well-known/openid-configuration", tenant_identifier
+ )
+ except AzureTenantResolutionError:
+ log.warning(
+ "Could not read the v1.0 OpenID metadata for Azure tenant %s
on %s; "
+ "only the v2.0 issuer will be accepted.",
+ tenant_identifier,
+ authority_host,
+ )
+ else:
+ v1_issuer = v1.get("issuer")
+ if isinstance(v1_issuer, str) and v1_issuer not in issuers:
+ issuers.append(v1_issuer)
+
+ resolved = (tuple(issuers), jwks_uri)
+ self._azure_tenant_metadata_cache[cache_key] = resolved
+ return resolved
+
def _decode_and_validate_azure_jwt(self, id_token: str) -> dict[str, str]:
verify_signature =
self.oauth_remotes["azure"].client_kwargs.get("verify_signature", True)
if verify_signature:
@@ -2560,24 +2709,32 @@ class
FabAirflowSecurityManagerOverride(AirflowSecurityManagerV2):
"in the azure provider's client_kwargs."
)
- tenant_guid = self._resolve_azure_tenant_guid(tenant_identifier)
+ authority_host = self._get_azure_authority_host()
+ jwks_uri: str | None = None
+ if authority_host == AZURE_COMMERCIAL_AUTHORITY_HOST:
+ # The commercial cloud's issuer forms are stable and
documented, so they are
+ # assembled from the tenant GUID. A GUID resolves without any
network call,
+ # which keeps the common deployment off the discovery endpoint
at login time.
+ tenant_guid =
self._resolve_azure_tenant_guid(tenant_identifier)
+ issuers: tuple[str, ...] = (
+ f"https://login.microsoftonline.com/{tenant_guid}/v2.0";,
+ f"https://sts.windows.net/{tenant_guid}/";,
+ )
+ else:
+ # National clouds use different issuer hosts and serve their
own signing keys,
+ # so both are read from the tenant's own metadata instead of
being guessed.
+ issuers, jwks_uri =
self._resolve_azure_tenant_metadata(tenant_identifier, authority_host)
claims_options = {
# The token must have been issued by the configured tenant.
Both the v1.0 and
# v2.0 issuer forms are accepted because either may be
returned depending on
# which endpoints the deployment is configured against.
- "iss": {
- "essential": True,
- "values": [
-
f"https://login.microsoftonline.com/{tenant_guid}/v2.0";,
- f"https://sts.windows.net/{tenant_guid}/";,
- ],
- },
+ "iss": {"essential": True, "values": list(issuers)},
# The token must have been minted for this application.
"aud": {"essential": True, "value":
self.oauth_remotes["azure"].client_id},
}
- keyset = JsonWebKey.import_key_set(self._get_microsoft_jwks()) #
type: ignore
+ keyset =
JsonWebKey.import_key_set(self._get_microsoft_jwks(jwks_uri)) # type: ignore
claims = authlib_jwt.decode(id_token, keyset,
claims_options=claims_options)
claims.validate()
return claims
diff --git
a/providers/fab/tests/unit/fab/auth_manager/security_manager/test_override.py
b/providers/fab/tests/unit/fab/auth_manager/security_manager/test_override.py
index de29ea3ccf0..68f648edddc 100644
---
a/providers/fab/tests/unit/fab/auth_manager/security_manager/test_override.py
+++
b/providers/fab/tests/unit/fab/auth_manager/security_manager/test_override.py
@@ -76,6 +76,7 @@ class EmptySecurityManager(FabAirflowSecurityManagerOverride):
# super() not called on purpose to avoid the whole chain of init calls
def __init__(self):
self._azure_tenant_guid_cache = {}
+ self._azure_tenant_metadata_cache = {}
class TestFabAirflowSecurityManagerOverride:
@@ -826,6 +827,186 @@ class TestFabAirflowSecurityManagerOverride:
assert claims["iss"] == f"https://sts.windows.net/{TENANT_GUID}/";
+ @pytest.mark.parametrize(
+ "authority_host",
+ [
+ pytest.param("login.microsoftonline.us", id="us-government"),
+ pytest.param("login.partner.microsoftonline.cn",
id="china-21vianet"),
+ ],
+ )
+ def
test_decode_and_validate_azure_jwt_national_cloud_uses_tenant_metadata(self,
authority_host):
+ """National clouds pin the issuer and key set that their own metadata
advertises."""
+ key = JsonWebKey.generate_key("RSA", 2048, options={"kid":
"test-kid"}, is_private=True)
+ public_key = key.as_dict(is_private=False, kid="test-kid")
+ issuer = f"https://{authority_host}/{TENANT_GUID}/v2.0";
+ jwks_uri =
f"https://{authority_host}/{TENANT_GUID}/discovery/v2.0/keys";
+ id_token = _create_azure_jwt(key=key, iss=issuer)
+
+ sm = EmptySecurityManager()
+ sm.oauth_remotes = {
+ "azure": SimpleNamespace(
+ client_kwargs={},
+ client_id=CLIENT_ID,
+
api_base_url=f"https://{authority_host}/{TENANT_GUID}/oauth2/v2.0/";,
+ access_token_url=None,
+ authorize_url=None,
+ )
+ }
+
+ v2 = _create_mock_response(json_data={"issuer": issuer, "jwks_uri":
jwks_uri})
+ v1 = _create_mock_response(json_data={"issuer":
f"https://sts.{authority_host}/{TENANT_GUID}/"})
+
+ with (
+ mock.patch.object(
+ EmptySecurityManager,
+ "_get_microsoft_jwks",
+ autospec=True,
+ return_value={"keys": [public_key]},
+ ) as mock_jwks,
+ mock.patch("requests.get", autospec=True, side_effect=[v2, v1]),
+ ):
+ claims = sm._decode_and_validate_azure_jwt(id_token)
+
+ assert claims["iss"] == issuer
+ # the key set comes from the tenant's own metadata, not the commercial
cloud
+ assert mock_jwks.call_args.args[1] == jwks_uri
+
+ def
test_decode_and_validate_azure_jwt_national_cloud_rejects_commercial_issuer(self):
+ """A token minted by the commercial cloud is not accepted for a
national cloud tenant."""
+ key = JsonWebKey.generate_key("RSA", 2048, options={"kid":
"test-kid"}, is_private=True)
+ public_key = key.as_dict(is_private=False, kid="test-kid")
+ authority_host = "login.microsoftonline.us"
+ id_token = _create_azure_jwt(key=key,
iss=f"https://login.microsoftonline.com/{TENANT_GUID}/v2.0";)
+
+ sm = EmptySecurityManager()
+ sm.oauth_remotes = {
+ "azure": SimpleNamespace(
+ client_kwargs={},
+ client_id=CLIENT_ID,
+
api_base_url=f"https://{authority_host}/{TENANT_GUID}/oauth2/v2.0/";,
+ access_token_url=None,
+ authorize_url=None,
+ )
+ }
+
+ v2 = _create_mock_response(
+ json_data={
+ "issuer": f"https://{authority_host}/{TENANT_GUID}/v2.0";,
+ "jwks_uri":
f"https://{authority_host}/{TENANT_GUID}/discovery/v2.0/keys";,
+ }
+ )
+ v1 = _create_mock_response(json_data={"issuer":
f"https://sts.{authority_host}/{TENANT_GUID}/"})
+
+ with (
+ mock.patch.object(
+ EmptySecurityManager,
+ "_get_microsoft_jwks",
+ autospec=True,
+ return_value={"keys": [public_key]},
+ ),
+ mock.patch("requests.get", autospec=True, side_effect=[v2, v1]),
+ ):
+ with pytest.raises(InvalidClaimError, match="invalid_claim:
Invalid claim 'iss'"):
+ sm._decode_and_validate_azure_jwt(id_token)
+
+ @pytest.mark.parametrize(
+ ("metadata", "expected_message"),
+ [
+ pytest.param(
+ {
+ "issuer": f"https://evil.example.com/{TENANT_GUID}/v2.0";,
+ "jwks_uri":
f"https://login.microsoftonline.us/{TENANT_GUID}/discovery/v2.0/keys";,
+ },
+ "outside the configured authority",
+ id="issuer-off-authority",
+ ),
+ pytest.param(
+ {
+ "issuer":
f"https://login.microsoftonline.us/{TENANT_GUID}/v2.0";,
+ "jwks_uri":
f"https://evil.example.com/{TENANT_GUID}/discovery/v2.0/keys";,
+ },
+ "outside the configured authority",
+ id="jwks-off-authority",
+ ),
+ pytest.param(
+ {"issuer":
f"https://login.microsoftonline.us/{TENANT_GUID}/v2.0"},
+ "missing 'issuer' or 'jwks_uri'",
+ id="missing-jwks-uri",
+ ),
+ ],
+ )
+ def test_resolve_azure_tenant_metadata_fails_closed(self, metadata,
expected_message):
+ """Metadata that points verification off the configured authority is
refused."""
+ sm = EmptySecurityManager()
+ with mock.patch(
+ "requests.get", autospec=True,
return_value=_create_mock_response(json_data=metadata)
+ ):
+ with pytest.raises(AzureTenantResolutionError,
match=expected_message):
+ sm._resolve_azure_tenant_metadata(TENANT_GUID,
"login.microsoftonline.us")
+
+ def test_resolve_azure_tenant_metadata_tolerates_missing_v1_document(self):
+ """A v1.0 document that cannot be read narrows the accepted issuers
rather than widening them."""
+ authority_host = "login.microsoftonline.us"
+ issuer = f"https://{authority_host}/{TENANT_GUID}/v2.0";
+ jwks_uri =
f"https://{authority_host}/{TENANT_GUID}/discovery/v2.0/keys";
+
+ sm = EmptySecurityManager()
+ v2 = _create_mock_response(json_data={"issuer": issuer, "jwks_uri":
jwks_uri})
+ v1 = _create_mock_response(status_code=500)
+
+ with mock.patch("requests.get", autospec=True, side_effect=[v2, v1]):
+ issuers, resolved_jwks =
sm._resolve_azure_tenant_metadata(TENANT_GUID, authority_host)
+
+ assert issuers == (issuer,)
+ assert resolved_jwks == jwks_uri
+
+ def test_resolve_azure_tenant_metadata_caches_success(self):
+ """A resolved tenant is not re-fetched on the next login."""
+ authority_host = "login.microsoftonline.us"
+ issuer = f"https://{authority_host}/{TENANT_GUID}/v2.0";
+ jwks_uri =
f"https://{authority_host}/{TENANT_GUID}/discovery/v2.0/keys";
+
+ sm = EmptySecurityManager()
+ v2 = _create_mock_response(json_data={"issuer": issuer, "jwks_uri":
jwks_uri})
+ v1 = _create_mock_response(json_data={"issuer":
f"https://sts.{authority_host}/{TENANT_GUID}/"})
+
+ with mock.patch("requests.get", autospec=True, side_effect=[v2, v1])
as mock_get:
+ first = sm._resolve_azure_tenant_metadata(TENANT_GUID,
authority_host)
+ second = sm._resolve_azure_tenant_metadata(TENANT_GUID,
authority_host)
+
+ assert first == second
+ assert mock_get.call_count == 2
+
+ @pytest.mark.parametrize(
+ ("api_base_url", "expected"),
+ [
+ pytest.param(None, "login.microsoftonline.com",
id="defaults-to-commercial"),
+ pytest.param(
+ f"https://login.microsoftonline.us/{TENANT_GUID}/oauth2/v2.0/";,
+ "login.microsoftonline.us",
+ id="us-government",
+ ),
+ pytest.param(
+ f"https://login.microsoftonline.de/{TENANT_GUID}/oauth2/v2.0/";,
+ "login.microsoftonline.com",
+ id="unknown-host-falls-back-to-commercial",
+ ),
+ ],
+ )
+ def test_get_azure_authority_host(self, api_base_url, expected):
+ """The authority host is taken from the configured endpoints,
defaulting to commercial."""
+ sm = EmptySecurityManager()
+ sm.oauth_remotes = {
+ "azure": SimpleNamespace(
+ client_kwargs={},
+ client_id=CLIENT_ID,
+ api_base_url=api_base_url,
+ access_token_url=None,
+ authorize_url=None,
+ )
+ }
+ assert sm._get_azure_authority_host() == expected
+
def test_decode_and_validate_azure_jwt_rejects_issuer_mismatch(self):
"""Tokens issued for a different tenant are rejected with
InvalidClaimError."""
key = JsonWebKey.generate_key("RSA", 2048, options={"kid":
"test-kid"}, is_private=True)
