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

问题描述

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

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

我已经完成了大部分设置，但我的集成测试包括对 Cloud Run 的几次调用，以验证经过身份验证的调用返回 200 和未经身份验证的返回 401。我遇到的困难是从 Cloud Build 发出签名请求。在手动部署和运行集成测试时，它们可以工作，但不能来自 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我尝试自己获取令牌而不是包裹，但仍然是 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}')

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

标签： pythongoogle-cloud-platformgoogle-cloud-build

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

问题描述

解决方案

推荐阅读