我想创建一个到存储桶的预签名上传 URL,并希望避免显式引用 json 密钥。
目前,我正在尝试使用默认 App Engine 服务帐户来执行此操作
我正在尝试遵循这个答案,但收到此错误:
AttributeError:您需要私钥来签署凭据。 您当前使用的凭据
仅包含 令牌。看 https://googleapis.dev/python/google-api-core/latest/auth.html#setting-up-a-service-account 了解更多详情。
我的云函数代码如下所示:
from google.cloud import storage
import datetime
import google.auth
def generate_upload_url(blob_name, additional_metadata: dict = {}):
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("my_bucket")
blob = bucket.blob(blob_name)
service_account_email = credentials.service_account_email
print(f"attempting to create signed url for {service_account_email}")
url = blob.generate_signed_url(
version="v4",
service_account_email=service_account_email,
access_token=credentials.token,
# This URL is valid for 120 minutes
expiration=datetime.timedelta(minutes=120),
# Allow PUT requests using this URL.
method="PUT",
content_type="application/octet-stream",
)
return url
def get_upload_url(request):
blob_name = get_param(request, "blob_name")
url = generate_upload_url(blob_name)
return url
当您使用签名 URL 的 v4 版本时,方法的第一行调用
ensure_signed_credentials
方法来检查当前服务帐户是否可以在独立模式下生成签名(因此使用私钥)。所以,这打破了当前的行为。
在函数的注释中,清楚地描述了需要服务帐户JSON文件
If you are on Google Compute Engine, you can't generate a signed URL.
Follow `Issue 922`_ for updates on this. If you'd like to be able to
generate a signed URL from GCE, you can use a standard service account
from a JSON file rather than a GCE service account.
所以,请使用 v2 版本。
作为替代方案,您还可以使用模拟服务帐户来创建具有临时提升权限的短期凭据。
要求:
IAM Service Account Credentials API
Service Account Token Creator
Service Account User
def get_impersonated_credentials():
scopes=['https://www.googleapis.com/auth/cloud-platform']
credentials, project = auth.default(scopes=scopes)
if credentials.token is None:
credentials.refresh(requests.Request())
signing_credentials = impersonated_credentials.Credentials(
source_credentials=credentials,
target_principal=credentials.service_account_email,
target_scopes=scopes,
lifetime=datetime.timedelta(seconds=3600),
delegates=[credentials.service_account_email]
)
return signing_credentials
然后在生成签名网址的函数中...
...
credentials=get_impersonated_credentials()
url = blob.generate_signed_url(
version="v4",
credentials=credentials,
# This URL is valid for 120 minutes
expiration=datetime.timedelta(minutes=120),
# Allow PUT requests using this URL.
method="PUT",
content_type="application/octet-stream",
...
在Cloud Run上测试成功。学分: