Repository: apache/airflow

Issue:
Custom FAB actions no longer work
### Apache Airflow version

2.9.0

### If "Other Airflow 2 version" selected, which one?

_No response_

### What happened?

Before the auth manager/FAB transition in 2.9.0, plugins were able to include custom actions. This no longer works:

```
[2024-04-19T21:24:50.050+0000] {app.py:1744} ERROR - Exception on /testappbuilderbaseview/ [GET]                                                             
Traceback (most recent call last):                                                                                                                           
  File "/home/airflow/.local/lib/python3.12/site-packages/flask/app.py", line 2529, in wsgi_app                                                              
    response = self.full_dispatch_request()                                                                                                                  
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^                                                                                                                  
  File "/home/airflow/.local/lib/python3.12/site-packages/flask/app.py", line 1825, in full_dispatch_request                                                 
    rv = self.handle_user_exception(e)                                                                                                                       
         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^                                                                                                                       
  File "/home/airflow/.local/lib/python3.12/site-packages/flask/app.py", line 1823, in full_dispatch_request                                                 
    rv = self.dispatch_request()                                                                                                                             
         ^^^^^^^^^^^^^^^^^^^^^^^                                                                                                                             
  File "/home/airflow/.local/lib/python3.12/site-packages/flask/app.py", line 1799, in dispatch_request                                                      
    return self.ensure_sync(self.view_functions[rule.endpoint])(**view_args)                                                                                 
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^                                                                                 
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/fab/auth_manager/decorators/auth.py", line 118, in decorated                     
    is_authorized=appbuilder.sm.check_authorization(permissions, dag_id),                                                                                    
                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^                                                                                     
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/fab/auth_manager/security_manager/override.py", line 2305, in check_authorization
    elif not self.has_access(*perm):                                                                                                                         
             ^^^^^^^^^^^^^^^^^^^^^^                                                                                                                          
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/www/security_manager.py", line 142, in has_access                                          
    return is_authorized_method(action_name, resource_pk, user)                                                                                              
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^                                                                                              
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/www/security_manager.py", line 340, in <lambda>                                            
    method=get_method_from_fab_action_map()[action],                                                                                                         
           ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^^^                                                                                                          
KeyError: 'can_do'                                                                                                                                           
```

### What you think should happen instead?

Custom actions should continue to function.

### How to reproduce

This is a simple plugin that reproduces the issue:

```
from airflow.plugins_manager import AirflowPlugin
from airflow.www.auth import has_access
from flask_appbuilder import expose, BaseView as AppBuilderBaseView


class TestAppBuilderBaseView(AppBuilderBaseView):
    default_view = "test"
    base_permissions = ['can_do']

    @expose("/")
    @has_access(
        [
            ("can_do", "TestAppBuilderBaseView"),
        ]
    )
    def test(self):
        return "hello"


v_appbuilder_view = TestAppBuilderBaseView()
v_appbuilder_package = {
    "name": "Test View",
    "category": "Test Plugin",
    "view": v_appbuilder_view,
}

class AirflowTestPlugin(AirflowPlugin):
    name = "test_plugin"
    appbuilder_views = [v_appbuilder_package]
```

Note that users with the new action + resource perm still face the above traceback.

### Operating System

Debian (official image)

### Versions of Apache Airflow Providers

apache-airflow-providers-fab==1.0.2

### Deployment

Official Apache Airflow Helm Chart

### Deployment details

_No response_

### Anything else?

_No response_

### Are you willing to submit PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)


Discussion:

Patch:
diff --git a/airflow/auth/managers/base_auth_manager.py b/airflow/auth/managers/base_auth_manager.py
index 4d5c249235..7bb4e92889 100644
--- a/airflow/auth/managers/base_auth_manager.py
+++ b/airflow/auth/managers/base_auth_manager.py
@@ -237,7 +237,7 @@ class BaseAuthManager(LoggingMixin):
 
     @abstractmethod
     def is_authorized_custom_view(
-        self, *, method: ResourceMethod, resource_name: str, user: BaseUser | None = None
+        self, *, method: ResourceMethod | str, resource_name: str, user: BaseUser | None = None
     ):
         """
         Return whether the user is authorized to perform a given action on a custom view.
@@ -246,7 +246,10 @@ class BaseAuthManager(LoggingMixin):
         the auth manager is used as part of the environment. It can also be a view defined as part of a
         plugin defined by a user.
 
-        :param method: the method to perform
+        :param method: the method to perform.
+            The method can also be a string if the action has been defined in a plugin.
+            In that case, the action can be anything (e.g. can_do).
+            See https://github.com/apache/airflow/issues/39144
         :param resource_name: the name of the resource
         :param user: the user to perform the action on. If not provided (or None), it uses the current user
         """
diff --git a/airflow/providers/amazon/aws/auth_manager/avp/entities.py b/airflow/providers/amazon/aws/auth_manager/avp/entities.py
index f2c6376729..8c2e8855b8 100644
--- a/airflow/providers/amazon/aws/auth_manager/avp/entities.py
+++ b/airflow/providers/amazon/aws/auth_manager/avp/entities.py
@@ -55,7 +55,7 @@ def get_entity_type(resource_type: AvpEntities) -> str:
     return AVP_PREFIX_ENTITIES + resource_type.value
 
 
