如何使用模拟服务账号运行Python代码访问GCP资源？

问题背景 #

我在谷歌搜索没找到易懂的方案，特求助：已知gsutil -i service-account-id ls -p project-id可以模拟服务账号访问GCP资源，但不知道怎么在Python代码里实现。具体场景：

• Project-A的Pub/Sub主题归xyz@gmail.com所有
• 我的Python代码部署在abc@gmail.com所有的Project-B中
• Project-A已创建服务账号，已授予abc@gmail.com该账号的模拟权限，且该服务账号拥有Pub/Sub Admin角色
• 需要无密钥修改现有Pub/Sub发布代码，访问Project-A的主题

实现方案 #

要实现服务账号模拟，不需要额外安装特殊包（google-cloud-pubsub已经依赖所需的身份验证库），核心是基于当前环境的默认凭据，创建一个模拟目标服务账号的凭据，再用这个凭据初始化Pub/Sub客户端。

关键步骤 #

• 利用google.auth获取当前环境的默认凭据（比如本地gcloud登录凭据、Cloud Run/GCE的内置服务账号凭据）
• 通过google.auth.impersonated_credentials.Credentials创建模拟凭据，指定要模拟的Project-A服务账号邮箱
• 用模拟凭据初始化pubsub_v1.PublisherClient

修改后的完整代码 #

"""Publishes multiple messages to a Pub/Sub topic with an error handler (使用服务账号模拟)"""
import os
from collections.abc import Callable
from concurrent import futures
from google.cloud import pubsub_v1
from google.auth import default
from google.auth.impersonated_credentials import Credentials

# 配置参数
PROJECT_A_ID = "project-a的项目ID"
TOPIC_ID = "myTopic1"
# 要模拟的Project-A服务账号邮箱
IMPERSONATE_SERVICE_ACCOUNT = "project-a的服务账号邮箱@project-a.iam.gserviceaccount.com"

# 获取当前环境的默认凭据（无需密钥文件）
base_credentials, _ = default()

# 创建模拟凭据
impersonated_creds = Credentials(
    source_credentials=base_credentials,
    target_principal=IMPERSONATE_SERVICE_ACCOUNT,
    # 所需的Pub/Scopes权限范围，根据需求调整
    target_scopes=[
        "https://www.googleapis.com/auth/pubsub"
    ],
    lifetime=3600  # 凭据有效期，最长3600秒
)

# 使用模拟凭据初始化PublisherClient
publisher = pubsub_v1.PublisherClient(credentials=impersonated_creds)
topic_path = publisher.topic_path(PROJECT_A_ID, TOPIC_ID)
publish_futures = []

def get_callback(publish_future: pubsub_v1.publisher.futures.Future, data: str) -> Callable[[pubsub_v1.publisher.futures.Future], None]:
    def callback(publish_future: pubsub_v1.publisher.futures.Future) -> None:
        try:
            # 等待60秒确认发布成功
            print(f"消息发布结果: {publish_future.result(timeout=60)}")
        except futures.TimeoutError:
            print(f"发布消息 {data} 超时")
    return callback

for i in range(4):
    data = str(i)
    publish_future = publisher.publish(topic_path, data.encode("utf-8"))
    publish_future.add_done_callback(get_callback(publish_future, data))
    publish_futures.append(publish_future)

# 等待所有发布任务完成
futures.wait(publish_futures, return_when=futures.ALL_COMPLETED)

print(f"已向 {topic_path} 发布消息（含错误处理）")

注意事项 #

• 确保当前环境的默认凭据（比如Project-B的服务账号或本地gcloud登录的abc@gmail.com）拥有模拟Project-A服务账号的权限（即roles/iam.serviceAccountTokenCreator角色）
• 模拟凭据的有效期最长为3600秒，若需要长期运行，可按需重新创建
• 目标服务账号必须拥有访问Project-A Pub/Sub主题的权限（已满足Pub/Sub Admin角色）

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