Fix inheritance chain in security manager

airflow/api_connexion/endpoints/role_and_permission_endpoint.py

@@ -40,14 +40,14 @@
 
 if TYPE_CHECKING:
     from airflow.api_connexion.types import APIResponse, UpdateMask
-    from airflow.www.security import AirflowSecurityManager
+    from airflow.www.security_manager import AirflowSecurityManagerV2
 
 
-def _check_action_and_resource(sm: AirflowSecurityManager, perms: list[tuple[str, str]]) -> None:
+def _check_action_and_resource(sm: AirflowSecurityManagerV2, perms: list[tuple[str, str]]) -> None:
     """
     Check if the action or resource exists and otherwise raise 400.
 
-    This function is intended for use in the REST API because it raise 400
+    This function is intended for use in the REST API because it raises an HTTP error 400
     """
     for action, resource in perms:
         if not sm.get_action(action):

airflow/auth/managers/base_auth_manager.py

@@ -24,9 +24,11 @@
 from airflow.utils.log.logging_mixin import LoggingMixin
 
 if TYPE_CHECKING:
+    from flask import Flask
+
     from airflow.auth.managers.models.base_user import BaseUser
     from airflow.cli.cli_config import CLICommand
-    from airflow.www.security import AirflowSecurityManager
+    from airflow.www.security_manager import AirflowSecurityManagerV2
 
 
 class BaseAuthManager(LoggingMixin):
@@ -36,8 +38,9 @@ class BaseAuthManager(LoggingMixin):
     Auth managers are responsible for any user management related operation such as login, logout, authz, ...
     """
 
-    def __init__(self):
-        self._security_manager: AirflowSecurityManager | None = None
+    def __init__(self, app: Flask) -> None:
+        self._security_manager: AirflowSecurityManagerV2 | None = None
+        self.app = app
 
     @staticmethod
     def get_cli_commands() -> list[CLICommand]:
@@ -80,22 +83,24 @@ def get_security_manager_override_class(self) -> type:
         Return the security manager override class.
 
         The security manager override class is responsible for overriding the default security manager
-        class airflow.www.security.AirflowSecurityManager with a custom implementation. This class is
-        essentially inherited from airflow.www.security.AirflowSecurityManager.
+        class airflow.www.security_manager.AirflowSecurityManagerV2 with a custom implementation.
+        This class is essentially inherited from airflow.www.security_manager.AirflowSecurityManagerV2.
 
-        By default, return an empty class.
+        By default, return the generic AirflowSecurityManagerV2.
         """
-        return object
+        from airflow.www.security_manager import AirflowSecurityManagerV2
+
+        return AirflowSecurityManagerV2
 
     @property
-    def security_manager(self) -> AirflowSecurityManager:
+    def security_manager(self) -> AirflowSecurityManagerV2:
         """Get the security manager."""
         if not self._security_manager:
             raise AirflowException("Security manager not defined.")
         return self._security_manager
 
     @security_manager.setter
-    def security_manager(self, security_manager: AirflowSecurityManager):
+    def security_manager(self, security_manager: AirflowSecurityManagerV2):
         """
         Set the security manager.
 

airflow/auth/managers/fab/cli_commands/role_command.py

@@ -29,7 +29,7 @@
 from airflow.utils import cli as cli_utils
 from airflow.utils.cli import suppress_logs_and_warning
 from airflow.utils.providers_configuration_loader import providers_configuration_loaded
-from airflow.www.security import EXISTING_ROLES
+from airflow.www.security_manager import EXISTING_ROLES
 
 if TYPE_CHECKING:
     from airflow.auth.managers.fab.models import Action, Permission, Resource, Role

airflow/auth/managers/fab/fab_auth_manager.py

@@ -17,6 +17,7 @@
 # under the License.
 from __future__ import annotations
 
+import warnings
 from typing import TYPE_CHECKING
 
 from airflow import AirflowException
@@ -90,8 +91,24 @@ def is_logged_in(self) -> bool:
     def get_security_manager_override_class(self) -> type:
         """Return the security manager override."""
         from airflow.auth.managers.fab.security_manager.override import FabAirflowSecurityManagerOverride
-
-        return FabAirflowSecurityManagerOverride
+        from airflow.www.security import AirflowSecurityManager
+
+        sm_from_config = self.app.config.get("SECURITY_MANAGER_CLASS")
+        if sm_from_config:
+            if not issubclass(sm_from_config, AirflowSecurityManager):
+                raise Exception(
+                    """Your CUSTOM_SECURITY_MANAGER must extend FabAirflowSecurityManagerOverride,
+                     not FAB's own security manager."""
+                )
+            if not issubclass(sm_from_config, FabAirflowSecurityManagerOverride):
+                warnings.warn(
+                    "Please make your custom security manager inherit from "
+                    "FabAirflowSecurityManagerOverride instead of AirflowSecurityManager.",
+                    DeprecationWarning,
+                )
+            return sm_from_config
+
+        return FabAirflowSecurityManagerOverride  # default choice
 
     def url_for(self, *args, **kwargs):
         """Wrapper to allow mocking without having to import at the top of the file."""