Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/tornado/http1connection.py: 13%

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 

16"""Client and server implementations of HTTP/1.x. 

17 

18.. versionadded:: 4.0 

19""" 

20 

21import asyncio 

22import logging 

23import re 

24import types 

25import zlib 

26from collections.abc import Awaitable, Callable 

27from typing import Optional, Type, cast 

28 

29from tornado import gen, httputil, iostream 

30from tornado.concurrent import ( 

31 Future, 

32 future_add_done_callback, 

33 future_set_result_unless_cancelled, 

34) 

35from tornado.escape import native_str, utf8 

36from tornado.log import app_log, gen_log 

37from tornado.util import GzipDecompressor 

38 

39CR_OR_LF_RE = re.compile(b"\r|\n") 

40 

41 

42# The maximum number of informational (1xx) responses to accept before the 

43# real response. Each one is processed with a recursive call to 

44# _read_message, so an unbounded number of them would exhaust the stack. 

45# There is no legitimate use for more than a handful. 

46_MAX_1XX_RESPONSES = 10 

47 

48# Reads that can be satisfied from data that has already arrived complete 

49# without returning control to the IOLoop, so a peer that sends a body 

50# faster than we can process it could otherwise monopolize the IOLoop for 

51# the duration of the body. While reading a body, we yield to the IOLoop 

52# after processing this many bytes. Each chunk of a chunked body also counts 

53# as _CHUNK_YIELD_COST bytes, since the per-chunk overhead dominates when 

54# the chunks are small. 

55_YIELD_THRESHOLD = 256 * 1024 

56_CHUNK_YIELD_COST = 1024 

57 

58 

59def _is_done(future: Awaitable) -> bool: 

60 # IOStream's read methods are declared to return Awaitables, but they 

61 # are actually Futures which may already be resolved. 

62 return isinstance(future, Future) and future.done() 

63 

64 

65class _QuietException(Exception): 

66 def __init__(self) -> None: 

67 pass 

68 

69 

70class _ExceptionLoggingContext: 

71 """Used with the ``with`` statement when calling delegate methods to 

72 log any exceptions with the given logger. Any exceptions caught are 

73 converted to _QuietException 

74 """ 

75 

76 def __init__(self, logger: logging.Logger) -> None: 

77 self.logger = logger 

78 

79 def __enter__(self) -> None: 

80 pass 

81 

82 def __exit__( 

83 self, 

84 typ: "Optional[Type[BaseException]]", 

85 value: BaseException | None, 

86 tb: types.TracebackType, 

87 ) -> None: 

88 if value is not None: 

89 assert typ is not None 

90 # Let HTTPInputError pass through to higher-level handler 

91 if isinstance(value, httputil.HTTPInputError): 

92 return None 

93 self.logger.error("Uncaught exception", exc_info=(typ, value, tb)) 

94 raise _QuietException 

95 

96 

97class HTTP1ConnectionParameters: 

98 """Parameters for `.HTTP1Connection` and `.HTTP1ServerConnection`.""" 

99 

100 def __init__( 

101 self, 

102 no_keep_alive: bool = False, 

103 chunk_size: int | None = None, 

104 max_header_size: int | None = None, 

105 header_timeout: float | None = None, 

106 max_body_size: int | None = None, 

107 body_timeout: float | None = None, 

108 decompress: bool = False, 

109 ) -> None: 

110 """ 

111 :arg bool no_keep_alive: If true, always close the connection after 

112 one request. 

113 :arg int chunk_size: how much data to read into memory at once 

114 :arg int max_header_size: maximum amount of data for HTTP headers 

115 :arg float header_timeout: how long to wait for all headers (seconds) 

116 :arg int max_body_size: maximum amount of data for body 

117 :arg float body_timeout: how long to wait while reading body (seconds) 

118 :arg bool decompress: if true, decode incoming 

119 ``Content-Encoding: gzip`` 

120 """ 

121 self.no_keep_alive = no_keep_alive 

122 self.chunk_size = chunk_size or 65536 

123 self.max_header_size = max_header_size or 65536 

124 self.header_timeout = header_timeout 

125 self.max_body_size = max_body_size 

126 self.body_timeout = body_timeout 

127 self.decompress = decompress 

128 

129 

