diff options
| author | Xusheng <xusheng@vector35.com> | 2021-07-01 12:34:30 +0800 |
|---|---|---|
| committer | Xusheng Li <xushengli@protonmail.com> | 2021-07-05 10:11:47 +0800 |
| commit | 13ff200ba134b8704f37eca99c42b70dab0d17dc (patch) | |
| tree | 876b3db3c53f704b31152aeaede1f2edc9d27b24 /python/examples/kaitai/zip.py | |
| parent | 4039a800349dcd587833d3e27d9137e4649818f4 (diff) | |
Remove snippet and kaitai plugin. Note their individual links.
Diffstat (limited to 'python/examples/kaitai/zip.py')
| -rw-r--r-- | python/examples/kaitai/zip.py | 502 |
1 files changed, 0 insertions, 502 deletions
diff --git a/python/examples/kaitai/zip.py b/python/examples/kaitai/zip.py deleted file mode 100644 index 9fe85264..00000000 --- a/python/examples/kaitai/zip.py +++ /dev/null @@ -1,502 +0,0 @@ -# This is a generated file! Please edit source .ksy file and use kaitai-struct-compiler to rebuild - -from pkg_resources import parse_version -from .kaitaistruct import __version__ as ks_version, KaitaiStruct, KaitaiStream, BytesIO -from enum import Enum -import collections - - -if parse_version(ks_version) < parse_version('0.7'): - raise Exception("Incompatible Kaitai Struct Python API: 0.7 or later is required, but you have %s" % (ks_version)) - -class Zip(KaitaiStruct): - """ - .. seealso:: - Source - https://pkware.cachefly.net/webdocs/casestudies/APPNOTE.TXT - """ - - class Compression(Enum): - none = 0 - shrunk = 1 - reduced_1 = 2 - reduced_2 = 3 - reduced_3 = 4 - reduced_4 = 5 - imploded = 6 - deflated = 8 - enhanced_deflated = 9 - pkware_dcl_imploded = 10 - bzip2 = 12 - lzma = 14 - ibm_terse = 18 - ibm_lz77_z = 19 - ppmd = 98 - - class ExtraCodes(Enum): - zip64 = 1 - av_info = 7 - os2 = 9 - ntfs = 10 - openvms = 12 - pkware_unix = 13 - file_stream_and_fork_descriptors = 14 - patch_descriptor = 15 - pkcs7 = 20 - x509_cert_id_and_signature_for_file = 21 - x509_cert_id_for_central_dir = 22 - strong_encryption_header = 23 - record_management_controls = 24 - pkcs7_enc_recip_cert_list = 25 - ibm_s390_uncomp = 101 - ibm_s390_comp = 102 - poszip_4690 = 18064 - extended_timestamp = 21589 - infozip_unix = 30805 - infozip_unix_var_size = 30837 - SEQ_FIELDS = ["sections"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['sections']['start'] = self._io.pos() - self.sections = [] - i = 0 - while not self._io.is_eof(): - if not 'arr' in self._debug['sections']: - self._debug['sections']['arr'] = [] - self._debug['sections']['arr'].append({'start': self._io.pos()}) - _t_sections = self._root.PkSection(self._io, self, self._root) - _t_sections._read() - self.sections.append(_t_sections) - self._debug['sections']['arr'][len(self.sections) - 1]['end'] = self._io.pos() - i += 1 - - self._debug['sections']['end'] = self._io.pos() - - class LocalFile(KaitaiStruct): - SEQ_FIELDS = ["header", "body"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['header']['start'] = self._io.pos() - self.header = self._root.LocalFileHeader(self._io, self, self._root) - self.header._read() - self._debug['header']['end'] = self._io.pos() - self._debug['body']['start'] = self._io.pos() - self.body = self._io.read_bytes(self.header.compressed_size) - self._debug['body']['end'] = self._io.pos() - - - class ExtraField(KaitaiStruct): - SEQ_FIELDS = ["code", "size", "body"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['code']['start'] = self._io.pos() - self.code = KaitaiStream.resolve_enum(self._root.ExtraCodes, self._io.read_u2le()) - self._debug['code']['end'] = self._io.pos() - self._debug['size']['start'] = self._io.pos() - self.size = self._io.read_u2le() - self._debug['size']['end'] = self._io.pos() - self._debug['body']['start'] = self._io.pos() - _on = self.code - if _on == self._root.ExtraCodes.ntfs: - self._raw_body = self._io.read_bytes(self.size) - io = KaitaiStream(BytesIO(self._raw_body)) - self.body = self._root.ExtraField.Ntfs(io, self, self._root) - self.body._read() - elif _on == self._root.ExtraCodes.extended_timestamp: - self._raw_body = self._io.read_bytes(self.size) - io = KaitaiStream(BytesIO(self._raw_body)) - self.body = self._root.ExtraField.ExtendedTimestamp(io, self, self._root) - self.body._read() - elif _on == self._root.ExtraCodes.infozip_unix_var_size: - self._raw_body = self._io.read_bytes(self.size) - io = KaitaiStream(BytesIO(self._raw_body)) - self.body = self._root.ExtraField.InfozipUnixVarSize(io, self, self._root) - self.body._read() - else: - self.body = self._io.read_bytes(self.size) - self._debug['body']['end'] = self._io.pos() - - class Ntfs(KaitaiStruct): - """ - .. seealso:: - Source - https://github.com/LuaDist/zip/blob/master/proginfo/extrafld.txt#L191 - """ - SEQ_FIELDS = ["reserved", "attributes"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['reserved']['start'] = self._io.pos() - self.reserved = self._io.read_u4le() - self._debug['reserved']['end'] = self._io.pos() - self._debug['attributes']['start'] = self._io.pos() - self.attributes = [] - i = 0 - while not self._io.is_eof(): - if not 'arr' in self._debug['attributes']: - self._debug['attributes']['arr'] = [] - self._debug['attributes']['arr'].append({'start': self._io.pos()}) - _t_attributes = self._root.ExtraField.Ntfs.Attribute(self._io, self, self._root) - _t_attributes._read() - self.attributes.append(_t_attributes) - self._debug['attributes']['arr'][len(self.attributes) - 1]['end'] = self._io.pos() - i += 1 - - self._debug['attributes']['end'] = self._io.pos() - - class Attribute(KaitaiStruct): - SEQ_FIELDS = ["tag", "size", "body"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['tag']['start'] = self._io.pos() - self.tag = self._io.read_u2le() - self._debug['tag']['end'] = self._io.pos() - self._debug['size']['start'] = self._io.pos() - self.size = self._io.read_u2le() - self._debug['size']['end'] = self._io.pos() - self._debug['body']['start'] = self._io.pos() - _on = self.tag - if _on == 1: - self._raw_body = self._io.read_bytes(self.size) - io = KaitaiStream(BytesIO(self._raw_body)) - self.body = self._root.ExtraField.Ntfs.Attribute1(io, self, self._root) - self.body._read() - else: - self.body = self._io.read_bytes(self.size) - self._debug['body']['end'] = self._io.pos() - - - class Attribute1(KaitaiStruct): - SEQ_FIELDS = ["last_mod_time", "last_access_time", "creation_time"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['last_mod_time']['start'] = self._io.pos() - self.last_mod_time = self._io.read_u8le() - self._debug['last_mod_time']['end'] = self._io.pos() - self._debug['last_access_time']['start'] = self._io.pos() - self.last_access_time = self._io.read_u8le() - self._debug['last_access_time']['end'] = self._io.pos() - self._debug['creation_time']['start'] = self._io.pos() - self.creation_time = self._io.read_u8le() - self._debug['creation_time']['end'] = self._io.pos() - - - - class ExtendedTimestamp(KaitaiStruct): - """ - .. seealso:: - Source - https://github.com/LuaDist/zip/blob/master/proginfo/extrafld.txt#L817 - """ - SEQ_FIELDS = ["flags", "mod_time", "access_time", "create_time"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['flags']['start'] = self._io.pos() - self.flags = self._io.read_u1() - self._debug['flags']['end'] = self._io.pos() - self._debug['mod_time']['start'] = self._io.pos() - self.mod_time = self._io.read_u4le() - self._debug['mod_time']['end'] = self._io.pos() - if not (self._io.is_eof()): - self._debug['access_time']['start'] = self._io.pos() - self.access_time = self._io.read_u4le() - self._debug['access_time']['end'] = self._io.pos() - - if not (self._io.is_eof()): - self._debug['create_time']['start'] = self._io.pos() - self.create_time = self._io.read_u4le() - self._debug['create_time']['end'] = self._io.pos() - - - - class InfozipUnixVarSize(KaitaiStruct): - """ - .. seealso:: - Source - https://github.com/LuaDist/zip/blob/master/proginfo/extrafld.txt#L1339 - """ - SEQ_FIELDS = ["version", "uid_size", "uid", "gid_size", "gid"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['version']['start'] = self._io.pos() - self.version = self._io.read_u1() - self._debug['version']['end'] = self._io.pos() - self._debug['uid_size']['start'] = self._io.pos() - self.uid_size = self._io.read_u1() - self._debug['uid_size']['end'] = self._io.pos() - self._debug['uid']['start'] = self._io.pos() - self.uid = self._io.read_bytes(self.uid_size) - self._debug['uid']['end'] = self._io.pos() - self._debug['gid_size']['start'] = self._io.pos() - self.gid_size = self._io.read_u1() - self._debug['gid_size']['end'] = self._io.pos() - self._debug['gid']['start'] = self._io.pos() - self.gid = self._io.read_bytes(self.gid_size) - self._debug['gid']['end'] = self._io.pos() - - - - class CentralDirEntry(KaitaiStruct): - """ - .. seealso:: - - 4.3.12 - https://pkware.cachefly.net/webdocs/casestudies/APPNOTE.TXT - """ - SEQ_FIELDS = ["version_made_by", "version_needed_to_extract", "flags", "compression_method", "last_mod_file_time", "last_mod_file_date", "crc32", "compressed_size", "uncompressed_size", "file_name_len", "extra_len", "comment_len", "disk_number_start", "int_file_attr", "ext_file_attr", "local_header_offset", "file_name", "extra", "comment"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['version_made_by']['start'] = self._io.pos() - self.version_made_by = self._io.read_u2le() - self._debug['version_made_by']['end'] = self._io.pos() - self._debug['version_needed_to_extract']['start'] = self._io.pos() - self.version_needed_to_extract = self._io.read_u2le() - self._debug['version_needed_to_extract']['end'] = self._io.pos() - self._debug['flags']['start'] = self._io.pos() - self.flags = self._io.read_u2le() - self._debug['flags']['end'] = self._io.pos() - self._debug['compression_method']['start'] = self._io.pos() - self.compression_method = KaitaiStream.resolve_enum(self._root.Compression, self._io.read_u2le()) - self._debug['compression_method']['end'] = self._io.pos() - self._debug['last_mod_file_time']['start'] = self._io.pos() - self.last_mod_file_time = self._io.read_u2le() - self._debug['last_mod_file_time']['end'] = self._io.pos() - self._debug['last_mod_file_date']['start'] = self._io.pos() - self.last_mod_file_date = self._io.read_u2le() - self._debug['last_mod_file_date']['end'] = self._io.pos() - self._debug['crc32']['start'] = self._io.pos() - self.crc32 = self._io.read_u4le() - self._debug['crc32']['end'] = self._io.pos() - self._debug['compressed_size']['start'] = self._io.pos() - self.compressed_size = self._io.read_u4le() - self._debug['compressed_size']['end'] = self._io.pos() - self._debug['uncompressed_size']['start'] = self._io.pos() - self.uncompressed_size = self._io.read_u4le() - self._debug['uncompressed_size']['end'] = self._io.pos() - self._debug['file_name_len']['start'] = self._io.pos() - self.file_name_len = self._io.read_u2le() - self._debug['file_name_len']['end'] = self._io.pos() - self._debug['extra_len']['start'] = self._io.pos() - self.extra_len = self._io.read_u2le() - self._debug['extra_len']['end'] = self._io.pos() - self._debug['comment_len']['start'] = self._io.pos() - self.comment_len = self._io.read_u2le() - self._debug['comment_len']['end'] = self._io.pos() - self._debug['disk_number_start']['start'] = self._io.pos() - self.disk_number_start = self._io.read_u2le() - self._debug['disk_number_start']['end'] = self._io.pos() - self._debug['int_file_attr']['start'] = self._io.pos() - self.int_file_attr = self._io.read_u2le() - self._debug['int_file_attr']['end'] = self._io.pos() - self._debug['ext_file_attr']['start'] = self._io.pos() - self.ext_file_attr = self._io.read_u4le() - self._debug['ext_file_attr']['end'] = self._io.pos() - self._debug['local_header_offset']['start'] = self._io.pos() - self.local_header_offset = self._io.read_s4le() - self._debug['local_header_offset']['end'] = self._io.pos() - self._debug['file_name']['start'] = self._io.pos() - self.file_name = (self._io.read_bytes(self.file_name_len)).decode(u"UTF-8") - self._debug['file_name']['end'] = self._io.pos() - self._debug['extra']['start'] = self._io.pos() - self._raw_extra = self._io.read_bytes(self.extra_len) - io = KaitaiStream(BytesIO(self._raw_extra)) - self.extra = self._root.Extras(io, self, self._root) - self.extra._read() - self._debug['extra']['end'] = self._io.pos() - self._debug['comment']['start'] = self._io.pos() - self.comment = (self._io.read_bytes(self.comment_len)).decode(u"UTF-8") - self._debug['comment']['end'] = self._io.pos() - - @property - def local_header(self): - if hasattr(self, '_m_local_header'): - return self._m_local_header if hasattr(self, '_m_local_header') else None - - _pos = self._io.pos() - self._io.seek(self.local_header_offset) - self._debug['_m_local_header']['start'] = self._io.pos() - self._m_local_header = self._root.PkSection(self._io, self, self._root) - self._m_local_header._read() - self._debug['_m_local_header']['end'] = self._io.pos() - self._io.seek(_pos) - return self._m_local_header if hasattr(self, '_m_local_header') else None - - - class PkSection(KaitaiStruct): - SEQ_FIELDS = ["magic", "section_type", "body"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['magic']['start'] = self._io.pos() - self.magic = self._io.ensure_fixed_contents(b"\x50\x4B") - self._debug['magic']['end'] = self._io.pos() - self._debug['section_type']['start'] = self._io.pos() - self.section_type = self._io.read_u2le() - self._debug['section_type']['end'] = self._io.pos() - self._debug['body']['start'] = self._io.pos() - _on = self.section_type - if _on == 513: - self.body = self._root.CentralDirEntry(self._io, self, self._root) - self.body._read() - elif _on == 1027: - self.body = self._root.LocalFile(self._io, self, self._root) - self.body._read() - elif _on == 1541: - self.body = self._root.EndOfCentralDir(self._io, self, self._root) - self.body._read() - self._debug['body']['end'] = self._io.pos() - - - class Extras(KaitaiStruct): - SEQ_FIELDS = ["entries"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['entries']['start'] = self._io.pos() - self.entries = [] - i = 0 - while not self._io.is_eof(): - if not 'arr' in self._debug['entries']: - self._debug['entries']['arr'] = [] - self._debug['entries']['arr'].append({'start': self._io.pos()}) - _t_entries = self._root.ExtraField(self._io, self, self._root) - _t_entries._read() - self.entries.append(_t_entries) - self._debug['entries']['arr'][len(self.entries) - 1]['end'] = self._io.pos() - i += 1 - - self._debug['entries']['end'] = self._io.pos() - - - class LocalFileHeader(KaitaiStruct): - SEQ_FIELDS = ["version", "flags", "compression_method", "file_mod_time", "file_mod_date", "crc32", "compressed_size", "uncompressed_size", "file_name_len", "extra_len", "file_name", "extra"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['version']['start'] = self._io.pos() - self.version = self._io.read_u2le() - self._debug['version']['end'] = self._io.pos() - self._debug['flags']['start'] = self._io.pos() - self.flags = self._io.read_u2le() - self._debug['flags']['end'] = self._io.pos() - self._debug['compression_method']['start'] = self._io.pos() - self.compression_method = KaitaiStream.resolve_enum(self._root.Compression, self._io.read_u2le()) - self._debug['compression_method']['end'] = self._io.pos() - self._debug['file_mod_time']['start'] = self._io.pos() - self.file_mod_time = self._io.read_u2le() - self._debug['file_mod_time']['end'] = self._io.pos() - self._debug['file_mod_date']['start'] = self._io.pos() - self.file_mod_date = self._io.read_u2le() - self._debug['file_mod_date']['end'] = self._io.pos() - self._debug['crc32']['start'] = self._io.pos() - self.crc32 = self._io.read_u4le() - self._debug['crc32']['end'] = self._io.pos() - self._debug['compressed_size']['start'] = self._io.pos() - self.compressed_size = self._io.read_u4le() - self._debug['compressed_size']['end'] = self._io.pos() - self._debug['uncompressed_size']['start'] = self._io.pos() - self.uncompressed_size = self._io.read_u4le() - self._debug['uncompressed_size']['end'] = self._io.pos() - self._debug['file_name_len']['start'] = self._io.pos() - self.file_name_len = self._io.read_u2le() - self._debug['file_name_len']['end'] = self._io.pos() - self._debug['extra_len']['start'] = self._io.pos() - self.extra_len = self._io.read_u2le() - self._debug['extra_len']['end'] = self._io.pos() - self._debug['file_name']['start'] = self._io.pos() - self.file_name = (self._io.read_bytes(self.file_name_len)).decode(u"UTF-8") - self._debug['file_name']['end'] = self._io.pos() - self._debug['extra']['start'] = self._io.pos() - self._raw_extra = self._io.read_bytes(self.extra_len) - io = KaitaiStream(BytesIO(self._raw_extra)) - self.extra = self._root.Extras(io, self, self._root) - self.extra._read() - self._debug['extra']['end'] = self._io.pos() - - - class EndOfCentralDir(KaitaiStruct): - SEQ_FIELDS = ["disk_of_end_of_central_dir", "disk_of_central_dir", "qty_central_dir_entries_on_disk", "qty_central_dir_entries_total", "central_dir_size", "central_dir_offset", "comment_len", "comment"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['disk_of_end_of_central_dir']['start'] = self._io.pos() - self.disk_of_end_of_central_dir = self._io.read_u2le() - self._debug['disk_of_end_of_central_dir']['end'] = self._io.pos() - self._debug['disk_of_central_dir']['start'] = self._io.pos() - self.disk_of_central_dir = self._io.read_u2le() - self._debug['disk_of_central_dir']['end'] = self._io.pos() - self._debug['qty_central_dir_entries_on_disk']['start'] = self._io.pos() - self.qty_central_dir_entries_on_disk = self._io.read_u2le() - self._debug['qty_central_dir_entries_on_disk']['end'] = self._io.pos() - self._debug['qty_central_dir_entries_total']['start'] = self._io.pos() - self.qty_central_dir_entries_total = self._io.read_u2le() - self._debug['qty_central_dir_entries_total']['end'] = self._io.pos() - self._debug['central_dir_size']['start'] = self._io.pos() - self.central_dir_size = self._io.read_u4le() - self._debug['central_dir_size']['end'] = self._io.pos() - self._debug['central_dir_offset']['start'] = self._io.pos() - self.central_dir_offset = self._io.read_u4le() - self._debug['central_dir_offset']['end'] = self._io.pos() - self._debug['comment_len']['start'] = self._io.pos() - self.comment_len = self._io.read_u2le() - self._debug['comment_len']['end'] = self._io.pos() - self._debug['comment']['start'] = self._io.pos() - self.comment = (self._io.read_bytes(self.comment_len)).decode(u"UTF-8") - self._debug['comment']['end'] = self._io.pos() - - - |
