Use permission constants (apache#11389)

Use constants for permission resource and action names.

airflow/api_connexion/endpoints/config_endpoint.py

@@ -20,6 +20,7 @@
 from airflow.api_connexion import security
 from airflow.api_connexion.schemas.config_schema import Config, ConfigOption, ConfigSection, config_schema
 from airflow.configuration import conf
+from airflow.security import permissions
 from airflow.settings import json
 
 LINE_SEP = '\n'  # `\n` cannot appear in f-strings
@@ -61,7 +62,7 @@ def _config_to_json(config: Config) -> str:
     return json.dumps(config_schema.dump(config), indent=4)
 
 
-@security.requires_access([("can_read", "Config")])
+@security.requires_access([(permissions.ACTION_CAN_READ, permissions.RESOURCE_CONFIG)])
 def get_config() -> Response:
     """
     Get current configuration.

airflow/api_connexion/endpoints/connection_endpoint.py

@@ -30,10 +30,11 @@
     connection_schema,
 )
 from airflow.models import Connection
+from airflow.security import permissions
 from airflow.utils.session import provide_session
 
 
-@security.requires_access([("can_delete", "Connection")])
+@security.requires_access([(permissions.ACTION_CAN_DELETE, permissions.RESOURCE_CONNECTION)])
 @provide_session
 def delete_connection(connection_id, session):
     """
@@ -49,7 +50,7 @@ def delete_connection(connection_id, session):
     return NoContent, 204
 
 
-@security.requires_access([("can_read", "Connection")])
+@security.requires_access([(permissions.ACTION_CAN_READ, permissions.RESOURCE_CONNECTION)])
 @provide_session
 def get_connection(connection_id, session):
     """
@@ -64,7 +65,7 @@ def get_connection(connection_id, session):
     return connection_collection_item_schema.dump(connection)
 
 
-@security.requires_access([("can_read", "Connection")])
+@security.requires_access([(permissions.ACTION_CAN_READ, permissions.RESOURCE_CONNECTION)])
 @format_parameters({'limit': check_limit})
 @provide_session
 def get_connections(session, limit, offset=0):
@@ -79,7 +80,7 @@ def get_connections(session, limit, offset=0):
     )
 
 
-@security.requires_access([("can_edit", "Connection")])
+@security.requires_access([(permissions.ACTION_CAN_EDIT, permissions.RESOURCE_CONNECTION)])
 @provide_session
 def patch_connection(connection_id, session, update_mask=None):
     """
@@ -115,7 +116,7 @@ def patch_connection(connection_id, session, update_mask=None):
     return connection_schema.dump(connection)
 
 
-@security.requires_access([("can_create", "Connection")])
+@security.requires_access([(permissions.ACTION_CAN_CREATE, permissions.RESOURCE_CONNECTION)])
 @provide_session
 def post_connection(session):
     """

airflow/api_connexion/endpoints/dag_endpoint.py

@@ -32,7 +32,7 @@
 from airflow.utils.session import provide_session
 
 
-@security.requires_access([("can_read", permissions.RESOURCE_DAGS)])
+@security.requires_access([(permissions.ACTION_CAN_READ, permissions.RESOURCE_DAGS)])
 @provide_session
 def get_dag(dag_id, session):
     """
@@ -46,7 +46,7 @@ def get_dag(dag_id, session):
     return dag_schema.dump(dag)
 
 
-@security.requires_access([("can_read", permissions.RESOURCE_DAGS)])
+@security.requires_access([(permissions.ACTION_CAN_READ, permissions.RESOURCE_DAGS)])
 def get_dag_details(dag_id):
     """
     Get details of DAG.
@@ -57,7 +57,7 @@ def get_dag_details(dag_id):
     return dag_detail_schema.dump(dag)
 
 
-@security.requires_access([("can_read", permissions.RESOURCE_DAGS)])
+@security.requires_access([(permissions.ACTION_CAN_READ, permissions.RESOURCE_DAGS)])
 @format_parameters({'limit': check_limit})
 def get_dags(limit, offset=0):
     """
@@ -70,7 +70,7 @@ def get_dags(limit, offset=0):
     return dags_collection_schema.dump(DAGCollection(dags=dags, total_entries=total_entries))
 
 
-@security.requires_access([("can_edit", permissions.RESOURCE_DAGS)])
+@security.requires_access([(permissions.ACTION_CAN_EDIT, permissions.RESOURCE_DAGS)])
 @provide_session
 def patch_dag(session, dag_id, update_mask=None):
     """

airflow/api_connexion/endpoints/dag_run_endpoint.py

@@ -33,7 +33,12 @@
 from airflow.utils.types import DagRunType
 
 
-@security.requires_access([("can_read", permissions.RESOURCE_DAGS), ("can_delete", "DagRun")])
+@security.requires_access(
+    [
+        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAGS),
+        (permissions.ACTION_CAN_DELETE, permissions.RESOURCE_DAG_RUN),
+    ]
+)
 @provide_session
 def delete_dag_run(dag_id, dag_run_id, session):
     """
@@ -44,7 +49,12 @@ def delete_dag_run(dag_id, dag_run_id, session):
     return NoContent, 204
 
 
-@security.requires_access([("can_read", permissions.RESOURCE_DAGS), ("can_read", "DagRun")])
+@security.requires_access(
+    [
+        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAGS),
+        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_RUN),
+    ]
+)
 @provide_session
 def get_dag_run(dag_id, dag_run_id, session):
     """
@@ -59,7 +69,12 @@ def get_dag_run(dag_id, dag_run_id, session):
     return dagrun_schema.dump(dag_run)
 
 
-@security.requires_access([("can_read", permissions.RESOURCE_DAGS), ("can_read", "DagRun")])
+@security.requires_access(
+    [
+        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAGS),
+        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_RUN),
+    ]
+)
 @format_parameters(
     {
         'start_date_gte': format_datetime,
@@ -158,7 +173,12 @@ def _apply_date_filters_to_query(
     return query
 
 
-@security.requires_access([("can_read", permissions.RESOURCE_DAGS), ("can_read", "DagRun")])
+@security.requires_access(
+    [
+        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAGS),
+        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_RUN),
+    ]
+)
 @provide_session
 def get_dag_runs_batch(session):
     """
