从 Google Cloud Build 发出授权请求

Question

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

运行单元测试
在没有流量的情况下部署到云端运行
运行集成测试
迁移云中的流量运行

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

理想情况下，我想使用 Cloud Build 服务帐户来调用 Cloud 运行就像我通常在 AWS 中所做的那样，但我找不到如何从 Cloud 访问它运行纳尔。因此，我改为从 Secret Manager 检索凭证文件。此凭据文件来自新创建的具有 Cloud 运行调用者角色的服务帐户：

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 的 class，并且尝试了 __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}')

知道我做错了什么吗？如果您需要任何澄清，请告诉我。提前致谢！

Answer 1

首先，你的代码太复杂了。如果您想根据运行时环境利用应用程序默认凭证 (ADC)，仅这些行就足够了

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

在Google 云平台上，由于metadata server，将使用环境服务帐户。在您的本地环境中，您需要将环境变量 GOOGLE_APPLICATION_CREDENTIALS 设置为服务帐户密钥文件的路径

注意：您只能使用服务帐户凭据（在 GCP 或您的环境中）生成 id_token，使用您的用户帐户是不可能的

这里的问题是，它在 Cloud Build 上不起作用。我不知道为什么，但无法使用 Cloud Build 元数据服务器生成 id_token。所以，我写了 an article on this 和一个可能的解决方法

从 Google Cloud Build 发出授权请求

Making authorised requests from Google Cloud Build

python

google-cloud-platform

google-cloud-build