(self, s)
| 316 | return b"HTTP1" |
| 317 | |
| 318 | def post_dissect(self, s): |
| 319 | self._original_len = len(s) |
| 320 | encodings = self._get_encodings() |
| 321 | # Un-chunkify |
| 322 | if conf.contribs["http"]["auto_chunk"] and "chunked" in encodings: |
| 323 | data = b"" |
| 324 | while s: |
| 325 | length, _, body = s.partition(b"\r\n") |
| 326 | try: |
| 327 | length = int(length, 16) |
| 328 | except ValueError: |
| 329 | # Not a valid chunk. Ignore |
| 330 | break |
| 331 | else: |
| 332 | load = body[:length] |
| 333 | if body[length : length + 2] != b"\r\n": |
| 334 | # Invalid chunk. Ignore |
| 335 | break |
| 336 | s = body[length + 2 :] |
| 337 | data += load |
| 338 | if not s: |
| 339 | s = data |
| 340 | if not conf.contribs["http"]["auto_compression"]: |
| 341 | return s |
| 342 | # Decompress |
| 343 | try: |
| 344 | if "deflate" in encodings: |
| 345 | import zlib |
| 346 | |
| 347 | s = zlib.decompress(s) |
| 348 | elif "gzip" in encodings: |
| 349 | s = gzip.decompress(s) |
| 350 | elif "compress" in encodings: |
| 351 | if _is_lzw_available: |
| 352 | s = lzw.decompress(s) |
| 353 | else: |
| 354 | log_loading.info( |
| 355 | "Can't import lzw. compress decompression " "will be ignored !" |
| 356 | ) |
| 357 | elif "br" in encodings: |
| 358 | if _is_brotli_available: |
| 359 | s = brotli.decompress(s) |
| 360 | else: |
| 361 | log_loading.info( |
| 362 | "Can't import brotli. brotli decompression " "will be ignored !" |
| 363 | ) |
| 364 | elif "zstd" in encodings: |
| 365 | if _is_zstd_available: |
| 366 | # Using its streaming API since its simple API could handle |
| 367 | # only cases where there is content size data embedded in |
| 368 | # the frame |
| 369 | bio = io.BytesIO(s) |
| 370 | reader = zstandard.ZstdDecompressor().stream_reader(bio) |
| 371 | s = reader.read() |
| 372 | else: |
| 373 | log_loading.info( |
| 374 | "Can't import zstandard. zstd decompression " |
| 375 | "will be ignored !" |
nothing calls this directly
no test coverage detected