Python websockets 模块，InvalidHandshake() 实例源码

def websocket_upgrade(http):
    request_headers = dict(http.headers)
    response_headers = []

    def get_header(key):
        key = key.lower().encode('utf-8')
        return request_headers.get(key, b'').decode('utf-8')

    def set_header(key, val):
        response_headers.append((key.encode('utf-8'), val.encode('utf-8')))

    try:
        key = websockets.handshake.check_request(get_header)
        websockets.handshake.build_response(set_header, key)
    except websockets.InvalidHandshake:
        http.loop.create_task(http.channels['reply'].send({
            'status': 400,
            'headers': [[b'content-type', b'text/plain']],
            'content': b'Invalid WebSocket handshake'
        }))
        return

    protocol = WebSocketProtocol(http, response_headers)
    protocol.connection_made(http.transport, http.message)
    http.transport.set_protocol(protocol)

def bad_websocket_handshake(*args, **kwargs):
        raise InvalidHandshake("Bad status code: 200")

def upgrade_to_websocket(http_protocol):
    request_headers = dict(http_protocol.headers)
    response_headers = []

    def get_header(key):
        key = key.lower().encode('utf-8')
        return request_headers.get(key, b'').decode('utf-8')

    def set_header(key, value):
        response_headers.append((key.encode('utf-8'), value.encode('utf-8')))

    try:
        key = websockets.handshake.check_request(get_header)
        websockets.handshake.build_response(set_header, key)
    except websockets.InvalidHandshake:
        http_protocol.send({
            'status': 400,
            'headers': [[b'Content-Type', b'text/plain']],
            'content': b'Invalid Handshake'})
    websocket = WebSocketProtocol(http_protocol, response_headers)
    websocket.connection_made(http_protocol.transport,
                              http_protocol.parser.get_http_version(),
                              http_protocol.parser.get_method(),
                              http_protocol.url,
                              http_protocol.headers)
    http_protocol.transport.set_protocol(websocket)

def websocket_handshake(self, request, subprotocols=None):
        # let the websockets package do the handshake with the client
        headers = []

        def get_header(k):
            return request.headers.get(k, '')

        def set_header(k, v):
            headers.append((k, v))

        try:
            key = handshake.check_request(get_header)
            handshake.build_response(set_header, key)
        except InvalidHandshake:
            raise InvalidUsage('Invalid websocket request')

        subprotocol = None
        if subprotocols and 'Sec-Websocket-Protocol' in request.headers:
            # select a subprotocol
            client_subprotocols = [p.strip() for p in request.headers[
                'Sec-Websocket-Protocol'].split(',')]
            for p in client_subprotocols:
                if p in subprotocols:
                    subprotocol = p
                    set_header('Sec-Websocket-Protocol', subprotocol)
                    break

        # write the 101 response back to the client
        rv = b'HTTP/1.1 101 Switching Protocols\r\n'
        for k, v in headers:
            rv += k.encode('utf-8') + b': ' + v.encode('utf-8') + b'\r\n'
        rv += b'\r\n'
        request.transport.write(rv)

        # hook up the websocket protocol
        self.websocket = WebSocketCommonProtocol(
            max_size=self.websocket_max_size,
            max_queue=self.websocket_max_queue
        )
        self.websocket.subprotocol = subprotocol
        self.websocket.connection_made(request.transport)
        self.websocket.connection_open()
        return self.websocket