google/appengine/api/app_identity/app_identity_stub.py

#!/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.
#




"""App identity stub service implementation.

This service behaves the same as the production service, except using
constant values instead of app-specific values:
* Signing key is constant; in production this rotates.
* Public key is constant; in production this varies per app.
* Service account name is constant; in production this varies per app.
"""








import binascii
import time

try:
  from Crypto.Hash import SHA256
  from Crypto.PublicKey import RSA
  from Crypto.Util import number
  CRYPTO_LIB_INSTALLED = True
except ImportError, e:
  CRYPTO_LIB_INSTALLED = False

try:
  import rsa
  RSA_LIB_INSTALLED = True
except ImportError, e:
  RSA_LIB_INSTALLED = False

from google.appengine.api import apiproxy_stub

APP_SERVICE_ACCOUNT_NAME = 'test@localhost'
APP_DEFAULT_GCS_BUCKET_NAME = 'app_default_bucket'

SIGNING_KEY_NAME = 'key'


N = 19119371788959611760073322421014045870056498252163411380847152703712917776733759011400972099255719579701566470175077491500050513917658074590935646529525468755348555932670175295728802986097707368373781743941167574738113348515272061138933984990014969297930973127363812200790406743271047572192133912023914306041356562363557723417403707408838823620411045628159183655215061768071407845537324145892973481372872161981015237572556138317222082306397041309823528068650373958169977675424007883635551170458356632131122901683151395297447872184074888239102348331222079943386530179883880518236689216575776729057173406091195993394637
MODULUS_BYTES = 256

E = 65536L

D = 16986504444572720056487621821047100642841595850137583213470349776864799280835251113078612103869013355016302383270733509621770011190160658118800356360958694229960556902751935956316359959542321272425222634888969943798180994410031448370776358545990991384123912313866752051562052322103544805811361355593091450379904792608637886965065110019212136239200637553477192566763015004249754677600683846556806159369233241157779976231822757855748068765507787598014034587835400718727569389998321277712761796543890788269130617890866139616903097422259980026836628018133574943835504630997228592718738382001678104796538128020421537193913
X509_PUBLIC_CERT = """
-----BEGIN CERTIFICATE-----
MIIC/jCCAeagAwIBAgIIQTBFcRw3moMwDQYJKoZIhvcNAQEFBQAwIjEgMB4GA1UE
AxMXcm9ib3RqYXZhLmEuYXBwc3BvdC5jb20wHhcNMTEwMjIzMTUwNzQ5WhcNMTEw
MjI0MTYwNzQ5WjAiMSAwHgYDVQQDExdyb2JvdGphdmEuYS5hcHBzcG90LmNvbTCC
ASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBAJd0YJCQWvQMa+7L/orCt3D0
hVtkdAkeGSikuT4U7mNrxBuOaAbxCIGhRbUe2p+uvRF6MZtLvoU1h9qEFo/wAVDO
HN4WHhw3VLl/OVuredRfe8bBTi0KqdgUBrKr8V61n26N3B4Ma9dkTMbcODC/XCfP
IRJnTIf4Z1vnoEfWQEJDfW9QLJFyJF17hpp9l5S1uuMJBxjYMsZ3ExLqSFhM7IbN
1PDBAb6zGtI7b9AVP+gxS1hjXiJoZA32IWINAZiPV+0k925ecsV0BkI0zV4Ta06F
JexNx040y5ivr4C214GRUM3UKihirTcEOBS1a7SRi5wCPh/wT0A8gN6NNbTNjc0C
AwEAAaM4MDYwDAYDVR0TAQH/BAIwADAOBgNVHQ8BAf8EBAMCB4AwFgYDVR0lAQH/
BAwwCgYIKwYBBQUHAwIwDQYJKoZIhvcNAQEFBQADggEBAD+h2D+XGIHWMwPCA2DN
JgMhN1yTTJ8dtwbiQIhfy8xjOJbrzZaSEX8g2gDm50qaEl5TYHHr2zvAI1UMWdR4
nx9TN7I9u3GoOcQsmn9TaOKkBDpMv8sPtFBal3AR5PwR5Sq8/4L/M22LX/TN0eIF
Y4LnkW+X/h442N8a1oXn05UYtFo+p/6emZb1S84WZAnONGtF5D1Z6HuX4ikDI5m+
iZbwm47mLkV8yuTZGKI1gJsWmAsElPkoWVy2X0t69ecBOYyn3wMmQhkLk2+7lLlD
/c4kygP/941fe1Wb/T9yGeBXFwEvJ4jWbX93Q4Xhk9UgHlso9xkCu9QeWFvJqufR
5Cc=
-----END CERTIFICATE-----
"""


