如何限制对airflow.models的访问 实现多租户DagRun元数据权限隔离

核心思路是从身份校验、查询入口拦截、行级数据过滤三个层面做全链路权限控制，避免租户越权访问非自有DAG的运行元数据

具体实现步骤如下：

• 第一步：统一DAG租户标识规则
  要求所有租户上传的DAG必须在配置中添加唯一的tenant_id标签，平台侧可以在DAG入库扫描阶段校验该标签的合法性，防止租户冒充其他身份。
• 第二步：开启Airflow行级权限控制
  Airflow 2.x及以上版本自带基于FAB的行级安全（Row Level Security）能力，你可以直接给DagRun、DAG等核心表配置过滤规则，绑定租户角色后，用户无论通过UI、API还是代码调用查询DagRun数据，都会自动过滤掉非所属租户的记录，越权查询时会返回空结果或权限报错。
  如果你使用的是1.x版本无行级权限能力，可以自行封装统一的元数据查询工具类，所有DagRun相关查询必须走该工具类，工具类内部强制添加tenant_id过滤条件，禁止租户直接调用原生的DagRun查询接口。
• 第三步：改造自定义宏函数，拦截非法查询
  你原本的宏函数允许租户传入任意DAG对象，存在越权风险，改造后直接从执行上下文取当前运行的DAG，并增加租户身份校验，改造后的示例代码如下：

  from airflow.exceptions import AirflowSecurityException
  
  def get_current_dag_last_run_end_date(**context):
      # 从执行上下文获取当前运行的DAG，禁止用户传入任意DAG对象
      current_dag = context["dag"]
      # 获取当前执行用户所属租户、当前DAG绑定的租户
      current_user_tenant = context["user"].extra.get("tenant_id")
      dag_tenant = current_dag.tags.get("tenant_id")
      # 租户身份校验不通过直接抛权限异常
      if not current_user_tenant or not dag_tenant or current_user_tenant != dag_tenant:
          raise AirflowSecurityException("无权限访问该DAG的运行元数据")
      last_dag_run = current_dag.get_last_dagrun()
      return last_dag_run.end_date if last_dag_run else None
  

• 第四步：物理隔离DAG目录
  给每个租户分配独立的DAG上传目录，限制租户的文件访问权限，仅能修改自有目录下的DAG文件，避免租户篡改DAG的tenant_id标签伪造身份。
• 第五步：禁止直接访问元数据库
  关闭租户自定义代码访问Airflow元数据库的权限，所有元数据查询必须走平台封装的安全接口或宏，避免租户直接执行SQL拉取全量DagRun数据。

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