Coverage for /pythoncovmergedfiles/medio/medio/src/aiohttp/aiohttp/http_parser.py: 90%

1import abc 

2import asyncio 

3import re 

4import string 

5from contextlib import suppress 

6from enum import IntEnum 

7from typing import ( 

8 Final, 

9 Generic, 

10 List, 

11 NamedTuple, 

12 Optional, 

13 Pattern, 

14 Set, 

15 Tuple, 

16 Type, 

17 TypeVar, 

18 Union, 

19) 

20 

21from multidict import CIMultiDict, CIMultiDictProxy, istr 

22from yarl import URL 

23 

24from . import hdrs 

25from .base_protocol import BaseProtocol 

26from .compression_utils import HAS_BROTLI, BrotliDecompressor, ZLibDecompressor 

27from .helpers import NO_EXTENSIONS, BaseTimerContext 

28from .http_exceptions import ( 

29 BadHttpMessage, 

30 BadStatusLine, 

31 ContentEncodingError, 

32 ContentLengthError, 

33 InvalidHeader, 

34 LineTooLong, 

35 TransferEncodingError, 

36) 

37from .http_writer import HttpVersion, HttpVersion10 

38from .log import internal_logger 

39from .streams import EMPTY_PAYLOAD, StreamReader 

40from .typedefs import RawHeaders 

41 

42__all__ = ( 

43 "HeadersParser", 

44 "HttpParser", 

45 "HttpRequestParser", 

46 "HttpResponseParser", 

47 "RawRequestMessage", 

48 "RawResponseMessage", 

49) 

50 

51ASCIISET: Final[Set[str]] = set(string.printable) 

52 

53# See https://tools.ietf.org/html/rfc7230#section-3.1.1 

54# and https://tools.ietf.org/html/rfc7230#appendix-B 

55# 

56# method = token 

57# tchar = "!" / "#" / "$" / "%" / "&" / "'" / "*" / "+" / "-" / "." / 

58# "^" / "_" / "`" / "|" / "~" / DIGIT / ALPHA 

59# token = 1*tchar 

60METHRE: Final[Pattern[str]] = re.compile(r"[!#$%&'*+\-.^_`|~0-9A-Za-z]+") 

61VERSRE: Final[Pattern[str]] = re.compile(r"HTTP/(\d+).(\d+)") 

62HDRRE: Final[Pattern[bytes]] = re.compile(rb"[\x00-\x1F\x7F()<>@,;:\[\]={} \t\\\\\"]") 

63 

64 

65class RawRequestMessage(NamedTuple): 

66 method: str 

67 path: str 

68 version: HttpVersion 

69 headers: CIMultiDictProxy[str] 

70 raw_headers: RawHeaders 

71 should_close: bool 

72 compression: Optional[str] 

73 upgrade: bool 

74 chunked: bool 

75 url: URL 

76 

77 

78class RawResponseMessage(NamedTuple): 

79 version: HttpVersion 

80 code: int 

81 reason: str 

82 headers: CIMultiDictProxy[str] 

83 raw_headers: RawHeaders 

84 should_close: bool 

85 compression: Optional[str] 

86 upgrade: bool 

87 chunked: bool 

88 

89 

90_MsgT = TypeVar("_MsgT", RawRequestMessage, RawResponseMessage) 

91 

92 

93class ParseState(IntEnum): 

94 PARSE_NONE = 0 

95 PARSE_LENGTH = 1 

96 PARSE_CHUNKED = 2 

97 PARSE_UNTIL_EOF = 3 

98 

99 

100class ChunkState(IntEnum): 

101 PARSE_CHUNKED_SIZE = 0 

102 PARSE_CHUNKED_CHUNK = 1 

103 PARSE_CHUNKED_CHUNK_EOF = 2 

104 PARSE_MAYBE_TRAILERS = 3 

105 PARSE_TRAILERS = 4 

106 

107 

108class HeadersParser: 

109 def __init__( 

110 self, 

111 max_line_size: int = 8190, 

112 max_field_size: int = 8190, 

113 ) -> None: 

114 self.max_line_size = max_line_size 

115 self.max_field_size = max_field_size 

116 

