diff options
| author | Xusheng <xusheng@vector35.com> | 2021-07-01 12:34:30 +0800 |
|---|---|---|
| committer | Xusheng Li <xushengli@protonmail.com> | 2021-07-05 10:11:47 +0800 |
| commit | 13ff200ba134b8704f37eca99c42b70dab0d17dc (patch) | |
| tree | 876b3db3c53f704b31152aeaede1f2edc9d27b24 /python/examples/kaitai/windows_minidump.py | |
| parent | 4039a800349dcd587833d3e27d9137e4649818f4 (diff) | |
Remove snippet and kaitai plugin. Note their individual links.
Diffstat (limited to 'python/examples/kaitai/windows_minidump.py')
| -rw-r--r-- | python/examples/kaitai/windows_minidump.py | 540 |
1 files changed, 0 insertions, 540 deletions
diff --git a/python/examples/kaitai/windows_minidump.py b/python/examples/kaitai/windows_minidump.py deleted file mode 100644 index 8a9a0502..00000000 --- a/python/examples/kaitai/windows_minidump.py +++ /dev/null @@ -1,540 +0,0 @@ -# This is a generated file! Please edit source .ksy file and use kaitai-struct-compiler to rebuild - -from pkg_resources import parse_version -from .kaitaistruct import __version__ as ks_version, KaitaiStruct, KaitaiStream, BytesIO -from enum import Enum -import collections - - -if parse_version(ks_version) < parse_version('0.7'): - raise Exception("Incompatible Kaitai Struct Python API: 0.7 or later is required, but you have %s" % (ks_version)) - -class WindowsMinidump(KaitaiStruct): - """Windows MiniDump (MDMP) file provides a concise way to store process - core dumps, which is useful for debugging. Given its small size, - modularity, some cross-platform features and native support in some - debuggers, it is particularly useful for crash reporting, and is - used for that purpose in Windows and Google Chrome projects. - - The file itself is a container, which contains a number of typed - "streams", which contain some data according to its type attribute. - - .. seealso:: - Source - https://msdn.microsoft.com/en-us/library/ms680378(VS.85).aspx - """ - - class StreamTypes(Enum): - unused = 0 - reserved_0 = 1 - reserved_1 = 2 - thread_list = 3 - module_list = 4 - memory_list = 5 - exception = 6 - system_info = 7 - thread_ex_list = 8 - memory_64_list = 9 - comment_a = 10 - comment_w = 11 - handle_data = 12 - function_table = 13 - unloaded_module_list = 14 - misc_info = 15 - memory_info_list = 16 - thread_info_list = 17 - handle_operation_list = 18 - SEQ_FIELDS = ["magic1", "magic2", "version", "num_streams", "ofs_streams", "checksum", "timestamp", "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['magic1']['start'] = self._io.pos() - self.magic1 = self._io.ensure_fixed_contents(b"\x4D\x44\x4D\x50") - self._debug['magic1']['end'] = self._io.pos() - self._debug['magic2']['start'] = self._io.pos() - self.magic2 = self._io.ensure_fixed_contents(b"\x93\xA7") - self._debug['magic2']['end'] = self._io.pos() - self._debug['version']['start'] = self._io.pos() - self.version = self._io.read_u2le() - self._debug['version']['end'] = self._io.pos() - self._debug['num_streams']['start'] = self._io.pos() - self.num_streams = self._io.read_u4le() - self._debug['num_streams']['end'] = self._io.pos() - self._debug['ofs_streams']['start'] = self._io.pos() - self.ofs_streams = self._io.read_u4le() - self._debug['ofs_streams']['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['timestamp']['start'] = self._io.pos() - self.timestamp = self._io.read_u4le() - self._debug['timestamp']['end'] = self._io.pos() - self._debug['flags']['start'] = self._io.pos() - self.flags = self._io.read_u8le() - self._debug['flags']['end'] = self._io.pos() - - class ThreadList(KaitaiStruct): - """ - .. seealso:: - Source - https://msdn.microsoft.com/en-us/library/ms680515(v=vs.85).aspx - """ - SEQ_FIELDS = ["num_threads", "threads"] - 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_threads']['start'] = self._io.pos() - self.num_threads = self._io.read_u4le() - self._debug['num_threads']['end'] = self._io.pos() - self._debug['threads']['start'] = self._io.pos() - self.threads = [None] * (self.num_threads) - for i in range(self.num_threads): - if not 'arr' in self._debug['threads']: - self._debug['threads']['arr'] = [] - self._debug['threads']['arr'].append({'start': self._io.pos()}) - _t_threads = self._root.Thread(self._io, self, self._root) - _t_threads._read() - self.threads[i] = _t_threads - self._debug['threads']['arr'][i]['end'] = self._io.pos() - - self._debug['threads']['end'] = self._io.pos() - - - class LocationDescriptor(KaitaiStruct): - """ - .. seealso:: - Source - https://msdn.microsoft.com/en-us/library/ms680383(v=vs.85).aspx - """ - SEQ_FIELDS = ["len_data", "ofs_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['len_data']['start'] = self._io.pos() - self.len_data = self._io.read_u4le() - self._debug['len_data']['end'] = self._io.pos() - self._debug['ofs_data']['start'] = self._io.pos() - self.ofs_data = self._io.read_u4le() - self._debug['ofs_data']['end'] = self._io.pos() - - @property - def data(self): - if hasattr(self, '_m_data'): - return self._m_data if hasattr(self, '_m_data') else None - - io = self._root._io - _pos = io.pos() - io.seek(self.ofs_data) - self._debug['_m_data']['start'] = io.pos() - self._m_data = io.read_bytes(self.len_data) - self._debug['_m_data']['end'] = io.pos() - io.seek(_pos) - return self._m_data if hasattr(self, '_m_data') else None - - - class MinidumpString(KaitaiStruct): - """Specific string serialization scheme used in MiniDump format is - actually a simple 32-bit length-prefixed UTF-16 string. - - .. seealso:: - Source - https://msdn.microsoft.com/en-us/library/ms680395(v=vs.85).aspx - """ - SEQ_FIELDS = ["len_str", "str"] - 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['len_str']['start'] = self._io.pos() - self.len_str = self._io.read_u4le() - self._debug['len_str']['end'] = self._io.pos() - self._debug['str']['start'] = self._io.pos() - self.str = (self._io.read_bytes(self.len_str)).decode(u"UTF-16LE") - self._debug['str']['end'] = self._io.pos() - - - class SystemInfo(KaitaiStruct): - """"System info" stream provides basic information about the - hardware and operating system which produces this dump. - - .. seealso:: - Source - https://msdn.microsoft.com/en-us/library/ms680396(v=vs.85).aspx - """ - - class CpuArchs(Enum): - intel = 0 - arm = 5 - ia64 = 6 - amd64 = 9 - unknown = 65535 - SEQ_FIELDS = ["cpu_arch", "cpu_level", "cpu_revision", "num_cpus", "os_type", "os_ver_major", "os_ver_minor", "os_build", "os_platform", "ofs_service_pack", "os_suite_mask", "reserved2"] - 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['cpu_arch']['start'] = self._io.pos() - self.cpu_arch = KaitaiStream.resolve_enum(self._root.SystemInfo.CpuArchs, self._io.read_u2le()) - self._debug['cpu_arch']['end'] = self._io.pos() - self._debug['cpu_level']['start'] = self._io.pos() - self.cpu_level = self._io.read_u2le() - self._debug['cpu_level']['end'] = self._io.pos() - self._debug['cpu_revision']['start'] = self._io.pos() - self.cpu_revision = self._io.read_u2le() - self._debug['cpu_revision']['end'] = self._io.pos() - self._debug['num_cpus']['start'] = self._io.pos() - self.num_cpus = self._io.read_u1() - self._debug['num_cpus']['end'] = self._io.pos() - self._debug['os_type']['start'] = self._io.pos() - self.os_type = self._io.read_u1() - self._debug['os_type']['end'] = self._io.pos() - self._debug['os_ver_major']['start'] = self._io.pos() - self.os_ver_major = self._io.read_u4le() - self._debug['os_ver_major']['end'] = self._io.pos() - self._debug['os_ver_minor']['start'] = self._io.pos() - self.os_ver_minor = self._io.read_u4le() - self._debug['os_ver_minor']['end'] = self._io.pos() - self._debug['os_build']['start'] = self._io.pos() - self.os_build = self._io.read_u4le() - self._debug['os_build']['end'] = self._io.pos() - self._debug['os_platform']['start'] = self._io.pos() - self.os_platform = self._io.read_u4le() - self._debug['os_platform']['end'] = self._io.pos() - self._debug['ofs_service_pack']['start'] = self._io.pos() - self.ofs_service_pack = self._io.read_u4le() - self._debug['ofs_service_pack']['end'] = self._io.pos() - self._debug['os_suite_mask']['start'] = self._io.pos() - self.os_suite_mask = self._io.read_u2le() - self._debug['os_suite_mask']['end'] = self._io.pos() - self._debug['reserved2']['start'] = self._io.pos() - self.reserved2 = self._io.read_u2le() - self._debug['reserved2']['end'] = self._io.pos() - - @property - def service_pack(self): - if hasattr(self, '_m_service_pack'): - return self._m_service_pack if hasattr(self, '_m_service_pack') else None - - if self.ofs_service_pack > 0: - io = self._root._io - _pos = io.pos() - io.seek(self.ofs_service_pack) - self._debug['_m_service_pack']['start'] = io.pos() - self._m_service_pack = self._root.MinidumpString(io, self, self._root) - self._m_service_pack._read() - self._debug['_m_service_pack']['end'] = io.pos() - io.seek(_pos) - - return self._m_service_pack if hasattr(self, '_m_service_pack') else None - - - class ExceptionRecord(KaitaiStruct): - """ - .. seealso:: - Source - https://msdn.microsoft.com/en-us/library/ms680367(v=vs.85).aspx - """ - SEQ_FIELDS = ["code", "flags", "inner_exception", "addr", "num_params", "reserved", "params"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['code']['start'] = self._io.pos() - self.code = self._io.read_u4le() - self._debug['code']['end'] = self._io.pos() - self._debug['flags']['start'] = self._io.pos() - self.flags = self._io.read_u4le() - self._debug['flags']['end'] = self._io.pos() - self._debug['inner_exception']['start'] = self._io.pos() - self.inner_exception = self._io.read_u8le() - self._debug['inner_exception']['end'] = self._io.pos() - self._debug['addr']['start'] = self._io.pos() - self.addr = self._io.read_u8le() - self._debug['addr']['end'] = self._io.pos() - self._debug['num_params']['start'] = self._io.pos() - self.num_params = self._io.read_u4le() - self._debug['num_params']['end'] = self._io.pos() - self._debug['reserved']['start'] = self._io.pos() - self.reserved = self._io.read_u4le() - self._debug['reserved']['end'] = self._io.pos() - self._debug['params']['start'] = self._io.pos() - self.params = [None] * (15) - for i in range(15): - if not 'arr' in self._debug['params']: - self._debug['params']['arr'] = [] - self._debug['params']['arr'].append({'start': self._io.pos()}) - self.params[i] = self._io.read_u8le() - self._debug['params']['arr'][i]['end'] = self._io.pos() - - self._debug['params']['end'] = self._io.pos() - - - class MiscInfo(KaitaiStruct): - """ - .. seealso:: - Source - https://msdn.microsoft.com/en-us/library/ms680389(v=vs.85).aspx - """ - SEQ_FIELDS = ["len_info", "flags1", "process_id", "process_create_time", "process_user_time", "process_kernel_time", "cpu_max_mhz", "cpu_cur_mhz", "cpu_limit_mhz", "cpu_max_idle_state", "cpu_cur_idle_state"] - 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['len_info']['start'] = self._io.pos() - self.len_info = self._io.read_u4le() - self._debug['len_info']['end'] = self._io.pos() - self._debug['flags1']['start'] = self._io.pos() - self.flags1 = self._io.read_u4le() - self._debug['flags1']['end'] = self._io.pos() - self._debug['process_id']['start'] = self._io.pos() - self.process_id = self._io.read_u4le() - self._debug['process_id']['end'] = self._io.pos() - self._debug['process_create_time']['start'] = self._io.pos() - self.process_create_time = self._io.read_u4le() - self._debug['process_create_time']['end'] = self._io.pos() - self._debug['process_user_time']['start'] = self._io.pos() - self.process_user_time = self._io.read_u4le() - self._debug['process_user_time']['end'] = self._io.pos() - self._debug['process_kernel_time']['start'] = self._io.pos() - self.process_kernel_time = self._io.read_u4le() - self._debug['process_kernel_time']['end'] = self._io.pos() - self._debug['cpu_max_mhz']['start'] = self._io.pos() - self.cpu_max_mhz = self._io.read_u4le() - self._debug['cpu_max_mhz']['end'] = self._io.pos() - self._debug['cpu_cur_mhz']['start'] = self._io.pos() - self.cpu_cur_mhz = self._io.read_u4le() - self._debug['cpu_cur_mhz']['end'] = self._io.pos() - self._debug['cpu_limit_mhz']['start'] = self._io.pos() - self.cpu_limit_mhz = self._io.read_u4le() - self._debug['cpu_limit_mhz']['end'] = self._io.pos() - self._debug['cpu_max_idle_state']['start'] = self._io.pos() - self.cpu_max_idle_state = self._io.read_u4le() - self._debug['cpu_max_idle_state']['end'] = self._io.pos() - self._debug['cpu_cur_idle_state']['start'] = self._io.pos() - self.cpu_cur_idle_state = self._io.read_u4le() - self._debug['cpu_cur_idle_state']['end'] = self._io.pos() - - - class Dir(KaitaiStruct): - """ - .. seealso:: - Source - https://msdn.microsoft.com/en-us/library/ms680365(v=vs.85).aspx - """ - SEQ_FIELDS = ["stream_type", "len_data", "ofs_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['stream_type']['start'] = self._io.pos() - self.stream_type = KaitaiStream.resolve_enum(self._root.StreamTypes, self._io.read_u4le()) - self._debug['stream_type']['end'] = self._io.pos() - self._debug['len_data']['start'] = self._io.pos() - self.len_data = self._io.read_u4le() - self._debug['len_data']['end'] = self._io.pos() - self._debug['ofs_data']['start'] = self._io.pos() - self.ofs_data = self._io.read_u4le() - self._debug['ofs_data']['end'] = self._io.pos() - - @property - def data(self): - if hasattr(self, '_m_data'): - return self._m_data if hasattr(self, '_m_data') else None - - _pos = self._io.pos() - self._io.seek(self.ofs_data) - self._debug['_m_data']['start'] = self._io.pos() - _on = self.stream_type - if _on == self._root.StreamTypes.memory_list: - self._raw__m_data = self._io.read_bytes(self.len_data) - io = KaitaiStream(BytesIO(self._raw__m_data)) - self._m_data = self._root.MemoryList(io, self, self._root) - self._m_data._read() - elif _on == self._root.StreamTypes.misc_info: - self._raw__m_data = self._io.read_bytes(self.len_data) - io = KaitaiStream(BytesIO(self._raw__m_data)) - self._m_data = self._root.MiscInfo(io, self, self._root) - self._m_data._read() - elif _on == self._root.StreamTypes.thread_list: - self._raw__m_data = self._io.read_bytes(self.len_data) - io = KaitaiStream(BytesIO(self._raw__m_data)) - self._m_data = self._root.ThreadList(io, self, self._root) - self._m_data._read() - elif _on == self._root.StreamTypes.exception: - self._raw__m_data = self._io.read_bytes(self.len_data) - io = KaitaiStream(BytesIO(self._raw__m_data)) - self._m_data = self._root.ExceptionStream(io, self, self._root) - self._m_data._read() - elif _on == self._root.StreamTypes.system_info: - self._raw__m_data = self._io.read_bytes(self.len_data) - io = KaitaiStream(BytesIO(self._raw__m_data)) - self._m_data = self._root.SystemInfo(io, self, self._root) - self._m_data._read() - else: - self._m_data = self._io.read_bytes(self.len_data) - self._debug['_m_data']['end'] = self._io.pos() - self._io.seek(_pos) - return self._m_data if hasattr(self, '_m_data') else None - - - class Thread(KaitaiStruct): - """ - .. seealso:: - Source - https://msdn.microsoft.com/en-us/library/ms680517(v=vs.85).aspx - """ - SEQ_FIELDS = ["thread_id", "suspend_count", "priority_class", "priority", "teb", "stack", "thread_context"] - 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['thread_id']['start'] = self._io.pos() - self.thread_id = self._io.read_u4le() - self._debug['thread_id']['end'] = self._io.pos() - self._debug['suspend_count']['start'] = self._io.pos() - self.suspend_count = self._io.read_u4le() - self._debug['suspend_count']['end'] = self._io.pos() - self._debug['priority_class']['start'] = self._io.pos() - self.priority_class = self._io.read_u4le() - self._debug['priority_class']['end'] = self._io.pos() - self._debug['priority']['start'] = self._io.pos() - self.priority = self._io.read_u4le() - self._debug['priority']['end'] = self._io.pos() - self._debug['teb']['start'] = self._io.pos() - self.teb = self._io.read_u8le() - self._debug['teb']['end'] = self._io.pos() - self._debug['stack']['start'] = self._io.pos() - self.stack = self._root.MemoryDescriptor(self._io, self, self._root) - self.stack._read() - self._debug['stack']['end'] = self._io.pos() - self._debug['thread_context']['start'] = self._io.pos() - self.thread_context = self._root.LocationDescriptor(self._io, self, self._root) - self.thread_context._read() - self._debug['thread_context']['end'] = self._io.pos() - - - class MemoryList(KaitaiStruct): - """ - .. seealso:: - Source - https://msdn.microsoft.com/en-us/library/ms680387(v=vs.85).aspx - """ - SEQ_FIELDS = ["num_mem_ranges", "mem_ranges"] - 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_mem_ranges']['start'] = self._io.pos() - self.num_mem_ranges = self._io.read_u4le() - self._debug['num_mem_ranges']['end'] = self._io.pos() - self._debug['mem_ranges']['start'] = self._io.pos() - self.mem_ranges = [None] * (self.num_mem_ranges) - for i in range(self.num_mem_ranges): - if not 'arr' in self._debug['mem_ranges']: - self._debug['mem_ranges']['arr'] = [] - self._debug['mem_ranges']['arr'].append({'start': self._io.pos()}) - _t_mem_ranges = self._root.MemoryDescriptor(self._io, self, self._root) - _t_mem_ranges._read() - self.mem_ranges[i] = _t_mem_ranges - self._debug['mem_ranges']['arr'][i]['end'] = self._io.pos() - - self._debug['mem_ranges']['end'] = self._io.pos() - - - class MemoryDescriptor(KaitaiStruct): - """ - .. seealso:: - Source - https://msdn.microsoft.com/en-us/library/ms680384(v=vs.85).aspx - """ - SEQ_FIELDS = ["addr_memory_range", "memory"] - 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['addr_memory_range']['start'] = self._io.pos() - self.addr_memory_range = self._io.read_u8le() - self._debug['addr_memory_range']['end'] = self._io.pos() - self._debug['memory']['start'] = self._io.pos() - self.memory = self._root.LocationDescriptor(self._io, self, self._root) - self.memory._read() - self._debug['memory']['end'] = self._io.pos() - - - class ExceptionStream(KaitaiStruct): - """ - .. seealso:: - Source - https://msdn.microsoft.com/en-us/library/ms680368(v=vs.85).aspx - """ - SEQ_FIELDS = ["thread_id", "reserved", "exception_rec", "thread_context"] - 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['thread_id']['start'] = self._io.pos() - self.thread_id = self._io.read_u4le() - self._debug['thread_id']['end'] = self._io.pos() - self._debug['reserved']['start'] = self._io.pos() - self.reserved = self._io.read_u4le() - self._debug['reserved']['end'] = self._io.pos() - self._debug['exception_rec']['start'] = self._io.pos() - self.exception_rec = self._root.ExceptionRecord(self._io, self, self._root) - self.exception_rec._read() - self._debug['exception_rec']['end'] = self._io.pos() - self._debug['thread_context']['start'] = self._io.pos() - self.thread_context = self._root.LocationDescriptor(self._io, self, self._root) - self.thread_context._read() - self._debug['thread_context']['end'] = self._io.pos() - - - @property - def streams(self): - if hasattr(self, '_m_streams'): - return self._m_streams if hasattr(self, '_m_streams') else None - - _pos = self._io.pos() - self._io.seek(self.ofs_streams) - self._debug['_m_streams']['start'] = self._io.pos() - self._m_streams = [None] * (self.num_streams) - for i in range(self.num_streams): - if not 'arr' in self._debug['_m_streams']: - self._debug['_m_streams']['arr'] = [] - self._debug['_m_streams']['arr'].append({'start': self._io.pos()}) - _t__m_streams = self._root.Dir(self._io, self, self._root) - _t__m_streams._read() - self._m_streams[i] = _t__m_streams - self._debug['_m_streams']['arr'][i]['end'] = self._io.pos() - - self._debug['_m_streams']['end'] = self._io.pos() - self._io.seek(_pos) - return self._m_streams if hasattr(self, '_m_streams') else None - - |