130class HTTP1Connection(httputil.HTTPConnection): 

131 """Implements the HTTP/1.x protocol. 

132 

133 This class can be on its own for clients, or via `HTTP1ServerConnection` 

134 for servers. 

135 """ 

136 

137 def __init__( 

138 self, 

139 stream: iostream.IOStream, 

140 is_client: bool, 

141 params: HTTP1ConnectionParameters | None = None, 

142 context: object | None = None, 

143 ) -> None: 

144 """ 

145 :arg stream: an `.IOStream` 

146 :arg bool is_client: client or server 

147 :arg params: a `.HTTP1ConnectionParameters` instance or ``None`` 

148 :arg context: an opaque application-defined object that can be accessed 

149 as ``connection.context``. 

150 """ 

151 self.is_client = is_client 

152 self.stream = stream 

153 if params is None: 

154 params = HTTP1ConnectionParameters() 

155 self.params = params 

156 self.context = context 

157 self.no_keep_alive = params.no_keep_alive 

158 # The body limits can be altered by the delegate, so save them 

159 # here instead of just referencing self.params later. 

160 self._max_body_size = ( 

161 self.params.max_body_size 

162 if self.params.max_body_size is not None 

163 else self.stream.max_buffer_size 

164 ) 

165 self._body_timeout = self.params.body_timeout 

166 # _write_finished is set to True when finish() has been called, 

167 # i.e. there will be no more data sent. Data may still be in the 

168 # stream's write buffer. 

169 self._write_finished = False 

170 # True when we have read the entire incoming body. 

171 self._read_finished = False 

172 # _finish_future resolves when all data has been written and flushed 

173 # to the IOStream. 

174 self._finish_future: Future[None] = Future() 

175 # If true, the connection should be closed after this request 

176 # (after the response has been written in the server side, 

177 # and after it has been read in the client) 

178 self._disconnect_on_finish = False 

179 self._clear_callbacks() 

180 # Save the start lines after we read or write them; they 

181 # affect later processing (e.g. 304 responses and HEAD methods 

182 # have content-length but no bodies) 

183 self._request_start_line: httputil.RequestStartLine | None = None 

184 self._response_start_line: httputil.ResponseStartLine | None = None 

185 self._request_headers: httputil.HTTPHeaders | None = None 

186 # True if we are writing output with chunked encoding. 

187 self._chunking_output = False 

188 # While reading a body with a content-length, this is the 

189 # amount left to read. 

190 self._expected_content_remaining: int | None = None 

191 # A Future for our outgoing writes, returned by IOStream.write. 

192 self._pending_write: Future[None] | None = None 

193 

194 def read_response(self, delegate: httputil.HTTPMessageDelegate) -> Awaitable[bool]: 

195 """Read a single HTTP response. 

196 

197 Typical client-mode usage is to write a request using `write_headers`, 

198 `write`, and `finish`, and then call ``read_response``. 

199 

200 :arg delegate: a `.HTTPMessageDelegate` 

201 

202 Returns a `.Future` that resolves to a bool after the full response has 

203 been read. The result is true if the stream is still open. 

204 """ 

205 if self.params.decompress: 

206 delegate = _GzipMessageDelegate( 

207 delegate, self.params.chunk_size, self._max_body_size 

208 ) 

209 return self._read_message(delegate) 

210 

211 async def _read_message(self, delegate: httputil.HTTPMessageDelegate) -> bool: 

212 try: 

213 return await self._read_message_inner(delegate) 

214 except _QuietException: 

215 # The delegate raised an exception, which has already been logged, 

216 # and the delegate has been told that the connection is closed. 

217 # Close it, instead of letting _QuietException escape to our caller. 

218 self.close() 

219 return False 

220 

221 async def _read_message_inner( 

222 self, delegate: httputil.HTTPMessageDelegate, num_1xx: int = 0 

223 ) -> bool: 

224 need_delegate_close = False 

225 try: 

226 header_future = self.stream.read_until_regex( 

227 b"\r?\n\r?\n", max_bytes=self.params.max_header_size 

228 ) 

229 if self.params.header_timeout is None: 

230 header_data = await header_future 

231 else: 

232 try: 

