websocket.py 25 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671
  1. import base64
  2. import codecs
  3. import collections
  4. import errno
  5. from random import Random
  6. from socket import error as SocketError
  7. import string
  8. import struct
  9. import sys
  10. import time
  11. try:
  12. from hashlib import md5, sha1
  13. except ImportError: # pragma NO COVER
  14. from md5 import md5
  15. from sha import sha as sha1
  16. from eventlet import semaphore
  17. from eventlet import wsgi
  18. from eventlet.green import socket
  19. from eventlet.support import get_errno, six
  20. # Python 2's utf8 decoding is more lenient than we'd like
  21. # In order to pass autobahn's testsuite we need stricter validation
  22. # if available...
  23. for _mod in ('wsaccel.utf8validator', 'autobahn.utf8validator'):
  24. # autobahn has it's own python-based validator. in newest versions
  25. # this prefers to use wsaccel, a cython based implementation, if available.
  26. # wsaccel may also be installed w/out autobahn, or with a earlier version.
  27. try:
  28. utf8validator = __import__(_mod, {}, {}, [''])
  29. except ImportError:
  30. utf8validator = None
  31. else:
  32. break
  33. ACCEPTABLE_CLIENT_ERRORS = set((errno.ECONNRESET, errno.EPIPE))
  34. __all__ = ["WebSocketWSGI", "WebSocket"]
  35. PROTOCOL_GUID = b'258EAFA5-E914-47DA-95CA-C5AB0DC85B11'
  36. VALID_CLOSE_STATUS = set(
  37. list(range(1000, 1004)) +
  38. list(range(1007, 1012)) +
  39. # 3000-3999: reserved for use by libraries, frameworks,
  40. # and applications
  41. list(range(3000, 4000)) +
  42. # 4000-4999: reserved for private use and thus can't
  43. # be registered
  44. list(range(4000, 5000))
  45. )
  46. class BadRequest(Exception):
  47. def __init__(self, status='400 Bad Request', body=None, headers=None):
  48. super(Exception, self).__init__()
  49. self.status = status
  50. self.body = body
  51. self.headers = headers
  52. class WebSocketWSGI(object):
  53. """Wraps a websocket handler function in a WSGI application.
  54. Use it like this::
  55. @websocket.WebSocketWSGI
  56. def my_handler(ws):
  57. from_browser = ws.wait()
  58. ws.send("from server")
  59. The single argument to the function will be an instance of
  60. :class:`WebSocket`. To close the socket, simply return from the
  61. function. Note that the server will log the websocket request at
  62. the time of closure.
  63. """
  64. def __init__(self, handler):
  65. self.handler = handler
  66. self.protocol_version = None
  67. self.support_legacy_versions = True
  68. self.supported_protocols = []
  69. self.origin_checker = None
  70. @classmethod
  71. def configured(cls,
  72. handler=None,
  73. supported_protocols=None,
  74. origin_checker=None,
  75. support_legacy_versions=False):
  76. def decorator(handler):
  77. inst = cls(handler)
  78. inst.support_legacy_versions = support_legacy_versions
  79. inst.origin_checker = origin_checker
  80. if supported_protocols:
  81. inst.supported_protocols = supported_protocols
  82. return inst
  83. if handler is None:
  84. return decorator
  85. return decorator(handler)
  86. def __call__(self, environ, start_response):
  87. http_connection_parts = [
  88. part.strip()
  89. for part in environ.get('HTTP_CONNECTION', '').lower().split(',')]
  90. if not ('upgrade' in http_connection_parts and
  91. environ.get('HTTP_UPGRADE', '').lower() == 'websocket'):
  92. # need to check a few more things here for true compliance
  93. start_response('400 Bad Request', [('Connection', 'close')])
  94. return []
  95. try:
  96. if 'HTTP_SEC_WEBSOCKET_VERSION' in environ:
  97. ws = self._handle_hybi_request(environ)
  98. elif self.support_legacy_versions:
  99. ws = self._handle_legacy_request(environ)
  100. else:
  101. raise BadRequest()
  102. except BadRequest as e:
  103. status = e.status
  104. body = e.body or b''
  105. headers = e.headers or []
  106. start_response(status,
  107. [('Connection', 'close'), ] + headers)
  108. return [body]
  109. try:
  110. self.handler(ws)
  111. except socket.error as e:
  112. if get_errno(e) not in ACCEPTABLE_CLIENT_ERRORS:
  113. raise
  114. # Make sure we send the closing frame
  115. ws._send_closing_frame(True)
  116. # use this undocumented feature of eventlet.wsgi to ensure that it
  117. # doesn't barf on the fact that we didn't call start_response
  118. return wsgi.ALREADY_HANDLED
  119. def _handle_legacy_request(self, environ):
  120. if 'eventlet.input' in environ:
  121. sock = environ['eventlet.input'].get_socket()
  122. elif 'gunicorn.socket' in environ:
  123. sock = environ['gunicorn.socket']
  124. else:
  125. raise Exception('No eventlet.input or gunicorn.socket present in environ.')
  126. if 'HTTP_SEC_WEBSOCKET_KEY1' in environ:
  127. self.protocol_version = 76
  128. if 'HTTP_SEC_WEBSOCKET_KEY2' not in environ:
  129. raise BadRequest()
  130. else:
  131. self.protocol_version = 75
  132. if self.protocol_version == 76:
  133. key1 = self._extract_number(environ['HTTP_SEC_WEBSOCKET_KEY1'])
  134. key2 = self._extract_number(environ['HTTP_SEC_WEBSOCKET_KEY2'])
  135. # There's no content-length header in the request, but it has 8
  136. # bytes of data.
  137. environ['wsgi.input'].content_length = 8
  138. key3 = environ['wsgi.input'].read(8)
  139. key = struct.pack(">II", key1, key2) + key3
  140. response = md5(key).digest()
  141. # Start building the response
  142. scheme = 'ws'
  143. if environ.get('wsgi.url_scheme') == 'https':
  144. scheme = 'wss'
  145. location = '%s://%s%s%s' % (
  146. scheme,
  147. environ.get('HTTP_HOST'),
  148. environ.get('SCRIPT_NAME'),
  149. environ.get('PATH_INFO')
  150. )
  151. qs = environ.get('QUERY_STRING')
  152. if qs is not None:
  153. location += '?' + qs
  154. if self.protocol_version == 75:
  155. handshake_reply = (
  156. b"HTTP/1.1 101 Web Socket Protocol Handshake\r\n"
  157. b"Upgrade: WebSocket\r\n"
  158. b"Connection: Upgrade\r\n"
  159. b"WebSocket-Origin: " + six.b(environ.get('HTTP_ORIGIN')) + b"\r\n"
  160. b"WebSocket-Location: " + six.b(location) + b"\r\n\r\n"
  161. )
  162. elif self.protocol_version == 76:
  163. handshake_reply = (
  164. b"HTTP/1.1 101 WebSocket Protocol Handshake\r\n"
  165. b"Upgrade: WebSocket\r\n"
  166. b"Connection: Upgrade\r\n"
  167. b"Sec-WebSocket-Origin: " + six.b(environ.get('HTTP_ORIGIN')) + b"\r\n"
  168. b"Sec-WebSocket-Protocol: " +
  169. six.b(environ.get('HTTP_SEC_WEBSOCKET_PROTOCOL', 'default')) + b"\r\n"
  170. b"Sec-WebSocket-Location: " + six.b(location) + b"\r\n"
  171. b"\r\n" + response
  172. )
  173. else: # pragma NO COVER
  174. raise ValueError("Unknown WebSocket protocol version.")
  175. sock.sendall(handshake_reply)
  176. return WebSocket(sock, environ, self.protocol_version)
  177. def _handle_hybi_request(self, environ):
  178. if 'eventlet.input' in environ:
  179. sock = environ['eventlet.input'].get_socket()
  180. elif 'gunicorn.socket' in environ:
  181. sock = environ['gunicorn.socket']
  182. else:
  183. raise Exception('No eventlet.input or gunicorn.socket present in environ.')
  184. hybi_version = environ['HTTP_SEC_WEBSOCKET_VERSION']
  185. if hybi_version not in ('8', '13', ):
  186. raise BadRequest(status='426 Upgrade Required',
  187. headers=[('Sec-WebSocket-Version', '8, 13')])
  188. self.protocol_version = int(hybi_version)
  189. if 'HTTP_SEC_WEBSOCKET_KEY' not in environ:
  190. # That's bad.
  191. raise BadRequest()
  192. origin = environ.get(
  193. 'HTTP_ORIGIN',
  194. (environ.get('HTTP_SEC_WEBSOCKET_ORIGIN', '')
  195. if self.protocol_version <= 8 else ''))
  196. if self.origin_checker is not None:
  197. if not self.origin_checker(environ.get('HTTP_HOST'), origin):
  198. raise BadRequest(status='403 Forbidden')
  199. protocols = environ.get('HTTP_SEC_WEBSOCKET_PROTOCOL', None)
  200. negotiated_protocol = None
  201. if protocols:
  202. for p in (i.strip() for i in protocols.split(',')):
  203. if p in self.supported_protocols:
  204. negotiated_protocol = p
  205. break
  206. # extensions = environ.get('HTTP_SEC_WEBSOCKET_EXTENSIONS', None)
  207. # if extensions:
  208. # extensions = [i.strip() for i in extensions.split(',')]
  209. key = environ['HTTP_SEC_WEBSOCKET_KEY']
  210. response = base64.b64encode(sha1(six.b(key) + PROTOCOL_GUID).digest())
  211. handshake_reply = [b"HTTP/1.1 101 Switching Protocols",
  212. b"Upgrade: websocket",
  213. b"Connection: Upgrade",
  214. b"Sec-WebSocket-Accept: " + response]
  215. if negotiated_protocol:
  216. handshake_reply.append(b"Sec-WebSocket-Protocol: " + six.b(negotiated_protocol))
  217. sock.sendall(b'\r\n'.join(handshake_reply) + b'\r\n\r\n')
  218. return RFC6455WebSocket(sock, environ, self.protocol_version,
  219. protocol=negotiated_protocol)
  220. def _extract_number(self, value):
  221. """
  222. Utility function which, given a string like 'g98sd 5[]221@1', will
  223. return 9852211. Used to parse the Sec-WebSocket-Key headers.
  224. """
  225. out = ""
  226. spaces = 0
  227. for char in value:
  228. if char in string.digits:
  229. out += char
  230. elif char == " ":
  231. spaces += 1
  232. return int(out) // spaces
  233. class WebSocket(object):
  234. """A websocket object that handles the details of
  235. serialization/deserialization to the socket.
  236. The primary way to interact with a :class:`WebSocket` object is to
  237. call :meth:`send` and :meth:`wait` in order to pass messages back
  238. and forth with the browser. Also available are the following
  239. properties:
  240. path
  241. The path value of the request. This is the same as the WSGI PATH_INFO variable,
  242. but more convenient.
  243. protocol
  244. The value of the Websocket-Protocol header.
  245. origin
  246. The value of the 'Origin' header.
  247. environ
  248. The full WSGI environment for this request.
  249. """
  250. def __init__(self, sock, environ, version=76):
  251. """
  252. :param socket: The eventlet socket
  253. :type socket: :class:`eventlet.greenio.GreenSocket`
  254. :param environ: The wsgi environment
  255. :param version: The WebSocket spec version to follow (default is 76)
  256. """
  257. self.socket = sock
  258. self.origin = environ.get('HTTP_ORIGIN')
  259. self.protocol = environ.get('HTTP_WEBSOCKET_PROTOCOL')
  260. self.path = environ.get('PATH_INFO')
  261. self.environ = environ
  262. self.version = version
  263. self.websocket_closed = False
  264. self._buf = b""
  265. self._msgs = collections.deque()
  266. self._sendlock = semaphore.Semaphore()
  267. @staticmethod
  268. def _pack_message(message):
  269. """Pack the message inside ``00`` and ``FF``
  270. As per the dataframing section (5.3) for the websocket spec
  271. """
  272. if isinstance(message, six.text_type):
  273. message = message.encode('utf-8')
  274. elif not isinstance(message, six.binary_type):
  275. message = six.b(str(message))
  276. packed = b"\x00" + message + b"\xFF"
  277. return packed
  278. def _parse_messages(self):
  279. """ Parses for messages in the buffer *buf*. It is assumed that
  280. the buffer contains the start character for a message, but that it
  281. may contain only part of the rest of the message.
  282. Returns an array of messages, and the buffer remainder that
  283. didn't contain any full messages."""
  284. msgs = []
  285. end_idx = 0
  286. buf = self._buf
  287. while buf:
  288. frame_type = six.indexbytes(buf, 0)
  289. if frame_type == 0:
  290. # Normal message.
  291. end_idx = buf.find(b"\xFF")
  292. if end_idx == -1: # pragma NO COVER
  293. break
  294. msgs.append(buf[1:end_idx].decode('utf-8', 'replace'))
  295. buf = buf[end_idx + 1:]
  296. elif frame_type == 255:
  297. # Closing handshake.
  298. assert six.indexbytes(buf, 1) == 0, "Unexpected closing handshake: %r" % buf
  299. self.websocket_closed = True
  300. break
  301. else:
  302. raise ValueError("Don't understand how to parse this type of message: %r" % buf)
  303. self._buf = buf
  304. return msgs
  305. def send(self, message):
  306. """Send a message to the browser.
  307. *message* should be convertable to a string; unicode objects should be
  308. encodable as utf-8. Raises socket.error with errno of 32
  309. (broken pipe) if the socket has already been closed by the client."""
  310. packed = self._pack_message(message)
  311. # if two greenthreads are trying to send at the same time
  312. # on the same socket, sendlock prevents interleaving and corruption
  313. self._sendlock.acquire()
  314. try:
  315. self.socket.sendall(packed)
  316. finally:
  317. self._sendlock.release()
  318. def wait(self):
  319. """Waits for and deserializes messages.
  320. Returns a single message; the oldest not yet processed. If the client
  321. has already closed the connection, returns None. This is different
  322. from normal socket behavior because the empty string is a valid
  323. websocket message."""
  324. while not self._msgs:
  325. # Websocket might be closed already.
  326. if self.websocket_closed:
  327. return None
  328. # no parsed messages, must mean buf needs more data
  329. delta = self.socket.recv(8096)
  330. if delta == b'':
  331. return None
  332. self._buf += delta
  333. msgs = self._parse_messages()
  334. self._msgs.extend(msgs)
  335. return self._msgs.popleft()
  336. def _send_closing_frame(self, ignore_send_errors=False):
  337. """Sends the closing frame to the client, if required."""
  338. if self.version == 76 and not self.websocket_closed:
  339. try:
  340. self.socket.sendall(b"\xff\x00")
  341. except SocketError:
  342. # Sometimes, like when the remote side cuts off the connection,
  343. # we don't care about this.
  344. if not ignore_send_errors: # pragma NO COVER
  345. raise
  346. self.websocket_closed = True
  347. def close(self):
  348. """Forcibly close the websocket; generally it is preferable to
  349. return from the handler method."""
  350. self._send_closing_frame()
  351. self.socket.shutdown(True)
  352. self.socket.close()
  353. class ConnectionClosedError(Exception):
  354. pass
  355. class FailedConnectionError(Exception):
  356. def __init__(self, status, message):
  357. super(FailedConnectionError, self).__init__(status, message)
  358. self.message = message
  359. self.status = status
  360. class ProtocolError(ValueError):
  361. pass
  362. class RFC6455WebSocket(WebSocket):
  363. def __init__(self, sock, environ, version=13, protocol=None, client=False):
  364. super(RFC6455WebSocket, self).__init__(sock, environ, version)
  365. self.iterator = self._iter_frames()
  366. self.client = client
  367. self.protocol = protocol
  368. class UTF8Decoder(object):
  369. def __init__(self):
  370. if utf8validator:
  371. self.validator = utf8validator.Utf8Validator()
  372. else:
  373. self.validator = None
  374. decoderclass = codecs.getincrementaldecoder('utf8')
  375. self.decoder = decoderclass()
  376. def reset(self):
  377. if self.validator:
  378. self.validator.reset()
  379. self.decoder.reset()
  380. def decode(self, data, final=False):
  381. if self.validator:
  382. valid, eocp, c_i, t_i = self.validator.validate(data)
  383. if not valid:
  384. raise ValueError('Data is not valid unicode')
  385. return self.decoder.decode(data, final)
  386. def _get_bytes(self, numbytes):
  387. data = b''
  388. while len(data) < numbytes:
  389. d = self.socket.recv(numbytes - len(data))
  390. if not d:
  391. raise ConnectionClosedError()
  392. data = data + d
  393. return data
  394. class Message(object):
  395. def __init__(self, opcode, decoder=None):
  396. self.decoder = decoder
  397. self.data = []
  398. self.finished = False
  399. self.opcode = opcode
  400. def push(self, data, final=False):
  401. if self.decoder:
  402. data = self.decoder.decode(data, final=final)
  403. self.finished = final
  404. self.data.append(data)
  405. def getvalue(self):
  406. return ('' if self.decoder else b'').join(self.data)
  407. @staticmethod
  408. def _apply_mask(data, mask, length=None, offset=0):
  409. if length is None:
  410. length = len(data)
  411. cnt = range(length)
  412. return b''.join(six.int2byte(six.indexbytes(data, i) ^ mask[(offset + i) % 4]) for i in cnt)
  413. def _handle_control_frame(self, opcode, data):
  414. if opcode == 8: # connection close
  415. if not data:
  416. status = 1000
  417. elif len(data) > 1:
  418. status = struct.unpack_from('!H', data)[0]
  419. if not status or status not in VALID_CLOSE_STATUS:
  420. raise FailedConnectionError(
  421. 1002,
  422. "Unexpected close status code.")
  423. try:
  424. data = self.UTF8Decoder().decode(data[2:], True)
  425. except (UnicodeDecodeError, ValueError):
  426. raise FailedConnectionError(
  427. 1002,
  428. "Close message data should be valid UTF-8.")
  429. else:
  430. status = 1002
  431. self.close(close_data=(status, ''))
  432. raise ConnectionClosedError()
  433. elif opcode == 9: # ping
  434. self.send(data, control_code=0xA)
  435. elif opcode == 0xA: # pong
  436. pass
  437. else:
  438. raise FailedConnectionError(
  439. 1002, "Unknown control frame received.")
  440. def _iter_frames(self):
  441. fragmented_message = None
  442. try:
  443. while True:
  444. message = self._recv_frame(message=fragmented_message)
  445. if message.opcode & 8:
  446. self._handle_control_frame(
  447. message.opcode, message.getvalue())
  448. continue
  449. if fragmented_message and message is not fragmented_message:
  450. raise RuntimeError('Unexpected message change.')
  451. fragmented_message = message
  452. if message.finished:
  453. data = fragmented_message.getvalue()
  454. fragmented_message = None
  455. yield data
  456. except FailedConnectionError:
  457. exc_typ, exc_val, exc_tb = sys.exc_info()
  458. self.close(close_data=(exc_val.status, exc_val.message))
  459. except ConnectionClosedError:
  460. return
  461. except Exception:
  462. self.close(close_data=(1011, 'Internal Server Error'))
  463. raise
  464. def _recv_frame(self, message=None):
  465. recv = self._get_bytes
  466. header = recv(2)
  467. a, b = struct.unpack('!BB', header)
  468. finished = a >> 7 == 1
  469. rsv123 = a >> 4 & 7
  470. if rsv123:
  471. # must be zero
  472. raise FailedConnectionError(
  473. 1002,
  474. "RSV1, RSV2, RSV3: MUST be 0 unless an extension is"
  475. " negotiated that defines meanings for non-zero values.")
  476. opcode = a & 15
  477. if opcode not in (0, 1, 2, 8, 9, 0xA):
  478. raise FailedConnectionError(1002, "Unknown opcode received.")
  479. masked = b & 128 == 128
  480. if not masked and not self.client:
  481. raise FailedConnectionError(1002, "A client MUST mask all frames"
  482. " that it sends to the server")
  483. length = b & 127
  484. if opcode & 8:
  485. if not finished:
  486. raise FailedConnectionError(1002, "Control frames must not"
  487. " be fragmented.")
  488. if length > 125:
  489. raise FailedConnectionError(
  490. 1002,
  491. "All control frames MUST have a payload length of 125"
  492. " bytes or less")
  493. elif opcode and message:
  494. raise FailedConnectionError(
  495. 1002,
  496. "Received a non-continuation opcode within"
  497. " fragmented message.")
  498. elif not opcode and not message:
  499. raise FailedConnectionError(
  500. 1002,
  501. "Received continuation opcode with no previous"
  502. " fragments received.")
  503. if length == 126:
  504. length = struct.unpack('!H', recv(2))[0]
  505. elif length == 127:
  506. length = struct.unpack('!Q', recv(8))[0]
  507. if masked:
  508. mask = struct.unpack('!BBBB', recv(4))
  509. received = 0
  510. if not message or opcode & 8:
  511. decoder = self.UTF8Decoder() if opcode == 1 else None
  512. message = self.Message(opcode, decoder=decoder)
  513. if not length:
  514. message.push(b'', final=finished)
  515. else:
  516. while received < length:
  517. d = self.socket.recv(length - received)
  518. if not d:
  519. raise ConnectionClosedError()
  520. dlen = len(d)
  521. if masked:
  522. d = self._apply_mask(d, mask, length=dlen, offset=received)
  523. received = received + dlen
  524. try:
  525. message.push(d, final=finished)
  526. except (UnicodeDecodeError, ValueError):
  527. raise FailedConnectionError(
  528. 1007, "Text data must be valid utf-8")
  529. return message
  530. @staticmethod
  531. def _pack_message(message, masked=False,
  532. continuation=False, final=True, control_code=None):
  533. is_text = False
  534. if isinstance(message, six.text_type):
  535. message = message.encode('utf-8')
  536. is_text = True
  537. length = len(message)
  538. if not length:
  539. # no point masking empty data
  540. masked = False
  541. if control_code:
  542. if control_code not in (8, 9, 0xA):
  543. raise ProtocolError('Unknown control opcode.')
  544. if continuation or not final:
  545. raise ProtocolError('Control frame cannot be a fragment.')
  546. if length > 125:
  547. raise ProtocolError('Control frame data too large (>125).')
  548. header = struct.pack('!B', control_code | 1 << 7)
  549. else:
  550. opcode = 0 if continuation else (1 if is_text else 2)
  551. header = struct.pack('!B', opcode | (1 << 7 if final else 0))
  552. lengthdata = 1 << 7 if masked else 0
  553. if length > 65535:
  554. lengthdata = struct.pack('!BQ', lengthdata | 127, length)
  555. elif length > 125:
  556. lengthdata = struct.pack('!BH', lengthdata | 126, length)
  557. else:
  558. lengthdata = struct.pack('!B', lengthdata | length)
  559. if masked:
  560. # NOTE: RFC6455 states:
  561. # A server MUST NOT mask any frames that it sends to the client
  562. rand = Random(time.time())
  563. mask = [rand.getrandbits(8) for _ in six.moves.xrange(4)]
  564. message = RFC6455WebSocket._apply_mask(message, mask, length)
  565. maskdata = struct.pack('!BBBB', *mask)
  566. else:
  567. maskdata = b''
  568. return b''.join((header, lengthdata, maskdata, message))
  569. def wait(self):
  570. for i in self.iterator:
  571. return i
  572. def _send(self, frame):
  573. self._sendlock.acquire()
  574. try:
  575. self.socket.sendall(frame)
  576. finally:
  577. self._sendlock.release()
  578. def send(self, message, **kw):
  579. kw['masked'] = self.client
  580. payload = self._pack_message(message, **kw)
  581. self._send(payload)
  582. def _send_closing_frame(self, ignore_send_errors=False, close_data=None):
  583. if self.version in (8, 13) and not self.websocket_closed:
  584. if close_data is not None:
  585. status, msg = close_data
  586. if isinstance(msg, six.text_type):
  587. msg = msg.encode('utf-8')
  588. data = struct.pack('!H', status) + msg
  589. else:
  590. data = ''
  591. try:
  592. self.send(data, control_code=8)
  593. except SocketError:
  594. # Sometimes, like when the remote side cuts off the connection,
  595. # we don't care about this.
  596. if not ignore_send_errors: # pragma NO COVER
  597. raise
  598. self.websocket_closed = True
  599. def close(self, close_data=None):
  600. """Forcibly close the websocket; generally it is preferable to
  601. return from the handler method."""
  602. self._send_closing_frame(close_data=close_data)
  603. self.socket.shutdown(socket.SHUT_WR)
  604. self.socket.close()