App Identity Python API 概览

借助 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 应用通过 http://<your_app_id>.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 操作配额。此存储分区的最大存储空间为 5 GB,通过为应用启用结算功能并购买额外的存储空间使其成为付费存储分区,您可以增加其存储空间。

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

此页内容是否有用?请给出您的反馈和评价:

发送以下问题的反馈:

此网页
适用于 Python 2 的 App Engine 标准环境