233 header_data = await gen.with_timeout( 

234 self.stream.io_loop.time() + self.params.header_timeout, 

235 header_future, 

236 quiet_exceptions=iostream.StreamClosedError, 

237 ) 

238 except TimeoutError: 

239 self.close() 

240 return False 

241 start_line_str, headers = self._parse_headers(header_data) 

242 if self.is_client: 

243 resp_start_line = httputil.parse_response_start_line(start_line_str) 

244 self._response_start_line = resp_start_line 

245 start_line: httputil.RequestStartLine | httputil.ResponseStartLine = ( 

246 resp_start_line 

247 ) 

248 # TODO: this will need to change to support client-side keepalive 

249 self._disconnect_on_finish = False 

250 else: 

251 req_start_line = httputil.parse_request_start_line(start_line_str) 

252 self._request_start_line = req_start_line 

253 self._request_headers = headers 

254 start_line = req_start_line 

255 self._disconnect_on_finish = not self._can_keep_alive( 

256 req_start_line, headers 

257 ) 

258 need_delegate_close = True 

259 with _ExceptionLoggingContext(app_log): 

260 header_recv_future = delegate.headers_received(start_line, headers) 

261 if header_recv_future is not None: 

262 await header_recv_future 

263 if self.stream is None: 

264 # We've been detached. 

265 need_delegate_close = False 

266 return False 

267 skip_body = False 

268 if self.is_client: 

269 assert isinstance(start_line, httputil.ResponseStartLine) 

270 if ( 

271 self._request_start_line is not None 

272 and self._request_start_line.method == "HEAD" 

273 ): 

274 skip_body = True 

275 code = start_line.code 

276 if code == 304: 

277 # 304 responses may include the content-length header 

278 # but do not actually have a body. 

279 # http://tools.ietf.org/html/rfc7230#section-3.3 

280 skip_body = True 

281 if 100 <= code < 200: 

282 # 1xx responses should never indicate the presence of 

283 # a body. 

284 if "Content-Length" in headers or "Transfer-Encoding" in headers: 

285 raise httputil.HTTPInputError( 

286 "Response code %d cannot have body" % code 

287 ) 

288 if num_1xx >= _MAX_1XX_RESPONSES: 

289 raise httputil.HTTPInputError("Too many 1xx responses") 

290 # TODO: client delegates will get headers_received twice 

291 # in the case of a 100-continue. Document or change? 

292 # 

293 # The recursive call reads the real response and owns 

294 # the delegate from here on, so there is nothing left 

295 # for this frame to do. Clear need_delegate_close so 

296 # that the finally block does not call 

297 # on_connection_close() on an already-finished 

298 # delegate. 

299 need_delegate_close = False 

300 return await self._read_message_inner(delegate, num_1xx + 1) 

301 else: 

302 if headers.get("Expect") == "100-continue" and not self._write_finished: 

303 self.stream.write(b"HTTP/1.1 100 (Continue)\r\n\r\n") 

304 if not skip_body: 

305 body_future = self._read_body( 

306 resp_start_line.code if self.is_client else 0, headers, delegate 

307 ) 

308 if body_future is not None: 

309 if self._body_timeout is None: 

310 await body_future 

311 else: 

312 try: 

313 await gen.with_timeout( 

314 self.stream.io_loop.time() + self._body_timeout, 

315 body_future, 

316 quiet_exceptions=iostream.StreamClosedError, 

317 ) 

318 except TimeoutError: 

319 gen_log.info("Timeout reading body from %s", self.context) 

320 self.stream.close() 

321 return False 

322 self._read_finished = True 

323 if not self._write_finished or self.is_client: 

324 with _ExceptionLoggingContext(app_log): 

325 delegate.finish() 

326 # If finish() raised, the delegate is told that the connection 

327 # is closed, as when any of its other methods raise. 

328 need_delegate_close = False 

329 # If we're waiting for the application to produce an asynchronous 

330 # response, and we're not detached, register a close callback 

331 # on the stream (we didn't need one while we were reading) 

332 if ( 

333 not self._finish_future.done() 

334 and self.stream is not None 

335 and not self.stream.closed() 

336 ): 

337 self.stream.set_close_callback(self._on_connection_close) 

338 await self._finish_future 

339 if self.is_client and self._disconnect_on_finish: 

340 self.close() 

341 if self.stream is None: 

