Python 2 不再受社区支持。我们建议您将 Python 2 应用迁移到 Python 3

App Identity Python API 概览

地区 ID

REGION_ID 是 Google 根据您在创建应用时选择的地区分配的缩写代码。此代码不对应于国家/地区或省,尽管某些地区 ID 可能类似于常用国家/地区代码和省代码。在 App Engine 网址中包含 REGION_ID.r 对于现有应用是可选项,但在不久后将成为所有新应用的必要项。

为了确保顺利过渡,我们正在逐步更新 App Engine 以使用地区 ID。如果我们尚未更新您的 Google Cloud 项目,则您不会看到应用的地区 ID。由于该 ID 对于现有应用是可选的,因此您在现有应用可以使用地区 ID 后无需更新网址或进行其他更改。

详细了解地区 ID

借助 App Identity API,应用能够找到其应用 ID(也称为项目 ID)。App Engine 应用可使用该 ID 向其他 App Engine 应用、Google API 以及第三方应用和服务声明其身份。应用 ID 也可用于生成网址或电子邮件地址,或者用于做出运行时决策。

获取项目 ID

您可以使用 app_identity.get_application_id() 方法查找项目 ID。WSGI 或 CGI 环境公开了由 API 处理的一些实现细节。

获取应用主机名

默认情况下,App Engine 应用通过 https://PROJECT_ID.REGION_ID.r.appspot.com 形式的网址提供,其中项目 ID 是主机名的一部分。如果应用是通过自定义网域提供的,则可能需要检索整个主机名的组成部分。您可以使用 app_identity.get_default_version_hostname() 方法完成该操作。

向其他 App Engine 应用声明身份

如果要确定向您的 App Engine 应用发出请求的 App Engine 应用的身份,则可以使用请求标头 X-Appengine-Inbound-Appid。此标头由 URLFetch 服务添加到请求中,而且用户无法修改,因此它安全地标明了发出请求的应用的项目 ID(如果存在)。

要求

  • 只有对应用的 appspot.com 网域进行的调用才包含 X-Appengine-Inbound-Appid 标头。对自定义网域的调用不包含该标头。
  • 您的请求必须设置为不遵循重定向。将 urlfetch.fetch() follow_redirects 参数设置为 False

在应用处理程序中,您可以通过读取 X-Appengine-Inbound-Appid 标头并将其与允许发出请求的 ID 列表进行比对来检查传入 ID。例如:

import webapp2

class MainPage(webapp2.RequestHandler):
    allowed_app_ids = [
        'other-app-id',
        'other-app-id-2'
    ]

    def get(self):
        incoming_app_id = self.request.headers.get(
            'X-Appengine-Inbound-Appid', None)

        if incoming_app_id not in self.allowed_app_ids:
            self.abort(403)

        self.response.write('This is a protected page.')

app = webapp2.WSGIApplication([
    ('/', MainPage)
], debug=True)

向 Google API 声明身份

Google API 使用 OAuth 2.0 协议进行身份验证和授权。App Identity API 可以创建 OAuth 令牌,这些令牌可用于声明请求的来源是应用本身。get_access_token() 方法会返回一个范围或范围列表的访问令牌。然后,可以在调用的 HTTP 标头中设置此令牌,以识别调用应用。

以下示例说明了如何使用 App Identity API 对 Cloud Storage API 进行身份验证,并检索项目中所有存储分区的列表。
import json
import logging

from google.appengine.api import app_identity
from google.appengine.api import urlfetch
import webapp2

class MainPage(webapp2.RequestHandler):
    def get(self):
        auth_token, _ = app_identity.get_access_token(
            'https://www.googleapis.com/auth/cloud-platform')
        logging.info(
            'Using token {} to represent identity {}'.format(
                auth_token, app_identity.get_service_account_name()))

        response = urlfetch.fetch(
            'https://www.googleapis.com/storage/v1/b?project={}'.format(
                app_identity.get_application_id()),
            method=urlfetch.GET,
            headers={
                'Authorization': 'Bearer {}'.format(auth_token)
            }
        )

        if response.status_code != 200:
            raise Exception(
                'Call failed. Status code {}. Body {}'.format(
                    response.status_code, response.content))

        result = json.loads(response.content)
        self.response.headers['Content-Type'] = 'application/json'
        self.response.write(json.dumps(result, indent=2))

app = webapp2.WSGIApplication([
    ('/', MainPage)
], debug=True)

请注意,应用的身份由服务帐号名称表示,通常为 applicationid@appspot.gserviceaccount.com。您可以使用 get_service_account_name() 方法获取确切的值。对于提供 ACL 的服务,您可以通过向此帐号授予访问权限来向应用授予访问权限。

向第三方服务声明身份

get_access_token() 生成的令牌仅适用于 Google 服务。但是,您可以使用基础签名技术向其他服务声明应用的身份。sign_blob() 方法将使用应用特有的私钥为字节签名,并且 get_public_certificates() 方法将返回可用于验证签名的证书。

下列示例说明了如何对 blob 签名并验证其签名:

import base64

from Crypto.Hash import SHA256
from Crypto.PublicKey import RSA
from Crypto.Signature import PKCS1_v1_5
from Crypto.Util.asn1 import DerSequence
from google.appengine.api import app_identity
import webapp2

def verify_signature(data, signature, x509_certificate):
    """Verifies a signature using the given x.509 public key certificate."""

    # PyCrypto 2.6 doesn't support x.509 certificates directly, so we'll need
    # to extract the public key from it manually.
    # This code is based on https://github.com/google/oauth2client/blob/master
    # /oauth2client/_pycrypto_crypt.py
    pem_lines = x509_certificate.replace(b' ', b'').split()
    cert_der = base64.urlsafe_b64decode(b''.join(pem_lines[1:-1]))
    cert_seq = DerSequence()
    cert_seq.decode(cert_der)
    tbs_seq = DerSequence()
    tbs_seq.decode(cert_seq[0])
    public_key = RSA.importKey(tbs_seq[6])

    signer = PKCS1_v1_5.new(public_key)
    digest = SHA256.new(data)

    return signer.verify(digest, signature)

def verify_signed_by_app(data, signature):
    """Checks the signature and data against all currently valid certificates
    for the application."""
    public_certificates = app_identity.get_public_certificates()

    for cert in public_certificates:
        if verify_signature(data, signature, cert.x509_certificate_pem):
            return True

    return False

class MainPage(webapp2.RequestHandler):
    def get(self):
        message = 'Hello, world!'
        signing_key_name, signature = app_identity.sign_blob(message)
        verified = verify_signed_by_app(message, signature)

        self.response.content_type = 'text/plain'
        self.response.write('Message: {}\n'.format(message))
        self.response.write(
            'Signature: {}\n'.format(base64.b64encode(signature)))
        self.response.write('Verified: {}\n'.format(verified))

app = webapp2.WSGIApplication([
    ('/', MainPage)
], debug=True)

获取默认的 Cloud Storage 存储分区名称

每个应用都可以有一个默认的 Cloud Storage 存储分区,其中包括 5 GB 免费存储空间和免费的 I/O 操作配额

要获取默认存储分区的名称,可以使用 App Identity API。调用 google.appengine.api.app_identity.app_identity.get_default_gcs_bucket_name