This page hosts a formal specification of Zchunk using Kaitai Struct. This specification can be automatically translated into a variety of programming languages to get a parsing library.
All Python code generated by Kaitai Struct depends on the Kaitai Struct runtime library for Python. You must add this dependency to your project before you can parse or serialize any data.
The Python runtime library can be installed from PyPI:
python3 -m pip install kaitaistructParse a local file and get structure in memory:
data = Zchunk.from_file("path/to/local/file.zck")
Or parse structure from a bytes:
from kaitaistruct import KaitaiStream, BytesIO
raw = b"\x00\x01\x02..."
data = Zchunk(KaitaiStream(BytesIO(raw)))
After that, one can get various attributes from the structure by invoking getter methods like:
data.dict # => Custom dictionary used when compressing each chunk. It's compressed itself
without a dictionary.
The official zchunk specification calls this section "Compressed Dict".
It's also called a "dictionary chunk". `zck_read_header -c` presents it as
"chunk 0" (which is always shown in the chunk table, but can have size 0
if the dictionary is not in use).
# This is a generated file! Please edit source .ksy file and use kaitai-struct-compiler to rebuild
# type: ignore
import kaitaistruct
from kaitaistruct import KaitaiStruct, KaitaiStream, BytesIO
from enum import IntEnum
if getattr(kaitaistruct, 'API_VERSION', (0, 9)) < (0, 11):
raise Exception("Incompatible Kaitai Struct Python API: 0.11 or later is required, but you have %s" % (kaitaistruct.__version__))
class Zchunk(KaitaiStruct):
"""
.. seealso::
Source - https://github.com/zchunk/zchunk/blob/99e51afa38c723e7c25834c2c3b305d20ef55d04/zchunk_format.txt
"""
class ChecksumTypes(IntEnum):
sha1 = 0
sha256 = 1
sha512 = 2
sha512_128 = 3
class CompressionTypes(IntEnum):
none = 0
zstd = 2
def __init__(self, _io, _parent=None, _root=None):
super(Zchunk, self).__init__(_io)
self._parent = _parent
self._root = _root or self
self._read()
def _read(self):
self.lead = Zchunk.HeaderLead(self._io, self, self._root)
self._raw_header_rest = self._io.read_bytes(self.lead.len_header_rest.value)
_io__raw_header_rest = KaitaiStream(BytesIO(self._raw_header_rest))
self.header_rest = Zchunk.HeaderWithoutLead(_io__raw_header_rest, self, self._root)
self.dict = self._io.read_bytes(self.header_rest.index.len_dict.value)
if (not (self.lead.is_detached_header)):
pass
self.chunks = []
for i in range(len(self.header_rest.index.chunks_metadata)):
self.chunks.append(self._io.read_bytes(self.header_rest.index.chunks_metadata[i].len_chunk.value))
def _fetch_instances(self):
pass
self.lead._fetch_instances()
self.header_rest._fetch_instances()
if (not (self.lead.is_detached_header)):
pass
for i in range(len(self.chunks)):
pass
class ChecksumType(KaitaiStruct):
def __init__(self, _io, _parent=None, _root=None):
super(Zchunk.ChecksumType, self).__init__(_io)
self._parent = _parent
self._root = _root
self._read()
def _read(self):
self.raw = Zchunk.CompressedInteger(self._io, self, self._root)
_ = self.raw
if not self.len_checksum != 0:
raise kaitaistruct.ValidationExprError(self.raw, self._io, u"/types/checksum_type/seq/0")
def _fetch_instances(self):
pass
self.raw._fetch_instances()
@property
def len_checksum(self):
if hasattr(self, '_m_len_checksum'):
return self._m_len_checksum
self._m_len_checksum = (20 if self.value == Zchunk.ChecksumTypes.sha1 else (32 if self.value == Zchunk.ChecksumTypes.sha256 else (64 if self.value == Zchunk.ChecksumTypes.sha512 else (16 if self.value == Zchunk.ChecksumTypes.sha512_128 else 0))))
return getattr(self, '_m_len_checksum', None)
@property
def value(self):
if hasattr(self, '_m_value'):
return self._m_value
self._m_value = KaitaiStream.resolve_enum(Zchunk.ChecksumTypes, self.raw.value)
return getattr(self, '_m_value', None)
class Chunk(KaitaiStruct):
def __init__(self, len_checksum, has_data_streams, has_uncompressed_source, _io, _parent=None, _root=None):
super(Zchunk.Chunk, self).__init__(_io)
self._parent = _parent
self._root = _root
self.len_checksum = len_checksum
self.has_data_streams = has_data_streams
self.has_uncompressed_source = has_uncompressed_source
self._read()
def _read(self):
if self.has_data_streams:
pass
self.chunk_stream = Zchunk.CompressedInteger(self._io, self, self._root)
self.chunk_checksum = self._io.read_bytes(self.len_checksum)
if self.has_uncompressed_source:
pass
self.uncompressed_chunk_checksum = self._io.read_bytes(self.len_checksum)
self.len_chunk = Zchunk.CompressedInteger(self._io, self, self._root)
self.len_uncompressed_chunk = Zchunk.CompressedInteger(self._io, self, self._root)
def _fetch_instances(self):
pass
if self.has_data_streams:
pass
self.chunk_stream._fetch_instances()
if self.has_uncompressed_source:
pass
self.len_chunk._fetch_instances()
self.len_uncompressed_chunk._fetch_instances()
class CompressedInteger(KaitaiStruct):
"""Like `/common/vlq_base128_le` (LEB128), but the logic of the
"continuation" flag in the most significant bit is inverted, so instead of
`has_next`, it is called `is_last` (if the highest bit is set to zero, it
means "continue", whereas in standard LEB128, the highest bit set to
**one** means "continue"). Therefore, we cannot simply import
`/common/vlq_base128_le` and use it, because it is incompatible.
"""
def __init__(self, _io, _parent=None, _root=None):
super(Zchunk.CompressedInteger, self).__init__(_io)
self._parent = _parent
self._root = _root
self._read()
def _read(self):
self.groups = []
i = 0
while True:
_ = Zchunk.CompressedInteger.Group(i, self._io, self, self._root)
self.groups.append(_)
if _.is_last:
break
i += 1
def _fetch_instances(self):
pass
for i in range(len(self.groups)):
pass
self.groups[i]._fetch_instances()
class Group(KaitaiStruct):
"""One byte group, clearly divided into 7-bit "value" chunk and 1-bit "continuation" flag.
"""
def __init__(self, idx, _io, _parent=None, _root=None):
super(Zchunk.CompressedInteger.Group, self).__init__(_io)
self._parent = _parent
self._root = _root
self.idx = idx
self._read()
def _read(self):
self.is_last = self._io.read_bits_int_be(1) != 0
if not self.is_last == (True if self.idx == 9 else self.is_last):
raise kaitaistruct.ValidationNotEqualError((True if self.idx == 9 else self.is_last), self.is_last, self._io, u"/types/compressed_integer/types/group/seq/0")
self.value = self._io.read_bits_int_be(7)
if not self.value <= (1 if self.idx == 9 else 127):
raise kaitaistruct.ValidationGreaterThanError((1 if self.idx == 9 else 127), self.value, self._io, u"/types/compressed_integer/types/group/seq/1")
def _fetch_instances(self):
pass
@property
def len(self):
if hasattr(self, '_m_len'):
return self._m_len
self._m_len = len(self.groups)
return getattr(self, '_m_len', None)
@property
def value(self):
"""Resulting unsigned value as normal integer."""
if hasattr(self, '_m_value'):
return self._m_value
self._m_value = (((((((((self.groups[0].value | (self.groups[1].value << 7 if self.len >= 2 else 0)) | (self.groups[2].value << 14 if self.len >= 3 else 0)) | (self.groups[3].value << 21 if self.len >= 4 else 0)) | (self.groups[4].value << 28 if self.len >= 5 else 0)) | (self.groups[5].value << 35 if self.len >= 6 else 0)) | (self.groups[6].value << 42 if self.len >= 7 else 0)) | (self.groups[7].value << 49 if self.len >= 8 else 0)) | (self.groups[8].value << 56 if self.len >= 9 else 0)) | (self.groups[9].value << 63 if self.len >= 10 else 0))
return getattr(self, '_m_value', None)
class HeaderLead(KaitaiStruct):
def __init__(self, _io, _parent=None, _root=None):
super(Zchunk.HeaderLead, self).__init__(_io)
self._parent = _parent
self._root = _root
self._read()
def _read(self):
self.magic = self._io.read_bytes(5)
if not ((self.magic == b"\x00\x5A\x43\x4B\x31") or (self.magic == b"\x00\x5A\x48\x52\x31")) :
raise kaitaistruct.ValidationNotAnyOfError(self.magic, self._io, u"/types/header_lead/seq/0")
self.overall_checksum_type = Zchunk.ChecksumType(self._io, self, self._root)
self.len_header_rest = Zchunk.CompressedInteger(self._io, self, self._root)
self.header_checksum = self._io.read_bytes(self.overall_checksum_type.len_checksum)
def _fetch_instances(self):
pass
self.overall_checksum_type._fetch_instances()
self.len_header_rest._fetch_instances()
@property
def is_detached_header(self):
"""Determines whether this file is a zchunk detached header (`.zhr`). If
not, it is a complete zchunk file (`.zck`).
"""
if hasattr(self, '_m_is_detached_header'):
return self._m_is_detached_header
self._m_is_detached_header = KaitaiStream.byte_array_index(self.magic, 2) == 72
return getattr(self, '_m_is_detached_header', None)
class HeaderWithoutLead(KaitaiStruct):
def __init__(self, _io, _parent=None, _root=None):
super(Zchunk.HeaderWithoutLead, self).__init__(_io)
self._parent = _parent
self._root = _root
self._read()
def _read(self):
self.preface = Zchunk.Preface(self._io, self, self._root)
self.len_index = Zchunk.CompressedInteger(self._io, self, self._root)
self._raw_index = self._io.read_bytes(self.len_index.value)
_io__raw_index = KaitaiStream(BytesIO(self._raw_index))
self.index = Zchunk.Index(_io__raw_index, self, self._root)
self.num_signatures = Zchunk.CompressedInteger(self._io, self, self._root)
_ = self.num_signatures
if not _.value == 0:
raise kaitaistruct.ValidationExprError(self.num_signatures, self._io, u"/types/header_without_lead/seq/3")
def _fetch_instances(self):
pass
self.preface._fetch_instances()
self.len_index._fetch_instances()
self.index._fetch_instances()
self.num_signatures._fetch_instances()
class Index(KaitaiStruct):
def __init__(self, _io, _parent=None, _root=None):
super(Zchunk.Index, self).__init__(_io)
self._parent = _parent
self._root = _root
self._read()
def _read(self):
self.chunk_checksum_type = Zchunk.ChecksumType(self._io, self, self._root)
self.num_chunks = Zchunk.CompressedInteger(self._io, self, self._root)
_ = self.num_chunks
if not _.value >= 1:
raise kaitaistruct.ValidationExprError(self.num_chunks, self._io, u"/types/index/seq/1")
if self._parent.preface.has_data_streams:
pass
self.dict_stream = Zchunk.CompressedInteger(self._io, self, self._root)
_ = self.dict_stream
if not _.value == 0:
raise kaitaistruct.ValidationExprError(self.dict_stream, self._io, u"/types/index/seq/2")
self.dict_checksum = self._io.read_bytes(self.chunk_checksum_type.len_checksum)
if self._parent.preface.has_uncompressed_source:
pass
self.uncompressed_dict_checksum = self._io.read_bytes(self.chunk_checksum_type.len_checksum)
self.len_dict = Zchunk.CompressedInteger(self._io, self, self._root)
self.len_uncompressed_dict = Zchunk.CompressedInteger(self._io, self, self._root)
self.chunks_metadata = []
for i in range(self.num_data_chunks):
self.chunks_metadata.append(Zchunk.Chunk(self.chunk_checksum_type.len_checksum, self._parent.preface.has_data_streams, self._parent.preface.has_uncompressed_source, self._io, self, self._root))
def _fetch_instances(self):
pass
self.chunk_checksum_type._fetch_instances()
self.num_chunks._fetch_instances()
if self._parent.preface.has_data_streams:
pass
self.dict_stream._fetch_instances()
if self._parent.preface.has_uncompressed_source:
pass
self.len_dict._fetch_instances()
self.len_uncompressed_dict._fetch_instances()
for i in range(len(self.chunks_metadata)):
pass
self.chunks_metadata[i]._fetch_instances()
@property
def num_data_chunks(self):
"""Number of data chunks. `num_chunks` counts the dictionary as chunk 0,
so it is one greater than this number.
"""
if hasattr(self, '_m_num_data_chunks'):
return self._m_num_data_chunks
self._m_num_data_chunks = self.num_chunks.value - 1
return getattr(self, '_m_num_data_chunks', None)
class OptionalElement(KaitaiStruct):
def __init__(self, _io, _parent=None, _root=None):
super(Zchunk.OptionalElement, self).__init__(_io)
self._parent = _parent
self._root = _root
self._read()
def _read(self):
self.element_id = Zchunk.CompressedInteger(self._io, self, self._root)
self.len_data = Zchunk.CompressedInteger(self._io, self, self._root)
self.data = self._io.read_bytes(self.len_data.value)
def _fetch_instances(self):
pass
self.element_id._fetch_instances()
self.len_data._fetch_instances()
class Preface(KaitaiStruct):
def __init__(self, _io, _parent=None, _root=None):
super(Zchunk.Preface, self).__init__(_io)
self._parent = _parent
self._root = _root
self._read()
def _read(self):
self.data_checksum = self._io.read_bytes(self._root.lead.overall_checksum_type.len_checksum)
self.flags = Zchunk.CompressedInteger(self._io, self, self._root)
_ = self.flags
if not _.value <= 7:
raise kaitaistruct.ValidationExprError(self.flags, self._io, u"/types/preface/seq/1")
self.compression_type_int = Zchunk.CompressedInteger(self._io, self, self._root)
_ = self.compression_type_int
if not ((_.value == int(Zchunk.CompressionTypes.none)) or (_.value == int(Zchunk.CompressionTypes.zstd))) :
raise kaitaistruct.ValidationExprError(self.compression_type_int, self._io, u"/types/preface/seq/2")
if self.has_optional_elements:
pass
self.num_optional_elements = Zchunk.CompressedInteger(self._io, self, self._root)
_ = self.num_optional_elements
if not _.value >= 1:
raise kaitaistruct.ValidationExprError(self.num_optional_elements, self._io, u"/types/preface/seq/3")
if self.has_optional_elements:
pass
self.optional_elements = []
for i in range(self.num_optional_elements.value):
self.optional_elements.append(Zchunk.OptionalElement(self._io, self, self._root))
def _fetch_instances(self):
pass
self.flags._fetch_instances()
self.compression_type_int._fetch_instances()
if self.has_optional_elements:
pass
self.num_optional_elements._fetch_instances()
if self.has_optional_elements:
pass
for i in range(len(self.optional_elements)):
pass
self.optional_elements[i]._fetch_instances()
@property
def compression_type(self):
if hasattr(self, '_m_compression_type'):
return self._m_compression_type
self._m_compression_type = KaitaiStream.resolve_enum(Zchunk.CompressionTypes, self.compression_type_int.value)
return getattr(self, '_m_compression_type', None)
@property
def has_data_streams(self):
if hasattr(self, '_m_has_data_streams'):
return self._m_has_data_streams
self._m_has_data_streams = self.flags.value & 1 != 0
return getattr(self, '_m_has_data_streams', None)
@property
def has_optional_elements(self):
if hasattr(self, '_m_has_optional_elements'):
return self._m_has_optional_elements
self._m_has_optional_elements = self.flags.value & 2 != 0
return getattr(self, '_m_has_optional_elements', None)
@property
def has_uncompressed_source(self):
"""The file may be applied against an uncompressed source. This adds an
uncompressed checksum to every index entry, including the dictionary.
"""
if hasattr(self, '_m_has_uncompressed_source'):
return self._m_has_uncompressed_source
self._m_has_uncompressed_source = self.flags.value & 4 != 0
return getattr(self, '_m_has_uncompressed_source', None)