无需使用Google SDK，如何在Cloudflare Workers中访问Google CloudTasks API

为什么API密钥认证失败 #

Google Cloud的资源管理类API（包括CloudTasks）不支持API密钥认证，API密钥仅适用于公开访问的API（比如地图、翻译类公开接口），所以必须使用OAuth2服务账号令牌认证。

核心实现流程（无需SDK/CLI） #

整个流程只依赖Cloudflare Workers原生提供的Web Crypto API和fetch，不需要第三方依赖，步骤如下：

1. 前置准备 #

• 从Google Cloud IAM控制台下载服务账号JSON密钥文件，提取client_email（服务账号邮箱）、private_key（私钥内容）两个字段
• 将上述两个字段存入Cloudflare Workers的加密环境变量，避免硬编码泄露
• 给服务账号授予对应CloudTasks资源的操作权限，比如Cloud Tasks Editor角色

2. 生成JWT并交换OAuth2访问令牌 #

可直接运行的代码片段：

// 从环境变量读取服务账号信息
const SERVICE_ACCOUNT_EMAIL = env.SERVICE_ACCOUNT_EMAIL
const SERVICE_ACCOUNT_PRIVATE_KEY = env.SERVICE_ACCOUNT_PRIVATE_KEY.replace(/\\n/g, '\n')
// CloudTasks API的权限 scope
const SCOPE = 'https://www.googleapis.com/auth/cloudtasks'

// 导入私钥用于签名
async function importPrivateKey(pem) {
  const pemHeader = '-----BEGIN PRIVATE KEY-----'
  const pemFooter = '-----END PRIVATE KEY-----'
  const pemContents = pem.replace(pemHeader, '').replace(pemFooter, '').replace(/\s/g, '')
  const binaryDer = Uint8Array.from(atob(pemContents), c => c.charCodeAt(0))
  return crypto.subtle.importKey(
    'pkcs8',
    binaryDer.buffer,
    { name: 'RSASSA-PKCS1-v1_5', hash: 'SHA-256' },
    false,
    ['sign']
  )
}

// 生成签名JWT
async function generateJWT() {
  const iat = Math.floor(Date.now() / 1000)
  const exp = iat + 3600 // 令牌有效期1小时
  const header = { alg: 'RS256', typ: 'JWT' }
  const payload = {
    iss: SERVICE_ACCOUNT_EMAIL,
    scope: SCOPE,
    aud: 'https://oauth2.googleapis.com/token',
    iat: iat,
    exp: exp
  }
  const encodedHeader = btoa(JSON.stringify(header)).replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, '')
  const encodedPayload = btoa(JSON.stringify(payload)).replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, '')
  const data = `${encodedHeader}.${encodedPayload}`
  const privateKey = await importPrivateKey(SERVICE_ACCOUNT_PRIVATE_KEY)
  const signature = await crypto.subtle.sign(
    'RSASSA-PKCS1-v1_5',
    privateKey,
    new TextEncoder().encode(data)
  )
  const encodedSignature = btoa(String.fromCharCode(...new Uint8Array(signature)))
    .replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, '')
  return `${encodedHeader}.${encodedPayload}.${encodedSignature}`
}

// 交换获取OAuth2访问令牌
async function getAccessToken() {
  // 可选：将令牌缓存到 Workers KV，减少重复请求
  const jwt = await generateJWT()
  const res = await fetch('https://oauth2.googleapis.com/token', {
    method: 'POST',
    headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
    body: `grant_type=urn:ietf:params:oauth:grant-type:jwt-bearer&assertion=${jwt}`
  })
  const data = await res.json()
  return data.access_token
}

3. 调用CloudTasks API示例 #

拿到访问令牌后，直接在请求头携带即可：

async function createCloudTask(queuePath, taskBody) {
  const accessToken = await getAccessToken()
  const res = await fetch(`https://cloudtasks.googleapis.com/v2/${queuePath}/tasks`, {
    method: 'POST',
    headers: {
      'Authorization': `Bearer ${accessToken}`,
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({ task: taskBody })
  })
  return res.json()
}

常见错误排查 #

• 如果返回401：检查JWT的payload字段是否正确，尤其是iss、aud、scope是否拼写正确，私钥是否完整、换行符是否处理正确
• 如果返回403：检查服务账号是否有对应CloudTasks队列的操作权限，是否开启了CloudTasks API
• 之前使用workers-jwt失败大概率是scope参数填写错误、或者JWT的aud字段没有设置为Google的令牌接口地址

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