stouputils.archive.repair.scanner module#

Byte level scanning of a zip archive whose structure cannot be trusted.

The standard library refuses to open a damaged archive, so every offset here is treated as a hint: signatures are searched for, headers are bounds checked, and anything unreadable is reported as None.

class LocalHeader( method: int, csize: int, flags: int, name: str, data_start: int, )[source]#

Bases: object

Local file header, once read and bounds checked.

method: int[source]#: Compression method, 0 for stored and 8 for deflated.

csize: int[source]#: Compressed size announced by the header, often wrong in a damaged archive.

flags: int[source]#: General purpose bit flags.

name: str[source]#: Entry name as decoded from the header.

data_start: int[source]#: Offset of the first compressed byte, right after the name and extra fields.

class CentralEntry( name: str, method: int, csize: int, local_offset: int, )[source]#

Bases: object

Central directory entry, which usually survives better than the local headers.

name: str[source]#: Entry name as decoded from the directory.

method: int[source]#: Compression method announced by the directory.

csize: int[source]#: Compressed size announced by the directory.

local_offset: int[source]#: Announced offset of the matching local header, used as a hint only.

class ZipScanner(data: bytes)[source]#

Bases: object

Read only view over the bytes of an archive, with every lookup tolerant to corruption. .. rubric:: Examples

>>> import io, zipfile
>>> buffer = io.BytesIO()
>>> with zipfile.ZipFile(buffer, "w") as archive:
...     archive.writestr("pack.mcmeta", '{"pack": {}}')
>>> scanner = ZipScanner(buffer.getvalue())
>>> [entry.name for entry in scanner.central_entries()]
['pack.mcmeta']

LOCAL_SIGNATURE: ClassVar[bytes] = b'PK\x03\x04'[source]#: Magic bytes starting a local file header.

CENTRAL_SIGNATURE: ClassVar[bytes] = b'PK\x01\x02'[source]#: Magic bytes starting a central directory entry.

EOCD_SIGNATURE: ClassVar[bytes] = b'PK\x05\x06'[source]#: Magic bytes starting the end of central directory record.

LOCAL_HEADER_SIZE: ClassVar[int] = 30[source]#: Size of a local file header, before its variable length name and extra fields.

CENTRAL_HEADER_SIZE: ClassVar[int] = 46[source]#: Size of a central directory entry, before its variable length name, extra and comment fields.

UTF8_NAME_FLAG: ClassVar[int] = 2048[source]#: General purpose bit telling that the entry name is utf-8 encoded instead of cp437.

SEARCH_BACKWARD: ClassVar[int] = 32[source]#: How far before a broken offset hint a local header is looked for.

SEARCH_FORWARD: ClassVar[int] = 8192[source]#: How far after a broken offset hint a local header is looked for.

data: bytes[source]#: Whole archive, read in memory once.

signature_positions: list[int][source]#: Sorted offsets of every zip signature, used to guess where a compressed stream ends.

property size: int[source]#: Total number of bytes of the archive.

find_all(signature: bytes) → list[int][source]#

Collect every offset where the signature appears.

Parameters:: signature – Magic bytes to look for
Returns:: Offsets in increasing order

next_signature(start: int) → int[source]#

Offset of the first zip signature at or after the given position, or the end of the archive.

Parameters:: start – Offset to search from
Returns:: Offset of the next signature, or the archive size when there is none

static decode_name(raw_name: bytes, flags: int) → str[source]#

Decode an entry name with the encoding announced by the header flags.

Parameters:

raw_name – Name as stored in the archive
flags – General purpose bit flags of the entry

Returns:

Decoded name, with unreadable bytes replaced

Examples

>>> ZipScanner.decode_name(b"assets/", 0)
'assets/'

static sanitize_name( name: str, fallback_index: int, ) → str[source]#

Turn a possibly damaged entry name into a name safe to write in the repaired archive.

Parameters:

name – Decoded entry name
fallback_index – Index used to name an entry whose name is empty

Returns:

Sanitized name

Examples

>>> ZipScanner.sanitize_name("\\assets\\icon.png", 0)
'assets/icon.png'
>>> ZipScanner.sanitize_name("", 7)
'recovered_7'

find_local_header_near(offset_hint: int) → int[source]#

Look for a local header around an offset announced by the central directory.

Parameters:: offset_hint – Offset announced by the central directory
Returns:: Offset of the closest local header, or -1 when none is found nearby

read_local_header( offset: int, ) → LocalHeader | None[source]#

Read the local header at the given offset.

Parameters:: offset – Offset of the local signature
Returns:: Header, or None when the bytes there cannot be read as one

decode_range( method: int, data_start: int, end: int, ) → tuple[bytes, int] | None[source]#

Decompress a candidate byte range.

Parameters:

method – Compression method, 0 for stored and 8 for deflated
data_start – Offset of the first compressed byte
end – Offset where the compressed stream is assumed to end

Returns:

Content and offset right after it, or None when it does not decode

extract_content( method: int, data_start: int, size_hint: int | None, ) → tuple[bytes, int] | None[source]#

Decompress an entry, trying the announced size first and then guessed ends.

Parameters:

method – Compression method, 0 for stored and 8 for deflated
data_start – Offset of the first compressed byte
size_hint – Compressed size announced by a header, when there is one

Returns:

Content and offset right after it, or None when nothing decodes

central_entries() → list[CentralEntry][source]#: Read every readable central directory entry. :returns: Entries in the order they appear in the archive