Implement is_authorized() in auth manager

airflow/auth/managers/base_auth_manager.py

@@ -18,9 +18,14 @@
 from __future__ import annotations
 
 from abc import abstractmethod
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Sequence
 
+from airflow.auth.managers.models.authorized_action import AuthorizedAction
+from airflow.auth.managers.models.resource_action import ResourceAction
+from airflow.auth.managers.models.resource_details import ResourceDetails
 from airflow.exceptions import AirflowException
+from airflow.models.dag import DagModel
+from airflow.security.permissions import RESOURCE_DAG, RESOURCE_DAG_PREFIX
 from airflow.utils.log.logging_mixin import LoggingMixin
 
 if TYPE_CHECKING:
@@ -63,6 +68,69 @@ def get_user_id(self) -> str:
     def is_logged_in(self) -> bool:
         """Return whether the user is logged in."""
 
+    @abstractmethod
+    def is_authorized(
+        self,
+        action: ResourceAction,
+        resource_type: str,
+        resource_details: ResourceDetails | None = None,
+        user: BaseUser | None = None,
+    ) -> bool:
+        """
+        Return whether the user is authorized to perform a given action.
+
+        .. code-block:: python
+
+            # Check whether the logged-in user has permission to read the DAG "my_dag_id"
+            get_auth_manager().is_authorized(
+                Action.GET,
+                Resource.DAG,
+                ResourceDetails(
+                    id="my_dag_id",
+                ),
+            )
+
+        :param action: the action to perform
+        :param resource_type: the type of resource the user attempts to perform the action on
+        :param resource_details: optional details about the resource itself
+        :param user: the user to perform the action on. If not provided (or None), it uses the current user
+        """
+
+    def is_all_authorized(
+        self,
+        actions: Sequence[AuthorizedAction],
+    ) -> bool:
+        """
+        Wrapper around `is_authorized` to check whether the user is authorized to perform several actions.
+
+        :param actions: the list of actions to check. Each item represents the list of parameters of
+            `is_authorized`
+        """
+        return all(
+            self.is_authorized(
+                action=action.action,
+                resource_type=action.resource_type,
+                resource_details=action.resource_details,
+                user=action.user,
+            )
+            for action in actions
+        )
+
+    def _get_root_dag_id(self, dag_id: str) -> str:
+        """
+        Return the root DAG id in case of sub DAG, return the DAG id otherwise.
+
+        :param dag_id: the DAG id
+        """
+        if "." in dag_id:
+            dm = (
+                self.security_manager.appbuilder.get_session.query(DagModel.dag_id, DagModel.root_dag_id)
+                .filter(DagModel.dag_id == dag_id)
+                .first()
+            )
+            return dm.root_dag_id or dm.dag_id
+        return dag_id
+
     @abstractmethod
     def get_url_login(self, **kwargs) -> str:
         """Return the login page url."""
@@ -102,3 +170,8 @@ def security_manager(self, security_manager: AirflowSecurityManager):
         :param security_manager: the security manager
         """
         self._security_manager = security_manager
+
+    @staticmethod
+    def is_dag_resource(resource_name: str) -> bool:
+        """Determines if a resource relates to a DAG."""
+        return resource_name == RESOURCE_DAG or resource_name.startswith(RESOURCE_DAG_PREFIX)

airflow/auth/managers/fab/fab_auth_manager.py

@@ -26,14 +26,33 @@
     SYNC_PERM_COMMAND,
     USERS_COMMANDS,
 )
+from airflow.auth.managers.models.base_user import BaseUser
+from airflow.auth.managers.models.resource_action import ResourceAction
+from airflow.auth.managers.models.resource_details import ResourceDetails
 from airflow.cli.cli_config import (
     CLICommand,
     GroupCommand,
 )
+from airflow.security.permissions import (
+    ACTION_CAN_ACCESS_MENU,
+    ACTION_CAN_CREATE,
+    ACTION_CAN_DELETE,
+    ACTION_CAN_EDIT,
+    ACTION_CAN_READ,
+    RESOURCE_DAG,
+    RESOURCE_DAG_PREFIX,
+)
 
 if TYPE_CHECKING:
     from airflow.auth.managers.fab.models import User
 
+_MAP_ACTION_NAME_TO_FAB_ACTION_NAME = {
+    ResourceAction.POST: ACTION_CAN_CREATE,
+    ResourceAction.GET: ACTION_CAN_READ,
+    ResourceAction.PUT: ACTION_CAN_EDIT,
+    ResourceAction.DELETE: ACTION_CAN_DELETE,
+}
+
 
 class FabAuthManager(BaseAuthManager):
     """
@@ -85,6 +104,42 @@ def is_logged_in(self) -> bool:
         """Return whether the user is logged in."""
         return not self.get_user().is_anonymous
 