@@ -194,7 +214,12 @@ def get_dag_runs_batch(session):
     return dagrun_collection_schema.dump(DAGRunCollection(dag_runs=dag_runs, total_entries=total_entries))
 
 
-@security.requires_access([("can_read", permissions.RESOURCE_DAGS), ("can_create", "DagRun")])
+@security.requires_access(
+    [
+        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAGS),
+        (permissions.ACTION_CAN_CREATE, permissions.RESOURCE_DAG_RUN),
+    ]
+)
 @provide_session
 def post_dag_run(dag_id, session):
     """

airflow/api_connexion/endpoints/dag_source_endpoint.py

@@ -23,11 +23,13 @@
 from airflow.api_connexion.exceptions import NotFound
 from airflow.api_connexion.schemas.dag_source_schema import dag_source_schema
 from airflow.models.dagcode import DagCode
+from airflow.security import permissions
+
 
 log = logging.getLogger(__name__)
 
 
-@security.requires_access([("can_read", "DagCode")])
+@security.requires_access([(permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_CODE)])
 def get_dag_source(file_token: str):
     """
     Get source code using file token

airflow/api_connexion/endpoints/event_log_endpoint.py

@@ -27,10 +27,11 @@
     event_log_schema,
 )
 from airflow.models import Log
+from airflow.security import permissions
 from airflow.utils.session import provide_session
 
 
-@security.requires_access([('can_read', 'Log')])
+@security.requires_access([(permissions.ACTION_CAN_READ, permissions.RESOURCE_LOG)])
 @provide_session
 def get_event_log(event_log_id, session):
     """
@@ -42,7 +43,7 @@ def get_event_log(event_log_id, session):
     return event_log_schema.dump(event_log)
 
 
-@security.requires_access([('can_read', 'Log')])
+@security.requires_access([(permissions.ACTION_CAN_READ, permissions.RESOURCE_LOG)])
 @format_parameters({'limit': check_limit})
 @provide_session
 def get_event_logs(session, limit, offset=None):

airflow/api_connexion/endpoints/extra_link_endpoint.py

@@ -29,10 +29,10 @@
 
 @security.requires_access(
     [
-        ('can_read', permissions.RESOURCE_DAGS),
-        ('can_read', 'DagRun'),
-        ('can_read', 'Task'),
-        ('can_read', 'TaskInstance'),
+        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAGS),
+        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_RUN),
+        (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK),
+        (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK_INSTANCE),
     ]
 )
 @provide_session

airflow/api_connexion/endpoints/import_error_endpoint.py

@@ -26,10 +26,11 @@
     import_error_schema,
 )
 from airflow.models.errors import ImportError  # pylint: disable=redefined-builtin
+from airflow.security import permissions
 from airflow.utils.session import provide_session
 
 
-@security.requires_access([('can_read', 'ImportError')])
+@security.requires_access([(permissions.ACTION_CAN_READ, permissions.RESOURCE_IMPORT_ERROR)])
 @provide_session
 def get_import_error(import_error_id, session):
     """
