<s> """<STR_LIT>""" <EOL> from mod_pywebsocket import common <EOL> from mod_pywebsocket import http_header_util <EOL> class AbortedByUserException ( Exception ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class HandshakeException ( Exception ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , name , status = None ) : <EOL> super ( HandshakeException , self ) . __init__ ( name ) <EOL> self . status = status <EOL> class VersionException ( Exception ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , name , supported_versions = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> super ( VersionException , self ) . __init__ ( name ) <EOL> self . supported_versions = supported_versions <EOL> def get_default_port ( is_secure ) : <EOL> if is_secure : <EOL> return common . DEFAULT_WEB_SOCKET_SECURE_PORT <EOL> else : <EOL> return common . DEFAULT_WEB_SOCKET_PORT <EOL> def validate_subprotocol ( subprotocol ) : <EOL> """<STR_LIT>""" <EOL> if not subprotocol : <EOL> raise HandshakeException ( '<STR_LIT>' ) <EOL> state = http_header_util . ParsingState ( subprotocol ) <EOL> token = http_header_util . consume_token ( state ) <EOL> rest = http_header_util . peek ( state ) <EOL> if rest is not None : <EOL> raise HandshakeException ( '<STR_LIT>' <EOL> '<STR_LIT>' % rest ) <EOL> def parse_host_header ( request ) : <EOL> fields = request . headers_in [ common . HOST_HEADER ] . split ( '<STR_LIT::>' , <NUM_LIT:1> ) <EOL> if len ( fields ) == <NUM_LIT:1> : <EOL> return fields [ <NUM_LIT:0> ] , get_default_port ( request . is_https ( ) ) <EOL> try : <EOL> return fields [ <NUM_LIT:0> ] , int ( fields [ <NUM_LIT:1> ] ) <EOL> except ValueError , e : <EOL> raise HandshakeException ( '<STR_LIT>' % e ) <EOL> def format_header ( name , value ) : <EOL> return '<STR_LIT>' % ( name , value ) <EOL> def get_mandatory_header ( request , key ) : <EOL> value = request . headers_in . get ( key ) <EOL> if value is None : <EOL> raise HandshakeException ( '<STR_LIT>' % key ) <EOL> return value <EOL> def validate_mandatory_header ( request , key , expected_value , fail_status = None ) : <EOL> value = get_mandatory_header ( request , key ) <EOL> if value . lower ( ) != expected_value . lower ( ) : <EOL> raise HandshakeException ( <EOL> '<STR_LIT>' % <EOL> ( expected_value , key , value ) , status = fail_status ) <EOL> def check_request_line ( request ) : <EOL> if request . method != '<STR_LIT:GET>' : <EOL> raise HandshakeException ( '<STR_LIT>' % request . method ) <EOL> if request . protocol != '<STR_LIT>' : <EOL> raise HandshakeException ( '<STR_LIT>' % <EOL> request . protocol ) <EOL> def parse_token_list ( data ) : <EOL> """<STR_LIT>""" <EOL> state = http_header_util . ParsingState ( data ) <EOL> token_list = [ ] <EOL> while True : <EOL> token = http_header_util . consume_token ( state ) <EOL> if token is not None : <EOL> token_list . append ( token ) <EOL> http_header_util . consume_lwses ( state ) <EOL> if http_header_util . peek ( state ) is None : <EOL> break <EOL> if not http_header_util . consume_string ( state , '<STR_LIT:U+002C>' ) : <EOL> raise HandshakeException ( <EOL> '<STR_LIT>' % http_header_util . peek ( state ) ) <EOL> http_header_util . consume_lwses ( state ) <EOL> if len ( token_list ) == <NUM_LIT:0> : <EOL> raise HandshakeException ( '<STR_LIT>' ) <EOL> return token_list </s>
