Dataproc集群用Python发Pub/Sub消息遇ACCESS_TOKEN_SCOPE_INSUFFICIENT错误

错误信息 #

raise exceptions.from_grpc_error(exc) from exc
google.api_core.exceptions.PermissionDenied: 403 Request had insufficient authentication scopes. [reason: "ACCESS_TOKEN_SCOPE_INSUFFICIENT"
domain: "googleapis.com"
metadata {
  key: "method"
  value: "google.pubsub.v1.Publisher.Publish"
}

metadata {
  key: "service"
  value: "pubsub.googleapis.com"
}
]

问题详情 #

• 已为Dataproc集群分配具备Pub/Sub发布权限的服务账号，但运行Python代码发布消息时触发上述403错误
• 相同服务账号在Cloud Function中可正常运行
• 临时使用服务账号密钥文件能解决问题，但存在密钥暴露风险；尝试通过Secret Manager获取密钥也因权限问题报403

当前代码 #

认证代码

service_account_credentials = {"""  hidden for security reasons lol """} 

credentials = service_account.Credentials.from_service_account_info(
service_account_credentials)

发布代码

class EmailPublisher:
    def __init__(self, project_id: str, topic_id: str, credentials):
        self.publisher = pubsub_v1.PublisherClient(credentials=credentials)
        self.topic_path = self.publisher.topic_path(project_id, topic_id)

    def publish_message(self, message: str):
        data = str(message).encode("utf-8")
        future = self.publisher.publish(
        self.topic_path, data, origin="dataproc-python-pipeline", username="gcp"
        )
        logging.info(future.result())
        logging.info("Published messages with custom attributes to %s", self.topic_path)

解决方案 #

1. 调整Dataproc集群的访问范围（Scopes） #

Dataproc集群的服务账号访问范围会限制令牌能访问的服务，即使IAM权限足够，Scope不足也会触发ACCESS_TOKEN_SCOPE_INSUFFICIENT错误：

• 创建集群时：添加Pub/Sub的访问范围https://www.googleapis.com/auth/pubsub，可搭配默认范围default使用
• 已有集群更新：通过gcloud命令更新集群范围（需重启节点生效）：

gcloud dataproc clusters update <CLUSTER_NAME> \
    --region <REGION> \
    --scopes=https://www.googleapis.com/auth/pubsub,default

2. 使用集群默认应用凭据（无需手动加载密钥） #

GCP客户端库会自动从Dataproc环境中获取服务账号凭据，只要Scope和IAM权限配置正确，无需手动指定密钥：
修改发布代码，移除手动凭据初始化逻辑：

class EmailPublisher:
    def __init__(self, project_id: str, topic_id: str):
        # 自动获取集群关联服务账号的凭据
        self.publisher = pubsub_v1.PublisherClient()
        self.topic_path = self.publisher.topic_path(project_id, topic_id)

    def publish_message(self, message: str):
        data = str(message).encode("utf-8")
        future = self.publisher.publish(
            self.topic_path, data, origin="dataproc-python-pipeline", username="gcp"
        )
        logging.info(future.result())
        logging.info("Published messages with custom attributes to %s", self.topic_path)

3. 验证服务账号的IAM权限 #

确认集群绑定的服务账号已拥有roles/pubsub.publisher角色：

• 进入GCP IAM控制台，找到目标服务账号，检查是否已添加该角色
• 角色作用范围可指定为目标Pub/Sub主题或整个项目，按需选择

4. 修复Secret Manager访问权限（若需使用密钥） #

如果必须通过Secret Manager获取密钥，需完成两项配置：

• 为集群服务账号绑定roles/secretmanager.secretAccessor角色
• 确保集群访问范围包含https://www.googleapis.com/auth/cloud-platform（该范围覆盖Secret Manager权限）

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