如何在 Google Cloud Run 中生成 Blob 签名 URL?

2022-01-25 00:00:00 python google-cloud-storage google-cloud-run google-iam

问题描述

在 Google Cloud Run 下，您可以选择您的容器正在运行的服务帐号.使用默认计算服务帐号无法生成签名 url.

Under Google Cloud Run, you can select which service account your container is running. Using the default compute service account fails to generate a signed url.

此处列出的解决方法适用于 Google Cloud Compute - 如果您允许服务帐户的所有范围.在 Cloud Run 中似乎没有办法做到这一点(我找不到).

The work around listed here works on Google Cloud Compute -- if you allow all the scopes for the service account. There does not seem to be away to do that in Cloud Run (not that I can find).

https://github.com/googleapis/google-auth-library-python/issues/50

我尝试过的事情:

  1. 为服务帐号分配角色:roles/iam.serviceAccountTokenCreator
  2. 在虚拟机中验证相同 GCP 项目中的解决方法(与 Cloud Run 相比)
  3. 使用从私钥加载的服务帐户(通过 json 文件)验证代码在容器中本地运行.

from google.cloud import storage
client = storage.Client()
bucket = client.get_bucket('EXAMPLE_BUCKET')
blob = bucket.get_blob('libraries/image_1.png')
expires = datetime.now() + timedelta(seconds=86400)
blob.generate_signed_url(expiration=expires)

失败:

you need a private key to sign credentials.the credentials you are currently using <class 'google.auth.compute_engine.credentials.Credentials'> just contains a token. see https://googleapis.dev/python/google-api-core/latest/auth.html#setting-up-a-service-account for more details.
/usr/local/lib/python3.8/site-packages/google/cloud/storage/_signing.py, line 51, in ensure_signed_credentials

尝试添加解决方法，

Error calling the IAM signBytes API: 
{  "error": {  "code": 400,

    "message": "Request contains an invalid argument.",
    "status": "INVALID_ARGUMENT"  }
}
Exception Location: /usr/local/lib/python3.8/site-packages/google/auth/iam.py, line 81, in _make_signing_request

Github 问题中提到的解决方法代码:

Workaround code as mention in Github issue:

from google.cloud import storage
from google.auth.transport import requests
from google.auth import compute_engine
from datetime import datetime, timedelta

def get_signing_creds(credentials):
    auth_request = requests.Request()
    print(credentials.service_account_email)
    signing_credentials = compute_engine.IDTokenCredentials(auth_request, "", service_account_email=credentials.ser
vice_account_email)
    return signing_credentials


client = storage.Client()
bucket = client.get_bucket('EXAMPLE_BUCKET')
blob = bucket.get_blob('libraries/image_1.png')
expires = datetime.now() + timedelta(seconds=86400)
signing_creds = get_signing_creds(client._credentials)
url = blob.generate_signed_url(expiration=expires, credentials=signing_creds)
print(url)

如何在 Google Cloud Run 下生成签名 url?在这一点上，我可能不得不挂载我想避免的服务帐户密钥.

How do I generate a signed url under Google Cloud Run? At this point, it seems like I may have to mount the service account key which I wanted to avoid.

尝试澄清一下，服务帐户具有正确的权限 - 它在 GCE 和本地使用 JSON 私钥工作.

To try and clarify, the service account has the correct permissions - it works in GCE and locally with the JSON private key.

解决方案

是的，你可以，但我必须深入了解如何(如果你不关心细节，请跳到最后)

Yes you can, but I had to deep dive to find how (jump to the end if you don't care about the details)

如果您进入 _signing.py文件，第623行，可以看到这个

If you go in the _signing.py file, line 623, you can see this

if access_token and service_account_email:
   signature = _sign_message(string_to_sign, access_token, service_account_email)
...

如果您提供 access_token 和 service_account_email，则可以使用 _sign_message 方法.此方法使用 IAM 服务 SignBlob API 在这一行

If you provide the access_token and the service_account_email, you can use the _sign_message method. This method uses the IAM service SignBlob API at this line

这很重要，因为您现在可以在没有本地私钥的情况下对 blob 进行签名！所以，这解决了问题，下面的代码在 Cloud Run 上工作(我确定在 Cloud Function 上)

It's important because you can now sign blob without having locally the private key!! So, that solves the problem, and the following code works on Cloud Run (and I'm sure on Cloud Function)

def sign_url():
    from google.cloud import storage
    from datetime import datetime, timedelta

    import google.auth
    credentials, project_id = google.auth.default()

    # Perform a refresh request to get the access token of the current credentials (Else, it's None)
    from google.auth.transport import requests
    r = requests.Request()
    credentials.refresh(r)

    client = storage.Client()
    bucket = client.get_bucket('EXAMPLE_BUCKET')
    blob = bucket.get_blob('libraries/image_1.png')
    expires = datetime.now() + timedelta(seconds=86400)

    # In case of user credential use, define manually the service account to use (for development purpose only)
    service_account_email = "YOUR DEV SERVICE ACCOUNT"
    # If you use a service account credential, you can use the embedded email
    if hasattr(credentials, "service_account_email"):
        service_account_email = credentials.service_account_email

    url = blob.generate_signed_url(expiration=expires,service_account_email=service_account_email, access_token=credentials.token)
    return url, 200

如果不清楚，请告诉我

相关文章