Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def process_request(self, path, headers):
"""
This hook is called to determine if the websocket should return
an HTTP response and close.
Our behavior here is to start the ASGI application, and then wait
for either `accept` or `close` in order to determine if we should
close the connection.
"""
path_portion, _, query_string = path.partition("?")
websockets.handshake.check_request(headers)
subprotocols = []
for header in headers.get_all("Sec-WebSocket-Protocol"):
subprotocols.extend([token.strip() for token in header.split(",")])
headers = [
(name.encode("ascii"), value.encode("ascii"))
for name, value in headers.raw_items()
]
self.environ = {
"type": "websocket",
"scheme": self.scheme,
"server": self.server,
"client": self.client,
"method": "GET",
def handshake(self, response):
"""Websocket handshake, handled by `websockets`
"""
try:
headers = websockets.http.Headers(**self.request.headers)
key = websockets.handshake.check_request(headers)
websockets.handshake.build_response(response.headers, key)
except websockets.InvalidHandshake:
raise RuntimeError('Invalid websocket request')
subprotocol = None
ws_protocol = ','.join(headers.get_all('Sec-Websocket-Protocol'))
subprotocols = self.request.route.payload.get('subprotocols')
if subprotocols and ws_protocol:
# select a subprotocol
client_subprotocols = tuple(
(p.strip() for p in ws_protocol.split(',')))
for p in client_subprotocols:
if p in subprotocols:
subprotocol = p
response.headers['Sec-Websocket-Protocol'] = subprotocol
break
Try to perform the server side of the opening websocket handshake.
If it fails, switch self._type to HTTP and return.
Returns the (method, url, headers, body)
Copy of WebSocketServerProtocol.handshake with HTTP flavour.
"""
# Read handshake request.
try:
method, url, headers, body = yield from read_request(self.reader)
except Exception as exc:
raise HTTPError(code=500) from exc
get_header = lambda k: headers.get(k, '')
try:
key = check_request(get_header)
except InvalidHandshake:
self._type = 'HTTP' # switching to HTTP here
return (method, url, headers, body)
# Send handshake response. Since the headers only contain ASCII
# characters, we can keep this simple.
response = ['HTTP/1.1 101 Switching Protocols']
set_header = lambda k, v: response.append('{}: {}'.format(k, v))
set_header('Server', USER_AGENT)
build_response(set_header, key)
response.append('\r\n')
response = '\r\n'.join(response).encode()
self.writer.write(response)
self.state = 'OPEN'
self.opening_handshake.set_result(True)