适用于旧版捆绑服务的 App Identity API

区域 ID

REGION_ID 是 Google 根据您在创建应用时选择的区域分配的缩写代码。此代码不对应于国家/地区或省，尽管某些区域 ID 可能类似于常用国家/地区代码和省代码。对于 2020 年 2 月以后创建的应用，REGION_ID.r 包含在 App Engine 网址中。对于在此日期之前创建的现有应用，网址中的区域 ID 是可选的。

详细了解区域 ID。

确定

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

此 API 支持第一代运行时，并可在升级到相应的第二代运行时后使用。如果要更新到 App Engine Python 3 运行时，请参阅迁移指南以了解旧版捆绑服务的迁移选项。

获取项目 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 进行身份验证，并检索项目中所有存储桶的列表。

注意：Google 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)

请注意，应用的身份由服务账号名称表示，通常为 [e-mail]。您可以使用 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。