Source code for google.appengine.api.app_identity.app_identity

#!/usr/bin/env python
#
# Copyright 2007 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#





"""Provides access functions for the app identity service.

To learn more about the App Identity API, review the `Overview`_ document.

.. _Overview:
   https://cloud.google.com/appengine/docs/python/appidentity/
"""









import os
import time

from google.appengine.api import apiproxy_stub_map
from google.appengine.api import memcache
from google.appengine.api.app_identity import app_identity_service_pb
from google.appengine.runtime import apiproxy_errors

__all__ = ['BackendDeadlineExceeded',
           'BlobSizeTooLarge',
           'InternalError',
           'InvalidScope',
           'NotAllowed',
           'OperationNotImplemented',
           'Error',
           'create_rpc',
           'make_sign_blob_call',
           'make_get_public_certificates_call',
           'make_get_service_account_name_call',
           'sign_blob',
           'get_public_certificates',
           'PublicCertificate',
           'get_service_account_name',
           'get_application_id',
           'get_default_version_hostname',
           'get_access_token',
           'get_access_token_uncached',
           'make_get_access_token_call',
           'get_default_gcs_bucket_name',
           'make_get_default_gcs_bucket_name_call',
          ]


_APP_IDENTITY_SERVICE_NAME = 'app_identity_service'
_SIGN_FOR_APP_METHOD_NAME = 'SignForApp'
_GET_CERTS_METHOD_NAME = 'GetPublicCertificatesForApp'
_GET_SERVICE_ACCOUNT_NAME_METHOD_NAME = 'GetServiceAccountName'
_GET_DEFAULT_GCS_BUCKET_NAME_METHOD_NAME = 'GetDefaultGcsBucketName'
_GET_ACCESS_TOKEN_METHOD_NAME = 'GetAccessToken'
_PARTITION_SEPARATOR = '~'
_DOMAIN_SEPARATOR = ':'
_MEMCACHE_KEY_PREFIX = '_ah_app_identity_'
_MEMCACHE_NAMESPACE = '_ah_'



_TOKEN_EXPIRY_SAFETY_MARGIN = 300
_MAX_TOKEN_CACHE_SIZE = 100


_MAX_RANDOM_EXPIRY_DELTA = 60




_access_token_cache = {}





_random_cache_expiry_delta = (
    hash(time.time()) % (_MAX_RANDOM_EXPIRY_DELTA * 1000) / 1000.0)


