如何为暴露自定义REST API的Airflow插件添加访问控制

问题原因 #

直接使用Airflow官方的security.requires_access装饰器会触发RuntimeError: Working outside of application context，因为Airflow插件加载阶段早于Flask应用的完全初始化，此时没有可用的应用上下文，而该装饰器依赖上下文环境。

解决方案 #

方案一：手动检查用户权限（推荐） #

通过login_required确保用户已认证，再手动验证用户权限，避免上下文依赖问题：

import flask
import http
from airflow.www.app import csrf, login_required
from airflow import plugins_manager
from airflow.security import permissions
from airflow.www.security import get_user
import logging

LOG = logging.getLogger(__name__)

HELLO_BLUEPRINT = flask.Blueprint(
    'hello-api', __name__, url_prefix='/hello/v1/world')


@HELLO_BLUEPRINT.route('/', methods=['POST'])
@csrf.exempt
@login_required
def hello_world() -> flask.Response:
    # 获取当前登录用户
    user = get_user()
    
    # 验证用户是否具备触发DAGRun的权限（和官方API一致）
    if not user.can_access(permissions.ACTION_CAN_CREATE, permissions.RESOURCE_DAG_RUN):
        return flask.Response(
            status=http.HTTPStatus.UNAUTHORIZED,
            response="Unauthorized: Insufficient permissions"
        )
    
    LOG.info('hello_world POST called')
    return flask.Response(status=http.HTTPStatus.OK, response="Hello World !")


class HelloApiPlugin(plugins_manager.AirflowPlugin):
    """Integrates our flask Blueprint."""
    name = 'hello_api'
    flask_blueprints = [HELLO_BLUEPRINT]

• @login_required：强制用户通过Airflow的基础认证
• user.can_access()：验证用户是否拥有指定操作权限和资源，你可以根据API功能替换为其他权限（比如ACTION_CAN_READ + RESOURCE_DAG）

方案二：使用官方requires_access装饰器（延迟导入） #

通过延迟导入装饰器，确保在应用上下文存在时加载：

import flask
import http
from airflow.www.app import csrf
from airflow import plugins_manager
from airflow.security import permissions
import logging

LOG = logging.getLogger(__name__)

HELLO_BLUEPRINT = flask.Blueprint(
    'hello-api', __name__, url_prefix='/hello/v1/world')

def register_routes():
    # 延迟导入装饰器，避免加载时的上下文问题
    from airflow.www.security import requires_access

    @HELLO_BLUEPRINT.route('/', methods=['POST'])
    @csrf.exempt
    @requires_access(action=permissions.ACTION_CAN_CREATE, resource=permissions.RESOURCE_DAG_RUN)
    def hello_world():
        LOG.info('hello_world POST called')
        return flask.Response(status=http.HTTPStatus.OK, response="Hello World !")

register_routes()


class HelloApiPlugin(plugins_manager.AirflowPlugin):
    """Integrates our flask Blueprint."""
    name = 'hello_api'
    flask_blueprints = [HELLO_BLUEPRINT]

这种方式和官方API的权限控制逻辑完全一致，通过延迟导入解决了上下文依赖问题。

权限参数说明 #

所有可用的权限常量都在airflow.security.permissions模块中，常见的包括：

• 操作权限：ACTION_CAN_READ、ACTION_CAN_EDIT、ACTION_CAN_CREATE、ACTION_CAN_DELETE
• 资源类型：RESOURCE_DAG、RESOURCE_DAG_RUN、RESOURCE_TASK_INSTANCE、RESOURCE_USER

内容的提问来源于stack exchange，提问作者Tijo John