342 return False 

343 except httputil.HTTPInputError as e: 

344 gen_log.info("Malformed HTTP message from %s: %s", self.context, e) 

345 if not self.is_client: 

346 await self.stream.write(b"HTTP/1.1 400 Bad Request\r\n\r\n") 

347 elif not need_delegate_close: 

348 # The error came before headers_received, so the delegate 

349 # doesn't know about this message yet. Tell it the 

350 # connection is closed so a client request doesn't hang. 

351 need_delegate_close = True 

352 self.close() 

353 return False 

354 finally: 

355 if need_delegate_close: 

356 with _ExceptionLoggingContext(app_log): 

357 delegate.on_connection_close() 

358 header_future = None # type: ignore 

359 self._clear_callbacks() 

360 return True 

361 

362 def _clear_callbacks(self) -> None: 

363 """Clears the callback attributes. 

364 

365 This allows the request handler to be garbage collected more 

366 quickly in CPython by breaking up reference cycles. 

367 """ 

368 self._write_callback = None 

369 self._write_future: Future[None] | None = None 

370 self._close_callback: Callable[[], None] | None = None 

371 if self.stream is not None: 

372 self.stream.set_close_callback(None) 

373 

374 def set_close_callback(self, callback: Callable[[], None] | None) -> None: 

375 """Sets a callback that will be run when the connection is closed. 

376 

377 Note that this callback is slightly different from 

378 `.HTTPMessageDelegate.on_connection_close`: The 

379 `.HTTPMessageDelegate` method is called when the connection is 

380 closed while receiving a message. This callback is used when 

381 there is not an active delegate (for example, on the server 

382 side this callback is used if the client closes the connection 

383 after sending its request but before receiving all the 

384 response. 

385 """ 

386 self._close_callback = callback 

387 

388 def _on_connection_close(self) -> None: 

389 # Note that this callback is only registered on the IOStream 

390 # when we have finished reading the request and are waiting for 

391 # the application to produce its response. 

392 if self._close_callback is not None: 

393 callback = self._close_callback 

394 self._close_callback = None 

395 callback() 

396 if not self._finish_future.done(): 

397 future_set_result_unless_cancelled(self._finish_future, None) 

398 self._clear_callbacks() 

399 

400 def close(self) -> None: 

401 if self.stream is not None: 

402 self.stream.close() 

403 self._clear_callbacks() 

404 if not self._finish_future.done(): 

405 future_set_result_unless_cancelled(self._finish_future, None) 

406 

407 def detach(self) -> iostream.IOStream: 

408 """Take control of the underlying stream. 

409 

410 Returns the underlying `.IOStream` object and stops all further 

411 HTTP processing. May only be called during 

412 `.HTTPMessageDelegate.headers_received`. Intended for implementing 

413 protocols like websockets that tunnel over an HTTP handshake. 

414 """ 

415 self._clear_callbacks() 

416 stream = self.stream 

417 self.stream = None # type: ignore 

418 if not self._finish_future.done(): 

419 future_set_result_unless_cancelled(self._finish_future, None) 

420 return stream 

421 

422 def set_body_timeout(self, timeout: float) -> None: 

423 """Sets the body timeout for a single request. 

424 

425 Overrides the value from `.HTTP1ConnectionParameters`. 

426 """ 

427 self._body_timeout = timeout 

428 

429 def set_max_body_size(self, max_body_size: int) -> None: 

430 """Sets the body size limit for a single request. 

431 

432 Overrides the value from `.HTTP1ConnectionParameters`. 

433 """ 

434 self._max_body_size = max_body_size 

435 

436 def write_headers( 

437 self, 

438 start_line: httputil.RequestStartLine | httputil.ResponseStartLine, 

439 headers: httputil.HTTPHeaders, 

440 chunk: bytes | None = None, 

441 ) -> "Future[None]": 

442 """Implements `.HTTPConnection.write_headers`.""" 

443 lines = [] 

444 if self.is_client: 

445 assert isinstance(start_line, httputil.RequestStartLine) 

446 self._request_start_line = start_line 

447 lines.append(utf8(f"{start_line[0]} {start_line[1]} HTTP/1.1")) 

448 # Client requests with a non-empty body must have either a 