117 def parse_headers( 

118 self, lines: List[bytes] 

119 ) -> Tuple["CIMultiDictProxy[str]", RawHeaders]: 

120 headers: CIMultiDict[str] = CIMultiDict() 

121 raw_headers = [] 

122 

123 lines_idx = 1 

124 line = lines[1] 

125 line_count = len(lines) 

126 

127 while line: 

128 # Parse initial header name : value pair. 

129 try: 

130 bname, bvalue = line.split(b":", 1) 

131 except ValueError: 

132 raise InvalidHeader(line) from None 

133 

134 bname = bname.strip(b" \t") 

135 bvalue = bvalue.lstrip() 

136 if HDRRE.search(bname): 

137 raise InvalidHeader(bname) 

138 if len(bname) > self.max_field_size: 

139 raise LineTooLong( 

140 "request header name {}".format( 

141 bname.decode("utf8", "backslashreplace") 

142 ), 

143 str(self.max_field_size), 

144 str(len(bname)), 

145 ) 

146 

147 header_length = len(bvalue) 

148 

149 # next line 

150 lines_idx += 1 

151 line = lines[lines_idx] 

152 

153 # consume continuation lines 

154 continuation = line and line[0] in (32, 9) # (' ', '\t') 

155 

156 if continuation: 

157 bvalue_lst = [bvalue] 

158 while continuation: 

159 header_length += len(line) 

160 if header_length > self.max_field_size: 

161 raise LineTooLong( 

162 "request header field {}".format( 

163 bname.decode("utf8", "backslashreplace") 

164 ), 

165 str(self.max_field_size), 

166 str(header_length), 

167 ) 

168 bvalue_lst.append(line) 

169 

170 # next line 

171 lines_idx += 1 

172 if lines_idx < line_count: 

173 line = lines[lines_idx] 

174 if line: 

175 continuation = line[0] in (32, 9) # (' ', '\t') 

176 else: 

177 line = b"" 

178 break 

179 bvalue = b"".join(bvalue_lst) 

180 else: 

181 if header_length > self.max_field_size: 

182 raise LineTooLong( 

183 "request header field {}".format( 

184 bname.decode("utf8", "backslashreplace") 

185 ), 

186 str(self.max_field_size), 

187 str(header_length), 

188 ) 

189 

190 bvalue = bvalue.strip() 

191 name = bname.decode("utf-8", "surrogateescape") 

192 value = bvalue.decode("utf-8", "surrogateescape") 

193 

194 headers.add(name, value) 

195 raw_headers.append((bname, bvalue)) 

196 

197 return (CIMultiDictProxy(headers), tuple(raw_headers)) 

198 

199 

200class HttpParser(abc.ABC, Generic[_MsgT]): 

201 def __init__( 

202 self, 

203 protocol: BaseProtocol, 

204 loop: asyncio.AbstractEventLoop, 

205 limit: int, 

206 max_line_size: int = 8190, 

207 max_field_size: int = 8190, 

208 timer: Optional[BaseTimerContext] = None, 

209 code: Optional[int] = None, 

210 method: Optional[str] = None, 

211 readall: bool = False, 

212 payload_exception: Optional[Type[BaseException]] = None, 

213 response_with_body: bool = True, 

214 read_until_eof: bool = False, 

215 auto_decompress: bool = True, 

216 ) -> None: 

217 self.protocol = protocol 

218 self.loop = loop 

219 self.max_line_size = max_line_size 

220 self.max_field_size = max_field_size 

221 self.timer = timer 

222 self.code = code 

223 self.method = method 

224 self.readall = readall 

225 self.payload_exception = payload_exception 

226 self.response_with_body = response_with_body 

227 self.read_until_eof = read_until_eof 

228 

229 self._lines: List[bytes] = [] 

230 self._tail = b"" 

231 self._upgraded = False 

232 self._payload = None 

233 self._payload_parser: Optional[HttpPayloadParser] = None 

234 self._auto_decompress = auto_decompress 

235 self._limit = limit 

236 self._headers_parser = HeadersParser(max_line_size, max_field_size) 

237 

238 @abc.abstractmethod 

239 def parse_message(self, lines: List[bytes]) -> _MsgT: 

240 pass 

241 

