python - 从 Google Cloud Build 发出授权请求

Question

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

1. 运行单元测试
2. 在没有流量的情况下部署到 Cloud Run
3. 运行集成测试
4. 在 Cloud Run 中迁移流量

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

理想情况下，我想使用 Cloud Build 服务帐户来调用 Cloud Run，就像我通常在 AWS 中所做的那样，但我找不到如何从 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 引用我本地 pc 上的 json 文件，它仍然会收到 401。即使我从凭据文件所有者权限中授予服务帐户。另一种尝试是 __get_authorized_header_for_cloud_build ，我尝试自己更多地获取 token 而不是包，但仍然是 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 Cloud Platform 上，由于 metadata server，将使用环境服务帐户.在您的本地环境中，您需要将环境变量 GOOGLE_APPLICATION_CREDENTIALS 设置为服务帐户 key 文件的路径

注意:您只能使用服务帐户凭据(在 GCP 上或您的环境中)生成 id_token，而不能使用您的用户帐户生成 id_token

---

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