-def get_action_id(resource_type: AvpEntities, method: ResourceMethod):
+def get_action_id(resource_type: AvpEntities, method: ResourceMethod | str):
     """
     Return action id.
 
diff --git a/airflow/providers/amazon/aws/auth_manager/avp/facade.py b/airflow/providers/amazon/aws/auth_manager/avp/facade.py
index 010531155e..4bb9515004 100644
--- a/airflow/providers/amazon/aws/auth_manager/avp/facade.py
+++ b/airflow/providers/amazon/aws/auth_manager/avp/facade.py
@@ -75,7 +75,7 @@ class AwsAuthManagerAmazonVerifiedPermissionsFacade(LoggingMixin):
     def is_authorized(
         self,
         *,
-        method: ResourceMethod,
+        method: ResourceMethod | str,
         entity_type: AvpEntities,
         user: AwsAuthManagerUser | None,
         entity_id: str | None = None,
@@ -86,7 +86,10 @@ class AwsAuthManagerAmazonVerifiedPermissionsFacade(LoggingMixin):
 
         Check whether the user has permissions to access given resource.
 
-        :param method: the method to perform
+        :param method: the method to perform.
+            The method can also be a string if the action has been defined in a plugin.
+            In that case, the action can be anything (e.g. can_do).
+            See https://github.com/apache/airflow/issues/39144
         :param entity_type: the entity type the user accesses
         :param user: the user
         :param entity_id: the entity ID the user accesses. If not provided, all entities of the type will be
diff --git a/airflow/providers/amazon/aws/auth_manager/aws_auth_manager.py b/airflow/providers/amazon/aws/auth_manager/aws_auth_manager.py
index 57b9f9ea0c..f94e4de691 100644
--- a/airflow/providers/amazon/aws/auth_manager/aws_auth_manager.py
+++ b/airflow/providers/amazon/aws/auth_manager/aws_auth_manager.py
@@ -197,7 +197,7 @@ class AwsAuthManager(BaseAuthManager):
         )
 
     def is_authorized_custom_view(
-        self, *, method: ResourceMethod, resource_name: str, user: BaseUser | None = None
+        self, *, method: ResourceMethod | str, resource_name: str, user: BaseUser | None = None
     ):
         return self.avp_facade.is_authorized(
             method=method,
diff --git a/airflow/providers/fab/auth_manager/fab_auth_manager.py b/airflow/providers/fab/auth_manager/fab_auth_manager.py
index d01b3526bf..547bc626bb 100644
--- a/airflow/providers/fab/auth_manager/fab_auth_manager.py
+++ b/airflow/providers/fab/auth_manager/fab_auth_manager.py
@@ -268,11 +268,14 @@ class FabAuthManager(BaseAuthManager):
         )
 
     def is_authorized_custom_view(
-        self, *, method: ResourceMethod, resource_name: str, user: BaseUser | None = None
+        self, *, method: ResourceMethod | str, resource_name: str, user: BaseUser | None = None
     ):
         if not user:
             user = self.get_user()
-        fab_action_name = get_fab_action_from_method_map()[method]
+        if method in get_fab_action_from_method_map():
+            fab_action_name = get_fab_action_from_method_map()[method]
+        else:
+            fab_action_name = method
         return (fab_action_name, resource_name) in self._get_user_permissions(user)
 
     @provide_session
diff --git a/docs/apache-airflow/installation/prerequisites.rst b/docs/apache-airflow/installation/prerequisites.rst
index 07ccc903c7..c253217a29 100644
--- a/docs/apache-airflow/installation/prerequisites.rst
+++ b/docs/apache-airflow/installation/prerequisites.rst
@@ -28,7 +28,7 @@ Airflow™ is tested with:
   * MySQL: 8.0, `Innovation <https://dev.mysql.com/blog-archive/introducing-mysql-innovation-and-long-term-support-lts-versions>`_
   * SQLite: 3.15.0+
 
-* Kubernetes: 1.25, 1.26, 1.27, 1.28
+* Kubernetes: 1.25, 1.26, 1.27, 1.28, 1.29
 
 The minimum memory required we recommend Airflow to run with is 4GB, but the actual requirements depend
 wildly on the deployment options you have


Test patch:
diff --git a/tests/auth/managers/test_base_auth_manager.py b/tests/auth/managers/test_base_auth_manager.py
index 04191c4838..64d33f6065 100644
--- a/tests/auth/managers/test_base_auth_manager.py
+++ b/tests/auth/managers/test_base_auth_manager.py
@@ -95,7 +95,7 @@ class EmptyAuthManager(BaseAuthManager):
         raise NotImplementedError()
 
     def is_authorized_custom_view(
-        self, *, method: ResourceMethod, resource_name: str, user: BaseUser | None = None
+        self, *, method: ResourceMethod | str, resource_name: str, user: BaseUser | None = None
     ):
         raise NotImplementedError()
 
diff --git a/tests/providers/fab/auth_manager/test_fab_auth_manager.py b/tests/providers/fab/auth_manager/test_fab_auth_manager.py
index 72e63983b5..30ef64b281 100644
--- a/tests/providers/fab/auth_manager/test_fab_auth_manager.py
+++ b/tests/providers/fab/auth_manager/test_fab_auth_manager.py
@@ -392,10 +392,21 @@ class TestFabAuthManager:
                 [(ACTION_CAN_READ, "custom_resource2")],
                 False,
             ),
+            (
+                "DUMMY",
+                "custom_resource",
+                [("DUMMY", "custom_resource")],
+                True,
+            ),
         ],
     )
     def test_is_authorized_custom_view(
-        self, method: ResourceMethod, resource_name: str, user_permissions, expected_result, auth_manager
+        self,
+        method: ResourceMethod | str,
+        resource_name: str,
+        user_permissions,
+        expected_result,
+        auth_manager,
     ):
         user = Mock()
         user.perms = user_permissions