AIP 84 - Add auth for asset alias

airflow/api_fastapi/auth/managers/base_auth_manager.py

@@ -47,6 +47,7 @@
     )
     from airflow.api_fastapi.auth.managers.models.resource_details import (
         AccessView,
+        AssetAliasDetails,
         AssetDetails,
         ConfigurationDetails,
         ConnectionDetails,
@@ -174,6 +175,22 @@ def is_authorized_asset(
         :param details: optional details about the asset
         """
 
+    @abstractmethod
+    def is_authorized_asset_alias(
+        self,
+        *,
+        method: ResourceMethod,
+        user: T,
+        details: AssetAliasDetails | None = None,
+    ) -> bool:
+        """
+        Return whether the user is authorized to perform a given action on an asset alias.
+
+        :param method: the method to perform
+        :param user: the user to perform the action on
+        :param details: optional details about the asset alias
+        """
+
     @abstractmethod
     def is_authorized_pool(
         self,

airflow/api_fastapi/auth/managers/models/resource_details.py

@@ -49,6 +49,13 @@ class AssetDetails:
     id: str | None = None
 
 
+@dataclass
+class AssetAliasDetails:
+    """Represents the details of an asset alias."""
+
+    id: str | None = None
+
+
 @dataclass
 class PoolDetails:
     """Represents the details of a pool."""

airflow/api_fastapi/auth/managers/simple/simple_auth_manager.py

@@ -40,6 +40,7 @@
     from airflow.api_fastapi.auth.managers.base_auth_manager import ResourceMethod
     from airflow.api_fastapi.auth.managers.models.resource_details import (
         AccessView,
+        AssetAliasDetails,
         AssetDetails,
         ConfigurationDetails,
         ConnectionDetails,
@@ -180,6 +181,20 @@ def is_authorized_asset(
             user=user,
         )
 
+    def is_authorized_asset_alias(
+        self,
+        *,
+        method: ResourceMethod,
+        user: SimpleAuthManagerUser,
+        details: AssetAliasDetails | None = None,
+    ) -> bool:
+        return self._is_authorized(
+            method=method,
+            allow_get_role=SimpleAuthManagerRole.VIEWER,
+            allow_role=SimpleAuthManagerRole.OP,
+            user=user,
+        )
+
     def is_authorized_pool(
         self,
         *,

airflow/api_fastapi/core_api/openapi/v1-generated.yaml

@@ -616,6 +616,8 @@ paths:
       summary: Get Asset Aliases
       description: Get asset aliases.
       operationId: get_asset_aliases
+      security:
+      - OAuth2PasswordBearer: []
       parameters:
       - name: limit
         in: query
@@ -686,6 +688,8 @@ paths:
       summary: Get Asset Alias
       description: Get an asset alias.
       operationId: get_asset_alias
+      security:
+      - OAuth2PasswordBearer: []
       parameters:
       - name: asset_alias_id
         in: path

airflow/api_fastapi/core_api/routes/public/assets.py

@@ -53,7 +53,11 @@
 )
 from airflow.api_fastapi.core_api.datamodels.dag_run import DAGRunResponse
 from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
-from airflow.api_fastapi.core_api.security import requires_access_asset, requires_access_dag
+from airflow.api_fastapi.core_api.security import (
+    requires_access_asset,
+    requires_access_asset_alias,
+    requires_access_dag,
+)
 from airflow.api_fastapi.logging.decorators import action_logging
 from airflow.assets.manager import asset_manager
 from airflow.models.asset import (
@@ -130,6 +134,7 @@ def get_assets(
 @assets_router.get(
     "/assets/aliases",
     responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+    dependencies=[Depends(requires_access_asset_alias(method="GET"))],
 )
 def get_asset_aliases(
     limit: QueryLimit,
@@ -160,6 +165,7 @@ def get_asset_aliases(
 @assets_router.get(
     "/assets/aliases/{asset_alias_id}",
     responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+    dependencies=[Depends(requires_access_asset_alias(method="GET"))],
 )
 def get_asset_alias(asset_alias_id: int, session: SessionDep):
     """Get an asset alias."""

airflow/api_fastapi/core_api/security.py

@@ -26,6 +26,7 @@
 from airflow.api_fastapi.app import get_auth_manager
 from airflow.api_fastapi.auth.managers.models.base_user import BaseUser
 from airflow.api_fastapi.auth.managers.models.resource_details import (
+    AssetAliasDetails,
     AssetDetails,
     ConfigurationDetails,
     ConnectionDetails,
@@ -174,6 +175,22 @@ def inner(
     return inner
 
 
+def requires_access_asset_alias(method: ResourceMethod) -> Callable:
+    def inner(
+        request: Request,
+        user: Annotated[BaseUser | None, Depends(get_user)] = None,
+    ) -> None:
+        asset_alias_id: str | None = request.path_params.get("asset_alias_id")
+
+        _requires_access(
+            is_authorized_callback=lambda: get_auth_manager().is_authorized_asset_alias(
+                method=method, details=AssetAliasDetails(id=asset_alias_id), user=user
+            ),
+        )
+
+    return inner
+
+
 def _requires_access(
     *,
     is_authorized_callback: Callable[[], bool],

airflow/security/permissions.py

@@ -34,6 +34,7 @@
 RESOURCE_DAG_WARNING = "DAG Warnings"
 RESOURCE_CLUSTER_ACTIVITY = "Cluster Activity"
 RESOURCE_ASSET = "Assets"
+RESOURCE_ASSET_ALIAS = "Asset Aliases"
 RESOURCE_DOCS = "Documentation"
 RESOURCE_DOCS_MENU = "Docs"
 RESOURCE_IMPORT_ERROR = "ImportError"

providers/amazon/src/airflow/providers/amazon/aws/auth_manager/avp/entities.py

@@ -34,6 +34,7 @@ class AvpEntities(Enum):
 
     # Resource types
     ASSET = "Asset"
+    ASSET_ALIAS = "AssetAlias"
     CONFIGURATION = "Configuration"
     CONNECTION = "Connection"
     CUSTOM = "Custom"

providers/amazon/src/airflow/providers/amazon/aws/auth_manager/aws_auth_manager.py

@@ -55,7 +55,11 @@
         IsAuthorizedPoolRequest,
         IsAuthorizedVariableRequest,
     )
-    from airflow.api_fastapi.auth.managers.models.resource_details import AssetDetails, ConfigurationDetails
+    from airflow.api_fastapi.auth.managers.models.resource_details import (
+        AssetAliasDetails,
+        AssetDetails,
+        ConfigurationDetails,
+    )
 
 
 class AwsAuthManager(BaseAuthManager[AwsAuthManagerUser]):
@@ -158,6 +162,14 @@ def is_authorized_asset(
             method=method, entity_type=AvpEntities.ASSET, user=user, entity_id=asset_id
         )
 
+    def is_authorized_asset_alias(
+        self, *, method: ResourceMethod, user: AwsAuthManagerUser, details: AssetAliasDetails | None = None
+    ) -> bool:
+        asset_alias_id = details.id if details else None
+        return self.avp_facade.is_authorized(
+            method=method, entity_type=AvpEntities.ASSET_ALIAS, user=user, entity_id=asset_alias_id
+        )
+
     def is_authorized_pool(
         self, *, method: ResourceMethod, user: AwsAuthManagerUser, details: PoolDetails | None = None
     ) -> bool:

providers/amazon/tests/unit/amazon/aws/auth_manager/test_aws_auth_manager.py

@@ -43,9 +43,9 @@
 
 if TYPE_CHECKING:
     from airflow.api_fastapi.auth.managers.base_auth_manager import ResourceMethod
-    from airflow.api_fastapi.auth.managers.models.resource_details import AssetDetails
+    from airflow.api_fastapi.auth.managers.models.resource_details import AssetAliasDetails, AssetDetails
 else:
-    from airflow.providers.common.compat.assets import AssetDetails
+    from airflow.providers.common.compat.assets import AssetAliasDetails, AssetDetails
 
 
 mock = Mock()
@@ -229,6 +229,37 @@ def test_is_authorized_asset(
         )
         assert result
 
+    @pytest.mark.parametrize(
+        "details, user, expected_user, expected_entity_id",
+        [
+            (None, mock, ANY, None),
+            (AssetAliasDetails(id="1"), mock, mock, "1"),
+        ],
+    )
+    @patch.object(AwsAuthManager, "avp_facade")
+    def test_is_authorized_asset_alias(
+        self,
+        mock_avp_facade,
+        details,
+        user,
+        expected_user,
+        expected_entity_id,
+        auth_manager,
+    ):
+        is_authorized = Mock(return_value=True)
+        mock_avp_facade.is_authorized = is_authorized
+
+        method: ResourceMethod = "GET"
+        result = auth_manager.is_authorized_asset_alias(method=method, details=details, user=user)
+
+        is_authorized.assert_called_once_with(
+            method=method,
+            entity_type=AvpEntities.ASSET_ALIAS,
+            user=expected_user,
+            entity_id=expected_entity_id,
+        )
+        assert result
+
     @pytest.mark.parametrize(
         "details, user, expected_user, expected_entity_id",
         [

providers/common/compat/src/airflow/providers/common/compat/assets/__init__.py

@@ -25,12 +25,12 @@
 )
 
 if TYPE_CHECKING:
-    from airflow.api_fastapi.auth.managers.models.resource_details import AssetDetails
+    from airflow.api_fastapi.auth.managers.models.resource_details import AssetAliasDetails, AssetDetails
     from airflow.models.asset import expand_alias_to_assets
     from airflow.sdk.definitions.asset import Asset, AssetAlias, AssetAll, AssetAny
 else:
     if AIRFLOW_V_3_0_PLUS:
-        from airflow.api_fastapi.auth.managers.models.resource_details import AssetDetails
+        from airflow.api_fastapi.auth.managers.models.resource_details import AssetAliasDetails, AssetDetails
         from airflow.models.asset import expand_alias_to_assets
         from airflow.sdk.definitions.asset import Asset, AssetAlias, AssetAll, AssetAny
     else:
@@ -52,6 +52,7 @@
 __all__ = [
     "Asset",
     "AssetAlias",
+    "AssetAliasDetails",
     "AssetAll",
     "AssetAny",
     "AssetDetails",

providers/common/compat/src/airflow/providers/common/compat/security/permissions.py

@@ -19,12 +19,12 @@
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
-    from airflow.security.permissions import RESOURCE_ASSET
+    from airflow.security.permissions import RESOURCE_ASSET, RESOURCE_ASSET_ALIAS
 else:
     try:
-        from airflow.security.permissions import RESOURCE_ASSET
+        from airflow.security.permissions import RESOURCE_ASSET, RESOURCE_ASSET_ALIAS
     except ImportError:
         from airflow.security.permissions import RESOURCE_DATASET as RESOURCE_ASSET
 
 
-__all__ = ["RESOURCE_ASSET"]
+__all__ = ["RESOURCE_ASSET", "RESOURCE_ASSET_ALIAS"]

providers/fab/src/airflow/providers/fab/auth_manager/fab_auth_manager.py

@@ -97,12 +97,12 @@
     from airflow.cli.cli_config import (
         CLICommand,
     )
-    from airflow.providers.common.compat.assets import AssetDetails
+    from airflow.providers.common.compat.assets import AssetAliasDetails, AssetDetails
     from airflow.providers.fab.auth_manager.security_manager.override import FabAirflowSecurityManagerOverride
     from airflow.providers.fab.www.extensions.init_appbuilder import AirflowAppBuilder
-    from airflow.providers.fab.www.security.permissions import RESOURCE_ASSET
+    from airflow.providers.fab.www.security.permissions import RESOURCE_ASSET, RESOURCE_ASSET_ALIAS
 else:
-    from airflow.providers.common.compat.security.permissions import RESOURCE_ASSET
+    from airflow.providers.common.compat.security.permissions import RESOURCE_ASSET, RESOURCE_ASSET_ALIAS
 
 
 _MAP_DAG_ACCESS_ENTITY_TO_FAB_RESOURCE_TYPE: dict[DagAccessEntity, tuple[str, ...]] = {
@@ -319,6 +319,11 @@ def is_authorized_asset(
     ) -> bool:
         return self._is_authorized(method=method, resource_type=RESOURCE_ASSET, user=user)
 
+    def is_authorized_asset_alias(
+        self, *, method: ResourceMethod, user: User, details: AssetAliasDetails | None = None
+    ) -> bool:
+        return self._is_authorized(method=method, resource_type=RESOURCE_ASSET_ALIAS, user=user)
+
     def is_authorized_pool(
         self, *, method: ResourceMethod, user: User, details: PoolDetails | None = None
     ) -> bool:

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

@@ -112,9 +112,9 @@
 )
 
 if TYPE_CHECKING:
-    from airflow.providers.fab.www.security.permissions import RESOURCE_ASSET
+    from airflow.providers.fab.www.security.permissions import RESOURCE_ASSET, RESOURCE_ASSET_ALIAS
 else:
-    from airflow.providers.common.compat.security.permissions import RESOURCE_ASSET
+    from airflow.providers.common.compat.security.permissions import RESOURCE_ASSET, RESOURCE_ASSET_ALIAS
 
 log = logging.getLogger(__name__)
 
@@ -233,6 +233,7 @@ class FabAirflowSecurityManagerOverride(AirflowSecurityManagerV2):
         (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_CODE),
         (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_RUN),
         (permissions.ACTION_CAN_READ, RESOURCE_ASSET),
+        (permissions.ACTION_CAN_READ, RESOURCE_ASSET_ALIAS),
         (permissions.ACTION_CAN_READ, permissions.RESOURCE_CLUSTER_ACTIVITY),
         (permissions.ACTION_CAN_READ, permissions.RESOURCE_POOL),
         (permissions.ACTION_CAN_READ, permissions.RESOURCE_IMPORT_ERROR),

providers/fab/src/airflow/providers/fab/www/security/permissions.py

@@ -34,6 +34,7 @@
 RESOURCE_DAG_WARNING = "DAG Warnings"
 RESOURCE_CLUSTER_ACTIVITY = "Cluster Activity"
 RESOURCE_ASSET = "Assets"
+RESOURCE_ASSET_ALIAS = "Asset Aliases"
 RESOURCE_DOCS = "Documentation"
 RESOURCE_DOCS_MENU = "Docs"
 RESOURCE_IMPORT_ERROR = "ImportError"

providers/fab/tests/unit/fab/auth_manager/test_fab_auth_manager.py

@@ -45,7 +45,7 @@
     from airflow.providers.fab.auth_manager.fab_auth_manager import FabAuthManager
     from airflow.providers.fab.auth_manager.security_manager.override import FabAirflowSecurityManagerOverride
 
-from airflow.providers.common.compat.security.permissions import RESOURCE_ASSET
+from airflow.providers.common.compat.security.permissions import RESOURCE_ASSET, RESOURCE_ASSET_ALIAS
 from airflow.providers.fab.www.security.permissions import (
     ACTION_CAN_ACCESS_MENU,
     ACTION_CAN_CREATE,
@@ -74,6 +74,7 @@
     "is_authorized_configuration": RESOURCE_CONFIG,
     "is_authorized_connection": RESOURCE_CONNECTION,
     "is_authorized_asset": RESOURCE_ASSET,
+    "is_authorized_asset_alias": RESOURCE_ASSET_ALIAS,
     "is_authorized_variable": RESOURCE_VARIABLE,
 }
 

providers/fab/tests/unit/fab/auth_manager/test_security.py

@@ -60,9 +60,9 @@
 from tests_common.test_utils.permissions import _resource_name
 
 if TYPE_CHECKING:
-    from airflow.providers.fab.www.security.permissions import RESOURCE_ASSET
+    from airflow.providers.fab.www.security.permissions import RESOURCE_ASSET, RESOURCE_ASSET_ALIAS
 else:
-    from airflow.providers.common.compat.security.permissions import RESOURCE_ASSET
+    from airflow.providers.common.compat.security.permissions import RESOURCE_ASSET, RESOURCE_ASSET_ALIAS
 
 
 pytestmark = pytest.mark.db_test
@@ -419,6 +419,7 @@ def test_get_user_roles_for_anonymous_user(app, security_manager):
         (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_CODE),
         (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_RUN),
         (permissions.ACTION_CAN_READ, RESOURCE_ASSET),
+        (permissions.ACTION_CAN_READ, RESOURCE_ASSET_ALIAS),
         (permissions.ACTION_CAN_READ, permissions.RESOURCE_CLUSTER_ACTIVITY),
         (permissions.ACTION_CAN_READ, permissions.RESOURCE_IMPORT_ERROR),
         (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_WARNING),