<s> """<STR_LIT>""" <EOL> import socket <EOL> try : <EOL> import ssl <EOL> except ImportError : <EOL> pass <EOL> from urllib . parse import urlparse <EOL> import os <EOL> import array <EOL> import struct <EOL> import uuid <EOL> import hashlib <EOL> import base64 <EOL> import logging <EOL> """<STR_LIT>""" <EOL> VERSION = <NUM_LIT> <EOL> STATUS_NORMAL = <NUM_LIT:1000> <EOL> STATUS_GOING_AWAY = <NUM_LIT> <EOL> STATUS_PROTOCOL_ERROR = <NUM_LIT> <EOL> STATUS_UNSUPPORTED_DATA_TYPE = <NUM_LIT> <EOL> STATUS_STATUS_NOT_AVAILABLE = <NUM_LIT> <EOL> STATUS_ABNORMAL_CLOSED = <NUM_LIT> <EOL> STATUS_INVALID_PAYLOAD = <NUM_LIT> <EOL> STATUS_POLICY_VIOLATION = <NUM_LIT> <EOL> STATUS_MESSAGE_TOO_BIG = <NUM_LIT> <EOL> STATUS_INVALID_EXTENSION = <NUM_LIT> <EOL> STATUS_UNEXPECTED_CONDITION = <NUM_LIT> <EOL> STATUS_TLS_HANDSHAKE_ERROR = <NUM_LIT> <EOL> logger = logging . getLogger ( ) <EOL> class WebSocketException ( Exception ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class WebSocketConnectionClosedException ( WebSocketException ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> default_timeout = None <EOL> traceEnabled = False <EOL> def enableTrace ( tracable ) : <EOL> """<STR_LIT>""" <EOL> global traceEnabled <EOL> traceEnabled = tracable <EOL> if tracable : <EOL> if not logger . handlers : <EOL> logger . addHandler ( logging . StreamHandler ( ) ) <EOL> logger . setLevel ( logging . DEBUG ) <EOL> def setdefaulttimeout ( timeout ) : <EOL> """<STR_LIT>""" <EOL> global default_timeout <EOL> default_timeout = timeout <EOL> def getdefaulttimeout ( ) : <EOL> """<STR_LIT>""" <EOL> return default_timeout <EOL> def _parse_url ( url ) : <EOL> """<STR_LIT>""" <EOL> if "<STR_LIT::>" not in url : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> scheme , url = url . split ( "<STR_LIT::>" , <NUM_LIT:1> ) <EOL> parsed = urlparse ( url , scheme = "<STR_LIT:http>" ) <EOL> if parsed . hostname : <EOL> hostname = parsed . hostname <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> port = <NUM_LIT:0> <EOL> if parsed . port : <EOL> port = parsed . port <EOL> is_secure = False <EOL> if scheme == "<STR_LIT>" : <EOL> if not port : <EOL> port = <NUM_LIT> <EOL> elif scheme == "<STR_LIT>" : <EOL> is_secure = True <EOL> if not port : <EOL> port = <NUM_LIT> <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" % scheme ) <EOL> if parsed . path : <EOL> resource = parsed . path <EOL> else : <EOL> resource = "<STR_LIT:/>" <EOL> if parsed . query : <EOL> resource += "<STR_LIT:?>" + parsed . query <EOL> return ( hostname , port , resource , is_secure ) <EOL> def create_connection ( url , timeout = None , ** options ) : <EOL> """<STR_LIT>""" <EOL> sockopt = options . get ( "<STR_LIT>" , ( ) ) <EOL> websock = WebSocket ( sockopt = sockopt ) <EOL> websock . settimeout ( timeout != None and timeout or default_timeout ) <EOL> websock . connect ( url , ** options ) <EOL> return websock <EOL> _MAX_INTEGER = ( <NUM_LIT:1> << <NUM_LIT:32> ) - <NUM_LIT:1> <EOL> _AVAILABLE_KEY_CHARS = list ( range ( <NUM_LIT> , <NUM_LIT> + <NUM_LIT:1> ) ) + list ( range ( <NUM_LIT> , <NUM_LIT> + <NUM_LIT:1> ) ) <EOL> _MAX_CHAR_BYTE = ( <NUM_LIT:1> << <NUM_LIT:8> ) - <NUM_LIT:1> <EOL> def _create_sec_websocket_key ( ) : <EOL> uid = uuid . uuid4 ( ) <EOL> return base64 . encodebytes ( uid . bytes ) . strip ( ) . decode ( "<STR_LIT:utf-8>" ) <EOL> _HEADERS_TO_CHECK = { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } <EOL> class _SSLSocketWrapper ( object ) : <EOL> def __init__ ( self , sock ) : <EOL> self . _ssl = ssl . wrap_socket ( sock ) <EOL> def recv ( self , bufsize ) : <EOL> return self . _ssl . read ( bufsize ) <EOL> def send ( self , payload ) : <EOL> return self . _ssl . write ( payload ) <EOL> def close ( self ) : <EOL> self . _ssl . close ( ) <EOL> _BOOL_VALUES = ( <NUM_LIT:0> , <NUM_LIT:1> ) <EOL> def _is_bool ( * values ) : <EOL> for v in values : <EOL> if v not in _BOOL_VALUES : <EOL> return False <EOL> return True <EOL> class ABNF ( object ) : <EOL> """<STR_LIT>""" <EOL> OPCODE_TEXT = <NUM_LIT> <EOL> OPCODE_BINARY = <NUM_LIT> <EOL> OPCODE_CLOSE = <NUM_LIT> <EOL> OPCODE_PING = <NUM_LIT> <EOL> OPCODE_PONG = <NUM_LIT> <EOL> OPCODES = ( OPCODE_TEXT , OPCODE_BINARY , OPCODE_CLOSE , <EOL> OPCODE_PING , OPCODE_PONG ) <EOL> OPCODE_MAP = { <EOL> OPCODE_TEXT : "<STR_LIT:text>" , <EOL> OPCODE_BINARY : "<STR_LIT>" , <EOL> OPCODE_CLOSE : "<STR_LIT>" , <EOL> OPCODE_PING : "<STR_LIT>" , <EOL> OPCODE_PONG : "<STR_LIT>" <EOL> } <EOL> LENGTH_7 = <NUM_LIT> <EOL> LENGTH_16 = <NUM_LIT:1> << <NUM_LIT:16> <EOL> LENGTH_63 = <NUM_LIT:1> << <NUM_LIT> <EOL> def __init__ ( self , fin = <NUM_LIT:0> , rsv1 = <NUM_LIT:0> , rsv2 = <NUM_LIT:0> , rsv3 = <NUM_LIT:0> , <EOL> opcode = OPCODE_TEXT , mask = <NUM_LIT:1> , data = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> self . fin = fin <EOL> self . rsv1 = rsv1 <EOL> self . rsv2 = rsv2 <EOL> self . rsv3 = rsv3 <EOL> self . opcode = opcode <EOL> self . mask = mask <EOL> self . data = data <EOL> self . get_mask_key = os . urandom <EOL> @ staticmethod <EOL> def create_frame ( data , opcode ) : <EOL> """<STR_LIT>""" <EOL> if opcode == ABNF . OPCODE_TEXT and isinstance ( data , str ) : <EOL> data = data . encode ( "<STR_LIT:utf-8>" ) <EOL> return ABNF ( <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , opcode , <NUM_LIT:1> , data ) <EOL> def format ( self ) : <EOL> """<STR_LIT>""" <EOL> if not _is_bool ( self . fin , self . rsv1 , self . rsv2 , self . rsv3 ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if self . opcode not in ABNF . OPCODES : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> length = len ( self . data ) <EOL> if length >= ABNF . LENGTH_63 : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> frame_header = [ ] <EOL> frame_header = bytes ( ( self . fin << <NUM_LIT:7> <EOL> | self . rsv1 << <NUM_LIT:6> | self . rsv2 << <NUM_LIT:5> | self . rsv3 << <NUM_LIT:4> <EOL> | self . opcode , ) ) <EOL> if length < ABNF . LENGTH_7 : <EOL> frame_header += bytes ( ( self . mask << <NUM_LIT:7> | length , ) ) <EOL> elif length < ABNF . LENGTH_16 : <EOL> frame_header += bytes ( ( self . mask << <NUM_LIT:7> | <NUM_LIT> , ) ) <EOL> frame_header += struct . pack ( "<STR_LIT>" , length ) <EOL> else : <EOL> frame_header += bytes ( ( self . mask << <NUM_LIT:7> | <NUM_LIT> , ) ) <EOL> frame_header += struct . pack ( "<STR_LIT>" , length ) <EOL> if not self . mask : <EOL> return frame_header + self . data <EOL> else : <EOL> mask_key = self . get_mask_key ( <NUM_LIT:4> ) <EOL> return frame_header + self . _get_masked ( mask_key ) <EOL> def _get_masked ( self , mask_key ) : <EOL> s = ABNF . mask ( mask_key , self . data ) <EOL> return mask_key + s <EOL> @ staticmethod <EOL> def mask ( mask_key , data ) : <EOL> """<STR_LIT>""" <EOL> _m = array . array ( "<STR_LIT:B>" , mask_key ) <EOL> _d = array . array ( "<STR_LIT:B>" , data ) <EOL> for i in range ( len ( _d ) ) : <EOL> _d [ i ] ^= _m [ i % <NUM_LIT:4> ] <EOL> return _d . tobytes ( ) <EOL> class WebSocket ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , get_mask_key = None , sockopt = ( ) ) : <EOL> """<STR_LIT>""" <EOL> self . connected = False <EOL> self . sock = socket . socket ( ) <EOL> for opts in sockopt : <EOL> self . sock . setsockopt ( * opts ) <EOL> self . get_mask_key = get_mask_key <EOL> def set_mask_key ( self , func ) : <EOL> """<STR_LIT>""" <EOL> self . get_mask_key = func <EOL> def settimeout ( self , timeout ) : <EOL> """<STR_LIT>""" <EOL> self . sock . settimeout ( timeout ) <EOL> def gettimeout ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . sock . gettimeout ( ) <EOL> def connect ( self , url , ** options ) : <EOL> """<STR_LIT>""" <EOL> hostname , port , resource , is_secure = _parse_url ( url ) <EOL> self . sock . connect ( ( hostname , port ) ) <EOL> if is_secure : <EOL> self . sock = _SSLSocketWrapper ( self . sock ) <EOL> self . _handshake ( hostname , port , resource , ** options ) <EOL> def _handshake ( self , host , port , resource , ** options ) : <EOL> sock = self . sock <EOL> headers = [ ] <EOL> headers . append ( "<STR_LIT>" % resource ) <EOL> headers . append ( "<STR_LIT>" ) <EOL> headers . append ( "<STR_LIT>" ) <EOL> if port == <NUM_LIT> : <EOL> hostport = host <EOL> else : <EOL> hostport = "<STR_LIT>" % ( host , port ) <EOL> headers . append ( "<STR_LIT>" % hostport ) <EOL> if "<STR_LIT>" in options : <EOL> headers . append ( "<STR_LIT>" % options [ "<STR_LIT>" ] ) <EOL> else : <EOL> headers . append ( "<STR_LIT>" % hostport ) <EOL> key = _create_sec_websocket_key ( ) <EOL> headers . append ( "<STR_LIT>" % key ) <EOL> headers . append ( "<STR_LIT>" % VERSION ) <EOL> if "<STR_LIT>" in options : <EOL> headers . extend ( options [ "<STR_LIT>" ] ) <EOL> headers . append ( "<STR_LIT>" ) <EOL> headers . append ( "<STR_LIT>" ) <EOL> header_str = "<STR_LIT:\r\n>" . join ( headers ) <EOL> sock . send ( bytes ( header_str , "<STR_LIT:utf-8>" ) ) <EOL> if traceEnabled : <EOL> logger . debug ( "<STR_LIT>" ) <EOL> logger . debug ( header_str ) <EOL> logger . debug ( "<STR_LIT>" ) <EOL> status , resp_headers = self . _read_headers ( ) <EOL> if status != <NUM_LIT> : <EOL> self . close ( ) <EOL> raise WebSocketException ( "<STR_LIT>" % status ) <EOL> success = self . _validate_header ( resp_headers , key ) <EOL> if not success : <EOL> self . close ( ) <EOL> raise WebSocketException ( "<STR_LIT>" ) <EOL> self . connected = True <EOL> def _validate_header ( self , headers , key ) : <EOL> for k , v in _HEADERS_TO_CHECK . items ( ) : <EOL> r = headers . get ( k , None ) <EOL> if not r : <EOL> return False <EOL> r = r . lower ( ) <EOL> if v != r : <EOL> return False <EOL> result = headers . get ( "<STR_LIT>" , None ) <EOL> if not result : <EOL> return False <EOL> result = result . lower ( ) <EOL> value = key + "<STR_LIT>" <EOL> digest = hashlib . sha1 ( bytes ( value , "<STR_LIT:utf-8>" ) ) . digest ( ) <EOL> hashed = base64 . encodebytes ( digest ) . strip ( ) . decode ( "<STR_LIT:utf-8>" ) . lower ( ) <EOL> return hashed == result <EOL> def _read_headers ( self ) : <EOL> status = None <EOL> headers = { } <EOL> if traceEnabled : <EOL> logger . debug ( "<STR_LIT>" ) <EOL> while True : <EOL> line = self . _recv_line ( ) <EOL> if line == "<STR_LIT:\r\n>" : <EOL> break <EOL> line = line . strip ( ) <EOL> if traceEnabled : <EOL> logger . debug ( line ) <EOL> if not status : <EOL> status_info = line . split ( "<STR_LIT:U+0020>" , <NUM_LIT:2> ) <EOL> status = int ( status_info [ <NUM_LIT:1> ] ) <EOL> else : <EOL> kv = line . split ( "<STR_LIT::>" , <NUM_LIT:1> ) <EOL> if len ( kv ) == <NUM_LIT:2> : <EOL> key , value = kv <EOL> headers [ key . lower ( ) ] = value . strip ( ) . lower ( ) <EOL> else : <EOL> raise WebSocketException ( "<STR_LIT>" ) <EOL> if traceEnabled : <EOL> logger . debug ( "<STR_LIT>" ) <EOL> return status , headers <EOL> def send ( self , payload , opcode = ABNF . OPCODE_TEXT ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( payload , str ) : <EOL> payload = bytes ( payload , "<STR_LIT:utf-8>" ) <EOL> frame = ABNF . create_frame ( payload , opcode ) <EOL> if self . get_mask_key : <EOL> frame . get_mask_key = self . get_mask_key <EOL> data = frame . format ( ) <EOL> while data : <EOL> l = self . sock . send ( data ) <EOL> data = data [ l : ] <EOL> if traceEnabled : <EOL> logger . debug ( "<STR_LIT>" + repr ( data ) ) <EOL> def ping ( self , payload = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> self . send ( payload , ABNF . OPCODE_PING ) <EOL> def pong ( self , payload ) : <EOL> """<STR_LIT>""" <EOL> self . send ( payload , ABNF . OPCODE_PONG ) <EOL> def recv ( self ) : <EOL> """<STR_LIT>""" <EOL> opcode , data = self . recv_data ( ) <EOL> if opcode == ABNF . OPCODE_TEXT : <EOL> return data . decode ( "<STR_LIT:utf-8>" ) <EOL> return data <EOL> def recv_data ( self ) : <EOL> """<STR_LIT>""" <EOL> while True : <EOL> frame = self . recv_frame ( ) <EOL> if not frame : <EOL> raise WebSocketException ( "<STR_LIT>" % frame ) <EOL> elif frame . opcode in ( ABNF . OPCODE_TEXT , ABNF . OPCODE_BINARY ) : <EOL> return ( frame . opcode , frame . data ) <EOL> elif frame . opcode == ABNF . OPCODE_CLOSE : <EOL> self . send_close ( ) <EOL> return ( frame . opcode , None ) <EOL> elif frame . opcode == ABNF . OPCODE_PING : <EOL> self . pong ( frame . data ) <EOL> def recv_frame ( self ) : <EOL> """<STR_LIT>""" <EOL> header_bytes = self . _recv_strict ( <NUM_LIT:2> ) <EOL> if not header_bytes : <EOL> return None <EOL> b1 = header_bytes [ <NUM_LIT:0> ] <EOL> fin = b1 >> <NUM_LIT:7> & <NUM_LIT:1> <EOL> rsv1 = b1 >> <NUM_LIT:6> & <NUM_LIT:1> <EOL> rsv2 = b1 >> <NUM_LIT:5> & <NUM_LIT:1> <EOL> rsv3 = b1 >> <NUM_LIT:4> & <NUM_LIT:1> <EOL> opcode = b1 & <NUM_LIT> <EOL> b2 = header_bytes [ <NUM_LIT:1> ] <EOL> mask = b2 >> <NUM_LIT:7> & <NUM_LIT:1> <EOL> length = b2 & <NUM_LIT> <EOL> length_data = b"<STR_LIT>" <EOL> if length == <NUM_LIT> : <EOL> length_data = self . _recv_strict ( <NUM_LIT:2> ) <EOL> length = struct . unpack ( "<STR_LIT>" , length_data ) [ <NUM_LIT:0> ] <EOL> elif length == <NUM_LIT> : <EOL> length_data = self . _recv_strict ( <NUM_LIT:8> ) <EOL> length = struct . unpack ( "<STR_LIT>" , length_data ) [ <NUM_LIT:0> ] <EOL> mask_key = b"<STR_LIT>" <EOL> if mask : <EOL> mask_key = self . _recv_strict ( <NUM_LIT:4> ) <EOL> data = self . _recv_strict ( length ) <EOL> if traceEnabled : <EOL> recieved = header_bytes + length_data + mask_key + data <EOL> logger . debug ( "<STR_LIT>" + repr ( recieved ) ) <EOL> if mask : <EOL> data = ABNF . mask ( mask_key , data ) <EOL> frame = ABNF ( fin , rsv1 , rsv2 , rsv3 , opcode , mask , data ) <EOL> return frame <EOL> def send_close ( self , status = STATUS_NORMAL , reason = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> if status < <NUM_LIT:0> or status >= ABNF . LENGTH_16 : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . send ( struct . pack ( '<STR_LIT>' , status ) + reason , ABNF . OPCODE_CLOSE ) <EOL> def close ( self , status = STATUS_NORMAL , reason = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> if self . connected : <EOL> if status < <NUM_LIT:0> or status >= ABNF . LENGTH_16 : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> try : <EOL> self . send ( struct . pack ( '<STR_LIT>' , status ) + reason , ABNF . OPCODE_CLOSE ) <EOL> timeout = self . sock . gettimeout ( ) <EOL> self . sock . settimeout ( <NUM_LIT:3> ) <EOL> try : <EOL> frame = self . recv_frame ( ) <EOL> if logger . isEnabledFor ( logging . DEBUG ) : <EOL> logger . error ( "<STR_LIT>" + repr ( frame . data ) ) <EOL> except : <EOL> pass <EOL> self . sock . settimeout ( timeout ) <EOL> self . sock . shutdown ( socket . SHUT_RDWR ) <EOL> except : <EOL> pass <EOL> self . _closeInternal ( ) <EOL> def _closeInternal ( self ) : <EOL> self . connected = False <EOL> self . sock . close ( ) <EOL> def _recv ( self , bufsize ) : <EOL> bytes = self . sock . recv ( bufsize ) <EOL> if not bytes : <EOL> raise WebSocketConnectionClosedException ( ) <EOL> return bytes <EOL> def _recv_strict ( self , bufsize ) : <EOL> remaining = bufsize <EOL> bytes = b"<STR_LIT>" <EOL> while remaining : <EOL> bytes += self . _recv ( remaining ) <EOL> remaining = bufsize - len ( bytes ) <EOL> return bytes <EOL> def _recv_line ( self ) : <EOL> line = [ ] <EOL> while True : <EOL> c = self . _recv ( <NUM_LIT:1> ) <EOL> line . append ( c ) <EOL> if c == b"<STR_LIT:\n>" : <EOL> break <EOL> return b"<STR_LIT>" . join ( line ) . decode ( "<STR_LIT:utf-8>" ) <EOL> class WebSocketApp ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , url , header = [ ] , <EOL> on_open = None , on_message = None , on_error = None , <EOL> on_close = None , keep_running = True , get_mask_key = None , <EOL> sockopt = ( ) ) : <EOL> """<STR_LIT>""" <EOL> self . url = url <EOL> self . header = header <EOL> self . on_open = on_open <EOL> self . on_message = on_message <EOL> self . on_error = on_error <EOL> self . on_close = on_close <EOL> self . keep_running = keep_running <EOL> self . get_mask_key = get_mask_key <EOL> self . sock = None <EOL> def send ( self , data , opcode = ABNF . OPCODE_TEXT ) : <EOL> """<STR_LIT>""" <EOL> if self . sock . send ( data , opcode ) == <NUM_LIT:0> : <EOL> raise WebSocketConnectionClosedException ( ) <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> self . keep_running = False <EOL> self . sock . close ( ) <EOL> def run_forever ( self , sockopt = ( ) ) : <EOL> """<STR_LIT>""" <EOL> if self . sock : <EOL> raise WebSocketException ( "<STR_LIT>" ) <EOL> try : <EOL> self . sock = WebSocket ( self . get_mask_key , sockopt = sockopt ) <EOL> self . sock . connect ( self . url , header = self . header ) <EOL> self . _run_with_no_err ( self . on_open ) <EOL> while self . keep_running : <EOL> data = self . sock . recv ( ) <EOL> if data is None : <EOL> break <EOL> self . _run_with_no_err ( self . on_message , data ) <EOL> except Exception as e : <EOL> self . _run_with_no_err ( self . on_error , e ) <EOL> finally : <EOL> self . sock . close ( ) <EOL> self . _run_with_no_err ( self . on_close ) <EOL> self . sock = None <EOL> def _run_with_no_err ( self , callback , * args ) : <EOL> if callback : <EOL> try : <EOL> callback ( self , * args ) <EOL> except Exception as e : <EOL> if logger . isEnabledFor ( logging . DEBUG ) : <EOL> logger . error ( e ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> enableTrace ( True ) <EOL> ws = create_connection ( "<STR_LIT>" ) <EOL> print ( "<STR_LIT>" ) <EOL> ws . send ( "<STR_LIT>" ) <EOL> print ( "<STR_LIT>" ) <EOL> print ( "<STR_LIT>" ) <EOL> result = ws . recv ( ) <EOL> print ( "<STR_LIT>" % result ) <EOL> ws . close ( ) </s>
