Revert "Provider fab auth manager deprecated methods removed (apache#…

…41720)" (apache#41960)

This reverts commit b039183.

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

@@ -23,7 +23,8 @@
 import os
 import random
 import uuid
-from typing import Any, Callable, Collection, Iterable, Sequence
+import warnings
+from typing import TYPE_CHECKING, Any, Callable, Collection, Container, Iterable, Sequence
 
 import jwt
 import packaging.version
@@ -68,12 +69,13 @@
 from markupsafe import Markup
 from sqlalchemy import and_, func, inspect, literal, or_, select
 from sqlalchemy.exc import MultipleResultsFound
-from sqlalchemy.orm import joinedload
+from sqlalchemy.orm import Session, joinedload
 from werkzeug.security import check_password_hash, generate_password_hash
 
 from airflow import __version__ as airflow_version
+from airflow.auth.managers.utils.fab import get_method_from_fab_action_map
 from airflow.configuration import conf
-from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning, RemovedInAirflow3Warning
 from airflow.models import DagBag, DagModel
 from airflow.providers.fab.auth_manager.models import (
     Action,
@@ -106,10 +108,14 @@
 )
 from airflow.providers.fab.auth_manager.views.user_stats import CustomUserStatsChartView
 from airflow.security import permissions
+from airflow.utils.session import NEW_SESSION, provide_session
 from airflow.www.extensions.init_auth_manager import get_auth_manager
 from airflow.www.security_manager import AirflowSecurityManagerV2
 from airflow.www.session import AirflowDatabaseSessionInterface
 
+if TYPE_CHECKING:
+    from airflow.auth.managers.base_auth_manager import ResourceMethod
+
 log = logging.getLogger(__name__)
 
 # This is the limit of DB user sessions that we consider as "healthy". If you have more sessions that this
@@ -956,6 +962,70 @@ def create_db(self):
             log.exception(const.LOGMSG_ERR_SEC_CREATE_DB)
             exit(1)
 
+    def get_readable_dags(self, user) -> Iterable[DagModel]:
+        """Get the DAGs readable by authenticated user."""
+        warnings.warn(
+            "`get_readable_dags` has been deprecated. Please use `get_auth_manager().get_permitted_dag_ids` "
+            "instead.",
+            RemovedInAirflow3Warning,
+            stacklevel=2,
+        )
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore", RemovedInAirflow3Warning)
+            return self.get_accessible_dags([permissions.ACTION_CAN_READ], user)
+
+    def get_editable_dags(self, user) -> Iterable[DagModel]:
+        """Get the DAGs editable by authenticated user."""
+        warnings.warn(
+            "`get_editable_dags` has been deprecated. Please use `get_auth_manager().get_permitted_dag_ids` "
+            "instead.",
+            RemovedInAirflow3Warning,
+            stacklevel=2,
+        )
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore", RemovedInAirflow3Warning)
+            return self.get_accessible_dags([permissions.ACTION_CAN_EDIT], user)
+
+    @provide_session
+    def get_accessible_dags(
+        self,
+        user_actions: Container[str] | None,
+        user,
+        session: Session = NEW_SESSION,
+    ) -> Iterable[DagModel]:
+        warnings.warn(
+            "`get_accessible_dags` has been deprecated. Please use "
+            "`get_auth_manager().get_permitted_dag_ids` instead.",
+            RemovedInAirflow3Warning,
+            stacklevel=3,
+        )
+
+        dag_ids = self.get_accessible_dag_ids(user, user_actions, session)
+        return session.scalars(select(DagModel).where(DagModel.dag_id.in_(dag_ids)))
+
+    @provide_session
+    def get_accessible_dag_ids(
+        self,
+        user,
+        user_actions: Container[str] | None = None,
+        session: Session = NEW_SESSION,
+    ) -> set[str]:
+        warnings.warn(
+            "`get_accessible_dag_ids` has been deprecated. Please use "
+            "`get_auth_manager().get_permitted_dag_ids` instead.",
+            RemovedInAirflow3Warning,
+            stacklevel=3,
+        )
+        if not user_actions:
+            user_actions = [permissions.ACTION_CAN_EDIT, permissions.ACTION_CAN_READ]
+        method_from_fab_action_map = get_method_from_fab_action_map()
+        user_methods: Container[ResourceMethod] = [
+            method_from_fab_action_map[action]
+            for action in method_from_fab_action_map
+            if action in user_actions
+        ]
+        return get_auth_manager().get_permitted_dag_ids(user=user, methods=user_methods, session=session)
+
     @staticmethod
     def get_readable_dag_ids(user=None) -> set[str]:
         """Get the DAG IDs readable by authenticated user."""
