使用Python通过JWT获取Google BigQuery访问令牌遇401未授权错误

问题根源 #

你的代码存在几个关键错误：

1. 直接使用JWT令牌访问BigQuery API：签名后的JWT不是最终的OAuth2访问令牌，需要先向Google的token端点换取正式的access_token
2. Audience参数错误：生成JWT时aud应该设置为Google的token端点，而不是BigQuery的权限scope
3. 请求参数位置错误：grant_type应该放在请求体中，而非请求头里

修正后的代码 #

import google.auth.jwt as jwt
import google.auth.crypt
import time
import json
import requests

now = int(time.time())
sa_email = '你的服务账号邮箱'  # 填写完整的服务账号邮箱，比如xxx@project-id.iam.gserviceaccount.com
sa_keyfile = 'key.json'  # 确保该文件路径正确

# 修正payload：aud设置为token端点，scope指定BigQuery权限
payload = {
    'iat': now,
    'exp': now + 3600,
    'iss': sa_email,
    'aud': 'https://oauth2.googleapis.com/token',  # 正确的audience
    'scope': 'https://www.googleapis.com/auth/bigquery',  # 指定所需权限
    'sub': sa_email
}

# 签名生成JWT
signer = google.auth.crypt.RSASigner.from_service_account_file(sa_keyfile)
jwt_token = google.auth.jwt.encode(signer, payload)

# 向token端点请求获取access_token
token_url = 'https://oauth2.googleapis.com/token'
data = {
    'grant_type': 'urn:ietf:params:oauth:grant-type:jwt-bearer',
    'assertion': jwt_token.decode('utf-8')
}
response = requests.post(token_url, data=data)
response.raise_for_status()  # 检查获取token是否成功
token_data = response.json()
access_token = token_data['access_token']
print('获取到的access_token:', access_token)

# 使用access_token访问BigQuery API
headers = {
    'Authorization': f'Bearer {access_token}',
    'Content-type': 'application/json'
}
bigquery_url = 'https://www.googleapis.com/bigquery/v2/projects'
bq_response = requests.get(bigquery_url, headers=headers)
bq_response.raise_for_status()
print('BigQuery API响应:', bq_response.json())

额外检查项 #

• 确保服务账号邮箱填写完整，没有遗漏
• 确认key.json是从Google Cloud控制台下载的服务账号密钥文件
• 给该服务账号分配BigQuery数据查看者或更高权限（在Google Cloud IAM中配置）
• 检查本地时间是否准确，JWT的iat和exp依赖正确的时间戳

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