<s> from binascii import hexlify , a2b_base64 , b2a_base64 <EOL> from getpass import getpass <EOL> import os <EOL> import msgpack <EOL> import textwrap <EOL> import hmac <EOL> from hashlib import sha256 <EOL> import zlib <EOL> from attic . crypto import pbkdf2_sha256 , get_random_bytes , AES , bytes_to_long , long_to_bytes , bytes_to_int , num_aes_blocks <EOL> from attic . helpers import IntegrityError , get_keys_dir , Error <EOL> PREFIX = b'<STR_LIT>' * <NUM_LIT:8> <EOL> class UnsupportedPayloadError ( Error ) : <EOL> """<STR_LIT>""" <EOL> class KeyfileNotFoundError ( Error ) : <EOL> """<STR_LIT>""" <EOL> class HMAC ( hmac . HMAC ) : <EOL> """<STR_LIT>""" <EOL> def update ( self , msg ) : <EOL> self . inner . update ( msg ) <EOL> def key_creator ( repository , args ) : <EOL> if args . encryption == '<STR_LIT>' : <EOL> return KeyfileKey . create ( repository , args ) <EOL> elif args . encryption == '<STR_LIT>' : <EOL> return PassphraseKey . create ( repository , args ) <EOL> else : <EOL> return PlaintextKey . create ( repository , args ) <EOL> def key_factory ( repository , manifest_data ) : <EOL> if manifest_data [ <NUM_LIT:0> ] == KeyfileKey . TYPE : <EOL> return KeyfileKey . detect ( repository , manifest_data ) <EOL> elif manifest_data [ <NUM_LIT:0> ] == PassphraseKey . TYPE : <EOL> return PassphraseKey . detect ( repository , manifest_data ) <EOL> elif manifest_data [ <NUM_LIT:0> ] == PlaintextKey . TYPE : <EOL> return PlaintextKey . detect ( repository , manifest_data ) <EOL> else : <EOL> raise UnsupportedPayloadError ( manifest_data [ <NUM_LIT:0> ] ) <EOL> class KeyBase ( object ) : <EOL> def __init__ ( self ) : <EOL> self . TYPE_STR = bytes ( [ self . TYPE ] ) <EOL> def id_hash ( self , data ) : <EOL> """<STR_LIT>""" <EOL> def encrypt ( self , data ) : <EOL> pass <EOL> def decrypt ( self , id , data ) : <EOL> pass <EOL> class PlaintextKey ( KeyBase ) : <EOL> TYPE = <NUM_LIT> <EOL> chunk_seed = <NUM_LIT:0> <EOL> @ classmethod <EOL> def create ( cls , repository , args ) : <EOL> print ( '<STR_LIT>' ) <EOL> return cls ( ) <EOL> @ classmethod <EOL> def detect ( cls , repository , manifest_data ) : <EOL> return cls ( ) <EOL> def id_hash ( self , data ) : <EOL> return sha256 ( data ) . digest ( ) <EOL> def encrypt ( self , data ) : <EOL> return b'<STR_LIT>' . join ( [ self . TYPE_STR , zlib . compress ( data ) ] ) <EOL> def decrypt ( self , id , data ) : <EOL> if data [ <NUM_LIT:0> ] != self . TYPE : <EOL> raise IntegrityError ( '<STR_LIT>' ) <EOL> data = zlib . decompress ( memoryview ( data ) [ <NUM_LIT:1> : ] ) <EOL> if id and sha256 ( data ) . digest ( ) != id : <EOL> raise IntegrityError ( '<STR_LIT>' ) <EOL> return data <EOL> class AESKeyBase ( KeyBase ) : <EOL> """<STR_LIT>""" <EOL> PAYLOAD_OVERHEAD = <NUM_LIT:1> + <NUM_LIT:32> + <NUM_LIT:8> <EOL> def id_hash ( self , data ) : <EOL> """<STR_LIT>""" <EOL> return HMAC ( self . id_key , data , sha256 ) . digest ( ) <EOL> def encrypt ( self , data ) : <EOL> data = zlib . compress ( data ) <EOL> self . enc_cipher . reset ( ) <EOL> data = b'<STR_LIT>' . join ( ( self . enc_cipher . iv [ <NUM_LIT:8> : ] , self . enc_cipher . encrypt ( data ) ) ) <EOL> hmac = HMAC ( self . enc_hmac_key , data , sha256 ) . digest ( ) <EOL> return b'<STR_LIT>' . join ( ( self . TYPE_STR , hmac , data ) ) <EOL> def decrypt ( self , id , data ) : <EOL> if data [ <NUM_LIT:0> ] != self . TYPE : <EOL> raise IntegrityError ( '<STR_LIT>' ) <EOL> hmac = memoryview ( data ) [ <NUM_LIT:1> : <NUM_LIT> ] <EOL> if memoryview ( HMAC ( self . enc_hmac_key , memoryview ( data ) [ <NUM_LIT> : ] , sha256 ) . digest ( ) ) != hmac : <EOL> raise IntegrityError ( '<STR_LIT>' ) <EOL> self . dec_cipher . reset ( iv = PREFIX + data [ <NUM_LIT> : <NUM_LIT> ] ) <EOL> data = zlib . decompress ( self . dec_cipher . decrypt ( data [ <NUM_LIT> : ] ) ) <EOL> if id and HMAC ( self . id_key , data , sha256 ) . digest ( ) != id : <EOL> raise IntegrityError ( '<STR_LIT>' ) <EOL> return data <EOL> def extract_nonce ( self , payload ) : <EOL> if payload [ <NUM_LIT:0> ] != self . TYPE : <EOL> raise IntegrityError ( '<STR_LIT>' ) <EOL> nonce = bytes_to_long ( payload [ <NUM_LIT> : <NUM_LIT> ] ) <EOL> return nonce <EOL> def init_from_random_data ( self , data ) : <EOL> self . enc_key = data [ <NUM_LIT:0> : <NUM_LIT:32> ] <EOL> self . enc_hmac_key = data [ <NUM_LIT:32> : <NUM_LIT:64> ] <EOL> self . id_key = data [ <NUM_LIT:64> : <NUM_LIT> ] <EOL> self . chunk_seed = bytes_to_int ( data [ <NUM_LIT> : <NUM_LIT:100> ] ) <EOL> if self . chunk_seed & <NUM_LIT> : <EOL> self . chunk_seed = self . chunk_seed - <NUM_LIT> - <NUM_LIT:1> <EOL> def init_ciphers ( self , enc_iv = b'<STR_LIT>' ) : <EOL> self . enc_cipher = AES ( self . enc_key , enc_iv ) <EOL> self . dec_cipher = AES ( self . enc_key ) <EOL> class PassphraseKey ( AESKeyBase ) : <EOL> TYPE = <NUM_LIT> <EOL> iterations = <NUM_LIT> <EOL> @ classmethod <EOL> def create ( cls , repository , args ) : <EOL> key = cls ( ) <EOL> passphrase = os . environ . get ( '<STR_LIT>' ) <EOL> if passphrase is not None : <EOL> passphrase2 = passphrase <EOL> else : <EOL> passphrase , passphrase2 = <NUM_LIT:1> , <NUM_LIT:2> <EOL> while passphrase != passphrase2 : <EOL> passphrase = getpass ( '<STR_LIT>' ) <EOL> if not passphrase : <EOL> print ( '<STR_LIT>' ) <EOL> continue <EOL> passphrase2 = getpass ( '<STR_LIT>' ) <EOL> if passphrase != passphrase2 : <EOL> print ( '<STR_LIT>' ) <EOL> key . init ( repository , passphrase ) <EOL> if passphrase : <EOL> print ( '<STR_LIT>' ) <EOL> return key <EOL> @ classmethod <EOL> def detect ( cls , repository , manifest_data ) : <EOL> prompt = '<STR_LIT>' % repository . _location . orig <EOL> key = cls ( ) <EOL> passphrase = os . environ . get ( '<STR_LIT>' ) <EOL> if passphrase is None : <EOL> passphrase = getpass ( prompt ) <EOL> while True : <EOL> key . init ( repository , passphrase ) <EOL> try : <EOL> key . decrypt ( None , manifest_data ) <EOL> num_blocks = num_aes_blocks ( len ( manifest_data ) - <NUM_LIT> ) <EOL> key . init_ciphers ( PREFIX + long_to_bytes ( key . extract_nonce ( manifest_data ) + num_blocks ) ) <EOL> return key <EOL> except IntegrityError : <EOL> passphrase = getpass ( prompt ) <EOL> def init ( self , repository , passphrase ) : <EOL> self . init_from_random_data ( pbkdf2_sha256 ( passphrase . encode ( '<STR_LIT:utf-8>' ) , repository . id , self . iterations , <NUM_LIT:100> ) ) <EOL> self . init_ciphers ( ) <EOL> class KeyfileKey ( AESKeyBase ) : <EOL> FILE_ID = '<STR_LIT>' <EOL> TYPE = <NUM_LIT> <EOL> @ classmethod <EOL> def detect ( cls , repository , manifest_data ) : <EOL> key = cls ( ) <EOL> path = cls . find_key_file ( repository ) <EOL> prompt = '<STR_LIT>' % path <EOL> passphrase = os . environ . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> while not key . load ( path , passphrase ) : <EOL> passphrase = getpass ( prompt ) <EOL> num_blocks = num_aes_blocks ( len ( manifest_data ) - <NUM_LIT> ) <EOL> key . init_ciphers ( PREFIX + long_to_bytes ( key . extract_nonce ( manifest_data ) + num_blocks ) ) <EOL> return key <EOL> @ classmethod <EOL> def find_key_file ( cls , repository ) : <EOL> id = hexlify ( repository . id ) . decode ( '<STR_LIT:ascii>' ) <EOL> keys_dir = get_keys_dir ( ) <EOL> for name in os . listdir ( keys_dir ) : <EOL> filename = os . path . join ( keys_dir , name ) <EOL> with open ( filename , '<STR_LIT:r>' ) as fd : <EOL> line = fd . readline ( ) . strip ( ) <EOL> if line and line . startswith ( cls . FILE_ID ) and line [ <NUM_LIT:10> : ] == id : <EOL> return filename <EOL> raise KeyfileNotFoundError ( repository . _location . canonical_path ( ) , get_keys_dir ( ) ) <EOL> def load ( self , filename , passphrase ) : <EOL> with open ( filename , '<STR_LIT:r>' ) as fd : <EOL> cdata = a2b_base64 ( '<STR_LIT>' . join ( fd . readlines ( ) [ <NUM_LIT:1> : ] ) . encode ( '<STR_LIT:ascii>' ) ) <EOL> data = self . decrypt_key_file ( cdata , passphrase ) <EOL> if data : <EOL> key = msgpack . unpackb ( data ) <EOL> if key [ b'<STR_LIT:version>' ] != <NUM_LIT:1> : <EOL> raise IntegrityError ( '<STR_LIT>' ) <EOL> self . repository_id = key [ b'<STR_LIT>' ] <EOL> self . enc_key = key [ b'<STR_LIT>' ] <EOL> self . enc_hmac_key = key [ b'<STR_LIT>' ] <EOL> self . id_key = key [ b'<STR_LIT>' ] <EOL> self . chunk_seed = key [ b'<STR_LIT>' ] <EOL> self . path = filename <EOL> return True <EOL> def decrypt_key_file ( self , data , passphrase ) : <EOL> d = msgpack . unpackb ( data ) <EOL> assert d [ b'<STR_LIT:version>' ] == <NUM_LIT:1> <EOL> assert d [ b'<STR_LIT>' ] == b'<STR_LIT>' <EOL> key = pbkdf2_sha256 ( passphrase . encode ( '<STR_LIT:utf-8>' ) , d [ b'<STR_LIT>' ] , d [ b'<STR_LIT>' ] , <NUM_LIT:32> ) <EOL> data = AES ( key ) . decrypt ( d [ b'<STR_LIT:data>' ] ) <EOL> if HMAC ( key , data , sha256 ) . digest ( ) != d [ b'<STR_LIT>' ] : <EOL> return None <EOL> return data <EOL> def encrypt_key_file ( self , data , passphrase ) : <EOL> salt = get_random_bytes ( <NUM_LIT:32> ) <EOL> iterations = <NUM_LIT> <EOL> key = pbkdf2_sha256 ( passphrase . encode ( '<STR_LIT:utf-8>' ) , salt , iterations , <NUM_LIT:32> ) <EOL> hash = HMAC ( key , data , sha256 ) . digest ( ) <EOL> cdata = AES ( key ) . encrypt ( data ) <EOL> d = { <EOL> '<STR_LIT:version>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : salt , <EOL> '<STR_LIT>' : iterations , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : hash , <EOL> '<STR_LIT:data>' : cdata , <EOL> } <EOL> return msgpack . packb ( d ) <EOL> def save ( self , path , passphrase ) : <EOL> key = { <EOL> '<STR_LIT:version>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : self . repository_id , <EOL> '<STR_LIT>' : self . enc_key , <EOL> '<STR_LIT>' : self . enc_hmac_key , <EOL> '<STR_LIT>' : self . id_key , <EOL> '<STR_LIT>' : self . chunk_seed , <EOL> } <EOL> data = self . encrypt_key_file ( msgpack . packb ( key ) , passphrase ) <EOL> with open ( path , '<STR_LIT:w>' ) as fd : <EOL> fd . write ( '<STR_LIT>' % ( self . FILE_ID , hexlify ( self . repository_id ) . decode ( '<STR_LIT:ascii>' ) ) ) <EOL> fd . write ( '<STR_LIT:\n>' . join ( textwrap . wrap ( b2a_base64 ( data ) . decode ( '<STR_LIT:ascii>' ) ) ) ) <EOL> fd . write ( '<STR_LIT:\n>' ) <EOL> self . path = path <EOL> def change_passphrase ( self ) : <EOL> passphrase , passphrase2 = <NUM_LIT:1> , <NUM_LIT:2> <EOL> while passphrase != passphrase2 : <EOL> passphrase = getpass ( '<STR_LIT>' ) <EOL> passphrase2 = getpass ( '<STR_LIT>' ) <EOL> if passphrase != passphrase2 : <EOL> print ( '<STR_LIT>' ) <EOL> self . save ( self . path , passphrase ) <EOL> print ( '<STR_LIT>' % self . path ) <EOL> @ classmethod <EOL> def create ( cls , repository , args ) : <EOL> filename = args . repository . to_key_filename ( ) <EOL> path = filename <EOL> i = <NUM_LIT:1> <EOL> while os . path . exists ( path ) : <EOL> i += <NUM_LIT:1> <EOL> path = filename + '<STR_LIT>' % i <EOL> passphrase = os . environ . get ( '<STR_LIT>' ) <EOL> if passphrase is not None : <EOL> passphrase2 = passphrase <EOL> else : <EOL> passphrase , passphrase2 = <NUM_LIT:1> , <NUM_LIT:2> <EOL> while passphrase != passphrase2 : <EOL> passphrase = getpass ( '<STR_LIT>' ) <EOL> passphrase2 = getpass ( '<STR_LIT>' ) <EOL> if passphrase != passphrase2 : <EOL> print ( '<STR_LIT>' ) <EOL> key = cls ( ) <EOL> key . repository_id = repository . id <EOL> key . init_from_random_data ( get_random_bytes ( <NUM_LIT:100> ) ) <EOL> key . init_ciphers ( ) <EOL> key . save ( path , passphrase ) <EOL> print ( '<STR_LIT>' % key . path ) <EOL> print ( '<STR_LIT>' ) <EOL> return key </s>
