<s> from hashlib import sha256 <EOL> import hmac <EOL> import random <EOL> import struct <EOL> from twimp . handshake import Handshaker <EOL> from twimp . primitives import _s_ulong_b as _s_ulong <EOL> from twimp . utils import ms_time <EOL> LOG_CATEGORY = '<STR_LIT>' <EOL> import twimp . log <EOL> log = twimp . log . get_logger ( LOG_CATEGORY ) <EOL> def _hmac ( key , msg , digestmod = sha256 ) : <EOL> return hmac . new ( key , msg , digestmod ) . digest ( ) <EOL> _shared_key_suffix = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) . decode ( '<STR_LIT>' ) <EOL> _fms_key = '<STR_LIT>' <EOL> _full_fms_key = _fms_key + _shared_key_suffix <EOL> _fp_key = '<STR_LIT>' <EOL> _full_fp_key = _fp_key + _shared_key_suffix <EOL> def _make_offset_extractor ( pos , count , mod , shift ) : <EOL> return ( lambda data : sum ( map ( ord , data [ pos : pos + count ] ) ) % mod + shift ) <EOL> ( OFFSET_SCHEME_1 , OFFSET_SCHEME_2 ) = range ( <NUM_LIT:2> ) <EOL> schemes = [ <EOL> ( _make_offset_extractor ( <NUM_LIT:8> , <NUM_LIT:4> , <NUM_LIT> , <NUM_LIT:12> ) , ) , <EOL> ( _make_offset_extractor ( <NUM_LIT> , <NUM_LIT:4> , <NUM_LIT> , <NUM_LIT> ) , ) , <EOL> ] <EOL> schemes_by_client_ver = [ <EOL> ( ( <NUM_LIT:10> , <NUM_LIT:0> , <NUM_LIT:32> , <NUM_LIT:0> ) , OFFSET_SCHEME_2 , ( <NUM_LIT:10> , <NUM_LIT:0> , <NUM_LIT:32> , <NUM_LIT:2> ) ) , <EOL> ( ( <NUM_LIT:9> , <NUM_LIT:0> , <NUM_LIT> , <NUM_LIT:0> ) , OFFSET_SCHEME_1 , ( <NUM_LIT:9> , <NUM_LIT:0> , <NUM_LIT> , <NUM_LIT:0> ) ) , <EOL> ] <EOL> def find_client_offset_scheme ( version ) : <EOL> for min_scheme_version , scheme , _cli_version in schemes_by_client_ver : <EOL> if version >= min_scheme_version : <EOL> return scheme <EOL> return None <EOL> DEFAULT_SERVER_COMPAT_VERSION = ( <NUM_LIT:3> , <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> DEFAULT_CLIENT_COMPAT_VERSION = schemes_by_client_ver [ <NUM_LIT:0> ] [ <NUM_LIT:2> ] <EOL> NO_VERSION = ( <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> _s_ts_ver = struct . Struct ( '<STR_LIT>' ) <EOL> def generate_random_bytes ( count ) : <EOL> n = random . getrandbits ( count * <NUM_LIT:8> ) <EOL> fmt = '<STR_LIT>' % ( count * <NUM_LIT:2> ) <EOL> return ( fmt % n ) . decode ( '<STR_LIT>' ) <EOL> class CryptoHandshaker ( Handshaker ) : <EOL> server_compat_version = DEFAULT_SERVER_COMPAT_VERSION <EOL> client_compat_version = DEFAULT_CLIENT_COMPAT_VERSION <EOL> compat_version = None <EOL> strict = True <EOL> def __init__ ( self , protocol , epoch_base , is_client = False ) : <EOL> Handshaker . __init__ ( self , protocol , epoch_base , is_client = is_client ) <EOL> self . _digest_offset_extractor = None <EOL> if self . is_client : <EOL> self . compat_version = self . client_compat_version <EOL> scheme = find_client_offset_scheme ( self . compat_version ) <EOL> if scheme is not None : <EOL> self . _digest_offset_extractor = schemes [ scheme ] [ <NUM_LIT:0> ] <EOL> else : <EOL> self . compat_version = NO_VERSION <EOL> else : <EOL> self . compat_version = self . server_compat_version <EOL> def select_own_key_short ( self ) : <EOL> if self . is_client : <EOL> return _fp_key <EOL> return _fms_key <EOL> def select_own_key ( self ) : <EOL> if self . is_client : <EOL> return _full_fp_key <EOL> else : <EOL> return _full_fms_key <EOL> def select_other_key_short ( self ) : <EOL> if self . is_client : <EOL> return _fms_key <EOL> return _fp_key <EOL> def select_other_key ( self ) : <EOL> if self . is_client : <EOL> return _full_fms_key <EOL> else : <EOL> return _full_fp_key <EOL> def _check_client_scheme ( self , scheme , data ) : <EOL> digest_offset_extractor , = schemes [ scheme ] <EOL> offset = digest_offset_extractor ( data ) <EOL> key = self . select_other_key_short ( ) <EOL> digest = _hmac ( key , data [ : offset ] + data [ offset + <NUM_LIT:32> : ] ) <EOL> return buffer ( data , offset , <NUM_LIT:32> ) == buffer ( digest ) <EOL> def discover_client_scheme ( self , context ) : <EOL> ts , v0 , v1 , v2 , v3 = _s_ts_ver . unpack_from ( context , <NUM_LIT:0> ) <EOL> client_ver = ( v0 , v1 , v2 , v3 ) <EOL> if client_ver == NO_VERSION : <EOL> log . info ( "<STR_LIT>" ) <EOL> return None <EOL> log . debug ( '<STR_LIT>' , client_ver ) <EOL> scheme = find_client_offset_scheme ( client_ver ) <EOL> log . debug ( '<STR_LIT>' , scheme ) <EOL> if scheme is not None : <EOL> if self . _check_client_scheme ( scheme , context ) : <EOL> log . debug ( '<STR_LIT>' , scheme ) <EOL> self . _digest_offset_extractor = schemes [ scheme ] [ <NUM_LIT:0> ] <EOL> return scheme <EOL> log . debug ( '<STR_LIT>' ) <EOL> for i in xrange ( len ( schemes ) ) : <EOL> if i == scheme : <EOL> continue <EOL> if self . _check_client_scheme ( i , context ) : <EOL> log . debug ( '<STR_LIT>' , i ) <EOL> self . _digest_offset_extractor = schemes [ i ] [ <NUM_LIT:0> ] <EOL> return i <EOL> log . info ( "<STR_LIT>" ) <EOL> if not self . strict : <EOL> log . debug ( '<STR_LIT>' , OFFSET_SCHEME_1 ) <EOL> return OFFSET_SCHEME_1 <EOL> return None <EOL> def generate_base_request ( self ) : <EOL> return ( _s_ts_ver . pack ( ms_time ( self . epoch_time ( ) ) , <EOL> * self . compat_version ) + <EOL> generate_random_bytes ( self . packet_bytes - <NUM_LIT:8> ) ) <EOL> def generate_request ( self , context = None ) : <EOL> base_request = self . generate_base_request ( ) <EOL> if self . is_client : <EOL> if not self . _digest_offset_extractor : <EOL> return base_request <EOL> else : <EOL> scheme = self . discover_client_scheme ( context ) <EOL> if scheme is None : <EOL> return base_request <EOL> request = buffer ( base_request , <NUM_LIT:0> , self . packet_bytes - <NUM_LIT:32> ) <EOL> offset = self . _digest_offset_extractor ( request ) <EOL> key = self . select_own_key_short ( ) <EOL> digest = _hmac ( key , request ) <EOL> return request [ : offset ] + digest + request [ offset : ] <EOL> def generate_simple_response ( self , request ) : <EOL> return ( buffer ( request , <NUM_LIT:0> , <NUM_LIT:4> ) + <EOL> ( buffer ( _s_ulong . pack ( ms_time ( self . epoch_time ( ) ) ) ) + <EOL> buffer ( request , <NUM_LIT:8> ) ) ) <EOL> def generate_response ( self , request ) : <EOL> if not self . _digest_offset_extractor : <EOL> return self . generate_simple_response ( request ) <EOL> offset = self . _digest_offset_extractor ( request ) <EOL> req_digest = buffer ( request , offset , <NUM_LIT:32> ) <EOL> key = self . select_own_key ( ) <EOL> digest_key = _hmac ( key , req_digest ) <EOL> response = generate_random_bytes ( self . packet_bytes - <NUM_LIT:32> ) <EOL> digest = _hmac ( digest_key , response ) <EOL> return response + digest <EOL> def verify_simple_response ( self , request , response ) : <EOL> return ( buffer ( request , <NUM_LIT:0> , <NUM_LIT:4> ) == buffer ( response , <NUM_LIT:0> , <NUM_LIT:4> ) and <EOL> buffer ( request , <NUM_LIT:8> ) == buffer ( response , <NUM_LIT:8> ) ) <EOL> def verify_response ( self , request , response ) : <EOL> if not self . _digest_offset_extractor : <EOL> return self . verify_simple_response ( request , response ) <EOL> offset = self . _digest_offset_extractor ( request ) <EOL> req_digest = buffer ( request , offset , <NUM_LIT:32> ) <EOL> key = self . select_other_key ( ) <EOL> digest_key = _hmac ( key , req_digest ) <EOL> digest = _hmac ( digest_key , buffer ( response , <NUM_LIT:0> , self . packet_bytes - <NUM_LIT:32> ) ) <EOL> if not self . strict : <EOL> return True <EOL> return buffer ( response , len ( response ) - <NUM_LIT:32> ) == buffer ( digest ) </s>