[docs]class Error(Exception): """Base error type."""
[docs]class BackendDeadlineExceeded(Error): """The communication to the backend service timed out."""
[docs]class BlobSizeTooLarge(Error): """The size of the blob to sign is larger than the allowed limit."""
[docs]class InternalError(Error): """An unspecified internal failure occurred."""
[docs]class InvalidScope(Error): """The scope is invalid."""
[docs]class NotAllowed(Error): """The operation is not allowed."""
[docs]class OperationNotImplemented(Error): """The operation is not implemented for the service account."""
def _to_app_identity_error(error): """Translates an application error to an external error, if possible. Args: error: An `ApplicationError` to translate. Returns: error: The App Identity API-specific error message. """ error_map = { app_identity_service_pb.AppIdentityServiceError.NOT_A_VALID_APP: InternalError, app_identity_service_pb.AppIdentityServiceError.DEADLINE_EXCEEDED: BackendDeadlineExceeded, app_identity_service_pb.AppIdentityServiceError.BLOB_TOO_LARGE: BlobSizeTooLarge, app_identity_service_pb.AppIdentityServiceError.UNKNOWN_ERROR: InternalError, app_identity_service_pb.AppIdentityServiceError.UNKNOWN_SCOPE: InvalidScope, app_identity_service_pb.AppIdentityServiceError.NOT_ALLOWED: NotAllowed, app_identity_service_pb.AppIdentityServiceError.NOT_IMPLEMENTED: OperationNotImplemented, } if error.application_error in error_map: return error_map[error.application_error](error.error_detail) else: return InternalError('%s: %s' % (error.application_error, error.error_detail))
[docs]class PublicCertificate(object): """Class that specifies information about a public certificate. Attributes: key_name: Name of the certificate. x509_certificate_pem: `X.509 certificates`_ in PEM format. .. _X.509 certificates: https://www.ietf.org/rfc/rfc2459.txt """ def __init__(self, key_name, x509_certificate_pem): self.key_name = key_name self.x509_certificate_pem = x509_certificate_pem
[docs]def create_rpc(deadline=None, callback=None): """Creates an RPC object for use with the App Identity API. Args: deadline: Optional deadline in seconds for the operation; the default value is a system-specific deadline, typically 5 seconds. callback: Optional callable to invoke on completion. Returns: An `apiproxy_stub_map.UserRPC` object specialized for this service. """ return apiproxy_stub_map.UserRPC(_APP_IDENTITY_SERVICE_NAME, deadline, callback)
[docs]def make_sign_blob_call(rpc, bytes_to_sign): """Executes the RPC call to sign a blob. Args: rpc: A UserRPC instance. bytes_to_sign: Blob that must be signed. Returns: A tuple that contains the signing key name and the signature. Raises: TypeError: If `bytes_to_sign` is not a string. """ if not isinstance(bytes_to_sign, str): raise TypeError('bytes_to_sign must be str: %s' % bytes_to_sign) request = app_identity_service_pb.SignForAppRequest() request.set_bytes_to_sign(bytes_to_sign) response = app_identity_service_pb.SignForAppResponse() def signing_for_app_result(rpc): """Checks success, handles exceptions, and returns the converted RPC result. This method waits for the RPC if it has not yet finished and calls the post-call hooks on the first invocation. Args: rpc: A UserRPC object. Returns: A tuple that contains signing key name and signature. """ assert rpc.service == _APP_IDENTITY_SERVICE_NAME, repr(rpc.service) assert rpc.method == _SIGN_FOR_APP_METHOD_NAME, repr(rpc.method) try: rpc.check_success() except apiproxy_errors.ApplicationError, err: raise _to_app_identity_error(err) return (response.key_name(), response.signature_bytes()) rpc.make_call(_SIGN_FOR_APP_METHOD_NAME, request, response, signing_for_app_result)
[docs]def make_get_public_certificates_call(rpc): """Executes the RPC call to get a list of public certificates. Args: rpc: A UserRPC instance. Returns: A list of `PublicCertificate` objects. """ request = app_identity_service_pb.GetPublicCertificateForAppRequest() response = app_identity_service_pb.GetPublicCertificateForAppResponse() def get_certs_result(rpc): """Checks success, handles exceptions, and returns the converted RPC result. This method waits for the RPC if it has not yet finished and calls the post-call hooks on the first invocation. Args: rpc: A UserRPC object. Returns: A list of `PublicCertificate` objects. """ assert rpc.service == _APP_IDENTITY_SERVICE_NAME, repr(rpc.service) assert rpc.method == _GET_CERTS_METHOD_NAME, repr(rpc.method) try: rpc.check_success() except apiproxy_errors.ApplicationError, err: raise _to_app_identity_error(err) result = [] for cert in response.public_certificate_list_list(): result.append(PublicCertificate( cert.key_name(), cert.x509_certificate_pem())) return result rpc.make_call(_GET_CERTS_METHOD_NAME, request, response, get_certs_result)
[docs]def make_get_service_account_name_call(rpc): """Gets the service account name of the app. Args: rpc: A UserRPC object. Returns: Service account name of the app. """ request = app_identity_service_pb.GetServiceAccountNameRequest() response = app_identity_service_pb.GetServiceAccountNameResponse() if rpc.deadline is not None: request.set_deadline(rpc.deadline) def get_service_account_name_result(rpc): """Checks success, handles exceptions, and returns the converted RPC result. This method waits for the RPC if it has not yet finished and calls the post-call hooks on the first invocation. Args: rpc: A UserRPC object. Returns: A string of the service account name of the app. """ assert rpc.service == _APP_IDENTITY_SERVICE_NAME, repr(rpc.service) assert rpc.method == _GET_SERVICE_ACCOUNT_NAME_METHOD_NAME, repr(rpc.method) try: rpc.check_success() except apiproxy_errors.ApplicationError, err: raise _to_app_identity_error(err) return response.service_account_name() rpc.make_call(_GET_SERVICE_ACCOUNT_NAME_METHOD_NAME, request, response, get_service_account_name_result)
[docs]def make_get_default_gcs_bucket_name_call(rpc): """Gets the default Google Cloud Storage bucket name for the app. Args: rpc: A UserRPC object. Returns: The default Google Cloud Storage bucket name for the app. """ request = app_identity_service_pb.GetDefaultGcsBucketNameRequest() response = app_identity_service_pb.GetDefaultGcsBucketNameResponse() if rpc.deadline is not None: request.set_deadline(rpc.deadline) def get_default_gcs_bucket_name_result(rpc): """Checks success, handles exceptions, and returns the converted RPC result. This method waits for the RPC if it has not yet finished and calls the post-call hooks on the first invocation. Args: rpc: A UserRPC object. Returns: A string of the name of the app's default Google Cloud Storage bucket. """ assert rpc.service == _APP_IDENTITY_SERVICE_NAME, repr(rpc.service) assert rpc.method == _GET_DEFAULT_GCS_BUCKET_NAME_METHOD_NAME, ( repr(rpc.method)) try: rpc.check_success() except apiproxy_errors.ApplicationError, err: raise _to_app_identity_error(err) return response.default_gcs_bucket_name() or None rpc.make_call(_GET_DEFAULT_GCS_BUCKET_NAME_METHOD_NAME, request, response, get_default_gcs_bucket_name_result)
[docs]def sign_blob(bytes_to_sign, deadline=None): """Signs a blob. Args: bytes_to_sign: The blob that must be signed. deadline: Optional deadline in seconds for the operation; the default value is a system-specific deadline, typically 5 seconds. Returns: A tuple containing the signing key name and signature. """ rpc = create_rpc(deadline) make_sign_blob_call(rpc, bytes_to_sign) rpc.wait() return rpc.get_result()
[docs]def get_public_certificates(deadline=None): """Gets public certificates. Args: deadline: Optional deadline in seconds for the operation; the default value is a system-specific deadline, typically 5 seconds. Returns: A list of `PublicCertificate` objects. """ rpc = create_rpc(deadline) make_get_public_certificates_call(rpc) rpc.wait() return rpc.get_result()
[docs]def get_service_account_name(deadline=None): """Gets the service account name of the app. Args: deadline: Optional deadline in seconds for the operation; the default value is a system-specific deadline, typically 5 seconds. Returns: The service account name of the app. """ rpc = create_rpc(deadline) make_get_service_account_name_call(rpc) rpc.wait() return rpc.get_result()
[docs]def get_default_gcs_bucket_name(deadline=None): """Gets the default Google Cloud Storage bucket name for the app. Args: deadline: Optional deadline in seconds for the operation; the default value is a system-specific deadline, typically 5 seconds. Returns: Default bucket name for the app. """ rpc = create_rpc(deadline) make_get_default_gcs_bucket_name_call(rpc) rpc.wait() return rpc.get_result()
def _ParseFullAppId(app_id): """Parses a full app ID into `partition`, `domain_name`, and `display_app_id`. Args: app_id: The full partitioned app ID. Returns: A tuple `(partition, domain_name, display_app_id)`. The partition and domain name might be empty. """ partition = '' psep = app_id.find(_PARTITION_SEPARATOR) if psep > 0: partition = app_id[:psep] app_id = app_id[psep+1:] domain_name = '' dsep = app_id.find(_DOMAIN_SEPARATOR) if dsep > 0: domain_name = app_id[:dsep] app_id = app_id[dsep+1:] return partition, domain_name, app_id
[docs]def get_application_id(): """Gets the application ID of an app. Returns: The application ID of the app. """ full_app_id = os.getenv('APPLICATION_ID') _, domain_name, display_app_id = _ParseFullAppId(full_app_id) if domain_name: return '%s%s%s' % (domain_name, _DOMAIN_SEPARATOR, display_app_id) return display_app_id
[docs]def get_default_version_hostname(): """Gets the standard host name of the default version of the app. For example, if your `application_id` is `my-app`, then the result might be `my-app.appspot.com`. Returns: The standard host name of the default version of the application. """ return os.getenv('DEFAULT_VERSION_HOSTNAME')
[docs]def make_get_access_token_call(rpc, scopes, service_account_id=None): """Generates the OAuth 2.0 access token to act on behalf of the application. This method is asynchronous and uncached. Most developers should use `get_access_token` instead. Args: rpc: An RPC object. scopes: The requested API scope string, or a list of strings. Raises: InvalidScope: If the scopes are unspecified or invalid. """ request = app_identity_service_pb.GetAccessTokenRequest() if not scopes: raise InvalidScope('No scopes specified.') if isinstance(scopes, basestring): request.add_scope(scopes) else: for scope in scopes: request.add_scope(scope) if service_account_id: if isinstance(service_account_id, (int, long)): request.set_service_account_id(service_account_id) elif isinstance(service_account_id, basestring): request.set_service_account_name(service_account_id) else: raise TypeError() response = app_identity_service_pb.GetAccessTokenResponse() def get_access_token_result(rpc): """Checks success, handles exceptions, and returns the converted RPC result. This method waits for the RPC if it has not yet finished, and calls the post-call hooks on the first invocation. Args: rpc: A UserRPC object. Returns: A `Pair`, `Access` token string and the expiration time in seconds since the epoch. """ assert rpc.service == _APP_IDENTITY_SERVICE_NAME, repr(rpc.service) assert rpc.method == _GET_ACCESS_TOKEN_METHOD_NAME, repr(rpc.method) try: rpc.check_success() except apiproxy_errors.ApplicationError, err: raise _to_app_identity_error(err) return response.access_token(), response.expiration_time() rpc.make_call(_GET_ACCESS_TOKEN_METHOD_NAME, request, response, get_access_token_result)
[docs]def get_access_token_uncached(scopes, deadline=None, service_account_id=None): """Generates the OAuth 2.0 access token to act on behalf of the application. This method is asynchronous and uncached. Most developers should use `get_access_token` instead. Args: scopes: The requested API scope string, or a list of strings. deadline: Optional deadline in seconds for the operation; the default value is a system-specific deadline, typically 5 seconds. Returns: A `Pair`, `Access` token string and the expiration time in seconds since the epoch. """ rpc = create_rpc(deadline) make_get_access_token_call(rpc, scopes, service_account_id=service_account_id) rpc.wait() return rpc.get_result()
[docs]def get_access_token(scopes, service_account_id=None): """The OAuth 2.0 access token to act on behalf of the application. This token will be cached. A token will be generated and cached for the service account for the App Engine application. Each application has an associated Google account. This function returns an OAuth 2.0 access token that corresponds to the running app. Access tokens are safe to cache and reuse until their expiry time as returned. This method caches access tokens using both an in-process cache and memcache. Args: scopes: The requested API scope string, or a list of strings. Returns: A `Pair`, `Access` token string and the expiration time in seconds since the epoch. """ cache_key = _MEMCACHE_KEY_PREFIX + str(scopes) if service_account_id: cache_key += ',%s' % service_account_id cached = _access_token_cache.get(cache_key) if cached is not None: access_token, expires_at = cached safe_expiry = (expires_at - _TOKEN_EXPIRY_SAFETY_MARGIN - _random_cache_expiry_delta) if time.time() < safe_expiry: return access_token, expires_at memcache_value = memcache.get(cache_key, namespace=_MEMCACHE_NAMESPACE) if memcache_value: access_token, expires_at = memcache_value else: access_token, expires_at = get_access_token_uncached( scopes, service_account_id=service_account_id) memcache_expiry = expires_at - _TOKEN_EXPIRY_SAFETY_MARGIN memcache_expiry -= _MAX_RANDOM_EXPIRY_DELTA memcache_expiry -= 10 memcache.add(cache_key, (access_token, expires_at), memcache_expiry, namespace=_MEMCACHE_NAMESPACE) if len(_access_token_cache) >= _MAX_TOKEN_CACHE_SIZE: _access_token_cache.clear() _access_token_cache[cache_key] = (access_token, expires_at) return access_token, expires_at

Send feedback about...

App Engine standard environment for Python