Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

For investigative purposes, use stream-inflate for everything #83

Closed
wants to merge 1 commit into from
Closed

For investigative purposes, use stream-inflate for everything #83

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyproject.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@ dev = [
]
ci = [
"pycryptodome==3.10.1",
"stream-inflate==0.0.12",
"stream-inflate==0.0.15",
"coverage==6.2",
"pytest==6.2.5",
"pytest-cov==3.0.0",
Expand Down
50 changes: 29 additions & 21 deletions stream_unzip.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
from Crypto.Util import Counter
from Crypto.Protocol.KDF import PBKDF2

from stream_inflate import stream_inflate64
from stream_inflate import stream_inflate, stream_inflate64


def stream_unzip(zipfile_chunks, password=None, chunk_size=65536, allow_zip64=True):
Expand Down Expand Up @@ -112,32 +112,40 @@ def _num_unused():

return _decompress, _is_done, _num_unused

def get_decompressor_deflate():
dobj = zlib.decompressobj(wbits=-zlib.MAX_WBITS)
# def get_decompressor_deflate():
# dobj = zlib.decompressobj(wbits=-zlib.MAX_WBITS)

def _decompress_single(compressed_chunk):
try:
return dobj.decompress(compressed_chunk, chunk_size)
except zlib.error as e:
raise DeflateError() from e
# def _decompress_single(compressed_chunk):
# try:
# return dobj.decompress(compressed_chunk, chunk_size)
# except zlib.error as e:
# raise DeflateError() from e

def _decompress(compressed_chunk):
uncompressed_chunk = _decompress_single(compressed_chunk)
if uncompressed_chunk:
yield uncompressed_chunk
# def _decompress(compressed_chunk):
# uncompressed_chunk = _decompress_single(compressed_chunk)
# if uncompressed_chunk:
# yield uncompressed_chunk

while dobj.unconsumed_tail and not dobj.eof:
uncompressed_chunk = _decompress_single(dobj.unconsumed_tail)
if uncompressed_chunk:
yield uncompressed_chunk
# while dobj.unconsumed_tail and not dobj.eof:
# uncompressed_chunk = _decompress_single(dobj.unconsumed_tail)
# if uncompressed_chunk:
# yield uncompressed_chunk

def _is_done():
return dobj.eof
# def _is_done():
# return dobj.eof

def _num_unused():
return len(dobj.unused_data)
# def _num_unused():
# return len(dobj.unused_data)

return _decompress, _is_done, _num_unused
# return _decompress, _is_done, _num_unused

def get_decompressor_deflate():
uncompressed_chunks, is_done, num_bytes_unconsumed = stream_inflate()

def _decompress(compressed_chunk):
yield from uncompressed_chunks((compressed_chunk,))

return _decompress, is_done, num_bytes_unconsumed

def get_decompressor_deflate64():
uncompressed_chunks, is_done, num_bytes_unconsumed = stream_inflate64()
Expand Down
Loading