242 def feed_eof(self) -> Optional[_MsgT]: 

243 if self._payload_parser is not None: 

244 self._payload_parser.feed_eof() 

245 self._payload_parser = None 

246 else: 

247 # try to extract partial message 

248 if self._tail: 

249 self._lines.append(self._tail) 

250 

251 if self._lines: 

252 if self._lines[-1] != "\r\n": 

253 self._lines.append(b"") 

254 with suppress(Exception): 

255 return self.parse_message(self._lines) 

256 return None 

257 

258 def feed_data( 

259 self, 

260 data: bytes, 

261 SEP: bytes = b"\r\n", 

262 EMPTY: bytes = b"", 

263 CONTENT_LENGTH: istr = hdrs.CONTENT_LENGTH, 

264 METH_CONNECT: str = hdrs.METH_CONNECT, 

265 SEC_WEBSOCKET_KEY1: istr = hdrs.SEC_WEBSOCKET_KEY1, 

266 ) -> Tuple[List[Tuple[_MsgT, StreamReader]], bool, bytes]: 

267 messages = [] 

268 

269 if self._tail: 

270 data, self._tail = self._tail + data, b"" 

271 

272 data_len = len(data) 

273 start_pos = 0 

274 loop = self.loop 

275 

276 while start_pos < data_len: 

277 # read HTTP message (request/response line + headers), \r\n\r\n 

278 # and split by lines 

279 if self._payload_parser is None and not self._upgraded: 

280 pos = data.find(SEP, start_pos) 

281 # consume \r\n 

282 if pos == start_pos and not self._lines: 

283 start_pos = pos + 2 

284 continue 

285 

286 if pos >= start_pos: 

287 # line found 

288 self._lines.append(data[start_pos:pos]) 

289 start_pos = pos + 2 

290 

291 # \r\n\r\n found 

292 if self._lines[-1] == EMPTY: 

293 try: 

294 msg: _MsgT = self.parse_message(self._lines) 

295 finally: 

296 self._lines.clear() 

297 

298 def get_content_length() -> Optional[int]: 

299 # payload length 

300 length_hdr = msg.headers.get(CONTENT_LENGTH) 

301 if length_hdr is None: 

302 return None 

303 

304 try: 

305 length = int(length_hdr) 

306 except ValueError: 

307 raise InvalidHeader(CONTENT_LENGTH) 

308 

309 if length < 0: 

310 raise InvalidHeader(CONTENT_LENGTH) 

311 

312 return length 

313 

314 length = get_content_length() 

315 # do not support old websocket spec 

316 if SEC_WEBSOCKET_KEY1 in msg.headers: 

317 raise InvalidHeader(SEC_WEBSOCKET_KEY1) 

318 

319 self._upgraded = msg.upgrade 

320 

321 method = getattr(msg, "method", self.method) 

322 

323 assert self.protocol is not None 

324 # calculate payload 

325 if ( 

326 (length is not None and length > 0) 

327 or msg.chunked 

328 and not msg.upgrade 

329 ): 

330 payload = StreamReader( 

331 self.protocol, 

332 timer=self.timer, 

333 loop=loop, 

334 limit=self._limit, 

335 ) 

336 payload_parser = HttpPayloadParser( 

337 payload, 

338 length=length, 

339 chunked=msg.chunked, 

340 method=method, 

341 compression=msg.compression, 

342 code=self.code, 

343 readall=self.readall, 

344 response_with_body=self.response_with_body, 

345 auto_decompress=self._auto_decompress, 

346 ) 

347 if not payload_parser.done: 

348 self._payload_parser = payload_parser 

349 elif method == METH_CONNECT: 

350 assert isinstance(msg, RawRequestMessage) 

351 payload = StreamReader( 

352 self.protocol, 

353 timer=self.timer, 

354 loop=loop, 

355 limit=self._limit, 

356 ) 

357 self._upgraded = True 

358 self._payload_parser = HttpPayloadParser( 

359 payload, 

360 method=msg.method, 

361 compression=msg.compression, 

362 readall=True, 

363 auto_decompress=self._auto_decompress, 

364 ) 

365 else: 

366 if ( 

367 getattr(msg, "code", 100) >= 199 

368 and length is None 

369 and self.read_until_eof 

370 ): 

