如何在multiprocessing Pool的每个进程中输入MFA代码并完成认证

问题根源 #

1. boto3客户端跨进程不安全：主进程创建的s3_client会被复制到子进程中，但boto3客户端内部维护的连接状态、认证信息等无法在进程间安全共享，会引发请求异常。
2. 多进程抢占输入流：两个子进程同时请求MFA输入，终端的标准输入被多个进程抢占，导致输入的MFA代码无法被正确读取，引发认证失败和崩溃。

解决思路 #

在主进程中一次性获取MFA代码，生成带MFA认证的会话凭证，然后让每个子进程使用该凭证独立创建boto3会话和客户端，避免进程间共享客户端，同时只需要用户输入一次MFA。

修改后的代码 #

import multiprocessing
import boto3

def load_to_s3(args):
    file_path, mfa_serial, mfa_token, bucket, target_path = args
    # 子进程内创建独立的boto3会话，使用主进程传入的MFA凭证
    session = boto3.Session()
    # 用MFA令牌获取临时凭证
    sts_client = session.client('sts')
    response = sts_client.get_session_token(
        SerialNumber=mfa_serial,
        TokenCode=mfa_token
    )
    # 使用临时凭证创建S3客户端
    s3_client = boto3.client(
        's3',
        aws_access_key_id=response['Credentials']['AccessKeyId'],
        aws_secret_access_key=response['Credentials']['SecretAccessKey'],
        aws_session_token=response['Credentials']['SessionToken']
    )
    # 执行上传
    s3_client.upload_file(file_path, bucket, target_path)
    return f"上传成功: {file_path}"

if __name__ == '__main__':
    # 配置参数（替换为你的实际信息）
    bucket = "your-bucket-name"
    target_path = "target/path/"
    file_path_chunks = ["file1.txt", "file2.txt"]  # 你的文件列表
    mfa_serial = "arn:aws:iam::123456789012:mfa/your-mfa-device-name"  # 你的MFA设备ARN

    # 主进程中一次性获取MFA代码
    mfa_token = input("请输入MFA代码: ")

    # 构造每个子进程的参数，包含MFA信息
    task_args = [(fp, mfa_serial, mfa_token, bucket, target_path) for fp in file_path_chunks]

    # 创建进程池执行任务
    with multiprocessing.Pool(processes=2) as pool:
        response_list = pool.map(load_to_s3, task_args)
    
    for res in response_list:
        print(res)

关键说明 #

• 主进程统一获取MFA：只需要用户输入一次MFA代码，避免多进程抢占输入流的问题。
• 子进程独立创建客户端：每个子进程都基于临时凭证创建自己的S3客户端，保证进程间无状态共享，避免跨进程的客户端安全问题。
• 临时凭证有效期：STS临时凭证默认有效期为12小时，足够完成大部分上传任务，如果需要更长时间可以在get_session_token中指定DurationSeconds参数（最大不超过43200秒）。

替代方案：使用线程池 #

如果你的上传任务以IO密集型为主，也可以考虑使用concurrent.futures.ThreadPoolExecutor替代进程池。因为boto3客户端是线程安全的，主进程创建带MFA认证的客户端后，线程可以安全复用，且只需要输入一次MFA：

import concurrent.futures
import boto3

def load_to_s3(file_path):
    s3_client.upload_file(file_path, bucket, target_path)
    return f"上传成功: {file_path}"

if __name__ == '__main__':
    bucket = "your-bucket-name"
    target_path = "target/path/"
    file_path_chunks = ["file1.txt", "file2.txt"]
    mfa_serial = "arn:aws:iam::123456789012:mfa/your-mfa-device-name"

    # 主进程创建带MFA认证的会话和客户端
    session = boto3.Session()
    sts_client = session.client('sts')
    mfa_token = input("请输入MFA代码: ")
    creds = sts_client.get_session_token(SerialNumber=mfa_serial, TokenCode=mfa_token)['Credentials']
    s3_client = boto3.client(
        's3',
        aws_access_key_id=creds['AccessKeyId'],
        aws_secret_access_key=creds['SecretAccessKey'],
        aws_session_token=creds['SessionToken']
    )

    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
        results = executor.map(load_to_s3, file_path_chunks)
    
    for res in results:
        print(res)

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