从GoogleCloudBuild发出授权请求

我正在尝试在 Google Cloud Build 中设置部署街道。为此，我想：

1. 运行单元测试
2. 无流量部署到 Cloud Run
3. 运行集成测试
4. 在 Cloud Run 中迁移流量

我已经完成了大部分设置，但我的集成测试包括对 Cloud Run 的几次调用，以验证经过身份验证的调用返回 200 和未经身份验证的返回 401。我遇到的困难是从 Cloud Build 发出签名请求。手动部署和运行集成测试时，它们可以工作，但不能通过 Cloud Build。

理想情况下，我想像在 AWS 中一样使用 Cloud Build 服务帐户来调用 Cloud Run，但我找不到如何从 Cloud Runner 访问它。因此，我从 Secret Manager 中检索凭证文件。此凭据文件来自新创建的具有 Cloud Run Invoker 角色的服务帐户：

steps:
  - name: gcr.io/cloud-builders/gcloud
    id: get-github-ssh-secret
    entrypoint: 'bash'
    args: [ '-c', 'gcloud secrets version access latest --secret=name-of-secret > /root/service-account/credentials.json' ]
    volumes:
      - name: 'service-account'
        path: /root/service-account
...
  - name: python:3.8.7
    id: integration-tests
    entrypoint: /bin/sh
    args:
      - '-c'
      - |-
        if [ $_STAGE != "prod" ]; then 
          python -m pip install -r requirements.txt
          python -m pytest test/integration --disable-warnings ; 
        fi
    volumes:
      - name: 'service-account'
        path: /root/service-account

对于集成测试，我创建了一个名为 Authorizer 的类，__get_authorized_header_for_cloud_build并且__get_authorized_header_for_cloud_build2尝试：

import json
import time
import urllib
from typing import Optional

import google.auth
import requests
from google import auth
from google.auth.transport.requests import AuthorizedSession
from google.oauth2 import service_account
import jwt


class Authorizer(object):
    cloudbuild_credential_path = "/root/service-account/credentials.json"

    # Permissions to request for Access Token
    scopes = ["https://www.googleapis.com/auth/cloud-platform"]

    def get_authorized_header(self, receiving_service_url) -> dict:
        auth_header = self.__get_authorized_header_for_current_user() 
                      or self.__get_authorized_header_for_cloud_build(receiving_service_url)
        return auth_header

    def __get_authorized_header_for_current_user(self) -> Optional[dict]:
        credentials, _ = auth.default()
        auth_req = google.auth.transport.requests.Request()
        credentials.refresh(auth_req)
        if hasattr(credentials, "id_token"):
            authorized_header = {"Authorization": f'Bearer {credentials.id_token}'}
            auth_req.session.close()
            print("Got auth header for current user with auth.default()")
            return authorized_header

    def __get_authorized_header_for_cloud_build2(self, receiving_service_url) -> dict:
        credentials = service_account.Credentials.from_service_account_file(
            self.cloudbuild_credential_path, scopes=self.scopes)
        auth_req = google.auth.transport.requests.Request()
        credentials.refresh(auth_req)
        return {"Authorization": f'Bearer {credentials.token}'}

    def __get_authorized_header_for_cloud_build(self, receiving_service_url) -> dict:
        with open(self.cloudbuild_credential_path, 'r') as f:
            data = f.read()
        credentials_json = json.loads(data)

        signed_jwt = self.__create_signed_jwt(credentials_json, receiving_service_url)
        token = self.__exchange_jwt_for_token(signed_jwt)
        return {"Authorization": f'Bearer {token}'}

    def __create_signed_jwt(self, credentials_json, run_service_url):
        iat = time.time()
        exp = iat + 3600
        payload = {
            'iss': credentials_json['client_email'],
            'sub': credentials_json['client_email'],
            'target_audience': run_service_url,
            'aud': 'https://www.googleapis.com/oauth2/v4/token',
            'iat': iat,
            'exp': exp
        }
        additional_headers = {
            'kid': credentials_json['private_key_id']
        }
        signed_jwt = jwt.encode(
            payload,
            credentials_json['private_key'],
            headers=additional_headers,
            algorithm='RS256'
        )
        return signed_jwt

    def __exchange_jwt_for_token(self, signed_jwt):
        body = {
            'grant_type': 'urn:ietf:params:oauth:grant-type:jwt-bearer',
            'assertion': signed_jwt
        }
        token_request = requests.post(
            url='https://www.googleapis.com/oauth2/v4/token',
            headers={
                'Content-Type': 'application/x-www-form-urlencoded'
            },
            data=urllib.parse.urlencode(body)
        )
        return token_request.json()['id_token']

因此，在本地运行时，__get_authorized_header_for_current_user正在使用和工作。在 Cloud Build 中运行时，__get_authorized_header_for_cloud_build使用。但即使暂时禁用__get_authorized_header_for_current_user并让 cloudbuild_credential_path 引用我本地电脑上的 json 文件，它也会不断收到 401。即使我从凭据文件所有者权限中授予服务帐户。另一种尝试是__get_authorized_header_for_cloud_build我尝试通过自己而不是包获得更多令牌，但仍然是 401。

为了完整起见，集成测试看起来有点像这样：

class NameOfViewIntegrationTestCase(unittest.TestCase):
    base_url = "https://**.a.run.app"
    name_of_call_url = base_url + "/name-of-call"

    def setUp(self) -> None:
        self._authorizer = Authorizer()

    def test_name_of_call__authorized__ok_result(self) -> None:
        # Arrange
        url = self.name_of_call_url 

        # Act
        response = requests.post(url, headers=self._authorizer.get_authorized_header(url))

        # Arrange
        self.assertTrue(response.ok, msg=f'{response.status_code}: {response.text}')

知道我在这里做错了什么吗？如果您需要对某些事情进行澄清，请告诉我。提前致谢！

回答

from google.oauth2.id_token import fetch_id_token
from google.auth.transport import requests
r = requests.Request()
print(fetch_id_token(r,"<AUDIENCE>"))