diff options
| author | Andrew Lamoureux <andrew@vector35.com> | 2019-03-18 17:34:09 -0400 |
|---|---|---|
| committer | Rusty Wagner <rusty@vector35.com> | 2019-03-20 13:00:17 -0400 |
| commit | 815f5485eb50eff3b483a6bcc0c3d3d11569f8ca (patch) | |
| tree | 5128136ac86971ed8729e8c19cb00bcf8f220acc /python/examples/kaitai/kaitai_struct_formats/filesystem | |
| parent | 8e9322ed8b5826dd62bc931d83e0828d42f7969f (diff) | |
kaitai: all formats compiled and included
Diffstat (limited to 'python/examples/kaitai/kaitai_struct_formats/filesystem')
14 files changed, 3864 insertions, 0 deletions
diff --git a/python/examples/kaitai/kaitai_struct_formats/filesystem/__init__.py b/python/examples/kaitai/kaitai_struct_formats/filesystem/__init__.py new file mode 100644 index 00000000..e69de29b --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/filesystem/__init__.py diff --git a/python/examples/kaitai/kaitai_struct_formats/filesystem/apm_partition_table.py b/python/examples/kaitai/kaitai_struct_formats/filesystem/apm_partition_table.py new file mode 100644 index 00000000..18b05757 --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/filesystem/apm_partition_table.py @@ -0,0 +1,194 @@ +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 ApmPartitionTable(KaitaiStruct): + """ + .. seealso:: + Specification taken from https://en.wikipedia.org/wiki/Apple_Partition_Map + """ + 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 PartitionEntry(KaitaiStruct): + SEQ_FIELDS = ["magic", "reserved_1", "number_of_partitions", "partition_start", "partition_size", "partition_name", "partition_type", "data_start", "data_size", "partition_status", "boot_code_start", "boot_code_size", "boot_loader_address", "reserved_2", "boot_code_entry", "reserved_3", "boot_code_cksum", "processor_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['magic']['start'] = self._io.pos() + self.magic = self._io.ensure_fixed_contents(b"\x50\x4D") + self._debug['magic']['end'] = self._io.pos() + self._debug['reserved_1']['start'] = self._io.pos() + self.reserved_1 = self._io.read_bytes(2) + self._debug['reserved_1']['end'] = self._io.pos() + self._debug['number_of_partitions']['start'] = self._io.pos() + self.number_of_partitions = self._io.read_u4be() + self._debug['number_of_partitions']['end'] = self._io.pos() + self._debug['partition_start']['start'] = self._io.pos() + self.partition_start = self._io.read_u4be() + self._debug['partition_start']['end'] = self._io.pos() + self._debug['partition_size']['start'] = self._io.pos() + self.partition_size = self._io.read_u4be() + self._debug['partition_size']['end'] = self._io.pos() + self._debug['partition_name']['start'] = self._io.pos() + self.partition_name = (KaitaiStream.bytes_terminate(self._io.read_bytes(32), 0, False)).decode(u"ascii") + self._debug['partition_name']['end'] = self._io.pos() + self._debug['partition_type']['start'] = self._io.pos() + self.partition_type = (KaitaiStream.bytes_terminate(self._io.read_bytes(32), 0, False)).decode(u"ascii") + self._debug['partition_type']['end'] = self._io.pos() + self._debug['data_start']['start'] = self._io.pos() + self.data_start = self._io.read_u4be() + self._debug['data_start']['end'] = self._io.pos() + self._debug['data_size']['start'] = self._io.pos() + self.data_size = self._io.read_u4be() + self._debug['data_size']['end'] = self._io.pos() + self._debug['partition_status']['start'] = self._io.pos() + self.partition_status = self._io.read_u4be() + self._debug['partition_status']['end'] = self._io.pos() + self._debug['boot_code_start']['start'] = self._io.pos() + self.boot_code_start = self._io.read_u4be() + self._debug['boot_code_start']['end'] = self._io.pos() + self._debug['boot_code_size']['start'] = self._io.pos() + self.boot_code_size = self._io.read_u4be() + self._debug['boot_code_size']['end'] = self._io.pos() + self._debug['boot_loader_address']['start'] = self._io.pos() + self.boot_loader_address = self._io.read_u4be() + self._debug['boot_loader_address']['end'] = self._io.pos() + self._debug['reserved_2']['start'] = self._io.pos() + self.reserved_2 = self._io.read_bytes(4) + self._debug['reserved_2']['end'] = self._io.pos() + self._debug['boot_code_entry']['start'] = self._io.pos() + self.boot_code_entry = self._io.read_u4be() + self._debug['boot_code_entry']['end'] = self._io.pos() + self._debug['reserved_3']['start'] = self._io.pos() + self.reserved_3 = self._io.read_bytes(4) + self._debug['reserved_3']['end'] = self._io.pos() + self._debug['boot_code_cksum']['start'] = self._io.pos() + self.boot_code_cksum = self._io.read_u4be() + self._debug['boot_code_cksum']['end'] = self._io.pos() + self._debug['processor_type']['start'] = self._io.pos() + self.processor_type = (KaitaiStream.bytes_terminate(self._io.read_bytes(16), 0, False)).decode(u"ascii") + self._debug['processor_type']['end'] = self._io.pos() + + @property + def partition(self): + if hasattr(self, '_m_partition'): + return self._m_partition if hasattr(self, '_m_partition') else None + + if (self.partition_status & 1) != 0: + io = self._root._io + _pos = io.pos() + io.seek((self.partition_start * self._root.sector_size)) + self._debug['_m_partition']['start'] = io.pos() + self._m_partition = io.read_bytes((self.partition_size * self._root.sector_size)) + self._debug['_m_partition']['end'] = io.pos() + io.seek(_pos) + + return self._m_partition if hasattr(self, '_m_partition') else None + + @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.data_start * self._root.sector_size)) + self._debug['_m_data']['start'] = io.pos() + self._m_data = io.read_bytes((self.data_size * self._root.sector_size)) + self._debug['_m_data']['end'] = io.pos() + io.seek(_pos) + return self._m_data if hasattr(self, '_m_data') else None + + @property + def boot_code(self): + if hasattr(self, '_m_boot_code'): + return self._m_boot_code if hasattr(self, '_m_boot_code') else None + + io = self._root._io + _pos = io.pos() + io.seek((self.boot_code_start * self._root.sector_size)) + self._debug['_m_boot_code']['start'] = io.pos() + self._m_boot_code = io.read_bytes(self.boot_code_size) + self._debug['_m_boot_code']['end'] = io.pos() + io.seek(_pos) + return self._m_boot_code if hasattr(self, '_m_boot_code') else None + + + @property + def sector_size(self): + """0x200 (512) bytes for disks, 0x1000 (4096) bytes is not supported by APM + 0x800 (2048) bytes for CDROM + """ + if hasattr(self, '_m_sector_size'): + return self._m_sector_size if hasattr(self, '_m_sector_size') else None + + self._m_sector_size = 512 + return self._m_sector_size if hasattr(self, '_m_sector_size') else None + + @property + def partition_lookup(self): + """Every partition entry contains the number of partition entries. + We parse the first entry, to know how many to parse, including the first one. + No logic is given what to do if other entries have a different number. + """ + if hasattr(self, '_m_partition_lookup'): + return self._m_partition_lookup if hasattr(self, '_m_partition_lookup') else None + + io = self._root._io + _pos = io.pos() + io.seek(self._root.sector_size) + self._debug['_m_partition_lookup']['start'] = io.pos() + self._raw__m_partition_lookup = io.read_bytes(self.sector_size) + io = KaitaiStream(BytesIO(self._raw__m_partition_lookup)) + self._m_partition_lookup = self._root.PartitionEntry(io, self, self._root) + self._m_partition_lookup._read() + self._debug['_m_partition_lookup']['end'] = io.pos() + io.seek(_pos) + return self._m_partition_lookup if hasattr(self, '_m_partition_lookup') else None + + @property + def partition_entries(self): + if hasattr(self, '_m_partition_entries'): + return self._m_partition_entries if hasattr(self, '_m_partition_entries') else None + + io = self._root._io + _pos = io.pos() + io.seek(self._root.sector_size) + self._debug['_m_partition_entries']['start'] = io.pos() + self._raw__m_partition_entries = [None] * (self._root.partition_lookup.number_of_partitions) + self._m_partition_entries = [None] * (self._root.partition_lookup.number_of_partitions) + for i in range(self._root.partition_lookup.number_of_partitions): + if not 'arr' in self._debug['_m_partition_entries']: + self._debug['_m_partition_entries']['arr'] = [] + self._debug['_m_partition_entries']['arr'].append({'start': io.pos()}) + self._raw__m_partition_entries[i] = io.read_bytes(self.sector_size) + io = KaitaiStream(BytesIO(self._raw__m_partition_entries[i])) + _t__m_partition_entries = self._root.PartitionEntry(io, self, self._root) + _t__m_partition_entries._read() + self._m_partition_entries[i] = _t__m_partition_entries + self._debug['_m_partition_entries']['arr'][i]['end'] = io.pos() + + self._debug['_m_partition_entries']['end'] = io.pos() + io.seek(_pos) + return self._m_partition_entries if hasattr(self, '_m_partition_entries') else None + + diff --git a/python/examples/kaitai/kaitai_struct_formats/filesystem/apple_single_double.py b/python/examples/kaitai/kaitai_struct_formats/filesystem/apple_single_double.py new file mode 100644 index 00000000..39a65abc --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/filesystem/apple_single_double.py @@ -0,0 +1,180 @@ +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 AppleSingleDouble(KaitaiStruct): + """AppleSingle and AppleDouble files are used by certain Mac + applications (e.g. Finder) to store Mac-specific file attributes on + filesystems that do not support that. + + Syntactically, both formats are the same, the only difference is how + they are being used: + + * AppleSingle means that only one file will be created on external + filesystem that will hold both the data (AKA "data fork" in Apple + terminology), and the attributes (AKA "resource fork"). + * AppleDouble means that two files will be created: a normal file + that keeps the data ("data fork") is kept separately from an + auxiliary file that contains attributes ("resource fork"), which + is kept with the same name, but starting with an extra dot and + underscore `._` to keep it hidden. + + In modern practice (Mac OS X), Finder only uses AppleDouble to keep + compatibility with other OSes, as virtually nobody outside of Mac + understands how to access data in AppleSingle container. + + .. seealso:: + Source - http://kaiser-edv.de/documents/AppleSingle_AppleDouble.pdf + """ + + class FileType(Enum): + apple_single = 333312 + apple_double = 333319 + SEQ_FIELDS = ["magic", "version", "reserved", "num_entries", "entries"] + def __init__(self, _io, _parent=None, _root=None): + self._io = _io + self._parent = _parent + self._root = _root if _root else self + self._debug = collections.defaultdict(dict) + + def _read(self): + self._debug['magic']['start'] = self._io.pos() + self.magic = KaitaiStream.resolve_enum(self._root.FileType, self._io.read_u4be()) + self._debug['magic']['end'] = self._io.pos() + self._debug['version']['start'] = self._io.pos() + self.version = self._io.read_u4be() + self._debug['version']['end'] = self._io.pos() + self._debug['reserved']['start'] = self._io.pos() + self.reserved = self._io.read_bytes(16) + self._debug['reserved']['end'] = self._io.pos() + self._debug['num_entries']['start'] = self._io.pos() + self.num_entries = self._io.read_u2be() + self._debug['num_entries']['end'] = self._io.pos() + self._debug['entries']['start'] = self._io.pos() + self.entries = [None] * (self.num_entries) + for i in range(self.num_entries): + if not 'arr' in self._debug['entries']: + self._debug['entries']['arr'] = [] + self._debug['entries']['arr'].append({'start': self._io.pos()}) + _t_entries = self._root.Entry(self._io, self, self._root) + _t_entries._read() + self.entries[i] = _t_entries + self._debug['entries']['arr'][i]['end'] = self._io.pos() + + self._debug['entries']['end'] = self._io.pos() + + class Entry(KaitaiStruct): + + class Types(Enum): + data_fork = 1 + resource_fork = 2 + real_name = 3 + comment = 4 + icon_bw = 5 + icon_color = 6 + file_dates_info = 8 + finder_info = 9 + macintosh_file_info = 10 + prodos_file_info = 11 + msdos_file_info = 12 + afp_short_name = 13 + afp_file_info = 14 + afp_directory_id = 15 + SEQ_FIELDS = ["type", "ofs_body", "len_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['type']['start'] = self._io.pos() + self.type = KaitaiStream.resolve_enum(self._root.Entry.Types, self._io.read_u4be()) + self._debug['type']['end'] = self._io.pos() + self._debug['ofs_body']['start'] = self._io.pos() + self.ofs_body = self._io.read_u4be() + self._debug['ofs_body']['end'] = self._io.pos() + self._debug['len_body']['start'] = self._io.pos() + self.len_body = self._io.read_u4be() + self._debug['len_body']['end'] = self._io.pos() + + @property + def body(self): + if hasattr(self, '_m_body'): + return self._m_body if hasattr(self, '_m_body') else None + + _pos = self._io.pos() + self._io.seek(self.ofs_body) + self._debug['_m_body']['start'] = self._io.pos() + _on = self.type + if _on == self._root.Entry.Types.finder_info: + self._raw__m_body = self._io.read_bytes(self.len_body) + io = KaitaiStream(BytesIO(self._raw__m_body)) + self._m_body = self._root.FinderInfo(io, self, self._root) + self._m_body._read() + else: + self._m_body = self._io.read_bytes(self.len_body) + self._debug['_m_body']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_body if hasattr(self, '_m_body') else None + + + class FinderInfo(KaitaiStruct): + """Information specific to Finder. + + .. seealso:: + older Inside Macintosh, Volume II page 84 or Volume IV page 104. + """ + SEQ_FIELDS = ["file_type", "file_creator", "flags", "location", "folder_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['file_type']['start'] = self._io.pos() + self.file_type = self._io.read_bytes(4) + self._debug['file_type']['end'] = self._io.pos() + self._debug['file_creator']['start'] = self._io.pos() + self.file_creator = self._io.read_bytes(4) + self._debug['file_creator']['end'] = self._io.pos() + self._debug['flags']['start'] = self._io.pos() + self.flags = self._io.read_u2be() + self._debug['flags']['end'] = self._io.pos() + self._debug['location']['start'] = self._io.pos() + self.location = self._root.Point(self._io, self, self._root) + self.location._read() + self._debug['location']['end'] = self._io.pos() + self._debug['folder_id']['start'] = self._io.pos() + self.folder_id = self._io.read_u2be() + self._debug['folder_id']['end'] = self._io.pos() + + + class Point(KaitaiStruct): + """Specifies 2D coordinate in QuickDraw grid.""" + SEQ_FIELDS = ["x", "y"] + 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['x']['start'] = self._io.pos() + self.x = self._io.read_u2be() + self._debug['x']['end'] = self._io.pos() + self._debug['y']['start'] = self._io.pos() + self.y = self._io.read_u2be() + self._debug['y']['end'] = self._io.pos() + + + diff --git a/python/examples/kaitai/kaitai_struct_formats/filesystem/cramfs.py b/python/examples/kaitai/kaitai_struct_formats/filesystem/cramfs.py new file mode 100644 index 00000000..bb8efdd1 --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/filesystem/cramfs.py @@ -0,0 +1,340 @@ +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 Cramfs(KaitaiStruct): + SEQ_FIELDS = ["super_block"] + 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['super_block']['start'] = self._io.pos() + self.super_block = self._root.SuperBlockStruct(self._io, self, self._root) + self.super_block._read() + self._debug['super_block']['end'] = self._io.pos() + + class SuperBlockStruct(KaitaiStruct): + SEQ_FIELDS = ["magic", "size", "flags", "future", "signature", "fsid", "name", "root"] + 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"\x45\x3D\xCD\x28") + self._debug['magic']['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['flags']['start'] = self._io.pos() + self.flags = self._io.read_u4le() + self._debug['flags']['end'] = self._io.pos() + self._debug['future']['start'] = self._io.pos() + self.future = self._io.read_u4le() + self._debug['future']['end'] = self._io.pos() + self._debug['signature']['start'] = self._io.pos() + self.signature = self._io.ensure_fixed_contents(b"\x43\x6F\x6D\x70\x72\x65\x73\x73\x65\x64\x20\x52\x4F\x4D\x46\x53") + self._debug['signature']['end'] = self._io.pos() + self._debug['fsid']['start'] = self._io.pos() + self.fsid = self._root.Info(self._io, self, self._root) + self.fsid._read() + self._debug['fsid']['end'] = self._io.pos() + self._debug['name']['start'] = self._io.pos() + self.name = (self._io.read_bytes(16)).decode(u"ASCII") + self._debug['name']['end'] = self._io.pos() + self._debug['root']['start'] = self._io.pos() + self.root = self._root.Inode(self._io, self, self._root) + self.root._read() + self._debug['root']['end'] = self._io.pos() + + @property + def flag_fsid_v2(self): + if hasattr(self, '_m_flag_fsid_v2'): + return self._m_flag_fsid_v2 if hasattr(self, '_m_flag_fsid_v2') else None + + self._m_flag_fsid_v2 = ((self.flags >> 0) & 1) + return self._m_flag_fsid_v2 if hasattr(self, '_m_flag_fsid_v2') else None + + @property + def flag_holes(self): + if hasattr(self, '_m_flag_holes'): + return self._m_flag_holes if hasattr(self, '_m_flag_holes') else None + + self._m_flag_holes = ((self.flags >> 8) & 1) + return self._m_flag_holes if hasattr(self, '_m_flag_holes') else None + + @property + def flag_wrong_signature(self): + if hasattr(self, '_m_flag_wrong_signature'): + return self._m_flag_wrong_signature if hasattr(self, '_m_flag_wrong_signature') else None + + self._m_flag_wrong_signature = ((self.flags >> 9) & 1) + return self._m_flag_wrong_signature if hasattr(self, '_m_flag_wrong_signature') else None + + @property + def flag_sorted_dirs(self): + if hasattr(self, '_m_flag_sorted_dirs'): + return self._m_flag_sorted_dirs if hasattr(self, '_m_flag_sorted_dirs') else None + + self._m_flag_sorted_dirs = ((self.flags >> 1) & 1) + return self._m_flag_sorted_dirs if hasattr(self, '_m_flag_sorted_dirs') else None + + @property + def flag_shifted_root_offset(self): + if hasattr(self, '_m_flag_shifted_root_offset'): + return self._m_flag_shifted_root_offset if hasattr(self, '_m_flag_shifted_root_offset') else None + + self._m_flag_shifted_root_offset = ((self.flags >> 10) & 1) + return self._m_flag_shifted_root_offset if hasattr(self, '_m_flag_shifted_root_offset') else None + + + class ChunkedDataInode(KaitaiStruct): + SEQ_FIELDS = ["block_end_index", "raw_blocks"] + 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['block_end_index']['start'] = self._io.pos() + self.block_end_index = [None] * (((self._parent.size + self._root.page_size) - 1) // self._root.page_size) + for i in range(((self._parent.size + self._root.page_size) - 1) // self._root.page_size): + if not 'arr' in self._debug['block_end_index']: + self._debug['block_end_index']['arr'] = [] + self._debug['block_end_index']['arr'].append({'start': self._io.pos()}) + self.block_end_index[i] = self._io.read_u4le() + self._debug['block_end_index']['arr'][i]['end'] = self._io.pos() + + self._debug['block_end_index']['end'] = self._io.pos() + self._debug['raw_blocks']['start'] = self._io.pos() + self.raw_blocks = self._io.read_bytes_full() + self._debug['raw_blocks']['end'] = self._io.pos() + + + class Inode(KaitaiStruct): + + class FileType(Enum): + fifo = 1 + chrdev = 2 + dir = 4 + blkdev = 6 + reg_file = 8 + symlink = 10 + socket = 12 + SEQ_FIELDS = ["mode", "uid", "size_gid", "namelen_offset", "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['mode']['start'] = self._io.pos() + self.mode = self._io.read_u2le() + self._debug['mode']['end'] = self._io.pos() + self._debug['uid']['start'] = self._io.pos() + self.uid = self._io.read_u2le() + self._debug['uid']['end'] = self._io.pos() + self._debug['size_gid']['start'] = self._io.pos() + self.size_gid = self._io.read_u4le() + self._debug['size_gid']['end'] = self._io.pos() + self._debug['namelen_offset']['start'] = self._io.pos() + self.namelen_offset = self._io.read_u4le() + self._debug['namelen_offset']['end'] = self._io.pos() + self._debug['name']['start'] = self._io.pos() + self.name = (self._io.read_bytes(self.namelen)).decode(u"utf-8") + self._debug['name']['end'] = self._io.pos() + + @property + def attr(self): + if hasattr(self, '_m_attr'): + return self._m_attr if hasattr(self, '_m_attr') else None + + self._m_attr = ((self.mode >> 9) & 7) + return self._m_attr if hasattr(self, '_m_attr') else None + + @property + def as_reg_file(self): + if hasattr(self, '_m_as_reg_file'): + return self._m_as_reg_file if hasattr(self, '_m_as_reg_file') else None + + io = self._root._io + _pos = io.pos() + io.seek(self.offset) + self._debug['_m_as_reg_file']['start'] = io.pos() + self._m_as_reg_file = self._root.ChunkedDataInode(io, self, self._root) + self._m_as_reg_file._read() + self._debug['_m_as_reg_file']['end'] = io.pos() + io.seek(_pos) + return self._m_as_reg_file if hasattr(self, '_m_as_reg_file') else None + + @property + def perm_u(self): + if hasattr(self, '_m_perm_u'): + return self._m_perm_u if hasattr(self, '_m_perm_u') else None + + self._m_perm_u = ((self.mode >> 6) & 7) + return self._m_perm_u if hasattr(self, '_m_perm_u') else None + + @property + def as_symlink(self): + if hasattr(self, '_m_as_symlink'): + return self._m_as_symlink if hasattr(self, '_m_as_symlink') else None + + io = self._root._io + _pos = io.pos() + io.seek(self.offset) + self._debug['_m_as_symlink']['start'] = io.pos() + self._m_as_symlink = self._root.ChunkedDataInode(io, self, self._root) + self._m_as_symlink._read() + self._debug['_m_as_symlink']['end'] = io.pos() + io.seek(_pos) + return self._m_as_symlink if hasattr(self, '_m_as_symlink') else None + + @property + def perm_o(self): + if hasattr(self, '_m_perm_o'): + return self._m_perm_o if hasattr(self, '_m_perm_o') else None + + self._m_perm_o = (self.mode & 7) + return self._m_perm_o if hasattr(self, '_m_perm_o') else None + + @property + def size(self): + if hasattr(self, '_m_size'): + return self._m_size if hasattr(self, '_m_size') else None + + self._m_size = (self.size_gid & 16777215) + return self._m_size if hasattr(self, '_m_size') else None + + @property + def gid(self): + if hasattr(self, '_m_gid'): + return self._m_gid if hasattr(self, '_m_gid') else None + + self._m_gid = (self.size_gid >> 24) + return self._m_gid if hasattr(self, '_m_gid') else None + + @property + def perm_g(self): + if hasattr(self, '_m_perm_g'): + return self._m_perm_g if hasattr(self, '_m_perm_g') else None + + self._m_perm_g = ((self.mode >> 3) & 7) + return self._m_perm_g if hasattr(self, '_m_perm_g') else None + + @property + def namelen(self): + if hasattr(self, '_m_namelen'): + return self._m_namelen if hasattr(self, '_m_namelen') else None + + self._m_namelen = ((self.namelen_offset & 63) << 2) + return self._m_namelen if hasattr(self, '_m_namelen') else None + + @property + def as_dir(self): + if hasattr(self, '_m_as_dir'): + return self._m_as_dir if hasattr(self, '_m_as_dir') else None + + io = self._root._io + _pos = io.pos() + io.seek(self.offset) + self._debug['_m_as_dir']['start'] = io.pos() + self._raw__m_as_dir = io.read_bytes(self.size) + io = KaitaiStream(BytesIO(self._raw__m_as_dir)) + self._m_as_dir = self._root.DirInode(io, self, self._root) + self._m_as_dir._read() + self._debug['_m_as_dir']['end'] = io.pos() + io.seek(_pos) + return self._m_as_dir if hasattr(self, '_m_as_dir') else None + + @property + def type(self): + if hasattr(self, '_m_type'): + return self._m_type if hasattr(self, '_m_type') else None + + self._m_type = KaitaiStream.resolve_enum(self._root.Inode.FileType, ((self.mode >> 12) & 15)) + return self._m_type if hasattr(self, '_m_type') else None + + @property + def offset(self): + if hasattr(self, '_m_offset'): + return self._m_offset if hasattr(self, '_m_offset') else None + + self._m_offset = (((self.namelen_offset >> 6) & 67108863) << 2) + return self._m_offset if hasattr(self, '_m_offset') else None + + + class DirInode(KaitaiStruct): + SEQ_FIELDS = ["children"] + 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): + if self._io.size() > 0: + self._debug['children']['start'] = self._io.pos() + self.children = [] + i = 0 + while not self._io.is_eof(): + if not 'arr' in self._debug['children']: + self._debug['children']['arr'] = [] + self._debug['children']['arr'].append({'start': self._io.pos()}) + _t_children = self._root.Inode(self._io, self, self._root) + _t_children._read() + self.children.append(_t_children) + self._debug['children']['arr'][len(self.children) - 1]['end'] = self._io.pos() + i += 1 + + self._debug['children']['end'] = self._io.pos() + + + + class Info(KaitaiStruct): + SEQ_FIELDS = ["crc", "edition", "blocks", "files"] + 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['crc']['start'] = self._io.pos() + self.crc = self._io.read_u4le() + self._debug['crc']['end'] = self._io.pos() + self._debug['edition']['start'] = self._io.pos() + self.edition = self._io.read_u4le() + self._debug['edition']['end'] = self._io.pos() + self._debug['blocks']['start'] = self._io.pos() + self.blocks = self._io.read_u4le() + self._debug['blocks']['end'] = self._io.pos() + self._debug['files']['start'] = self._io.pos() + self.files = self._io.read_u4le() + self._debug['files']['end'] = self._io.pos() + + + @property + def page_size(self): + if hasattr(self, '_m_page_size'): + return self._m_page_size if hasattr(self, '_m_page_size') else None + + self._m_page_size = 4096 + return self._m_page_size if hasattr(self, '_m_page_size') else None + + diff --git a/python/examples/kaitai/kaitai_struct_formats/filesystem/ext2.py b/python/examples/kaitai/kaitai_struct_formats/filesystem/ext2.py new file mode 100644 index 00000000..f912475b --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/filesystem/ext2.py @@ -0,0 +1,535 @@ +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 Ext2(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 SuperBlockStruct(KaitaiStruct): + + class StateEnum(Enum): + valid_fs = 1 + error_fs = 2 + + class ErrorsEnum(Enum): + act_continue = 1 + act_ro = 2 + act_panic = 3 + SEQ_FIELDS = ["inodes_count", "blocks_count", "r_blocks_count", "free_blocks_count", "free_inodes_count", "first_data_block", "log_block_size", "log_frag_size", "blocks_per_group", "frags_per_group", "inodes_per_group", "mtime", "wtime", "mnt_count", "max_mnt_count", "magic", "state", "errors", "minor_rev_level", "lastcheck", "checkinterval", "creator_os", "rev_level", "def_resuid", "def_resgid", "first_ino", "inode_size", "block_group_nr", "feature_compat", "feature_incompat", "feature_ro_compat", "uuid", "volume_name", "last_mounted", "algo_bitmap", "prealloc_blocks", "prealloc_dir_blocks", "padding1", "journal_uuid", "journal_inum", "journal_dev", "last_orphan", "hash_seed", "def_hash_version"] + 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['inodes_count']['start'] = self._io.pos() + self.inodes_count = self._io.read_u4le() + self._debug['inodes_count']['end'] = self._io.pos() + self._debug['blocks_count']['start'] = self._io.pos() + self.blocks_count = self._io.read_u4le() + self._debug['blocks_count']['end'] = self._io.pos() + self._debug['r_blocks_count']['start'] = self._io.pos() + self.r_blocks_count = self._io.read_u4le() + self._debug['r_blocks_count']['end'] = self._io.pos() + self._debug['free_blocks_count']['start'] = self._io.pos() + self.free_blocks_count = self._io.read_u4le() + self._debug['free_blocks_count']['end'] = self._io.pos() + self._debug['free_inodes_count']['start'] = self._io.pos() + self.free_inodes_count = self._io.read_u4le() + self._debug['free_inodes_count']['end'] = self._io.pos() + self._debug['first_data_block']['start'] = self._io.pos() + self.first_data_block = self._io.read_u4le() + self._debug['first_data_block']['end'] = self._io.pos() + self._debug['log_block_size']['start'] = self._io.pos() + self.log_block_size = self._io.read_u4le() + self._debug['log_block_size']['end'] = self._io.pos() + self._debug['log_frag_size']['start'] = self._io.pos() + self.log_frag_size = self._io.read_u4le() + self._debug['log_frag_size']['end'] = self._io.pos() + self._debug['blocks_per_group']['start'] = self._io.pos() + self.blocks_per_group = self._io.read_u4le() + self._debug['blocks_per_group']['end'] = self._io.pos() + self._debug['frags_per_group']['start'] = self._io.pos() + self.frags_per_group = self._io.read_u4le() + self._debug['frags_per_group']['end'] = self._io.pos() + self._debug['inodes_per_group']['start'] = self._io.pos() + self.inodes_per_group = self._io.read_u4le() + self._debug['inodes_per_group']['end'] = self._io.pos() + self._debug['mtime']['start'] = self._io.pos() + self.mtime = self._io.read_u4le() + self._debug['mtime']['end'] = self._io.pos() + self._debug['wtime']['start'] = self._io.pos() + self.wtime = self._io.read_u4le() + self._debug['wtime']['end'] = self._io.pos() + self._debug['mnt_count']['start'] = self._io.pos() + self.mnt_count = self._io.read_u2le() + self._debug['mnt_count']['end'] = self._io.pos() + self._debug['max_mnt_count']['start'] = self._io.pos() + self.max_mnt_count = self._io.read_u2le() + self._debug['max_mnt_count']['end'] = self._io.pos() + self._debug['magic']['start'] = self._io.pos() + self.magic = self._io.ensure_fixed_contents(b"\x53\xEF") + self._debug['magic']['end'] = self._io.pos() + self._debug['state']['start'] = self._io.pos() + self.state = KaitaiStream.resolve_enum(self._root.SuperBlockStruct.StateEnum, self._io.read_u2le()) + self._debug['state']['end'] = self._io.pos() + self._debug['errors']['start'] = self._io.pos() + self.errors = KaitaiStream.resolve_enum(self._root.SuperBlockStruct.ErrorsEnum, self._io.read_u2le()) + self._debug['errors']['end'] = self._io.pos() + self._debug['minor_rev_level']['start'] = self._io.pos() + self.minor_rev_level = self._io.read_u2le() + self._debug['minor_rev_level']['end'] = self._io.pos() + self._debug['lastcheck']['start'] = self._io.pos() + self.lastcheck = self._io.read_u4le() + self._debug['lastcheck']['end'] = self._io.pos() + self._debug['checkinterval']['start'] = self._io.pos() + self.checkinterval = self._io.read_u4le() + self._debug['checkinterval']['end'] = self._io.pos() + self._debug['creator_os']['start'] = self._io.pos() + self.creator_os = self._io.read_u4le() + self._debug['creator_os']['end'] = self._io.pos() + self._debug['rev_level']['start'] = self._io.pos() + self.rev_level = self._io.read_u4le() + self._debug['rev_level']['end'] = self._io.pos() + self._debug['def_resuid']['start'] = self._io.pos() + self.def_resuid = self._io.read_u2le() + self._debug['def_resuid']['end'] = self._io.pos() + self._debug['def_resgid']['start'] = self._io.pos() + self.def_resgid = self._io.read_u2le() + self._debug['def_resgid']['end'] = self._io.pos() + self._debug['first_ino']['start'] = self._io.pos() + self.first_ino = self._io.read_u4le() + self._debug['first_ino']['end'] = self._io.pos() + self._debug['inode_size']['start'] = self._io.pos() + self.inode_size = self._io.read_u2le() + self._debug['inode_size']['end'] = self._io.pos() + self._debug['block_group_nr']['start'] = self._io.pos() + self.block_group_nr = self._io.read_u2le() + self._debug['block_group_nr']['end'] = self._io.pos() + self._debug['feature_compat']['start'] = self._io.pos() + self.feature_compat = self._io.read_u4le() + self._debug['feature_compat']['end'] = self._io.pos() + self._debug['feature_incompat']['start'] = self._io.pos() + self.feature_incompat = self._io.read_u4le() + self._debug['feature_incompat']['end'] = self._io.pos() + self._debug['feature_ro_compat']['start'] = self._io.pos() + self.feature_ro_compat = self._io.read_u4le() + self._debug['feature_ro_compat']['end'] = self._io.pos() + self._debug['uuid']['start'] = self._io.pos() + self.uuid = self._io.read_bytes(16) + self._debug['uuid']['end'] = self._io.pos() + self._debug['volume_name']['start'] = self._io.pos() + self.volume_name = self._io.read_bytes(16) + self._debug['volume_name']['end'] = self._io.pos() + self._debug['last_mounted']['start'] = self._io.pos() + self.last_mounted = self._io.read_bytes(64) + self._debug['last_mounted']['end'] = self._io.pos() + self._debug['algo_bitmap']['start'] = self._io.pos() + self.algo_bitmap = self._io.read_u4le() + self._debug['algo_bitmap']['end'] = self._io.pos() + self._debug['prealloc_blocks']['start'] = self._io.pos() + self.prealloc_blocks = self._io.read_u1() + self._debug['prealloc_blocks']['end'] = self._io.pos() + self._debug['prealloc_dir_blocks']['start'] = self._io.pos() + self.prealloc_dir_blocks = self._io.read_u1() + self._debug['prealloc_dir_blocks']['end'] = self._io.pos() + self._debug['padding1']['start'] = self._io.pos() + self.padding1 = self._io.read_bytes(2) + self._debug['padding1']['end'] = self._io.pos() + self._debug['journal_uuid']['start'] = self._io.pos() + self.journal_uuid = self._io.read_bytes(16) + self._debug['journal_uuid']['end'] = self._io.pos() + self._debug['journal_inum']['start'] = self._io.pos() + self.journal_inum = self._io.read_u4le() + self._debug['journal_inum']['end'] = self._io.pos() + self._debug['journal_dev']['start'] = self._io.pos() + self.journal_dev = self._io.read_u4le() + self._debug['journal_dev']['end'] = self._io.pos() + self._debug['last_orphan']['start'] = self._io.pos() + self.last_orphan = self._io.read_u4le() + self._debug['last_orphan']['end'] = self._io.pos() + self._debug['hash_seed']['start'] = self._io.pos() + self.hash_seed = [None] * (4) + for i in range(4): + if not 'arr' in self._debug['hash_seed']: + self._debug['hash_seed']['arr'] = [] + self._debug['hash_seed']['arr'].append({'start': self._io.pos()}) + self.hash_seed[i] = self._io.read_u4le() + self._debug['hash_seed']['arr'][i]['end'] = self._io.pos() + + self._debug['hash_seed']['end'] = self._io.pos() + self._debug['def_hash_version']['start'] = self._io.pos() + self.def_hash_version = self._io.read_u1() + self._debug['def_hash_version']['end'] = self._io.pos() + + @property + def block_size(self): + if hasattr(self, '_m_block_size'): + return self._m_block_size if hasattr(self, '_m_block_size') else None + + self._m_block_size = (1024 << self.log_block_size) + return self._m_block_size if hasattr(self, '_m_block_size') else None + + @property + def block_group_count(self): + if hasattr(self, '_m_block_group_count'): + return self._m_block_group_count if hasattr(self, '_m_block_group_count') else None + + self._m_block_group_count = self.blocks_count // self.blocks_per_group + return self._m_block_group_count if hasattr(self, '_m_block_group_count') else None + + + class DirEntry(KaitaiStruct): + + class FileTypeEnum(Enum): + unknown = 0 + reg_file = 1 + dir = 2 + chrdev = 3 + blkdev = 4 + fifo = 5 + sock = 6 + symlink = 7 + SEQ_FIELDS = ["inode_ptr", "rec_len", "name_len", "file_type", "name", "padding"] + 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['inode_ptr']['start'] = self._io.pos() + self.inode_ptr = self._io.read_u4le() + self._debug['inode_ptr']['end'] = self._io.pos() + self._debug['rec_len']['start'] = self._io.pos() + self.rec_len = self._io.read_u2le() + self._debug['rec_len']['end'] = self._io.pos() + self._debug['name_len']['start'] = self._io.pos() + self.name_len = self._io.read_u1() + self._debug['name_len']['end'] = self._io.pos() + self._debug['file_type']['start'] = self._io.pos() + self.file_type = KaitaiStream.resolve_enum(self._root.DirEntry.FileTypeEnum, self._io.read_u1()) + self._debug['file_type']['end'] = self._io.pos() + self._debug['name']['start'] = self._io.pos() + self.name = (self._io.read_bytes(self.name_len)).decode(u"UTF-8") + self._debug['name']['end'] = self._io.pos() + self._debug['padding']['start'] = self._io.pos() + self.padding = self._io.read_bytes(((self.rec_len - self.name_len) - 8)) + self._debug['padding']['end'] = self._io.pos() + + @property + def inode(self): + if hasattr(self, '_m_inode'): + return self._m_inode if hasattr(self, '_m_inode') else None + + self._m_inode = self._root.bg1.block_groups[(self.inode_ptr - 1) // self._root.bg1.super_block.inodes_per_group].inodes[((self.inode_ptr - 1) % self._root.bg1.super_block.inodes_per_group)] + return self._m_inode if hasattr(self, '_m_inode') else None + + + class Inode(KaitaiStruct): + SEQ_FIELDS = ["mode", "uid", "size", "atime", "ctime", "mtime", "dtime", "gid", "links_count", "blocks", "flags", "osd1", "block", "generation", "file_acl", "dir_acl", "faddr", "osd2"] + 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['mode']['start'] = self._io.pos() + self.mode = self._io.read_u2le() + self._debug['mode']['end'] = self._io.pos() + self._debug['uid']['start'] = self._io.pos() + self.uid = self._io.read_u2le() + self._debug['uid']['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['atime']['start'] = self._io.pos() + self.atime = self._io.read_u4le() + self._debug['atime']['end'] = self._io.pos() + self._debug['ctime']['start'] = self._io.pos() + self.ctime = self._io.read_u4le() + self._debug['ctime']['end'] = self._io.pos() + self._debug['mtime']['start'] = self._io.pos() + self.mtime = self._io.read_u4le() + self._debug['mtime']['end'] = self._io.pos() + self._debug['dtime']['start'] = self._io.pos() + self.dtime = self._io.read_u4le() + self._debug['dtime']['end'] = self._io.pos() + self._debug['gid']['start'] = self._io.pos() + self.gid = self._io.read_u2le() + self._debug['gid']['end'] = self._io.pos() + self._debug['links_count']['start'] = self._io.pos() + self.links_count = self._io.read_u2le() + self._debug['links_count']['end'] = self._io.pos() + self._debug['blocks']['start'] = self._io.pos() + self.blocks = self._io.read_u4le() + self._debug['blocks']['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['osd1']['start'] = self._io.pos() + self.osd1 = self._io.read_u4le() + self._debug['osd1']['end'] = self._io.pos() + self._debug['block']['start'] = self._io.pos() + self.block = [None] * (15) + for i in range(15): + if not 'arr' in self._debug['block']: + self._debug['block']['arr'] = [] + self._debug['block']['arr'].append({'start': self._io.pos()}) + _t_block = self._root.BlockPtr(self._io, self, self._root) + _t_block._read() + self.block[i] = _t_block + self._debug['block']['arr'][i]['end'] = self._io.pos() + + self._debug['block']['end'] = self._io.pos() + self._debug['generation']['start'] = self._io.pos() + self.generation = self._io.read_u4le() + self._debug['generation']['end'] = self._io.pos() + self._debug['file_acl']['start'] = self._io.pos() + self.file_acl = self._io.read_u4le() + self._debug['file_acl']['end'] = self._io.pos() + self._debug['dir_acl']['start'] = self._io.pos() + self.dir_acl = self._io.read_u4le() + self._debug['dir_acl']['end'] = self._io.pos() + self._debug['faddr']['start'] = self._io.pos() + self.faddr = self._io.read_u4le() + self._debug['faddr']['end'] = self._io.pos() + self._debug['osd2']['start'] = self._io.pos() + self.osd2 = self._io.read_bytes(12) + self._debug['osd2']['end'] = self._io.pos() + + @property + def as_dir(self): + if hasattr(self, '_m_as_dir'): + return self._m_as_dir if hasattr(self, '_m_as_dir') else None + + io = self.block[0].body._io + _pos = io.pos() + io.seek(0) + self._debug['_m_as_dir']['start'] = io.pos() + self._m_as_dir = self._root.Dir(io, self, self._root) + self._m_as_dir._read() + self._debug['_m_as_dir']['end'] = io.pos() + io.seek(_pos) + return self._m_as_dir if hasattr(self, '_m_as_dir') else None + + + class BlockPtr(KaitaiStruct): + SEQ_FIELDS = ["ptr"] + 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['ptr']['start'] = self._io.pos() + self.ptr = self._io.read_u4le() + self._debug['ptr']['end'] = self._io.pos() + + @property + def body(self): + if hasattr(self, '_m_body'): + return self._m_body if hasattr(self, '_m_body') else None + + _pos = self._io.pos() + self._io.seek((self.ptr * self._root.bg1.super_block.block_size)) + self._debug['_m_body']['start'] = self._io.pos() + self._raw__m_body = self._io.read_bytes(self._root.bg1.super_block.block_size) + io = KaitaiStream(BytesIO(self._raw__m_body)) + self._m_body = self._root.RawBlock(io, self, self._root) + self._m_body._read() + self._debug['_m_body']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_body if hasattr(self, '_m_body') else None + + + class Dir(KaitaiStruct): + SEQ_FIELDS = ["entries"] + def __init__(self, _io, _parent=None, _root=None): + self._io = _io + self._parent = _parent + self._root = _root if _root else self + self._debug = collections.defaultdict(dict) + + def _read(self): + self._debug['entries']['start'] = self._io.pos() + self.entries = [] + i = 0 + while not self._io.is_eof(): + if not 'arr' in self._debug['entries']: + self._debug['entries']['arr'] = [] + self._debug['entries']['arr'].append({'start': self._io.pos()}) + _t_entries = self._root.DirEntry(self._io, self, self._root) + _t_entries._read() + self.entries.append(_t_entries) + self._debug['entries']['arr'][len(self.entries) - 1]['end'] = self._io.pos() + i += 1 + + self._debug['entries']['end'] = self._io.pos() + + + class BlockGroup(KaitaiStruct): + SEQ_FIELDS = ["super_block", "block_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['super_block']['start'] = self._io.pos() + self._raw_super_block = self._io.read_bytes(1024) + io = KaitaiStream(BytesIO(self._raw_super_block)) + self.super_block = self._root.SuperBlockStruct(io, self, self._root) + self.super_block._read() + self._debug['super_block']['end'] = self._io.pos() + self._debug['block_groups']['start'] = self._io.pos() + self.block_groups = [None] * (self.super_block.block_group_count) + for i in range(self.super_block.block_group_count): + if not 'arr' in self._debug['block_groups']: + self._debug['block_groups']['arr'] = [] + self._debug['block_groups']['arr'].append({'start': self._io.pos()}) + _t_block_groups = self._root.Bgd(self._io, self, self._root) + _t_block_groups._read() + self.block_groups[i] = _t_block_groups + self._debug['block_groups']['arr'][i]['end'] = self._io.pos() + + self._debug['block_groups']['end'] = self._io.pos() + + + class Bgd(KaitaiStruct): + SEQ_FIELDS = ["block_bitmap_block", "inode_bitmap_block", "inode_table_block", "free_blocks_count", "free_inodes_count", "used_dirs_count", "pad_reserved"] + 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['block_bitmap_block']['start'] = self._io.pos() + self.block_bitmap_block = self._io.read_u4le() + self._debug['block_bitmap_block']['end'] = self._io.pos() + self._debug['inode_bitmap_block']['start'] = self._io.pos() + self.inode_bitmap_block = self._io.read_u4le() + self._debug['inode_bitmap_block']['end'] = self._io.pos() + self._debug['inode_table_block']['start'] = self._io.pos() + self.inode_table_block = self._io.read_u4le() + self._debug['inode_table_block']['end'] = self._io.pos() + self._debug['free_blocks_count']['start'] = self._io.pos() + self.free_blocks_count = self._io.read_u2le() + self._debug['free_blocks_count']['end'] = self._io.pos() + self._debug['free_inodes_count']['start'] = self._io.pos() + self.free_inodes_count = self._io.read_u2le() + self._debug['free_inodes_count']['end'] = self._io.pos() + self._debug['used_dirs_count']['start'] = self._io.pos() + self.used_dirs_count = self._io.read_u2le() + self._debug['used_dirs_count']['end'] = self._io.pos() + self._debug['pad_reserved']['start'] = self._io.pos() + self.pad_reserved = self._io.read_bytes((2 + 12)) + self._debug['pad_reserved']['end'] = self._io.pos() + + @property + def block_bitmap(self): + if hasattr(self, '_m_block_bitmap'): + return self._m_block_bitmap if hasattr(self, '_m_block_bitmap') else None + + _pos = self._io.pos() + self._io.seek((self.block_bitmap_block * self._root.bg1.super_block.block_size)) + self._debug['_m_block_bitmap']['start'] = self._io.pos() + self._m_block_bitmap = self._io.read_bytes(1024) + self._debug['_m_block_bitmap']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_block_bitmap if hasattr(self, '_m_block_bitmap') else None + + @property + def inode_bitmap(self): + if hasattr(self, '_m_inode_bitmap'): + return self._m_inode_bitmap if hasattr(self, '_m_inode_bitmap') else None + + _pos = self._io.pos() + self._io.seek((self.inode_bitmap_block * self._root.bg1.super_block.block_size)) + self._debug['_m_inode_bitmap']['start'] = self._io.pos() + self._m_inode_bitmap = self._io.read_bytes(1024) + self._debug['_m_inode_bitmap']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_inode_bitmap if hasattr(self, '_m_inode_bitmap') else None + + @property + def inodes(self): + if hasattr(self, '_m_inodes'): + return self._m_inodes if hasattr(self, '_m_inodes') else None + + _pos = self._io.pos() + self._io.seek((self.inode_table_block * self._root.bg1.super_block.block_size)) + self._debug['_m_inodes']['start'] = self._io.pos() + self._m_inodes = [None] * (self._root.bg1.super_block.inodes_per_group) + for i in range(self._root.bg1.super_block.inodes_per_group): + if not 'arr' in self._debug['_m_inodes']: + self._debug['_m_inodes']['arr'] = [] + self._debug['_m_inodes']['arr'].append({'start': self._io.pos()}) + _t__m_inodes = self._root.Inode(self._io, self, self._root) + _t__m_inodes._read() + self._m_inodes[i] = _t__m_inodes + self._debug['_m_inodes']['arr'][i]['end'] = self._io.pos() + + self._debug['_m_inodes']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_inodes if hasattr(self, '_m_inodes') else None + + + class RawBlock(KaitaiStruct): + SEQ_FIELDS = ["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['body']['start'] = self._io.pos() + self.body = self._io.read_bytes(self._root.bg1.super_block.block_size) + self._debug['body']['end'] = self._io.pos() + + + @property + def bg1(self): + if hasattr(self, '_m_bg1'): + return self._m_bg1 if hasattr(self, '_m_bg1') else None + + _pos = self._io.pos() + self._io.seek(1024) + self._debug['_m_bg1']['start'] = self._io.pos() + self._m_bg1 = self._root.BlockGroup(self._io, self, self._root) + self._m_bg1._read() + self._debug['_m_bg1']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_bg1 if hasattr(self, '_m_bg1') else None + + @property + def root_dir(self): + if hasattr(self, '_m_root_dir'): + return self._m_root_dir if hasattr(self, '_m_root_dir') else None + + self._m_root_dir = self.bg1.block_groups[0].inodes[1].as_dir + return self._m_root_dir if hasattr(self, '_m_root_dir') else None + + diff --git a/python/examples/kaitai/kaitai_struct_formats/filesystem/gpt_partition_table.py b/python/examples/kaitai/kaitai_struct_formats/filesystem/gpt_partition_table.py new file mode 100644 index 00000000..49a56371 --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/filesystem/gpt_partition_table.py @@ -0,0 +1,173 @@ +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 GptPartitionTable(KaitaiStruct): + """ + .. seealso:: + Specification taken from https://en.wikipedia.org/wiki/GUID_Partition_Table + """ + 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 PartitionEntry(KaitaiStruct): + SEQ_FIELDS = ["type_guid", "guid", "first_lba", "last_lba", "attributes", "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['type_guid']['start'] = self._io.pos() + self.type_guid = self._io.read_bytes(16) + self._debug['type_guid']['end'] = self._io.pos() + self._debug['guid']['start'] = self._io.pos() + self.guid = self._io.read_bytes(16) + self._debug['guid']['end'] = self._io.pos() + self._debug['first_lba']['start'] = self._io.pos() + self.first_lba = self._io.read_u8le() + self._debug['first_lba']['end'] = self._io.pos() + self._debug['last_lba']['start'] = self._io.pos() + self.last_lba = self._io.read_u8le() + self._debug['last_lba']['end'] = self._io.pos() + self._debug['attributes']['start'] = self._io.pos() + self.attributes = self._io.read_u8le() + self._debug['attributes']['end'] = self._io.pos() + self._debug['name']['start'] = self._io.pos() + self.name = (self._io.read_bytes(72)).decode(u"UTF-16LE") + self._debug['name']['end'] = self._io.pos() + + + class PartitionHeader(KaitaiStruct): + SEQ_FIELDS = ["signature", "revision", "header_size", "crc32_header", "reserved", "current_lba", "backup_lba", "first_usable_lba", "last_usable_lba", "disk_guid", "entries_start", "entries_count", "entries_size", "crc32_array"] + 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['signature']['start'] = self._io.pos() + self.signature = self._io.ensure_fixed_contents(b"\x45\x46\x49\x20\x50\x41\x52\x54") + self._debug['signature']['end'] = self._io.pos() + self._debug['revision']['start'] = self._io.pos() + self.revision = self._io.read_u4le() + self._debug['revision']['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['crc32_header']['start'] = self._io.pos() + self.crc32_header = self._io.read_u4le() + self._debug['crc32_header']['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['current_lba']['start'] = self._io.pos() + self.current_lba = self._io.read_u8le() + self._debug['current_lba']['end'] = self._io.pos() + self._debug['backup_lba']['start'] = self._io.pos() + self.backup_lba = self._io.read_u8le() + self._debug['backup_lba']['end'] = self._io.pos() + self._debug['first_usable_lba']['start'] = self._io.pos() + self.first_usable_lba = self._io.read_u8le() + self._debug['first_usable_lba']['end'] = self._io.pos() + self._debug['last_usable_lba']['start'] = self._io.pos() + self.last_usable_lba = self._io.read_u8le() + self._debug['last_usable_lba']['end'] = self._io.pos() + self._debug['disk_guid']['start'] = self._io.pos() + self.disk_guid = self._io.read_bytes(16) + self._debug['disk_guid']['end'] = self._io.pos() + self._debug['entries_start']['start'] = self._io.pos() + self.entries_start = self._io.read_u8le() + self._debug['entries_start']['end'] = self._io.pos() + self._debug['entries_count']['start'] = self._io.pos() + self.entries_count = self._io.read_u4le() + self._debug['entries_count']['end'] = self._io.pos() + self._debug['entries_size']['start'] = self._io.pos() + self.entries_size = self._io.read_u4le() + self._debug['entries_size']['end'] = self._io.pos() + self._debug['crc32_array']['start'] = self._io.pos() + self.crc32_array = self._io.read_u4le() + self._debug['crc32_array']['end'] = self._io.pos() + + @property + def entries(self): + if hasattr(self, '_m_entries'): + return self._m_entries if hasattr(self, '_m_entries') else None + + io = self._root._io + _pos = io.pos() + io.seek((self.entries_start * self._root.sector_size)) + self._debug['_m_entries']['start'] = io.pos() + self._raw__m_entries = [None] * (self.entries_count) + self._m_entries = [None] * (self.entries_count) + for i in range(self.entries_count): + if not 'arr' in self._debug['_m_entries']: + self._debug['_m_entries']['arr'] = [] + self._debug['_m_entries']['arr'].append({'start': io.pos()}) + self._raw__m_entries[i] = io.read_bytes(self.entries_size) + io = KaitaiStream(BytesIO(self._raw__m_entries[i])) + _t__m_entries = self._root.PartitionEntry(io, self, self._root) + _t__m_entries._read() + self._m_entries[i] = _t__m_entries + self._debug['_m_entries']['arr'][i]['end'] = io.pos() + + self._debug['_m_entries']['end'] = io.pos() + io.seek(_pos) + return self._m_entries if hasattr(self, '_m_entries') else None + + + @property + def sector_size(self): + if hasattr(self, '_m_sector_size'): + return self._m_sector_size if hasattr(self, '_m_sector_size') else None + + self._m_sector_size = 512 + return self._m_sector_size if hasattr(self, '_m_sector_size') else None + + @property + def primary(self): + if hasattr(self, '_m_primary'): + return self._m_primary if hasattr(self, '_m_primary') else None + + io = self._root._io + _pos = io.pos() + io.seek(self._root.sector_size) + self._debug['_m_primary']['start'] = io.pos() + self._m_primary = self._root.PartitionHeader(io, self, self._root) + self._m_primary._read() + self._debug['_m_primary']['end'] = io.pos() + io.seek(_pos) + return self._m_primary if hasattr(self, '_m_primary') else None + + @property + def backup(self): + if hasattr(self, '_m_backup'): + return self._m_backup if hasattr(self, '_m_backup') else None + + io = self._root._io + _pos = io.pos() + io.seek((self._io.size() - self._root.sector_size)) + self._debug['_m_backup']['start'] = io.pos() + self._m_backup = self._root.PartitionHeader(io, self, self._root) + self._m_backup._read() + self._debug['_m_backup']['end'] = io.pos() + io.seek(_pos) + return self._m_backup if hasattr(self, '_m_backup') else None + + diff --git a/python/examples/kaitai/kaitai_struct_formats/filesystem/iso9660.py b/python/examples/kaitai/kaitai_struct_formats/filesystem/iso9660.py new file mode 100644 index 00000000..2ed28663 --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/filesystem/iso9660.py @@ -0,0 +1,536 @@ +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 Iso9660(KaitaiStruct): + """ISO9660 is standard filesystem used on read-only optical discs + (mostly CD-ROM). The standard was based on earlier High Sierra + Format (HSF), proposed for CD-ROMs in 1985, and, after several + revisions, it was accepted as ISO9960:1998. + + The format emphasizes portability (thus having pretty minimal + features and very conservative file names standards) and sequential + access (which favors disc devices with relatively slow rotation + speed). + """ + 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 VolDescPrimary(KaitaiStruct): + """ + .. seealso:: + Source - http://wiki.osdev.org/ISO_9660#The_Primary_Volume_Descriptor + """ + SEQ_FIELDS = ["unused1", "system_id", "volume_id", "unused2", "vol_space_size", "unused3", "vol_set_size", "vol_seq_num", "logical_block_size", "path_table_size", "lba_path_table_le", "lba_opt_path_table_le", "lba_path_table_be", "lba_opt_path_table_be", "root_dir", "vol_set_id", "publisher_id", "data_preparer_id", "application_id", "copyright_file_id", "abstract_file_id", "bibliographic_file_id", "vol_create_datetime", "vol_mod_datetime", "vol_expire_datetime", "vol_effective_datetime", "file_structure_version", "unused4", "application_area"] + 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['unused1']['start'] = self._io.pos() + self.unused1 = self._io.ensure_fixed_contents(b"\x00") + self._debug['unused1']['end'] = self._io.pos() + self._debug['system_id']['start'] = self._io.pos() + self.system_id = (self._io.read_bytes(32)).decode(u"UTF-8") + self._debug['system_id']['end'] = self._io.pos() + self._debug['volume_id']['start'] = self._io.pos() + self.volume_id = (self._io.read_bytes(32)).decode(u"UTF-8") + self._debug['volume_id']['end'] = self._io.pos() + self._debug['unused2']['start'] = self._io.pos() + self.unused2 = self._io.ensure_fixed_contents(b"\x00\x00\x00\x00\x00\x00\x00\x00") + self._debug['unused2']['end'] = self._io.pos() + self._debug['vol_space_size']['start'] = self._io.pos() + self.vol_space_size = self._root.U4bi(self._io, self, self._root) + self.vol_space_size._read() + self._debug['vol_space_size']['end'] = self._io.pos() + self._debug['unused3']['start'] = self._io.pos() + self.unused3 = self._io.ensure_fixed_contents(b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00") + self._debug['unused3']['end'] = self._io.pos() + self._debug['vol_set_size']['start'] = self._io.pos() + self.vol_set_size = self._root.U2bi(self._io, self, self._root) + self.vol_set_size._read() + self._debug['vol_set_size']['end'] = self._io.pos() + self._debug['vol_seq_num']['start'] = self._io.pos() + self.vol_seq_num = self._root.U2bi(self._io, self, self._root) + self.vol_seq_num._read() + self._debug['vol_seq_num']['end'] = self._io.pos() + self._debug['logical_block_size']['start'] = self._io.pos() + self.logical_block_size = self._root.U2bi(self._io, self, self._root) + self.logical_block_size._read() + self._debug['logical_block_size']['end'] = self._io.pos() + self._debug['path_table_size']['start'] = self._io.pos() + self.path_table_size = self._root.U4bi(self._io, self, self._root) + self.path_table_size._read() + self._debug['path_table_size']['end'] = self._io.pos() + self._debug['lba_path_table_le']['start'] = self._io.pos() + self.lba_path_table_le = self._io.read_u4le() + self._debug['lba_path_table_le']['end'] = self._io.pos() + self._debug['lba_opt_path_table_le']['start'] = self._io.pos() + self.lba_opt_path_table_le = self._io.read_u4le() + self._debug['lba_opt_path_table_le']['end'] = self._io.pos() + self._debug['lba_path_table_be']['start'] = self._io.pos() + self.lba_path_table_be = self._io.read_u4be() + self._debug['lba_path_table_be']['end'] = self._io.pos() + self._debug['lba_opt_path_table_be']['start'] = self._io.pos() + self.lba_opt_path_table_be = self._io.read_u4be() + self._debug['lba_opt_path_table_be']['end'] = self._io.pos() + self._debug['root_dir']['start'] = self._io.pos() + self._raw_root_dir = self._io.read_bytes(34) + io = KaitaiStream(BytesIO(self._raw_root_dir)) + self.root_dir = self._root.DirEntry(io, self, self._root) + self.root_dir._read() + self._debug['root_dir']['end'] = self._io.pos() + self._debug['vol_set_id']['start'] = self._io.pos() + self.vol_set_id = (self._io.read_bytes(128)).decode(u"UTF-8") + self._debug['vol_set_id']['end'] = self._io.pos() + self._debug['publisher_id']['start'] = self._io.pos() + self.publisher_id = (self._io.read_bytes(128)).decode(u"UTF-8") + self._debug['publisher_id']['end'] = self._io.pos() + self._debug['data_preparer_id']['start'] = self._io.pos() + self.data_preparer_id = (self._io.read_bytes(128)).decode(u"UTF-8") + self._debug['data_preparer_id']['end'] = self._io.pos() + self._debug['application_id']['start'] = self._io.pos() + self.application_id = (self._io.read_bytes(128)).decode(u"UTF-8") + self._debug['application_id']['end'] = self._io.pos() + self._debug['copyright_file_id']['start'] = self._io.pos() + self.copyright_file_id = (self._io.read_bytes(38)).decode(u"UTF-8") + self._debug['copyright_file_id']['end'] = self._io.pos() + self._debug['abstract_file_id']['start'] = self._io.pos() + self.abstract_file_id = (self._io.read_bytes(36)).decode(u"UTF-8") + self._debug['abstract_file_id']['end'] = self._io.pos() + self._debug['bibliographic_file_id']['start'] = self._io.pos() + self.bibliographic_file_id = (self._io.read_bytes(37)).decode(u"UTF-8") + self._debug['bibliographic_file_id']['end'] = self._io.pos() + self._debug['vol_create_datetime']['start'] = self._io.pos() + self.vol_create_datetime = self._root.DecDatetime(self._io, self, self._root) + self.vol_create_datetime._read() + self._debug['vol_create_datetime']['end'] = self._io.pos() + self._debug['vol_mod_datetime']['start'] = self._io.pos() + self.vol_mod_datetime = self._root.DecDatetime(self._io, self, self._root) + self.vol_mod_datetime._read() + self._debug['vol_mod_datetime']['end'] = self._io.pos() + self._debug['vol_expire_datetime']['start'] = self._io.pos() + self.vol_expire_datetime = self._root.DecDatetime(self._io, self, self._root) + self.vol_expire_datetime._read() + self._debug['vol_expire_datetime']['end'] = self._io.pos() + self._debug['vol_effective_datetime']['start'] = self._io.pos() + self.vol_effective_datetime = self._root.DecDatetime(self._io, self, self._root) + self.vol_effective_datetime._read() + self._debug['vol_effective_datetime']['end'] = self._io.pos() + self._debug['file_structure_version']['start'] = self._io.pos() + self.file_structure_version = self._io.read_u1() + self._debug['file_structure_version']['end'] = self._io.pos() + self._debug['unused4']['start'] = self._io.pos() + self.unused4 = self._io.read_u1() + self._debug['unused4']['end'] = self._io.pos() + self._debug['application_area']['start'] = self._io.pos() + self.application_area = self._io.read_bytes(512) + self._debug['application_area']['end'] = self._io.pos() + + @property + def path_table(self): + if hasattr(self, '_m_path_table'): + return self._m_path_table if hasattr(self, '_m_path_table') else None + + _pos = self._io.pos() + self._io.seek((self.lba_path_table_le * self._root.sector_size)) + self._debug['_m_path_table']['start'] = self._io.pos() + self._raw__m_path_table = self._io.read_bytes(self.path_table_size.le) + io = KaitaiStream(BytesIO(self._raw__m_path_table)) + self._m_path_table = self._root.PathTableLe(io, self, self._root) + self._m_path_table._read() + self._debug['_m_path_table']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_path_table if hasattr(self, '_m_path_table') else None + + + class VolDescBootRecord(KaitaiStruct): + SEQ_FIELDS = ["boot_system_id", "boot_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['boot_system_id']['start'] = self._io.pos() + self.boot_system_id = (self._io.read_bytes(32)).decode(u"UTF-8") + self._debug['boot_system_id']['end'] = self._io.pos() + self._debug['boot_id']['start'] = self._io.pos() + self.boot_id = (self._io.read_bytes(32)).decode(u"UTF-8") + self._debug['boot_id']['end'] = self._io.pos() + + + class Datetime(KaitaiStruct): + SEQ_FIELDS = ["year", "month", "day", "hour", "minute", "sec", "timezone"] + 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['year']['start'] = self._io.pos() + self.year = self._io.read_u1() + self._debug['year']['end'] = self._io.pos() + self._debug['month']['start'] = self._io.pos() + self.month = self._io.read_u1() + self._debug['month']['end'] = self._io.pos() + self._debug['day']['start'] = self._io.pos() + self.day = self._io.read_u1() + self._debug['day']['end'] = self._io.pos() + self._debug['hour']['start'] = self._io.pos() + self.hour = self._io.read_u1() + self._debug['hour']['end'] = self._io.pos() + self._debug['minute']['start'] = self._io.pos() + self.minute = self._io.read_u1() + self._debug['minute']['end'] = self._io.pos() + self._debug['sec']['start'] = self._io.pos() + self.sec = self._io.read_u1() + self._debug['sec']['end'] = self._io.pos() + self._debug['timezone']['start'] = self._io.pos() + self.timezone = self._io.read_u1() + self._debug['timezone']['end'] = self._io.pos() + + + class DirEntry(KaitaiStruct): + SEQ_FIELDS = ["len", "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['len']['start'] = self._io.pos() + self.len = self._io.read_u1() + self._debug['len']['end'] = self._io.pos() + if self.len > 0: + self._debug['body']['start'] = self._io.pos() + self._raw_body = self._io.read_bytes((self.len - 1)) + io = KaitaiStream(BytesIO(self._raw_body)) + self.body = self._root.DirEntryBody(io, self, self._root) + self.body._read() + self._debug['body']['end'] = self._io.pos() + + + + class VolDesc(KaitaiStruct): + SEQ_FIELDS = ["type", "magic", "version", "vol_desc_boot_record", "vol_desc_primary"] + 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 = self._io.read_u1() + self._debug['type']['end'] = self._io.pos() + self._debug['magic']['start'] = self._io.pos() + self.magic = self._io.ensure_fixed_contents(b"\x43\x44\x30\x30\x31") + self._debug['magic']['end'] = self._io.pos() + self._debug['version']['start'] = self._io.pos() + self.version = self._io.read_u1() + self._debug['version']['end'] = self._io.pos() + if self.type == 0: + self._debug['vol_desc_boot_record']['start'] = self._io.pos() + self.vol_desc_boot_record = self._root.VolDescBootRecord(self._io, self, self._root) + self.vol_desc_boot_record._read() + self._debug['vol_desc_boot_record']['end'] = self._io.pos() + + if self.type == 1: + self._debug['vol_desc_primary']['start'] = self._io.pos() + self.vol_desc_primary = self._root.VolDescPrimary(self._io, self, self._root) + self.vol_desc_primary._read() + self._debug['vol_desc_primary']['end'] = self._io.pos() + + + + class PathTableEntryLe(KaitaiStruct): + SEQ_FIELDS = ["len_dir_name", "len_ext_attr_rec", "lba_extent", "parent_dir_idx", "dir_name", "padding"] + 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_dir_name']['start'] = self._io.pos() + self.len_dir_name = self._io.read_u1() + self._debug['len_dir_name']['end'] = self._io.pos() + self._debug['len_ext_attr_rec']['start'] = self._io.pos() + self.len_ext_attr_rec = self._io.read_u1() + self._debug['len_ext_attr_rec']['end'] = self._io.pos() + self._debug['lba_extent']['start'] = self._io.pos() + self.lba_extent = self._io.read_u4le() + self._debug['lba_extent']['end'] = self._io.pos() + self._debug['parent_dir_idx']['start'] = self._io.pos() + self.parent_dir_idx = self._io.read_u2le() + self._debug['parent_dir_idx']['end'] = self._io.pos() + self._debug['dir_name']['start'] = self._io.pos() + self.dir_name = (self._io.read_bytes(self.len_dir_name)).decode(u"UTF-8") + self._debug['dir_name']['end'] = self._io.pos() + if (self.len_dir_name % 2) == 1: + self._debug['padding']['start'] = self._io.pos() + self.padding = self._io.read_u1() + self._debug['padding']['end'] = self._io.pos() + + + + class DirEntries(KaitaiStruct): + SEQ_FIELDS = ["entries"] + def __init__(self, _io, _parent=None, _root=None): + self._io = _io + self._parent = _parent + self._root = _root if _root else self + self._debug = collections.defaultdict(dict) + + def _read(self): + self._debug['entries']['start'] = self._io.pos() + self.entries = [] + i = 0 + while True: + if not 'arr' in self._debug['entries']: + self._debug['entries']['arr'] = [] + self._debug['entries']['arr'].append({'start': self._io.pos()}) + _t_entries = self._root.DirEntry(self._io, self, self._root) + _t_entries._read() + _ = _t_entries + self.entries.append(_) + self._debug['entries']['arr'][len(self.entries) - 1]['end'] = self._io.pos() + if _.len == 0: + break + i += 1 + self._debug['entries']['end'] = self._io.pos() + + + class U4bi(KaitaiStruct): + SEQ_FIELDS = ["le", "be"] + 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['le']['start'] = self._io.pos() + self.le = self._io.read_u4le() + self._debug['le']['end'] = self._io.pos() + self._debug['be']['start'] = self._io.pos() + self.be = self._io.read_u4be() + self._debug['be']['end'] = self._io.pos() + + + class U2bi(KaitaiStruct): + SEQ_FIELDS = ["le", "be"] + 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['le']['start'] = self._io.pos() + self.le = self._io.read_u2le() + self._debug['le']['end'] = self._io.pos() + self._debug['be']['start'] = self._io.pos() + self.be = self._io.read_u2be() + self._debug['be']['end'] = self._io.pos() + + + class PathTableLe(KaitaiStruct): + """ + .. seealso:: + Source - http://wiki.osdev.org/ISO_9660#The_Path_Table + """ + SEQ_FIELDS = ["entries"] + def __init__(self, _io, _parent=None, _root=None): + self._io = _io + self._parent = _parent + self._root = _root if _root else self + self._debug = collections.defaultdict(dict) + + def _read(self): + self._debug['entries']['start'] = self._io.pos() + self.entries = [] + i = 0 + while not self._io.is_eof(): + if not 'arr' in self._debug['entries']: + self._debug['entries']['arr'] = [] + self._debug['entries']['arr'].append({'start': self._io.pos()}) + _t_entries = self._root.PathTableEntryLe(self._io, self, self._root) + _t_entries._read() + self.entries.append(_t_entries) + self._debug['entries']['arr'][len(self.entries) - 1]['end'] = self._io.pos() + i += 1 + + self._debug['entries']['end'] = self._io.pos() + + + class DecDatetime(KaitaiStruct): + """ + .. seealso:: + Source - http://wiki.osdev.org/ISO_9660#Date.2Ftime_format + """ + SEQ_FIELDS = ["year", "month", "day", "hour", "minute", "sec", "sec_hundreds", "timezone"] + 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['year']['start'] = self._io.pos() + self.year = (self._io.read_bytes(4)).decode(u"ASCII") + self._debug['year']['end'] = self._io.pos() + self._debug['month']['start'] = self._io.pos() + self.month = (self._io.read_bytes(2)).decode(u"ASCII") + self._debug['month']['end'] = self._io.pos() + self._debug['day']['start'] = self._io.pos() + self.day = (self._io.read_bytes(2)).decode(u"ASCII") + self._debug['day']['end'] = self._io.pos() + self._debug['hour']['start'] = self._io.pos() + self.hour = (self._io.read_bytes(2)).decode(u"ASCII") + self._debug['hour']['end'] = self._io.pos() + self._debug['minute']['start'] = self._io.pos() + self.minute = (self._io.read_bytes(2)).decode(u"ASCII") + self._debug['minute']['end'] = self._io.pos() + self._debug['sec']['start'] = self._io.pos() + self.sec = (self._io.read_bytes(2)).decode(u"ASCII") + self._debug['sec']['end'] = self._io.pos() + self._debug['sec_hundreds']['start'] = self._io.pos() + self.sec_hundreds = (self._io.read_bytes(2)).decode(u"ASCII") + self._debug['sec_hundreds']['end'] = self._io.pos() + self._debug['timezone']['start'] = self._io.pos() + self.timezone = self._io.read_u1() + self._debug['timezone']['end'] = self._io.pos() + + + class DirEntryBody(KaitaiStruct): + SEQ_FIELDS = ["len_ext_attr_rec", "lba_extent", "size_extent", "datetime", "file_flags", "file_unit_size", "interleave_gap_size", "vol_seq_num", "len_file_name", "file_name", "padding", "rest"] + 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_ext_attr_rec']['start'] = self._io.pos() + self.len_ext_attr_rec = self._io.read_u1() + self._debug['len_ext_attr_rec']['end'] = self._io.pos() + self._debug['lba_extent']['start'] = self._io.pos() + self.lba_extent = self._root.U4bi(self._io, self, self._root) + self.lba_extent._read() + self._debug['lba_extent']['end'] = self._io.pos() + self._debug['size_extent']['start'] = self._io.pos() + self.size_extent = self._root.U4bi(self._io, self, self._root) + self.size_extent._read() + self._debug['size_extent']['end'] = self._io.pos() + self._debug['datetime']['start'] = self._io.pos() + self.datetime = self._root.Datetime(self._io, self, self._root) + self.datetime._read() + self._debug['datetime']['end'] = self._io.pos() + self._debug['file_flags']['start'] = self._io.pos() + self.file_flags = self._io.read_u1() + self._debug['file_flags']['end'] = self._io.pos() + self._debug['file_unit_size']['start'] = self._io.pos() + self.file_unit_size = self._io.read_u1() + self._debug['file_unit_size']['end'] = self._io.pos() + self._debug['interleave_gap_size']['start'] = self._io.pos() + self.interleave_gap_size = self._io.read_u1() + self._debug['interleave_gap_size']['end'] = self._io.pos() + self._debug['vol_seq_num']['start'] = self._io.pos() + self.vol_seq_num = self._root.U2bi(self._io, self, self._root) + self.vol_seq_num._read() + self._debug['vol_seq_num']['end'] = self._io.pos() + self._debug['len_file_name']['start'] = self._io.pos() + self.len_file_name = self._io.read_u1() + self._debug['len_file_name']['end'] = self._io.pos() + self._debug['file_name']['start'] = self._io.pos() + self.file_name = (self._io.read_bytes(self.len_file_name)).decode(u"UTF-8") + self._debug['file_name']['end'] = self._io.pos() + if (self.len_file_name % 2) == 0: + self._debug['padding']['start'] = self._io.pos() + self.padding = self._io.read_u1() + self._debug['padding']['end'] = self._io.pos() + + self._debug['rest']['start'] = self._io.pos() + self.rest = self._io.read_bytes_full() + self._debug['rest']['end'] = self._io.pos() + + @property + def extent_as_dir(self): + if hasattr(self, '_m_extent_as_dir'): + return self._m_extent_as_dir if hasattr(self, '_m_extent_as_dir') else None + + if (self.file_flags & 2) != 0: + io = self._root._io + _pos = io.pos() + io.seek((self.lba_extent.le * self._root.sector_size)) + self._debug['_m_extent_as_dir']['start'] = io.pos() + self._raw__m_extent_as_dir = io.read_bytes(self.size_extent.le) + io = KaitaiStream(BytesIO(self._raw__m_extent_as_dir)) + self._m_extent_as_dir = self._root.DirEntries(io, self, self._root) + self._m_extent_as_dir._read() + self._debug['_m_extent_as_dir']['end'] = io.pos() + io.seek(_pos) + + return self._m_extent_as_dir if hasattr(self, '_m_extent_as_dir') else None + + @property + def extent_as_file(self): + if hasattr(self, '_m_extent_as_file'): + return self._m_extent_as_file if hasattr(self, '_m_extent_as_file') else None + + if (self.file_flags & 2) == 0: + io = self._root._io + _pos = io.pos() + io.seek((self.lba_extent.le * self._root.sector_size)) + self._debug['_m_extent_as_file']['start'] = io.pos() + self._m_extent_as_file = io.read_bytes(self.size_extent.le) + self._debug['_m_extent_as_file']['end'] = io.pos() + io.seek(_pos) + + return self._m_extent_as_file if hasattr(self, '_m_extent_as_file') else None + + + @property + def sector_size(self): + if hasattr(self, '_m_sector_size'): + return self._m_sector_size if hasattr(self, '_m_sector_size') else None + + self._m_sector_size = 2048 + return self._m_sector_size if hasattr(self, '_m_sector_size') else None + + @property + def primary_vol_desc(self): + if hasattr(self, '_m_primary_vol_desc'): + return self._m_primary_vol_desc if hasattr(self, '_m_primary_vol_desc') else None + + _pos = self._io.pos() + self._io.seek((16 * self.sector_size)) + self._debug['_m_primary_vol_desc']['start'] = self._io.pos() + self._m_primary_vol_desc = self._root.VolDesc(self._io, self, self._root) + self._m_primary_vol_desc._read() + self._debug['_m_primary_vol_desc']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_primary_vol_desc if hasattr(self, '_m_primary_vol_desc') else None + + diff --git a/python/examples/kaitai/kaitai_struct_formats/filesystem/luks.py b/python/examples/kaitai/kaitai_struct_formats/filesystem/luks.py new file mode 100644 index 00000000..c4e33287 --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/filesystem/luks.py @@ -0,0 +1,145 @@ +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 Luks(KaitaiStruct): + """Linux Unified Key Setup (LUKS) is a format specification for storing disk + encryption parameters and up to 8 user keys (which can unlock the master key). + + .. seealso:: + Source - https://gitlab.com/cryptsetup/cryptsetup/wikis/LUKS-standard/on-disk-format.pdf + """ + SEQ_FIELDS = ["partition_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['partition_header']['start'] = self._io.pos() + self.partition_header = self._root.PartitionHeader(self._io, self, self._root) + self.partition_header._read() + self._debug['partition_header']['end'] = self._io.pos() + + class PartitionHeader(KaitaiStruct): + SEQ_FIELDS = ["magic", "version", "cipher_name_specification", "cipher_mode_specification", "hash_specification", "payload_offset", "number_of_key_bytes", "master_key_checksum", "master_key_salt_parameter", "master_key_iterations_parameter", "uuid", "key_slots"] + 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"\x4C\x55\x4B\x53\xBA\xBE") + self._debug['magic']['end'] = self._io.pos() + self._debug['version']['start'] = self._io.pos() + self.version = self._io.ensure_fixed_contents(b"\x00\x01") + self._debug['version']['end'] = self._io.pos() + self._debug['cipher_name_specification']['start'] = self._io.pos() + self.cipher_name_specification = (self._io.read_bytes(32)).decode(u"ASCII") + self._debug['cipher_name_specification']['end'] = self._io.pos() + self._debug['cipher_mode_specification']['start'] = self._io.pos() + self.cipher_mode_specification = (self._io.read_bytes(32)).decode(u"ASCII") + self._debug['cipher_mode_specification']['end'] = self._io.pos() + self._debug['hash_specification']['start'] = self._io.pos() + self.hash_specification = (self._io.read_bytes(32)).decode(u"ASCII") + self._debug['hash_specification']['end'] = self._io.pos() + self._debug['payload_offset']['start'] = self._io.pos() + self.payload_offset = self._io.read_u4be() + self._debug['payload_offset']['end'] = self._io.pos() + self._debug['number_of_key_bytes']['start'] = self._io.pos() + self.number_of_key_bytes = self._io.read_u4be() + self._debug['number_of_key_bytes']['end'] = self._io.pos() + self._debug['master_key_checksum']['start'] = self._io.pos() + self.master_key_checksum = self._io.read_bytes(20) + self._debug['master_key_checksum']['end'] = self._io.pos() + self._debug['master_key_salt_parameter']['start'] = self._io.pos() + self.master_key_salt_parameter = self._io.read_bytes(32) + self._debug['master_key_salt_parameter']['end'] = self._io.pos() + self._debug['master_key_iterations_parameter']['start'] = self._io.pos() + self.master_key_iterations_parameter = self._io.read_u4be() + self._debug['master_key_iterations_parameter']['end'] = self._io.pos() + self._debug['uuid']['start'] = self._io.pos() + self.uuid = (self._io.read_bytes(40)).decode(u"ASCII") + self._debug['uuid']['end'] = self._io.pos() + self._debug['key_slots']['start'] = self._io.pos() + self.key_slots = [None] * (8) + for i in range(8): + if not 'arr' in self._debug['key_slots']: + self._debug['key_slots']['arr'] = [] + self._debug['key_slots']['arr'].append({'start': self._io.pos()}) + _t_key_slots = self._root.PartitionHeader.KeySlot(self._io, self, self._root) + _t_key_slots._read() + self.key_slots[i] = _t_key_slots + self._debug['key_slots']['arr'][i]['end'] = self._io.pos() + + self._debug['key_slots']['end'] = self._io.pos() + + class KeySlot(KaitaiStruct): + + class KeySlotStates(Enum): + disabled_key_slot = 57005 + enabled_key_slot = 11301363 + SEQ_FIELDS = ["state_of_key_slot", "iteration_parameter", "salt_parameter", "start_sector_of_key_material", "number_of_anti_forensic_stripes"] + 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['state_of_key_slot']['start'] = self._io.pos() + self.state_of_key_slot = KaitaiStream.resolve_enum(self._root.PartitionHeader.KeySlot.KeySlotStates, self._io.read_u4be()) + self._debug['state_of_key_slot']['end'] = self._io.pos() + self._debug['iteration_parameter']['start'] = self._io.pos() + self.iteration_parameter = self._io.read_u4be() + self._debug['iteration_parameter']['end'] = self._io.pos() + self._debug['salt_parameter']['start'] = self._io.pos() + self.salt_parameter = self._io.read_bytes(32) + self._debug['salt_parameter']['end'] = self._io.pos() + self._debug['start_sector_of_key_material']['start'] = self._io.pos() + self.start_sector_of_key_material = self._io.read_u4be() + self._debug['start_sector_of_key_material']['end'] = self._io.pos() + self._debug['number_of_anti_forensic_stripes']['start'] = self._io.pos() + self.number_of_anti_forensic_stripes = self._io.read_u4be() + self._debug['number_of_anti_forensic_stripes']['end'] = self._io.pos() + + @property + def key_material(self): + if hasattr(self, '_m_key_material'): + return self._m_key_material if hasattr(self, '_m_key_material') else None + + _pos = self._io.pos() + self._io.seek((self.start_sector_of_key_material * 512)) + self._debug['_m_key_material']['start'] = self._io.pos() + self._m_key_material = self._io.read_bytes((self._parent.number_of_key_bytes * self.number_of_anti_forensic_stripes)) + self._debug['_m_key_material']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_key_material if hasattr(self, '_m_key_material') else None + + + + @property + def payload(self): + if hasattr(self, '_m_payload'): + return self._m_payload if hasattr(self, '_m_payload') else None + + _pos = self._io.pos() + self._io.seek((self.partition_header.payload_offset * 512)) + self._debug['_m_payload']['start'] = self._io.pos() + self._m_payload = self._io.read_bytes_full() + self._debug['_m_payload']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_payload if hasattr(self, '_m_payload') else None + + diff --git a/python/examples/kaitai/kaitai_struct_formats/filesystem/lvm2.py b/python/examples/kaitai/kaitai_struct_formats/filesystem/lvm2.py new file mode 100644 index 00000000..459f92ca --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/filesystem/lvm2.py @@ -0,0 +1,342 @@ +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 Lvm2(KaitaiStruct): + """### Building a test file + + ``` + dd if=/dev/zero of=image.img bs=512 count=$(( 4 * 1024 * 2 )) + sudo losetup /dev/loop1 image.img + sudo pvcreate /dev/loop1 + sudo vgcreate vg_test /dev/loop1 + sudo lvcreate --name lv_test1 vg_test + sudo losetup -d /dev/loop1 + ``` + + .. seealso:: + Source - https://github.com/libyal/libvslvm/blob/master/documentation/Logical%20Volume%20Manager%20(LVM)%20format.asciidoc + """ + SEQ_FIELDS = ["pv"] + 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['pv']['start'] = self._io.pos() + self.pv = self._root.PhysicalVolume(self._io, self, self._root) + self.pv._read() + self._debug['pv']['end'] = self._io.pos() + + class PhysicalVolume(KaitaiStruct): + SEQ_FIELDS = ["empty_sector", "label"] + 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['empty_sector']['start'] = self._io.pos() + self.empty_sector = self._io.read_bytes(self._root.sector_size) + self._debug['empty_sector']['end'] = self._io.pos() + self._debug['label']['start'] = self._io.pos() + self.label = self._root.PhysicalVolume.Label(self._io, self, self._root) + self.label._read() + self._debug['label']['end'] = self._io.pos() + + class Label(KaitaiStruct): + SEQ_FIELDS = ["label_header", "volume_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['label_header']['start'] = self._io.pos() + self.label_header = self._root.PhysicalVolume.Label.LabelHeader(self._io, self, self._root) + self.label_header._read() + self._debug['label_header']['end'] = self._io.pos() + self._debug['volume_header']['start'] = self._io.pos() + self.volume_header = self._root.PhysicalVolume.Label.VolumeHeader(self._io, self, self._root) + self.volume_header._read() + self._debug['volume_header']['end'] = self._io.pos() + + class LabelHeader(KaitaiStruct): + SEQ_FIELDS = ["signature", "sector_number", "checksum", "label_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['signature']['start'] = self._io.pos() + self.signature = self._io.ensure_fixed_contents(b"\x4C\x41\x42\x45\x4C\x4F\x4E\x45") + self._debug['signature']['end'] = self._io.pos() + self._debug['sector_number']['start'] = self._io.pos() + self.sector_number = self._io.read_u8le() + self._debug['sector_number']['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['label_header_']['start'] = self._io.pos() + self.label_header_ = self._root.PhysicalVolume.Label.LabelHeader.LabelHeader(self._io, self, self._root) + self.label_header_._read() + self._debug['label_header_']['end'] = self._io.pos() + + class LabelHeader(KaitaiStruct): + SEQ_FIELDS = ["data_offset", "type_indicator"] + 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['data_offset']['start'] = self._io.pos() + self.data_offset = self._io.read_u4le() + self._debug['data_offset']['end'] = self._io.pos() + self._debug['type_indicator']['start'] = self._io.pos() + self.type_indicator = self._io.ensure_fixed_contents(b"\x4C\x56\x4D\x32\x20\x30\x30\x31") + self._debug['type_indicator']['end'] = self._io.pos() + + + + class VolumeHeader(KaitaiStruct): + SEQ_FIELDS = ["id", "size", "data_area_descriptors", "metadata_area_descriptors"] + 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_bytes(32)).decode(u"ascii") + self._debug['id']['end'] = self._io.pos() + self._debug['size']['start'] = self._io.pos() + self.size = self._io.read_u8le() + self._debug['size']['end'] = self._io.pos() + self._debug['data_area_descriptors']['start'] = self._io.pos() + self.data_area_descriptors = [] + i = 0 + while True: + if not 'arr' in self._debug['data_area_descriptors']: + self._debug['data_area_descriptors']['arr'] = [] + self._debug['data_area_descriptors']['arr'].append({'start': self._io.pos()}) + _t_data_area_descriptors = self._root.PhysicalVolume.Label.VolumeHeader.DataAreaDescriptor(self._io, self, self._root) + _t_data_area_descriptors._read() + _ = _t_data_area_descriptors + self.data_area_descriptors.append(_) + self._debug['data_area_descriptors']['arr'][len(self.data_area_descriptors) - 1]['end'] = self._io.pos() + if ((_.size != 0) and (_.offset != 0)) : + break + i += 1 + self._debug['data_area_descriptors']['end'] = self._io.pos() + self._debug['metadata_area_descriptors']['start'] = self._io.pos() + self.metadata_area_descriptors = [] + i = 0 + while True: + if not 'arr' in self._debug['metadata_area_descriptors']: + self._debug['metadata_area_descriptors']['arr'] = [] + self._debug['metadata_area_descriptors']['arr'].append({'start': self._io.pos()}) + _t_metadata_area_descriptors = self._root.PhysicalVolume.Label.VolumeHeader.MetadataAreaDescriptor(self._io, self, self._root) + _t_metadata_area_descriptors._read() + _ = _t_metadata_area_descriptors + self.metadata_area_descriptors.append(_) + self._debug['metadata_area_descriptors']['arr'][len(self.metadata_area_descriptors) - 1]['end'] = self._io.pos() + if ((_.size != 0) and (_.offset != 0)) : + break + i += 1 + self._debug['metadata_area_descriptors']['end'] = self._io.pos() + + class DataAreaDescriptor(KaitaiStruct): + SEQ_FIELDS = ["offset", "size"] + 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['offset']['start'] = self._io.pos() + self.offset = self._io.read_u8le() + self._debug['offset']['end'] = self._io.pos() + self._debug['size']['start'] = self._io.pos() + self.size = self._io.read_u8le() + self._debug['size']['end'] = self._io.pos() + + @property + def data(self): + if hasattr(self, '_m_data'): + return self._m_data if hasattr(self, '_m_data') else None + + if self.size != 0: + _pos = self._io.pos() + self._io.seek(self.offset) + self._debug['_m_data']['start'] = self._io.pos() + self._m_data = (self._io.read_bytes(self.size)).decode(u"ascii") + self._debug['_m_data']['end'] = self._io.pos() + self._io.seek(_pos) + + return self._m_data if hasattr(self, '_m_data') else None + + + class MetadataAreaDescriptor(KaitaiStruct): + SEQ_FIELDS = ["offset", "size"] + 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['offset']['start'] = self._io.pos() + self.offset = self._io.read_u8le() + self._debug['offset']['end'] = self._io.pos() + self._debug['size']['start'] = self._io.pos() + self.size = self._io.read_u8le() + self._debug['size']['end'] = self._io.pos() + + @property + def data(self): + if hasattr(self, '_m_data'): + return self._m_data if hasattr(self, '_m_data') else None + + if self.size != 0: + _pos = self._io.pos() + self._io.seek(self.offset) + self._debug['_m_data']['start'] = self._io.pos() + self._raw__m_data = self._io.read_bytes(self.size) + io = KaitaiStream(BytesIO(self._raw__m_data)) + self._m_data = self._root.PhysicalVolume.Label.VolumeHeader.MetadataArea(io, self, self._root) + self._m_data._read() + self._debug['_m_data']['end'] = self._io.pos() + self._io.seek(_pos) + + return self._m_data if hasattr(self, '_m_data') else None + + + class MetadataArea(KaitaiStruct): + """According to `[REDHAT]` the metadata area is a circular buffer. New metadata is appended to the old metadata and then the pointer to the start of it is updated. The metadata area, therefore, can contain copies of older versions of the metadata.""" + 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.PhysicalVolume.Label.VolumeHeader.MetadataArea.MetadataAreaHeader(self._io, self, self._root) + self.header._read() + self._debug['header']['end'] = self._io.pos() + + class MetadataAreaHeader(KaitaiStruct): + SEQ_FIELDS = ["checksum", "signature", "version", "metadata_area_offset", "metadata_area_size", "raw_location_descriptors"] + 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['checksum']['start'] = self._io.pos() + self.checksum = self._root.PhysicalVolume.Label.VolumeHeader.MetadataArea.MetadataAreaHeader(self._io, self, self._root) + self.checksum._read() + self._debug['checksum']['end'] = self._io.pos() + self._debug['signature']['start'] = self._io.pos() + self.signature = self._io.ensure_fixed_contents(b"\x20\x4C\x56\x4D\x32\x20\x78\x5B\x35\x41\x25\x72\x30\x4E\x2A\x3E") + self._debug['signature']['end'] = self._io.pos() + self._debug['version']['start'] = self._io.pos() + self.version = self._io.read_u4le() + self._debug['version']['end'] = self._io.pos() + self._debug['metadata_area_offset']['start'] = self._io.pos() + self.metadata_area_offset = self._io.read_u8le() + self._debug['metadata_area_offset']['end'] = self._io.pos() + self._debug['metadata_area_size']['start'] = self._io.pos() + self.metadata_area_size = self._io.read_u8le() + self._debug['metadata_area_size']['end'] = self._io.pos() + self._debug['raw_location_descriptors']['start'] = self._io.pos() + self.raw_location_descriptors = [] + i = 0 + while True: + if not 'arr' in self._debug['raw_location_descriptors']: + self._debug['raw_location_descriptors']['arr'] = [] + self._debug['raw_location_descriptors']['arr'].append({'start': self._io.pos()}) + _t_raw_location_descriptors = self._root.PhysicalVolume.Label.VolumeHeader.MetadataArea.MetadataAreaHeader.RawLocationDescriptor(self._io, self, self._root) + _t_raw_location_descriptors._read() + _ = _t_raw_location_descriptors + self.raw_location_descriptors.append(_) + self._debug['raw_location_descriptors']['arr'][len(self.raw_location_descriptors) - 1]['end'] = self._io.pos() + if ((_.offset != 0) and (_.size != 0) and (_.checksum != 0)) : + break + i += 1 + self._debug['raw_location_descriptors']['end'] = self._io.pos() + + class RawLocationDescriptor(KaitaiStruct): + """The data area size can be 0. It is assumed it represents the remaining available data.""" + + class RawLocationDescriptorFlags(Enum): + raw_location_ignored = 1 + SEQ_FIELDS = ["offset", "size", "checksum", "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['offset']['start'] = self._io.pos() + self.offset = self._io.read_u8le() + self._debug['offset']['end'] = self._io.pos() + self._debug['size']['start'] = self._io.pos() + self.size = self._io.read_u8le() + self._debug['size']['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['flags']['start'] = self._io.pos() + self.flags = KaitaiStream.resolve_enum(self._root.PhysicalVolume.Label.VolumeHeader.MetadataArea.MetadataAreaHeader.RawLocationDescriptor.RawLocationDescriptorFlags, self._io.read_u4le()) + self._debug['flags']['end'] = self._io.pos() + + + @property + def metadata(self): + if hasattr(self, '_m_metadata'): + return self._m_metadata if hasattr(self, '_m_metadata') else None + + _pos = self._io.pos() + self._io.seek(self.metadata_area_offset) + self._debug['_m_metadata']['start'] = self._io.pos() + self._m_metadata = self._io.read_bytes(self.metadata_area_size) + self._debug['_m_metadata']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_metadata if hasattr(self, '_m_metadata') else None + + + + + + + @property + def sector_size(self): + if hasattr(self, '_m_sector_size'): + return self._m_sector_size if hasattr(self, '_m_sector_size') else None + + self._m_sector_size = 512 + return self._m_sector_size if hasattr(self, '_m_sector_size') else None + + diff --git a/python/examples/kaitai/kaitai_struct_formats/filesystem/mbr_partition_table.py b/python/examples/kaitai/kaitai_struct_formats/filesystem/mbr_partition_table.py new file mode 100644 index 00000000..558dd14c --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/filesystem/mbr_partition_table.py @@ -0,0 +1,117 @@ +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 MbrPartitionTable(KaitaiStruct): + """MBR (Master Boot Record) partition table is a traditional way of + MS-DOS to partition larger hard disc drives into distinct + partitions. + + This table is stored in the end of the boot sector (first sector) of + the drive, after the bootstrap code. Original DOS 2.0 specification + allowed only 4 partitions per disc, but DOS 3.2 introduced concept + of "extended partitions", which work as nested extra "boot records" + which are pointed to by original ("primary") partitions in MBR. + """ + SEQ_FIELDS = ["bootstrap_code", "partitions", "boot_signature"] + 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_code']['start'] = self._io.pos() + self.bootstrap_code = self._io.read_bytes(446) + self._debug['bootstrap_code']['end'] = self._io.pos() + self._debug['partitions']['start'] = self._io.pos() + self.partitions = [None] * (4) + for i in range(4): + if not 'arr' in self._debug['partitions']: + self._debug['partitions']['arr'] = [] + self._debug['partitions']['arr'].append({'start': self._io.pos()}) + _t_partitions = self._root.PartitionEntry(self._io, self, self._root) + _t_partitions._read() + self.partitions[i] = _t_partitions + self._debug['partitions']['arr'][i]['end'] = self._io.pos() + + self._debug['partitions']['end'] = self._io.pos() + self._debug['boot_signature']['start'] = self._io.pos() + self.boot_signature = self._io.ensure_fixed_contents(b"\x55\xAA") + self._debug['boot_signature']['end'] = self._io.pos() + + class PartitionEntry(KaitaiStruct): + SEQ_FIELDS = ["status", "chs_start", "partition_type", "chs_end", "lba_start", "num_sectors"] + 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['status']['start'] = self._io.pos() + self.status = self._io.read_u1() + self._debug['status']['end'] = self._io.pos() + self._debug['chs_start']['start'] = self._io.pos() + self.chs_start = self._root.Chs(self._io, self, self._root) + self.chs_start._read() + self._debug['chs_start']['end'] = self._io.pos() + self._debug['partition_type']['start'] = self._io.pos() + self.partition_type = self._io.read_u1() + self._debug['partition_type']['end'] = self._io.pos() + self._debug['chs_end']['start'] = self._io.pos() + self.chs_end = self._root.Chs(self._io, self, self._root) + self.chs_end._read() + self._debug['chs_end']['end'] = self._io.pos() + self._debug['lba_start']['start'] = self._io.pos() + self.lba_start = self._io.read_u4le() + self._debug['lba_start']['end'] = self._io.pos() + self._debug['num_sectors']['start'] = self._io.pos() + self.num_sectors = self._io.read_u4le() + self._debug['num_sectors']['end'] = self._io.pos() + + + class Chs(KaitaiStruct): + SEQ_FIELDS = ["head", "b2", "b3"] + 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['head']['start'] = self._io.pos() + self.head = self._io.read_u1() + self._debug['head']['end'] = self._io.pos() + self._debug['b2']['start'] = self._io.pos() + self.b2 = self._io.read_u1() + self._debug['b2']['end'] = self._io.pos() + self._debug['b3']['start'] = self._io.pos() + self.b3 = self._io.read_u1() + self._debug['b3']['end'] = self._io.pos() + + @property + def sector(self): + if hasattr(self, '_m_sector'): + return self._m_sector if hasattr(self, '_m_sector') else None + + self._m_sector = (self.b2 & 63) + return self._m_sector if hasattr(self, '_m_sector') else None + + @property + def cylinder(self): + if hasattr(self, '_m_cylinder'): + return self._m_cylinder if hasattr(self, '_m_cylinder') else None + + self._m_cylinder = (self.b3 + ((self.b2 & 192) << 2)) + return self._m_cylinder if hasattr(self, '_m_cylinder') else None + + + diff --git a/python/examples/kaitai/kaitai_struct_formats/filesystem/tr_dos_image.py b/python/examples/kaitai/kaitai_struct_formats/filesystem/tr_dos_image.py new file mode 100644 index 00000000..ce0bdc3b --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/filesystem/tr_dos_image.py @@ -0,0 +1,318 @@ +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 TrDosImage(KaitaiStruct): + """.trd file is a raw dump of TR-DOS (ZX-Spectrum) floppy. .trd files are + headerless and contain consequent "logical tracks", each logical track + consists of 16 256-byte sectors. + + Logical tracks are defined the same way as used by TR-DOS: for single-side + floppies it's just a physical track number, for two-side floppies sides are + interleaved, i.e. logical_track_num = (physical_track_num << 1) | side + + So, this format definition is more for TR-DOS filesystem than for .trd files, + which are formatless. + + Strings (file names, disk label, disk password) are padded with spaces and use + ZX Spectrum character set, including UDGs, block drawing chars and Basic + tokens. ASCII range is mostly standard ASCII, with few characters (^, `, DEL) + replaced with (up arrow, pound, copyright symbol). + + .trd file can be smaller than actual floppy disk, if last logical tracks are + empty (contain no file data) they can be omitted. + """ + + class DiskType(Enum): + type_80_tracks_double_side = 22 + type_40_tracks_double_side = 23 + type_80_tracks_single_side = 24 + type_40_tracks_single_side = 25 + SEQ_FIELDS = ["files"] + 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['files']['start'] = self._io.pos() + self.files = [] + i = 0 + while True: + if not 'arr' in self._debug['files']: + self._debug['files']['arr'] = [] + self._debug['files']['arr'].append({'start': self._io.pos()}) + _t_files = self._root.File(self._io, self, self._root) + _t_files._read() + _ = _t_files + self.files.append(_) + self._debug['files']['arr'][len(self.files) - 1]['end'] = self._io.pos() + if _.is_terminator: + break + i += 1 + self._debug['files']['end'] = self._io.pos() + + class VolumeInfo(KaitaiStruct): + SEQ_FIELDS = ["catalog_end", "unused", "first_free_sector_sector", "first_free_sector_track", "disk_type", "num_files", "num_free_sectors", "tr_dos_id", "unused_2", "password", "unused_3", "num_deleted_files", "label", "unused_4"] + 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['catalog_end']['start'] = self._io.pos() + self.catalog_end = self._io.ensure_fixed_contents(b"\x00") + self._debug['catalog_end']['end'] = self._io.pos() + self._debug['unused']['start'] = self._io.pos() + self.unused = self._io.read_bytes(224) + self._debug['unused']['end'] = self._io.pos() + self._debug['first_free_sector_sector']['start'] = self._io.pos() + self.first_free_sector_sector = self._io.read_u1() + self._debug['first_free_sector_sector']['end'] = self._io.pos() + self._debug['first_free_sector_track']['start'] = self._io.pos() + self.first_free_sector_track = self._io.read_u1() + self._debug['first_free_sector_track']['end'] = self._io.pos() + self._debug['disk_type']['start'] = self._io.pos() + self.disk_type = KaitaiStream.resolve_enum(self._root.DiskType, self._io.read_u1()) + self._debug['disk_type']['end'] = self._io.pos() + self._debug['num_files']['start'] = self._io.pos() + self.num_files = self._io.read_u1() + self._debug['num_files']['end'] = self._io.pos() + self._debug['num_free_sectors']['start'] = self._io.pos() + self.num_free_sectors = self._io.read_u2le() + self._debug['num_free_sectors']['end'] = self._io.pos() + self._debug['tr_dos_id']['start'] = self._io.pos() + self.tr_dos_id = self._io.ensure_fixed_contents(b"\x10") + self._debug['tr_dos_id']['end'] = self._io.pos() + self._debug['unused_2']['start'] = self._io.pos() + self.unused_2 = self._io.read_bytes(2) + self._debug['unused_2']['end'] = self._io.pos() + self._debug['password']['start'] = self._io.pos() + self.password = self._io.read_bytes(9) + self._debug['password']['end'] = self._io.pos() + self._debug['unused_3']['start'] = self._io.pos() + self.unused_3 = self._io.read_bytes(1) + self._debug['unused_3']['end'] = self._io.pos() + self._debug['num_deleted_files']['start'] = self._io.pos() + self.num_deleted_files = self._io.read_u1() + self._debug['num_deleted_files']['end'] = self._io.pos() + self._debug['label']['start'] = self._io.pos() + self.label = self._io.read_bytes(8) + self._debug['label']['end'] = self._io.pos() + self._debug['unused_4']['start'] = self._io.pos() + self.unused_4 = self._io.read_bytes(3) + self._debug['unused_4']['end'] = self._io.pos() + + @property + def num_tracks(self): + if hasattr(self, '_m_num_tracks'): + return self._m_num_tracks if hasattr(self, '_m_num_tracks') else None + + self._m_num_tracks = (40 if (self.disk_type.value & 1) != 0 else 80) + return self._m_num_tracks if hasattr(self, '_m_num_tracks') else None + + @property + def num_sides(self): + if hasattr(self, '_m_num_sides'): + return self._m_num_sides if hasattr(self, '_m_num_sides') else None + + self._m_num_sides = (1 if (self.disk_type.value & 8) != 0 else 2) + return self._m_num_sides if hasattr(self, '_m_num_sides') else None + + + class PositionAndLengthCode(KaitaiStruct): + SEQ_FIELDS = ["start_address", "length"] + 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_address']['start'] = self._io.pos() + self.start_address = self._io.read_u2le() + self._debug['start_address']['end'] = self._io.pos() + self._debug['length']['start'] = self._io.pos() + self.length = self._io.read_u2le() + self._debug['length']['end'] = self._io.pos() + + + class Filename(KaitaiStruct): + SEQ_FIELDS = ["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['name']['start'] = self._io.pos() + self.name = self._io.read_bytes(8) + self._debug['name']['end'] = self._io.pos() + + @property + def first_byte(self): + if hasattr(self, '_m_first_byte'): + return self._m_first_byte if hasattr(self, '_m_first_byte') else None + + _pos = self._io.pos() + self._io.seek(0) + self._debug['_m_first_byte']['start'] = self._io.pos() + self._m_first_byte = self._io.read_u1() + self._debug['_m_first_byte']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_first_byte if hasattr(self, '_m_first_byte') else None + + + class PositionAndLengthPrint(KaitaiStruct): + SEQ_FIELDS = ["extent_no", "reserved", "length"] + 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['extent_no']['start'] = self._io.pos() + self.extent_no = self._io.read_u1() + self._debug['extent_no']['end'] = self._io.pos() + self._debug['reserved']['start'] = self._io.pos() + self.reserved = self._io.read_u1() + self._debug['reserved']['end'] = self._io.pos() + self._debug['length']['start'] = self._io.pos() + self.length = self._io.read_u2le() + self._debug['length']['end'] = self._io.pos() + + + class PositionAndLengthGeneric(KaitaiStruct): + SEQ_FIELDS = ["reserved", "length"] + def __init__(self, _io, _parent=None, _root=None): + self._io = _io + self._parent = _parent + self._root = _root if _root else self + self._debug = collections.defaultdict(dict) + + def _read(self): + self._debug['reserved']['start'] = self._io.pos() + self.reserved = self._io.read_u2le() + self._debug['reserved']['end'] = self._io.pos() + self._debug['length']['start'] = self._io.pos() + self.length = self._io.read_u2le() + self._debug['length']['end'] = self._io.pos() + + + class PositionAndLengthBasic(KaitaiStruct): + SEQ_FIELDS = ["program_and_data_length", "program_length"] + 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['program_and_data_length']['start'] = self._io.pos() + self.program_and_data_length = self._io.read_u2le() + self._debug['program_and_data_length']['end'] = self._io.pos() + self._debug['program_length']['start'] = self._io.pos() + self.program_length = self._io.read_u2le() + self._debug['program_length']['end'] = self._io.pos() + + + class File(KaitaiStruct): + SEQ_FIELDS = ["name", "extension", "position_and_length", "length_sectors", "starting_sector", "starting_track"] + 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']['start'] = self._io.pos() + self._raw_name = self._io.read_bytes(8) + io = KaitaiStream(BytesIO(self._raw_name)) + self.name = self._root.Filename(io, self, self._root) + self.name._read() + self._debug['name']['end'] = self._io.pos() + self._debug['extension']['start'] = self._io.pos() + self.extension = self._io.read_u1() + self._debug['extension']['end'] = self._io.pos() + self._debug['position_and_length']['start'] = self._io.pos() + _on = self.extension + if _on == 66: + self.position_and_length = self._root.PositionAndLengthBasic(self._io, self, self._root) + self.position_and_length._read() + elif _on == 67: + self.position_and_length = self._root.PositionAndLengthCode(self._io, self, self._root) + self.position_and_length._read() + elif _on == 35: + self.position_and_length = self._root.PositionAndLengthPrint(self._io, self, self._root) + self.position_and_length._read() + else: + self.position_and_length = self._root.PositionAndLengthGeneric(self._io, self, self._root) + self.position_and_length._read() + self._debug['position_and_length']['end'] = self._io.pos() + self._debug['length_sectors']['start'] = self._io.pos() + self.length_sectors = self._io.read_u1() + self._debug['length_sectors']['end'] = self._io.pos() + self._debug['starting_sector']['start'] = self._io.pos() + self.starting_sector = self._io.read_u1() + self._debug['starting_sector']['end'] = self._io.pos() + self._debug['starting_track']['start'] = self._io.pos() + self.starting_track = self._io.read_u1() + self._debug['starting_track']['end'] = self._io.pos() + + @property + def is_deleted(self): + if hasattr(self, '_m_is_deleted'): + return self._m_is_deleted if hasattr(self, '_m_is_deleted') else None + + self._m_is_deleted = self.name.first_byte == 1 + return self._m_is_deleted if hasattr(self, '_m_is_deleted') else None + + @property + def is_terminator(self): + if hasattr(self, '_m_is_terminator'): + return self._m_is_terminator if hasattr(self, '_m_is_terminator') else None + + self._m_is_terminator = self.name.first_byte == 0 + return self._m_is_terminator if hasattr(self, '_m_is_terminator') else None + + @property + def contents(self): + if hasattr(self, '_m_contents'): + return self._m_contents if hasattr(self, '_m_contents') else None + + _pos = self._io.pos() + self._io.seek((((self.starting_track * 256) * 16) + (self.starting_sector * 256))) + self._debug['_m_contents']['start'] = self._io.pos() + self._m_contents = self._io.read_bytes((self.length_sectors * 256)) + self._debug['_m_contents']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_contents if hasattr(self, '_m_contents') else None + + + @property + def volume_info(self): + if hasattr(self, '_m_volume_info'): + return self._m_volume_info if hasattr(self, '_m_volume_info') else None + + _pos = self._io.pos() + self._io.seek(2048) + self._debug['_m_volume_info']['start'] = self._io.pos() + self._m_volume_info = self._root.VolumeInfo(self._io, self, self._root) + self._m_volume_info._read() + self._debug['_m_volume_info']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_volume_info if hasattr(self, '_m_volume_info') else None + + diff --git a/python/examples/kaitai/kaitai_struct_formats/filesystem/vdi.py b/python/examples/kaitai/kaitai_struct_formats/filesystem/vdi.py new file mode 100644 index 00000000..8dd8a775 --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/filesystem/vdi.py @@ -0,0 +1,459 @@ +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 Vdi(KaitaiStruct): + """A native VirtualBox file format + Images for testing can be downloaded from + * https://www.osboxes.org/virtualbox-images/ + * https://virtualboxes.org/images/ + * https://virtualboximages.com/ + or you can convert images of other formats. + + .. seealso:: + Source - https://github.com/qemu/qemu/blob/master/block/vdi.c + """ + + class ImageType(Enum): + dynamic = 1 + static = 2 + undo = 3 + diff = 4 + 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.Header(self._io, self, self._root) + self.header._read() + self._debug['header']['end'] = self._io.pos() + + class Header(KaitaiStruct): + SEQ_FIELDS = ["text", "signature", "version", "header_size_optional", "header_main"] + 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['text']['start'] = self._io.pos() + self.text = (self._io.read_bytes(64)).decode(u"utf-8") + self._debug['text']['end'] = self._io.pos() + self._debug['signature']['start'] = self._io.pos() + self.signature = self._io.ensure_fixed_contents(b"\x7F\x10\xDA\xBE") + self._debug['signature']['end'] = self._io.pos() + self._debug['version']['start'] = self._io.pos() + self.version = self._root.Header.Version(self._io, self, self._root) + self.version._read() + self._debug['version']['end'] = self._io.pos() + if self.subheader_size_is_dynamic: + self._debug['header_size_optional']['start'] = self._io.pos() + self.header_size_optional = self._io.read_u4le() + self._debug['header_size_optional']['end'] = self._io.pos() + + self._debug['header_main']['start'] = self._io.pos() + self._raw_header_main = self._io.read_bytes(self.header_size) + io = KaitaiStream(BytesIO(self._raw_header_main)) + self.header_main = self._root.Header.HeaderMain(io, self, self._root) + self.header_main._read() + self._debug['header_main']['end'] = self._io.pos() + + class Uuid(KaitaiStruct): + SEQ_FIELDS = ["uuid"] + 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['uuid']['start'] = self._io.pos() + self.uuid = self._io.read_bytes(16) + self._debug['uuid']['end'] = self._io.pos() + + + class Version(KaitaiStruct): + SEQ_FIELDS = ["major", "minor"] + 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['major']['start'] = self._io.pos() + self.major = self._io.read_u2le() + self._debug['major']['end'] = self._io.pos() + self._debug['minor']['start'] = self._io.pos() + self.minor = self._io.read_u2le() + self._debug['minor']['end'] = self._io.pos() + + + class HeaderMain(KaitaiStruct): + SEQ_FIELDS = ["image_type", "image_flags", "description", "blocks_map_offset", "offset_data", "geometry", "reserved1", "disk_size", "block_data_size", "block_metadata_size", "blocks_in_image", "blocks_allocated", "uuid_image", "uuid_last_snap", "uuid_link", "uuid_parent", "lchc_geometry"] + 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['image_type']['start'] = self._io.pos() + self.image_type = KaitaiStream.resolve_enum(self._root.ImageType, self._io.read_u4le()) + self._debug['image_type']['end'] = self._io.pos() + self._debug['image_flags']['start'] = self._io.pos() + self.image_flags = self._root.Header.HeaderMain.Flags(self._io, self, self._root) + self.image_flags._read() + self._debug['image_flags']['end'] = self._io.pos() + self._debug['description']['start'] = self._io.pos() + self.description = (self._io.read_bytes(256)).decode(u"utf-8") + self._debug['description']['end'] = self._io.pos() + if self._parent.version.major >= 1: + self._debug['blocks_map_offset']['start'] = self._io.pos() + self.blocks_map_offset = self._io.read_u4le() + self._debug['blocks_map_offset']['end'] = self._io.pos() + + if self._parent.version.major >= 1: + self._debug['offset_data']['start'] = self._io.pos() + self.offset_data = self._io.read_u4le() + self._debug['offset_data']['end'] = self._io.pos() + + self._debug['geometry']['start'] = self._io.pos() + self.geometry = self._root.Header.HeaderMain.Geometry(self._io, self, self._root) + self.geometry._read() + self._debug['geometry']['end'] = self._io.pos() + if self._parent.version.major >= 1: + self._debug['reserved1']['start'] = self._io.pos() + self.reserved1 = self._io.read_u4le() + self._debug['reserved1']['end'] = self._io.pos() + + self._debug['disk_size']['start'] = self._io.pos() + self.disk_size = self._io.read_u8le() + self._debug['disk_size']['end'] = self._io.pos() + self._debug['block_data_size']['start'] = self._io.pos() + self.block_data_size = self._io.read_u4le() + self._debug['block_data_size']['end'] = self._io.pos() + if self._parent.version.major >= 1: + self._debug['block_metadata_size']['start'] = self._io.pos() + self.block_metadata_size = self._io.read_u4le() + self._debug['block_metadata_size']['end'] = self._io.pos() + + self._debug['blocks_in_image']['start'] = self._io.pos() + self.blocks_in_image = self._io.read_u4le() + self._debug['blocks_in_image']['end'] = self._io.pos() + self._debug['blocks_allocated']['start'] = self._io.pos() + self.blocks_allocated = self._io.read_u4le() + self._debug['blocks_allocated']['end'] = self._io.pos() + self._debug['uuid_image']['start'] = self._io.pos() + self.uuid_image = self._root.Header.Uuid(self._io, self, self._root) + self.uuid_image._read() + self._debug['uuid_image']['end'] = self._io.pos() + self._debug['uuid_last_snap']['start'] = self._io.pos() + self.uuid_last_snap = self._root.Header.Uuid(self._io, self, self._root) + self.uuid_last_snap._read() + self._debug['uuid_last_snap']['end'] = self._io.pos() + self._debug['uuid_link']['start'] = self._io.pos() + self.uuid_link = self._root.Header.Uuid(self._io, self, self._root) + self.uuid_link._read() + self._debug['uuid_link']['end'] = self._io.pos() + if self._parent.version.major >= 1: + self._debug['uuid_parent']['start'] = self._io.pos() + self.uuid_parent = self._root.Header.Uuid(self._io, self, self._root) + self.uuid_parent._read() + self._debug['uuid_parent']['end'] = self._io.pos() + + if ((self._parent.version.major >= 1) and ((self._io.pos() + 16) <= self._io.size())) : + self._debug['lchc_geometry']['start'] = self._io.pos() + self.lchc_geometry = self._root.Header.HeaderMain.Geometry(self._io, self, self._root) + self.lchc_geometry._read() + self._debug['lchc_geometry']['end'] = self._io.pos() + + + class Geometry(KaitaiStruct): + SEQ_FIELDS = ["cylinders", "heads", "sectors", "sector_size"] + 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['cylinders']['start'] = self._io.pos() + self.cylinders = self._io.read_u4le() + self._debug['cylinders']['end'] = self._io.pos() + self._debug['heads']['start'] = self._io.pos() + self.heads = self._io.read_u4le() + self._debug['heads']['end'] = self._io.pos() + self._debug['sectors']['start'] = self._io.pos() + self.sectors = self._io.read_u4le() + self._debug['sectors']['end'] = self._io.pos() + self._debug['sector_size']['start'] = self._io.pos() + self.sector_size = self._io.read_u4le() + self._debug['sector_size']['end'] = self._io.pos() + + + class Flags(KaitaiStruct): + SEQ_FIELDS = ["reserved0", "zero_expand", "reserved1", "diff", "fixed", "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['reserved0']['start'] = self._io.pos() + self.reserved0 = self._io.read_bits_int(15) + self._debug['reserved0']['end'] = self._io.pos() + self._debug['zero_expand']['start'] = self._io.pos() + self.zero_expand = self._io.read_bits_int(1) != 0 + self._debug['zero_expand']['end'] = self._io.pos() + self._debug['reserved1']['start'] = self._io.pos() + self.reserved1 = self._io.read_bits_int(6) + self._debug['reserved1']['end'] = self._io.pos() + self._debug['diff']['start'] = self._io.pos() + self.diff = self._io.read_bits_int(1) != 0 + self._debug['diff']['end'] = self._io.pos() + self._debug['fixed']['start'] = self._io.pos() + self.fixed = self._io.read_bits_int(1) != 0 + self._debug['fixed']['end'] = self._io.pos() + self._debug['reserved2']['start'] = self._io.pos() + self.reserved2 = self._io.read_bits_int(8) + self._debug['reserved2']['end'] = self._io.pos() + + + + @property + def header_size(self): + if hasattr(self, '_m_header_size'): + return self._m_header_size if hasattr(self, '_m_header_size') else None + + self._m_header_size = (self.header_size_optional if self.subheader_size_is_dynamic else 336) + return self._m_header_size if hasattr(self, '_m_header_size') else None + + @property + def blocks_map_offset(self): + if hasattr(self, '_m_blocks_map_offset'): + return self._m_blocks_map_offset if hasattr(self, '_m_blocks_map_offset') else None + + self._m_blocks_map_offset = self.header_main.blocks_map_offset + return self._m_blocks_map_offset if hasattr(self, '_m_blocks_map_offset') else None + + @property + def subheader_size_is_dynamic(self): + if hasattr(self, '_m_subheader_size_is_dynamic'): + return self._m_subheader_size_is_dynamic if hasattr(self, '_m_subheader_size_is_dynamic') else None + + self._m_subheader_size_is_dynamic = self.version.major >= 1 + return self._m_subheader_size_is_dynamic if hasattr(self, '_m_subheader_size_is_dynamic') else None + + @property + def blocks_offset(self): + if hasattr(self, '_m_blocks_offset'): + return self._m_blocks_offset if hasattr(self, '_m_blocks_offset') else None + + self._m_blocks_offset = self.header_main.offset_data + return self._m_blocks_offset if hasattr(self, '_m_blocks_offset') else None + + @property + def block_size(self): + if hasattr(self, '_m_block_size'): + return self._m_block_size if hasattr(self, '_m_block_size') else None + + self._m_block_size = (self.header_main.block_metadata_size + self.header_main.block_data_size) + return self._m_block_size if hasattr(self, '_m_block_size') else None + + @property + def blocks_map_size(self): + if hasattr(self, '_m_blocks_map_size'): + return self._m_blocks_map_size if hasattr(self, '_m_blocks_map_size') else None + + self._m_blocks_map_size = ((((self.header_main.blocks_in_image * 4) + self.header_main.geometry.sector_size) - 1) // self.header_main.geometry.sector_size * self.header_main.geometry.sector_size) + return self._m_blocks_map_size if hasattr(self, '_m_blocks_map_size') else None + + + class BlocksMap(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 = [None] * (self._root.header.header_main.blocks_in_image) + for i in range(self._root.header.header_main.blocks_in_image): + if not 'arr' in self._debug['index']: + self._debug['index']['arr'] = [] + self._debug['index']['arr'].append({'start': self._io.pos()}) + _t_index = self._root.BlocksMap.BlockIndex(self._io, self, self._root) + _t_index._read() + self.index[i] = _t_index + self._debug['index']['arr'][i]['end'] = self._io.pos() + + self._debug['index']['end'] = self._io.pos() + + class BlockIndex(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_u4le() + self._debug['index']['end'] = self._io.pos() + + @property + def is_allocated(self): + if hasattr(self, '_m_is_allocated'): + return self._m_is_allocated if hasattr(self, '_m_is_allocated') else None + + self._m_is_allocated = self.index < self._root.block_discarded + return self._m_is_allocated if hasattr(self, '_m_is_allocated') else None + + @property + def block(self): + if hasattr(self, '_m_block'): + return self._m_block if hasattr(self, '_m_block') else None + + if self.is_allocated: + self._m_block = self._root.disk.blocks[self.index] + + return self._m_block if hasattr(self, '_m_block') else None + + + + class Disk(KaitaiStruct): + SEQ_FIELDS = ["blocks"] + 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['blocks']['start'] = self._io.pos() + self.blocks = [None] * (self._root.header.header_main.blocks_in_image) + for i in range(self._root.header.header_main.blocks_in_image): + if not 'arr' in self._debug['blocks']: + self._debug['blocks']['arr'] = [] + self._debug['blocks']['arr'].append({'start': self._io.pos()}) + _t_blocks = self._root.Disk.Block(self._io, self, self._root) + _t_blocks._read() + self.blocks[i] = _t_blocks + self._debug['blocks']['arr'][i]['end'] = self._io.pos() + + self._debug['blocks']['end'] = self._io.pos() + + class Block(KaitaiStruct): + SEQ_FIELDS = ["metadata", "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['metadata']['start'] = self._io.pos() + self.metadata = self._io.read_bytes(self._root.header.header_main.block_metadata_size) + self._debug['metadata']['end'] = self._io.pos() + self._debug['data']['start'] = self._io.pos() + self._raw_data = [] + self.data = [] + i = 0 + while not self._io.is_eof(): + if not 'arr' in self._debug['data']: + self._debug['data']['arr'] = [] + self._debug['data']['arr'].append({'start': self._io.pos()}) + self._raw_data.append(self._io.read_bytes(self._root.header.header_main.block_data_size)) + io = KaitaiStream(BytesIO(self._raw_data[-1])) + _t_data = self._root.Disk.Block.Sector(io, self, self._root) + _t_data._read() + self.data.append(_t_data) + self._debug['data']['arr'][len(self.data) - 1]['end'] = self._io.pos() + i += 1 + + self._debug['data']['end'] = self._io.pos() + + class Sector(KaitaiStruct): + SEQ_FIELDS = ["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['data']['start'] = self._io.pos() + self.data = self._io.read_bytes(self._root.header.header_main.geometry.sector_size) + self._debug['data']['end'] = self._io.pos() + + + + + @property + def block_discarded(self): + if hasattr(self, '_m_block_discarded'): + return self._m_block_discarded if hasattr(self, '_m_block_discarded') else None + + self._m_block_discarded = 4294967294 + return self._m_block_discarded if hasattr(self, '_m_block_discarded') else None + + @property + def block_unallocated(self): + if hasattr(self, '_m_block_unallocated'): + return self._m_block_unallocated if hasattr(self, '_m_block_unallocated') else None + + self._m_block_unallocated = 4294967295 + return self._m_block_unallocated if hasattr(self, '_m_block_unallocated') else None + + @property + def blocks_map(self): + """block_index = offset_in_virtual_disk / block_size actual_data_offset = blocks_map[block_index]*block_size+metadata_size+offset_in_block + The blocks_map will take up blocks_in_image_max * sizeof(uint32_t) bytes; since the blocks_map is read and written in a single operation, its size needs to be limited to INT_MAX; furthermore, when opening an image, the blocks_map size is rounded up to be aligned on BDRV_SECTOR_SIZE. Therefore this should satisfy the following: blocks_in_image_max * sizeof(uint32_t) + BDRV_SECTOR_SIZE == INT_MAX + 1 (INT_MAX + 1 is the first value not representable as an int) This guarantees that any value below or equal to the constant will, when multiplied by sizeof(uint32_t) and rounded up to a BDRV_SECTOR_SIZE boundary, still be below or equal to INT_MAX. + """ + if hasattr(self, '_m_blocks_map'): + return self._m_blocks_map if hasattr(self, '_m_blocks_map') else None + + _pos = self._io.pos() + self._io.seek(self.header.blocks_map_offset) + self._debug['_m_blocks_map']['start'] = self._io.pos() + self._raw__m_blocks_map = self._io.read_bytes(self.header.blocks_map_size) + io = KaitaiStream(BytesIO(self._raw__m_blocks_map)) + self._m_blocks_map = self._root.BlocksMap(io, self, self._root) + self._m_blocks_map._read() + self._debug['_m_blocks_map']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_blocks_map if hasattr(self, '_m_blocks_map') else None + + @property + def disk(self): + if hasattr(self, '_m_disk'): + return self._m_disk if hasattr(self, '_m_disk') else None + + _pos = self._io.pos() + self._io.seek(self.header.blocks_offset) + self._debug['_m_disk']['start'] = self._io.pos() + self._m_disk = self._root.Disk(self._io, self, self._root) + self._m_disk._read() + self._debug['_m_disk']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_disk if hasattr(self, '_m_disk') else None + + diff --git a/python/examples/kaitai/kaitai_struct_formats/filesystem/vfat.py b/python/examples/kaitai/kaitai_struct_formats/filesystem/vfat.py new file mode 100644 index 00000000..38ec9d3a --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/filesystem/vfat.py @@ -0,0 +1,360 @@ +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 Vfat(KaitaiStruct): + SEQ_FIELDS = ["boot_sector"] + 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['boot_sector']['start'] = self._io.pos() + self.boot_sector = self._root.BootSector(self._io, self, self._root) + self.boot_sector._read() + self._debug['boot_sector']['end'] = self._io.pos() + + class ExtBiosParamBlockFat32(KaitaiStruct): + """Extended BIOS Parameter Block for FAT32.""" + SEQ_FIELDS = ["ls_per_fat", "has_active_fat", "reserved1", "active_fat_id", "reserved2", "fat_version", "root_dir_start_clus", "ls_fs_info", "boot_sectors_copy_start_ls", "reserved3", "phys_drive_num", "reserved4", "ext_boot_sign", "volume_id", "partition_volume_label", "fs_type_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['ls_per_fat']['start'] = self._io.pos() + self.ls_per_fat = self._io.read_u4le() + self._debug['ls_per_fat']['end'] = self._io.pos() + self._debug['has_active_fat']['start'] = self._io.pos() + self.has_active_fat = self._io.read_bits_int(1) != 0 + self._debug['has_active_fat']['end'] = self._io.pos() + self._debug['reserved1']['start'] = self._io.pos() + self.reserved1 = self._io.read_bits_int(3) + self._debug['reserved1']['end'] = self._io.pos() + self._debug['active_fat_id']['start'] = self._io.pos() + self.active_fat_id = self._io.read_bits_int(4) + self._debug['active_fat_id']['end'] = self._io.pos() + self._io.align_to_byte() + self._debug['reserved2']['start'] = self._io.pos() + self.reserved2 = self._io.ensure_fixed_contents(b"\x00") + self._debug['reserved2']['end'] = self._io.pos() + self._debug['fat_version']['start'] = self._io.pos() + self.fat_version = self._io.read_u2le() + self._debug['fat_version']['end'] = self._io.pos() + self._debug['root_dir_start_clus']['start'] = self._io.pos() + self.root_dir_start_clus = self._io.read_u4le() + self._debug['root_dir_start_clus']['end'] = self._io.pos() + self._debug['ls_fs_info']['start'] = self._io.pos() + self.ls_fs_info = self._io.read_u2le() + self._debug['ls_fs_info']['end'] = self._io.pos() + self._debug['boot_sectors_copy_start_ls']['start'] = self._io.pos() + self.boot_sectors_copy_start_ls = self._io.read_u2le() + self._debug['boot_sectors_copy_start_ls']['end'] = self._io.pos() + self._debug['reserved3']['start'] = self._io.pos() + self.reserved3 = self._io.read_bytes(12) + self._debug['reserved3']['end'] = self._io.pos() + self._debug['phys_drive_num']['start'] = self._io.pos() + self.phys_drive_num = self._io.read_u1() + self._debug['phys_drive_num']['end'] = self._io.pos() + self._debug['reserved4']['start'] = self._io.pos() + self.reserved4 = self._io.read_u1() + self._debug['reserved4']['end'] = self._io.pos() + self._debug['ext_boot_sign']['start'] = self._io.pos() + self.ext_boot_sign = self._io.read_u1() + self._debug['ext_boot_sign']['end'] = self._io.pos() + self._debug['volume_id']['start'] = self._io.pos() + self.volume_id = self._io.read_bytes(4) + self._debug['volume_id']['end'] = self._io.pos() + self._debug['partition_volume_label']['start'] = self._io.pos() + self.partition_volume_label = (KaitaiStream.bytes_strip_right(self._io.read_bytes(11), 32)).decode(u"ASCII") + self._debug['partition_volume_label']['end'] = self._io.pos() + self._debug['fs_type_str']['start'] = self._io.pos() + self.fs_type_str = (KaitaiStream.bytes_strip_right(self._io.read_bytes(8), 32)).decode(u"ASCII") + self._debug['fs_type_str']['end'] = self._io.pos() + + + class BootSector(KaitaiStruct): + SEQ_FIELDS = ["jmp_instruction", "oem_name", "bpb", "ebpb_fat16", "ebpb_fat32"] + 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['jmp_instruction']['start'] = self._io.pos() + self.jmp_instruction = self._io.read_bytes(3) + self._debug['jmp_instruction']['end'] = self._io.pos() + self._debug['oem_name']['start'] = self._io.pos() + self.oem_name = (KaitaiStream.bytes_strip_right(self._io.read_bytes(8), 32)).decode(u"ASCII") + self._debug['oem_name']['end'] = self._io.pos() + self._debug['bpb']['start'] = self._io.pos() + self.bpb = self._root.BiosParamBlock(self._io, self, self._root) + self.bpb._read() + self._debug['bpb']['end'] = self._io.pos() + if not (self.is_fat32): + self._debug['ebpb_fat16']['start'] = self._io.pos() + self.ebpb_fat16 = self._root.ExtBiosParamBlockFat16(self._io, self, self._root) + self.ebpb_fat16._read() + self._debug['ebpb_fat16']['end'] = self._io.pos() + + if self.is_fat32: + self._debug['ebpb_fat32']['start'] = self._io.pos() + self.ebpb_fat32 = self._root.ExtBiosParamBlockFat32(self._io, self, self._root) + self.ebpb_fat32._read() + self._debug['ebpb_fat32']['end'] = self._io.pos() + + + @property + def pos_fats(self): + """Offset of FATs in bytes from start of filesystem.""" + if hasattr(self, '_m_pos_fats'): + return self._m_pos_fats if hasattr(self, '_m_pos_fats') else None + + self._m_pos_fats = (self.bpb.bytes_per_ls * self.bpb.num_reserved_ls) + return self._m_pos_fats if hasattr(self, '_m_pos_fats') else None + + @property + def ls_per_fat(self): + if hasattr(self, '_m_ls_per_fat'): + return self._m_ls_per_fat if hasattr(self, '_m_ls_per_fat') else None + + self._m_ls_per_fat = (self.ebpb_fat32.ls_per_fat if self.is_fat32 else self.bpb.ls_per_fat) + return self._m_ls_per_fat if hasattr(self, '_m_ls_per_fat') else None + + @property + def ls_per_root_dir(self): + """Size of root directory in logical sectors. + + .. seealso:: + FAT: General Overview of On-Disk Format, section "FAT Data Structure" + """ + if hasattr(self, '_m_ls_per_root_dir'): + return self._m_ls_per_root_dir if hasattr(self, '_m_ls_per_root_dir') else None + + self._m_ls_per_root_dir = (((self.bpb.max_root_dir_rec * 32) + self.bpb.bytes_per_ls) - 1) // self.bpb.bytes_per_ls + return self._m_ls_per_root_dir if hasattr(self, '_m_ls_per_root_dir') else None + + @property + def is_fat32(self): + """Determines if filesystem is FAT32 (true) or FAT12/16 (false) + by analyzing some preliminary conditions in BPB. Used to + determine whether we should parse post-BPB data as + `ext_bios_param_block_fat16` or `ext_bios_param_block_fat32`. + """ + if hasattr(self, '_m_is_fat32'): + return self._m_is_fat32 if hasattr(self, '_m_is_fat32') else None + + self._m_is_fat32 = self.bpb.max_root_dir_rec == 0 + return self._m_is_fat32 if hasattr(self, '_m_is_fat32') else None + + @property + def size_fat(self): + """Size of one FAT in bytes.""" + if hasattr(self, '_m_size_fat'): + return self._m_size_fat if hasattr(self, '_m_size_fat') else None + + self._m_size_fat = (self.bpb.bytes_per_ls * self.ls_per_fat) + return self._m_size_fat if hasattr(self, '_m_size_fat') else None + + @property + def pos_root_dir(self): + """Offset of root directory in bytes from start of filesystem.""" + if hasattr(self, '_m_pos_root_dir'): + return self._m_pos_root_dir if hasattr(self, '_m_pos_root_dir') else None + + self._m_pos_root_dir = (self.bpb.bytes_per_ls * (self.bpb.num_reserved_ls + (self.ls_per_fat * self.bpb.num_fats))) + return self._m_pos_root_dir if hasattr(self, '_m_pos_root_dir') else None + + @property + def size_root_dir(self): + """Size of root directory in bytes.""" + if hasattr(self, '_m_size_root_dir'): + return self._m_size_root_dir if hasattr(self, '_m_size_root_dir') else None + + self._m_size_root_dir = (self.ls_per_root_dir * self.bpb.bytes_per_ls) + return self._m_size_root_dir if hasattr(self, '_m_size_root_dir') else None + + + class BiosParamBlock(KaitaiStruct): + SEQ_FIELDS = ["bytes_per_ls", "ls_per_clus", "num_reserved_ls", "num_fats", "max_root_dir_rec", "total_ls_2", "media_code", "ls_per_fat", "ps_per_track", "num_heads", "num_hidden_sectors", "total_ls_4"] + 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['bytes_per_ls']['start'] = self._io.pos() + self.bytes_per_ls = self._io.read_u2le() + self._debug['bytes_per_ls']['end'] = self._io.pos() + self._debug['ls_per_clus']['start'] = self._io.pos() + self.ls_per_clus = self._io.read_u1() + self._debug['ls_per_clus']['end'] = self._io.pos() + self._debug['num_reserved_ls']['start'] = self._io.pos() + self.num_reserved_ls = self._io.read_u2le() + self._debug['num_reserved_ls']['end'] = self._io.pos() + self._debug['num_fats']['start'] = self._io.pos() + self.num_fats = self._io.read_u1() + self._debug['num_fats']['end'] = self._io.pos() + self._debug['max_root_dir_rec']['start'] = self._io.pos() + self.max_root_dir_rec = self._io.read_u2le() + self._debug['max_root_dir_rec']['end'] = self._io.pos() + self._debug['total_ls_2']['start'] = self._io.pos() + self.total_ls_2 = self._io.read_u2le() + self._debug['total_ls_2']['end'] = self._io.pos() + self._debug['media_code']['start'] = self._io.pos() + self.media_code = self._io.read_u1() + self._debug['media_code']['end'] = self._io.pos() + self._debug['ls_per_fat']['start'] = self._io.pos() + self.ls_per_fat = self._io.read_u2le() + self._debug['ls_per_fat']['end'] = self._io.pos() + self._debug['ps_per_track']['start'] = self._io.pos() + self.ps_per_track = self._io.read_u2le() + self._debug['ps_per_track']['end'] = self._io.pos() + self._debug['num_heads']['start'] = self._io.pos() + self.num_heads = self._io.read_u2le() + self._debug['num_heads']['end'] = self._io.pos() + self._debug['num_hidden_sectors']['start'] = self._io.pos() + self.num_hidden_sectors = self._io.read_u4le() + self._debug['num_hidden_sectors']['end'] = self._io.pos() + self._debug['total_ls_4']['start'] = self._io.pos() + self.total_ls_4 = self._io.read_u4le() + self._debug['total_ls_4']['end'] = self._io.pos() + + + class RootDirectoryRec(KaitaiStruct): + SEQ_FIELDS = ["file_name", "attribute", "reserved", "time", "date", "start_clus", "file_size"] + 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['file_name']['start'] = self._io.pos() + self.file_name = self._io.read_bytes(11) + self._debug['file_name']['end'] = self._io.pos() + self._debug['attribute']['start'] = self._io.pos() + self.attribute = self._io.read_u1() + self._debug['attribute']['end'] = self._io.pos() + self._debug['reserved']['start'] = self._io.pos() + self.reserved = self._io.read_bytes(10) + self._debug['reserved']['end'] = self._io.pos() + self._debug['time']['start'] = self._io.pos() + self.time = self._io.read_u2le() + self._debug['time']['end'] = self._io.pos() + self._debug['date']['start'] = self._io.pos() + self.date = self._io.read_u2le() + self._debug['date']['end'] = self._io.pos() + self._debug['start_clus']['start'] = self._io.pos() + self.start_clus = self._io.read_u2le() + self._debug['start_clus']['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() + + + class RootDirectory(KaitaiStruct): + SEQ_FIELDS = ["records"] + 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['records']['start'] = self._io.pos() + self.records = [None] * (self._root.boot_sector.bpb.max_root_dir_rec) + for i in range(self._root.boot_sector.bpb.max_root_dir_rec): + if not 'arr' in self._debug['records']: + self._debug['records']['arr'] = [] + self._debug['records']['arr'].append({'start': self._io.pos()}) + _t_records = self._root.RootDirectoryRec(self._io, self, self._root) + _t_records._read() + self.records[i] = _t_records + self._debug['records']['arr'][i]['end'] = self._io.pos() + + self._debug['records']['end'] = self._io.pos() + + + class ExtBiosParamBlockFat16(KaitaiStruct): + """Extended BIOS Parameter Block (DOS 4.0+, OS/2 1.0+). Used only + for FAT12 and FAT16. + """ + SEQ_FIELDS = ["phys_drive_num", "reserved1", "ext_boot_sign", "volume_id", "partition_volume_label", "fs_type_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['phys_drive_num']['start'] = self._io.pos() + self.phys_drive_num = self._io.read_u1() + self._debug['phys_drive_num']['end'] = self._io.pos() + self._debug['reserved1']['start'] = self._io.pos() + self.reserved1 = self._io.read_u1() + self._debug['reserved1']['end'] = self._io.pos() + self._debug['ext_boot_sign']['start'] = self._io.pos() + self.ext_boot_sign = self._io.read_u1() + self._debug['ext_boot_sign']['end'] = self._io.pos() + self._debug['volume_id']['start'] = self._io.pos() + self.volume_id = self._io.read_bytes(4) + self._debug['volume_id']['end'] = self._io.pos() + self._debug['partition_volume_label']['start'] = self._io.pos() + self.partition_volume_label = (KaitaiStream.bytes_strip_right(self._io.read_bytes(11), 32)).decode(u"ASCII") + self._debug['partition_volume_label']['end'] = self._io.pos() + self._debug['fs_type_str']['start'] = self._io.pos() + self.fs_type_str = (KaitaiStream.bytes_strip_right(self._io.read_bytes(8), 32)).decode(u"ASCII") + self._debug['fs_type_str']['end'] = self._io.pos() + + + @property + def fats(self): + if hasattr(self, '_m_fats'): + return self._m_fats if hasattr(self, '_m_fats') else None + + _pos = self._io.pos() + self._io.seek(self.boot_sector.pos_fats) + self._debug['_m_fats']['start'] = self._io.pos() + self._m_fats = [None] * (self.boot_sector.bpb.num_fats) + for i in range(self.boot_sector.bpb.num_fats): + if not 'arr' in self._debug['_m_fats']: + self._debug['_m_fats']['arr'] = [] + self._debug['_m_fats']['arr'].append({'start': self._io.pos()}) + self._m_fats[i] = self._io.read_bytes(self.boot_sector.size_fat) + self._debug['_m_fats']['arr'][i]['end'] = self._io.pos() + + self._debug['_m_fats']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_fats if hasattr(self, '_m_fats') else None + + @property + def root_dir(self): + if hasattr(self, '_m_root_dir'): + return self._m_root_dir if hasattr(self, '_m_root_dir') else None + + _pos = self._io.pos() + self._io.seek(self.boot_sector.pos_root_dir) + self._debug['_m_root_dir']['start'] = self._io.pos() + self._raw__m_root_dir = self._io.read_bytes(self.boot_sector.size_root_dir) + io = KaitaiStream(BytesIO(self._raw__m_root_dir)) + self._m_root_dir = self._root.RootDirectory(io, self, self._root) + self._m_root_dir._read() + self._debug['_m_root_dir']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_root_dir if hasattr(self, '_m_root_dir') else None + + diff --git a/python/examples/kaitai/kaitai_struct_formats/filesystem/vmware_vmdk.py b/python/examples/kaitai/kaitai_struct_formats/filesystem/vmware_vmdk.py new file mode 100644 index 00000000..79ad7e9a --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/filesystem/vmware_vmdk.py @@ -0,0 +1,165 @@ +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 VmwareVmdk(KaitaiStruct): + """ + .. seealso:: + Source - https://github.com/libyal/libvmdk/blob/master/documentation/VMWare%20Virtual%20Disk%20Format%20(VMDK).asciidoc#41-file-header + """ + + class CompressionMethods(Enum): + none = 0 + deflate = 1 + SEQ_FIELDS = ["magic", "version", "flags", "size_max", "size_grain", "start_descriptor", "size_descriptor", "num_grain_table_entries", "start_secondary_grain", "start_primary_grain", "size_metadata", "is_dirty", "stuff", "compression_method"] + 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"\x4B\x44\x4D\x56") + self._debug['magic']['end'] = self._io.pos() + self._debug['version']['start'] = self._io.pos() + self.version = self._io.read_s4le() + self._debug['version']['end'] = self._io.pos() + self._debug['flags']['start'] = self._io.pos() + self.flags = self._root.HeaderFlags(self._io, self, self._root) + self.flags._read() + self._debug['flags']['end'] = self._io.pos() + self._debug['size_max']['start'] = self._io.pos() + self.size_max = self._io.read_s8le() + self._debug['size_max']['end'] = self._io.pos() + self._debug['size_grain']['start'] = self._io.pos() + self.size_grain = self._io.read_s8le() + self._debug['size_grain']['end'] = self._io.pos() + self._debug['start_descriptor']['start'] = self._io.pos() + self.start_descriptor = self._io.read_s8le() + self._debug['start_descriptor']['end'] = self._io.pos() + self._debug['size_descriptor']['start'] = self._io.pos() + self.size_descriptor = self._io.read_s8le() + self._debug['size_descriptor']['end'] = self._io.pos() + self._debug['num_grain_table_entries']['start'] = self._io.pos() + self.num_grain_table_entries = self._io.read_s4le() + self._debug['num_grain_table_entries']['end'] = self._io.pos() + self._debug['start_secondary_grain']['start'] = self._io.pos() + self.start_secondary_grain = self._io.read_s8le() + self._debug['start_secondary_grain']['end'] = self._io.pos() + self._debug['start_primary_grain']['start'] = self._io.pos() + self.start_primary_grain = self._io.read_s8le() + self._debug['start_primary_grain']['end'] = self._io.pos() + self._debug['size_metadata']['start'] = self._io.pos() + self.size_metadata = self._io.read_s8le() + self._debug['size_metadata']['end'] = self._io.pos() + self._debug['is_dirty']['start'] = self._io.pos() + self.is_dirty = self._io.read_u1() + self._debug['is_dirty']['end'] = self._io.pos() + self._debug['stuff']['start'] = self._io.pos() + self.stuff = self._io.read_bytes(4) + self._debug['stuff']['end'] = self._io.pos() + self._debug['compression_method']['start'] = self._io.pos() + self.compression_method = KaitaiStream.resolve_enum(self._root.CompressionMethods, self._io.read_u2le()) + self._debug['compression_method']['end'] = self._io.pos() + + class HeaderFlags(KaitaiStruct): + """ + .. seealso:: + Source - https://github.com/libyal/libvmdk/blob/master/documentation/VMWare%20Virtual%20Disk%20Format%20(VMDK).asciidoc#411-flags + """ + SEQ_FIELDS = ["reserved1", "zeroed_grain_table_entry", "use_secondary_grain_dir", "valid_new_line_detection_test", "reserved2", "reserved3", "has_metadata", "has_compressed_grain", "reserved4"] + 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['reserved1']['start'] = self._io.pos() + self.reserved1 = self._io.read_bits_int(5) + self._debug['reserved1']['end'] = self._io.pos() + self._debug['zeroed_grain_table_entry']['start'] = self._io.pos() + self.zeroed_grain_table_entry = self._io.read_bits_int(1) != 0 + self._debug['zeroed_grain_table_entry']['end'] = self._io.pos() + self._debug['use_secondary_grain_dir']['start'] = self._io.pos() + self.use_secondary_grain_dir = self._io.read_bits_int(1) != 0 + self._debug['use_secondary_grain_dir']['end'] = self._io.pos() + self._debug['valid_new_line_detection_test']['start'] = self._io.pos() + self.valid_new_line_detection_test = self._io.read_bits_int(1) != 0 + self._debug['valid_new_line_detection_test']['end'] = self._io.pos() + self._io.align_to_byte() + self._debug['reserved2']['start'] = self._io.pos() + self.reserved2 = self._io.read_u1() + self._debug['reserved2']['end'] = self._io.pos() + self._debug['reserved3']['start'] = self._io.pos() + self.reserved3 = self._io.read_bits_int(6) + self._debug['reserved3']['end'] = self._io.pos() + self._debug['has_metadata']['start'] = self._io.pos() + self.has_metadata = self._io.read_bits_int(1) != 0 + self._debug['has_metadata']['end'] = self._io.pos() + self._debug['has_compressed_grain']['start'] = self._io.pos() + self.has_compressed_grain = self._io.read_bits_int(1) != 0 + self._debug['has_compressed_grain']['end'] = self._io.pos() + self._io.align_to_byte() + self._debug['reserved4']['start'] = self._io.pos() + self.reserved4 = self._io.read_u1() + self._debug['reserved4']['end'] = self._io.pos() + + + @property + def len_sector(self): + if hasattr(self, '_m_len_sector'): + return self._m_len_sector if hasattr(self, '_m_len_sector') else None + + self._m_len_sector = 512 + return self._m_len_sector if hasattr(self, '_m_len_sector') else None + + @property + def descriptor(self): + if hasattr(self, '_m_descriptor'): + return self._m_descriptor if hasattr(self, '_m_descriptor') else None + + _pos = self._io.pos() + self._io.seek((self.start_descriptor * self._root.len_sector)) + self._debug['_m_descriptor']['start'] = self._io.pos() + self._m_descriptor = self._io.read_bytes((self.size_descriptor * self._root.len_sector)) + self._debug['_m_descriptor']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_descriptor if hasattr(self, '_m_descriptor') else None + + @property + def grain_primary(self): + if hasattr(self, '_m_grain_primary'): + return self._m_grain_primary if hasattr(self, '_m_grain_primary') else None + + _pos = self._io.pos() + self._io.seek((self.start_primary_grain * self._root.len_sector)) + self._debug['_m_grain_primary']['start'] = self._io.pos() + self._m_grain_primary = self._io.read_bytes((self.size_grain * self._root.len_sector)) + self._debug['_m_grain_primary']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_grain_primary if hasattr(self, '_m_grain_primary') else None + + @property + def grain_secondary(self): + if hasattr(self, '_m_grain_secondary'): + return self._m_grain_secondary if hasattr(self, '_m_grain_secondary') else None + + _pos = self._io.pos() + self._io.seek((self.start_secondary_grain * self._root.len_sector)) + self._debug['_m_grain_secondary']['start'] = self._io.pos() + self._m_grain_secondary = self._io.read_bytes((self.size_grain * self._root.len_sector)) + self._debug['_m_grain_secondary']['end'] = self._io.pos() + self._io.seek(_pos) + return self._m_grain_secondary if hasattr(self, '_m_grain_secondary') else None + + |
