<s> """<STR_LIT>""" <EOL> from __future__ import with_statement <EOL> import json <EOL> import threading <EOL> import time <EOL> import urllib <EOL> from oauth2client import client <EOL> from oauth2client import service_account <EOL> from pyasn1 . codec . der import decoder <EOL> from pyasn1_modules . rfc2459 import Certificate <EOL> import rsa <EOL> from google . appengine . api import urlfetch <EOL> from google . appengine . api . app_identity import app_identity_service_pb <EOL> from google . appengine . api . app_identity import app_identity_stub <EOL> from google . appengine . runtime import apiproxy_errors <EOL> def BitStringToByteString ( bs ) : <EOL> """<STR_LIT>""" <EOL> def BitsToInt ( bits ) : <EOL> return sum ( v * ( <NUM_LIT:2> ** ( <NUM_LIT:7> - j ) ) for j , v in enumerate ( bits ) ) <EOL> return str ( bytearray ( [ BitsToInt ( bs [ i : i + <NUM_LIT:8> ] ) for i in range ( <NUM_LIT:0> , len ( bs ) , <NUM_LIT:8> ) ] ) ) <EOL> class DefaultCredentialsBasedAppIdentityServiceStub ( <EOL> app_identity_stub . AppIdentityServiceStub ) : <EOL> """<STR_LIT>""" <EOL> THREADSAFE = True <EOL> def __init__ ( self , service_name = '<STR_LIT>' ) : <EOL> super ( DefaultCredentialsBasedAppIdentityServiceStub , <EOL> self ) . __init__ ( service_name ) <EOL> client . NO_GCE_CHECK = '<STR_LIT:True>' <EOL> self . _credentials = ( <EOL> client . GoogleCredentials . get_application_default ( ) ) <EOL> self . _access_token_cache_lock = threading . Lock ( ) <EOL> self . _access_token_cache = { } <EOL> self . _x509_init_lock = threading . Lock ( ) <EOL> self . _default_gcs_bucket_name = ( <EOL> app_identity_stub . APP_DEFAULT_GCS_BUCKET_NAME ) <EOL> self . _x509 = None <EOL> self . _signing_key = None <EOL> self . _non_service_account_credentials = not isinstance ( <EOL> self . _credentials , service_account . _ServiceAccountCredentials ) <EOL> def _PopulateX509 ( self ) : <EOL> with self . _x509_init_lock : <EOL> if self . _x509 is None : <EOL> url = ( '<STR_LIT>' <EOL> % urllib . unquote_plus ( self . _credentials . service_account_email ) ) <EOL> response = urlfetch . fetch ( <EOL> url = url , <EOL> validate_certificate = True , <EOL> method = urlfetch . GET ) <EOL> if response . status_code != <NUM_LIT:200> : <EOL> raise apiproxy_errors . ApplicationError ( <EOL> app_identity_service_pb . AppIdentityServiceError . UNKNOWN_ERROR , <EOL> '<STR_LIT>' % ( <EOL> url , response . status_code , response . content ) ) <EOL> message = '<STR_LIT>' <EOL> _ , signature = self . _credentials . sign_blob ( message ) <EOL> for signing_key , x509 in json . loads ( response . content ) . items ( ) : <EOL> der = rsa . pem . load_pem ( x509 , '<STR_LIT>' ) <EOL> asn1_cert , _ = decoder . decode ( der , asn1Spec = Certificate ( ) ) <EOL> key_bitstring = ( <EOL> asn1_cert [ '<STR_LIT>' ] <EOL> [ '<STR_LIT>' ] <EOL> [ '<STR_LIT>' ] ) <EOL> key_bytearray = BitStringToByteString ( key_bitstring ) <EOL> public_key = rsa . PublicKey . load_pkcs1 ( key_bytearray , '<STR_LIT>' ) <EOL> try : <EOL> if rsa . pkcs1 . verify ( message , signature , public_key ) : <EOL> self . _x509 = x509 <EOL> self . _signing_key = signing_key <EOL> return <EOL> except rsa . pkcs1 . VerificationError : <EOL> pass <EOL> raise apiproxy_errors . ApplicationError ( <EOL> app_identity_service_pb . AppIdentityServiceError . UNKNOWN_ERROR , <EOL> '<STR_LIT>' % url ) <EOL> def _Dynamic_SignForApp ( self , request , response ) : <EOL> """<STR_LIT>""" <EOL> if self . _non_service_account_credentials : <EOL> return super ( DefaultCredentialsBasedAppIdentityServiceStub , <EOL> self ) . _Dynamic_SignForApp ( request , response ) <EOL> self . _PopulateX509 ( ) <EOL> private_key_id , signature = self . _credentials . sign_blob ( <EOL> request . bytes_to_sign ( ) ) <EOL> assert private_key_id == self . _signing_key <EOL> response . set_signature_bytes ( signature ) <EOL> response . set_key_name ( self . _signing_key ) <EOL> def _Dynamic_GetPublicCertificatesForApp ( self , request , response ) : <EOL> """<STR_LIT>""" <EOL> if self . _non_service_account_credentials : <EOL> return super ( DefaultCredentialsBasedAppIdentityServiceStub , <EOL> self ) . _Dynamic_GetPublicCertificatesForApp ( request , response ) <EOL> self . _PopulateX509 ( ) <EOL> certificate = response . add_public_certificate_list ( ) <EOL> certificate . set_key_name ( self . _signing_key ) <EOL> certificate . set_x509_certificate_pem ( self . _x509 ) <EOL> def _Dynamic_GetServiceAccountName ( self , request , response ) : <EOL> """<STR_LIT>""" <EOL> if self . _non_service_account_credentials : <EOL> response . set_service_account_name ( '<STR_LIT>' ) <EOL> else : <EOL> response . set_service_account_name ( self . _credentials . service_account_email ) <EOL> def _Dynamic_GetDefaultGcsBucketName ( self , unused_request , response ) : <EOL> """<STR_LIT>""" <EOL> response . set_default_gcs_bucket_name ( self . _default_gcs_bucket_name ) <EOL> def SetDefaultGcsBucketName ( self , default_gcs_bucket_name ) : <EOL> if default_gcs_bucket_name : <EOL> self . _default_gcs_bucket_name = default_gcs_bucket_name <EOL> else : <EOL> self . _default_gcs_bucket_name = ( <EOL> app_identity_stub . APP_DEFAULT_GCS_BUCKET_NAME ) <EOL> def _Dynamic_GetAccessToken ( self , request , response ) : <EOL> """<STR_LIT>""" <EOL> scope = '<STR_LIT:U+0020>' . join ( request . scope_list ( ) ) <EOL> with self . _access_token_cache_lock : <EOL> rv = self . _access_token_cache . get ( scope , None ) <EOL> now = int ( time . time ( ) ) <EOL> if not ( rv and rv [ '<STR_LIT>' ] > ( now + <NUM_LIT> ) ) : <EOL> credentials = self . _credentials <EOL> if credentials . create_scoped_required ( ) : <EOL> credentials = credentials . create_scoped ( request . scope_list ( ) ) <EOL> token = credentials . get_access_token ( ) <EOL> rv = { <EOL> '<STR_LIT>' : token . access_token , <EOL> '<STR_LIT>' : now + token . expires_in , <EOL> } <EOL> with self . _access_token_cache_lock : <EOL> self . _access_token_cache [ scope ] = rv <EOL> response . set_access_token ( rv [ '<STR_LIT>' ] ) <EOL> response . set_expiration_time ( rv [ '<STR_LIT>' ] ) </s>
