<s> """<STR_LIT>""" <EOL> import base64 <EOL> import copy <EOL> import datetime <EOL> import json <EOL> import time <EOL> from oauth2client import GOOGLE_REVOKE_URI <EOL> from oauth2client import GOOGLE_TOKEN_URI <EOL> from oauth2client . _helpers import _json_encode <EOL> from oauth2client . _helpers import _from_bytes <EOL> from oauth2client . _helpers import _urlsafe_b64encode <EOL> from oauth2client import util <EOL> from oauth2client . client import AssertionCredentials <EOL> from oauth2client . client import EXPIRY_FORMAT <EOL> from oauth2client . client import SERVICE_ACCOUNT <EOL> from oauth2client import crypt <EOL> _PASSWORD_DEFAULT = '<STR_LIT>' <EOL> _PKCS12_KEY = '<STR_LIT>' <EOL> _PKCS12_ERROR = r"""<STR_LIT>""" <EOL> class ServiceAccountCredentials ( AssertionCredentials ) : <EOL> """<STR_LIT>""" <EOL> MAX_TOKEN_LIFETIME_SECS = <NUM_LIT> <EOL> """<STR_LIT>""" <EOL> NON_SERIALIZED_MEMBERS = ( <EOL> frozenset ( [ '<STR_LIT>' ] ) | <EOL> AssertionCredentials . NON_SERIALIZED_MEMBERS ) <EOL> """<STR_LIT>""" <EOL> _private_key_pkcs8_pem = None <EOL> _private_key_pkcs12 = None <EOL> _private_key_password = None <EOL> def __init__ ( self , <EOL> service_account_email , <EOL> signer , <EOL> scopes = '<STR_LIT>' , <EOL> private_key_id = None , <EOL> client_id = None , <EOL> user_agent = None , <EOL> ** kwargs ) : <EOL> super ( ServiceAccountCredentials , self ) . __init__ ( <EOL> None , user_agent = user_agent ) <EOL> self . _service_account_email = service_account_email <EOL> self . _signer = signer <EOL> self . _scopes = util . scopes_to_string ( scopes ) <EOL> self . _private_key_id = private_key_id <EOL> self . client_id = client_id <EOL> self . _user_agent = user_agent <EOL> self . _kwargs = kwargs <EOL> def _to_json ( self , strip , to_serialize = None ) : <EOL> """<STR_LIT>""" <EOL> if to_serialize is None : <EOL> to_serialize = copy . copy ( self . __dict__ ) <EOL> pkcs12_val = to_serialize . get ( _PKCS12_KEY ) <EOL> if pkcs12_val is not None : <EOL> to_serialize [ _PKCS12_KEY ] = base64 . b64encode ( pkcs12_val ) <EOL> return super ( ServiceAccountCredentials , self ) . _to_json ( <EOL> strip , to_serialize = to_serialize ) <EOL> @ classmethod <EOL> def _from_parsed_json_keyfile ( cls , keyfile_dict , scopes ) : <EOL> """<STR_LIT>""" <EOL> creds_type = keyfile_dict . get ( '<STR_LIT:type>' ) <EOL> if creds_type != SERVICE_ACCOUNT : <EOL> raise ValueError ( '<STR_LIT>' , creds_type , <EOL> '<STR_LIT>' , SERVICE_ACCOUNT ) <EOL> service_account_email = keyfile_dict [ '<STR_LIT>' ] <EOL> private_key_pkcs8_pem = keyfile_dict [ '<STR_LIT>' ] <EOL> private_key_id = keyfile_dict [ '<STR_LIT>' ] <EOL> client_id = keyfile_dict [ '<STR_LIT>' ] <EOL> signer = crypt . Signer . from_string ( private_key_pkcs8_pem ) <EOL> credentials = cls ( service_account_email , signer , scopes = scopes , <EOL> private_key_id = private_key_id , <EOL> client_id = client_id ) <EOL> credentials . _private_key_pkcs8_pem = private_key_pkcs8_pem <EOL> return credentials <EOL> @ classmethod <EOL> def from_json_keyfile_name ( cls , filename , scopes = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> with open ( filename , '<STR_LIT:r>' ) as file_obj : <EOL> client_credentials = json . load ( file_obj ) <EOL> return cls . _from_parsed_json_keyfile ( client_credentials , scopes ) <EOL> @ classmethod <EOL> def from_json_keyfile_dict ( cls , keyfile_dict , scopes = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> return cls . _from_parsed_json_keyfile ( keyfile_dict , scopes ) <EOL> @ classmethod <EOL> def _from_p12_keyfile_contents ( cls , service_account_email , <EOL> private_key_pkcs12 , <EOL> private_key_password = None , scopes = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> if private_key_password is None : <EOL> private_key_password = _PASSWORD_DEFAULT <EOL> if crypt . Signer is not crypt . OpenSSLSigner : <EOL> raise NotImplementedError ( _PKCS12_ERROR ) <EOL> signer = crypt . Signer . from_string ( private_key_pkcs12 , <EOL> private_key_password ) <EOL> credentials = cls ( service_account_email , signer , scopes = scopes ) <EOL> credentials . _private_key_pkcs12 = private_key_pkcs12 <EOL> credentials . _private_key_password = private_key_password <EOL> return credentials <EOL> @ classmethod <EOL> def from_p12_keyfile ( cls , service_account_email , filename , <EOL> private_key_password = None , scopes = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> with open ( filename , '<STR_LIT:rb>' ) as file_obj : <EOL> private_key_pkcs12 = file_obj . read ( ) <EOL> return cls . _from_p12_keyfile_contents ( <EOL> service_account_email , private_key_pkcs12 , <EOL> private_key_password = private_key_password , scopes = scopes ) <EOL> @ classmethod <EOL> def from_p12_keyfile_buffer ( cls , service_account_email , file_buffer , <EOL> private_key_password = None , scopes = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> private_key_pkcs12 = file_buffer . read ( ) <EOL> return cls . _from_p12_keyfile_contents ( <EOL> service_account_email , private_key_pkcs12 , <EOL> private_key_password = private_key_password , scopes = scopes ) <EOL> def _generate_assertion ( self ) : <EOL> """<STR_LIT>""" <EOL> now = int ( time . time ( ) ) <EOL> payload = { <EOL> '<STR_LIT>' : self . token_uri , <EOL> '<STR_LIT>' : self . _scopes , <EOL> '<STR_LIT>' : now , <EOL> '<STR_LIT>' : now + self . MAX_TOKEN_LIFETIME_SECS , <EOL> '<STR_LIT>' : self . _service_account_email , <EOL> } <EOL> payload . update ( self . _kwargs ) <EOL> return crypt . make_signed_jwt ( self . _signer , payload , <EOL> key_id = self . _private_key_id ) <EOL> def sign_blob ( self , blob ) : <EOL> """<STR_LIT>""" <EOL> return self . _private_key_id , self . _signer . sign ( blob ) <EOL> @ property <EOL> def service_account_email ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _service_account_email <EOL> @ property <EOL> def serialization_data ( self ) : <EOL> return { <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : self . _service_account_email , <EOL> '<STR_LIT>' : self . _private_key_id , <EOL> '<STR_LIT>' : self . _private_key_pkcs8_pem , <EOL> '<STR_LIT>' : self . client_id , <EOL> } <EOL> @ classmethod <EOL> def from_json ( cls , json_data ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( json_data , dict ) : <EOL> json_data = json . loads ( _from_bytes ( json_data ) ) <EOL> private_key_pkcs8_pem = None <EOL> pkcs12_val = json_data . get ( _PKCS12_KEY ) <EOL> password = None <EOL> if pkcs12_val is None : <EOL> private_key_pkcs8_pem = json_data [ '<STR_LIT>' ] <EOL> signer = crypt . Signer . from_string ( private_key_pkcs8_pem ) <EOL> else : <EOL> pkcs12_val = base64 . b64decode ( pkcs12_val ) <EOL> password = json_data [ '<STR_LIT>' ] <EOL> signer = crypt . Signer . from_string ( pkcs12_val , password ) <EOL> credentials = cls ( <EOL> json_data [ '<STR_LIT>' ] , <EOL> signer , <EOL> scopes = json_data [ '<STR_LIT>' ] , <EOL> private_key_id = json_data [ '<STR_LIT>' ] , <EOL> client_id = json_data [ '<STR_LIT>' ] , <EOL> user_agent = json_data [ '<STR_LIT>' ] , <EOL> ** json_data [ '<STR_LIT>' ] <EOL> ) <EOL> if private_key_pkcs8_pem is not None : <EOL> credentials . _private_key_pkcs8_pem = private_key_pkcs8_pem <EOL> if pkcs12_val is not None : <EOL> credentials . _private_key_pkcs12 = pkcs12_val <EOL> if password is not None : <EOL> credentials . _private_key_password = password <EOL> credentials . invalid = json_data [ '<STR_LIT>' ] <EOL> credentials . access_token = json_data [ '<STR_LIT>' ] <EOL> credentials . token_uri = json_data [ '<STR_LIT>' ] <EOL> credentials . revoke_uri = json_data [ '<STR_LIT>' ] <EOL> token_expiry = json_data . get ( '<STR_LIT>' , None ) <EOL> if token_expiry is not None : <EOL> credentials . token_expiry = datetime . datetime . strptime ( <EOL> token_expiry , EXPIRY_FORMAT ) <EOL> return credentials <EOL> def create_scoped_required ( self ) : <EOL> return not self . _scopes <EOL> def create_scoped ( self , scopes ) : <EOL> result = self . __class__ ( self . _service_account_email , <EOL> self . _signer , <EOL> scopes = scopes , <EOL> private_key_id = self . _private_key_id , <EOL> client_id = self . client_id , <EOL> user_agent = self . _user_agent , <EOL> ** self . _kwargs ) <EOL> result . token_uri = self . token_uri <EOL> result . revoke_uri = self . revoke_uri <EOL> result . _private_key_pkcs8_pem = self . _private_key_pkcs8_pem <EOL> result . _private_key_pkcs12 = self . _private_key_pkcs12 <EOL> result . _private_key_password = self . _private_key_password <EOL> return result <EOL> def create_delegated ( self , sub ) : <EOL> """<STR_LIT>""" <EOL> new_kwargs = dict ( self . _kwargs ) <EOL> new_kwargs [ '<STR_LIT>' ] = sub <EOL> result = self . __class__ ( self . _service_account_email , <EOL> self . _signer , <EOL> scopes = self . _scopes , <EOL> private_key_id = self . _private_key_id , <EOL> client_id = self . client_id , <EOL> user_agent = self . _user_agent , <EOL> ** new_kwargs ) <EOL> result . token_uri = self . token_uri <EOL> result . revoke_uri = self . revoke_uri <EOL> result . _private_key_pkcs8_pem = self . _private_key_pkcs8_pem <EOL> result . _private_key_pkcs12 = self . _private_key_pkcs12 <EOL> result . _private_key_password = self . _private_key_password <EOL> return result </s>
