feat: Added authentik as new identity provider

docs/security.rst

@@ -343,13 +343,35 @@ Specify a list of OAUTH_PROVIDERS in **config.py** that you want to allow for yo
                 "authorize_url": "https://login.microsoftonline.com/AZURE_TENANT_ID/oauth2/authorize",
             },
         },
+        {
+            "name": "authentik",
+            "token_key": "access_token",
+            "icon": "fa-fingerprint",
+            "remote_app": {
+                "api_base_url": "https://authentik.mydomain.com",
+                "client_kwargs": {
+                    "scope": "email profile",
+                    "verify_signature": True,
+                },
+                "access_token_url": (
+                    "https://authentik.mydomain.com/application/o/token/"
+                ),
+                "authorize_url": (
+                    "https://authentik.mydomain.com/application/o/authorize/"
+                ),
+                "request_token_url": None,
+                "client_id": "CLIENT_ID",
+                "client_secret": "CLIENT_SECRET",
+                'jwks_uri': 'https://authentik.mydomain.com/application/o/APPLICATION_NAME/jwks/',
+            },
+        },
     ]
 
 This needs a small explanation, you basically have five special keys:
 
 :name: the name of the provider:
     you can choose whatever you want, but FAB has builtin logic in `BaseSecurityManager.get_oauth_user_info()` for:
-    'azure', 'github', 'google', 'keycloak', 'keycloak_before_17', 'linkedin', 'okta', 'openshift', 'twitter'
+    'authentik', 'azure', 'github', 'google', 'keycloak', 'keycloak_before_17', 'linkedin', 'okta', 'openshift', 'twitter'
 
 :icon: the font-awesome icon for this provider
 

flask_appbuilder/exceptions.py

@@ -58,3 +58,11 @@ class OAuthProviderUnknown(FABException):
     """
 
     ...
+
+
+class InvalidLoginAttempt(FABException):
+    """
+    When the credentials entered could not be verified
+    """
+
+    ...

flask_appbuilder/security/manager.py

@@ -4,7 +4,7 @@
 from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union
 
 from flask import Flask, g, session, url_for
-from flask_appbuilder.exceptions import OAuthProviderUnknown
+from flask_appbuilder.exceptions import InvalidLoginAttempt, 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
@@ -680,6 +680,18 @@ def get_oauth_user_info(
                 "last_name": data.get("family_name", ""),
                 "email": data.get("email", ""),
             }
+        # for Authentik
+        if provider == "authentik":
+            id_token = resp["id_token"]
+            me = self._get_authentik_token_info(id_token)
+            log.debug("User info from authentik: %s", me)
+            return {
+                "email": me["preferred_username"],
+                "first_name": me.get("given_name", ""),
+                "username": me["nickname"],
+                "role_keys": me.get("groups", []),
+            }
+
         raise OAuthProviderUnknown()
 
     def _get_microsoft_jwks(self) -> List[Dict[str, Any]]:
@@ -701,6 +713,48 @@ def _decode_and_validate_azure_jwt(self, id_token: str) -> Dict[str, str]:
 
         return jwt.decode(id_token, options={"verify_signature": False})
 
+    def _get_authentik_jwks(self, jwks_url) -> dict:
+        import requests
+
+        resp = requests.get(jwks_url)
+        if resp.status_code == 200:
+            return resp.json()
+        return False
+
+    def _validate_jwt(self, id_token, jwks):
+        from authlib.jose import JsonWebKey, jwt as authlib_jwt
+
+        keyset = JsonWebKey.import_key_set(jwks)
+        claims = authlib_jwt.decode(id_token, keyset)
+        claims.validate()
+        log.info("JWT token is validated")
+        return claims
+
+    def _get_authentik_token_info(self, id_token):
+        me = jwt.decode(id_token, options={"verify_signature": False})
+
+        verify_signature = self.oauth_remotes["authentik"].client_kwargs.get(
+            "verify_signature", True
+        )
+        if verify_signature:
+            # Validate the token using authentik certificate
+            jwks_uri = self.oauth_remotes["authentik"].server_metadata.get("jwks_uri")
+            if jwks_uri:
+                jwks = self._get_authentik_jwks(jwks_uri)
+                if jwks:
+                    return self._validate_jwt(id_token, jwks)
+            else:
+                log.error(
+                    "jwks_uri not specified in OAuth Providers, "
+                    "could not verify token signature"
+                )
+        else:
+            # Return the token info without validating
+            log.warning("JWT token is not validated!")
+            return me
+
+        raise InvalidLoginAttempt("OAuth signature verify failed")
+
     def register_views(self):
         if not self.appbuilder.app.config.get("FAB_ADD_SECURITY_VIEWS", True):
             return