Verify Azure AD OAuth id_token signatures by default in FAB auth manager

docs/spelling_wordlist.txt

@@ -121,6 +121,7 @@ auditable
 Auth
 auth
 authenticator
+Authentik
 Authlib
 authMechanism
 authorised

providers/fab/docs/changelog.rst

@@ -20,6 +20,13 @@
 Changelog
 ---------
 
+.. note::
+    The Azure AD OAuth provider in the FAB auth manager now verifies the ``id_token``
+    signature by default: ``verify_signature`` now defaults to ``True`` (previously
+    ``False``), consistent with the Authentik provider. Deployments that intentionally
+    relied on skipping signature verification must set ``verify_signature: False``
+    explicitly in the Azure provider ``client_kwargs`` to keep the previous behaviour.
+
 3.7.2
 .....
 

providers/fab/src/airflow/providers/fab/auth_manager/security_manager/override.py

@@ -2424,7 +2424,7 @@ def _get_microsoft_jwks(self) -> list[dict[str, Any]]:
         return requests.get(MICROSOFT_KEY_SET_URL, timeout=30).json()
 
     def _decode_and_validate_azure_jwt(self, id_token: str) -> dict[str, str]:
-        verify_signature = self.oauth_remotes["azure"].client_kwargs.get("verify_signature", False)
+        verify_signature = self.oauth_remotes["azure"].client_kwargs.get("verify_signature", True)
         if verify_signature:
             from authlib.jose import JsonWebKey, jwt as authlib_jwt
 
@@ -2433,6 +2433,7 @@ def _decode_and_validate_azure_jwt(self, id_token: str) -> dict[str, str]:
             claims.validate()
             return claims
 
+        log.warning("JWT token is not validated!")
         _parts = id_token.split(".")
         _payload = _parts[1] + "=" * (-len(_parts[1]) % 4)
         return json.loads(base64.urlsafe_b64decode(_payload))

providers/fab/tests/unit/fab/auth_manager/security_manager/test_override.py

@@ -472,6 +472,38 @@ def test_get_oauth_user_info_azure_with_groups_config(self):
             assert user_info["email"] == "jane.smith@example.com"
             assert user_info["role_keys"] == ["admin-group", "viewer-group"]
 
+    def test_decode_and_validate_azure_jwt_verifies_signature_by_default(self):
+        """Azure AD id_token signatures are verified by default (verify_signature defaults to True)."""
+        sm = EmptySecurityManager()
+        # client_kwargs does not set verify_signature -> it must default to verifying
+        sm.oauth_remotes = {"azure": Mock(client_kwargs={})}
+
+        with mock.patch.object(
+            EmptySecurityManager, "_get_microsoft_jwks", side_effect=RuntimeError("verify-branch-reached")
+        ) as mock_jwks:
+            with pytest.raises(RuntimeError, match="verify-branch-reached"):
+                sm._decode_and_validate_azure_jwt("header.payload.signature")
+
+        # entering the verifying branch means the Microsoft JWKS were fetched
+        mock_jwks.assert_called_once()
+
+    def test_decode_and_validate_azure_jwt_skips_verification_when_opted_out(self):
+        """With verify_signature explicitly False, the token is decoded without signature verification."""
+        import base64
+        import json as _json
+
+        payload = base64.urlsafe_b64encode(_json.dumps({"oid": "user-1"}).encode()).decode().rstrip("=")
+        id_token = f"header.{payload}.signature"
+
+        sm = EmptySecurityManager()
+        sm.oauth_remotes = {"azure": Mock(client_kwargs={"verify_signature": False})}
+
+        with mock.patch.object(EmptySecurityManager, "_get_microsoft_jwks") as mock_jwks:
+            result = sm._decode_and_validate_azure_jwt(id_token)
+
+        mock_jwks.assert_not_called()
+        assert result == {"oid": "user-1"}
+
 
 def test_ldap_search_escapes_username_and_validates_filter():
     """Test that LDAP search properly escapes username and validates search filter."""