PREFIX = '3031300d060960864801650304020105000420'
LEN_OF_PREFIX = 19
HEADER1 = '0001'
HEADER2 = '00'
PADDING = 'ff'


LENGTH_OF_SHA256_HASH = 32


class AppIdentityServiceStub(apiproxy_stub.APIProxyStub):
  """A stub for the AppIdentityService API for offline development.

  Provides stub functions which allow a developer to test integration before
  deployment.
  """
  THREADSAFE = True

  def __init__(self, service_name='app_identity_service'):
    """Constructor."""
    super(AppIdentityServiceStub, self).__init__(service_name)
    self.__default_gcs_bucket_name = APP_DEFAULT_GCS_BUCKET_NAME

  def _Dynamic_SignForApp(self, request, response):
    """Implementation of AppIdentityService::SignForApp."""
    bytes_to_sign = request.bytes_to_sign()
    if RSA_LIB_INSTALLED:




      signature_bytes = rsa.pkcs1.sign(
          bytes_to_sign,
          rsa.key.PrivateKey(N, E, D, 3, 5),
          'SHA-256')
    elif CRYPTO_LIB_INSTALLED:


      rsa_obj = RSA.construct((N, E, D))
      hash_obj = SHA256.new()
      hash_obj.update(bytes_to_sign)
      padding_length = MODULUS_BYTES - LEN_OF_PREFIX - LENGTH_OF_SHA256_HASH - 3
      emsa = (HEADER1 + (PADDING * padding_length) + HEADER2 +
              PREFIX + hash_obj.hexdigest())
      sig = rsa_obj.sign(binascii.a2b_hex(emsa), '')
      signature_bytes = number.long_to_bytes(sig[0])
    else:
      raise NotImplementedError("""Unable to import the pycrypto module,
                                SignForApp is disabled.""")
    response.set_signature_bytes(signature_bytes)
    response.set_key_name(SIGNING_KEY_NAME)

  def _Dynamic_GetPublicCertificatesForApp(self, request, response):
    """Implementation of AppIdentityService::GetPublicCertificatesForApp"""
    cert = response.add_public_certificate_list()
    cert.set_key_name(SIGNING_KEY_NAME)
    cert.set_x509_certificate_pem(X509_PUBLIC_CERT)

  def _Dynamic_GetServiceAccountName(self, request, response):
    """Implementation of AppIdentityService::GetServiceAccountName"""
    response.set_service_account_name(APP_SERVICE_ACCOUNT_NAME)

  def _Dynamic_GetDefaultGcsBucketName(self, unused_request, response):
    """Implementation of AppIdentityService::GetDefaultGcsBucketName."""
    response.set_default_gcs_bucket_name(self.__default_gcs_bucket_name)

  def SetDefaultGcsBucketName(self, default_gcs_bucket_name):
    if default_gcs_bucket_name:
      self.__default_gcs_bucket_name = default_gcs_bucket_name
    else:
      self.__default_gcs_bucket_name = APP_DEFAULT_GCS_BUCKET_NAME

  def _Dynamic_GetAccessToken(self, request, response):
    """Implementation of AppIdentityService::GetAccessToken.

    This API returns an invalid token, as the dev_appserver does not have
    access to an actual service account.
    """
    token = ':'.join(request.scope_list())
    service_account_id = request.service_account_id()
    if service_account_id:
      token += '.%d' % service_account_id
    if request.has_service_account_name():
      token += '.%s' % request.service_account_name()
    response.set_access_token('InvalidToken:%s:%s' % (token, time.time() % 100))

    response.set_expiration_time(int(time.time()) + 1800)

  @staticmethod
  def Create(email_address=None, private_key_path=None):
    if email_address:
      from google.appengine.api.app_identity import app_identity_keybased_stub

      return app_identity_keybased_stub.KeyBasedAppIdentityServiceStub(
          email_address=email_address,
          private_key_path=private_key_path)
    else:
      return AppIdentityServiceStub()