#NEWLINE# def _PopulateX509( self ) : #NEWLINE# #INDENT# with self . __x509_init_lock : #NEWLINE# #INDENT# if ( not self . __x509 ) : #NEWLINE# #INDENT# url = ( 'https://www.googleapis.com/service_accounts/v1/metadata/x509/%s' % urllib . unquote_plus ( self . __email_address ) ) #NEWLINE# resp = urlfetch . fetch ( url = url , validate_certificate = True , method = urlfetch . GET ) #NEWLINE# if ( resp . status_code != 200 ) : #NEWLINE# #INDENT# raise apiproxy_errors . ApplicationError ( app_identity_service_pb . AppIdentityServiceError . UNKNOWN_ERROR , ( 'Unable to load X509 cert: %s Response code: %i, Content: %s' % ( url , resp . status_code , resp . content ) ) ) #NEWLINE# #UNINDENT# msg = 'test' #NEWLINE# sig = rsa . pkcs1 . sign ( msg , self . __private_key , 'SHA-256' ) #NEWLINE# for ( signing_key , x509 ) in json . loads ( resp . content ) . items ( ) : #NEWLINE# #INDENT# der = rsa . pem . load_pem ( x509 , 'CERTIFICATE' ) #NEWLINE# ( asn1_cert , _ ) = decoder . decode ( x509 , asn1Spec = Certificate ( ) ) #NEWLINE# key_bitstring = asn1_cert [ 'tbsCertificate' ] [ 'subjectPublicKeyInfo' ] [ 'subjectPublicKey' ] #NEWLINE# key_bytearray = BitStringToByteString ( key_bitstring ) #NEWLINE# pub = rsa . PublicKey . load_pkcs1 ( key_bytearray , 'DER' ) #NEWLINE# try : #NEWLINE# #INDENT# if rsa . pkcs1 . verify ( msg , sig , pub ) : #NEWLINE# #INDENT# self . __x509 = x509 #NEWLINE# self . __signing_key = signing_key #NEWLINE# return #NEWLINE# #UNINDENT# #UNINDENT# except rsa . pkcs1 . VerificationError : #NEWLINE# #INDENT# pass #NEWLINE# #UNINDENT# #UNINDENT# raise apiproxy_errors . ApplicationError ( app_identity_service_pb . AppIdentityServiceError . UNKNOWN_ERROR , ( 'Unable to find matching X509 cert for private key: %s' % url ) )