http1connection.py 32 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751
  1. #
  2. # Copyright 2014 Facebook
  3. #
  4. # Licensed under the Apache License, Version 2.0 (the "License"); you may
  5. # not use this file except in compliance with the License. You may obtain
  6. # a copy of the License at
  7. #
  8. # http://www.apache.org/licenses/LICENSE-2.0
  9. #
  10. # Unless required by applicable law or agreed to in writing, software
  11. # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  12. # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13. # License for the specific language governing permissions and limitations
  14. # under the License.
  15. """Client and server implementations of HTTP/1.x.
  16. .. versionadded:: 4.0
  17. """
  18. from __future__ import absolute_import, division, print_function
  19. import re
  20. import warnings
  21. from tornado.concurrent import (Future, future_add_done_callback,
  22. future_set_result_unless_cancelled)
  23. from tornado.escape import native_str, utf8
  24. from tornado import gen
  25. from tornado import httputil
  26. from tornado import iostream
  27. from tornado.log import gen_log, app_log
  28. from tornado import stack_context
  29. from tornado.util import GzipDecompressor, PY3
  30. class _QuietException(Exception):
  31. def __init__(self):
  32. pass
  33. class _ExceptionLoggingContext(object):
  34. """Used with the ``with`` statement when calling delegate methods to
  35. log any exceptions with the given logger. Any exceptions caught are
  36. converted to _QuietException
  37. """
  38. def __init__(self, logger):
  39. self.logger = logger
  40. def __enter__(self):
  41. pass
  42. def __exit__(self, typ, value, tb):
  43. if value is not None:
  44. self.logger.error("Uncaught exception", exc_info=(typ, value, tb))
  45. raise _QuietException
  46. class HTTP1ConnectionParameters(object):
  47. """Parameters for `.HTTP1Connection` and `.HTTP1ServerConnection`.
  48. """
  49. def __init__(self, no_keep_alive=False, chunk_size=None,
  50. max_header_size=None, header_timeout=None, max_body_size=None,
  51. body_timeout=None, decompress=False):
  52. """
  53. :arg bool no_keep_alive: If true, always close the connection after
  54. one request.
  55. :arg int chunk_size: how much data to read into memory at once
  56. :arg int max_header_size: maximum amount of data for HTTP headers
  57. :arg float header_timeout: how long to wait for all headers (seconds)
  58. :arg int max_body_size: maximum amount of data for body
  59. :arg float body_timeout: how long to wait while reading body (seconds)
  60. :arg bool decompress: if true, decode incoming
  61. ``Content-Encoding: gzip``
  62. """
  63. self.no_keep_alive = no_keep_alive
  64. self.chunk_size = chunk_size or 65536
  65. self.max_header_size = max_header_size or 65536
  66. self.header_timeout = header_timeout
  67. self.max_body_size = max_body_size
  68. self.body_timeout = body_timeout
  69. self.decompress = decompress
  70. class HTTP1Connection(httputil.HTTPConnection):
  71. """Implements the HTTP/1.x protocol.
  72. This class can be on its own for clients, or via `HTTP1ServerConnection`
  73. for servers.
  74. """
  75. def __init__(self, stream, is_client, params=None, context=None):
  76. """
  77. :arg stream: an `.IOStream`
  78. :arg bool is_client: client or server
  79. :arg params: a `.HTTP1ConnectionParameters` instance or ``None``
  80. :arg context: an opaque application-defined object that can be accessed
  81. as ``connection.context``.
  82. """
  83. self.is_client = is_client
  84. self.stream = stream
  85. if params is None:
  86. params = HTTP1ConnectionParameters()
  87. self.params = params
  88. self.context = context
  89. self.no_keep_alive = params.no_keep_alive
  90. # The body limits can be altered by the delegate, so save them
  91. # here instead of just referencing self.params later.
  92. self._max_body_size = (self.params.max_body_size or
  93. self.stream.max_buffer_size)
  94. self._body_timeout = self.params.body_timeout
  95. # _write_finished is set to True when finish() has been called,
  96. # i.e. there will be no more data sent. Data may still be in the
  97. # stream's write buffer.
  98. self._write_finished = False
  99. # True when we have read the entire incoming body.
  100. self._read_finished = False
  101. # _finish_future resolves when all data has been written and flushed
  102. # to the IOStream.
  103. self._finish_future = Future()
  104. # If true, the connection should be closed after this request
  105. # (after the response has been written in the server side,
  106. # and after it has been read in the client)
  107. self._disconnect_on_finish = False
  108. self._clear_callbacks()
  109. # Save the start lines after we read or write them; they
  110. # affect later processing (e.g. 304 responses and HEAD methods
  111. # have content-length but no bodies)
  112. self._request_start_line = None
  113. self._response_start_line = None
  114. self._request_headers = None
  115. # True if we are writing output with chunked encoding.
  116. self._chunking_output = None
  117. # While reading a body with a content-length, this is the
  118. # amount left to read.
  119. self._expected_content_remaining = None
  120. # A Future for our outgoing writes, returned by IOStream.write.
  121. self._pending_write = None
  122. def read_response(self, delegate):
  123. """Read a single HTTP response.
  124. Typical client-mode usage is to write a request using `write_headers`,
  125. `write`, and `finish`, and then call ``read_response``.
  126. :arg delegate: a `.HTTPMessageDelegate`
  127. Returns a `.Future` that resolves to None after the full response has
  128. been read.
  129. """
  130. if self.params.decompress:
  131. delegate = _GzipMessageDelegate(delegate, self.params.chunk_size)
  132. return self._read_message(delegate)
  133. @gen.coroutine
  134. def _read_message(self, delegate):
  135. need_delegate_close = False
  136. try:
  137. header_future = self.stream.read_until_regex(
  138. b"\r?\n\r?\n",
  139. max_bytes=self.params.max_header_size)
  140. if self.params.header_timeout is None:
  141. header_data = yield header_future
  142. else:
  143. try:
  144. header_data = yield gen.with_timeout(
  145. self.stream.io_loop.time() + self.params.header_timeout,
  146. header_future,
  147. quiet_exceptions=iostream.StreamClosedError)
  148. except gen.TimeoutError:
  149. self.close()
  150. raise gen.Return(False)
  151. start_line, headers = self._parse_headers(header_data)
  152. if self.is_client:
  153. start_line = httputil.parse_response_start_line(start_line)
  154. self._response_start_line = start_line
  155. else:
  156. start_line = httputil.parse_request_start_line(start_line)
  157. self._request_start_line = start_line
  158. self._request_headers = headers
  159. self._disconnect_on_finish = not self._can_keep_alive(
  160. start_line, headers)
  161. need_delegate_close = True
  162. with _ExceptionLoggingContext(app_log):
  163. header_future = delegate.headers_received(start_line, headers)
  164. if header_future is not None:
  165. yield header_future
  166. if self.stream is None:
  167. # We've been detached.
  168. need_delegate_close = False
  169. raise gen.Return(False)
  170. skip_body = False
  171. if self.is_client:
  172. if (self._request_start_line is not None and
  173. self._request_start_line.method == 'HEAD'):
  174. skip_body = True
  175. code = start_line.code
  176. if code == 304:
  177. # 304 responses may include the content-length header
  178. # but do not actually have a body.
  179. # http://tools.ietf.org/html/rfc7230#section-3.3
  180. skip_body = True
  181. if code >= 100 and code < 200:
  182. # 1xx responses should never indicate the presence of
  183. # a body.
  184. if ('Content-Length' in headers or
  185. 'Transfer-Encoding' in headers):
  186. raise httputil.HTTPInputError(
  187. "Response code %d cannot have body" % code)
  188. # TODO: client delegates will get headers_received twice
  189. # in the case of a 100-continue. Document or change?
  190. yield self._read_message(delegate)
  191. else:
  192. if (headers.get("Expect") == "100-continue" and
  193. not self._write_finished):
  194. self.stream.write(b"HTTP/1.1 100 (Continue)\r\n\r\n")
  195. if not skip_body:
  196. body_future = self._read_body(
  197. start_line.code if self.is_client else 0, headers, delegate)
  198. if body_future is not None:
  199. if self._body_timeout is None:
  200. yield body_future
  201. else:
  202. try:
  203. yield gen.with_timeout(
  204. self.stream.io_loop.time() + self._body_timeout,
  205. body_future,
  206. quiet_exceptions=iostream.StreamClosedError)
  207. except gen.TimeoutError:
  208. gen_log.info("Timeout reading body from %s",
  209. self.context)
  210. self.stream.close()
  211. raise gen.Return(False)
  212. self._read_finished = True
  213. if not self._write_finished or self.is_client:
  214. need_delegate_close = False
  215. with _ExceptionLoggingContext(app_log):
  216. delegate.finish()
  217. # If we're waiting for the application to produce an asynchronous
  218. # response, and we're not detached, register a close callback
  219. # on the stream (we didn't need one while we were reading)
  220. if (not self._finish_future.done() and
  221. self.stream is not None and
  222. not self.stream.closed()):
  223. self.stream.set_close_callback(self._on_connection_close)
  224. yield self._finish_future
  225. if self.is_client and self._disconnect_on_finish:
  226. self.close()
  227. if self.stream is None:
  228. raise gen.Return(False)
  229. except httputil.HTTPInputError as e:
  230. gen_log.info("Malformed HTTP message from %s: %s",
  231. self.context, e)
  232. if not self.is_client:
  233. yield self.stream.write(b'HTTP/1.1 400 Bad Request\r\n\r\n')
  234. self.close()
  235. raise gen.Return(False)
  236. finally:
  237. if need_delegate_close:
  238. with _ExceptionLoggingContext(app_log):
  239. delegate.on_connection_close()
  240. header_future = None
  241. self._clear_callbacks()
  242. raise gen.Return(True)
  243. def _clear_callbacks(self):
  244. """Clears the callback attributes.
  245. This allows the request handler to be garbage collected more
  246. quickly in CPython by breaking up reference cycles.
  247. """
  248. self._write_callback = None
  249. self._write_future = None
  250. self._close_callback = None
  251. if self.stream is not None:
  252. self.stream.set_close_callback(None)
  253. def set_close_callback(self, callback):
  254. """Sets a callback that will be run when the connection is closed.
  255. Note that this callback is slightly different from
  256. `.HTTPMessageDelegate.on_connection_close`: The
  257. `.HTTPMessageDelegate` method is called when the connection is
  258. closed while recieving a message. This callback is used when
  259. there is not an active delegate (for example, on the server
  260. side this callback is used if the client closes the connection
  261. after sending its request but before receiving all the
  262. response.
  263. """
  264. self._close_callback = stack_context.wrap(callback)
  265. def _on_connection_close(self):
  266. # Note that this callback is only registered on the IOStream
  267. # when we have finished reading the request and are waiting for
  268. # the application to produce its response.
  269. if self._close_callback is not None:
  270. callback = self._close_callback
  271. self._close_callback = None
  272. callback()
  273. if not self._finish_future.done():
  274. future_set_result_unless_cancelled(self._finish_future, None)
  275. self._clear_callbacks()
  276. def close(self):
  277. if self.stream is not None:
  278. self.stream.close()
  279. self._clear_callbacks()
  280. if not self._finish_future.done():
  281. future_set_result_unless_cancelled(self._finish_future, None)
  282. def detach(self):
  283. """Take control of the underlying stream.
  284. Returns the underlying `.IOStream` object and stops all further
  285. HTTP processing. May only be called during
  286. `.HTTPMessageDelegate.headers_received`. Intended for implementing
  287. protocols like websockets that tunnel over an HTTP handshake.
  288. """
  289. self._clear_callbacks()
  290. stream = self.stream
  291. self.stream = None
  292. if not self._finish_future.done():
  293. future_set_result_unless_cancelled(self._finish_future, None)
  294. return stream
  295. def set_body_timeout(self, timeout):
  296. """Sets the body timeout for a single request.
  297. Overrides the value from `.HTTP1ConnectionParameters`.
  298. """
  299. self._body_timeout = timeout
  300. def set_max_body_size(self, max_body_size):
  301. """Sets the body size limit for a single request.
  302. Overrides the value from `.HTTP1ConnectionParameters`.
  303. """
  304. self._max_body_size = max_body_size
  305. def write_headers(self, start_line, headers, chunk=None, callback=None):
  306. """Implements `.HTTPConnection.write_headers`."""
  307. lines = []
  308. if self.is_client:
  309. self._request_start_line = start_line
  310. lines.append(utf8('%s %s HTTP/1.1' % (start_line[0], start_line[1])))
  311. # Client requests with a non-empty body must have either a
  312. # Content-Length or a Transfer-Encoding.
  313. self._chunking_output = (
  314. start_line.method in ('POST', 'PUT', 'PATCH') and
  315. 'Content-Length' not in headers and
  316. 'Transfer-Encoding' not in headers)
  317. else:
  318. self._response_start_line = start_line
  319. lines.append(utf8('HTTP/1.1 %d %s' % (start_line[1], start_line[2])))
  320. self._chunking_output = (
  321. # TODO: should this use
  322. # self._request_start_line.version or
  323. # start_line.version?
  324. self._request_start_line.version == 'HTTP/1.1' and
  325. # 1xx, 204 and 304 responses have no body (not even a zero-length
  326. # body), and so should not have either Content-Length or
  327. # Transfer-Encoding headers.
  328. start_line.code not in (204, 304) and
  329. (start_line.code < 100 or start_line.code >= 200) and
  330. # No need to chunk the output if a Content-Length is specified.
  331. 'Content-Length' not in headers and
  332. # Applications are discouraged from touching Transfer-Encoding,
  333. # but if they do, leave it alone.
  334. 'Transfer-Encoding' not in headers)
  335. # If connection to a 1.1 client will be closed, inform client
  336. if (self._request_start_line.version == 'HTTP/1.1' and self._disconnect_on_finish):
  337. headers['Connection'] = 'close'
  338. # If a 1.0 client asked for keep-alive, add the header.
  339. if (self._request_start_line.version == 'HTTP/1.0' and
  340. self._request_headers.get('Connection', '').lower() == 'keep-alive'):
  341. headers['Connection'] = 'Keep-Alive'
  342. if self._chunking_output:
  343. headers['Transfer-Encoding'] = 'chunked'
  344. if (not self.is_client and
  345. (self._request_start_line.method == 'HEAD' or
  346. start_line.code == 304)):
  347. self._expected_content_remaining = 0
  348. elif 'Content-Length' in headers:
  349. self._expected_content_remaining = int(headers['Content-Length'])
  350. else:
  351. self._expected_content_remaining = None
  352. # TODO: headers are supposed to be of type str, but we still have some
  353. # cases that let bytes slip through. Remove these native_str calls when those
  354. # are fixed.
  355. header_lines = (native_str(n) + ": " + native_str(v) for n, v in headers.get_all())
  356. if PY3:
  357. lines.extend(l.encode('latin1') for l in header_lines)
  358. else:
  359. lines.extend(header_lines)
  360. for line in lines:
  361. if b'\n' in line:
  362. raise ValueError('Newline in header: ' + repr(line))
  363. future = None
  364. if self.stream.closed():
  365. future = self._write_future = Future()
  366. future.set_exception(iostream.StreamClosedError())
  367. future.exception()
  368. else:
  369. if callback is not None:
  370. warnings.warn("callback argument is deprecated, use returned Future instead",
  371. DeprecationWarning)
  372. self._write_callback = stack_context.wrap(callback)
  373. else:
  374. future = self._write_future = Future()
  375. data = b"\r\n".join(lines) + b"\r\n\r\n"
  376. if chunk:
  377. data += self._format_chunk(chunk)
  378. self._pending_write = self.stream.write(data)
  379. future_add_done_callback(self._pending_write, self._on_write_complete)
  380. return future
  381. def _format_chunk(self, chunk):
  382. if self._expected_content_remaining is not None:
  383. self._expected_content_remaining -= len(chunk)
  384. if self._expected_content_remaining < 0:
  385. # Close the stream now to stop further framing errors.
  386. self.stream.close()
  387. raise httputil.HTTPOutputError(
  388. "Tried to write more data than Content-Length")
  389. if self._chunking_output and chunk:
  390. # Don't write out empty chunks because that means END-OF-STREAM
  391. # with chunked encoding
  392. return utf8("%x" % len(chunk)) + b"\r\n" + chunk + b"\r\n"
  393. else:
  394. return chunk
  395. def write(self, chunk, callback=None):
  396. """Implements `.HTTPConnection.write`.
  397. For backwards compatibility it is allowed but deprecated to
  398. skip `write_headers` and instead call `write()` with a
  399. pre-encoded header block.
  400. """
  401. future = None
  402. if self.stream.closed():
  403. future = self._write_future = Future()
  404. self._write_future.set_exception(iostream.StreamClosedError())
  405. self._write_future.exception()
  406. else:
  407. if callback is not None:
  408. warnings.warn("callback argument is deprecated, use returned Future instead",
  409. DeprecationWarning)
  410. self._write_callback = stack_context.wrap(callback)
  411. else:
  412. future = self._write_future = Future()
  413. self._pending_write = self.stream.write(self._format_chunk(chunk))
  414. self._pending_write.add_done_callback(self._on_write_complete)
  415. return future
  416. def finish(self):
  417. """Implements `.HTTPConnection.finish`."""
  418. if (self._expected_content_remaining is not None and
  419. self._expected_content_remaining != 0 and
  420. not self.stream.closed()):
  421. self.stream.close()
  422. raise httputil.HTTPOutputError(
  423. "Tried to write %d bytes less than Content-Length" %
  424. self._expected_content_remaining)
  425. if self._chunking_output:
  426. if not self.stream.closed():
  427. self._pending_write = self.stream.write(b"0\r\n\r\n")
  428. self._pending_write.add_done_callback(self._on_write_complete)
  429. self._write_finished = True
  430. # If the app finished the request while we're still reading,
  431. # divert any remaining data away from the delegate and
  432. # close the connection when we're done sending our response.
  433. # Closing the connection is the only way to avoid reading the
  434. # whole input body.
  435. if not self._read_finished:
  436. self._disconnect_on_finish = True
  437. # No more data is coming, so instruct TCP to send any remaining
  438. # data immediately instead of waiting for a full packet or ack.
  439. self.stream.set_nodelay(True)
  440. if self._pending_write is None:
  441. self._finish_request(None)
  442. else:
  443. future_add_done_callback(self._pending_write, self._finish_request)
  444. def _on_write_complete(self, future):
  445. exc = future.exception()
  446. if exc is not None and not isinstance(exc, iostream.StreamClosedError):
  447. future.result()
  448. if self._write_callback is not None:
  449. callback = self._write_callback
  450. self._write_callback = None
  451. self.stream.io_loop.add_callback(callback)
  452. if self._write_future is not None:
  453. future = self._write_future
  454. self._write_future = None
  455. future_set_result_unless_cancelled(future, None)
  456. def _can_keep_alive(self, start_line, headers):
  457. if self.params.no_keep_alive:
  458. return False
  459. connection_header = headers.get("Connection")
  460. if connection_header is not None:
  461. connection_header = connection_header.lower()
  462. if start_line.version == "HTTP/1.1":
  463. return connection_header != "close"
  464. elif ("Content-Length" in headers or
  465. headers.get("Transfer-Encoding", "").lower() == "chunked" or
  466. getattr(start_line, 'method', None) in ("HEAD", "GET")):
  467. # start_line may be a request or response start line; only
  468. # the former has a method attribute.
  469. return connection_header == "keep-alive"
  470. return False
  471. def _finish_request(self, future):
  472. self._clear_callbacks()
  473. if not self.is_client and self._disconnect_on_finish:
  474. self.close()
  475. return
  476. # Turn Nagle's algorithm back on, leaving the stream in its
  477. # default state for the next request.
  478. self.stream.set_nodelay(False)
  479. if not self._finish_future.done():
  480. future_set_result_unless_cancelled(self._finish_future, None)
  481. def _parse_headers(self, data):
  482. # The lstrip removes newlines that some implementations sometimes
  483. # insert between messages of a reused connection. Per RFC 7230,
  484. # we SHOULD ignore at least one empty line before the request.
  485. # http://tools.ietf.org/html/rfc7230#section-3.5
  486. data = native_str(data.decode('latin1')).lstrip("\r\n")
  487. # RFC 7230 section allows for both CRLF and bare LF.
  488. eol = data.find("\n")
  489. start_line = data[:eol].rstrip("\r")
  490. headers = httputil.HTTPHeaders.parse(data[eol:])
  491. return start_line, headers
  492. def _read_body(self, code, headers, delegate):
  493. if "Content-Length" in headers:
  494. if "Transfer-Encoding" in headers:
  495. # Response cannot contain both Content-Length and
  496. # Transfer-Encoding headers.
  497. # http://tools.ietf.org/html/rfc7230#section-3.3.3
  498. raise httputil.HTTPInputError(
  499. "Response with both Transfer-Encoding and Content-Length")
  500. if "," in headers["Content-Length"]:
  501. # Proxies sometimes cause Content-Length headers to get
  502. # duplicated. If all the values are identical then we can
  503. # use them but if they differ it's an error.
  504. pieces = re.split(r',\s*', headers["Content-Length"])
  505. if any(i != pieces[0] for i in pieces):
  506. raise httputil.HTTPInputError(
  507. "Multiple unequal Content-Lengths: %r" %
  508. headers["Content-Length"])
  509. headers["Content-Length"] = pieces[0]
  510. try:
  511. content_length = int(headers["Content-Length"])
  512. except ValueError:
  513. # Handles non-integer Content-Length value.
  514. raise httputil.HTTPInputError(
  515. "Only integer Content-Length is allowed: %s" % headers["Content-Length"])
  516. if content_length > self._max_body_size:
  517. raise httputil.HTTPInputError("Content-Length too long")
  518. else:
  519. content_length = None
  520. if code == 204:
  521. # This response code is not allowed to have a non-empty body,
  522. # and has an implicit length of zero instead of read-until-close.
  523. # http://www.w3.org/Protocols/rfc2616/rfc2616-sec4.html#sec4.3
  524. if ("Transfer-Encoding" in headers or
  525. content_length not in (None, 0)):
  526. raise httputil.HTTPInputError(
  527. "Response with code %d should not have body" % code)
  528. content_length = 0
  529. if content_length is not None:
  530. return self._read_fixed_body(content_length, delegate)
  531. if headers.get("Transfer-Encoding", "").lower() == "chunked":
  532. return self._read_chunked_body(delegate)
  533. if self.is_client:
  534. return self._read_body_until_close(delegate)
  535. return None
  536. @gen.coroutine
  537. def _read_fixed_body(self, content_length, delegate):
  538. while content_length > 0:
  539. body = yield self.stream.read_bytes(
  540. min(self.params.chunk_size, content_length), partial=True)
  541. content_length -= len(body)
  542. if not self._write_finished or self.is_client:
  543. with _ExceptionLoggingContext(app_log):
  544. ret = delegate.data_received(body)
  545. if ret is not None:
  546. yield ret
  547. @gen.coroutine
  548. def _read_chunked_body(self, delegate):
  549. # TODO: "chunk extensions" http://tools.ietf.org/html/rfc2616#section-3.6.1
  550. total_size = 0
  551. while True:
  552. chunk_len = yield self.stream.read_until(b"\r\n", max_bytes=64)
  553. chunk_len = int(chunk_len.strip(), 16)
  554. if chunk_len == 0:
  555. crlf = yield self.stream.read_bytes(2)
  556. if crlf != b'\r\n':
  557. raise httputil.HTTPInputError("improperly terminated chunked request")
  558. return
  559. total_size += chunk_len
  560. if total_size > self._max_body_size:
  561. raise httputil.HTTPInputError("chunked body too large")
  562. bytes_to_read = chunk_len
  563. while bytes_to_read:
  564. chunk = yield self.stream.read_bytes(
  565. min(bytes_to_read, self.params.chunk_size), partial=True)
  566. bytes_to_read -= len(chunk)
  567. if not self._write_finished or self.is_client:
  568. with _ExceptionLoggingContext(app_log):
  569. ret = delegate.data_received(chunk)
  570. if ret is not None:
  571. yield ret
  572. # chunk ends with \r\n
  573. crlf = yield self.stream.read_bytes(2)
  574. assert crlf == b"\r\n"
  575. @gen.coroutine
  576. def _read_body_until_close(self, delegate):
  577. body = yield self.stream.read_until_close()
  578. if not self._write_finished or self.is_client:
  579. with _ExceptionLoggingContext(app_log):
  580. delegate.data_received(body)
  581. class _GzipMessageDelegate(httputil.HTTPMessageDelegate):
  582. """Wraps an `HTTPMessageDelegate` to decode ``Content-Encoding: gzip``.
  583. """
  584. def __init__(self, delegate, chunk_size):
  585. self._delegate = delegate
  586. self._chunk_size = chunk_size
  587. self._decompressor = None
  588. def headers_received(self, start_line, headers):
  589. if headers.get("Content-Encoding") == "gzip":
  590. self._decompressor = GzipDecompressor()
  591. # Downstream delegates will only see uncompressed data,
  592. # so rename the content-encoding header.
  593. # (but note that curl_httpclient doesn't do this).
  594. headers.add("X-Consumed-Content-Encoding",
  595. headers["Content-Encoding"])
  596. del headers["Content-Encoding"]
  597. return self._delegate.headers_received(start_line, headers)
  598. @gen.coroutine
  599. def data_received(self, chunk):
  600. if self._decompressor:
  601. compressed_data = chunk
  602. while compressed_data:
  603. decompressed = self._decompressor.decompress(
  604. compressed_data, self._chunk_size)
  605. if decompressed:
  606. ret = self._delegate.data_received(decompressed)
  607. if ret is not None:
  608. yield ret
  609. compressed_data = self._decompressor.unconsumed_tail
  610. else:
  611. ret = self._delegate.data_received(chunk)
  612. if ret is not None:
  613. yield ret
  614. def finish(self):
  615. if self._decompressor is not None:
  616. tail = self._decompressor.flush()
  617. if tail:
  618. # I believe the tail will always be empty (i.e.
  619. # decompress will return all it can). The purpose
  620. # of the flush call is to detect errors such
  621. # as truncated input. But in case it ever returns
  622. # anything, treat it as an extra chunk
  623. self._delegate.data_received(tail)
  624. return self._delegate.finish()
  625. def on_connection_close(self):
  626. return self._delegate.on_connection_close()
  627. class HTTP1ServerConnection(object):
  628. """An HTTP/1.x server."""
  629. def __init__(self, stream, params=None, context=None):
  630. """
  631. :arg stream: an `.IOStream`
  632. :arg params: a `.HTTP1ConnectionParameters` or None
  633. :arg context: an opaque application-defined object that is accessible
  634. as ``connection.context``
  635. """
  636. self.stream = stream
  637. if params is None:
  638. params = HTTP1ConnectionParameters()
  639. self.params = params
  640. self.context = context
  641. self._serving_future = None
  642. @gen.coroutine
  643. def close(self):
  644. """Closes the connection.
  645. Returns a `.Future` that resolves after the serving loop has exited.
  646. """
  647. self.stream.close()
  648. # Block until the serving loop is done, but ignore any exceptions
  649. # (start_serving is already responsible for logging them).
  650. try:
  651. yield self._serving_future
  652. except Exception:
  653. pass
  654. def start_serving(self, delegate):
  655. """Starts serving requests on this connection.
  656. :arg delegate: a `.HTTPServerConnectionDelegate`
  657. """
  658. assert isinstance(delegate, httputil.HTTPServerConnectionDelegate)
  659. self._serving_future = self._server_request_loop(delegate)
  660. # Register the future on the IOLoop so its errors get logged.
  661. self.stream.io_loop.add_future(self._serving_future,
  662. lambda f: f.result())
  663. @gen.coroutine
  664. def _server_request_loop(self, delegate):
  665. try:
  666. while True:
  667. conn = HTTP1Connection(self.stream, False,
  668. self.params, self.context)
  669. request_delegate = delegate.start_request(self, conn)
  670. try:
  671. ret = yield conn.read_response(request_delegate)
  672. except (iostream.StreamClosedError,
  673. iostream.UnsatisfiableReadError):
  674. return
  675. except _QuietException:
  676. # This exception was already logged.
  677. conn.close()
  678. return
  679. except Exception:
  680. gen_log.error("Uncaught exception", exc_info=True)
  681. conn.close()
  682. return
  683. if not ret:
  684. return
  685. yield gen.moment
  686. finally:
  687. delegate.on_close(self)