我们开发了一个自动化管道，主要在 AWS 上执行任务，然后在 Google Cloud 上执行一些下游工作。这些任务通过 AWS StepFunctions/Lambda 部署在 AWS 上，我们需要将处理过的文件从 AWS 传递到 Google Cloud Storage（通过 Google Transfer Service）。但是，我在将 AWS 和 GCP 部分连接在一起时遇到了麻烦。
我有一个用于实现谷歌传输服务的 Lambda 函数（通过谷歌的 python 客户端库），但是我不断收到错误消息：
module initialization error: Could not automatically determine
credentials. Please set GOOGLE_APPLICATION_CREDENTIALS or
explicitly create credential and re-run the application. For more
information, please see
https://developers.google.com/accounts/docs/application-default-
credentials.
我的 S3 到 GCS Lambda 函数（我称之为“切换”）中有一个方法可以设置 GOOGLE_APPLICATION_CREDENTIALS 环境变量，但它显然不起作用。
这是 Handoff Lambda 函数：
"""
Creates a one-time transfer from Amazon S3 to Google Cloud Storage.
"""
import requests
import boto3
from botocore.client import ClientError
from requests.exceptions import RequestException
from google.cloud import storage
import googleapiclient.discovery
import logging
import uuid
import os
import base64
EVENT_GCP_CREDS_KEY = 'gcp_creds_b64_cypher_text'
GCP_CREDENTIALS_FILE_NAME = 'service_creds.json'
# Establish clients
kms = boto3.client('kms')
storagetransfer = googleapiclient.discovery.build('storagetransfer', 'v1')
def handler(event, context):
description = "-".join("transfer-job", event['queue'])
project_id = event['project_id']
source_bucket = event['results_uri'] + "final-cohort-vcf/"
sink_bucket = event['sink_bucket']
include_prefix = event['cohort_prefix'] + ".gt.vcf.gz"
access_key = event['aws_access_key']
secret_access_key = event['aws_secret_key']
return gcp_storage_read_op_verification(event)
now = datetime.datetime.now()
day = now.day
month = now.month
year = now.year
# Add 7 hours because the time has to be in UTC
hours = now.hour + 7
minutes = now.minute + 2
transfer_job = {
'description': description,
'status': 'ENABLED',
'projectId': project_id,
'schedule': {
'scheduleStartDate': {
'day': day,
'month': month,
'year': year
},
'scheduleEndDate': {
'day': day,
'month': month,
'year': year
},
'startTimeOfDay': {
'hours': hours,
'minutes': minutes
}
},
'transferSpec': {
'objectConditions': {
'includePrefixes': [
include_prefix
]
},
'awsS3DataSource': {
'bucketName': source_bucket,
'awsAccessKey': {
'accessKeyId': access_key,
'secretAccessKey': secret_access_key
}
},
'gcsDataSink': {
'bucketName': sink_bucket
}
}
}
result = storagetransfer.transferJobs().create(body=transfer_job).execute()
print('Returned transferJob: {}'.format(
json.dumps(result, indent=4)))
return result
def _gcp_credentials_hack(event):
"""
A hack to enable GCP client usage via Application Default Credentials. Uses an encoded, encrypted string of the
GCP service account JSON from the event object. Has the side effecst of
1. Writing the credentials in plaintext to /tmp/service_creds.json
2. Referencing this location in environment variable
:param event: Passed in by the container, expecting to find key: gcp_creds_b64_cypher_text
:return: None
"""
# Get blob from event
cypher_text_blob = event['GCP_creds']
# Decode cypher_text from base64, into bytes
cypher_text_bytes = base64.b64decode(cypher_text_blob)
# Call KMS to decrypt GCP credentials cypher text
# Permisssions for this action should be given by a policy attached this function's invocation role
decrypt_response = kms.decrypt(CiphertextBlob=cypher_text_bytes)
# Process the plaintext from the result object from kms.decrypt(..) into a utf-8 encoded bytes object
gcp_credentials_bytes = decrypt_response['Plaintext']
# Decode the utf8-encoded bytes object into a Python str of th GCP credentials
gcp_credentials = gcp_credentials_bytes.decode('utf-8')
# Write the credentials in plaintext to /tmp
# - Can we gaurantee that only approved agents can read this file?
# YES. See comment.
gcp_credentials_file_local = os.path.sep.join(('/tmp', GCP_CREDENTIALS_FILE_NAME))
logger.debug('writing credentials to {}'.format(gcp_credentials_file_local))
with open(gcp_credentials_file_local, 'w') as credentials_fh:
credentials_fh.write(gcp_credentials)
# Set Application Default Credentials environment path to tmp file we created
os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = gcp_credentials_file_local
def _get_blob_from_gcs_public_landsat_bucket():
"""
Create Google Cloud Storage client and use it to return a blob from the public landsat image bucket.
:return: blob retrieved from public landsat bucket using Google Cloud Storage client library
"""
# Create GCP storage client instance, using Application Default Credentials (hack)
client = storage.Client()
# Use GCP storage client to get public landsat bucket object
public_landsat_bucket = client.get_bucket(bucket_name='gcp-public-data-landsat')
# return the first blob in the bucket
for blob in public_landsat_bucket.list_blobs():
return blob
def _get_blob_from_hail_project_bucket():
"""
Create Google Cloud Storage client and use it to return a blob from the public landsat image bucket.
:return: blob retrieved from public landsat bucket using Google Cloud Storage client library
"""
# Create GCP storage client instance, using Application Default Credentials (hack)
client = storage.Client()
# Use GCP storage client to get public landsat bucket object
public_landsat_bucket = client.get_bucket(bucket_name='liftover-results')
# return the first blob in the bucket
for blob in public_landsat_bucket.list_blobs():
return blob
def gcp_storage_read_op_verification(event):
"""
:param event: event from handler that contains the encrypted GCP service credentials
:return: first blob id from GCS public landsat bucket
"""
try:
_gcp_credentials_hack(event=event)
# If credentials hack succeeded, this will return a blob from the GCS plublic landsat public using GCS
# python client
landsat_blob = _get_blob_from_gcs_public_landsat_bucket()
liftover_blob = _get_blob_from_hail_project_bucket()
# return dict of blob.id for response jsonification
return {
"gcp-public-data-landsat-random-blob-id": landsat_blob.id,
"gcp-private-hail-bucket-random-blob-id": liftover_blob.id,
}
except BaseException as e:
# Very broad exception class.. using it for time being, until have opportunity to find the fine-grained
# exceptions that can be raised from the statement suite
logger.error(e)
return FAIL_TOKEN
Appears to be safe. See https://forums.aws.amazon.com/message.jspa?messageID=761306
"Thanks for reaching out to us. I believe what you are referring to is Lambda container re-usage. All the
resources associated with a Lambda container, including the /tmp storage, are isolated from other Lambda
containers.
https://aws.amazon.com/blogs/compute/container-reuse-in-lambda/
Indeed there is the possibility of reusing the same Lambda container if your function is executed at close
time intervals. Important to note is that the container will only be reused for your particular Lambda
function, other functions from your account or different accounts will run in other isolated containers.
So there is a 1-to-1 association between a Lambda function and its container.
After a certain time interval of a Lambda function not being invoked, its associated container is deleted
and in the process all the data stored in the memory or disk associated with the container is destroyed as
well."
...
2. With the client library bootstrapped, use the storage client to retrieve the public landsat image
bucket, and return a blob.
3. Finally return that blob's id, as proof of the execution.
:param event: event from handler that contains the encrypted GCP service credentials
:return: first blob id from GCS public landsat bucket
"""
# Use encrypted GCP storage crendentials from event to enable file-based Application Default Credentials.
# This is somewhat of a bad hack since they must exist in plaintext on a filesystem. Although, see above, this
# is supposed to be private between containers.
#
# TODO: If using this hack often, would be better to delete the plaintext creds file after the GCP API call
# Possibly using a context manager. E.g.
#
# with gcp_creds_hack:
# _get_blob
#
# Implemented like
#
# from contextlib import contextmanager
#
# @contextmanager
# def gcp_creds_hack(file):
# # write the plaintext creds file, file, to /tmp
# yield
# # delete the plaintext creds file, file, from /tmp
#
try:
_gcp_credentials_hack(event=event)
# If credentials hack succeeded, this will return a blob from the GCS plublic landsat public using GCS
# python client
landsat_blob = _get_blob_from_gcs_public_landsat_bucket()
liftover_blob = _get_blob_from_hail_project_bucket()
# return dict of blob.id for response jsonification
return {
"gcp-public-data-landsat-random-blob-id": landsat_blob.id,
"gcp-private-hail-bucket-random-blob-id": liftover_blob.id,
}
except BaseException as e:
# Very broad exception class.. using it for time being, until have opportunity to find the fine-grained
# exceptions that can be raised from the statement suite
logger.error(e)
return FAIL_TOKEN
在事件对象中传递的 GCP 凭据源于管道用户输入到服务帐户 json 文件的本地文件路径。然后我使用一个函数来读取这个 json 文件并通过以下脚本对其进行加密：
import boto3
import base64
class LambdaModeUseCases(object):
@staticmethod
def encrypt_gcp_creds(gcp_creds_file, key_id):
# get GCP creds as string
with open(gcp_creds_file, 'r') as gcp_creds_fh:
gcp_creds = gcp_creds_fh.read().replace('\n', '')
# kms:Encrypt (need key_id or arn)
kms_client = boto3.client('kms')
encryption_response = {}
try:
encryption_response = kms_client.encrypt(
KeyId=key_id,
Plaintext=gcp_creds.encode('utf-8'),
)
except ClientError as ce:
print('Failed calling kms.encrypt with key {key} on text {text}'.format(key=key_id, text=gcp_creds))
raise ce
print('Parsing kms_client.encrypt(..) response')
cypher_text_blob = 'FAILED'
if 'CiphertextBlob' in encryption_response:
cypher_text_blob = encryption_response['CiphertextBlob']
else:
print('Wait able to call kms.encrypt(..) without a botocore.client.ClientError.')
print('But failed to find CiphertextBlob in response: {response}'.format(
response=encryption_response
))
raise Exception('Failed to find CiphertextBlob in kms.encrypt() response.')
print('Base64 encoding...')
encrypted_gcp_creds_b64 = base64.b64encode(cypher_text_blob)
print('b64: {}'.format(encrypted_gcp_creds_b64))
encrypted_gcp_creds_str = encrypted_gcp_creds_b64.decode("utf-8")
print('str: {}'.format(encrypted_gcp_creds_str))
return encrypted_gcp_creds_str
这个加密的凭证对象随后被传递到 Handoff Lambda 函数中。
关于出了什么问题或更简化的方法的任何想法？