371 payload = StreamReader( 

372 self.protocol, 

373 timer=self.timer, 

374 loop=loop, 

375 limit=self._limit, 

376 ) 

377 payload_parser = HttpPayloadParser( 

378 payload, 

379 length=length, 

380 chunked=msg.chunked, 

381 method=method, 

382 compression=msg.compression, 

383 code=self.code, 

384 readall=True, 

385 response_with_body=self.response_with_body, 

386 auto_decompress=self._auto_decompress, 

387 ) 

388 if not payload_parser.done: 

389 self._payload_parser = payload_parser 

390 else: 

391 payload = EMPTY_PAYLOAD 

392 

393 messages.append((msg, payload)) 

394 else: 

395 self._tail = data[start_pos:] 

396 data = EMPTY 

397 break 

398 

399 # no parser, just store 

400 elif self._payload_parser is None and self._upgraded: 

401 assert not self._lines 

402 break 

403 

404 # feed payload 

405 elif data and start_pos < data_len: 

406 assert not self._lines 

407 assert self._payload_parser is not None 

408 try: 

409 eof, data = self._payload_parser.feed_data(data[start_pos:]) 

410 except BaseException as exc: 

411 if self.payload_exception is not None: 

412 self._payload_parser.payload.set_exception( 

413 self.payload_exception(str(exc)) 

414 ) 

415 else: 

416 self._payload_parser.payload.set_exception(exc) 

417 

418 eof = True 

419 data = b"" 

420 

421 if eof: 

422 start_pos = 0 

423 data_len = len(data) 

424 self._payload_parser = None 

425 continue 

426 else: 

427 break 

428 

429 if data and start_pos < data_len: 

430 data = data[start_pos:] 

431 else: 

432 data = EMPTY 

433 

434 return messages, self._upgraded, data 

435 

436 def parse_headers( 

437 self, lines: List[bytes] 

438 ) -> Tuple[ 

439 "CIMultiDictProxy[str]", RawHeaders, Optional[bool], Optional[str], bool, bool 

440 ]: 

441 """Parses RFC 5322 headers from a stream. 

442 

443 Line continuations are supported. Returns list of header name 

444 and value pairs. Header name is in upper case. 

445 """ 

446 headers, raw_headers = self._headers_parser.parse_headers(lines) 

447 close_conn = None 

448 encoding = None 

449 upgrade = False 

450 chunked = False 

451 

452 # keep-alive 

453 conn = headers.get(hdrs.CONNECTION) 

454 if conn: 

455 v = conn.lower() 

456 if v == "close": 

457 close_conn = True 

458 elif v == "keep-alive": 

459 close_conn = False 

460 elif v == "upgrade": 

461 upgrade = True 

462 

463 # encoding 

464 enc = headers.get(hdrs.CONTENT_ENCODING) 

465 if enc: 

466 enc = enc.lower() 

467 if enc in ("gzip", "deflate", "br"): 

468 encoding = enc 

469 

470 # chunking 

471 te = headers.get(hdrs.TRANSFER_ENCODING) 

472 if te is not None: 

473 if "chunked" == te.lower(): 

474 chunked = True 

475 else: 

476 raise BadHttpMessage("Request has invalid `Transfer-Encoding`") 

477 

478 if hdrs.CONTENT_LENGTH in headers: 

479 raise BadHttpMessage( 

480 "Content-Length can't be present with Transfer-Encoding", 

481 ) 

482 

483 return (headers, raw_headers, close_conn, encoding, upgrade, chunked) 

484 

485 def set_upgraded(self, val: bool) -> None: 

486 """Set connection upgraded (to websocket) mode. 

487 

488 :param bool val: new state. 

489 """ 

490 self._upgraded = val 

491 

492 

493class HttpRequestParser(HttpParser[RawRequestMessage]): 

494 """Read request status line. 

495 

496 Exception .http_exceptions.BadStatusLine 

497 could be raised in case of any errors in status line. 

498 Returns RawRequestMessage. 

499 """ 

500 

501 def parse_message(self, lines: List[bytes]) -> RawRequestMessage: 

502 # request line 

503 line = lines[0].decode("utf-8", "surrogateescape") 