449 # Content-Length or a Transfer-Encoding. If Content-Length is not 

450 # present we'll add our Transfer-Encoding below. 

451 self._chunking_output = ( 

452 start_line.method in ("POST", "PUT", "PATCH") 

453 and "Content-Length" not in headers 

454 ) 

455 else: 

456 assert isinstance(start_line, httputil.ResponseStartLine) 

457 assert self._request_start_line is not None 

458 assert self._request_headers is not None 

459 self._response_start_line = start_line 

460 lines.append(utf8("HTTP/1.1 %d %s" % (start_line[1], start_line[2]))) 

461 self._chunking_output = ( 

462 # TODO: should this use 

463 # self._request_start_line.version or 

464 # start_line.version? 

465 self._request_start_line.version == "HTTP/1.1" 

466 # Omit payload header field for HEAD request. 

467 and self._request_start_line.method != "HEAD" 

468 # 1xx, 204 and 304 responses have no body (not even a zero-length 

469 # body), and so should not have either Content-Length or 

470 # Transfer-Encoding headers. 

471 and start_line.code not in (204, 304) 

472 and (start_line.code < 100 or start_line.code >= 200) 

473 # No need to chunk the output if a Content-Length is specified. 

474 and "Content-Length" not in headers 

475 ) 

476 # If connection to a 1.1 client will be closed, inform client 

477 if ( 

478 self._request_start_line.version == "HTTP/1.1" 

479 and self._disconnect_on_finish 

480 ): 

481 headers["Connection"] = "close" 

482 # If a 1.0 client asked for keep-alive, add the header. 

483 if ( 

484 self._request_start_line.version == "HTTP/1.0" 

485 and self._request_headers.get("Connection", "").lower() == "keep-alive" 

486 ): 

487 headers["Connection"] = "Keep-Alive" 

488 if self._chunking_output: 

489 headers["Transfer-Encoding"] = "chunked" 

490 if not self.is_client and ( 

491 self._request_start_line.method == "HEAD" 

492 or cast(httputil.ResponseStartLine, start_line).code == 304 

493 ): 

494 self._expected_content_remaining = 0 

495 elif "Content-Length" in headers: 

496 self._expected_content_remaining = parse_int(headers["Content-Length"]) 

497 else: 

498 self._expected_content_remaining = None 

499 # TODO: headers are supposed to be of type str, but we still have some 

500 # cases that let bytes slip through. Remove these native_str calls when those 

501 # are fixed. 

502 header_lines = ( 

503 native_str(n) + ": " + native_str(v) for n, v in headers.get_all() 

504 ) 

505 lines.extend(line.encode("latin1") for line in header_lines) 

506 for line in lines: 

507 if CR_OR_LF_RE.search(line): 

508 raise ValueError("Illegal characters (CR or LF) in header: %r" % line) 

509 future = None 

510 if self.stream.closed(): 

511 future = self._write_future = Future() 

512 future.set_exception(iostream.StreamClosedError()) 

513 future.exception() 

514 else: 

515 future = self._write_future = Future() 

516 data = b"\r\n".join(lines) + b"\r\n\r\n" 

517 if chunk: 

518 data += self._format_chunk(chunk) 

519 self._pending_write = self.stream.write(data) 

520 future_add_done_callback(self._pending_write, self._on_write_complete) 

521 return future 

522 

523 def _format_chunk(self, chunk: bytes) -> bytes: 

524 if self._expected_content_remaining is not None: 

525 self._expected_content_remaining -= len(chunk) 

526 if self._expected_content_remaining < 0: 

527 # Close the stream now to stop further framing errors. 

528 self.stream.close() 

529 raise httputil.HTTPOutputError( 

530 "Tried to write more data than Content-Length" 

531 ) 

532 if self._chunking_output and chunk: 

533 # Don't write out empty chunks because that means END-OF-STREAM 

534 # with chunked encoding 

535 return utf8("%x" % len(chunk)) + b"\r\n" + chunk + b"\r\n" 

536 else: 

537 return chunk 

538 

539 def write(self, chunk: bytes) -> "Future[None]": 

540 """Implements `.HTTPConnection.write`. 

541 

542 For backwards compatibility it is allowed but deprecated to 

543 skip `write_headers` and instead call `write()` with a 

544 pre-encoded header block. 

545 """ 