+    def is_authorized(
+        self,
+        action: ResourceAction,
+        resource_type: str,
+        resource_details: ResourceDetails | None = None,
+        user: BaseUser | None = None,
+    ) -> bool:
+        """
+        Return whether the user is authorized to perform a given action.
+
+        :param action: the action to perform
+        :param resource_type: the type of resource the user attempts to perform the action on
+        :param resource_details: optional details about the resource itself
+        :param user: the user to perform the action on. If not provided (or None), it uses the current user
+        """
+        if not user:
+            user = self.get_user()
+
+        fab_action = self._get_fab_action(action)
+        user_permissions = self._get_user_permissions(user)
+
+        if (fab_action, resource_type) in user_permissions:
+            return True
+
+        if self.is_dag_resource(resource_type):
+            # Check whether the user has permissions to access all DAGs
+            if (fab_action, RESOURCE_DAG) in user_permissions:
+                return True
+
+            if resource_details and resource_details.id:
+                # Check whether the user has permissions to access a specific DAG
+                resource_dag_name = self._resource_name_for_dag(resource_details.id)
+                return (fab_action, resource_dag_name) in user_permissions
+
+        return False
+
     def get_security_manager_override_class(self) -> type:
         """Return the security manager override."""
         from airflow.auth.managers.fab.security_manager.override import FabAirflowSecurityManagerOverride
@@ -117,3 +172,49 @@ def get_url_user_profile(self) -> str | None:
         if not self.security_manager.user_view:
             return None
         return self.url_for(f"{self.security_manager.user_view.endpoint}.userinfo")
+
+    @staticmethod
+    def _get_fab_action(action: ResourceAction) -> str:
+        """
+        Convert the action to a FAB action.
+
+        :param action: the action to convert
+
+        :meta private:
+        """
+        if action not in _MAP_ACTION_NAME_TO_FAB_ACTION_NAME:
+            raise AirflowException(f"Unknown action: {action}")
+        return _MAP_ACTION_NAME_TO_FAB_ACTION_NAME[action]
+
+    def _resource_name_for_dag(self, dag_id: str) -> str:
+        """
+        Returns the FAB resource name for a DAG id.
+
+        :param dag_id: the DAG id
+
+        :meta private:
+        """
+        root_dag_id = self._get_root_dag_id(dag_id)
+        if root_dag_id == RESOURCE_DAG:
+            return root_dag_id
+        if root_dag_id.startswith(RESOURCE_DAG_PREFIX):
+            return root_dag_id
+        return f"{RESOURCE_DAG_PREFIX}{root_dag_id}"
+
+    @staticmethod
+    def _get_user_permissions(user: BaseUser):
+        """
+        Return the user permissions.
+
+        ACTION_CAN_READ and ACTION_CAN_ACCESS_MENU are merged into because they are very similar.
+        We can assume that if a user has permissions to read variables, they also have permissions to access
+        the menu "Variables".
+
+        :param user: the user to get permissions for
+
+        :meta private:
+        """
+        return [
+            (ACTION_CAN_READ if perm[0] == ACTION_CAN_ACCESS_MENU else perm[0], perm[1])
+            for perm in user.perms
+        ]

airflow/auth/managers/models/authorized_action.py

@@ -0,0 +1,34 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from airflow.auth.managers.models.base_user import BaseUser
+from airflow.auth.managers.models.resource_action import ResourceAction
+from airflow.auth.managers.models.resource_details import ResourceDetails
+
+
+@dataclass
+class AuthorizedAction:
+    """Represents an action that is being checked for authorization."""
+
+    action: ResourceAction
+    resource_type: str
+    resource_details: ResourceDetails | None = None
+    user: BaseUser | None = None

airflow/auth/managers/models/base_user.py

@@ -18,6 +18,7 @@
 from __future__ import annotations
 
 from abc import abstractmethod
+from typing import Any
 
 
 class BaseUser:
@@ -37,6 +38,11 @@ def is_active(self) -> bool:
     def is_anonymous(self) -> bool:
         ...
 
+    @property
+    @abstractmethod
+    def perms(self) -> Any:
+        ...
+
     @abstractmethod
     def get_id(self) -> str:
         ...

airflow/auth/managers/models/resource_action.py

@@ -0,0 +1,38 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from enum import Enum
+
+
+class ResourceAction(Enum):
+    """
+    Define the type of action/operation the user is doing.
+
+    This is used when doing authorization check to define the type of action/operation
+    the user is doing.
+    """
+
+    # Create a resource
+    POST = "POST"
+    # Read a resource
+    GET = "GET"
+    # Update a resource
+    PUT = "PUT"
+    # Delete a resource
+    DELETE = "DELETE"

airflow/auth/managers/models/resource_details.py

@@ -0,0 +1,31 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass
+class ResourceDetails:
+    """
+    Represents the details of a resource.
+
+    All fields must be optional. These details can be used in authorization decision.
+    """
+
+    id: str | None = None

airflow/configuration.py

@@ -35,14 +35,13 @@
 from contextlib import contextmanager
 from copy import deepcopy
 from json.decoder import JSONDecodeError
-from typing import IO, Any, Dict, Generator, Iterable, Pattern, Set, Tuple, Union
+from typing import IO, TYPE_CHECKING, Any, Dict, Generator, Iterable, Pattern, Set, Tuple, Union
 from urllib.parse import urlsplit
 
 import re2
 from packaging.version import parse as parse_version
 from typing_extensions import overload
 
-from airflow.auth.managers.base_auth_manager import BaseAuthManager
 from airflow.exceptions import AirflowConfigException
 from airflow.secrets import DEFAULT_SECRETS_SEARCH_PATH, BaseSecretsBackend
 from airflow.utils import yaml
@@ -51,6 +50,9 @@
 from airflow.utils.providers_configuration_loader import providers_configuration_loaded
 from airflow.utils.weight_rule import WeightRule
 
+if TYPE_CHECKING:
+    from airflow.auth.managers.base_auth_manager import BaseAuthManager
+
 log = logging.getLogger(__name__)
 
 # show Airflow's deprecation warnings