@@ -1013,6 +1083,17 @@ def create_dag_specific_permissions(self) -> None:
             if dag.access_control is not None:
                 self.sync_perm_for_dag(root_dag_id, dag.access_control)
 
+    def prefixed_dag_id(self, dag_id: str) -> str:
+        """Return the permission name for a DAG id."""
+        warnings.warn(
+            "`prefixed_dag_id` has been deprecated. "
+            "Please use `airflow.security.permissions.resource_name` instead.",
+            RemovedInAirflow3Warning,
+            stacklevel=2,
+        )
+        root_dag_id = self._get_root_dag_id(dag_id)
+        return self._resource_name(root_dag_id, permissions.RESOURCE_DAG)
+
     def is_dag_resource(self, resource_name: str) -> bool:
         """Determine if a resource belongs to a DAG or all DAGs."""
         if resource_name == permissions.RESOURCE_DAG:
@@ -1340,6 +1421,20 @@ def permission_exists_in_one_or_more_roles(
     def perms_include_action(self, perms, action_name):
         return any(perm.action and perm.action.name == action_name for perm in perms)
 
+    def init_role(self, role_name, perms) -> None:
+        """
+        Initialize the role with actions and related resources.
+
+        :param role_name:
+        :param perms:
+        """
+        warnings.warn(
+            "`init_role` has been deprecated. Please use `bulk_sync_roles` instead.",
+            RemovedInAirflow3Warning,
+            stacklevel=2,
+        )
+        self.bulk_sync_roles([{"role": role_name, "perms": perms}])
+
     def bulk_sync_roles(self, roles: Iterable[dict[str, Any]]) -> None:
         """Sync the provided roles and permissions."""
         existing_roles = self._get_all_roles_with_permissions()

tests/providers/fab/auth_manager/test_security.py

@@ -266,6 +266,25 @@ def _assert_user_does_not_have_dag_perms(dag_id, perms, user=None):
     return _assert_user_does_not_have_dag_perms
 
 
+@pytest.mark.parametrize(
+    "role",
+    [{"name": "MyRole7", "permissions": [("can_some_other_action", "AnotherBaseView")], "create": False}],
+    indirect=True,
+)
+def test_init_role_baseview(app, security_manager, role):
+    _, params = role
+
+    with pytest.warns(
+        DeprecationWarning,
+        match="`init_role` has been deprecated\\. Please use `bulk_sync_roles` instead\\.",
+    ):
+        security_manager.init_role(params["name"], params["permissions"])
+
+    _role = security_manager.find_role(params["name"])
+    assert _role is not None
+    assert len(_role.permissions) == len(params["permissions"])
+
+
 @pytest.mark.parametrize(
     "role",
     [{"name": "MyRole3", "permissions": [("can_some_action", "SomeBaseView")]}],
@@ -983,6 +1002,17 @@ def test_get_all_roles_with_permissions(security_manager):
     assert "Admin" in roles
 
 
+def test_prefixed_dag_id_is_deprecated(security_manager):
+    with pytest.warns(
+        DeprecationWarning,
+        match=(
+            "`prefixed_dag_id` has been deprecated. "
+            "Please use `airflow.security.permissions.resource_name` instead."
+        ),
+    ):
+        security_manager.prefixed_dag_id("hello")
+
+
 def test_permissions_work_for_dags_with_dot_in_dagname(
     app, security_manager, assert_user_has_dag_perms, assert_user_does_not_have_dag_perms, session
 ):