546 future = None 

547 if self.stream.closed(): 

548 future = self._write_future = Future() 

549 self._write_future.set_exception(iostream.StreamClosedError()) 

550 self._write_future.exception() 

551 else: 

552 future = self._write_future = Future() 

553 self._pending_write = self.stream.write(self._format_chunk(chunk)) 

554 future_add_done_callback(self._pending_write, self._on_write_complete) 

555 return future 

556 

557 def finish(self) -> None: 

558 """Implements `.HTTPConnection.finish`.""" 

559 if ( 

560 self._expected_content_remaining is not None 

561 and self._expected_content_remaining != 0 

562 and not self.stream.closed() 

563 ): 

564 self.stream.close() 

565 raise httputil.HTTPOutputError( 

566 "Tried to write %d bytes less than Content-Length" 

567 % self._expected_content_remaining 

568 ) 

569 if self._chunking_output: 

570 if not self.stream.closed(): 

571 self._pending_write = self.stream.write(b"0\r\n\r\n") 

572 self._pending_write.add_done_callback(self._on_write_complete) 

573 self._write_finished = True 

574 # If the app finished the request while we're still reading, 

575 # divert any remaining data away from the delegate and 

576 # close the connection when we're done sending our response. 

577 # Closing the connection is the only way to avoid reading the 

578 # whole input body. 

579 if not self._read_finished: 

580 self._disconnect_on_finish = True 

581 # No more data is coming, so instruct TCP to send any remaining 

582 # data immediately instead of waiting for a full packet or ack. 

583 self.stream.set_nodelay(True) 

584 if self._pending_write is None: 

585 self._finish_request(None) 

586 else: 

587 future_add_done_callback(self._pending_write, self._finish_request) 

588 

589 def _on_write_complete(self, future: "Future[None]") -> None: 

590 exc = future.exception() 

591 if exc is not None and not isinstance(exc, iostream.StreamClosedError): 

592 future.result() 

593 if self._write_callback is not None: 

594 callback = self._write_callback 

595 self._write_callback = None 

596 self.stream.io_loop.add_callback(callback) 

597 if self._write_future is not None: 

598 future = self._write_future 

599 self._write_future = None 

600 future_set_result_unless_cancelled(future, None) 

601 

602 def _can_keep_alive( 

603 self, start_line: httputil.RequestStartLine, headers: httputil.HTTPHeaders 

604 ) -> bool: 

605 if self.params.no_keep_alive: 

606 return False 

607 connection_header = headers.get("Connection") 

608 if connection_header is not None: 

609 connection_header = connection_header.lower() 

610 if start_line.version == "HTTP/1.1": 

611 return connection_header != "close" 

612 elif ( 

613 "Content-Length" in headers 

614 or is_transfer_encoding_chunked(headers) 

615 or getattr(start_line, "method", None) in ("HEAD", "GET") 

616 ): 

617 # start_line may be a request or response start line; only 

618 # the former has a method attribute. 

619 return connection_header == "keep-alive" 

620 return False 

621 

622 def _finish_request(self, future: "Optional[Future[None]]") -> None: 

623 self._clear_callbacks() 

624 if not self.is_client and self._disconnect_on_finish: 

625 self.close() 

626 return 

627 # Turn Nagle's algorithm back on, leaving the stream in its 

628 # default state for the next request. 

629 self.stream.set_nodelay(False) 

630 if not self._finish_future.done(): 

631 future_set_result_unless_cancelled(self._finish_future, None) 

632 

633 def _parse_headers(self, data: bytes) -> tuple[str, httputil.HTTPHeaders]: 

634 # The lstrip removes newlines that some implementations sometimes 

635 # insert between messages of a reused connection. Per RFC 7230, 

636 # we SHOULD ignore at least one empty line before the request. 

637 # http://tools.ietf.org/html/rfc7230#section-3.5 

638 data_str = native_str(data.decode("latin1")).lstrip("\r\n") 

639 # RFC 7230 section allows for both CRLF and bare LF. 

640 eol = data_str.find("\n") 

641 start_line = data_str[:eol].rstrip("\r") 

642 headers = httputil.HTTPHeaders.parse(data_str[eol:]) 

643 return start_line, headers 

644 