@@ -45,7 +46,7 @@ def get_import_error(import_error_id, session):
     return import_error_schema.dump(error)
 
 
-@security.requires_access([('can_read', 'ImportError')])
+@security.requires_access([(permissions.ACTION_CAN_READ, permissions.RESOURCE_IMPORT_ERROR)])
 @format_parameters({'limit': check_limit})
 @provide_session
 def get_import_errors(session, limit, offset=None):

airflow/api_connexion/endpoints/log_endpoint.py

@@ -29,7 +29,11 @@
 
 
 @security.requires_access(
-    [('can_read', permissions.RESOURCE_DAGS), ('can_read', 'DagRun'), ('can_read', 'Task')]
+    [
+        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAGS),
+        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_RUN),
+        (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK),
+    ]
 )
 @provide_session
 def get_log(session, dag_id, dag_run_id, task_id, task_try_number, full_content=False, token=None):

airflow/api_connexion/endpoints/pool_endpoint.py

@@ -24,10 +24,11 @@
 from airflow.api_connexion.parameters import check_limit, format_parameters
 from airflow.api_connexion.schemas.pool_schema import PoolCollection, pool_collection_schema, pool_schema
 from airflow.models.pool import Pool
+from airflow.security import permissions
 from airflow.utils.session import provide_session
 
 
-@security.requires_access([("can_delete", "Pool")])
+@security.requires_access([(permissions.ACTION_CAN_DELETE, permissions.RESOURCE_POOL)])
 @provide_session
 def delete_pool(pool_name: str, session):
     """
@@ -41,7 +42,7 @@ def delete_pool(pool_name: str, session):
         return Response(status=204)
 
 
-@security.requires_access([("can_read", "Pool")])
+@security.requires_access([(permissions.ACTION_CAN_READ, permissions.RESOURCE_POOL)])
 @provide_session
 def get_pool(pool_name, session):
     """
@@ -53,7 +54,7 @@ def get_pool(pool_name, session):
     return pool_schema.dump(obj)
 
 
-@security.requires_access([("can_read", "Pool")])
+@security.requires_access([(permissions.ACTION_CAN_READ, permissions.RESOURCE_POOL)])
 @format_parameters({'limit': check_limit})
 @provide_session
 def get_pools(session, limit, offset=None):
@@ -65,7 +66,7 @@ def get_pools(session, limit, offset=None):
     return pool_collection_schema.dump(PoolCollection(pools=pools, total_entries=total_entries))
 
 
-@security.requires_access([("can_edit", "Pool")])
+@security.requires_access([(permissions.ACTION_CAN_EDIT, permissions.RESOURCE_POOL)])
 @provide_session
 def patch_pool(pool_name, session, update_mask=None):
     """
@@ -116,7 +117,7 @@ def patch_pool(pool_name, session, update_mask=None):
     return pool_schema.dump(pool)
 
 
-@security.requires_access([("can_create", "Pool")])
+@security.requires_access([(permissions.ACTION_CAN_CREATE, permissions.RESOURCE_POOL)])
 @provide_session
 def post_pool(session):
     """

airflow/api_connexion/endpoints/task_endpoint.py

@@ -24,7 +24,12 @@
 from airflow.security import permissions
 
 
-@security.requires_access([("can_read", permissions.RESOURCE_DAGS), ("can_read", "Task")])
+@security.requires_access(
+    [
+        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAGS),
+        (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK),
+    ]
+)
 def get_task(dag_id, task_id):
     """
     Get simplified representation of a task.
@@ -40,7 +45,12 @@ def get_task(dag_id, task_id):
     return task_schema.dump(task)
 
 
-@security.requires_access([("can_read", permissions.RESOURCE_DAGS), ("can_read", "Task")])
+@security.requires_access(
+    [
+        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAGS),
+        (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK),
+    ]
+)
 def get_tasks(dag_id):
     """
     Get tasks for DAG