504 try: 

505 method, path, version = line.split(None, 2) 

506 except ValueError: 

507 raise BadStatusLine(line) from None 

508 

509 if len(path) > self.max_line_size: 

510 raise LineTooLong( 

511 "Status line is too long", str(self.max_line_size), str(len(path)) 

512 ) 

513 

514 # method 

515 if not METHRE.match(method): 

516 raise BadStatusLine(method) 

517 

518 # version 

519 try: 

520 if version.startswith("HTTP/"): 

521 n1, n2 = version[5:].split(".", 1) 

522 version_o = HttpVersion(int(n1), int(n2)) 

523 else: 

524 raise BadStatusLine(version) 

525 except Exception: 

526 raise BadStatusLine(version) 

527 

528 if method == "CONNECT": 

529 # authority-form, 

530 # https://datatracker.ietf.org/doc/html/rfc7230#section-5.3.3 

531 url = URL.build(authority=path, encoded=True) 

532 elif path.startswith("/"): 

533 # origin-form, 

534 # https://datatracker.ietf.org/doc/html/rfc7230#section-5.3.1 

535 path_part, _hash_separator, url_fragment = path.partition("#") 

536 path_part, _question_mark_separator, qs_part = path_part.partition("?") 

537 

538 # NOTE: `yarl.URL.build()` is used to mimic what the Cython-based 

539 # NOTE: parser does, otherwise it results into the same 

540 # NOTE: HTTP Request-Line input producing different 

541 # NOTE: `yarl.URL()` objects 

542 url = URL.build( 

543 path=path_part, 

544 query_string=qs_part, 

545 fragment=url_fragment, 

546 encoded=True, 

547 ) 

548 else: 

549 # absolute-form for proxy maybe, 

550 # https://datatracker.ietf.org/doc/html/rfc7230#section-5.3.2 

551 url = URL(path, encoded=True) 

552 

553 # read headers 

554 ( 

555 headers, 

556 raw_headers, 

557 close, 

558 compression, 

559 upgrade, 

560 chunked, 

561 ) = self.parse_headers(lines) 

562 

563 if close is None: # then the headers weren't set in the request 

564 if version_o <= HttpVersion10: # HTTP 1.0 must asks to not close 

565 close = True 

566 else: # HTTP 1.1 must ask to close. 

567 close = False 

568 

569 return RawRequestMessage( 

570 method, 

571 path, 

572 version_o, 

573 headers, 

574 raw_headers, 

575 close, 

576 compression, 

577 upgrade, 

578 chunked, 

579 url, 

580 ) 

581 

582 

583class HttpResponseParser(HttpParser[RawResponseMessage]): 

584 """Read response status line and headers. 

585 

586 BadStatusLine could be raised in case of any errors in status line. 

587 Returns RawResponseMessage. 

588 """ 

589 

590 def parse_message(self, lines: List[bytes]) -> RawResponseMessage: 

591 line = lines[0].decode("utf-8", "surrogateescape") 

592 try: 

593 version, status = line.split(None, 1) 

594 except ValueError: 

595 raise BadStatusLine(line) from None 

596 

597 try: 

598 status, reason = status.split(None, 1) 

599 except ValueError: 

600 reason = "" 

601 

602 if len(reason) > self.max_line_size: 

603 raise LineTooLong( 

604 "Status line is too long", str(self.max_line_size), str(len(reason)) 

605 ) 

606 

607 # version 

608 match = VERSRE.match(version) 

609 if match is None: 

610 raise BadStatusLine(line) 

611 version_o = HttpVersion(int(match.group(1)), int(match.group(2))) 

612 

613 # The status code is a three-digit number 

614 try: 

615 status_i = int(status) 

616 except ValueError: 

617 raise BadStatusLine(line) from None 

618 

619 if status_i > 999: 

620 raise BadStatusLine(line) 

621 

622 # read headers 

623 ( 

624 headers, 

625 raw_headers, 

626 close, 

627 compression, 

628 upgrade, 

629 chunked, 

630 ) = self.parse_headers(lines) 

631 

632 if close is None: 

633 close = version_o <= HttpVersion10 

634 