645 def _read_body( 

646 self, 

647 code: int, 

648 headers: httputil.HTTPHeaders, 

649 delegate: httputil.HTTPMessageDelegate, 

650 ) -> Awaitable[None] | None: 

651 if "Content-Length" in headers: 

652 if "," in headers["Content-Length"]: 

653 # Proxies sometimes cause Content-Length headers to get 

654 # duplicated. If all the values are identical then we can 

655 # use them but if they differ it's an error. 

656 pieces = re.split(r",[ \t]*", headers["Content-Length"]) 

657 if any(i != pieces[0] for i in pieces): 

658 raise httputil.HTTPInputError( 

659 "Multiple unequal Content-Lengths: %r" 

660 % headers["Content-Length"] 

661 ) 

662 headers["Content-Length"] = pieces[0] 

663 

664 try: 

665 content_length: int | None = parse_int(headers["Content-Length"]) 

666 except ValueError: 

667 # Handles non-integer Content-Length value. 

668 raise httputil.HTTPInputError( 

669 "Only integer Content-Length is allowed: %s" 

670 % headers["Content-Length"] 

671 ) 

672 

673 if cast(int, content_length) > self._max_body_size: 

674 raise httputil.HTTPInputError("Content-Length too long") 

675 else: 

676 content_length = None 

677 

678 is_chunked = is_transfer_encoding_chunked(headers) 

679 

680 if code == 204: 

681 # This response code is not allowed to have a non-empty body, 

682 # and has an implicit length of zero instead of read-until-close. 

683 # http://www.w3.org/Protocols/rfc2616/rfc2616-sec4.html#sec4.3 

684 if is_chunked or content_length not in (None, 0): 

685 raise httputil.HTTPInputError( 

686 "Response with code %d should not have body" % code 

687 ) 

688 content_length = 0 

689 

690 if is_chunked: 

691 return self._read_chunked_body(delegate) 

692 if content_length is not None: 

693 return self._read_fixed_body(content_length, delegate) 

694 if self.is_client: 

695 return self._read_body_until_close(delegate) 

696 return None 

697 

698 async def _deliver_body( 

699 self, delegate: httputil.HTTPMessageDelegate, data: bytes 

700 ) -> None: 

701 if not self._write_finished or self.is_client: 

702 with _ExceptionLoggingContext(app_log): 

703 ret = delegate.data_received(data) 

704 if ret is not None: 

705 await ret 

706 

707 async def _read_fixed_body( 

708 self, content_length: int, delegate: httputil.HTTPMessageDelegate 

709 ) -> None: 

710 work = 0 # See _YIELD_THRESHOLD 

711 while content_length > 0: 

712 body = await self.stream.read_bytes( 

713 min(self.params.chunk_size, content_length), partial=True 

714 ) 

715 content_length -= len(body) 

716 await self._deliver_body(delegate, body) 

717 work += len(body) 

718 if work >= _YIELD_THRESHOLD: 

719 work = 0 

720 await asyncio.sleep(0) 

721 

722 async def _read_chunked_body(self, delegate: httputil.HTTPMessageDelegate) -> None: 

723 # TODO: "chunk extensions" http://tools.ietf.org/html/rfc2616#section-3.6.1 

724 total_size = 0 

725 # Chunk boundaries are not meaningful (and may be changed by 

726 # intermediaries), so the contents of consecutive small chunks are 

727 # coalesced before being passed to the delegate. This bounds the 

728 # number of data_received calls (and the overhead of delegates that 

729 # buffer each piece they receive) regardless of how small the chunks 

730 # are. Pending data is delivered whenever we would have to wait for 

731 # the network, so this does not add latency for streaming delegates. 

732 pending = bytearray() 

733 work = 0 # See _YIELD_THRESHOLD 

734 

735 def read(future: Awaitable[bytes]) -> Awaitable[bytes]: 

736 if pending and not _is_done(future): 

737 return flush_and_read(future) 

738 return future 

739 

740 async def flush_and_read(future: Awaitable[bytes]) -> bytes: 

741 data = bytes(pending) 

742 pending.clear() 

743 await self._deliver_body(delegate, data) 

744 return await future 

745 

746 while True: 

747 work += _CHUNK_YIELD_COST 

748 if work >= _YIELD_THRESHOLD: 

