From 815f5485eb50eff3b483a6bcc0c3d3d11569f8ca Mon Sep 17 00:00:00 2001 From: Andrew Lamoureux Date: Mon, 18 Mar 2019 17:34:09 -0400 Subject: kaitai: all formats compiled and included --- .../kaitai_struct_formats/executable/__init__.py | 0 .../kaitai/kaitai_struct_formats/executable/dex.py | 1084 ++++++++++++++++++++ .../kaitai_struct_formats/executable/dos_mz.py | 123 +++ .../kaitai_struct_formats/executable/java_class.py | 952 +++++++++++++++++ .../executable/python_pyc_27.py | 508 +++++++++ .../kaitai/kaitai_struct_formats/executable/swf.py | 404 ++++++++ .../executable/vlq_base128_le.py | 108 ++ 7 files changed, 3179 insertions(+) create mode 100644 python/examples/kaitai/kaitai_struct_formats/executable/__init__.py create mode 100644 python/examples/kaitai/kaitai_struct_formats/executable/dex.py create mode 100644 python/examples/kaitai/kaitai_struct_formats/executable/dos_mz.py create mode 100644 python/examples/kaitai/kaitai_struct_formats/executable/java_class.py create mode 100644 python/examples/kaitai/kaitai_struct_formats/executable/python_pyc_27.py create mode 100644 python/examples/kaitai/kaitai_struct_formats/executable/swf.py create mode 100644 python/examples/kaitai/kaitai_struct_formats/executable/vlq_base128_le.py (limited to 'python/examples/kaitai/kaitai_struct_formats/executable') diff --git a/python/examples/kaitai/kaitai_struct_formats/executable/__init__.py b/python/examples/kaitai/kaitai_struct_formats/executable/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/python/examples/kaitai/kaitai_struct_formats/executable/dex.py b/python/examples/kaitai/kaitai_struct_formats/executable/dex.py new file mode 100644 index 00000000..8a3ddaa4 --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/executable/dex.py @@ -0,0 +1,1084 @@ +from __future__ import absolute_import +# 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)) + +from vlq_base128_le import VlqBase128Le +class Dex(KaitaiStruct): + """Android OS applications executables are typically stored in its own + format, optimized for more efficient execution in Dalvik virtual + machine. + + This format is loosely similar to Java .class file format and + generally holds the similar set of data: i.e. classes, methods, + fields, annotations, etc. + + .. seealso:: + Source - https://source.android.com/devices/tech/dalvik/dex-format + """ + + class ClassAccessFlags(Enum): + public = 1 + private = 2 + protected = 4 + static = 8 + final = 16 + interface = 512 + abstract = 1024 + synthetic = 4096 + annotation = 8192 + enum = 16384 + SEQ_FIELDS = ["header"] + 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.HeaderItem(self._io, self, self._root) + self.header._read() + self._debug['header']['end'] = self._io.pos() + + class HeaderItem(KaitaiStruct): + + class EndianConstant(Enum): + endian_constant = 305419896 + reverse_endian_constant = 2018915346 + SEQ_FIELDS = ["magic", "version_str", "checksum", "signature", "file_size", "header_size", "endian_tag", "link_size", "link_off", "map_off", "string_ids_size", "string_ids_off", "type_ids_size", "type_ids_off", "proto_ids_size", "proto_ids_off", "field_ids_size", "field_ids_off", "method_ids_size", "method_ids_off", "class_defs_size", "class_defs_off", "data_size", "data_off"] + 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"\x64\x65\x78\x0A") + self._debug['magic']['end'] = self._io.pos() + self._debug['version_str']['start'] = self._io.pos() + self.version_str = (KaitaiStream.bytes_terminate(self._io.read_bytes(4), 0, False)).decode(u"ascii") + self._debug['version_str']['end'] = self._io.pos() + self._debug['checksum']['start'] = self._io.pos() + self.checksum = self._io.read_u4le() + self._debug['checksum']['end'] = self._io.pos() + self._debug['signature']['start'] = self._io.pos() + self.signature = self._io.read_bytes(20) + self._debug['signature']['end'] = self._io.pos() + self._debug['file_size']['start'] = self._io.pos() + self.file_size = self._io.read_u4le() + self._debug['file_size']['end'] = self._io.pos() + self._debug['header_size']['start'] = self._io.pos() + self.header_size = self._io.read_u4le() + self._debug['header_size']['end'] = self._io.pos() + self._debug['endian_tag']['start'] = self._io.pos() + self.endian_tag = KaitaiStream.resolve_enum(self._root.HeaderItem.EndianConstant, self._io.read_u4le()) + self._debug['endian_tag']['end'] = self._io.pos() + self._debug['link_size']['start'] = self._io.pos() + self.link_size = self._io.read_u4le() + self._debug['link_size']['end'] = self._io.pos() + self._debug['link_off']['start'] = self._io.pos() + self.link_off = self._io.read_u4le() + self._debug['link_off']['end'] = self._io.pos() + self._debug['map_off']['start'] = self._io.pos() + self.map_off = self._io.read_u4le() + self._debug['map_off']['end'] = self._io.pos() + self._debug['string_ids_size']['start'] = self._io.pos() + self.string_ids_size = self._io.read_u4le() + self._debug['string_ids_size']['end'] = self._io.pos() + self._debug['string_ids_off']['start'] = self._io.pos() + self.string_ids_off = self._io.read_u4le() + self._debug['string_ids_off']['end'] = self._io.pos() + self._debug['type_ids_size']['start'] = self._io.pos() + self.type_ids_size = self._io.read_u4le() + self._debug['type_ids_size']['end'] = self._io.pos() + self._debug['type_ids_off']['start'] = self._io.pos() + self.type_ids_off = self._io.read_u4le() + self._debug['type_ids_off']['end'] = self._io.pos() + self._debug['proto_ids_size']['start'] = self._io.pos() + self.proto_ids_size = self._io.read_u4le() + self._debug['proto_ids_size']['end'] = self._io.pos() + self._debug['proto_ids_off']['start'] = self._io.pos() + self.proto_ids_off = self._io.read_u4le() + self._debug['proto_ids_off']['end'] = self._io.pos() + self._debug['field_ids_size']['start'] = self._io.pos() + self.field_ids_size = self._io.read_u4le() + self._debug['field_ids_size']['end'] = self._io.pos() + self._debug['field_ids_off']['start'] = self._io.pos() + self.field_ids_off = self._io.read_u4le() + self._debug['field_ids_off']['end'] = self._io.pos() + self._debug['method_ids_size']['start'] = self._io.pos() + self.method_ids_size = self._io.read_u4le() + self._debug['method_ids_size']['end'] = self._io.pos() + self._debug['method_ids_off']['start'] = self._io.pos() + self.method_ids_off = self._io.read_u4le() + self._debug['method_ids_off']['end'] = self._io.pos() + self._debug['class_defs_size']['start'] = self._io.pos() + self.class_defs_size = self._io.read_u4le() + self._debug['class_defs_size']['end'] = self._io.pos() + self._debug['class_defs_off']['start'] = self._io.pos() + self.class_defs_off = self._io.read_u4le() + self._debug['class_defs_off']['end'] = self._io.pos() + self._debug['data_size']['start'] = self._io.pos() + self.data_size = self._io.read_u4le() + self._debug['data_size']['end'] = self._io.pos() + self._debug['data_off']['start'] = self._io.pos() + self.data_off = self._io.read_u4le() + self._debug['data_off']['end'] = self._io.pos() + + + class MapList(KaitaiStruct): + SEQ_FIELDS = ["size", "list"] + 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['size']['start'] = self._io.pos() + self.size = self._io.read_u4le() + self._debug['size']['end'] = self._io.pos() + self._debug['list']['start'] = self._io.pos() + self.list = [None] * (self.size) + for i in range(self.size): + if not 'arr' in self._debug['list']: + self._debug['list']['arr'] = [] + self._debug['list']['arr'].append({'start': self._io.pos()}) + _t_list = self._root.MapItem(self._io, self, self._root) + _t_list._read() + self.list[i] = _t_list + self._debug['list']['arr'][i]['end'] = self._io.pos() + + self._debug['list']['end'] = self._io.pos() + + + class EncodedValue(KaitaiStruct): + + class ValueTypeEnum(Enum): + byte = 0 + short = 2 + char = 3 + int = 4 + long = 6 + float = 16 + double = 17 + method_type = 21 + method_handle = 22 + string = 23 + type = 24 + field = 25 + method = 26 + enum = 27 + array = 28 + annotation = 29 + null = 30 + boolean = 31 + SEQ_FIELDS = ["value_arg", "value_type", "value"] + 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['value_arg']['start'] = self._io.pos() + self.value_arg = self._io.read_bits_int(3) + self._debug['value_arg']['end'] = self._io.pos() + self._debug['value_type']['start'] = self._io.pos() + self.value_type = KaitaiStream.resolve_enum(self._root.EncodedValue.ValueTypeEnum, self._io.read_bits_int(5)) + self._debug['value_type']['end'] = self._io.pos() + self._io.align_to_byte() + self._debug['value']['start'] = self._io.pos() + _on = self.value_type + if _on == self._root.EncodedValue.ValueTypeEnum.int: + self.value = self._io.read_s4le() + elif _on == self._root.EncodedValue.ValueTypeEnum.annotation: + self.value = self._root.EncodedAnnotation(self._io, self, self._root) + self.value._read() + elif _on == self._root.EncodedValue.ValueTypeEnum.long: + self.value = self._io.read_s8le() + elif _on == self._root.EncodedValue.ValueTypeEnum.method_handle: + self.value = self._io.read_u4le() + elif _on == self._root.EncodedValue.ValueTypeEnum.byte: + self.value = self._io.read_s1() + elif _on == self._root.EncodedValue.ValueTypeEnum.array: + self.value = self._root.EncodedArray(self._io, self, self._root) + self.value._read() + elif _on == self._root.EncodedValue.ValueTypeEnum.method_type: + self.value = self._io.read_u4le() + elif _on == self._root.EncodedValue.ValueTypeEnum.short: + self.value = self._io.read_s2le() + elif _on == self._root.EncodedValue.ValueTypeEnum.method: + self.value = self._io.read_u4le() + elif _on == self._root.EncodedValue.ValueTypeEnum.double: + self.value = self._io.read_f8le() + elif _on == self._root.EncodedValue.ValueTypeEnum.float: + self.value = self._io.read_f4le() + elif _on == self._root.EncodedValue.ValueTypeEnum.type: + self.value = self._io.read_u4le() + elif _on == self._root.EncodedValue.ValueTypeEnum.enum: + self.value = self._io.read_u4le() + elif _on == self._root.EncodedValue.ValueTypeEnum.field: + self.value = self._io.read_u4le() + elif _on == self._root.EncodedValue.ValueTypeEnum.string: + self.value = self._io.read_u4le() + elif _on == self._root.EncodedValue.ValueTypeEnum.char: + self.value = self._io.read_u2le() + self._debug['value']['end'] = self._io.pos() + + + class CallSiteIdItem(KaitaiStruct): + SEQ_FIELDS = ["call_site_off"] + 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['call_site_off']['start'] = self._io.pos() + self.call_site_off = self._io.read_u4le() + self._debug['call_site_off']['end'] = self._io.pos() + + + class MethodIdItem(KaitaiStruct): + SEQ_FIELDS = ["class_idx", "proto_idx", "name_idx"] + 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['class_idx']['start'] = self._io.pos() + self.class_idx = self._io.read_u2le() + self._debug['class_idx']['end'] = self._io.pos() + self._debug['proto_idx']['start'] = self._io.pos() + self.proto_idx = self._io.read_u2le() + self._debug['proto_idx']['end'] = self._io.pos() + self._debug['name_idx']['start'] = self._io.pos() + self.name_idx = self._io.read_u4le() + self._debug['name_idx']['end'] = self._io.pos() + + @property + def class_name(self): + """the definer of this method.""" + if hasattr(self, '_m_class_name'): + return self._m_class_name if hasattr(self, '_m_class_name') else None + + self._m_class_name = self._root.type_ids[self.class_idx].type_name + return self._m_class_name if hasattr(self, '_m_class_name') else None + + @property + def proto_desc(self): + """the short-form descriptor of the prototype of this method.""" + if hasattr(self, '_m_proto_desc'): + return self._m_proto_desc if hasattr(self, '_m_proto_desc') else None + + self._m_proto_desc = self._root.proto_ids[self.proto_idx].shorty_desc + return self._m_proto_desc if hasattr(self, '_m_proto_desc') else None + + @property + def method_name(self): + """the name of this method.""" + if hasattr(self, '_m_method_name'): + return self._m_method_name if hasattr(self, '_m_method_name') else None + + self._m_method_name = self._root.string_ids[self.name_idx].value.data + return self._m_method_name if hasattr(self, '_m_method_name') else None + + + class TypeItem(KaitaiStruct): + SEQ_FIELDS = ["type_idx"] + 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['type_idx']['start'] = self._io.pos() + self.type_idx = self._io.read_u2le() + self._debug['type_idx']['end'] = self._io.pos() + + @property + def value(self): + if hasattr(self, '_m_value'): + return self._m_value if hasattr(self, '_m_value') else None + + self._m_value = self._root.type_ids[self.type_idx].type_name + return self._m_value if hasattr(self, '_m_value') else None + + + class TypeIdItem(KaitaiStruct): + SEQ_FIELDS = ["descriptor_idx"] + 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['descriptor_idx']['start'] = self._io.pos() + self.descriptor_idx = self._io.read_u4le() + self._debug['descriptor_idx']['end'] = self._io.pos() + + @property + def type_name(self): + if hasattr(self, '_m_type_name'): + return self._m_type_name if hasattr(self, '_m_type_name') else None + + self._m_type_name = self._root.string_ids[self.descriptor_idx].value.data + return self._m_type_name if hasattr(self, '_m_type_name') else None + + + class AnnotationElement(KaitaiStruct): + SEQ_FIELDS = ["name_idx", "value"] + 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['name_idx']['start'] = self._io.pos() + self.name_idx = VlqBase128Le(self._io) + self.name_idx._read() + self._debug['name_idx']['end'] = self._io.pos() + self._debug['value']['start'] = self._io.pos() + self.value = self._root.EncodedValue(self._io, self, self._root) + self.value._read() + self._debug['value']['end'] = self._io.pos() + + + class EncodedField(KaitaiStruct): + SEQ_FIELDS = ["field_idx_diff", "access_flags"] + 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['field_idx_diff']['start'] = self._io.pos() + self.field_idx_diff = VlqBase128Le(self._io) + self.field_idx_diff._read() + self._debug['field_idx_diff']['end'] = self._io.pos() + self._debug['access_flags']['start'] = self._io.pos() + self.access_flags = VlqBase128Le(self._io) + self.access_flags._read() + self._debug['access_flags']['end'] = self._io.pos() + + + class EncodedArrayItem(KaitaiStruct): + SEQ_FIELDS = ["value"] + 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['value']['start'] = self._io.pos() + self.value = self._root.EncodedArray(self._io, self, self._root) + self.value._read() + self._debug['value']['end'] = self._io.pos() + + + class ClassDataItem(KaitaiStruct): + SEQ_FIELDS = ["static_fields_size", "instance_fields_size", "direct_methods_size", "virtual_methods_size", "static_fields", "instance_fields", "direct_methods", "virtual_methods"] + 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['static_fields_size']['start'] = self._io.pos() + self.static_fields_size = VlqBase128Le(self._io) + self.static_fields_size._read() + self._debug['static_fields_size']['end'] = self._io.pos() + self._debug['instance_fields_size']['start'] = self._io.pos() + self.instance_fields_size = VlqBase128Le(self._io) + self.instance_fields_size._read() + self._debug['instance_fields_size']['end'] = self._io.pos() + self._debug['direct_methods_size']['start'] = self._io.pos() + self.direct_methods_size = VlqBase128Le(self._io) + self.direct_methods_size._read() + self._debug['direct_methods_size']['end'] = self._io.pos() + self._debug['virtual_methods_size']['start'] = self._io.pos() + self.virtual_methods_size = VlqBase128Le(self._io) + self.virtual_methods_size._read() + self._debug['virtual_methods_size']['end'] = self._io.pos() + self._debug['static_fields']['start'] = self._io.pos() + self.static_fields = [None] * (self.static_fields_size.value) + for i in range(self.static_fields_size.value): + if not 'arr' in self._debug['static_fields']: + self._debug['static_fields']['arr'] = [] + self._debug['static_fields']['arr'].append({'start': self._io.pos()}) + _t_static_fields = self._root.EncodedField(self._io, self, self._root) + _t_static_fields._read() + self.static_fields[i] = _t_static_fields + self._debug['static_fields']['arr'][i]['end'] = self._io.pos() + + self._debug['static_fields']['end'] = self._io.pos() + self._debug['instance_fields']['start'] = self._io.pos() + self.instance_fields = [None] * (self.instance_fields_size.value) + for i in range(self.instance_fields_size.value): + if not 'arr' in self._debug['instance_fields']: + self._debug['instance_fields']['arr'] = [] + self._debug['instance_fields']['arr'].append({'start': self._io.pos()}) + _t_instance_fields = self._root.EncodedField(self._io, self, self._root) + _t_instance_fields._read() + self.instance_fields[i] = _t_instance_fields + self._debug['instance_fields']['arr'][i]['end'] = self._io.pos() + + self._debug['instance_fields']['end'] = self._io.pos() + self._debug['direct_methods']['start'] = self._io.pos() + self.direct_methods = [None] * (self.direct_methods_size.value) + for i in range(self.direct_methods_size.value): + if not 'arr' in self._debug['direct_methods']: + self._debug['direct_methods']['arr'] = [] + self._debug['direct_methods']['arr'].append({'start': self._io.pos()}) + _t_direct_methods = self._root.EncodedMethod(self._io, self, self._root) + _t_direct_methods._read() + self.direct_methods[i] = _t_direct_methods + self._debug['direct_methods']['arr'][i]['end'] = self._io.pos() + + self._debug['direct_methods']['end'] = self._io.pos() + self._debug['virtual_methods']['start'] = self._io.pos() + self.virtual_methods = [None] * (self.virtual_methods_size.value) + for i in range(self.virtual_methods_size.value): + if not 'arr' in self._debug['virtual_methods']: + self._debug['virtual_methods']['arr'] = [] + self._debug['virtual_methods']['arr'].append({'start': self._io.pos()}) + _t_virtual_methods = self._root.EncodedMethod(self._io, self, self._root) + _t_virtual_methods._read() + self.virtual_methods[i] = _t_virtual_methods + self._debug['virtual_methods']['arr'][i]['end'] = self._io.pos() + + self._debug['virtual_methods']['end'] = self._io.pos() + + + class FieldIdItem(KaitaiStruct): + SEQ_FIELDS = ["class_idx", "type_idx", "name_idx"] + 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['class_idx']['start'] = self._io.pos() + self.class_idx = self._io.read_u2le() + self._debug['class_idx']['end'] = self._io.pos() + self._debug['type_idx']['start'] = self._io.pos() + self.type_idx = self._io.read_u2le() + self._debug['type_idx']['end'] = self._io.pos() + self._debug['name_idx']['start'] = self._io.pos() + self.name_idx = self._io.read_u4le() + self._debug['name_idx']['end'] = self._io.pos() + + @property + def class_name(self): + """the definer of this field.""" + if hasattr(self, '_m_class_name'): + return self._m_class_name if hasattr(self, '_m_class_name') else None + + self._m_class_name = self._root.type_ids[self.class_idx].type_name + return self._m_class_name if hasattr(self, '_m_class_name') else None + + @property + def type_name(self): + """the type of this field.""" + if hasattr(self, '_m_type_name'): + return self._m_type_name if hasattr(self, '_m_type_name') else None + + self._m_type_name = self._root.type_ids[self.type_idx].type_name + return self._m_type_name if hasattr(self, '_m_type_name') else None + + @property + def field_name(self): + """the name of this field.""" + if hasattr(self, '_m_field_name'): + return self._m_field_name if hasattr(self, '_m_field_name') else None + + self._m_field_name = self._root.string_ids[self.name_idx].value.data + return self._m_field_name if hasattr(self, '_m_field_name') else None + + + class EncodedAnnotation(KaitaiStruct): + SEQ_FIELDS = ["type_idx", "size", "elements"] + 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['type_idx']['start'] = self._io.pos() + self.type_idx = VlqBase128Le(self._io) + self.type_idx._read() + self._debug['type_idx']['end'] = self._io.pos() + self._debug['size']['start'] = self._io.pos() + self.size = VlqBase128Le(self._io) + self.size._read() + self._debug['size']['end'] = self._io.pos() + self._debug['elements']['start'] = self._io.pos() + self.elements = [None] * (self.size.value) + for i in range(self.size.value): + if not 'arr' in self._debug['elements']: + self._debug['elements']['arr'] = [] + self._debug['elements']['arr'].append({'start': self._io.pos()}) + _t_elements = self._root.AnnotationElement(self._io, self, self._root) + _t_elements._read() + self.elements[i] = _t_elements + self._debug['elements']['arr'][i]['end'] = self._io.pos() + + self._debug['elements']['end'] = self._io.pos() + + + class ClassDefItem(KaitaiStruct): + SEQ_FIELDS = ["class_idx", "access_flags", "superclass_idx", "interfaces_off", "source_file_idx", "annotations_off", "class_data_off", "static_values_off"] + 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['class_idx']['start'] = self._io.pos() + self.class_idx = self._io.read_u4le() + self._debug['class_idx']['end'] = self._io.pos() + self._debug['access_flags']['start'] = self._io.pos() + self.access_flags = KaitaiStream.resolve_enum(self._root.ClassAccessFlags, self._io.read_u4le()) + self._debug['access_flags']['end'] = self._io.pos() + self._debug['superclass_idx']['start'] = self._io.pos() + self.superclass_idx = self._io.read_u4le() + self._debug['superclass_idx']['end'] = self._io.pos() + self._debug['interfaces_off']['start'] = self._io.pos() + self.interfaces_off = self._io.read_u4le() + self._debug['interfaces_off']['end'] = self._io.pos() + self._debug['source_file_idx']['start'] = self._io.pos() + self.source_file_idx = self._io.read_u4le() + self._debug['source_file_idx']['end'] = self._io.pos() + self._debug['annotations_off']['start'] = self._io.pos() + self.annotations_off = self._io.read_u4le() + self._debug['annotations_off']['end'] = self._io.pos() + self._debug['class_data_off']['start'] = self._io.pos() + self.class_data_off = self._io.read_u4le() + self._debug['class_data_off']['end'] = self._io.pos() + self._debug['static_values_off']['start'] = self._io.pos() + self.static_values_off = self._io.read_u4le() + self._debug['static_values_off']['end'] = self._io.pos() + + @property + def type_name(self): + if hasattr(self, '_m_type_name'): + return self._m_type_name if hasattr(self, '_m_type_name') else None + + self._m_type_name = self._root.type_ids[self.class_idx].type_name + return self._m_type_name if hasattr(self, '_m_type_name') else None + + @property + def class_data(self): + if hasattr(self, '_m_class_data'): + return self._m_class_data if hasattr(self, '_m_class_data') else None + + if self.class_data_off != 0: + _pos = self._io.pos() + self._io.seek(self.class_data_off) + self._debug['_m_class_data']['start'] = self._io.pos() + self._m_class_data = self._root.ClassDataItem(self._io, self, self._root) + self._m_class_data._read() + self._debug['_m_class_data']['end'] = self._io.pos() + self._io.seek(_pos) + + return self._m_class_data if hasattr(self, '_m_class_data') else None + + @property + def static_values(self): + if hasattr(self, '_m_static_values'): + return self._m_static_values if hasattr(self, '_m_static_values') else None + + if self.static_values_off != 0: + _pos = self._io.pos() + self._io.seek(self.static_values_off) + self._debug['_m_static_values']['start'] = self._io.pos() + self._m_static_values = self._root.EncodedArrayItem(self._io, self, self._root) + self._m_static_values._read() + self._debug['_m_static_values']['end'] = self._io.pos() + self._io.seek(_pos) + + return self._m_static_values if hasattr(self, '_m_static_values') else None + + + class TypeList(KaitaiStruct): + SEQ_FIELDS = ["size", "list"] + 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['size']['start'] = self._io.pos() + self.size = self._io.read_u4le() + self._debug['size']['end'] = self._io.pos() + self._debug['list']['start'] = self._io.pos() + self.list = [None] * (self.size) + for i in range(self.size): + if not 'arr' in self._debug['list']: + self._debug['list']['arr'] = [] + self._debug['list']['arr'].append({'start': self._io.pos()}) + _t_list = self._root.TypeItem(self._io, self, self._root) + _t_list._read() + self.list[i] = _t_list + self._debug['list']['arr'][i]['end'] = self._io.pos() + + self._debug['list']['end'] = self._io.pos() + + + class StringIdItem(KaitaiStruct): + SEQ_FIELDS = ["string_data_off"] + 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['string_data_off']['start'] = self._io.pos() + self.string_data_off = self._io.read_u4le() + self._debug['string_data_off']['end'] = self._io.pos() + + class StringDataItem(KaitaiStruct): + SEQ_FIELDS = ["utf16_size", "data"] + 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['utf16_size']['start'] = self._io.pos() + self.utf16_size = VlqBase128Le(self._io) + self.utf16_size._read() + self._debug['utf16_size']['end'] = self._io.pos() + self._debug['data']['start'] = self._io.pos() + self.data = (self._io.read_bytes(self.utf16_size.value)).decode(u"ascii") + self._debug['data']['end'] = self._io.pos() + + + @property + def value(self): + if hasattr(self, '_m_value'): + return self._m_value if hasattr(self, '_m_value') else None + + _pos = self._io.pos() + self._io.seek(self.string_data_off) + self._debug['_m_value']['start'] = self._io.pos() + self._m_value = self._root.StringIdItem.StringDataItem(self._io, self, self._root) + self._m_value._read() + self._debug['_m_value']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_value if hasattr(self, '_m_value') else None + + + class ProtoIdItem(KaitaiStruct): + SEQ_FIELDS = ["shorty_idx", "return_type_idx", "parameters_off"] + 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['shorty_idx']['start'] = self._io.pos() + self.shorty_idx = self._io.read_u4le() + self._debug['shorty_idx']['end'] = self._io.pos() + self._debug['return_type_idx']['start'] = self._io.pos() + self.return_type_idx = self._io.read_u4le() + self._debug['return_type_idx']['end'] = self._io.pos() + self._debug['parameters_off']['start'] = self._io.pos() + self.parameters_off = self._io.read_u4le() + self._debug['parameters_off']['end'] = self._io.pos() + + @property + def shorty_desc(self): + """short-form descriptor string of this prototype, as pointed to by shorty_idx.""" + if hasattr(self, '_m_shorty_desc'): + return self._m_shorty_desc if hasattr(self, '_m_shorty_desc') else None + + self._m_shorty_desc = self._root.string_ids[self.shorty_idx].value.data + return self._m_shorty_desc if hasattr(self, '_m_shorty_desc') else None + + @property + def params_types(self): + """list of parameter types for this prototype.""" + if hasattr(self, '_m_params_types'): + return self._m_params_types if hasattr(self, '_m_params_types') else None + + if self.parameters_off != 0: + io = self._root._io + _pos = io.pos() + io.seek(self.parameters_off) + self._debug['_m_params_types']['start'] = io.pos() + self._m_params_types = self._root.TypeList(io, self, self._root) + self._m_params_types._read() + self._debug['_m_params_types']['end'] = io.pos() + io.seek(_pos) + + return self._m_params_types if hasattr(self, '_m_params_types') else None + + @property + def return_type(self): + """return type of this prototype.""" + if hasattr(self, '_m_return_type'): + return self._m_return_type if hasattr(self, '_m_return_type') else None + + self._m_return_type = self._root.type_ids[self.return_type_idx].type_name + return self._m_return_type if hasattr(self, '_m_return_type') else None + + + class EncodedMethod(KaitaiStruct): + SEQ_FIELDS = ["method_idx_diff", "access_flags", "code_off"] + 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['method_idx_diff']['start'] = self._io.pos() + self.method_idx_diff = VlqBase128Le(self._io) + self.method_idx_diff._read() + self._debug['method_idx_diff']['end'] = self._io.pos() + self._debug['access_flags']['start'] = self._io.pos() + self.access_flags = VlqBase128Le(self._io) + self.access_flags._read() + self._debug['access_flags']['end'] = self._io.pos() + self._debug['code_off']['start'] = self._io.pos() + self.code_off = VlqBase128Le(self._io) + self.code_off._read() + self._debug['code_off']['end'] = self._io.pos() + + + class MapItem(KaitaiStruct): + + class MapItemType(Enum): + header_item = 0 + string_id_item = 1 + type_id_item = 2 + proto_id_item = 3 + field_id_item = 4 + method_id_item = 5 + class_def_item = 6 + call_site_id_item = 7 + method_handle_item = 8 + map_list = 4096 + type_list = 4097 + annotation_set_ref_list = 4098 + annotation_set_item = 4099 + class_data_item = 8192 + code_item = 8193 + string_data_item = 8194 + debug_info_item = 8195 + annotation_item = 8196 + encoded_array_item = 8197 + annotations_directory_item = 8198 + SEQ_FIELDS = ["type", "unused", "size", "offset"] + 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['type']['start'] = self._io.pos() + self.type = KaitaiStream.resolve_enum(self._root.MapItem.MapItemType, self._io.read_u2le()) + self._debug['type']['end'] = self._io.pos() + self._debug['unused']['start'] = self._io.pos() + self.unused = self._io.read_u2le() + self._debug['unused']['end'] = self._io.pos() + self._debug['size']['start'] = self._io.pos() + self.size = self._io.read_u4le() + self._debug['size']['end'] = self._io.pos() + self._debug['offset']['start'] = self._io.pos() + self.offset = self._io.read_u4le() + self._debug['offset']['end'] = self._io.pos() + + + class EncodedArray(KaitaiStruct): + SEQ_FIELDS = ["size", "values"] + 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['size']['start'] = self._io.pos() + self.size = VlqBase128Le(self._io) + self.size._read() + self._debug['size']['end'] = self._io.pos() + self._debug['values']['start'] = self._io.pos() + self.values = [None] * (self.size.value) + for i in range(self.size.value): + if not 'arr' in self._debug['values']: + self._debug['values']['arr'] = [] + self._debug['values']['arr'].append({'start': self._io.pos()}) + _t_values = self._root.EncodedValue(self._io, self, self._root) + _t_values._read() + self.values[i] = _t_values + self._debug['values']['arr'][i]['end'] = self._io.pos() + + self._debug['values']['end'] = self._io.pos() + + + @property + def string_ids(self): + """string identifiers list. + + These are identifiers for all the strings used by this file, either for + internal naming (e.g., type descriptors) or as constant objects referred to by code. + + This list must be sorted by string contents, using UTF-16 code point values + (not in a locale-sensitive manner), and it must not contain any duplicate entries. + """ + if hasattr(self, '_m_string_ids'): + return self._m_string_ids if hasattr(self, '_m_string_ids') else None + + _pos = self._io.pos() + self._io.seek(self.header.string_ids_off) + self._debug['_m_string_ids']['start'] = self._io.pos() + self._m_string_ids = [None] * (self.header.string_ids_size) + for i in range(self.header.string_ids_size): + if not 'arr' in self._debug['_m_string_ids']: + self._debug['_m_string_ids']['arr'] = [] + self._debug['_m_string_ids']['arr'].append({'start': self._io.pos()}) + _t__m_string_ids = self._root.StringIdItem(self._io, self, self._root) + _t__m_string_ids._read() + self._m_string_ids[i] = _t__m_string_ids + self._debug['_m_string_ids']['arr'][i]['end'] = self._io.pos() + + self._debug['_m_string_ids']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_string_ids if hasattr(self, '_m_string_ids') else None + + @property + def method_ids(self): + """method identifiers list. + + These are identifiers for all methods referred to by this file, + whether defined in the file or not. + + This list must be sorted, where the defining type (by type_id index + is the major order, method name (by string_id index) is the intermediate + order, and method prototype (by proto_id index) is the minor order. + + The list must not contain any duplicate entries. + """ + if hasattr(self, '_m_method_ids'): + return self._m_method_ids if hasattr(self, '_m_method_ids') else None + + _pos = self._io.pos() + self._io.seek(self.header.method_ids_off) + self._debug['_m_method_ids']['start'] = self._io.pos() + self._m_method_ids = [None] * (self.header.method_ids_size) + for i in range(self.header.method_ids_size): + if not 'arr' in self._debug['_m_method_ids']: + self._debug['_m_method_ids']['arr'] = [] + self._debug['_m_method_ids']['arr'].append({'start': self._io.pos()}) + _t__m_method_ids = self._root.MethodIdItem(self._io, self, self._root) + _t__m_method_ids._read() + self._m_method_ids[i] = _t__m_method_ids + self._debug['_m_method_ids']['arr'][i]['end'] = self._io.pos() + + self._debug['_m_method_ids']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_method_ids if hasattr(self, '_m_method_ids') else None + + @property + def link_data(self): + """data used in statically linked files. + + The format of the data in this section is left unspecified by this document. + + This section is empty in unlinked files, and runtime implementations may + use it as they see fit. + """ + if hasattr(self, '_m_link_data'): + return self._m_link_data if hasattr(self, '_m_link_data') else None + + _pos = self._io.pos() + self._io.seek(self.header.link_off) + self._debug['_m_link_data']['start'] = self._io.pos() + self._m_link_data = self._io.read_bytes(self.header.link_size) + self._debug['_m_link_data']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_link_data if hasattr(self, '_m_link_data') else None + + @property + def map(self): + if hasattr(self, '_m_map'): + return self._m_map if hasattr(self, '_m_map') else None + + _pos = self._io.pos() + self._io.seek(self.header.map_off) + self._debug['_m_map']['start'] = self._io.pos() + self._m_map = self._root.MapList(self._io, self, self._root) + self._m_map._read() + self._debug['_m_map']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_map if hasattr(self, '_m_map') else None + + @property + def class_defs(self): + """class definitions list. + + The classes must be ordered such that a given class's superclass and + implemented interfaces appear in the list earlier than the referring class. + + Furthermore, it is invalid for a definition for the same-named class to + appear more than once in the list. + """ + if hasattr(self, '_m_class_defs'): + return self._m_class_defs if hasattr(self, '_m_class_defs') else None + + _pos = self._io.pos() + self._io.seek(self.header.class_defs_off) + self._debug['_m_class_defs']['start'] = self._io.pos() + self._m_class_defs = [None] * (self.header.class_defs_size) + for i in range(self.header.class_defs_size): + if not 'arr' in self._debug['_m_class_defs']: + self._debug['_m_class_defs']['arr'] = [] + self._debug['_m_class_defs']['arr'].append({'start': self._io.pos()}) + _t__m_class_defs = self._root.ClassDefItem(self._io, self, self._root) + _t__m_class_defs._read() + self._m_class_defs[i] = _t__m_class_defs + self._debug['_m_class_defs']['arr'][i]['end'] = self._io.pos() + + self._debug['_m_class_defs']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_class_defs if hasattr(self, '_m_class_defs') else None + + @property + def data(self): + """data area, containing all the support data for the tables listed above. + + Different items have different alignment requirements, and padding bytes + are inserted before each item if necessary to achieve proper alignment. + """ + if hasattr(self, '_m_data'): + return self._m_data if hasattr(self, '_m_data') else None + + _pos = self._io.pos() + self._io.seek(self.header.data_off) + self._debug['_m_data']['start'] = self._io.pos() + self._m_data = self._io.read_bytes(self.header.data_size) + self._debug['_m_data']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_data if hasattr(self, '_m_data') else None + + @property + def type_ids(self): + """type identifiers list. + + These are identifiers for all types (classes, arrays, or primitive types) + referred to by this file, whether defined in the file or not. + + This list must be sorted by string_id index, and it must not contain any duplicate entries. + """ + if hasattr(self, '_m_type_ids'): + return self._m_type_ids if hasattr(self, '_m_type_ids') else None + + _pos = self._io.pos() + self._io.seek(self.header.type_ids_off) + self._debug['_m_type_ids']['start'] = self._io.pos() + self._m_type_ids = [None] * (self.header.type_ids_size) + for i in range(self.header.type_ids_size): + if not 'arr' in self._debug['_m_type_ids']: + self._debug['_m_type_ids']['arr'] = [] + self._debug['_m_type_ids']['arr'].append({'start': self._io.pos()}) + _t__m_type_ids = self._root.TypeIdItem(self._io, self, self._root) + _t__m_type_ids._read() + self._m_type_ids[i] = _t__m_type_ids + self._debug['_m_type_ids']['arr'][i]['end'] = self._io.pos() + + self._debug['_m_type_ids']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_type_ids if hasattr(self, '_m_type_ids') else None + + @property + def proto_ids(self): + """method prototype identifiers list. + + These are identifiers for all prototypes referred to by this file. + + This list must be sorted in return-type (by type_id index) major order, + and then by argument list (lexicographic ordering, individual arguments + ordered by type_id index). The list must not contain any duplicate entries. + """ + if hasattr(self, '_m_proto_ids'): + return self._m_proto_ids if hasattr(self, '_m_proto_ids') else None + + _pos = self._io.pos() + self._io.seek(self.header.proto_ids_off) + self._debug['_m_proto_ids']['start'] = self._io.pos() + self._m_proto_ids = [None] * (self.header.proto_ids_size) + for i in range(self.header.proto_ids_size): + if not 'arr' in self._debug['_m_proto_ids']: + self._debug['_m_proto_ids']['arr'] = [] + self._debug['_m_proto_ids']['arr'].append({'start': self._io.pos()}) + _t__m_proto_ids = self._root.ProtoIdItem(self._io, self, self._root) + _t__m_proto_ids._read() + self._m_proto_ids[i] = _t__m_proto_ids + self._debug['_m_proto_ids']['arr'][i]['end'] = self._io.pos() + + self._debug['_m_proto_ids']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_proto_ids if hasattr(self, '_m_proto_ids') else None + + @property + def field_ids(self): + """field identifiers list. + + These are identifiers for all fields referred to by this file, whether defined in the file or not. + + This list must be sorted, where the defining type (by type_id index) + is the major order, field name (by string_id index) is the intermediate + order, and type (by type_id index) is the minor order. + + The list must not contain any duplicate entries. + """ + if hasattr(self, '_m_field_ids'): + return self._m_field_ids if hasattr(self, '_m_field_ids') else None + + _pos = self._io.pos() + self._io.seek(self.header.field_ids_off) + self._debug['_m_field_ids']['start'] = self._io.pos() + self._m_field_ids = [None] * (self.header.field_ids_size) + for i in range(self.header.field_ids_size): + if not 'arr' in self._debug['_m_field_ids']: + self._debug['_m_field_ids']['arr'] = [] + self._debug['_m_field_ids']['arr'].append({'start': self._io.pos()}) + _t__m_field_ids = self._root.FieldIdItem(self._io, self, self._root) + _t__m_field_ids._read() + self._m_field_ids[i] = _t__m_field_ids + self._debug['_m_field_ids']['arr'][i]['end'] = self._io.pos() + + self._debug['_m_field_ids']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_field_ids if hasattr(self, '_m_field_ids') else None + + diff --git a/python/examples/kaitai/kaitai_struct_formats/executable/dos_mz.py b/python/examples/kaitai/kaitai_struct_formats/executable/dos_mz.py new file mode 100644 index 00000000..c718c41e --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/executable/dos_mz.py @@ -0,0 +1,123 @@ +from __future__ import absolute_import +# 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 +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 DosMz(KaitaiStruct): + """DOS MZ file format is a traditional format for executables in MS-DOS + environment. Many modern formats (i.e. Windows PE) still maintain + compatibility stub with this format. + + As opposed to .com file format (which basically sports one 64K code + segment of raw CPU instructions), DOS MZ .exe file format allowed + more flexible memory management, loading of larger programs and + added support for relocations. + """ + SEQ_FIELDS = ["hdr", "mz_header2", "relocations", "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['hdr']['start'] = self._io.pos() + self.hdr = self._root.MzHeader(self._io, self, self._root) + self.hdr._read() + self._debug['hdr']['end'] = self._io.pos() + self._debug['mz_header2']['start'] = self._io.pos() + self.mz_header2 = self._io.read_bytes((self.hdr.ofs_relocations - 28)) + self._debug['mz_header2']['end'] = self._io.pos() + self._debug['relocations']['start'] = self._io.pos() + self.relocations = [None] * (self.hdr.num_relocations) + for i in range(self.hdr.num_relocations): + if not 'arr' in self._debug['relocations']: + self._debug['relocations']['arr'] = [] + self._debug['relocations']['arr'].append({'start': self._io.pos()}) + _t_relocations = self._root.Relocation(self._io, self, self._root) + _t_relocations._read() + self.relocations[i] = _t_relocations + self._debug['relocations']['arr'][i]['end'] = self._io.pos() + + self._debug['relocations']['end'] = self._io.pos() + self._debug['body']['start'] = self._io.pos() + self.body = self._io.read_bytes_full() + self._debug['body']['end'] = self._io.pos() + + class MzHeader(KaitaiStruct): + SEQ_FIELDS = ["magic", "last_page_extra_bytes", "num_pages", "num_relocations", "header_size", "min_allocation", "max_allocation", "initial_ss", "initial_sp", "checksum", "initial_ip", "initial_cs", "ofs_relocations", "overlay_id"] + 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.read_bytes(2) + self._debug['magic']['end'] = self._io.pos() + self._debug['last_page_extra_bytes']['start'] = self._io.pos() + self.last_page_extra_bytes = self._io.read_u2le() + self._debug['last_page_extra_bytes']['end'] = self._io.pos() + self._debug['num_pages']['start'] = self._io.pos() + self.num_pages = self._io.read_u2le() + self._debug['num_pages']['end'] = self._io.pos() + self._debug['num_relocations']['start'] = self._io.pos() + self.num_relocations = self._io.read_u2le() + self._debug['num_relocations']['end'] = self._io.pos() + self._debug['header_size']['start'] = self._io.pos() + self.header_size = self._io.read_u2le() + self._debug['header_size']['end'] = self._io.pos() + self._debug['min_allocation']['start'] = self._io.pos() + self.min_allocation = self._io.read_u2le() + self._debug['min_allocation']['end'] = self._io.pos() + self._debug['max_allocation']['start'] = self._io.pos() + self.max_allocation = self._io.read_u2le() + self._debug['max_allocation']['end'] = self._io.pos() + self._debug['initial_ss']['start'] = self._io.pos() + self.initial_ss = self._io.read_u2le() + self._debug['initial_ss']['end'] = self._io.pos() + self._debug['initial_sp']['start'] = self._io.pos() + self.initial_sp = self._io.read_u2le() + self._debug['initial_sp']['end'] = self._io.pos() + self._debug['checksum']['start'] = self._io.pos() + self.checksum = self._io.read_u2le() + self._debug['checksum']['end'] = self._io.pos() + self._debug['initial_ip']['start'] = self._io.pos() + self.initial_ip = self._io.read_u2le() + self._debug['initial_ip']['end'] = self._io.pos() + self._debug['initial_cs']['start'] = self._io.pos() + self.initial_cs = self._io.read_u2le() + self._debug['initial_cs']['end'] = self._io.pos() + self._debug['ofs_relocations']['start'] = self._io.pos() + self.ofs_relocations = self._io.read_u2le() + self._debug['ofs_relocations']['end'] = self._io.pos() + self._debug['overlay_id']['start'] = self._io.pos() + self.overlay_id = self._io.read_u2le() + self._debug['overlay_id']['end'] = self._io.pos() + + + class Relocation(KaitaiStruct): + SEQ_FIELDS = ["ofs", "seg"] + 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['ofs']['start'] = self._io.pos() + self.ofs = self._io.read_u2le() + self._debug['ofs']['end'] = self._io.pos() + self._debug['seg']['start'] = self._io.pos() + self.seg = self._io.read_u2le() + self._debug['seg']['end'] = self._io.pos() + + + diff --git a/python/examples/kaitai/kaitai_struct_formats/executable/java_class.py b/python/examples/kaitai/kaitai_struct_formats/executable/java_class.py new file mode 100644 index 00000000..fd92ab69 --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/executable/java_class.py @@ -0,0 +1,952 @@ +from __future__ import absolute_import +# 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 +import collections +from enum import Enum + + +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 JavaClass(KaitaiStruct): + """ + .. seealso:: + Source - https://docs.oracle.com/javase/specs/jvms/se8/html/jvms-4.html#jvms-4.1 + """ + SEQ_FIELDS = ["magic", "version_minor", "version_major", "constant_pool_count", "constant_pool", "access_flags", "this_class", "super_class", "interfaces_count", "interfaces", "fields_count", "fields", "methods_count", "methods", "attributes_count", "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['magic']['start'] = self._io.pos() + self.magic = self._io.ensure_fixed_contents(b"\xCA\xFE\xBA\xBE") + self._debug['magic']['end'] = self._io.pos() + self._debug['version_minor']['start'] = self._io.pos() + self.version_minor = self._io.read_u2be() + self._debug['version_minor']['end'] = self._io.pos() + self._debug['version_major']['start'] = self._io.pos() + self.version_major = self._io.read_u2be() + self._debug['version_major']['end'] = self._io.pos() + self._debug['constant_pool_count']['start'] = self._io.pos() + self.constant_pool_count = self._io.read_u2be() + self._debug['constant_pool_count']['end'] = self._io.pos() + self._debug['constant_pool']['start'] = self._io.pos() + self.constant_pool = [None] * ((self.constant_pool_count - 1)) + for i in range((self.constant_pool_count - 1)): + if not 'arr' in self._debug['constant_pool']: + self._debug['constant_pool']['arr'] = [] + self._debug['constant_pool']['arr'].append({'start': self._io.pos()}) + _t_constant_pool = self._root.ConstantPoolEntry(self._io, self, self._root) + _t_constant_pool._read() + self.constant_pool[i] = _t_constant_pool + self._debug['constant_pool']['arr'][i]['end'] = self._io.pos() + + self._debug['constant_pool']['end'] = self._io.pos() + self._debug['access_flags']['start'] = self._io.pos() + self.access_flags = self._io.read_u2be() + self._debug['access_flags']['end'] = self._io.pos() + self._debug['this_class']['start'] = self._io.pos() + self.this_class = self._io.read_u2be() + self._debug['this_class']['end'] = self._io.pos() + self._debug['super_class']['start'] = self._io.pos() + self.super_class = self._io.read_u2be() + self._debug['super_class']['end'] = self._io.pos() + self._debug['interfaces_count']['start'] = self._io.pos() + self.interfaces_count = self._io.read_u2be() + self._debug['interfaces_count']['end'] = self._io.pos() + self._debug['interfaces']['start'] = self._io.pos() + self.interfaces = [None] * (self.interfaces_count) + for i in range(self.interfaces_count): + if not 'arr' in self._debug['interfaces']: + self._debug['interfaces']['arr'] = [] + self._debug['interfaces']['arr'].append({'start': self._io.pos()}) + self.interfaces[i] = self._io.read_u2be() + self._debug['interfaces']['arr'][i]['end'] = self._io.pos() + + self._debug['interfaces']['end'] = self._io.pos() + self._debug['fields_count']['start'] = self._io.pos() + self.fields_count = self._io.read_u2be() + self._debug['fields_count']['end'] = self._io.pos() + self._debug['fields']['start'] = self._io.pos() + self.fields = [None] * (self.fields_count) + for i in range(self.fields_count): + if not 'arr' in self._debug['fields']: + self._debug['fields']['arr'] = [] + self._debug['fields']['arr'].append({'start': self._io.pos()}) + _t_fields = self._root.FieldInfo(self._io, self, self._root) + _t_fields._read() + self.fields[i] = _t_fields + self._debug['fields']['arr'][i]['end'] = self._io.pos() + + self._debug['fields']['end'] = self._io.pos() + self._debug['methods_count']['start'] = self._io.pos() + self.methods_count = self._io.read_u2be() + self._debug['methods_count']['end'] = self._io.pos() + self._debug['methods']['start'] = self._io.pos() + self.methods = [None] * (self.methods_count) + for i in range(self.methods_count): + if not 'arr' in self._debug['methods']: + self._debug['methods']['arr'] = [] + self._debug['methods']['arr'].append({'start': self._io.pos()}) + _t_methods = self._root.MethodInfo(self._io, self, self._root) + _t_methods._read() + self.methods[i] = _t_methods + self._debug['methods']['arr'][i]['end'] = self._io.pos() + + self._debug['methods']['end'] = self._io.pos() + self._debug['attributes_count']['start'] = self._io.pos() + self.attributes_count = self._io.read_u2be() + self._debug['attributes_count']['end'] = self._io.pos() + self._debug['attributes']['start'] = self._io.pos() + self.attributes = [None] * (self.attributes_count) + for i in range(self.attributes_count): + if not 'arr' in self._debug['attributes']: + self._debug['attributes']['arr'] = [] + self._debug['attributes']['arr'].append({'start': self._io.pos()}) + _t_attributes = self._root.AttributeInfo(self._io, self, self._root) + _t_attributes._read() + self.attributes[i] = _t_attributes + self._debug['attributes']['arr'][i]['end'] = self._io.pos() + + self._debug['attributes']['end'] = self._io.pos() + + class FloatCpInfo(KaitaiStruct): + """ + .. seealso:: + Source - https://docs.oracle.com/javase/specs/jvms/se8/html/jvms-4.html#jvms-4.4.5 + """ + SEQ_FIELDS = ["value"] + 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['value']['start'] = self._io.pos() + self.value = self._io.read_f4be() + self._debug['value']['end'] = self._io.pos() + + + class AttributeInfo(KaitaiStruct): + """ + .. seealso:: + Source - https://docs.oracle.com/javase/specs/jvms/se8/html/jvms-4.html#jvms-4.7 + """ + SEQ_FIELDS = ["name_index", "attribute_length", "info"] + 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['name_index']['start'] = self._io.pos() + self.name_index = self._io.read_u2be() + self._debug['name_index']['end'] = self._io.pos() + self._debug['attribute_length']['start'] = self._io.pos() + self.attribute_length = self._io.read_u4be() + self._debug['attribute_length']['end'] = self._io.pos() + self._debug['info']['start'] = self._io.pos() + _on = self.name_as_str + if _on == u"SourceFile": + self._raw_info = self._io.read_bytes(self.attribute_length) + io = KaitaiStream(BytesIO(self._raw_info)) + self.info = self._root.AttributeInfo.AttrBodySourceFile(io, self, self._root) + self.info._read() + elif _on == u"LineNumberTable": + self._raw_info = self._io.read_bytes(self.attribute_length) + io = KaitaiStream(BytesIO(self._raw_info)) + self.info = self._root.AttributeInfo.AttrBodyLineNumberTable(io, self, self._root) + self.info._read() + elif _on == u"Exceptions": + self._raw_info = self._io.read_bytes(self.attribute_length) + io = KaitaiStream(BytesIO(self._raw_info)) + self.info = self._root.AttributeInfo.AttrBodyExceptions(io, self, self._root) + self.info._read() + elif _on == u"Code": + self._raw_info = self._io.read_bytes(self.attribute_length) + io = KaitaiStream(BytesIO(self._raw_info)) + self.info = self._root.AttributeInfo.AttrBodyCode(io, self, self._root) + self.info._read() + else: + self.info = self._io.read_bytes(self.attribute_length) + self._debug['info']['end'] = self._io.pos() + + class AttrBodyCode(KaitaiStruct): + """ + .. seealso:: + Source - https://docs.oracle.com/javase/specs/jvms/se8/html/jvms-4.html#jvms-4.7.3 + """ + SEQ_FIELDS = ["max_stack", "max_locals", "code_length", "code", "exception_table_length", "exception_table", "attributes_count", "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['max_stack']['start'] = self._io.pos() + self.max_stack = self._io.read_u2be() + self._debug['max_stack']['end'] = self._io.pos() + self._debug['max_locals']['start'] = self._io.pos() + self.max_locals = self._io.read_u2be() + self._debug['max_locals']['end'] = self._io.pos() + self._debug['code_length']['start'] = self._io.pos() + self.code_length = self._io.read_u4be() + self._debug['code_length']['end'] = self._io.pos() + self._debug['code']['start'] = self._io.pos() + self.code = self._io.read_bytes(self.code_length) + self._debug['code']['end'] = self._io.pos() + self._debug['exception_table_length']['start'] = self._io.pos() + self.exception_table_length = self._io.read_u2be() + self._debug['exception_table_length']['end'] = self._io.pos() + self._debug['exception_table']['start'] = self._io.pos() + self.exception_table = [None] * (self.exception_table_length) + for i in range(self.exception_table_length): + if not 'arr' in self._debug['exception_table']: + self._debug['exception_table']['arr'] = [] + self._debug['exception_table']['arr'].append({'start': self._io.pos()}) + _t_exception_table = self._root.AttributeInfo.AttrBodyCode.ExceptionEntry(self._io, self, self._root) + _t_exception_table._read() + self.exception_table[i] = _t_exception_table + self._debug['exception_table']['arr'][i]['end'] = self._io.pos() + + self._debug['exception_table']['end'] = self._io.pos() + self._debug['attributes_count']['start'] = self._io.pos() + self.attributes_count = self._io.read_u2be() + self._debug['attributes_count']['end'] = self._io.pos() + self._debug['attributes']['start'] = self._io.pos() + self.attributes = [None] * (self.attributes_count) + for i in range(self.attributes_count): + if not 'arr' in self._debug['attributes']: + self._debug['attributes']['arr'] = [] + self._debug['attributes']['arr'].append({'start': self._io.pos()}) + _t_attributes = self._root.AttributeInfo(self._io, self, self._root) + _t_attributes._read() + self.attributes[i] = _t_attributes + self._debug['attributes']['arr'][i]['end'] = self._io.pos() + + self._debug['attributes']['end'] = self._io.pos() + + class ExceptionEntry(KaitaiStruct): + """ + .. seealso:: + Source - https://docs.oracle.com/javase/specs/jvms/se8/html/jvms-4.html#jvms-4.7.3 + """ + SEQ_FIELDS = ["start_pc", "end_pc", "handler_pc", "catch_type"] + 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['start_pc']['start'] = self._io.pos() + self.start_pc = self._io.read_u2be() + self._debug['start_pc']['end'] = self._io.pos() + self._debug['end_pc']['start'] = self._io.pos() + self.end_pc = self._io.read_u2be() + self._debug['end_pc']['end'] = self._io.pos() + self._debug['handler_pc']['start'] = self._io.pos() + self.handler_pc = self._io.read_u2be() + self._debug['handler_pc']['end'] = self._io.pos() + self._debug['catch_type']['start'] = self._io.pos() + self.catch_type = self._io.read_u2be() + self._debug['catch_type']['end'] = self._io.pos() + + @property + def catch_exception(self): + if hasattr(self, '_m_catch_exception'): + return self._m_catch_exception if hasattr(self, '_m_catch_exception') else None + + if self.catch_type != 0: + self._m_catch_exception = self._root.constant_pool[(self.catch_type - 1)] + + return self._m_catch_exception if hasattr(self, '_m_catch_exception') else None + + + + class AttrBodyExceptions(KaitaiStruct): + """ + .. seealso:: + Source - https://docs.oracle.com/javase/specs/jvms/se8/html/jvms-4.html#jvms-4.7.5 + """ + SEQ_FIELDS = ["number_of_exceptions", "exceptions"] + 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['number_of_exceptions']['start'] = self._io.pos() + self.number_of_exceptions = self._io.read_u2be() + self._debug['number_of_exceptions']['end'] = self._io.pos() + self._debug['exceptions']['start'] = self._io.pos() + self.exceptions = [None] * (self.number_of_exceptions) + for i in range(self.number_of_exceptions): + if not 'arr' in self._debug['exceptions']: + self._debug['exceptions']['arr'] = [] + self._debug['exceptions']['arr'].append({'start': self._io.pos()}) + _t_exceptions = self._root.AttributeInfo.AttrBodyExceptions.ExceptionTableEntry(self._io, self, self._root) + _t_exceptions._read() + self.exceptions[i] = _t_exceptions + self._debug['exceptions']['arr'][i]['end'] = self._io.pos() + + self._debug['exceptions']['end'] = self._io.pos() + + class ExceptionTableEntry(KaitaiStruct): + SEQ_FIELDS = ["index"] + 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['index']['start'] = self._io.pos() + self.index = self._io.read_u2be() + self._debug['index']['end'] = self._io.pos() + + @property + def as_info(self): + if hasattr(self, '_m_as_info'): + return self._m_as_info if hasattr(self, '_m_as_info') else None + + self._m_as_info = self._root.constant_pool[(self.index - 1)].cp_info + return self._m_as_info if hasattr(self, '_m_as_info') else None + + @property + def name_as_str(self): + if hasattr(self, '_m_name_as_str'): + return self._m_name_as_str if hasattr(self, '_m_name_as_str') else None + + self._m_name_as_str = self.as_info.name_as_str + return self._m_name_as_str if hasattr(self, '_m_name_as_str') else None + + + + class AttrBodySourceFile(KaitaiStruct): + """ + .. seealso:: + Source - https://docs.oracle.com/javase/specs/jvms/se8/html/jvms-4.html#jvms-4.7.10 + """ + SEQ_FIELDS = ["sourcefile_index"] + 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['sourcefile_index']['start'] = self._io.pos() + self.sourcefile_index = self._io.read_u2be() + self._debug['sourcefile_index']['end'] = self._io.pos() + + @property + def sourcefile_as_str(self): + if hasattr(self, '_m_sourcefile_as_str'): + return self._m_sourcefile_as_str if hasattr(self, '_m_sourcefile_as_str') else None + + self._m_sourcefile_as_str = self._root.constant_pool[(self.sourcefile_index - 1)].cp_info.value + return self._m_sourcefile_as_str if hasattr(self, '_m_sourcefile_as_str') else None + + + class AttrBodyLineNumberTable(KaitaiStruct): + """ + .. seealso:: + Source - https://docs.oracle.com/javase/specs/jvms/se8/html/jvms-4.html#jvms-4.7.12 + """ + SEQ_FIELDS = ["line_number_table_length", "line_number_table"] + 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['line_number_table_length']['start'] = self._io.pos() + self.line_number_table_length = self._io.read_u2be() + self._debug['line_number_table_length']['end'] = self._io.pos() + self._debug['line_number_table']['start'] = self._io.pos() + self.line_number_table = [None] * (self.line_number_table_length) + for i in range(self.line_number_table_length): + if not 'arr' in self._debug['line_number_table']: + self._debug['line_number_table']['arr'] = [] + self._debug['line_number_table']['arr'].append({'start': self._io.pos()}) + _t_line_number_table = self._root.AttributeInfo.AttrBodyLineNumberTable.LineNumberTableEntry(self._io, self, self._root) + _t_line_number_table._read() + self.line_number_table[i] = _t_line_number_table + self._debug['line_number_table']['arr'][i]['end'] = self._io.pos() + + self._debug['line_number_table']['end'] = self._io.pos() + + class LineNumberTableEntry(KaitaiStruct): + SEQ_FIELDS = ["start_pc", "line_number"] + 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['start_pc']['start'] = self._io.pos() + self.start_pc = self._io.read_u2be() + self._debug['start_pc']['end'] = self._io.pos() + self._debug['line_number']['start'] = self._io.pos() + self.line_number = self._io.read_u2be() + self._debug['line_number']['end'] = self._io.pos() + + + + @property + def name_as_str(self): + if hasattr(self, '_m_name_as_str'): + return self._m_name_as_str if hasattr(self, '_m_name_as_str') else None + + self._m_name_as_str = self._root.constant_pool[(self.name_index - 1)].cp_info.value + return self._m_name_as_str if hasattr(self, '_m_name_as_str') else None + + + class MethodRefCpInfo(KaitaiStruct): + """ + .. seealso:: + Source - https://docs.oracle.com/javase/specs/jvms/se8/html/jvms-4.html#jvms-4.4.2 + """ + SEQ_FIELDS = ["class_index", "name_and_type_index"] + 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['class_index']['start'] = self._io.pos() + self.class_index = self._io.read_u2be() + self._debug['class_index']['end'] = self._io.pos() + self._debug['name_and_type_index']['start'] = self._io.pos() + self.name_and_type_index = self._io.read_u2be() + self._debug['name_and_type_index']['end'] = self._io.pos() + + @property + def class_as_info(self): + if hasattr(self, '_m_class_as_info'): + return self._m_class_as_info if hasattr(self, '_m_class_as_info') else None + + self._m_class_as_info = self._root.constant_pool[(self.class_index - 1)].cp_info + return self._m_class_as_info if hasattr(self, '_m_class_as_info') else None + + @property + def name_and_type_as_info(self): + if hasattr(self, '_m_name_and_type_as_info'): + return self._m_name_and_type_as_info if hasattr(self, '_m_name_and_type_as_info') else None + + self._m_name_and_type_as_info = self._root.constant_pool[(self.name_and_type_index - 1)].cp_info + return self._m_name_and_type_as_info if hasattr(self, '_m_name_and_type_as_info') else None + + + class FieldInfo(KaitaiStruct): + """ + .. seealso:: + Source - https://docs.oracle.com/javase/specs/jvms/se8/html/jvms-4.html#jvms-4.5 + """ + SEQ_FIELDS = ["access_flags", "name_index", "descriptor_index", "attributes_count", "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['access_flags']['start'] = self._io.pos() + self.access_flags = self._io.read_u2be() + self._debug['access_flags']['end'] = self._io.pos() + self._debug['name_index']['start'] = self._io.pos() + self.name_index = self._io.read_u2be() + self._debug['name_index']['end'] = self._io.pos() + self._debug['descriptor_index']['start'] = self._io.pos() + self.descriptor_index = self._io.read_u2be() + self._debug['descriptor_index']['end'] = self._io.pos() + self._debug['attributes_count']['start'] = self._io.pos() + self.attributes_count = self._io.read_u2be() + self._debug['attributes_count']['end'] = self._io.pos() + self._debug['attributes']['start'] = self._io.pos() + self.attributes = [None] * (self.attributes_count) + for i in range(self.attributes_count): + if not 'arr' in self._debug['attributes']: + self._debug['attributes']['arr'] = [] + self._debug['attributes']['arr'].append({'start': self._io.pos()}) + _t_attributes = self._root.AttributeInfo(self._io, self, self._root) + _t_attributes._read() + self.attributes[i] = _t_attributes + self._debug['attributes']['arr'][i]['end'] = self._io.pos() + + self._debug['attributes']['end'] = self._io.pos() + + @property + def name_as_str(self): + if hasattr(self, '_m_name_as_str'): + return self._m_name_as_str if hasattr(self, '_m_name_as_str') else None + + self._m_name_as_str = self._root.constant_pool[(self.name_index - 1)].cp_info.value + return self._m_name_as_str if hasattr(self, '_m_name_as_str') else None + + + class DoubleCpInfo(KaitaiStruct): + """ + .. seealso:: + Source - https://docs.oracle.com/javase/specs/jvms/se8/html/jvms-4.html#jvms-4.4.6 + """ + SEQ_FIELDS = ["value"] + 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['value']['start'] = self._io.pos() + self.value = self._io.read_f8be() + self._debug['value']['end'] = self._io.pos() + + + class LongCpInfo(KaitaiStruct): + """ + .. seealso:: + Source - https://docs.oracle.com/javase/specs/jvms/se8/html/jvms-4.html#jvms-4.4.5 + """ + SEQ_FIELDS = ["value"] + 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['value']['start'] = self._io.pos() + self.value = self._io.read_u8be() + self._debug['value']['end'] = self._io.pos() + + + class InvokeDynamicCpInfo(KaitaiStruct): + """ + .. seealso:: + Source - https://docs.oracle.com/javase/specs/jvms/se8/html/jvms-4.html#jvms-4.4.10 + """ + SEQ_FIELDS = ["bootstrap_method_attr_index", "name_and_type_index"] + 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['bootstrap_method_attr_index']['start'] = self._io.pos() + self.bootstrap_method_attr_index = self._io.read_u2be() + self._debug['bootstrap_method_attr_index']['end'] = self._io.pos() + self._debug['name_and_type_index']['start'] = self._io.pos() + self.name_and_type_index = self._io.read_u2be() + self._debug['name_and_type_index']['end'] = self._io.pos() + + + class MethodHandleCpInfo(KaitaiStruct): + """ + .. seealso:: + Source - https://docs.oracle.com/javase/specs/jvms/se8/html/jvms-4.html#jvms-4.4.8 + """ + + class ReferenceKindEnum(Enum): + get_field = 1 + get_static = 2 + put_field = 3 + put_static = 4 + invoke_virtual = 5 + invoke_static = 6 + invoke_special = 7 + new_invoke_special = 8 + invoke_interface = 9 + SEQ_FIELDS = ["reference_kind", "reference_index"] + 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['reference_kind']['start'] = self._io.pos() + self.reference_kind = KaitaiStream.resolve_enum(self._root.MethodHandleCpInfo.ReferenceKindEnum, self._io.read_u1()) + self._debug['reference_kind']['end'] = self._io.pos() + self._debug['reference_index']['start'] = self._io.pos() + self.reference_index = self._io.read_u2be() + self._debug['reference_index']['end'] = self._io.pos() + + + class NameAndTypeCpInfo(KaitaiStruct): + """ + .. seealso:: + Source - https://docs.oracle.com/javase/specs/jvms/se8/html/jvms-4.html#jvms-4.4.6 + """ + SEQ_FIELDS = ["name_index", "descriptor_index"] + 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['name_index']['start'] = self._io.pos() + self.name_index = self._io.read_u2be() + self._debug['name_index']['end'] = self._io.pos() + self._debug['descriptor_index']['start'] = self._io.pos() + self.descriptor_index = self._io.read_u2be() + self._debug['descriptor_index']['end'] = self._io.pos() + + @property + def name_as_info(self): + if hasattr(self, '_m_name_as_info'): + return self._m_name_as_info if hasattr(self, '_m_name_as_info') else None + + self._m_name_as_info = self._root.constant_pool[(self.name_index - 1)].cp_info + return self._m_name_as_info if hasattr(self, '_m_name_as_info') else None + + @property + def name_as_str(self): + if hasattr(self, '_m_name_as_str'): + return self._m_name_as_str if hasattr(self, '_m_name_as_str') else None + + self._m_name_as_str = self.name_as_info.value + return self._m_name_as_str if hasattr(self, '_m_name_as_str') else None + + @property + def descriptor_as_info(self): + if hasattr(self, '_m_descriptor_as_info'): + return self._m_descriptor_as_info if hasattr(self, '_m_descriptor_as_info') else None + + self._m_descriptor_as_info = self._root.constant_pool[(self.descriptor_index - 1)].cp_info + return self._m_descriptor_as_info if hasattr(self, '_m_descriptor_as_info') else None + + @property + def descriptor_as_str(self): + if hasattr(self, '_m_descriptor_as_str'): + return self._m_descriptor_as_str if hasattr(self, '_m_descriptor_as_str') else None + + self._m_descriptor_as_str = self.descriptor_as_info.value + return self._m_descriptor_as_str if hasattr(self, '_m_descriptor_as_str') else None + + + class Utf8CpInfo(KaitaiStruct): + """ + .. seealso:: + Source - https://docs.oracle.com/javase/specs/jvms/se8/html/jvms-4.html#jvms-4.4.7 + """ + SEQ_FIELDS = ["str_len", "value"] + 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['str_len']['start'] = self._io.pos() + self.str_len = self._io.read_u2be() + self._debug['str_len']['end'] = self._io.pos() + self._debug['value']['start'] = self._io.pos() + self.value = (self._io.read_bytes(self.str_len)).decode(u"UTF-8") + self._debug['value']['end'] = self._io.pos() + + + class StringCpInfo(KaitaiStruct): + """ + .. seealso:: + Source - https://docs.oracle.com/javase/specs/jvms/se8/html/jvms-4.html#jvms-4.4.3 + """ + SEQ_FIELDS = ["string_index"] + 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['string_index']['start'] = self._io.pos() + self.string_index = self._io.read_u2be() + self._debug['string_index']['end'] = self._io.pos() + + + class MethodTypeCpInfo(KaitaiStruct): + """ + .. seealso:: + Source - https://docs.oracle.com/javase/specs/jvms/se8/html/jvms-4.html#jvms-4.4.9 + """ + SEQ_FIELDS = ["descriptor_index"] + 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['descriptor_index']['start'] = self._io.pos() + self.descriptor_index = self._io.read_u2be() + self._debug['descriptor_index']['end'] = self._io.pos() + + + class InterfaceMethodRefCpInfo(KaitaiStruct): + """ + .. seealso:: + Source - https://docs.oracle.com/javase/specs/jvms/se8/html/jvms-4.html#jvms-4.4.2 + """ + SEQ_FIELDS = ["class_index", "name_and_type_index"] + 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['class_index']['start'] = self._io.pos() + self.class_index = self._io.read_u2be() + self._debug['class_index']['end'] = self._io.pos() + self._debug['name_and_type_index']['start'] = self._io.pos() + self.name_and_type_index = self._io.read_u2be() + self._debug['name_and_type_index']['end'] = self._io.pos() + + @property + def class_as_info(self): + if hasattr(self, '_m_class_as_info'): + return self._m_class_as_info if hasattr(self, '_m_class_as_info') else None + + self._m_class_as_info = self._root.constant_pool[(self.class_index - 1)].cp_info + return self._m_class_as_info if hasattr(self, '_m_class_as_info') else None + + @property + def name_and_type_as_info(self): + if hasattr(self, '_m_name_and_type_as_info'): + return self._m_name_and_type_as_info if hasattr(self, '_m_name_and_type_as_info') else None + + self._m_name_and_type_as_info = self._root.constant_pool[(self.name_and_type_index - 1)].cp_info + return self._m_name_and_type_as_info if hasattr(self, '_m_name_and_type_as_info') else None + + + class ClassCpInfo(KaitaiStruct): + """ + .. seealso:: + Source - https://docs.oracle.com/javase/specs/jvms/se8/html/jvms-4.html#jvms-4.4.1 + """ + SEQ_FIELDS = ["name_index"] + 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['name_index']['start'] = self._io.pos() + self.name_index = self._io.read_u2be() + self._debug['name_index']['end'] = self._io.pos() + + @property + def name_as_info(self): + if hasattr(self, '_m_name_as_info'): + return self._m_name_as_info if hasattr(self, '_m_name_as_info') else None + + self._m_name_as_info = self._root.constant_pool[(self.name_index - 1)].cp_info + return self._m_name_as_info if hasattr(self, '_m_name_as_info') else None + + @property + def name_as_str(self): + if hasattr(self, '_m_name_as_str'): + return self._m_name_as_str if hasattr(self, '_m_name_as_str') else None + + self._m_name_as_str = self.name_as_info.value + return self._m_name_as_str if hasattr(self, '_m_name_as_str') else None + + + class ConstantPoolEntry(KaitaiStruct): + """ + .. seealso:: + Source - https://docs.oracle.com/javase/specs/jvms/se8/html/jvms-4.html#jvms-4.4 + """ + + class TagEnum(Enum): + utf8 = 1 + integer = 3 + float = 4 + long = 5 + double = 6 + class_type = 7 + string = 8 + field_ref = 9 + method_ref = 10 + interface_method_ref = 11 + name_and_type = 12 + method_handle = 15 + method_type = 16 + invoke_dynamic = 18 + SEQ_FIELDS = ["tag", "cp_info"] + 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 = KaitaiStream.resolve_enum(self._root.ConstantPoolEntry.TagEnum, self._io.read_u1()) + self._debug['tag']['end'] = self._io.pos() + self._debug['cp_info']['start'] = self._io.pos() + _on = self.tag + if _on == self._root.ConstantPoolEntry.TagEnum.interface_method_ref: + self.cp_info = self._root.InterfaceMethodRefCpInfo(self._io, self, self._root) + self.cp_info._read() + elif _on == self._root.ConstantPoolEntry.TagEnum.class_type: + self.cp_info = self._root.ClassCpInfo(self._io, self, self._root) + self.cp_info._read() + elif _on == self._root.ConstantPoolEntry.TagEnum.utf8: + self.cp_info = self._root.Utf8CpInfo(self._io, self, self._root) + self.cp_info._read() + elif _on == self._root.ConstantPoolEntry.TagEnum.method_type: + self.cp_info = self._root.MethodTypeCpInfo(self._io, self, self._root) + self.cp_info._read() + elif _on == self._root.ConstantPoolEntry.TagEnum.integer: + self.cp_info = self._root.IntegerCpInfo(self._io, self, self._root) + self.cp_info._read() + elif _on == self._root.ConstantPoolEntry.TagEnum.string: + self.cp_info = self._root.StringCpInfo(self._io, self, self._root) + self.cp_info._read() + elif _on == self._root.ConstantPoolEntry.TagEnum.float: + self.cp_info = self._root.FloatCpInfo(self._io, self, self._root) + self.cp_info._read() + elif _on == self._root.ConstantPoolEntry.TagEnum.long: + self.cp_info = self._root.LongCpInfo(self._io, self, self._root) + self.cp_info._read() + elif _on == self._root.ConstantPoolEntry.TagEnum.method_ref: + self.cp_info = self._root.MethodRefCpInfo(self._io, self, self._root) + self.cp_info._read() + elif _on == self._root.ConstantPoolEntry.TagEnum.double: + self.cp_info = self._root.DoubleCpInfo(self._io, self, self._root) + self.cp_info._read() + elif _on == self._root.ConstantPoolEntry.TagEnum.invoke_dynamic: + self.cp_info = self._root.InvokeDynamicCpInfo(self._io, self, self._root) + self.cp_info._read() + elif _on == self._root.ConstantPoolEntry.TagEnum.field_ref: + self.cp_info = self._root.FieldRefCpInfo(self._io, self, self._root) + self.cp_info._read() + elif _on == self._root.ConstantPoolEntry.TagEnum.method_handle: + self.cp_info = self._root.MethodHandleCpInfo(self._io, self, self._root) + self.cp_info._read() + elif _on == self._root.ConstantPoolEntry.TagEnum.name_and_type: + self.cp_info = self._root.NameAndTypeCpInfo(self._io, self, self._root) + self.cp_info._read() + self._debug['cp_info']['end'] = self._io.pos() + + + class MethodInfo(KaitaiStruct): + """ + .. seealso:: + Source - https://docs.oracle.com/javase/specs/jvms/se8/html/jvms-4.html#jvms-4.6 + """ + SEQ_FIELDS = ["access_flags", "name_index", "descriptor_index", "attributes_count", "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['access_flags']['start'] = self._io.pos() + self.access_flags = self._io.read_u2be() + self._debug['access_flags']['end'] = self._io.pos() + self._debug['name_index']['start'] = self._io.pos() + self.name_index = self._io.read_u2be() + self._debug['name_index']['end'] = self._io.pos() + self._debug['descriptor_index']['start'] = self._io.pos() + self.descriptor_index = self._io.read_u2be() + self._debug['descriptor_index']['end'] = self._io.pos() + self._debug['attributes_count']['start'] = self._io.pos() + self.attributes_count = self._io.read_u2be() + self._debug['attributes_count']['end'] = self._io.pos() + self._debug['attributes']['start'] = self._io.pos() + self.attributes = [None] * (self.attributes_count) + for i in range(self.attributes_count): + if not 'arr' in self._debug['attributes']: + self._debug['attributes']['arr'] = [] + self._debug['attributes']['arr'].append({'start': self._io.pos()}) + _t_attributes = self._root.AttributeInfo(self._io, self, self._root) + _t_attributes._read() + self.attributes[i] = _t_attributes + self._debug['attributes']['arr'][i]['end'] = self._io.pos() + + self._debug['attributes']['end'] = self._io.pos() + + @property + def name_as_str(self): + if hasattr(self, '_m_name_as_str'): + return self._m_name_as_str if hasattr(self, '_m_name_as_str') else None + + self._m_name_as_str = self._root.constant_pool[(self.name_index - 1)].cp_info.value + return self._m_name_as_str if hasattr(self, '_m_name_as_str') else None + + + class IntegerCpInfo(KaitaiStruct): + """ + .. seealso:: + Source - https://docs.oracle.com/javase/specs/jvms/se8/html/jvms-4.html#jvms-4.4.4 + """ + SEQ_FIELDS = ["value"] + 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['value']['start'] = self._io.pos() + self.value = self._io.read_u4be() + self._debug['value']['end'] = self._io.pos() + + + class FieldRefCpInfo(KaitaiStruct): + """ + .. seealso:: + Source - https://docs.oracle.com/javase/specs/jvms/se8/html/jvms-4.html#jvms-4.4.2 + """ + SEQ_FIELDS = ["class_index", "name_and_type_index"] + 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['class_index']['start'] = self._io.pos() + self.class_index = self._io.read_u2be() + self._debug['class_index']['end'] = self._io.pos() + self._debug['name_and_type_index']['start'] = self._io.pos() + self.name_and_type_index = self._io.read_u2be() + self._debug['name_and_type_index']['end'] = self._io.pos() + + @property + def class_as_info(self): + if hasattr(self, '_m_class_as_info'): + return self._m_class_as_info if hasattr(self, '_m_class_as_info') else None + + self._m_class_as_info = self._root.constant_pool[(self.class_index - 1)].cp_info + return self._m_class_as_info if hasattr(self, '_m_class_as_info') else None + + @property + def name_and_type_as_info(self): + if hasattr(self, '_m_name_and_type_as_info'): + return self._m_name_and_type_as_info if hasattr(self, '_m_name_and_type_as_info') else None + + self._m_name_and_type_as_info = self._root.constant_pool[(self.name_and_type_index - 1)].cp_info + return self._m_name_and_type_as_info if hasattr(self, '_m_name_and_type_as_info') else None + + + diff --git a/python/examples/kaitai/kaitai_struct_formats/executable/python_pyc_27.py b/python/examples/kaitai/kaitai_struct_formats/executable/python_pyc_27.py new file mode 100644 index 00000000..2ac4154d --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/executable/python_pyc_27.py @@ -0,0 +1,508 @@ +from __future__ import absolute_import +# 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 PythonPyc27(KaitaiStruct): + """Python interpreter runs .py files in 2 step process: first, it + produces bytecode, which it then executes. Translation of .py source + into bytecode is time-consuming, so Python dumps compiled bytecode + into .pyc files, to be reused from cache at later time if possible. + + .pyc file is essentially a raw dump of `py_object` (see `body`) with + a simple header prepended. + """ + + class Version(Enum): + v15 = 20121 + v16 = 50428 + v20 = 50823 + v21 = 60202 + v22 = 60717 + v23_a0 = 62011 + v23_a0b = 62021 + v24_a0 = 62041 + v24_a3 = 62051 + v24_b1 = 62061 + v25_a0 = 62071 + v25_a0b = 62081 + v25_a0c = 62091 + v25_a0d = 62092 + v25_b3 = 62101 + v25_b3b = 62111 + v25_c1 = 62121 + v25_c2 = 62131 + v26_a0 = 62151 + v26_a1 = 62161 + v27_a0 = 62171 + v27_a0b = 62181 + v27_a0c = 62191 + v27_a0d = 62201 + v27_a0e = 62211 + SEQ_FIELDS = ["version_magic", "crlf", "modification_timestamp", "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['version_magic']['start'] = self._io.pos() + self.version_magic = KaitaiStream.resolve_enum(self._root.Version, self._io.read_u2le()) + self._debug['version_magic']['end'] = self._io.pos() + self._debug['crlf']['start'] = self._io.pos() + self.crlf = self._io.read_u2le() + self._debug['crlf']['end'] = self._io.pos() + self._debug['modification_timestamp']['start'] = self._io.pos() + self.modification_timestamp = self._io.read_u4le() + self._debug['modification_timestamp']['end'] = self._io.pos() + self._debug['body']['start'] = self._io.pos() + self.body = self._root.PyObject(self._io, self, self._root) + self.body._read() + self._debug['body']['end'] = self._io.pos() + + class CodeObject(KaitaiStruct): + + class FlagsEnum(Enum): + has_args = 4 + has_kwargs = 8 + generator = 32 + SEQ_FIELDS = ["arg_count", "local_count", "stack_size", "flags", "code", "consts", "names", "var_names", "free_vars", "cell_vars", "filename", "name", "first_line_no", "lnotab"] + 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['arg_count']['start'] = self._io.pos() + self.arg_count = self._io.read_u4le() + self._debug['arg_count']['end'] = self._io.pos() + self._debug['local_count']['start'] = self._io.pos() + self.local_count = self._io.read_u4le() + self._debug['local_count']['end'] = self._io.pos() + self._debug['stack_size']['start'] = self._io.pos() + self.stack_size = self._io.read_u4le() + self._debug['stack_size']['end'] = self._io.pos() + self._debug['flags']['start'] = self._io.pos() + self.flags = KaitaiStream.resolve_enum(self._root.CodeObject.FlagsEnum, self._io.read_u4le()) + self._debug['flags']['end'] = self._io.pos() + self._debug['code']['start'] = self._io.pos() + self.code = self._root.Assembly(self._io, self, self._root) + self.code._read() + self._debug['code']['end'] = self._io.pos() + self._debug['consts']['start'] = self._io.pos() + self.consts = self._root.PyObject(self._io, self, self._root) + self.consts._read() + self._debug['consts']['end'] = self._io.pos() + self._debug['names']['start'] = self._io.pos() + self.names = self._root.PyObject(self._io, self, self._root) + self.names._read() + self._debug['names']['end'] = self._io.pos() + self._debug['var_names']['start'] = self._io.pos() + self.var_names = self._root.PyObject(self._io, self, self._root) + self.var_names._read() + self._debug['var_names']['end'] = self._io.pos() + self._debug['free_vars']['start'] = self._io.pos() + self.free_vars = self._root.PyObject(self._io, self, self._root) + self.free_vars._read() + self._debug['free_vars']['end'] = self._io.pos() + self._debug['cell_vars']['start'] = self._io.pos() + self.cell_vars = self._root.PyObject(self._io, self, self._root) + self.cell_vars._read() + self._debug['cell_vars']['end'] = self._io.pos() + self._debug['filename']['start'] = self._io.pos() + self.filename = self._root.PyObject(self._io, self, self._root) + self.filename._read() + self._debug['filename']['end'] = self._io.pos() + self._debug['name']['start'] = self._io.pos() + self.name = self._root.PyObject(self._io, self, self._root) + self.name._read() + self._debug['name']['end'] = self._io.pos() + self._debug['first_line_no']['start'] = self._io.pos() + self.first_line_no = self._io.read_u4le() + self._debug['first_line_no']['end'] = self._io.pos() + self._debug['lnotab']['start'] = self._io.pos() + self.lnotab = self._root.PyObject(self._io, self, self._root) + self.lnotab._read() + self._debug['lnotab']['end'] = self._io.pos() + + + class Assembly(KaitaiStruct): + SEQ_FIELDS = ["string_magic", "length", "items"] + 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['string_magic']['start'] = self._io.pos() + self.string_magic = self._io.ensure_fixed_contents(b"\x73") + self._debug['string_magic']['end'] = self._io.pos() + self._debug['length']['start'] = self._io.pos() + self.length = self._io.read_u4le() + self._debug['length']['end'] = self._io.pos() + self._debug['items']['start'] = self._io.pos() + self._raw_items = self._io.read_bytes(self.length) + io = KaitaiStream(BytesIO(self._raw_items)) + self.items = self._root.OpArgs(io, self, self._root) + self.items._read() + self._debug['items']['end'] = self._io.pos() + + + class OpArg(KaitaiStruct): + + class OpCodeEnum(Enum): + stop_code = 0 + pop_top = 1 + rot_two = 2 + rot_three = 3 + dup_top = 4 + rot_four = 5 + nop = 9 + unary_positive = 10 + unary_negative = 11 + unary_not = 12 + unary_convert = 13 + unary_invert = 15 + binary_power = 19 + binary_multiply = 20 + binary_divide = 21 + binary_modulo = 22 + binary_add = 23 + binary_subtract = 24 + binary_subscr = 25 + binary_floor_divide = 26 + binary_true_divide = 27 + inplace_floor_divide = 28 + inplace_true_divide = 29 + slice_0 = 30 + slice_1 = 31 + slice_2 = 32 + slice_3 = 33 + store_slice_0 = 40 + store_slice_1 = 41 + store_slice_2 = 42 + store_slice_3 = 43 + delete_slice_0 = 50 + delete_slice_1 = 51 + delete_slice_2 = 52 + delete_slice_3 = 53 + store_map = 54 + inplace_add = 55 + inplace_subtract = 56 + inplace_multiply = 57 + inplace_divide = 58 + inplace_modulo = 59 + store_subscr = 60 + delete_subscr = 61 + binary_lshift = 62 + binary_rshift = 63 + binary_and = 64 + binary_xor = 65 + binary_or = 66 + inplace_power = 67 + get_iter = 68 + print_expr = 70 + print_item = 71 + print_newline = 72 + print_item_to = 73 + print_newline_to = 74 + inplace_lshift = 75 + inplace_rshift = 76 + inplace_and = 77 + inplace_xor = 78 + inplace_or = 79 + break_loop = 80 + with_cleanup = 81 + load_locals = 82 + return_value = 83 + import_star = 84 + exec_stmt = 85 + yield_value = 86 + pop_block = 87 + end_finally = 88 + build_class = 89 + store_name = 90 + delete_name = 91 + unpack_sequence = 92 + for_iter = 93 + list_append = 94 + store_attr = 95 + delete_attr = 96 + store_global = 97 + delete_global = 98 + dup_topx = 99 + load_const = 100 + load_name = 101 + build_tuple = 102 + build_list = 103 + build_set = 104 + build_map = 105 + load_attr = 106 + compare_op = 107 + import_name = 108 + import_from = 109 + jump_forward = 110 + jump_if_false_or_pop = 111 + jump_if_true_or_pop = 112 + jump_absolute = 113 + pop_jump_if_false = 114 + pop_jump_if_true = 115 + load_global = 116 + continue_loop = 119 + setup_loop = 120 + setup_except = 121 + setup_finally = 122 + load_fast = 124 + store_fast = 125 + delete_fast = 126 + raise_varargs = 130 + call_function = 131 + make_function = 132 + build_slice = 133 + make_closure = 134 + load_closure = 135 + load_deref = 136 + store_deref = 137 + call_function_var = 140 + call_function_kw = 141 + call_function_var_kw = 142 + setup_with = 143 + extended_arg = 145 + set_add = 146 + map_add = 147 + SEQ_FIELDS = ["op_code", "arg"] + 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['op_code']['start'] = self._io.pos() + self.op_code = KaitaiStream.resolve_enum(self._root.OpArg.OpCodeEnum, self._io.read_u1()) + self._debug['op_code']['end'] = self._io.pos() + if self.op_code.value >= self._root.OpArg.OpCodeEnum.store_name.value: + self._debug['arg']['start'] = self._io.pos() + self.arg = self._io.read_u2le() + self._debug['arg']['end'] = self._io.pos() + + + + class PyObject(KaitaiStruct): + + class ObjectType(Enum): + tuple = 40 + py_false = 70 + none = 78 + string_ref = 82 + py_true = 84 + code_object = 99 + int = 105 + string = 115 + interned = 116 + unicode_string = 117 + SEQ_FIELDS = ["type", "value"] + 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['type']['start'] = self._io.pos() + self.type = KaitaiStream.resolve_enum(self._root.PyObject.ObjectType, self._io.read_u1()) + self._debug['type']['end'] = self._io.pos() + self._debug['value']['start'] = self._io.pos() + _on = self.type + if _on == self._root.PyObject.ObjectType.string: + self.value = self._root.PyObject.PyString(self._io, self, self._root) + self.value._read() + elif _on == self._root.PyObject.ObjectType.tuple: + self.value = self._root.PyObject.Tuple(self._io, self, self._root) + self.value._read() + elif _on == self._root.PyObject.ObjectType.int: + self.value = self._io.read_u4le() + elif _on == self._root.PyObject.ObjectType.py_true: + self.value = self._root.PyObject.PyTrue(self._io, self, self._root) + self.value._read() + elif _on == self._root.PyObject.ObjectType.py_false: + self.value = self._root.PyObject.PyFalse(self._io, self, self._root) + self.value._read() + elif _on == self._root.PyObject.ObjectType.none: + self.value = self._root.PyObject.PyNone(self._io, self, self._root) + self.value._read() + elif _on == self._root.PyObject.ObjectType.string_ref: + self.value = self._root.PyObject.StringRef(self._io, self, self._root) + self.value._read() + elif _on == self._root.PyObject.ObjectType.code_object: + self.value = self._root.CodeObject(self._io, self, self._root) + self.value._read() + elif _on == self._root.PyObject.ObjectType.interned: + self.value = self._root.PyObject.InternedString(self._io, self, self._root) + self.value._read() + self._debug['value']['end'] = self._io.pos() + + class PyNone(KaitaiStruct): + SEQ_FIELDS = [] + 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): + pass + + + class PyFalse(KaitaiStruct): + SEQ_FIELDS = [] + 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): + pass + + + class StringRef(KaitaiStruct): + SEQ_FIELDS = ["interned_list_index"] + 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['interned_list_index']['start'] = self._io.pos() + self.interned_list_index = self._io.read_u4le() + self._debug['interned_list_index']['end'] = self._io.pos() + + + class PyTrue(KaitaiStruct): + SEQ_FIELDS = [] + 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): + pass + + + class Tuple(KaitaiStruct): + SEQ_FIELDS = ["count", "items"] + 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['count']['start'] = self._io.pos() + self.count = self._io.read_u4le() + self._debug['count']['end'] = self._io.pos() + self._debug['items']['start'] = self._io.pos() + self.items = [None] * (self.count) + for i in range(self.count): + if not 'arr' in self._debug['items']: + self._debug['items']['arr'] = [] + self._debug['items']['arr'].append({'start': self._io.pos()}) + _t_items = self._root.PyObject(self._io, self, self._root) + _t_items._read() + self.items[i] = _t_items + self._debug['items']['arr'][i]['end'] = self._io.pos() + + self._debug['items']['end'] = self._io.pos() + + + class UnicodeString(KaitaiStruct): + SEQ_FIELDS = ["length", "data"] + 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['length']['start'] = self._io.pos() + self.length = self._io.read_u4le() + self._debug['length']['end'] = self._io.pos() + self._debug['data']['start'] = self._io.pos() + self.data = (self._io.read_bytes(self.length)).decode(u"utf-8") + self._debug['data']['end'] = self._io.pos() + + + class InternedString(KaitaiStruct): + SEQ_FIELDS = ["length", "data"] + 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['length']['start'] = self._io.pos() + self.length = self._io.read_u4le() + self._debug['length']['end'] = self._io.pos() + self._debug['data']['start'] = self._io.pos() + self.data = (self._io.read_bytes(self.length)).decode(u"utf-8") + self._debug['data']['end'] = self._io.pos() + + + class PyString(KaitaiStruct): + SEQ_FIELDS = ["length", "data"] + 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['length']['start'] = self._io.pos() + self.length = self._io.read_u4le() + self._debug['length']['end'] = self._io.pos() + self._debug['data']['start'] = self._io.pos() + self.data = self._io.read_bytes(self.length) + self._debug['data']['end'] = self._io.pos() + + + + class OpArgs(KaitaiStruct): + SEQ_FIELDS = ["items"] + 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['items']['start'] = self._io.pos() + self.items = [] + i = 0 + while not self._io.is_eof(): + if not 'arr' in self._debug['items']: + self._debug['items']['arr'] = [] + self._debug['items']['arr'].append({'start': self._io.pos()}) + _t_items = self._root.OpArg(self._io, self, self._root) + _t_items._read() + self.items.append(_t_items) + self._debug['items']['arr'][len(self.items) - 1]['end'] = self._io.pos() + i += 1 + + self._debug['items']['end'] = self._io.pos() + + + diff --git a/python/examples/kaitai/kaitai_struct_formats/executable/swf.py b/python/examples/kaitai/kaitai_struct_formats/executable/swf.py new file mode 100644 index 00000000..cb7022c2 --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/executable/swf.py @@ -0,0 +1,404 @@ +from __future__ import absolute_import +# 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 +import zlib + + +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 Swf(KaitaiStruct): + """SWF files are used by Adobe Flash (AKA Shockwave Flash, Macromedia + Flash) to encode rich interactive multimedia content and are, + essentially, a container for special bytecode instructions to play + back that content. In early 2000s, it was dominant rich multimedia + web format (.swf files were integrated into web pages and played + back with a browser plugin), but its usage largely declined in + 2010s, as HTML5 and performant browser-native solutions + (i.e. JavaScript engines and graphical approaches, such as WebGL) + emerged. + + There are a lot of versions of SWF (~36), format is somewhat + documented by Adobe. + + .. seealso:: + Source - https://www.adobe.com/content/dam/acom/en/devnet/pdf/swf-file-format-spec.pdf + """ + + class Compressions(Enum): + zlib = 67 + none = 70 + lzma = 90 + + class TagType(Enum): + end_of_file = 0 + place_object = 4 + remove_object = 5 + set_background_color = 9 + define_sound = 14 + place_object2 = 26 + remove_object2 = 28 + frame_label = 43 + export_assets = 56 + script_limits = 65 + file_attributes = 69 + place_object3 = 70 + symbol_class = 76 + metadata = 77 + define_scaling_grid = 78 + do_abc = 82 + define_scene_and_frame_label_data = 86 + SEQ_FIELDS = ["compression", "signature", "version", "len_file", "plain_body", "zlib_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['compression']['start'] = self._io.pos() + self.compression = KaitaiStream.resolve_enum(self._root.Compressions, self._io.read_u1()) + self._debug['compression']['end'] = self._io.pos() + self._debug['signature']['start'] = self._io.pos() + self.signature = self._io.ensure_fixed_contents(b"\x57\x53") + self._debug['signature']['end'] = self._io.pos() + self._debug['version']['start'] = self._io.pos() + self.version = self._io.read_u1() + self._debug['version']['end'] = self._io.pos() + self._debug['len_file']['start'] = self._io.pos() + self.len_file = self._io.read_u4le() + self._debug['len_file']['end'] = self._io.pos() + if self.compression == self._root.Compressions.none: + self._debug['plain_body']['start'] = self._io.pos() + self._raw_plain_body = self._io.read_bytes_full() + io = KaitaiStream(BytesIO(self._raw_plain_body)) + self.plain_body = self._root.SwfBody(io, self, self._root) + self.plain_body._read() + self._debug['plain_body']['end'] = self._io.pos() + + if self.compression == self._root.Compressions.zlib: + self._debug['zlib_body']['start'] = self._io.pos() + self._raw__raw_zlib_body = self._io.read_bytes_full() + self._raw_zlib_body = zlib.decompress(self._raw__raw_zlib_body) + io = KaitaiStream(BytesIO(self._raw_zlib_body)) + self.zlib_body = self._root.SwfBody(io, self, self._root) + self.zlib_body._read() + self._debug['zlib_body']['end'] = self._io.pos() + + + class Rgb(KaitaiStruct): + SEQ_FIELDS = ["r", "g", "b"] + 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['r']['start'] = self._io.pos() + self.r = self._io.read_u1() + self._debug['r']['end'] = self._io.pos() + self._debug['g']['start'] = self._io.pos() + self.g = self._io.read_u1() + self._debug['g']['end'] = self._io.pos() + self._debug['b']['start'] = self._io.pos() + self.b = self._io.read_u1() + self._debug['b']['end'] = self._io.pos() + + + class DoAbcBody(KaitaiStruct): + SEQ_FIELDS = ["flags", "name", "abcdata"] + 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_u4le() + self._debug['flags']['end'] = self._io.pos() + self._debug['name']['start'] = self._io.pos() + self.name = (self._io.read_bytes_term(0, False, True, True)).decode(u"ASCII") + self._debug['name']['end'] = self._io.pos() + self._debug['abcdata']['start'] = self._io.pos() + self.abcdata = self._io.read_bytes_full() + self._debug['abcdata']['end'] = self._io.pos() + + + class SwfBody(KaitaiStruct): + SEQ_FIELDS = ["rect", "frame_rate", "frame_count", "file_attributes_tag", "tags"] + 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['rect']['start'] = self._io.pos() + self.rect = self._root.Rect(self._io, self, self._root) + self.rect._read() + self._debug['rect']['end'] = self._io.pos() + self._debug['frame_rate']['start'] = self._io.pos() + self.frame_rate = self._io.read_u2le() + self._debug['frame_rate']['end'] = self._io.pos() + self._debug['frame_count']['start'] = self._io.pos() + self.frame_count = self._io.read_u2le() + self._debug['frame_count']['end'] = self._io.pos() + if self._root.version >= 8: + self._debug['file_attributes_tag']['start'] = self._io.pos() + self.file_attributes_tag = self._root.Tag(self._io, self, self._root) + self.file_attributes_tag._read() + self._debug['file_attributes_tag']['end'] = self._io.pos() + + self._debug['tags']['start'] = self._io.pos() + self.tags = [] + i = 0 + while not self._io.is_eof(): + if not 'arr' in self._debug['tags']: + self._debug['tags']['arr'] = [] + self._debug['tags']['arr'].append({'start': self._io.pos()}) + _t_tags = self._root.Tag(self._io, self, self._root) + _t_tags._read() + self.tags.append(_t_tags) + self._debug['tags']['arr'][len(self.tags) - 1]['end'] = self._io.pos() + i += 1 + + self._debug['tags']['end'] = self._io.pos() + + + class Rect(KaitaiStruct): + SEQ_FIELDS = ["b1", "skip"] + 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['b1']['start'] = self._io.pos() + self.b1 = self._io.read_u1() + self._debug['b1']['end'] = self._io.pos() + self._debug['skip']['start'] = self._io.pos() + self.skip = self._io.read_bytes(self.num_bytes) + self._debug['skip']['end'] = self._io.pos() + + @property + def num_bits(self): + if hasattr(self, '_m_num_bits'): + return self._m_num_bits if hasattr(self, '_m_num_bits') else None + + self._m_num_bits = (self.b1 >> 3) + return self._m_num_bits if hasattr(self, '_m_num_bits') else None + + @property + def num_bytes(self): + if hasattr(self, '_m_num_bytes'): + return self._m_num_bytes if hasattr(self, '_m_num_bytes') else None + + self._m_num_bytes = (((self.num_bits * 4) - 3) + 7) // 8 + return self._m_num_bytes if hasattr(self, '_m_num_bytes') else None + + + class Tag(KaitaiStruct): + SEQ_FIELDS = ["record_header", "tag_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['record_header']['start'] = self._io.pos() + self.record_header = self._root.RecordHeader(self._io, self, self._root) + self.record_header._read() + self._debug['record_header']['end'] = self._io.pos() + self._debug['tag_body']['start'] = self._io.pos() + _on = self.record_header.tag_type + if _on == self._root.TagType.define_sound: + self._raw_tag_body = self._io.read_bytes(self.record_header.len) + io = KaitaiStream(BytesIO(self._raw_tag_body)) + self.tag_body = self._root.DefineSoundBody(io, self, self._root) + self.tag_body._read() + elif _on == self._root.TagType.set_background_color: + self._raw_tag_body = self._io.read_bytes(self.record_header.len) + io = KaitaiStream(BytesIO(self._raw_tag_body)) + self.tag_body = self._root.Rgb(io, self, self._root) + self.tag_body._read() + elif _on == self._root.TagType.script_limits: + self._raw_tag_body = self._io.read_bytes(self.record_header.len) + io = KaitaiStream(BytesIO(self._raw_tag_body)) + self.tag_body = self._root.ScriptLimitsBody(io, self, self._root) + self.tag_body._read() + elif _on == self._root.TagType.do_abc: + self._raw_tag_body = self._io.read_bytes(self.record_header.len) + io = KaitaiStream(BytesIO(self._raw_tag_body)) + self.tag_body = self._root.DoAbcBody(io, self, self._root) + self.tag_body._read() + elif _on == self._root.TagType.export_assets: + self._raw_tag_body = self._io.read_bytes(self.record_header.len) + io = KaitaiStream(BytesIO(self._raw_tag_body)) + self.tag_body = self._root.SymbolClassBody(io, self, self._root) + self.tag_body._read() + elif _on == self._root.TagType.symbol_class: + self._raw_tag_body = self._io.read_bytes(self.record_header.len) + io = KaitaiStream(BytesIO(self._raw_tag_body)) + self.tag_body = self._root.SymbolClassBody(io, self, self._root) + self.tag_body._read() + else: + self.tag_body = self._io.read_bytes(self.record_header.len) + self._debug['tag_body']['end'] = self._io.pos() + + + class SymbolClassBody(KaitaiStruct): + SEQ_FIELDS = ["num_symbols", "symbols"] + 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['num_symbols']['start'] = self._io.pos() + self.num_symbols = self._io.read_u2le() + self._debug['num_symbols']['end'] = self._io.pos() + self._debug['symbols']['start'] = self._io.pos() + self.symbols = [None] * (self.num_symbols) + for i in range(self.num_symbols): + if not 'arr' in self._debug['symbols']: + self._debug['symbols']['arr'] = [] + self._debug['symbols']['arr'].append({'start': self._io.pos()}) + _t_symbols = self._root.SymbolClassBody.Symbol(self._io, self, self._root) + _t_symbols._read() + self.symbols[i] = _t_symbols + self._debug['symbols']['arr'][i]['end'] = self._io.pos() + + self._debug['symbols']['end'] = self._io.pos() + + class Symbol(KaitaiStruct): + SEQ_FIELDS = ["tag", "name"] + 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['name']['start'] = self._io.pos() + self.name = (self._io.read_bytes_term(0, False, True, True)).decode(u"ASCII") + self._debug['name']['end'] = self._io.pos() + + + + class DefineSoundBody(KaitaiStruct): + + class SamplingRates(Enum): + rate_5_5_khz = 0 + rate_11_khz = 1 + rate_22_khz = 2 + rate_44_khz = 3 + + class Bps(Enum): + sound_8_bit = 0 + sound_16_bit = 1 + + class Channels(Enum): + mono = 0 + stereo = 1 + SEQ_FIELDS = ["id", "format", "sampling_rate", "bits_per_sample", "num_channels", "num_samples"] + 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['id']['start'] = self._io.pos() + self.id = self._io.read_u2le() + self._debug['id']['end'] = self._io.pos() + self._debug['format']['start'] = self._io.pos() + self.format = self._io.read_bits_int(4) + self._debug['format']['end'] = self._io.pos() + self._debug['sampling_rate']['start'] = self._io.pos() + self.sampling_rate = KaitaiStream.resolve_enum(self._root.DefineSoundBody.SamplingRates, self._io.read_bits_int(2)) + self._debug['sampling_rate']['end'] = self._io.pos() + self._debug['bits_per_sample']['start'] = self._io.pos() + self.bits_per_sample = KaitaiStream.resolve_enum(self._root.DefineSoundBody.Bps, self._io.read_bits_int(1)) + self._debug['bits_per_sample']['end'] = self._io.pos() + self._debug['num_channels']['start'] = self._io.pos() + self.num_channels = KaitaiStream.resolve_enum(self._root.DefineSoundBody.Channels, self._io.read_bits_int(1)) + self._debug['num_channels']['end'] = self._io.pos() + self._io.align_to_byte() + self._debug['num_samples']['start'] = self._io.pos() + self.num_samples = self._io.read_u4le() + self._debug['num_samples']['end'] = self._io.pos() + + + class RecordHeader(KaitaiStruct): + SEQ_FIELDS = ["tag_code_and_length", "big_len"] + 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_code_and_length']['start'] = self._io.pos() + self.tag_code_and_length = self._io.read_u2le() + self._debug['tag_code_and_length']['end'] = self._io.pos() + if self.small_len == 63: + self._debug['big_len']['start'] = self._io.pos() + self.big_len = self._io.read_s4le() + self._debug['big_len']['end'] = self._io.pos() + + + @property + def tag_type(self): + if hasattr(self, '_m_tag_type'): + return self._m_tag_type if hasattr(self, '_m_tag_type') else None + + self._m_tag_type = KaitaiStream.resolve_enum(self._root.TagType, (self.tag_code_and_length >> 6)) + return self._m_tag_type if hasattr(self, '_m_tag_type') else None + + @property + def small_len(self): + if hasattr(self, '_m_small_len'): + return self._m_small_len if hasattr(self, '_m_small_len') else None + + self._m_small_len = (self.tag_code_and_length & 63) + return self._m_small_len if hasattr(self, '_m_small_len') else None + + @property + def len(self): + if hasattr(self, '_m_len'): + return self._m_len if hasattr(self, '_m_len') else None + + self._m_len = (self.big_len if self.small_len == 63 else self.small_len) + return self._m_len if hasattr(self, '_m_len') else None + + + class ScriptLimitsBody(KaitaiStruct): + SEQ_FIELDS = ["max_recursion_depth", "script_timeout_seconds"] + 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['max_recursion_depth']['start'] = self._io.pos() + self.max_recursion_depth = self._io.read_u2le() + self._debug['max_recursion_depth']['end'] = self._io.pos() + self._debug['script_timeout_seconds']['start'] = self._io.pos() + self.script_timeout_seconds = self._io.read_u2le() + self._debug['script_timeout_seconds']['end'] = self._io.pos() + + + diff --git a/python/examples/kaitai/kaitai_struct_formats/executable/vlq_base128_le.py b/python/examples/kaitai/kaitai_struct_formats/executable/vlq_base128_le.py new file mode 100644 index 00000000..407fdac7 --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/executable/vlq_base128_le.py @@ -0,0 +1,108 @@ +# 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 +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 VlqBase128Le(KaitaiStruct): + """A variable-length unsigned integer using base128 encoding. 1-byte groups + consist of 1-bit flag of continuation and 7-bit value chunk, and are ordered + "least significant group first", i.e. in "little-endian" manner. + + This particular encoding is specified and used in: + + * DWARF debug file format, where it's dubbed "unsigned LEB128" or "ULEB128". + http://dwarfstd.org/doc/dwarf-2.0.0.pdf - page 139 + * Google Protocol Buffers, where it's called "Base 128 Varints". + https://developers.google.com/protocol-buffers/docs/encoding?csw=1#varints + * Apache Lucene, where it's called "VInt" + http://lucene.apache.org/core/3_5_0/fileformats.html#VInt + * Apache Avro uses this as a basis for integer encoding, adding ZigZag on + top of it for signed ints + http://avro.apache.org/docs/current/spec.html#binary_encode_primitive + + More information on this encoding is available at https://en.wikipedia.org/wiki/LEB128 + + This particular implementation supports serialized values to up 8 bytes long. + """ + SEQ_FIELDS = ["groups"] + 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['groups']['start'] = self._io.pos() + self.groups = [] + i = 0 + while True: + if not 'arr' in self._debug['groups']: + self._debug['groups']['arr'] = [] + self._debug['groups']['arr'].append({'start': self._io.pos()}) + _t_groups = self._root.Group(self._io, self, self._root) + _t_groups._read() + _ = _t_groups + self.groups.append(_) + self._debug['groups']['arr'][len(self.groups) - 1]['end'] = self._io.pos() + if not (_.has_next): + break + i += 1 + self._debug['groups']['end'] = self._io.pos() + + class Group(KaitaiStruct): + """One byte group, clearly divided into 7-bit "value" chunk and 1-bit "continuation" flag. + """ + SEQ_FIELDS = ["b"] + 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['b']['start'] = self._io.pos() + self.b = self._io.read_u1() + self._debug['b']['end'] = self._io.pos() + + @property + def has_next(self): + """If true, then we have more bytes to read.""" + if hasattr(self, '_m_has_next'): + return self._m_has_next if hasattr(self, '_m_has_next') else None + + self._m_has_next = (self.b & 128) != 0 + return self._m_has_next if hasattr(self, '_m_has_next') else None + + @property + def value(self): + """The 7-bit (base128) numeric value chunk of this group.""" + if hasattr(self, '_m_value'): + return self._m_value if hasattr(self, '_m_value') else None + + self._m_value = (self.b & 127) + return self._m_value if hasattr(self, '_m_value') else None + + + @property + def len(self): + if hasattr(self, '_m_len'): + return self._m_len if hasattr(self, '_m_len') else None + + self._m_len = len(self.groups) + return self._m_len if hasattr(self, '_m_len') else None + + @property + def value(self): + """Resulting value as normal integer.""" + if hasattr(self, '_m_value'): + return self._m_value if hasattr(self, '_m_value') else None + + 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)) + return self._m_value if hasattr(self, '_m_value') else None + + -- cgit v1.3.1