<s> import os <EOL> import base64 <EOL> from Crypto import Random <EOL> from Crypto . Cipher import AES <EOL> from Crypto . PublicKey import RSA <EOL> from Crypto . Hash import MD5 <EOL> from Crypto . Hash import SHA <EOL> from Crypto . Hash import SHA256 <EOL> from hashlib import sha256 as HMAC_HASH <EOL> from hmac import HMAC as HMAC_FUNC <EOL> try : <EOL> from Crypto . Cipher import PKCS1_OAEP as RSA_PAD_SCHEME <EOL> except ImportError : <EOL> RSA_PAD_SCHEME = None <EOL> try : <EOL> from Crypto . Signature import PKCS1_v1_5 as RSA_SGN_SCHEME <EOL> except ImportError : <EOL> RSA_SGN_SCHEME = None <EOL> RSA_NULL_KEY_OBJ = RSA . _RSAobj ( None , None ) <EOL> AES_KEY_BIT_SIZE = <NUM_LIT:32> * <NUM_LIT:8> <EOL> AES_KEY_DIR_NAME = "<STR_LIT>" <EOL> AES_RAW_KEY_FILE = "<STR_LIT>" <EOL> AES_MSG_PAD_SIZE = <NUM_LIT:64> <EOL> RSA_KEY_BIT_SIZE = <NUM_LIT> <EOL> RSA_KEY_FMT_NAME = "<STR_LIT>" <EOL> RSA_KEY_DIR_NAME = "<STR_LIT>" <EOL> RSA_PUB_KEY_FILE = "<STR_LIT>" <EOL> RSA_PRI_KEY_FILE = "<STR_LIT>" <EOL> DATA_MARKER_BYTE = "<STR_LIT>" <EOL> DATA_PARTIT_BYTE = "<STR_LIT:\n>" <EOL> UNICODE_ENCODING = "<STR_LIT:utf-8>" <EOL> PWRD_HASH_ROUNDS = <NUM_LIT> <EOL> USR_DB_SALT_SIZE = <NUM_LIT:16> <EOL> MIN_AES_KEY_SIZE = <NUM_LIT:16> <EOL> MIN_PASSWORD_LEN = <NUM_LIT:12> <EOL> MD5LEG_HASH_FUNC = MD5 . new <EOL> SHA256_HASH_FUNC = SHA256 . new <EOL> GLOBAL_RAND_POOL = Random . new ( ) <EOL> def null_encode ( s ) : return s <EOL> def null_decode ( s ) : return s <EOL> def safe_decode ( s , decode_func = base64 . b64decode ) : <EOL> try : <EOL> r = decode_func ( s ) <EOL> except : <EOL> r = s <EOL> return r <EOL> def extract_message_and_auth_code ( raw_data_blob ) : <EOL> if ( raw_data_blob [ <NUM_LIT:0> ] != DATA_MARKER_BYTE ) : <EOL> return ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> i = <NUM_LIT:1> <EOL> j = raw_data_blob . find ( DATA_MARKER_BYTE , i ) <EOL> if ( j != - <NUM_LIT:1> ) : <EOL> msg = raw_data_blob [ i : j ] <EOL> mac = raw_data_blob [ j + <NUM_LIT:1> : ] <EOL> else : <EOL> msg = raw_data_blob [ i : ] <EOL> mac = "<STR_LIT>" <EOL> return ( msg , mac ) <EOL> def encrypt_sign_message ( aes_obj , raw_msg , use_macs ) : <EOL> assert ( type ( raw_msg ) == str ) <EOL> assert ( isinstance ( aes_obj , aes_cipher ) ) <EOL> ret_enc_msg = "<STR_LIT>" <EOL> ret_msg_mac = "<STR_LIT>" <EOL> if ( use_macs ) : <EOL> enc_msg_mac = aes_obj . encrypt_sign_bytes ( raw_msg ) <EOL> ret_enc_msg = DATA_MARKER_BYTE + enc_msg_mac [ <NUM_LIT:0> ] <EOL> ret_msg_mac = DATA_MARKER_BYTE + enc_msg_mac [ <NUM_LIT:1> ] <EOL> else : <EOL> raw_enc_msg = aes_obj . encrypt_encode_bytes ( raw_msg ) <EOL> ret_enc_msg = DATA_MARKER_BYTE + raw_enc_msg <EOL> return ( ret_enc_msg + ret_msg_mac + DATA_PARTIT_BYTE ) <EOL> def decrypt_auth_message ( aes_obj , raw_msg , use_macs ) : <EOL> assert ( type ( raw_msg ) == str ) <EOL> assert ( isinstance ( aes_obj , aes_cipher ) ) <EOL> enc_msg_mac = extract_message_and_auth_code ( raw_msg ) <EOL> if ( len ( enc_msg_mac [ <NUM_LIT:0> ] ) == <NUM_LIT:0> ) : <EOL> return "<STR_LIT>" <EOL> if ( use_macs ) : <EOL> dec_msg = aes_obj . auth_decrypt_bytes ( enc_msg_mac , safe_decode ) <EOL> else : <EOL> dec_msg = aes_obj . decode_decrypt_bytes ( enc_msg_mac [ <NUM_LIT:0> ] , safe_decode ) <EOL> return dec_msg <EOL> def verify_message_auth_code ( our_mac , msg_mac , ses_key ) : <EOL> msg_mac = HMAC_FUNC ( ses_key , msg_mac , HMAC_HASH ) <EOL> our_mac = HMAC_FUNC ( ses_key , our_mac , HMAC_HASH ) <EOL> msg_mac = msg_mac . digest ( ) <EOL> our_mac = our_mac . digest ( ) <EOL> num_val = <NUM_LIT:0> <EOL> if ( len ( msg_mac ) != len ( our_mac ) ) : <EOL> return False <EOL> for i in xrange ( len ( our_mac ) ) : <EOL> num_val += ( our_mac [ i ] == msg_mac [ i ] ) <EOL> return ( num_val == len ( our_mac ) ) <EOL> def int32_to_str ( n ) : <EOL> assert ( n >= ( <NUM_LIT:0> ) ) <EOL> assert ( n < ( <NUM_LIT:1> << <NUM_LIT:32> ) ) <EOL> s = "<STR_LIT>" <EOL> s += "<STR_LIT>" % ( ( n >> <NUM_LIT:0> ) & <NUM_LIT> ) <EOL> s += "<STR_LIT>" % ( ( n >> <NUM_LIT:8> ) & <NUM_LIT> ) <EOL> s += "<STR_LIT>" % ( ( n >> <NUM_LIT:16> ) & <NUM_LIT> ) <EOL> s += "<STR_LIT>" % ( ( n >> <NUM_LIT> ) & <NUM_LIT> ) <EOL> return s <EOL> def str_to_int32 ( s ) : <EOL> n = <NUM_LIT:0> <EOL> n += ( ord ( s [ <NUM_LIT:0> ] ) << <NUM_LIT:0> ) <EOL> n += ( ord ( s [ <NUM_LIT:1> ] ) << <NUM_LIT:8> ) <EOL> n += ( ord ( s [ <NUM_LIT:2> ] ) << <NUM_LIT:16> ) <EOL> n += ( ord ( s [ <NUM_LIT:3> ] ) << <NUM_LIT> ) <EOL> return n <EOL> def pad_str ( msg , bs ) : <EOL> num = bs - ( len ( msg ) % bs ) <EOL> ext = num * chr ( num ) <EOL> return ( msg + ext ) <EOL> def unpad_str ( msg , bs ) : <EOL> idx = len ( msg ) - <NUM_LIT:1> <EOL> cnt = ord ( msg [ idx : ] ) <EOL> return msg [ <NUM_LIT:0> : - cnt ] <EOL> def read_file ( file_name , file_mode ) : <EOL> try : <EOL> f = open ( file_name , file_mode ) <EOL> s = f . read ( ) <EOL> f = f . close ( ) <EOL> return s <EOL> except IOError : <EOL> pass <EOL> return "<STR_LIT>" <EOL> def write_file ( file_name , file_mode , file_data ) : <EOL> try : <EOL> f = open ( file_name , file_mode ) <EOL> os . chmod ( file_name , <NUM_LIT> ) <EOL> f . write ( "<STR_LIT:%s>" % file_data ) <EOL> f = f . close ( ) <EOL> except IOError : <EOL> pass <EOL> class rsa_cipher : <EOL> def __init__ ( self , key_dir = RSA_KEY_DIR_NAME ) : <EOL> self . set_rnd_gen ( Random . new ( ) ) <EOL> self . set_instance_keys ( key_dir ) <EOL> self . set_pad_scheme ( RSA_PAD_SCHEME ) <EOL> self . set_sgn_scheme ( RSA_SGN_SCHEME ) <EOL> def set_rnd_gen ( self , rnd_gen ) : self . rnd_gen = rnd_gen <EOL> def set_pub_key ( self , pub_key ) : self . pub_key = pub_key <EOL> def set_pri_key ( self , pri_key ) : self . pri_key = pri_key <EOL> def get_pub_key ( self ) : return self . pub_key <EOL> def get_pri_key ( self ) : return self . pri_key <EOL> def sanity_test_keys ( self ) : <EOL> pk = ( self . pri_key . publickey ( ) ) <EOL> b0 = ( pk == self . pub_key ) <EOL> b1 = ( pk . exportKey ( RSA_KEY_FMT_NAME ) == self . pub_key . exportKey ( RSA_KEY_FMT_NAME ) ) <EOL> b2 = ( ( not self . pub_key . has_private ( ) ) and self . pri_key . has_private ( ) ) <EOL> return ( b0 and b1 and b2 ) <EOL> def set_pad_scheme ( self , scheme ) : <EOL> if ( scheme == None ) : <EOL> self . enc_pad_scheme = None <EOL> self . dec_pad_scheme = None <EOL> else : <EOL> self . enc_pad_scheme = scheme . new ( self . pub_key ) <EOL> self . dec_pad_scheme = scheme . new ( self . pri_key ) <EOL> def set_sgn_scheme ( self , scheme ) : <EOL> if ( scheme == None ) : <EOL> self . msg_sign_scheme = None <EOL> self . msg_auth_scheme = None <EOL> else : <EOL> self . msg_sign_scheme = scheme . new ( self . pri_key ) <EOL> self . msg_auth_scheme = scheme . new ( self . pub_key ) <EOL> def set_instance_keys ( self , key_dir ) : <EOL> if ( key_dir == None ) : <EOL> self . set_pub_key ( RSA_NULL_KEY_OBJ ) <EOL> self . set_pri_key ( RSA_NULL_KEY_OBJ ) <EOL> return <EOL> if ( not self . import_keys ( key_dir ) ) : <EOL> self . generate_keys ( ) <EOL> assert ( self . sanity_test_keys ( ) ) <EOL> def generate_keys ( self , num_bits = RSA_KEY_BIT_SIZE ) : <EOL> self . set_pri_key ( RSA . generate ( num_bits , self . rnd_gen . read ) ) <EOL> self . set_pub_key ( self . pri_key . publickey ( ) ) <EOL> return True <EOL> def import_key ( self , key_str ) : <EOL> return ( RSA . importKey ( key_str ) ) <EOL> def import_keys ( self , key_dir ) : <EOL> assert ( len ( key_dir ) == <NUM_LIT:0> or key_dir [ - <NUM_LIT:1> ] == '<STR_LIT:/>' ) <EOL> pub_key_str = read_file ( key_dir + RSA_PUB_KEY_FILE , "<STR_LIT:r>" ) <EOL> pri_key_str = read_file ( key_dir + RSA_PRI_KEY_FILE , "<STR_LIT:r>" ) <EOL> if ( len ( pub_key_str ) != <NUM_LIT:0> and len ( pri_key_str ) != <NUM_LIT:0> ) : <EOL> self . set_pub_key ( self . import_key ( pub_key_str ) ) <EOL> self . set_pri_key ( self . import_key ( pri_key_str ) ) <EOL> return True <EOL> return False <EOL> def export_keys ( self , key_dir ) : <EOL> assert ( len ( key_dir ) != <NUM_LIT:0> ) <EOL> assert ( key_dir [ - <NUM_LIT:1> ] == '<STR_LIT:/>' ) <EOL> if ( not os . path . isdir ( key_dir ) ) : <EOL> os . mkdir ( key_dir , <NUM_LIT> ) <EOL> write_file ( key_dir + RSA_PUB_KEY_FILE , "<STR_LIT:w>" , self . pub_key . exportKey ( RSA_KEY_FMT_NAME ) ) <EOL> write_file ( key_dir + RSA_PRI_KEY_FILE , "<STR_LIT:w>" , self . pri_key . exportKey ( RSA_KEY_FMT_NAME ) ) <EOL> def encrypt_encode_bytes_utf8 ( self , raw_bytes , encode_func = base64 . b64encode ) : <EOL> return ( self . encrypt_encode_bytes ( raw_bytes . encode ( UNICODE_ENCODING ) , encode_func ) ) <EOL> def decode_decrypt_bytes_utf8 ( self , enc_bytes , decode_func = base64 . b64decode ) : <EOL> return ( self . decode_decrypt_bytes ( enc_bytes . encode ( UNICODE_ENCODING ) , decode_func ) ) <EOL> def encrypt_encode_bytes ( self , raw_bytes , encode_func = base64 . b64encode ) : <EOL> assert ( type ( raw_bytes ) == str ) <EOL> assert ( len ( raw_bytes ) != <NUM_LIT:0> ) <EOL> assert ( self . pub_key . size ( ) >= ( len ( raw_bytes ) * <NUM_LIT:8> ) ) <EOL> assert ( ord ( raw_bytes [ <NUM_LIT:0> ] ) != <NUM_LIT:0> ) <EOL> if ( self . enc_pad_scheme != None ) : <EOL> enc_bytes = self . enc_pad_scheme . encrypt ( raw_bytes ) <EOL> else : <EOL> enc_bytes = self . pub_key . encrypt ( raw_bytes , "<STR_LIT>" ) [ <NUM_LIT:0> ] <EOL> return ( encode_func ( enc_bytes ) ) <EOL> def decode_decrypt_bytes ( self , enc_bytes , decode_func = base64 . b64decode ) : <EOL> assert ( type ( enc_bytes ) == str ) <EOL> assert ( len ( enc_bytes ) != <NUM_LIT:0> ) <EOL> enc_bytes = decode_func ( enc_bytes ) <EOL> if ( self . dec_pad_scheme != None ) : <EOL> dec_bytes = self . dec_pad_scheme . decrypt ( enc_bytes ) <EOL> else : <EOL> dec_bytes = self . pri_key . decrypt ( enc_bytes ) <EOL> return dec_bytes <EOL> def sign_bytes_utf8 ( self , msg_bytes ) : <EOL> return ( self . sign_bytes ( msg_bytes . encode ( UNICODE_ENCODING ) ) ) <EOL> def auth_bytes_utf8 ( self , msg_bytes , sig_bytes ) : <EOL> return ( self . auth_bytes ( msg_bytes . encode ( UNICODE_ENCODING ) , sig_bytes ) ) <EOL> def sign_bytes ( self , msg_bytes ) : <EOL> assert ( type ( msg_bytes ) == str ) <EOL> assert ( len ( msg_bytes ) != <NUM_LIT:0> ) <EOL> msg_bytes = SHA256_HASH_FUNC ( msg_bytes ) <EOL> if ( self . msg_sign_scheme != None ) : <EOL> ret = self . msg_sign_scheme . sign ( msg_bytes ) <EOL> else : <EOL> ret = str ( self . pri_key . sign ( msg_bytes . digest ( ) , "<STR_LIT>" ) [ <NUM_LIT:0> ] ) <EOL> assert ( type ( ret ) == str ) <EOL> return ret <EOL> def auth_bytes ( self , msg_bytes , sig_bytes ) : <EOL> assert ( type ( msg_bytes ) == str ) <EOL> assert ( type ( sig_bytes ) == str ) <EOL> assert ( len ( msg_bytes ) != <NUM_LIT:0> ) <EOL> msg_bytes = SHA256_HASH_FUNC ( msg_bytes ) <EOL> if ( self . msg_auth_scheme != None ) : <EOL> ret = self . msg_auth_scheme . verify ( msg_bytes , sig_bytes ) <EOL> else : <EOL> ret = ( self . pub_key . verify ( msg_bytes . digest ( ) , ( long ( sig_bytes ) , long ( <NUM_LIT:0> ) ) ) ) <EOL> assert ( type ( ret ) == bool ) <EOL> return ret <EOL> class aes_cipher : <EOL> def __init__ ( self , key_dir = AES_KEY_DIR_NAME , padding_length = AES_MSG_PAD_SIZE ) : <EOL> assert ( type ( key_dir ) == str ) <EOL> assert ( ( padding_length % <NUM_LIT:16> ) == <NUM_LIT:0> ) <EOL> self . pad_length = padding_length <EOL> self . random_gen = Random . new ( ) <EOL> self . khash_func = SHA256_HASH_FUNC <EOL> self . set_instance_key ( key_dir ) <EOL> def set_instance_key ( self , key_dir ) : <EOL> if ( not self . import_key ( key_dir ) ) : <EOL> self . set_key ( self . generate_key ( "<STR_LIT>" ) ) <EOL> def generate_key ( self , raw_key , key_len = AES_KEY_BIT_SIZE ) : <EOL> if ( len ( raw_key ) == <NUM_LIT:0> ) : <EOL> key_str = self . random_gen . read ( key_len / <NUM_LIT:8> ) <EOL> key_str = self . khash_func ( key_str ) <EOL> else : <EOL> key_str = self . khash_func ( raw_key ) <EOL> return ( key_str . digest ( ) ) <EOL> def get_key ( self ) : return self . key_string <EOL> def set_key ( self , s ) : self . key_string = s <EOL> def import_key ( self , key_dir ) : <EOL> assert ( len ( key_dir ) == <NUM_LIT:0> or key_dir [ - <NUM_LIT:1> ] == '<STR_LIT:/>' ) <EOL> key_str = read_file ( key_dir + AES_RAW_KEY_FILE , "<STR_LIT:rb>" ) <EOL> if ( len ( key_str ) != <NUM_LIT:0> ) : <EOL> self . set_key ( key_str ) <EOL> return True <EOL> return False <EOL> def export_key ( self , key_dir ) : <EOL> assert ( len ( key_dir ) != <NUM_LIT:0> ) <EOL> assert ( key_dir [ - <NUM_LIT:1> ] == '<STR_LIT:/>' ) <EOL> if ( not os . path . isdir ( key_dir ) ) : <EOL> os . mkdir ( key_dir , <NUM_LIT> ) <EOL> write_file ( key_dir + AES_RAW_KEY_FILE , "<STR_LIT:wb>" , self . get_key ( ) ) <EOL> def encrypt_encode_bytes_utf8 ( self , raw_bytes , encode_func = base64 . b64encode ) : <EOL> return ( self . encrypt_encode_bytes ( raw_bytes . encode ( UNICODE_ENCODING ) , encode_func ) ) <EOL> def decode_decrypt_bytes_utf8 ( self , enc_bytes , decode_func = base64 . b64decode ) : <EOL> return ( self . decode_decrypt_bytes ( enc_bytes . encode ( UNICODE_ENCODING ) , decode_func ) ) <EOL> def encrypt_encode_bytes ( self , raw_bytes , encode_func = base64 . b64encode ) : <EOL> assert ( type ( raw_bytes ) == str ) <EOL> assert ( len ( raw_bytes ) != <NUM_LIT:0> ) <EOL> ini_vector = self . random_gen . read ( AES . block_size ) <EOL> aes_object = AES . new ( self . key_string , AES . MODE_CBC , ini_vector ) <EOL> pad_bytes = pad_str ( raw_bytes , self . pad_length ) <EOL> enc_bytes = aes_object . encrypt ( pad_bytes ) <EOL> return ( encode_func ( ini_vector + enc_bytes ) ) <EOL> def decode_decrypt_bytes ( self , enc_bytes , decode_func = base64 . b64decode ) : <EOL> assert ( type ( enc_bytes ) == str ) <EOL> assert ( len ( enc_bytes ) != <NUM_LIT:0> ) <EOL> enc_bytes = decode_func ( enc_bytes ) <EOL> ini_vector = enc_bytes [ <NUM_LIT:0> : AES . block_size ] <EOL> aes_object = AES . new ( self . key_string , AES . MODE_CBC , ini_vector ) <EOL> dec_bytes = aes_object . decrypt ( enc_bytes [ AES . block_size : ] ) <EOL> dec_bytes = unpad_str ( dec_bytes , self . pad_length ) <EOL> return dec_bytes <EOL> def encrypt_sign_bytes_utf8 ( self , raw_msg , encode_func = base64 . b64encode ) : <EOL> return ( self . encrypt_sign_bytes ( raw_msg . encode ( UNICODE_ENCODING ) , encode_func ) ) <EOL> def auth_decrypt_bytes_utf8 ( self , enc_msg , msg_mac , decode_func = base64 . b64decode ) : <EOL> return ( self . auth_decrypt_bytes ( ( enc_msg . encode ( UNICODE_ENCODING ) , msg_mac . encode ( UNICODE_ENCODING ) ) , decode_func ) ) <EOL> def encrypt_sign_bytes ( self , raw_msg , encode_func = base64 . b64encode ) : <EOL> assert ( type ( raw_msg ) == str ) <EOL> enc_msg = self . encrypt_encode_bytes ( raw_msg , null_encode ) <EOL> msg_mac = HMAC_FUNC ( self . get_key ( ) , enc_msg , HMAC_HASH ) <EOL> msg_mac = encode_func ( msg_mac . digest ( ) ) <EOL> enc_msg = encode_func ( enc_msg ) <EOL> return ( enc_msg , msg_mac ) <EOL> def auth_decrypt_bytes ( self , enc_msg , msg_mac , decode_func = base64 . b64decode ) : <EOL> assert ( type ( enc_msg ) == str ) <EOL> assert ( type ( msg_mac ) == str ) <EOL> msg_mac = decode_func ( msg_mac ) <EOL> enc_msg = decode_func ( enc_msg ) <EOL> our_mac = HMAC_FUNC ( self . get_key ( ) , enc_msg , HMAC_HASH ) <EOL> our_mac = our_mac . digest ( ) <EOL> if ( verify_message_auth_code ( our_mac , msg_mac , self . get_key ( ) ) ) : <EOL> return ( self . decode_decrypt_bytes ( enc_msg , null_decode ) ) <EOL> return "<STR_LIT>" </s>
