fix: azure user info claims and JWT decode

CHANGELOG.rst

@@ -4,7 +4,7 @@ Flask-AppBuilder ChangeLog
 Improvements and Bug fixes on 4.3.7
 -----------------------------------
 
-- fix: fix: swagger missing nonce (#2116) [Daniel Vaz Gaspar]
+- fix: swagger missing nonce (#2116) [Daniel Vaz Gaspar]
 
 Improvements and Bug fixes on 4.3.6
 -----------------------------------

docs/security.rst

@@ -305,6 +305,8 @@ Specify a list of OAUTH_PROVIDERS in **config.py** that you want to allow for yo
                 "client_kwargs": {
                     "scope": "User.read name preferred_username email profile upn",
                     "resource": "AZURE_APPLICATION_ID",
+                    # Optionally enforce signature JWT verification
+                    "verify_signature": False
                 },
                 "request_token_url": None,
                 "access_token_url": "https://login.microsoftonline.com/AZURE_TENANT_ID/oauth2/token",
@@ -347,10 +349,13 @@ You can give FlaskAppBuilder roles based on Oauth groups::
 To customize the userinfo retrieval, you can create your own method like this::
 
     @appbuilder.sm.oauth_user_info_getter
-    def my_user_info_getter(sm, provider, response=None):
+    def my_user_info_getter(
+        sm: SecurityManager,
+        provider: str,
+        response: Dict[str, Any]
+    ) -> Dict[str, Any]:
         if provider == "okta":
             me = sm.oauth_remotes[provider].get("userinfo")
-            log.debug("User info from Okta: {0}".format(me.data))
             return {
                 "username": "okta_" + me.data.get("sub", ""),
                 "first_name": me.data.get("given_name", ""),
@@ -365,11 +370,9 @@ To customize the userinfo retrieval, you can create your own method like this::
                 "email": me.json().get("email"),
                 "first_name": me.json().get("given_name", ""),
                 "last_name": me.json().get("family_name", ""),
-                "id": me.json().get("sub", ""),
                 "role_keys": ["User"], # set AUTH_ROLES_SYNC_AT_LOGIN = False
             }
-        else:
-            return {}
+        return {}
 
 On Flask-AppBuilder 3.4.0 the login page has changed.
 

examples/oauth/config.py

@@ -72,17 +72,17 @@
         "remote_app": {
             "client_id": os.environ.get("AZURE_APPLICATION_ID"),
             "client_secret": os.environ.get("AZURE_SECRET"),
-            "api_base_url": "https://login.microsoftonline.com/{AZURE_TENANT_ID}/oauth2",
+            "api_base_url": f"https://login.microsoftonline.com/{os.environ.get('AZURE_TENANT_ID')}/oauth2",
             "client_kwargs": {
                 "scope": "User.read name preferred_username email profile upn",
                 "resource": os.environ.get("AZURE_APPLICATION_ID"),
             },
             "request_token_url": None,
             "access_token_url": f"https://login.microsoftonline.com/"
-            f"{os.environ.get('AZURE_APPLICATION_ID')}/"
+            f"{os.environ.get('AZURE_TENANT_ID')}/"
             "oauth2/token",
             "authorize_url": f"https://login.microsoftonline.com/"
-            f"{os.environ.get('AZURE_APPLICATION_ID')}/"
+            f"{os.environ.get('AZURE_TENANT_ID')}/"
             f"oauth2/authorize",
         },
     },

flask_appbuilder/exceptions.py

@@ -50,3 +50,11 @@ class ApplyFilterException(FABException):
     """When executing an apply filter a SQLAlchemy exception happens"""
 
     ...
+
+
+class OAuthProviderUnknown(FABException):
+    """
+    When an OAuth provider is not supported/unknown
+    """
+
+    ...

flask_appbuilder/security/manager.py

@@ -1,18 +1,17 @@
 import datetime
-import json
 import logging
 import re
-from typing import Any, Dict, List, Optional, Set, Tuple, Union
+from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union
 
-from authlib.jose import JsonWebKey, jwt
 from flask import Flask, g, session, url_for
+from flask_appbuilder.exceptions import OAuthProviderUnknown
 from flask_babel import lazy_gettext as _
 from flask_jwt_extended import current_user as current_user_jwt
 from flask_jwt_extended import JWTManager
 from flask_limiter import Limiter
 from flask_limiter.util import get_remote_address
 from flask_login import current_user, LoginManager
-import requests
+import jwt
 from werkzeug.security import check_password_hash, generate_password_hash
 
 from .api import SecurityApi
@@ -271,7 +270,7 @@
             from authlib.integrations.flask_client import OAuth
 
             self.oauth = OAuth(app)
-            self.oauth_remotes = dict()
+            self.oauth_remotes = {}
             for _provider in self.oauth_providers:
                 provider_name = _provider["name"]
                 log.debug("OAuth providers init %s", provider_name)
@@ -519,7 +518,10 @@
         elif current_user_jwt:
             return current_user_jwt
 
-    def oauth_user_info_getter(self, f):
+    def oauth_user_info_getter(
+        self,
+        func: Callable[["BaseSecurityManager", str, Dict[str, Any]], Dict[str, Any]],
+    ):
         """
         Decorator function to be the OAuth user info getter
         for all the providers, receives provider and response
@@ -534,21 +536,11 @@
                 if provider == 'github':
                     me = sm.oauth_remotes[provider].get('user')
                     return {'username': me.data.get('login')}
-                else:
-                    return {}
+                return {}
         """
 
-        def wraps(provider, response=None):
-            ret = f(self, provider, response=response)
-            # Checks if decorator is well behaved and returns a dict as supposed.
-            if not type(ret) == dict:
-                log.error(
-                    "OAuth user info decorated function "
-                    "did not returned a dict, but: %s",
-                    type(ret),
-                )
-                return {}
-            return ret
+        def wraps(provider: str, response: Dict[str, Any] = None) -> Dict[str, Any]:
+            return func(self, provider, response)
 
         self.oauth_user_info = wraps
         return wraps
@@ -587,9 +579,11 @@
         )
         session["oauth_provider"] = provider
 