635 return RawResponseMessage( 

636 version_o, 

637 status_i, 

638 reason.strip(), 

639 headers, 

640 raw_headers, 

641 close, 

642 compression, 

643 upgrade, 

644 chunked, 

645 ) 

646 

647 

648class HttpPayloadParser: 

649 def __init__( 

650 self, 

651 payload: StreamReader, 

652 length: Optional[int] = None, 

653 chunked: bool = False, 

654 compression: Optional[str] = None, 

655 code: Optional[int] = None, 

656 method: Optional[str] = None, 

657 readall: bool = False, 

658 response_with_body: bool = True, 

659 auto_decompress: bool = True, 

660 ) -> None: 

661 self._length = 0 

662 self._type = ParseState.PARSE_NONE 

663 self._chunk = ChunkState.PARSE_CHUNKED_SIZE 

664 self._chunk_size = 0 

665 self._chunk_tail = b"" 

666 self._auto_decompress = auto_decompress 

667 self.done = False 

668 

669 # payload decompression wrapper 

670 if response_with_body and compression and self._auto_decompress: 

671 real_payload: Union[StreamReader, DeflateBuffer] = DeflateBuffer( 

672 payload, compression 

673 ) 

674 else: 

675 real_payload = payload 

676 

677 # payload parser 

678 if not response_with_body: 

679 # don't parse payload if it's not expected to be received 

680 self._type = ParseState.PARSE_NONE 

681 real_payload.feed_eof() 

682 self.done = True 

683 

684 elif chunked: 

685 self._type = ParseState.PARSE_CHUNKED 

686 elif length is not None: 

687 self._type = ParseState.PARSE_LENGTH 

688 self._length = length 

689 if self._length == 0: 

690 real_payload.feed_eof() 

691 self.done = True 

692 else: 

693 if readall and code != 204: 

694 self._type = ParseState.PARSE_UNTIL_EOF 

695 elif method in ("PUT", "POST"): 

696 internal_logger.warning( # pragma: no cover 

697 "Content-Length or Transfer-Encoding header is required" 

698 ) 

699 self._type = ParseState.PARSE_NONE 

700 real_payload.feed_eof() 

701 self.done = True 

702 

703 self.payload = real_payload 

704 

705 def feed_eof(self) -> None: 

706 if self._type == ParseState.PARSE_UNTIL_EOF: 

707 self.payload.feed_eof() 

708 elif self._type == ParseState.PARSE_LENGTH: 

709 raise ContentLengthError( 

710 "Not enough data for satisfy content length header." 

711 ) 

712 elif self._type == ParseState.PARSE_CHUNKED: 

713 raise TransferEncodingError( 

714 "Not enough data for satisfy transfer length header." 

715 ) 

716 

717 def feed_data( 

718 self, chunk: bytes, SEP: bytes = b"\r\n", CHUNK_EXT: bytes = b";" 

719 ) -> Tuple[bool, bytes]: 

720 # Read specified amount of bytes 

721 if self._type == ParseState.PARSE_LENGTH: 

722 required = self._length 

723 chunk_len = len(chunk) 

724 

725 if required >= chunk_len: 

726 self._length = required - chunk_len 

727 self.payload.feed_data(chunk, chunk_len) 

728 if self._length == 0: 

729 self.payload.feed_eof() 

730 return True, b"" 

731 else: 

732 self._length = 0 

733 self.payload.feed_data(chunk[:required], required) 

734 self.payload.feed_eof() 

735 return True, chunk[required:] 

736 

737 # Chunked transfer encoding parser 

738 elif self._type == ParseState.PARSE_CHUNKED: 

739 if self._chunk_tail: 

740 chunk = self._chunk_tail + chunk 

741 self._chunk_tail = b"" 

742 

743 while chunk: 

744 # read next chunk size 

745 if self._chunk == ChunkState.PARSE_CHUNKED_SIZE: 

746 pos = chunk.find(SEP) 

747 if pos >= 0: 

748 i = chunk.find(CHUNK_EXT, 0, pos) 

749 if i >= 0: 

750 size_b = chunk[:i] # strip chunk-extensions 

751 else: 

752 size_b = chunk[:pos] 

753 

754 try: 

755 size = int(bytes(size_b), 16) 

756 except ValueError: 