749 work = 0 

750 await asyncio.sleep(0) 

751 chunk_len_str = await read(self.stream.read_until(b"\r\n", max_bytes=64)) 

752 try: 

753 chunk_len = parse_hex_int(native_str(chunk_len_str[:-2])) 

754 except ValueError: 

755 raise httputil.HTTPInputError("invalid chunk size") 

756 if chunk_len == 0: 

757 crlf = await read(self.stream.read_bytes(2)) 

758 if crlf != b"\r\n": 

759 raise httputil.HTTPInputError( 

760 "improperly terminated chunked request" 

761 ) 

762 if pending: 

763 await self._deliver_body(delegate, bytes(pending)) 

764 return 

765 total_size += chunk_len 

766 if total_size > self._max_body_size: 

767 raise httputil.HTTPInputError("chunked body too large") 

768 bytes_to_read = chunk_len 

769 while bytes_to_read: 

770 chunk = await read( 

771 self.stream.read_bytes( 

772 min(bytes_to_read, self.params.chunk_size), partial=True 

773 ) 

774 ) 

775 bytes_to_read -= len(chunk) 

776 if not pending and len(chunk) == self.params.chunk_size: 

777 # Full-sized reads are passed through without copying. 

778 await self._deliver_body(delegate, chunk) 

779 else: 

780 pending += chunk 

781 if len(pending) >= self.params.chunk_size: 

782 data = bytes(pending[: self.params.chunk_size]) 

783 del pending[: self.params.chunk_size] 

784 await self._deliver_body(delegate, data) 

785 work += len(chunk) 

786 if work >= _YIELD_THRESHOLD: 

787 work = 0 

788 await asyncio.sleep(0) 

789 # chunk ends with \r\n 

790 crlf = await read(self.stream.read_bytes(2)) 

791 if crlf != b"\r\n": 

792 raise httputil.HTTPInputError("improperly terminated chunk") 

793 

794 async def _read_body_until_close( 

795 self, delegate: httputil.HTTPMessageDelegate 

796 ) -> None: 

797 # The body is terminated by the connection closing, so there is no 

798 # length known in advance. Read incrementally so that max_body_size 

799 # is enforced before an over-large body has been buffered, and so 

800 # that the body is not limited by the stream's read buffer size. 

801 total_size = 0 

802 work = 0 # See _YIELD_THRESHOLD 

803 while True: 

804 try: 

805 body = await self.stream.read_bytes( 

806 self.params.chunk_size, partial=True 

807 ) 

808 except iostream.StreamClosedError: 

809 # The connection closing is the normal end of this body. 

810 return 

811 total_size += len(body) 

812 if total_size > self._max_body_size: 

813 raise httputil.HTTPInputError("Body too long") 

814 await self._deliver_body(delegate, body) 

815 work += len(body) 

816 if work >= _YIELD_THRESHOLD: 

817 work = 0 

818 await asyncio.sleep(0) 

819 

820 

821class _GzipMessageDelegate(httputil.HTTPMessageDelegate): 

822 """Wraps an `HTTPMessageDelegate` to decode ``Content-Encoding: gzip``.""" 

823 

824 def __init__( 

825 self, 

826 delegate: httputil.HTTPMessageDelegate, 

827 chunk_size: int, 

828 max_body_size: int, 

829 ) -> None: 

830 self._delegate = delegate 

831 self._chunk_size = chunk_size 

832 self._max_body_size = max_body_size 

833 self._decompressed_body_size = 0 

834 self._decompressor: GzipDecompressor | None = None 

835 

836 def headers_received( 

837 self, 

838 start_line: httputil.RequestStartLine | httputil.ResponseStartLine, 

839 headers: httputil.HTTPHeaders, 

840 ) -> Awaitable[None] | None: 

841 if headers.get("Content-Encoding", "").lower() == "gzip": 

842 self._decompressor = GzipDecompressor() 

843 # Downstream delegates will only see uncompressed data, 

844 # so rename the content-encoding header. 

845 # (but note that curl_httpclient doesn't do this). 

846 headers.add("X-Consumed-Content-Encoding", headers["Content-Encoding"]) 

847 del headers["Content-Encoding"] 

848 return self._delegate.headers_received(start_line, headers)