mar*_*gas 5 google-app-engine google-cloud-storage google-app-engine-python
我正在使用App Engine编写新的应用程序,并且正如文档建议不使用Blobstore API,我正在使用Google云端存储客户端(GCS).一切都很好,但我希望能够将"签名的URL"返回给客户,这样他们就可以获得GCS资源而无需通过应用程序.我相信这是signet网址的用途.
但是如何测试呢?我可以成功地从客户端测试GCS调用,但我不知道如何使用urlfetch测试客户端的HTTP调用.
以下是一个完整的测试用例,说明了我的问题:
import base64
import mimetypes
import urllib
import urllib2
from datetime import datetime, timedelta
import time
from google.appengine.api import app_identity
from google.appengine.datastore import datastore_stub_util
from google.appengine.ext import testbed
from google.appengine.ext import ndb
import unittest
import cloudstorage
# IS THIS RIGHT ?
GCS_API_ACCESS_ENDPOINT = 'http://localhost:8000/_ah/gcs'
def sign_url(bucket_object, expires_after_seconds=60):
""" cloudstorage signed url to download cloudstorage object without login
Docs : https://cloud.google.com/storage/docs/access-control?hl=bg#Signed-URLs
API : https://cloud.google.com/storage/docs/reference-methods?hl=bg#getobject
"""
# source: https://github.com/voscausa/appengine-gcs-signed-url/blob/05b8a93e2777679d40af62cc5ffce933216e6a85/sign_url.py
method = 'GET'
gcs_filename = urllib.quote(bucket_object)
content_md5, content_type = None, None
# expiration : number of seconds since epoch
expiration_dt = datetime.utcnow() + timedelta(seconds=expires_after_seconds)
expiration = int(time.mktime(expiration_dt.timetuple()))
# Generate the string to sign.
signature_string = '\n'.join([
method,
content_md5 or '',
content_type or '',
str(expiration),
gcs_filename])
signature_bytes = app_identity.sign_blob(signature_string)[1]
google_access_id = app_identity.get_service_account_name()
# Set the right query parameters. we use a gae service account for the id
query_params = {'GoogleAccessId': google_access_id,
'Expires': str(expiration),
'Signature': base64.b64encode(signature_bytes)}
# Return the built URL.
result = '{endpoint}{resource}?{querystring}'.format(endpoint=GCS_API_ACCESS_ENDPOINT,
resource=gcs_filename,
querystring=urllib.urlencode(query_params))
return result
FILE_DATA = "This is file contents."
MIME = "text/plain"
class TestGCS(unittest.TestCase):
def setUp(self):
self.testbed = testbed.Testbed()
self.testbed.activate()
self.policy = datastore_stub_util.PseudoRandomHRConsistencyPolicy(probability=0)
self.testbed.init_datastore_v3_stub(consistency_policy=self.policy)
self.testbed.init_app_identity_stub()
self.testbed.init_memcache_stub()
self.testbed.init_urlfetch_stub()
self.testbed.init_blobstore_stub()
ndb.get_context().clear_cache()
def tearDown(self):
self.testbed.deactivate()
def test_gcs_works(self):
with cloudstorage.open('/mybucket/test.txt', 'w', content_type=MIME) as f:
f.write(FILE_DATA)
with cloudstorage.open('/mybucket/test.txt', 'r') as f:
data = f.read()
print(data)
self.assertEqual(data, FILE_DATA)
def test_signurl(self):
url = sign_url('/mybucket/test.txt')
# FIXME: Not yet working as we have no idea on how to access local GCS during the test.
result = urllib2.urlopen(url)
self.assertEqual(200, result.code)
self.assertEqual(FILE_DATA, result.read())
Run Code Online (Sandbox Code Playgroud)
您可以在 SDK 中测试 GCS 和 service_accounts,但是当您使用签名网址时,您没有本地 appengine GCS 服务。
但您可以使用服务帐户和谷歌云服务测试您的本地应用程序。
服务帐户可以非常轻松地向其他 Google API 和服务授权 appengine 请求。
要在 appengine SDK 中使用服务帐户,您必须在运行开发服务器时添加两个未记录的选项:
--appidentity_email_address=<SERVICE_ACCOUNT_EMAIL_ADDRESS>
--appidentity_private_key_path=<PEM_KEY_PATH>
此文档问题请求中的更多信息
您可以在appengine 云项目的开发人员控制台权限部分创建或查找服务帐户。
您可以为服务帐户创建并下载 p12 密钥。
使用 OpenSSL 将此 p12 密钥转换为 RSA pem 密钥。
我在 Windows 上使用了这个 OpenSSL 安装程序。
要在 Windows 中创建 pem 密钥文件,请使用:
openssl pkcs12 -in <P12_KEY_PATH> -nocerts -nodes -passin pass:notasecret | openssl rsa -out <PEM_KEY_PATH>
现在,您可以在开发服务器中使用您的云应用服务帐户,并使用app_identity对请求进行签名和授权。