<s> '''<STR_LIT>''' <EOL> """<STR_LIT>""" <EOL> import sys <EOL> import os . path <EOL> import json <EOL> sys . path . append ( <EOL> os . path . abspath ( os . path . join ( os . path . dirname ( __file__ ) , "<STR_LIT>" ) ) ) <EOL> from creds . credentials import Credentials <EOL> from creds . credentials import CredentialsInvalidError , OAuthCredentials <EOL> import helpers <EOL> CRYPTO_FERNET = True <EOL> try : <EOL> from cryptography . fernet import Fernet <EOL> except : <EOL> CRYPTO_FERNET = False <EOL> import fernet <EOL> fernet . Configuration . enforce_ttl = False <EOL> __copyright__ = "<STR_LIT>" <EOL> __status__ = "<STR_LIT>" <EOL> class CredentialManager ( object ) : <EOL> def __init__ ( self , filename , key = None ) : <EOL> def decrypt ( ciphertext ) : <EOL> if key is None : <EOL> return ciphertext <EOL> try : <EOL> value = ciphertext <EOL> if CRYPTO_FERNET == True : <EOL> f = Fernet ( key ) <EOL> value = f . decrypt ( ciphertext ) <EOL> else : <EOL> verifier = fernet . verifier ( key , ciphertext ) <EOL> if verifier . valid ( ) == True : <EOL> value = verifier . message <EOL> return value <EOL> except Exception as e : <EOL> pass <EOL> return ciphertext <EOL> def encrypt ( value ) : <EOL> if key is None : <EOL> return value <EOL> if CRYPTO_FERNET == True : <EOL> f = Fernet ( key ) <EOL> token = f . encrypt ( bytes ( value ) ) <EOL> else : <EOL> token = fernet . generate ( key , value ) <EOL> return token <EOL> self . filename = filename <EOL> self . encrypt = encrypt <EOL> self . decrypt = decrypt <EOL> def credential_object_for_type ( self , identifier , type ) : <EOL> if ( type == "<STR_LIT:password>" ) : <EOL> return Credentials ( identifier ) <EOL> raise NotImplementedError ( "<STR_LIT>" % ( type ) ) <EOL> def get_credentials_for ( self , identifier , check_valid = True ) : <EOL> file_creds = { } <EOL> try : <EOL> with open ( self . filename ) as fp : <EOL> file_creds = json . load ( fp , object_hook = helpers . convert ) <EOL> except IOError as ioe : <EOL> raise NameError ( "<STR_LIT>" + str ( ioe ) ) <EOL> user_creds = file_creds . get ( identifier ) <EOL> if user_creds is None : <EOL> raise NameError ( identifier ) <EOL> return <EOL> type = user_creds . get ( "<STR_LIT:type>" ) <EOL> credentials = self . credential_object_for_type ( identifier , type ) <EOL> credentials . credential_manager = self <EOL> credentials . encrypt = self . encrypt <EOL> credentials . decrypt = self . decrypt <EOL> credentials . read ( user_creds ) <EOL> if check_valid == True and credentials . valid ( ) == False : <EOL> if isinstance ( creds , OAuthCredentials ) : <EOL> try : <EOL> credentials . refreshToken ( ) <EOL> if credentials . valid ( ) : <EOL> if write == True : <EOL> credentials . write ( ) <EOL> return credentials <EOL> except NotImplementedError , nie : <EOL> pass <EOL> raise CredentialsInvalidError ( "<STR_LIT>" % identifier ) <EOL> return credentials <EOL> def get_or_create_credentials_for ( self , identifier , type , write = True , check_valid = True ) : <EOL> try : <EOL> creds = self . get_credentials_for ( identifier , check_valid = False ) <EOL> except NameError , ne : <EOL> creds = self . new_credentials_for ( identifier , type ) <EOL> if write == True : <EOL> creds . write ( ) <EOL> if check_valid == True and creds . valid ( ) == False : <EOL> if isinstance ( creds , OAuthCredentials ) : <EOL> try : <EOL> creds . refreshToken ( ) <EOL> if creds . valid ( ) : <EOL> if write == True : <EOL> creds . write ( ) <EOL> return creds <EOL> except NotImplementedError , nie : <EOL> pass <EOL> for attempt_numer in range ( <NUM_LIT:3> ) : <EOL> creds . new_credentials ( ) <EOL> if creds . valid ( ) : <EOL> if write == True : <EOL> creds . write ( ) <EOL> return creds <EOL> raise CredentialsInvalidError ( "<STR_LIT>" % identifier ) <EOL> return creds <EOL> def new_credentials_for ( self , identifier , type , server_data = None ) : <EOL> credentials = self . credential_object_for_type ( identifier , type ) <EOL> credentials . credential_manager = self <EOL> credentials . encrypt = self . encrypt <EOL> credentials . decrypt = self . decrypt <EOL> if ( server_data is not None ) : <EOL> credentials . server_data = server_data <EOL> credentials . new_credentials ( ) <EOL> return credentials <EOL> def write_back_credentials ( self , credentials ) : <EOL> file_creds = { } <EOL> try : <EOL> file_data = open ( self . filename ) <EOL> file_creds = json . load ( file_data ) <EOL> except IOError as ioe : <EOL> print "<STR_LIT>" <EOL> user_creds = file_creds . get ( credentials . identifier ) <EOL> if user_creds is None : <EOL> user_creds = { } <EOL> user_creds = credentials . _populate_user_creds ( user_creds ) <EOL> file_creds [ credentials . identifier ] = user_creds <EOL> with open ( self . filename , '<STR_LIT:w>' ) as outfile : <EOL> json . dump ( file_creds , outfile , indent = <NUM_LIT:2> , separators = ( '<STR_LIT:U+002C>' , '<STR_LIT>' ) ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> import os <EOL> credential_key = os . environ . get ( '<STR_LIT>' ) <EOL> credentials_file = "<STR_LIT>" <EOL> credential_manager = CredentialManager ( credentials_file , credential_key ) </s>