757 exc = TransferEncodingError( 

758 chunk[:pos].decode("ascii", "surrogateescape") 

759 ) 

760 self.payload.set_exception(exc) 

761 raise exc from None 

762 

763 chunk = chunk[pos + 2 :] 

764 if size == 0: # eof marker 

765 self._chunk = ChunkState.PARSE_MAYBE_TRAILERS 

766 else: 

767 self._chunk = ChunkState.PARSE_CHUNKED_CHUNK 

768 self._chunk_size = size 

769 self.payload.begin_http_chunk_receiving() 

770 else: 

771 self._chunk_tail = chunk 

772 return False, b"" 

773 

774 # read chunk and feed buffer 

775 if self._chunk == ChunkState.PARSE_CHUNKED_CHUNK: 

776 required = self._chunk_size 

777 chunk_len = len(chunk) 

778 

779 if required > chunk_len: 

780 self._chunk_size = required - chunk_len 

781 self.payload.feed_data(chunk, chunk_len) 

782 return False, b"" 

783 else: 

784 self._chunk_size = 0 

785 self.payload.feed_data(chunk[:required], required) 

786 chunk = chunk[required:] 

787 self._chunk = ChunkState.PARSE_CHUNKED_CHUNK_EOF 

788 self.payload.end_http_chunk_receiving() 

789 

790 # toss the CRLF at the end of the chunk 

791 if self._chunk == ChunkState.PARSE_CHUNKED_CHUNK_EOF: 

792 if chunk[:2] == SEP: 

793 chunk = chunk[2:] 

794 self._chunk = ChunkState.PARSE_CHUNKED_SIZE 

795 else: 

796 self._chunk_tail = chunk 

797 return False, b"" 

798 

799 # if stream does not contain trailer, after 0\r\n 

800 # we should get another \r\n otherwise 

801 # trailers needs to be skiped until \r\n\r\n 

802 if self._chunk == ChunkState.PARSE_MAYBE_TRAILERS: 

803 head = chunk[:2] 

804 if head == SEP: 

805 # end of stream 

806 self.payload.feed_eof() 

807 return True, chunk[2:] 

808 # Both CR and LF, or only LF may not be received yet. It is 

809 # expected that CRLF or LF will be shown at the very first 

810 # byte next time, otherwise trailers should come. The last 

811 # CRLF which marks the end of response might not be 

812 # contained in the same TCP segment which delivered the 

813 # size indicator. 

814 if not head: 

815 return False, b"" 

816 if head == SEP[:1]: 

817 self._chunk_tail = head 

818 return False, b"" 

819 self._chunk = ChunkState.PARSE_TRAILERS 

820 

821 # read and discard trailer up to the CRLF terminator 

822 if self._chunk == ChunkState.PARSE_TRAILERS: 

823 pos = chunk.find(SEP) 

824 if pos >= 0: 

825 chunk = chunk[pos + 2 :] 

826 self._chunk = ChunkState.PARSE_MAYBE_TRAILERS 

827 else: 

828 self._chunk_tail = chunk 

829 return False, b"" 

830 

831 # Read all bytes until eof 

832 elif self._type == ParseState.PARSE_UNTIL_EOF: 

833 self.payload.feed_data(chunk, len(chunk)) 

834 

835 return False, b"" 

836 

837 

838class DeflateBuffer: 

839 """DeflateStream decompress stream and feed data into specified stream.""" 

840 

841 def __init__(self, out: StreamReader, encoding: Optional[str]) -> None: 

842 self.out = out 

843 self.size = 0 

844 self.encoding = encoding 

845 self._started_decoding = False 

846 

847 self.decompressor: Union[BrotliDecompressor, ZLibDecompressor] 

848 if encoding == "br": 

849 if not HAS_BROTLI: # pragma: no cover 

850 raise ContentEncodingError( 

851 "Can not decode content-encoding: brotli (br). " 

852 "Please install `Brotli`" 

853 ) 

854 self.decompressor = BrotliDecompressor() 

855 else: 

856 self.decompressor = ZLibDecompressor(encoding=encoding) 

857 

858 def set_exception(self, exc: BaseException) -> None: 

859 self.out.set_exception(exc) 

860