Google Vault API导出blob下载报403错误的实现方案咨询

你遇到的可枚举Bucket、Blob列表但操作Blob返回403的问题，优先排查权限配置问题：

• 枚举Bucket、列举Blob属于桶级操作，仅需storage.buckets.get、storage.objects.list权限即可完成，你当前的授权已经覆盖这部分能力
• 调用blob.exists()、下载Blob属于对象级操作，如果你操作的是Google Vault自动导出生成的存储桶，这类桶默认使用对象级ACL控制访问，哪怕给账号授予存储管理员角色，也会因为没有单个对象的读取权限被拦截。直接到Bucket设置页开启统一Bucket级访问权限，等待1-2分钟权限全局生效后再重试，这步可以解决绝大多数同类403问题。

你不需要额外改造代码就能用用户账号认证，官方示例中提到的gcloud auth application-default login本身就支持用户账号身份：

1. 本地终端执行认证命令：

   gcloud auth application-default login
   

2. 在弹出的浏览器窗口中登录对目标存储桶有权限的用户账号，完成授权流程
3. 代码中无需传入服务账号密钥，直接初始化客户端即可自动读取本地的用户账号凭证，写法和你贴的示例完全一致：

   from google.cloud import storage
   # 自动加载本地gcloud生成的用户认证凭证
   gcp_client = storage.Client()
   

如果需要在代码中显式指定用户账号凭证、不依赖gcloud生成的默认凭证，可以参考如下实现，提前安装依赖pip install google-auth-oauthlib：

from google.cloud import storage
from google_auth_oauthlib.flow import InstalledAppFlow

# 替换为你在GCP控制台生成的OAuth客户端密钥文件路径
auth_flow = InstalledAppFlow.from_client_secrets_file(
    "your_oauth_client_secret.json",
    scopes=["https://www.googleapis.com/auth/devstorage.full_control"]
)
# 拉起本地浏览器完成用户登录授权
auth_flow.run_local_server(port=8080)
user_credentials = auth_flow.credentials

# 传入用户凭证初始化存储客户端，替换为你自己的项目ID
gcp_client = storage.Client(
    credentials=user_credentials,
    project="your-gcp-project-id"
)

你贴的官方示例存在缩进错误、缺少依赖导入的问题，修正后的版本如下：

import os
from google.cloud import storage
from googleapiclient.discovery import build

def download_exports(vault_service, matter_id):
    """
    Google Cloud storage service is authenticated by running
    `gcloud auth application-default login` and expects a billing enabled project
    in ENV variable `GOOGLE_CLOUD_PROJECT`
    """
    gcp_client = storage.Client()
    matter_id = os.environ['MATTERID']
    vault_service = build('vault', 'v1', credentials=gcp_client._credentials)
    resp = vault_service.matters().exports().list(matterId=matter_id).execute()
    
    for export in resp.get('exports', []):
        if 'cloudStorageSink' not in export:
            continue
        directory = export['name']
        if not os.path.exists(directory):
            os.makedirs(directory)
        print(export['id'])
        for sink_file in export['cloudStorageSink']['files']:
            file_name = f"{directory}/{sink_file['objectName'].split('/')[-1]}"
            object_uri = f"gs://{sink_file['bucketName']}/{sink_file['objectName']}"
            print(f'get {object_uri} to {file_name}')
            with open(file_name, 'wb+') as f:
                gcp_client.download_blob_to_file(object_uri, f)

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