本地单元测试 Google Cloud Storage 签名 URL

Question

我正在使用 App Engine 编写一个新应用程序，并且由于文档建议不要使用 Blobstore API，我正在使用 Google Cloud Storage 客户端 (GCS)。一切都很好，但我希望能够将“签名的 url”返回给客户端，这样他们就可以在不通过应用程序的情况下获取 GCS 资源。我相信这就是签名网址的用途。

但是如何测试呢？我可以成功测试来自客户端的 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())

Answer 1

您可以在 SDK 中测试 GCS 和 service_accounts，但使用签名 url 时没有本地 appengine GCS 服务。

但您可以使用服务帐户和谷歌云服务测试您的本地应用。

服务帐户可以很容易地授权 appengine 对其他 Google API 和服务的请求。

要在 appengine SDK 中使用服务帐户，您必须在 run the development server 时添加两个未记录的选项：

1. --appidentity_email_address=<SERVICE_ACCOUNT_EMAIL_ADDRESS>
2. --appidentity_private_key_path=<PEM_KEY_PATH>

更多信息在this request for documentation issue

您可以在您的 appengine 云项目的developers console permissions section 中创建或查找服务帐号。
您可以为服务帐号创建和下载 p12 密钥。

使用 OpenSSL 将此 p12 密钥转换为 RSA pem 密钥。

我在 Windows 上使用了this OpenSSL installer。

要在 Windows 中创建 pem 密钥文件，请使用：

openssl pkcs12 -in <P12_KEY_PATH> -nocerts -nodes -passin pass:notasecret | openssl rsa -out <PEM_KEY_PATH>

现在您可以在开发服务器中使用您的云应用服务帐号，并使用app_identity 对请求进行签名和授权。