如何在Airflow中处理多因素认证（MFA）以解决S3文件列表访问被拒问题

云原生阿岚

2026-4-29

嘿，这个问题我之前帮不少同行解决过——AWS MFA启用后Airflow的S3 DAG报Access Denied，核心原因是现在访问AWS资源必须携带临时MFA凭证，而默认的静态AK/SK已经不满足权限要求了。下面是几种实用的解决办法，你可以根据自己的Airflow部署情况来选：

方法1：用Airflow的AWS Connection配置MFA（推荐）

Airflow本身支持在AWS连接里直接配置MFA，不需要改DAG代码，步骤很简单：

打开Airflow UI，进入Admin > Connections，找到你正在使用的AWS连接（或者新建一个）

在Extra字段里填入以下JSON格式的配置：

{
  "aws_access_key_id": "你的AWS访问密钥ID",
  "aws_secret_access_key": "你的AWS秘密访问密钥",
  "mfa_serial": "arn:aws:iam::你的AWS账号ID:mfa/你的MFA设备名称",
  "region_name": "你的S3桶所在区域，比如us-east-1"
}

保存后，你的S3Hook或者AwsHook会自动读取这个配置，调用AWS STS服务获取临时凭证，就能正常访问S3了。

方法2：在DAG代码中手动获取临时MFA凭证

如果不想修改Airflow连接配置，也可以在DAG任务里手动实现凭证获取逻辑：

先确保Airflow Worker所在环境能访问AWS STS服务，并且你的AK/SK拥有sts:GetSessionToken权限

在任务函数里添加以下代码：

import boto3
from airflow.models import Variable

def list_s3_files():
    # 从Airflow变量或安全存储中读取MFA设备ARN
    mfa_serial = Variable.get("aws_mfa_serial")
    # 如果是自动运行的DAG，这里要从Secret Manager或支持API的MFA设备获取令牌，不能手动输入
    mfa_token = input("请输入MFA验证码：")
    
    # 获取临时凭证
    sts_client = boto3.client('sts')
    cred_response = sts_client.get_session_token(
        SerialNumber=mfa_serial,
        TokenCode=mfa_token
    )
    temp_creds = cred_response['Credentials']
    
    # 用临时凭证初始化S3客户端并列出文件
    s3_client = boto3.client(
        's3',
        aws_access_key_id=temp_creds['AccessKeyId'],
        aws_secret_access_key=temp_creds['SecretAccessKey'],
        aws_session_token=temp_creds['SessionToken']
    )
    bucket_contents = s3_client.list_objects_v2(Bucket="你的S3桶名称")
    print([obj['Key'] for obj in bucket_contents.get('Contents', [])])

注意：如果是无人值守的自动DAG，手动输入MFA令牌肯定不行，这时候建议用支持API调用的MFA设备，或者把令牌存储在AWS Secrets Manager里让DAG自动读取。

方法3：IAM角色+MFA强制策略（生产环境首选）

如果是生产环境，更安全的做法是给Airflow的执行角色绑定MFA强制访问策略：

创建一个IAM策略，拒绝未携带MFA的S3访问请求：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Deny",
      "Action": "s3:*",
      "Resource": "arn:aws:s3:::你的S3桶名称/*",
      "Condition": {
        "BoolIfExists": {
          "aws:MultiFactorAuthPresent": "false"
        }
      }
    }
  ]
}

给Airflow使用的IAM角色附加这个策略，同时允许该角色调用sts:AssumeRole并携带MFA参数
在Airflow的AWS连接里配置role_arn和mfa_serial，钩子会自动完成角色假设并获取带MFA的临时凭证

常见坑点提醒

确保你的MFA设备处于激活状态，输入的令牌是最新的（MFA令牌每30秒刷新一次）
旧版本的Airflow（比如1.x）的AwsHook可能不支持MFA配置，建议升级到2.x以上版本
检查IAM权限：无论是AK/SK还是IAM角色，都必须拥有sts:GetSessionToken或sts:AssumeRole的权限，否则无法获取临时凭证

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