-    def get_oauth_user_info(self, provider, resp):
+    def get_oauth_user_info(
+        self, provider: str, resp: Dict[str, Any]
+    ) -> Dict[str, Any]:
         """
-        Since there are different OAuth API's with different ways to
+        Since there are different OAuth APIs with different ways to
         retrieve user info
         """
         # for GITHUB
@@ -628,21 +622,14 @@
                 "last_name": data.get("family_name", ""),
                 "email": data.get("email", ""),
             }
-        # for Azure AD Tenant. Azure OAuth response contains
-        # JWT token which has user info.
-        # JWT token needs to be base64 decoded.
-        # https://docs.microsoft.com/en-us/azure/active-directory/develop/
-        # active-directory-protocols-oauth-code
         if provider == "azure":
-            log.debug("Azure response received:\n%s", json.dumps(resp, indent=4))
             me = self._decode_and_validate_azure_jwt(resp["id_token"])
-            log.debug("Decoded JWT:\n%s", json.dumps(me, indent=4))
+            log.debug("User info from Azure: %s", me)
+            # https://learn.microsoft.com/en-us/azure/active-directory/develop/id-token-claims-reference#payload-claims
             return {
-                "name": me.get("name", ""),
-                "email": me["upn"],
+                "email": me["email"],
                 "first_name": me.get("given_name", ""),
                 "last_name": me.get("family_name", ""),
-                "id": me["oid"],
                 "username": me["oid"],
                 "role_keys": me.get("roles", []),
             }
@@ -680,16 +667,26 @@
                 "last_name": data.get("family_name", ""),
                 "email": data.get("email", ""),
             }
-        else:
-            return {}
+        raise OAuthProviderUnknown()
+
+    def _get_microsoft_jwks(self) -> List[Dict[str, Any]]:
+        import requests
+
+        return requests.get(MICROSOFT_KEY_SET_URL).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
+        )
+        if verify_signature:
+            from authlib.jose import JsonWebKey, jwt as authlib_jwt
 
-    def _decode_and_validate_azure_jwt(self, id_token):
-        keyset = JsonWebKey.import_key_set(requests.get(MICROSOFT_KEY_SET_URL).json())
-        claims = jwt.decode(id_token, keyset)
-        claims.validate()
-        log.debug("Decoded JWT:\n%s", json.dumps(claims, indent=4))
+            keyset = JsonWebKey.import_key_set(self._get_microsoft_jwks())
+            claims = authlib_jwt.decode(id_token, keyset)
+            claims.validate()
+            return claims
 
-        return claims
+        return jwt.decode(id_token, options={"verify_signature": False})
 
     def register_views(self):
         if not self.appbuilder.app.config.get("FAB_ADD_SECURITY_VIEWS", True):