Upgrade FAB to 4.4.1 (#38319)

This upgrades FAB to version 4.4.1 and brings all the changes done
to security manager between v4.3.11 and v4.4.1. As usual we need
to manually synchronize the changes to security manager which we
override in FAB provider.

This one also has the nice effect of bumping limits for a number of
dependencies we have, such as Werkzeug (finally <4 rather than <3)
and Flask (<3 rather than <2.3).

Still SQLAlchemy is not bumped though.

airflow/api_connexion/endpoints/pool_endpoint.py

@@ -114,8 +114,11 @@ def patch_pool(
         update_mask = [i.strip() for i in update_mask]
         _patch_body = {}
         try:
+            # MyPy infers a List[Optional[str]]  type here but it should be a List[str]
+            # there is no way field is None here (UpdateMask is a List[str])
+            # so if pool_schema.declared_fields[field].attribute is None file is returned
             update_mask = [
-                pool_schema.declared_fields[field].attribute
+                pool_schema.declared_fields[field].attribute  # type: ignore[misc]
                 if pool_schema.declared_fields[field].attribute
                 else field
                 for field in update_mask

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

@@ -340,6 +340,43 @@ def __init__(self, appbuilder):
         # Setup Flask-Jwt-Extended
         self.create_jwt_manager()
 
+    def _get_authentik_jwks(self, jwks_url) -> dict:
+        import requests
+
+        resp = requests.get(jwks_url)
+        if resp.status_code == 200:
+            return resp.json()
+        return {}
+
+    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 AirflowException("OAuth signature verify failed")
+
     def register_views(self):
         """Register FAB auth manager related views."""
         if not self.appbuilder.get_app.config.get("FAB_ADD_SECURITY_VIEWS", True):
@@ -513,9 +550,10 @@ def reset_user_sessions(self, user: User) -> None:
     def load_user_jwt(self, _jwt_header, jwt_data):
         identity = jwt_data["sub"]
         user = self.load_user(identity)
-        # Set flask g.user to JWT user, we can't do it on before request
-        g.user = user
-        return user
+        if user.is_active:
+            # Set flask g.user to JWT user, we can't do it on before request
+            g.user = user
+            return user
 
     @property
     def auth_type(self):
@@ -647,6 +685,10 @@ def auth_user_registration_role_jmespath(self) -> str:
         """The JMESPATH role to use for user registration."""
         return self.appbuilder.get_app.config["AUTH_USER_REGISTRATION_ROLE_JMESPATH"]
 
+    @property
+    def auth_remote_user_env_var(self) -> str:
+        return self.appbuilder.get_app.config["AUTH_REMOTE_USER_ENV_VAR"]
+
     @property
     def api_login_allow_multiple_providers(self):
         return self.appbuilder.get_app.config["AUTH_API_LOGIN_ALLOW_MULTIPLE_PROVIDERS"]
@@ -790,6 +832,9 @@ def _init_config(self):
             app.config.setdefault("AUTH_LDAP_LASTNAME_FIELD", "sn")
             app.config.setdefault("AUTH_LDAP_EMAIL_FIELD", "mail")
 
+        if self.auth_type == AUTH_REMOTE_USER:
+            app.config.setdefault("AUTH_REMOTE_USER_ENV_VAR", "REMOTE_USER")
+
         # Rate limiting
         app.config.setdefault("AUTH_RATE_LIMITED", True)
         app.config.setdefault("AUTH_RATE_LIMIT", "5 per 40 second")
@@ -804,7 +849,12 @@ def _init_auth(self):
         if self.auth_type == AUTH_OID:
             from flask_openid import OpenID
 
+            log.warning(
+                "AUTH_OID is deprecated and will be removed in version 5. "
+                "Migrate to other authentication methods."
+            )
             self.oid = OpenID(app)
+
         if self.auth_type == AUTH_OAUTH:
             from authlib.integrations.flask_client import OAuth
 
@@ -1471,8 +1521,9 @@ def add_user(
             return False
 
     def load_user(self, user_id):
-        """Load user by ID."""
-        return self.get_user_by_id(int(user_id))
+        user = self.get_user_by_id(int(user_id))
+        if user.is_active:
+            return user
 
     def get_user_by_id(self, pk):
         return self.get_session.get(self.user_model, pk)
@@ -2208,6 +2259,19 @@ def get_oauth_user_info(self, provider: str, resp: dict[str, Any]) -> dict[str,
                 "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", []),
+            }
+
         else:
             return {}
 

airflow/providers/fab/provider.yaml

@@ -44,7 +44,7 @@ dependencies:
   # Every time we update FAB version here, please make sure that you review the classes and models in
   # `airflow/providers/fab/auth_manager/security_manager/override.py` with their upstream counterparts.
   # In particular, make sure any breaking changes, for example any new methods, are accounted for.
-  - flask-appbuilder==4.3.11
+  - flask-appbuilder==4.4.1
   - flask-login>=0.6.2
   - google-re2>=1.0
 

dev/breeze/tests/test_packages.py

@@ -166,7 +166,7 @@ def test_get_documentation_package_path():
             "",
             """
     "apache-airflow>=2.9.0",
-    "flask-appbuilder==4.3.11",
+    "flask-appbuilder==4.4.1",
     "flask-login>=0.6.2",
     "flask>=2.2,<2.3",
     "google-re2>=1.0",
@@ -178,7 +178,7 @@ def test_get_documentation_package_path():
             "dev0",
             """
     "apache-airflow>=2.9.0.dev0",
-    "flask-appbuilder==4.3.11",
+    "flask-appbuilder==4.4.1",
     "flask-login>=0.6.2",
     "flask>=2.2,<2.3",
     "google-re2>=1.0",
@@ -190,7 +190,7 @@ def test_get_documentation_package_path():
             "beta0",
             """
     "apache-airflow>=2.9.0b0",
-    "flask-appbuilder==4.3.11",
+    "flask-appbuilder==4.4.1",
     "flask-login>=0.6.2",
     "flask>=2.2,<2.3",
     "google-re2>=1.0",

generated/provider_dependencies.json

@@ -468,7 +468,7 @@
   "fab": {
     "deps": [
       "apache-airflow>=2.9.0",
-      "flask-appbuilder==4.3.11",
+      "flask-appbuilder==4.4.1",
       "flask-login>=0.6.2",
       "flask>=2.2,<2.3",
       "google-re2>=1.0"

pyproject.toml

@@ -702,7 +702,7 @@ exasol = [ # source: airflow/providers/exasol/provider.yaml
   "pyexasol>=0.5.1",
 ]
 fab = [ # source: airflow/providers/fab/provider.yaml
-  "flask-appbuilder==4.3.11",
+  "flask-appbuilder==4.4.1",
   "flask-login>=0.6.2",
   "flask>=2.2,<2.3",
   "google-re2>=1.0",

scripts/in_container/install_airflow_and_providers.py

@@ -469,36 +469,54 @@ def install_airflow_and_providers(
         use_packages_from_dist=use_packages_from_dist,
     )
     if installation_spec.airflow_package and install_airflow_with_constraints:
-        install_airflow_cmd = [
+        base_install_airflow_cmd = [
             "pip",
             "install",
             "--root-user-action",
             "ignore",
             installation_spec.airflow_package,
         ]
+        install_airflow_cmd = base_install_airflow_cmd.copy()
         console.print(f"\n[bright_blue]Installing airflow package: {installation_spec.airflow_package}")
         if installation_spec.airflow_constraints_location:
             console.print(f"[bright_blue]Use constraints: {installation_spec.airflow_constraints_location}")
             install_airflow_cmd.extend(["--constraint", installation_spec.airflow_constraints_location])
         console.print()
-        run_command(install_airflow_cmd, github_actions=github_actions, check=True)
+        result = run_command(install_airflow_cmd, github_actions=github_actions, check=False)
+        if result.returncode != 0:
+            console.print(
+                "[warning]Installation with constraints failed - might be because pre-installed provider"
+                " has conflicting dependencies in PyPI. Falling back to a non-constraint installation."
+            )
+            run_command(base_install_airflow_cmd, github_actions=github_actions, check=True)
     if installation_spec.provider_packages or not install_airflow_with_constraints:
-        install_providers_cmd = ["pip", "install", "--root-user-action", "ignore"]
+        base_install_providers_cmd = ["pip", "install", "--root-user-action", "ignore"]
         if not install_airflow_with_constraints and installation_spec.airflow_package:
-            install_providers_cmd.append(installation_spec.airflow_package)
+            base_install_providers_cmd.append(installation_spec.airflow_package)
         console.print("\n[bright_blue]Installing provider packages:")
         for provider_package in sorted(installation_spec.provider_packages):
             console.print(f"  {provider_package}")
         console.print()
         for provider_package in installation_spec.provider_packages:
-            install_providers_cmd.append(provider_package)
+            base_install_providers_cmd.append(provider_package)
+        install_providers_command = base_install_providers_cmd.copy()
         if installation_spec.provider_constraints_location:
             console.print(
                 f"[bright_blue]with constraints: {installation_spec.provider_constraints_location}\n"
             )
-            install_providers_cmd.extend(["--constraint", installation_spec.provider_constraints_location])
-        console.print()
-        run_command(install_providers_cmd, github_actions=github_actions, check=True)
+            install_providers_command.extend(
+                ["--constraint", installation_spec.provider_constraints_location]
+            )
+            console.print()
+            result = run_command(install_providers_command, github_actions=github_actions, check=False)
+            if result.returncode != 0:
+                console.print(
+                    "[warning]Installation with constraints failed - might be because pre-installed provider"
+                    " has conflicting dependencies in PyPI. Falling back to a non-constraint installation."
+                )
+                run_command(base_install_providers_cmd, github_actions=github_actions, check=True)
+        else:
+            run_command(base_install_providers_cmd, github_actions=github_actions, check=True)
     console.print("[green]Done!")
 
 

tests/www/views/test_session.py

@@ -97,7 +97,7 @@ def test_check_active_user(app, user_client):
     user.active = False
     resp = user_client.get("/home")
     assert resp.status_code == 302
-    assert "/logout" in resp.headers.get("Location")
+    assert "/login/?next=http%3A%2F%2Flocalhost%2Fhome" in resp.headers.get("Location")
 
 
 def test_check_deactivated_user_redirected_to_login(app, user_client):