<s> import key_storage <EOL> import binascii <EOL> import hashlib <EOL> import os <EOL> import re <EOL> import rebuild <EOL> import dss <EOL> import rsa <EOL> from coro . asn1 . ber import decode as ber_decode <EOL> from coro . ssh . keys import openssh_key_formats <EOL> from coro . ssh . util import str_xor <EOL> from coro . ssh . util . password import get_password <EOL> from Crypto . Cipher import DES <EOL> import openssh_known_hosts <EOL> import remote_host <EOL> class OpenSSH_Key_Storage ( key_storage . SSH_Key_Storage ) : <EOL> header = re . compile ( <EOL> rebuild . CONCAT ( <EOL> rebuild . NAME ( '<STR_LIT:name>' , '<STR_LIT>' ) , <EOL> '<STR_LIT::>' , <EOL> rebuild . SPLAT ( '<STR_LIT>' ) , <EOL> rebuild . NAME ( '<STR_LIT:value>' , '<STR_LIT>' ) <EOL> ) <EOL> ) <EOL> key_types = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def get_private_key_filenames ( self , username , private_key_filename ) : <EOL> """<STR_LIT>""" <EOL> if private_key_filename is None : <EOL> if username is None : <EOL> username = os . getlogin ( ) <EOL> home_dir = os . path . expanduser ( '<STR_LIT>' + username ) <EOL> result = [ ] <EOL> for key_type in self . key_types : <EOL> private_key_filename = os . path . join ( home_dir , '<STR_LIT>' , '<STR_LIT>' % key_type ) <EOL> result . append ( private_key_filename ) <EOL> return result <EOL> else : <EOL> return [ private_key_filename ] <EOL> get_private_key_filenames = classmethod ( get_private_key_filenames ) <EOL> def get_public_key_filenames ( self , username , public_key_filename ) : <EOL> """<STR_LIT>""" <EOL> if public_key_filename is None : <EOL> result = self . get_private_key_filenames ( username , None ) <EOL> result = map ( lambda x : x + '<STR_LIT>' , result ) <EOL> return result <EOL> else : <EOL> return [ public_key_filename ] <EOL> get_public_key_filenames = classmethod ( get_public_key_filenames ) <EOL> def load_keys ( self , username = None , private_key_filename = None , public_key_filename = None ) : <EOL> """<STR_LIT>""" <EOL> private_key_filenames = self . get_private_key_filenames ( username , private_key_filename ) <EOL> result = [ ] <EOL> for filename in private_key_filenames : <EOL> private_keys = self . load_private_keys ( private_key_filename = filename ) <EOL> if private_keys : <EOL> assert ( len ( private_keys ) == <NUM_LIT:1> ) <EOL> if public_key_filename is None : <EOL> pub_filename = filename + '<STR_LIT>' <EOL> else : <EOL> pub_filename = public_key_filename <EOL> public_keys = self . load_public_keys ( public_key_filename = pub_filename ) <EOL> if public_keys : <EOL> assert ( len ( public_keys ) == <NUM_LIT:1> ) <EOL> key = private_keys [ <NUM_LIT:0> ] <EOL> key . public_key = public_keys [ <NUM_LIT:0> ] . public_key <EOL> result . append ( key ) <EOL> return result <EOL> def load_private_keys ( self , username = None , private_key_filename = None ) : <EOL> """<STR_LIT>""" <EOL> private_key_filenames = self . get_private_key_filenames ( username , private_key_filename ) <EOL> result = [ ] <EOL> for filename in private_key_filenames : <EOL> try : <EOL> data = open ( filename ) . read ( ) <EOL> except IOError : <EOL> pass <EOL> else : <EOL> result . append ( self . parse_private_key ( data ) ) <EOL> return result <EOL> load_private_keys = classmethod ( load_private_keys ) <EOL> def load_public_keys ( self , username = None , public_key_filename = None ) : <EOL> """<STR_LIT>""" <EOL> public_key_filenames = self . get_public_key_filenames ( username , public_key_filename ) <EOL> result = [ ] <EOL> for filename in public_key_filenames : <EOL> try : <EOL> data = open ( filename ) . read ( ) <EOL> except IOError : <EOL> pass <EOL> else : <EOL> result . append ( self . parse_public_key ( data ) ) <EOL> return result <EOL> load_public_keys = classmethod ( load_public_keys ) <EOL> def parse_private_key ( self , private_key ) : <EOL> """<STR_LIT>""" <EOL> data = private_key . split ( '<STR_LIT:\n>' ) <EOL> self . _strip_empty_surrounding_lines ( data ) <EOL> for key_type in self . key_types : <EOL> if ( data [ <NUM_LIT:0> ] == '<STR_LIT>' % ( key_type . upper ( ) , ) and <EOL> data [ - <NUM_LIT:1> ] == '<STR_LIT>' % ( key_type . upper ( ) , ) ) : <EOL> break <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' % ( data [ <NUM_LIT:0> ] , data [ - <NUM_LIT:1> ] ) ) <EOL> key_data = [ ] <EOL> headers = { } <EOL> current_line = <NUM_LIT:1> <EOL> if '<STR_LIT::>' in data [ current_line ] : <EOL> continuation = <NUM_LIT:0> <EOL> current_value = [ ] <EOL> name = '<STR_LIT>' <EOL> while <NUM_LIT:1> : <EOL> line = data [ current_line ] <EOL> if line . startswith ( '<STR_LIT:U+0020>' ) or line . startswith ( '<STR_LIT:\t>' ) or continuation : <EOL> if line . endswith ( '<STR_LIT:\\>' ) : <EOL> line = line [ : - <NUM_LIT:1> ] <EOL> continuation = <NUM_LIT:1> <EOL> else : <EOL> continuation = <NUM_LIT:0> <EOL> current_value . append ( line . lstrip ( ) ) <EOL> current_line += <NUM_LIT:1> <EOL> continue <EOL> else : <EOL> if current_value : <EOL> headers [ name ] = '<STR_LIT>' . join ( current_value ) <EOL> current_value = [ ] <EOL> if not line : <EOL> break <EOL> match = self . header . match ( line ) <EOL> assert ( match is not None ) , '<STR_LIT>' % line <EOL> d = match . groupdict ( ) <EOL> name = d [ '<STR_LIT:name>' ] <EOL> value = d [ '<STR_LIT:value>' ] <EOL> if line . endswith ( '<STR_LIT:\\>' ) : <EOL> continuation = <NUM_LIT:1> <EOL> current_value . append ( value ) <EOL> current_line += <NUM_LIT:1> <EOL> while <NUM_LIT:1> : <EOL> if data [ current_line ] . startswith ( '<STR_LIT>' ) : <EOL> break <EOL> if data [ current_line ] : <EOL> key_data . append ( data [ current_line ] ) <EOL> current_line += <NUM_LIT:1> <EOL> key_data = '<STR_LIT>' . join ( key_data ) <EOL> key_data = binascii . a2b_base64 ( key_data ) <EOL> if '<STR_LIT>' in headers : <EOL> proc_type = headers [ '<STR_LIT>' ] . split ( '<STR_LIT:U+002C>' ) <EOL> proc_type = map ( lambda x : x . strip ( ) , proc_type ) <EOL> if len ( proc_type ) == <NUM_LIT:2> and proc_type [ <NUM_LIT:0> ] == '<STR_LIT:4>' and proc_type [ <NUM_LIT:1> ] == '<STR_LIT>' : <EOL> assert '<STR_LIT>' in headers , '<STR_LIT>' <EOL> dek_info = headers [ '<STR_LIT>' ] . split ( '<STR_LIT:U+002C>' ) <EOL> dek_info = map ( lambda x : x . strip ( ) , dek_info ) <EOL> assert ( len ( dek_info ) == <NUM_LIT:2> ) , '<STR_LIT>' % dek_info <EOL> assert ( dek_info [ <NUM_LIT:0> ] == '<STR_LIT>' ) , '<STR_LIT>' % dek_info [ <NUM_LIT:0> ] <EOL> iv = binascii . a2b_hex ( dek_info [ <NUM_LIT:1> ] ) <EOL> passphrase = self . ask_for_passphrase ( ) <EOL> a = hashlib . md5 ( passphrase + iv ) . digest ( ) <EOL> b = hashlib . md5 ( a + passphrase + iv ) . digest ( ) <EOL> passkey = ( a + b ) [ : <NUM_LIT> ] <EOL> key_data = self . des_ede3_cbc_decrypt ( key_data , iv , passkey ) <EOL> key_data , _ = ber_decode ( key_data ) <EOL> key_data = [ long ( x ) for x in key_data ] <EOL> if key_type not in keytype_map : <EOL> return None <EOL> key_obj = keytype_map [ key_type ] ( ) <EOL> key_obj . private_key = tuple ( key_data [ <NUM_LIT:1> : <NUM_LIT:6> ] ) <EOL> return key_obj <EOL> parse_private_key = classmethod ( parse_private_key ) <EOL> def ask_for_passphrase ( ) : <EOL> return get_password ( '<STR_LIT>' ) <EOL> ask_for_passphrase = staticmethod ( ask_for_passphrase ) <EOL> def des_ede3_cbc_decrypt ( data , iv , key ) : <EOL> assert ( len ( data ) % <NUM_LIT:8> == <NUM_LIT:0> ) , '<STR_LIT>' % len ( data ) <EOL> key1 = DES . new ( key [ <NUM_LIT:0> : <NUM_LIT:8> ] , DES . MODE_ECB ) <EOL> key2 = DES . new ( key [ <NUM_LIT:8> : <NUM_LIT:16> ] , DES . MODE_ECB ) <EOL> key3 = DES . new ( key [ <NUM_LIT:16> : <NUM_LIT> ] , DES . MODE_ECB ) <EOL> result = [ ] <EOL> prev = iv <EOL> for i in xrange ( <NUM_LIT:0> , len ( data ) , <NUM_LIT:8> ) : <EOL> block = data [ i : i + <NUM_LIT:8> ] <EOL> value = key1 . decrypt ( <EOL> key2 . encrypt ( <EOL> key3 . decrypt ( block ) ) ) <EOL> result . append ( str_xor ( prev , value ) ) <EOL> prev = block <EOL> return '<STR_LIT>' . join ( result ) <EOL> des_ede3_cbc_decrypt = staticmethod ( des_ede3_cbc_decrypt ) <EOL> def parse_public_key ( public_key ) : <EOL> """<STR_LIT>""" <EOL> key_match = openssh_key_formats . ssh2_key . match ( public_key ) <EOL> if not key_match : <EOL> return None <EOL> keytype = key_match . group ( '<STR_LIT>' ) <EOL> if keytype not in keytype_map : <EOL> return None <EOL> try : <EOL> key = binascii . a2b_base64 ( key_match . group ( '<STR_LIT>' ) ) <EOL> except binascii . Error : <EOL> return None <EOL> key_obj = keytype_map [ keytype ] ( ) <EOL> key_obj . set_public_key ( key ) <EOL> return key_obj <EOL> parse_public_key = staticmethod ( parse_public_key ) <EOL> def _strip_empty_surrounding_lines ( data ) : <EOL> while <NUM_LIT:1> : <EOL> if not data [ <NUM_LIT:0> ] : <EOL> del data [ <NUM_LIT:0> ] <EOL> else : <EOL> break <EOL> while <NUM_LIT:1> : <EOL> if not data [ - <NUM_LIT:1> ] : <EOL> del data [ - <NUM_LIT:1> ] <EOL> else : <EOL> break <EOL> _strip_empty_surrounding_lines = staticmethod ( _strip_empty_surrounding_lines ) <EOL> def get_authorized_keys_filename ( username , authorized_keys_filename = None ) : <EOL> if authorized_keys_filename is None : <EOL> if username is None : <EOL> username = os . getlogin ( ) <EOL> home_dir = os . path . expanduser ( '<STR_LIT>' + username ) <EOL> authorized_keys_filename = os . path . join ( home_dir , '<STR_LIT>' , '<STR_LIT>' ) <EOL> return authorized_keys_filename <EOL> get_authorized_keys_filename = staticmethod ( get_authorized_keys_filename ) <EOL> def verify ( self , host_id , server_key_types , public_host_key , username = None , port = <NUM_LIT> ) : <EOL> for key in server_key_types : <EOL> if public_host_key . name == key . name : <EOL> if self . _verify_contains ( host_id , public_host_key , username , port ) : <EOL> return <NUM_LIT:1> <EOL> return <NUM_LIT:0> <EOL> verify . __doc__ = key_storage . SSH_Key_Storage . verify . __doc__ <EOL> verify = classmethod ( verify ) <EOL> def _verify_contains ( host_id , key , username , port ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( host_id , remote_host . IPv4_Remote_Host_ID ) : <EOL> return <NUM_LIT:0> <EOL> hostfile = openssh_known_hosts . OpenSSH_Known_Hosts ( ) <EOL> return hostfile . check_for_host ( host_id , key , username , port ) <EOL> _verify_contains = staticmethod ( _verify_contains ) <EOL> def update_known_hosts ( host , public_host_key , username = None ) : <EOL> hostfile = openssh_known_hosts . OpenSSH_Known_Hosts ( ) <EOL> hostfile . update_known_hosts ( host , public_host_key , username ) <EOL> update_known_hosts . __doc__ = key_storage . SSH_Key_Storage . update_known_hosts . __doc__ <EOL> update_known_hosts = staticmethod ( update_known_hosts ) <EOL> keytype_map = { '<STR_LIT>' : dss . SSH_DSS , <EOL> '<STR_LIT>' : dss . SSH_DSS , <EOL> '<STR_LIT>' : dss . SSH_DSS , <EOL> '<STR_LIT>' : rsa . SSH_RSA , <EOL> '<STR_LIT>' : rsa . SSH_RSA , <EOL> } <EOL> import unittest <EOL> class ssh_key_storage_test_case ( unittest . TestCase ) : <EOL> pass <EOL> class load_dsa_test_case ( ssh_key_storage_test_case ) : <EOL> def runTest ( self ) : <EOL> public_key = '<STR_LIT>' <EOL> public_key_value = ( <NUM_LIT> L , <EOL> <NUM_LIT> L , <EOL> <NUM_LIT> L , <EOL> <NUM_LIT> L ) <EOL> private_key = """<STR_LIT>""" <EOL> private_key_value = ( <NUM_LIT> L , <EOL> <NUM_LIT> L , <EOL> <NUM_LIT> L , <EOL> <NUM_LIT> L , <EOL> <NUM_LIT> L ) <EOL> encrypted_private_key = """<STR_LIT>""" <EOL> encrypted_private_key_value = ( <NUM_LIT> L , <EOL> <NUM_LIT> L , <EOL> <NUM_LIT> L , <EOL> <NUM_LIT> L , <EOL> <NUM_LIT> L ) <EOL> a = OpenSSH_Key_Storage ( ) <EOL> dss_obj = a . parse_public_key ( public_key ) <EOL> self . assertEqual ( dss_obj . public_key , public_key_value ) <EOL> dss_obj = a . parse_private_key ( private_key ) <EOL> self . assertEqual ( dss_obj . private_key , private_key_value ) <EOL> class fixed_passphrase_OpenSSH_Key_Storage ( OpenSSH_Key_Storage ) : <EOL> def ask_for_passphrase ( ) : <EOL> return '<STR_LIT>' <EOL> ask_for_passphrase = staticmethod ( ask_for_passphrase ) <EOL> a = fixed_passphrase_OpenSSH_Key_Storage ( ) <EOL> dss_obj = a . parse_private_key ( encrypted_private_key ) <EOL> self . assertEqual ( dss_obj . private_key , encrypted_private_key_value ) <EOL> class load_rsa_test_case ( ssh_key_storage_test_case ) : <EOL> def runTest ( self ) : <EOL> public_key = '<STR_LIT>' <EOL> public_key_value = ( <NUM_LIT> L , <EOL> <NUM_LIT> L ) <EOL> private_key = """<STR_LIT>""" <EOL> private_key_value = ( <NUM_LIT> L , <EOL> <NUM_LIT> L , <EOL> <NUM_LIT> L , <EOL> <NUM_LIT> L , <EOL> <NUM_LIT> L ) <EOL> encrypted_private_key = """<STR_LIT>""" <EOL> encrypted_private_key_value = ( <NUM_LIT> L , <EOL> <NUM_LIT> L , <EOL> <NUM_LIT> L , <EOL> <NUM_LIT> L , <EOL> <NUM_LIT> L ) <EOL> a = OpenSSH_Key_Storage ( ) <EOL> rsa_obj = a . parse_public_key ( public_key ) <EOL> self . assertEqual ( rsa_obj . public_key , public_key_value ) <EOL> rsa_obj = a . parse_private_key ( private_key ) <EOL> self . assertEqual ( rsa_obj . private_key , private_key_value ) <EOL> class fixed_passphrase_OpenSSH_Key_Storage ( OpenSSH_Key_Storage ) : <EOL> def ask_for_passphrase ( ) : <EOL> return '<STR_LIT>' <EOL> ask_for_passphrase = staticmethod ( ask_for_passphrase ) <EOL> a = fixed_passphrase_OpenSSH_Key_Storage ( ) <EOL> rsa_obj = a . parse_private_key ( encrypted_private_key ) <EOL> self . assertEqual ( rsa_obj . private_key , encrypted_private_key_value ) <EOL> def suite ( ) : <EOL> suite = unittest . TestSuite ( ) <EOL> suite . addTest ( load_dsa_test_case ( ) ) <EOL> suite . addTest ( load_rsa_test_case ( ) ) <EOL> return suite <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( defaultTest = '<STR_LIT>' ) </s>
