diff options
| author | Andrew Lamoureux <andrew@vector35.com> | 2019-03-19 17:32:44 -0400 |
|---|---|---|
| committer | Rusty Wagner <rusty@vector35.com> | 2019-03-20 13:00:17 -0400 |
| commit | 24ead55316ac8497df5842caf4e1d51f2b799787 (patch) | |
| tree | 3dfdfae8a076445c9d24bf6d1b40e57f36ebf94a /python/examples/kaitai/kaitai_struct_formats/network | |
| parent | 411acbaa4a1d217e731d3df0b189445117be3b26 (diff) | |
kaitai: adjustments and rearrangements for python2/3 compat
Diffstat (limited to 'python/examples/kaitai/kaitai_struct_formats/network')
19 files changed, 0 insertions, 3670 deletions
diff --git a/python/examples/kaitai/kaitai_struct_formats/network/__init__.py b/python/examples/kaitai/kaitai_struct_formats/network/__init__.py deleted file mode 100644 index e69de29b..00000000 --- a/python/examples/kaitai/kaitai_struct_formats/network/__init__.py +++ /dev/null diff --git a/python/examples/kaitai/kaitai_struct_formats/network/bitcoin_transaction.py b/python/examples/kaitai/kaitai_struct_formats/network/bitcoin_transaction.py deleted file mode 100644 index 41d05158..00000000 --- a/python/examples/kaitai/kaitai_struct_formats/network/bitcoin_transaction.py +++ /dev/null @@ -1,202 +0,0 @@ -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 BitcoinTransaction(KaitaiStruct): - """ - .. seealso:: - Source - https://bitcoin.org/en/developer-guide#transactions - https://en.bitcoin.it/wiki/Transaction - """ - - class SighashType(Enum): - sighash_all = 1 - sighash_none = 2 - sighash_single = 3 - sighash_anyonecanpay = 80 - SEQ_FIELDS = ["version", "num_vins", "vins", "num_vouts", "vouts", "locktime"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['version']['start'] = self._io.pos() - self.version = self._io.read_u4le() - self._debug['version']['end'] = self._io.pos() - self._debug['num_vins']['start'] = self._io.pos() - self.num_vins = self._io.read_u1() - self._debug['num_vins']['end'] = self._io.pos() - self._debug['vins']['start'] = self._io.pos() - self.vins = [None] * (self.num_vins) - for i in range(self.num_vins): - if not 'arr' in self._debug['vins']: - self._debug['vins']['arr'] = [] - self._debug['vins']['arr'].append({'start': self._io.pos()}) - _t_vins = self._root.Vin(self._io, self, self._root) - _t_vins._read() - self.vins[i] = _t_vins - self._debug['vins']['arr'][i]['end'] = self._io.pos() - - self._debug['vins']['end'] = self._io.pos() - self._debug['num_vouts']['start'] = self._io.pos() - self.num_vouts = self._io.read_u1() - self._debug['num_vouts']['end'] = self._io.pos() - self._debug['vouts']['start'] = self._io.pos() - self.vouts = [None] * (self.num_vouts) - for i in range(self.num_vouts): - if not 'arr' in self._debug['vouts']: - self._debug['vouts']['arr'] = [] - self._debug['vouts']['arr'].append({'start': self._io.pos()}) - _t_vouts = self._root.Vout(self._io, self, self._root) - _t_vouts._read() - self.vouts[i] = _t_vouts - self._debug['vouts']['arr'][i]['end'] = self._io.pos() - - self._debug['vouts']['end'] = self._io.pos() - self._debug['locktime']['start'] = self._io.pos() - self.locktime = self._io.read_u4le() - self._debug['locktime']['end'] = self._io.pos() - - class Vout(KaitaiStruct): - SEQ_FIELDS = ["amount", "script_len", "script_pub_key"] - 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['amount']['start'] = self._io.pos() - self.amount = self._io.read_u8le() - self._debug['amount']['end'] = self._io.pos() - self._debug['script_len']['start'] = self._io.pos() - self.script_len = self._io.read_u1() - self._debug['script_len']['end'] = self._io.pos() - self._debug['script_pub_key']['start'] = self._io.pos() - self.script_pub_key = self._io.read_bytes(self.script_len) - self._debug['script_pub_key']['end'] = self._io.pos() - - - class PublicKey(KaitaiStruct): - SEQ_FIELDS = ["type", "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['type']['start'] = self._io.pos() - self.type = self._io.read_u1() - self._debug['type']['end'] = self._io.pos() - self._debug['x']['start'] = self._io.pos() - self.x = self._io.read_bytes(32) - self._debug['x']['end'] = self._io.pos() - self._debug['y']['start'] = self._io.pos() - self.y = self._io.read_bytes(32) - self._debug['y']['end'] = self._io.pos() - - - class Vin(KaitaiStruct): - SEQ_FIELDS = ["txid", "output_id", "script_len", "script_sig", "end_of_vin"] - 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['txid']['start'] = self._io.pos() - self.txid = self._io.read_bytes(32) - self._debug['txid']['end'] = self._io.pos() - self._debug['output_id']['start'] = self._io.pos() - self.output_id = self._io.read_u4le() - self._debug['output_id']['end'] = self._io.pos() - self._debug['script_len']['start'] = self._io.pos() - self.script_len = self._io.read_u1() - self._debug['script_len']['end'] = self._io.pos() - self._debug['script_sig']['start'] = self._io.pos() - self._raw_script_sig = self._io.read_bytes(self.script_len) - io = KaitaiStream(BytesIO(self._raw_script_sig)) - self.script_sig = self._root.ScriptSignature(io, self, self._root) - self.script_sig._read() - self._debug['script_sig']['end'] = self._io.pos() - self._debug['end_of_vin']['start'] = self._io.pos() - self.end_of_vin = self._io.ensure_fixed_contents(b"\xFF\xFF\xFF\xFF") - self._debug['end_of_vin']['end'] = self._io.pos() - - - class ScriptSignature(KaitaiStruct): - SEQ_FIELDS = ["sig_stack_len", "der_sig", "sig_type", "pubkey_stack_len", "pubkey"] - 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['sig_stack_len']['start'] = self._io.pos() - self.sig_stack_len = self._io.read_u1() - self._debug['sig_stack_len']['end'] = self._io.pos() - self._debug['der_sig']['start'] = self._io.pos() - self.der_sig = self._root.DerSignature(self._io, self, self._root) - self.der_sig._read() - self._debug['der_sig']['end'] = self._io.pos() - self._debug['sig_type']['start'] = self._io.pos() - self.sig_type = KaitaiStream.resolve_enum(self._root.SighashType, self._io.read_u1()) - self._debug['sig_type']['end'] = self._io.pos() - self._debug['pubkey_stack_len']['start'] = self._io.pos() - self.pubkey_stack_len = self._io.read_u1() - self._debug['pubkey_stack_len']['end'] = self._io.pos() - self._debug['pubkey']['start'] = self._io.pos() - self.pubkey = self._root.PublicKey(self._io, self, self._root) - self.pubkey._read() - self._debug['pubkey']['end'] = self._io.pos() - - - class DerSignature(KaitaiStruct): - SEQ_FIELDS = ["sequence", "sig_len", "sep_1", "sig_r_len", "sig_r", "sep_2", "sig_s_len", "sig_s"] - 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['sequence']['start'] = self._io.pos() - self.sequence = self._io.ensure_fixed_contents(b"\x30") - self._debug['sequence']['end'] = self._io.pos() - self._debug['sig_len']['start'] = self._io.pos() - self.sig_len = self._io.read_u1() - self._debug['sig_len']['end'] = self._io.pos() - self._debug['sep_1']['start'] = self._io.pos() - self.sep_1 = self._io.ensure_fixed_contents(b"\x02") - self._debug['sep_1']['end'] = self._io.pos() - self._debug['sig_r_len']['start'] = self._io.pos() - self.sig_r_len = self._io.read_u1() - self._debug['sig_r_len']['end'] = self._io.pos() - self._debug['sig_r']['start'] = self._io.pos() - self.sig_r = self._io.read_bytes(self.sig_r_len) - self._debug['sig_r']['end'] = self._io.pos() - self._debug['sep_2']['start'] = self._io.pos() - self.sep_2 = self._io.ensure_fixed_contents(b"\x02") - self._debug['sep_2']['end'] = self._io.pos() - self._debug['sig_s_len']['start'] = self._io.pos() - self.sig_s_len = self._io.read_u1() - self._debug['sig_s_len']['end'] = self._io.pos() - self._debug['sig_s']['start'] = self._io.pos() - self.sig_s = self._io.read_bytes(self.sig_s_len) - self._debug['sig_s']['end'] = self._io.pos() - - - diff --git a/python/examples/kaitai/kaitai_struct_formats/network/dns_packet.py b/python/examples/kaitai/kaitai_struct_formats/network/dns_packet.py deleted file mode 100644 index c90b66da..00000000 --- a/python/examples/kaitai/kaitai_struct_formats/network/dns_packet.py +++ /dev/null @@ -1,356 +0,0 @@ -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 DnsPacket(KaitaiStruct): - """(No support for Auth-Name + Add-Name for simplicity) - """ - - class ClassType(Enum): - in_class = 1 - cs = 2 - ch = 3 - hs = 4 - - class TypeType(Enum): - a = 1 - ns = 2 - md = 3 - mf = 4 - cname = 5 - soe = 6 - mb = 7 - mg = 8 - mr = 9 - null = 10 - wks = 11 - ptr = 12 - hinfo = 13 - minfo = 14 - mx = 15 - txt = 16 - SEQ_FIELDS = ["transaction_id", "flags", "qdcount", "ancount", "nscount", "arcount", "queries", "answers"] - 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['transaction_id']['start'] = self._io.pos() - self.transaction_id = self._io.read_u2be() - self._debug['transaction_id']['end'] = self._io.pos() - self._debug['flags']['start'] = self._io.pos() - self.flags = self._root.PacketFlags(self._io, self, self._root) - self.flags._read() - self._debug['flags']['end'] = self._io.pos() - self._debug['qdcount']['start'] = self._io.pos() - self.qdcount = self._io.read_u2be() - self._debug['qdcount']['end'] = self._io.pos() - self._debug['ancount']['start'] = self._io.pos() - self.ancount = self._io.read_u2be() - self._debug['ancount']['end'] = self._io.pos() - self._debug['nscount']['start'] = self._io.pos() - self.nscount = self._io.read_u2be() - self._debug['nscount']['end'] = self._io.pos() - self._debug['arcount']['start'] = self._io.pos() - self.arcount = self._io.read_u2be() - self._debug['arcount']['end'] = self._io.pos() - self._debug['queries']['start'] = self._io.pos() - self.queries = [None] * (self.qdcount) - for i in range(self.qdcount): - if not 'arr' in self._debug['queries']: - self._debug['queries']['arr'] = [] - self._debug['queries']['arr'].append({'start': self._io.pos()}) - _t_queries = self._root.Query(self._io, self, self._root) - _t_queries._read() - self.queries[i] = _t_queries - self._debug['queries']['arr'][i]['end'] = self._io.pos() - - self._debug['queries']['end'] = self._io.pos() - self._debug['answers']['start'] = self._io.pos() - self.answers = [None] * (self.ancount) - for i in range(self.ancount): - if not 'arr' in self._debug['answers']: - self._debug['answers']['arr'] = [] - self._debug['answers']['arr'].append({'start': self._io.pos()}) - _t_answers = self._root.Answer(self._io, self, self._root) - _t_answers._read() - self.answers[i] = _t_answers - self._debug['answers']['arr'][i]['end'] = self._io.pos() - - self._debug['answers']['end'] = self._io.pos() - - class PointerStruct(KaitaiStruct): - SEQ_FIELDS = ["value"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['value']['start'] = self._io.pos() - self.value = self._io.read_u1() - self._debug['value']['end'] = self._io.pos() - - @property - def contents(self): - if hasattr(self, '_m_contents'): - return self._m_contents if hasattr(self, '_m_contents') else None - - io = self._root._io - _pos = io.pos() - io.seek(self.value) - self._debug['_m_contents']['start'] = io.pos() - self._m_contents = self._root.DomainName(io, self, self._root) - self._m_contents._read() - self._debug['_m_contents']['end'] = io.pos() - io.seek(_pos) - return self._m_contents if hasattr(self, '_m_contents') else None - - - class Label(KaitaiStruct): - SEQ_FIELDS = ["length", "pointer", "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['length']['start'] = self._io.pos() - self.length = self._io.read_u1() - self._debug['length']['end'] = self._io.pos() - if self.is_pointer: - self._debug['pointer']['start'] = self._io.pos() - self.pointer = self._root.PointerStruct(self._io, self, self._root) - self.pointer._read() - self._debug['pointer']['end'] = self._io.pos() - - if not (self.is_pointer): - self._debug['name']['start'] = self._io.pos() - self.name = (self._io.read_bytes(self.length)).decode(u"ASCII") - self._debug['name']['end'] = self._io.pos() - - - @property - def is_pointer(self): - if hasattr(self, '_m_is_pointer'): - return self._m_is_pointer if hasattr(self, '_m_is_pointer') else None - - self._m_is_pointer = self.length == 192 - return self._m_is_pointer if hasattr(self, '_m_is_pointer') else None - - - class Query(KaitaiStruct): - SEQ_FIELDS = ["name", "type", "query_class"] - 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._root.DomainName(self._io, self, self._root) - self.name._read() - self._debug['name']['end'] = self._io.pos() - self._debug['type']['start'] = self._io.pos() - self.type = KaitaiStream.resolve_enum(self._root.TypeType, self._io.read_u2be()) - self._debug['type']['end'] = self._io.pos() - self._debug['query_class']['start'] = self._io.pos() - self.query_class = KaitaiStream.resolve_enum(self._root.ClassType, self._io.read_u2be()) - self._debug['query_class']['end'] = self._io.pos() - - - class DomainName(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 = [] - i = 0 - while True: - if not 'arr' in self._debug['name']: - self._debug['name']['arr'] = [] - self._debug['name']['arr'].append({'start': self._io.pos()}) - _t_name = self._root.Label(self._io, self, self._root) - _t_name._read() - _ = _t_name - self.name.append(_) - self._debug['name']['arr'][len(self.name) - 1]['end'] = self._io.pos() - if ((_.length == 0) or (_.length == 192)) : - break - i += 1 - self._debug['name']['end'] = self._io.pos() - - - class Address(KaitaiStruct): - SEQ_FIELDS = ["ip"] - 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['ip']['start'] = self._io.pos() - self.ip = [None] * (4) - for i in range(4): - if not 'arr' in self._debug['ip']: - self._debug['ip']['arr'] = [] - self._debug['ip']['arr'].append({'start': self._io.pos()}) - self.ip[i] = self._io.read_u1() - self._debug['ip']['arr'][i]['end'] = self._io.pos() - - self._debug['ip']['end'] = self._io.pos() - - - class Answer(KaitaiStruct): - SEQ_FIELDS = ["name", "type", "answer_class", "ttl", "rdlength", "ptrdname", "address"] - 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._root.DomainName(self._io, self, self._root) - self.name._read() - self._debug['name']['end'] = self._io.pos() - self._debug['type']['start'] = self._io.pos() - self.type = KaitaiStream.resolve_enum(self._root.TypeType, self._io.read_u2be()) - self._debug['type']['end'] = self._io.pos() - self._debug['answer_class']['start'] = self._io.pos() - self.answer_class = KaitaiStream.resolve_enum(self._root.ClassType, self._io.read_u2be()) - self._debug['answer_class']['end'] = self._io.pos() - self._debug['ttl']['start'] = self._io.pos() - self.ttl = self._io.read_s4be() - self._debug['ttl']['end'] = self._io.pos() - self._debug['rdlength']['start'] = self._io.pos() - self.rdlength = self._io.read_u2be() - self._debug['rdlength']['end'] = self._io.pos() - if self.type == self._root.TypeType.ptr: - self._debug['ptrdname']['start'] = self._io.pos() - self.ptrdname = self._root.DomainName(self._io, self, self._root) - self.ptrdname._read() - self._debug['ptrdname']['end'] = self._io.pos() - - if self.type == self._root.TypeType.a: - self._debug['address']['start'] = self._io.pos() - self.address = self._root.Address(self._io, self, self._root) - self.address._read() - self._debug['address']['end'] = self._io.pos() - - - - class PacketFlags(KaitaiStruct): - SEQ_FIELDS = ["flag"] - 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['flag']['start'] = self._io.pos() - self.flag = self._io.read_u2be() - self._debug['flag']['end'] = self._io.pos() - - @property - def qr(self): - if hasattr(self, '_m_qr'): - return self._m_qr if hasattr(self, '_m_qr') else None - - self._m_qr = ((self.flag & 32768) >> 15) - return self._m_qr if hasattr(self, '_m_qr') else None - - @property - def ra(self): - if hasattr(self, '_m_ra'): - return self._m_ra if hasattr(self, '_m_ra') else None - - self._m_ra = ((self.flag & 128) >> 7) - return self._m_ra if hasattr(self, '_m_ra') else None - - @property - def tc(self): - if hasattr(self, '_m_tc'): - return self._m_tc if hasattr(self, '_m_tc') else None - - self._m_tc = ((self.flag & 512) >> 9) - return self._m_tc if hasattr(self, '_m_tc') else None - - @property - def rcode(self): - if hasattr(self, '_m_rcode'): - return self._m_rcode if hasattr(self, '_m_rcode') else None - - self._m_rcode = ((self.flag & 15) >> 0) - return self._m_rcode if hasattr(self, '_m_rcode') else None - - @property - def opcode(self): - if hasattr(self, '_m_opcode'): - return self._m_opcode if hasattr(self, '_m_opcode') else None - - self._m_opcode = ((self.flag & 30720) >> 11) - return self._m_opcode if hasattr(self, '_m_opcode') else None - - @property - def aa(self): - if hasattr(self, '_m_aa'): - return self._m_aa if hasattr(self, '_m_aa') else None - - self._m_aa = ((self.flag & 1024) >> 10) - return self._m_aa if hasattr(self, '_m_aa') else None - - @property - def z(self): - if hasattr(self, '_m_z'): - return self._m_z if hasattr(self, '_m_z') else None - - self._m_z = ((self.flag & 64) >> 6) - return self._m_z if hasattr(self, '_m_z') else None - - @property - def rd(self): - if hasattr(self, '_m_rd'): - return self._m_rd if hasattr(self, '_m_rd') else None - - self._m_rd = ((self.flag & 256) >> 8) - return self._m_rd if hasattr(self, '_m_rd') else None - - @property - def cd(self): - if hasattr(self, '_m_cd'): - return self._m_cd if hasattr(self, '_m_cd') else None - - self._m_cd = ((self.flag & 16) >> 4) - return self._m_cd if hasattr(self, '_m_cd') else None - - @property - def ad(self): - if hasattr(self, '_m_ad'): - return self._m_ad if hasattr(self, '_m_ad') else None - - self._m_ad = ((self.flag & 32) >> 5) - return self._m_ad if hasattr(self, '_m_ad') else None - - - diff --git a/python/examples/kaitai/kaitai_struct_formats/network/ethernet_frame.py b/python/examples/kaitai/kaitai_struct_formats/network/ethernet_frame.py deleted file mode 100644 index e6baaccf..00000000 --- a/python/examples/kaitai/kaitai_struct_formats/network/ethernet_frame.py +++ /dev/null @@ -1,66 +0,0 @@ -from __future__ import absolute_import -# This is a generated file! Please edit source .ksy file and use kaitai-struct-compiler to rebuild - -from pkg_resources import parse_version -from ...kaitaistruct import __version__ as ks_version, KaitaiStruct, KaitaiStream, BytesIO -from enum import Enum -import collections - - -if parse_version(ks_version) < parse_version('0.7'): - raise Exception("Incompatible Kaitai Struct Python API: 0.7 or later is required, but you have %s" % (ks_version)) - -from ipv4_packet import Ipv4Packet -from ipv6_packet import Ipv6Packet -class EthernetFrame(KaitaiStruct): - """Ethernet frame is a OSI data link layer (layer 2) protocol data unit - for Ethernet networks. In practice, many other networks and/or - in-file dumps adopted the same format for encapsulation purposes. - - .. seealso:: - Source - https://ieeexplore.ieee.org/document/7428776 - """ - - class EtherTypeEnum(Enum): - ipv4 = 2048 - x_75_internet = 2049 - nbs_internet = 2050 - ecma_internet = 2051 - chaosnet = 2052 - x_25_level_3 = 2053 - arp = 2054 - ipv6 = 34525 - SEQ_FIELDS = ["dst_mac", "src_mac", "ether_type", "body"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['dst_mac']['start'] = self._io.pos() - self.dst_mac = self._io.read_bytes(6) - self._debug['dst_mac']['end'] = self._io.pos() - self._debug['src_mac']['start'] = self._io.pos() - self.src_mac = self._io.read_bytes(6) - self._debug['src_mac']['end'] = self._io.pos() - self._debug['ether_type']['start'] = self._io.pos() - self.ether_type = KaitaiStream.resolve_enum(self._root.EtherTypeEnum, self._io.read_u2be()) - self._debug['ether_type']['end'] = self._io.pos() - self._debug['body']['start'] = self._io.pos() - _on = self.ether_type - if _on == self._root.EtherTypeEnum.ipv4: - self._raw_body = self._io.read_bytes_full() - io = KaitaiStream(BytesIO(self._raw_body)) - self.body = Ipv4Packet(io) - self.body._read() - elif _on == self._root.EtherTypeEnum.ipv6: - self._raw_body = self._io.read_bytes_full() - io = KaitaiStream(BytesIO(self._raw_body)) - self.body = Ipv6Packet(io) - self.body._read() - else: - self.body = self._io.read_bytes_full() - self._debug['body']['end'] = self._io.pos() - - diff --git a/python/examples/kaitai/kaitai_struct_formats/network/hccap.py b/python/examples/kaitai/kaitai_struct_formats/network/hccap.py deleted file mode 100644 index db02ef14..00000000 --- a/python/examples/kaitai/kaitai_struct_formats/network/hccap.py +++ /dev/null @@ -1,110 +0,0 @@ -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 Hccap(KaitaiStruct): - """Native format of Hashcat password "recovery" utility. - - A sample of file for testing can be downloaded from https://web.archive.org/web/20150220013635if_/http://hashcat.net:80/misc/example_hashes/hashcat.hccap - - .. seealso:: - Source - https://hashcat.net/wiki/doku.php?id=hccap - """ - 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 = [] - i = 0 - while not self._io.is_eof(): - if not 'arr' in self._debug['records']: - self._debug['records']['arr'] = [] - self._debug['records']['arr'].append({'start': self._io.pos()}) - _t_records = self._root.HccapRecord(self._io, self, self._root) - _t_records._read() - self.records.append(_t_records) - self._debug['records']['arr'][len(self.records) - 1]['end'] = self._io.pos() - i += 1 - - self._debug['records']['end'] = self._io.pos() - - class HccapRecord(KaitaiStruct): - SEQ_FIELDS = ["essid", "mac_ap", "mac_station", "nonce_station", "nonce_ap", "eapol_buffer", "len_eapol", "keyver", "keymic"] - 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['essid']['start'] = self._io.pos() - self.essid = self._io.read_bytes(36) - self._debug['essid']['end'] = self._io.pos() - self._debug['mac_ap']['start'] = self._io.pos() - self.mac_ap = self._io.read_bytes(6) - self._debug['mac_ap']['end'] = self._io.pos() - self._debug['mac_station']['start'] = self._io.pos() - self.mac_station = self._io.read_bytes(6) - self._debug['mac_station']['end'] = self._io.pos() - self._debug['nonce_station']['start'] = self._io.pos() - self.nonce_station = self._io.read_bytes(32) - self._debug['nonce_station']['end'] = self._io.pos() - self._debug['nonce_ap']['start'] = self._io.pos() - self.nonce_ap = self._io.read_bytes(32) - self._debug['nonce_ap']['end'] = self._io.pos() - self._debug['eapol_buffer']['start'] = self._io.pos() - self._raw_eapol_buffer = self._io.read_bytes(256) - io = KaitaiStream(BytesIO(self._raw_eapol_buffer)) - self.eapol_buffer = self._root.EapolDummy(io, self, self._root) - self.eapol_buffer._read() - self._debug['eapol_buffer']['end'] = self._io.pos() - self._debug['len_eapol']['start'] = self._io.pos() - self.len_eapol = self._io.read_u4le() - self._debug['len_eapol']['end'] = self._io.pos() - self._debug['keyver']['start'] = self._io.pos() - self.keyver = self._io.read_u4le() - self._debug['keyver']['end'] = self._io.pos() - self._debug['keymic']['start'] = self._io.pos() - self.keymic = self._io.read_bytes(16) - self._debug['keymic']['end'] = self._io.pos() - - @property - def eapol(self): - if hasattr(self, '_m_eapol'): - return self._m_eapol if hasattr(self, '_m_eapol') else None - - io = self.eapol_buffer._io - _pos = io.pos() - io.seek(0) - self._debug['_m_eapol']['start'] = io.pos() - self._m_eapol = io.read_bytes(self.len_eapol) - self._debug['_m_eapol']['end'] = io.pos() - io.seek(_pos) - return self._m_eapol if hasattr(self, '_m_eapol') else None - - - class EapolDummy(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 - - - diff --git a/python/examples/kaitai/kaitai_struct_formats/network/hccapx.py b/python/examples/kaitai/kaitai_struct_formats/network/hccapx.py deleted file mode 100644 index 439db7d4..00000000 --- a/python/examples/kaitai/kaitai_struct_formats/network/hccapx.py +++ /dev/null @@ -1,101 +0,0 @@ -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 Hccapx(KaitaiStruct): - """Native format of Hashcat password "recovery" utility - - .. seealso:: - Source - https://hashcat.net/wiki/doku.php?id=hccapx - """ - 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 = [] - i = 0 - while not self._io.is_eof(): - if not 'arr' in self._debug['records']: - self._debug['records']['arr'] = [] - self._debug['records']['arr'].append({'start': self._io.pos()}) - _t_records = self._root.HccapxRecord(self._io, self, self._root) - _t_records._read() - self.records.append(_t_records) - self._debug['records']['arr'][len(self.records) - 1]['end'] = self._io.pos() - i += 1 - - self._debug['records']['end'] = self._io.pos() - - class HccapxRecord(KaitaiStruct): - SEQ_FIELDS = ["magic", "version", "ignore_replay_counter", "message_pair", "len_essid", "essid", "padding1", "keyver", "keymic", "mac_ap", "nonce_ap", "mac_station", "nonce_station", "len_eapol", "eapol", "padding2"] - 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"\x48\x43\x50\x58") - self._debug['magic']['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['ignore_replay_counter']['start'] = self._io.pos() - self.ignore_replay_counter = self._io.read_bits_int(1) != 0 - self._debug['ignore_replay_counter']['end'] = self._io.pos() - self._debug['message_pair']['start'] = self._io.pos() - self.message_pair = self._io.read_bits_int(7) - self._debug['message_pair']['end'] = self._io.pos() - self._io.align_to_byte() - self._debug['len_essid']['start'] = self._io.pos() - self.len_essid = self._io.read_u1() - self._debug['len_essid']['end'] = self._io.pos() - self._debug['essid']['start'] = self._io.pos() - self.essid = self._io.read_bytes(self.len_essid) - self._debug['essid']['end'] = self._io.pos() - self._debug['padding1']['start'] = self._io.pos() - self.padding1 = self._io.read_bytes((32 - self.len_essid)) - self._debug['padding1']['end'] = self._io.pos() - self._debug['keyver']['start'] = self._io.pos() - self.keyver = self._io.read_u1() - self._debug['keyver']['end'] = self._io.pos() - self._debug['keymic']['start'] = self._io.pos() - self.keymic = self._io.read_bytes(16) - self._debug['keymic']['end'] = self._io.pos() - self._debug['mac_ap']['start'] = self._io.pos() - self.mac_ap = self._io.read_bytes(6) - self._debug['mac_ap']['end'] = self._io.pos() - self._debug['nonce_ap']['start'] = self._io.pos() - self.nonce_ap = self._io.read_bytes(32) - self._debug['nonce_ap']['end'] = self._io.pos() - self._debug['mac_station']['start'] = self._io.pos() - self.mac_station = self._io.read_bytes(6) - self._debug['mac_station']['end'] = self._io.pos() - self._debug['nonce_station']['start'] = self._io.pos() - self.nonce_station = self._io.read_bytes(32) - self._debug['nonce_station']['end'] = self._io.pos() - self._debug['len_eapol']['start'] = self._io.pos() - self.len_eapol = self._io.read_u2le() - self._debug['len_eapol']['end'] = self._io.pos() - self._debug['eapol']['start'] = self._io.pos() - self.eapol = self._io.read_bytes(self.len_eapol) - self._debug['eapol']['end'] = self._io.pos() - self._debug['padding2']['start'] = self._io.pos() - self.padding2 = self._io.read_bytes((256 - self.len_eapol)) - self._debug['padding2']['end'] = self._io.pos() - - - diff --git a/python/examples/kaitai/kaitai_struct_formats/network/icmp_packet.py b/python/examples/kaitai/kaitai_struct_formats/network/icmp_packet.py deleted file mode 100644 index 45de7aae..00000000 --- a/python/examples/kaitai/kaitai_struct_formats/network/icmp_packet.py +++ /dev/null @@ -1,134 +0,0 @@ -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 IcmpPacket(KaitaiStruct): - - class IcmpTypeEnum(Enum): - echo_reply = 0 - destination_unreachable = 3 - source_quench = 4 - redirect = 5 - echo = 8 - time_exceeded = 11 - SEQ_FIELDS = ["icmp_type", "destination_unreachable", "time_exceeded", "echo"] - 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['icmp_type']['start'] = self._io.pos() - self.icmp_type = KaitaiStream.resolve_enum(self._root.IcmpTypeEnum, self._io.read_u1()) - self._debug['icmp_type']['end'] = self._io.pos() - if self.icmp_type == self._root.IcmpTypeEnum.destination_unreachable: - self._debug['destination_unreachable']['start'] = self._io.pos() - self.destination_unreachable = self._root.DestinationUnreachableMsg(self._io, self, self._root) - self.destination_unreachable._read() - self._debug['destination_unreachable']['end'] = self._io.pos() - - if self.icmp_type == self._root.IcmpTypeEnum.time_exceeded: - self._debug['time_exceeded']['start'] = self._io.pos() - self.time_exceeded = self._root.TimeExceededMsg(self._io, self, self._root) - self.time_exceeded._read() - self._debug['time_exceeded']['end'] = self._io.pos() - - if ((self.icmp_type == self._root.IcmpTypeEnum.echo) or (self.icmp_type == self._root.IcmpTypeEnum.echo_reply)) : - self._debug['echo']['start'] = self._io.pos() - self.echo = self._root.EchoMsg(self._io, self, self._root) - self.echo._read() - self._debug['echo']['end'] = self._io.pos() - - - class DestinationUnreachableMsg(KaitaiStruct): - - class DestinationUnreachableCode(Enum): - net_unreachable = 0 - host_unreachable = 1 - protocol_unreachable = 2 - port_unreachable = 3 - fragmentation_needed_and_df_set = 4 - source_route_failed = 5 - dst_net_unkown = 6 - sdt_host_unkown = 7 - src_isolated = 8 - net_prohibited_by_admin = 9 - host_prohibited_by_admin = 10 - net_unreachable_for_tos = 11 - host_unreachable_for_tos = 12 - communication_prohibited_by_admin = 13 - host_precedence_violation = 14 - precedence_cuttoff_in_effect = 15 - SEQ_FIELDS = ["code", "checksum"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['code']['start'] = self._io.pos() - self.code = KaitaiStream.resolve_enum(self._root.DestinationUnreachableMsg.DestinationUnreachableCode, self._io.read_u1()) - self._debug['code']['end'] = self._io.pos() - self._debug['checksum']['start'] = self._io.pos() - self.checksum = self._io.read_u2be() - self._debug['checksum']['end'] = self._io.pos() - - - class TimeExceededMsg(KaitaiStruct): - - class TimeExceededCode(Enum): - time_to_live_exceeded_in_transit = 0 - fragment_reassembly_time_exceeded = 1 - SEQ_FIELDS = ["code", "checksum"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['code']['start'] = self._io.pos() - self.code = KaitaiStream.resolve_enum(self._root.TimeExceededMsg.TimeExceededCode, self._io.read_u1()) - self._debug['code']['end'] = self._io.pos() - self._debug['checksum']['start'] = self._io.pos() - self.checksum = self._io.read_u2be() - self._debug['checksum']['end'] = self._io.pos() - - - class EchoMsg(KaitaiStruct): - SEQ_FIELDS = ["code", "checksum", "identifier", "seq_num", "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['code']['start'] = self._io.pos() - self.code = self._io.ensure_fixed_contents(b"\x00") - self._debug['code']['end'] = self._io.pos() - self._debug['checksum']['start'] = self._io.pos() - self.checksum = self._io.read_u2be() - self._debug['checksum']['end'] = self._io.pos() - self._debug['identifier']['start'] = self._io.pos() - self.identifier = self._io.read_u2be() - self._debug['identifier']['end'] = self._io.pos() - self._debug['seq_num']['start'] = self._io.pos() - self.seq_num = self._io.read_u2be() - self._debug['seq_num']['end'] = self._io.pos() - self._debug['data']['start'] = self._io.pos() - self.data = self._io.read_bytes_full() - self._debug['data']['end'] = self._io.pos() - - - diff --git a/python/examples/kaitai/kaitai_struct_formats/network/ipv4_packet.py b/python/examples/kaitai/kaitai_struct_formats/network/ipv4_packet.py deleted file mode 100644 index 0bc7bb95..00000000 --- a/python/examples/kaitai/kaitai_struct_formats/network/ipv4_packet.py +++ /dev/null @@ -1,158 +0,0 @@ -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)) - -from protocol_body import ProtocolBody -class Ipv4Packet(KaitaiStruct): - SEQ_FIELDS = ["b1", "b2", "total_length", "identification", "b67", "ttl", "protocol", "header_checksum", "src_ip_addr", "dst_ip_addr", "options", "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['b1']['start'] = self._io.pos() - self.b1 = self._io.read_u1() - self._debug['b1']['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['total_length']['start'] = self._io.pos() - self.total_length = self._io.read_u2be() - self._debug['total_length']['end'] = self._io.pos() - self._debug['identification']['start'] = self._io.pos() - self.identification = self._io.read_u2be() - self._debug['identification']['end'] = self._io.pos() - self._debug['b67']['start'] = self._io.pos() - self.b67 = self._io.read_u2be() - self._debug['b67']['end'] = self._io.pos() - self._debug['ttl']['start'] = self._io.pos() - self.ttl = self._io.read_u1() - self._debug['ttl']['end'] = self._io.pos() - self._debug['protocol']['start'] = self._io.pos() - self.protocol = self._io.read_u1() - self._debug['protocol']['end'] = self._io.pos() - self._debug['header_checksum']['start'] = self._io.pos() - self.header_checksum = self._io.read_u2be() - self._debug['header_checksum']['end'] = self._io.pos() - self._debug['src_ip_addr']['start'] = self._io.pos() - self.src_ip_addr = self._io.read_bytes(4) - self._debug['src_ip_addr']['end'] = self._io.pos() - self._debug['dst_ip_addr']['start'] = self._io.pos() - self.dst_ip_addr = self._io.read_bytes(4) - self._debug['dst_ip_addr']['end'] = self._io.pos() - self._debug['options']['start'] = self._io.pos() - self._raw_options = self._io.read_bytes((self.ihl_bytes - 20)) - io = KaitaiStream(BytesIO(self._raw_options)) - self.options = self._root.Ipv4Options(io, self, self._root) - self.options._read() - self._debug['options']['end'] = self._io.pos() - self._debug['body']['start'] = self._io.pos() - self._raw_body = self._io.read_bytes((self.total_length - self.ihl_bytes)) - io = KaitaiStream(BytesIO(self._raw_body)) - self.body = ProtocolBody(self.protocol, io) - self.body._read() - self._debug['body']['end'] = self._io.pos() - - class Ipv4Options(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.Ipv4Option(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 Ipv4Option(KaitaiStruct): - SEQ_FIELDS = ["b1", "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['b1']['start'] = self._io.pos() - self.b1 = self._io.read_u1() - self._debug['b1']['end'] = self._io.pos() - self._debug['len']['start'] = self._io.pos() - self.len = self._io.read_u1() - self._debug['len']['end'] = self._io.pos() - self._debug['body']['start'] = self._io.pos() - self.body = self._io.read_bytes(((self.len - 2) if self.len > 2 else 0)) - self._debug['body']['end'] = self._io.pos() - - @property - def copy(self): - if hasattr(self, '_m_copy'): - return self._m_copy if hasattr(self, '_m_copy') else None - - self._m_copy = ((self.b1 & 128) >> 7) - return self._m_copy if hasattr(self, '_m_copy') else None - - @property - def opt_class(self): - if hasattr(self, '_m_opt_class'): - return self._m_opt_class if hasattr(self, '_m_opt_class') else None - - self._m_opt_class = ((self.b1 & 96) >> 5) - return self._m_opt_class if hasattr(self, '_m_opt_class') else None - - @property - def number(self): - if hasattr(self, '_m_number'): - return self._m_number if hasattr(self, '_m_number') else None - - self._m_number = (self.b1 & 31) - return self._m_number if hasattr(self, '_m_number') else None - - - @property - def version(self): - if hasattr(self, '_m_version'): - return self._m_version if hasattr(self, '_m_version') else None - - self._m_version = ((self.b1 & 240) >> 4) - return self._m_version if hasattr(self, '_m_version') else None - - @property - def ihl(self): - if hasattr(self, '_m_ihl'): - return self._m_ihl if hasattr(self, '_m_ihl') else None - - self._m_ihl = (self.b1 & 15) - return self._m_ihl if hasattr(self, '_m_ihl') else None - - @property - def ihl_bytes(self): - if hasattr(self, '_m_ihl_bytes'): - return self._m_ihl_bytes if hasattr(self, '_m_ihl_bytes') else None - - self._m_ihl_bytes = (self.ihl * 4) - return self._m_ihl_bytes if hasattr(self, '_m_ihl_bytes') else None - - diff --git a/python/examples/kaitai/kaitai_struct_formats/network/ipv6_packet.py b/python/examples/kaitai/kaitai_struct_formats/network/ipv6_packet.py deleted file mode 100644 index c108d643..00000000 --- a/python/examples/kaitai/kaitai_struct_formats/network/ipv6_packet.py +++ /dev/null @@ -1,55 +0,0 @@ -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)) - -from protocol_body import ProtocolBody -class Ipv6Packet(KaitaiStruct): - SEQ_FIELDS = ["version", "traffic_class", "flow_label", "payload_length", "next_header_type", "hop_limit", "src_ipv6_addr", "dst_ipv6_addr", "next_header", "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['version']['start'] = self._io.pos() - self.version = self._io.read_bits_int(4) - self._debug['version']['end'] = self._io.pos() - self._debug['traffic_class']['start'] = self._io.pos() - self.traffic_class = self._io.read_bits_int(8) - self._debug['traffic_class']['end'] = self._io.pos() - self._debug['flow_label']['start'] = self._io.pos() - self.flow_label = self._io.read_bits_int(20) - self._debug['flow_label']['end'] = self._io.pos() - self._io.align_to_byte() - self._debug['payload_length']['start'] = self._io.pos() - self.payload_length = self._io.read_u2be() - self._debug['payload_length']['end'] = self._io.pos() - self._debug['next_header_type']['start'] = self._io.pos() - self.next_header_type = self._io.read_u1() - self._debug['next_header_type']['end'] = self._io.pos() - self._debug['hop_limit']['start'] = self._io.pos() - self.hop_limit = self._io.read_u1() - self._debug['hop_limit']['end'] = self._io.pos() - self._debug['src_ipv6_addr']['start'] = self._io.pos() - self.src_ipv6_addr = self._io.read_bytes(16) - self._debug['src_ipv6_addr']['end'] = self._io.pos() - self._debug['dst_ipv6_addr']['start'] = self._io.pos() - self.dst_ipv6_addr = self._io.read_bytes(16) - self._debug['dst_ipv6_addr']['end'] = self._io.pos() - self._debug['next_header']['start'] = self._io.pos() - self.next_header = ProtocolBody(self.next_header_type, self._io) - self.next_header._read() - self._debug['next_header']['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() - - diff --git a/python/examples/kaitai/kaitai_struct_formats/network/microsoft_network_monitor_v2.py b/python/examples/kaitai/kaitai_struct_formats/network/microsoft_network_monitor_v2.py deleted file mode 100644 index 7fec8613..00000000 --- a/python/examples/kaitai/kaitai_struct_formats/network/microsoft_network_monitor_v2.py +++ /dev/null @@ -1,307 +0,0 @@ -from __future__ import absolute_import -# This is a generated file! Please edit source .ksy file and use kaitai-struct-compiler to rebuild - -from pkg_resources import parse_version -from ...kaitaistruct import __version__ as ks_version, KaitaiStruct, KaitaiStream, BytesIO -from enum import Enum -import collections - - -if parse_version(ks_version) < parse_version('0.7'): - raise Exception("Incompatible Kaitai Struct Python API: 0.7 or later is required, but you have %s" % (ks_version)) - -from windows_systemtime import WindowsSystemtime -from ethernet_frame import EthernetFrame -class MicrosoftNetworkMonitorV2(KaitaiStruct): - """Microsoft Network Monitor (AKA Netmon) is a proprietary Microsoft's - network packet sniffing and analysis tool. It can save captured - traffic as .cap files, which usually contain the packets and may - contain some additional info - enhanced network info, calculated - statistics, etc. - - There are at least 2 different versions of the format: v1 and - v2. Netmon v3 seems to use the same file format as v1. - - .. seealso:: - Source - https://msdn.microsoft.com/en-us/library/windows/desktop/ee817717.aspx - """ - - class Linktype(Enum): - null_linktype = 0 - ethernet = 1 - ax25 = 3 - ieee802_5 = 6 - arcnet_bsd = 7 - slip = 8 - ppp = 9 - fddi = 10 - ppp_hdlc = 50 - ppp_ether = 51 - atm_rfc1483 = 100 - raw = 101 - c_hdlc = 104 - ieee802_11 = 105 - frelay = 107 - loop = 108 - linux_sll = 113 - ltalk = 114 - pflog = 117 - ieee802_11_prism = 119 - ip_over_fc = 122 - sunatm = 123 - ieee802_11_radiotap = 127 - arcnet_linux = 129 - apple_ip_over_ieee1394 = 138 - mtp2_with_phdr = 139 - mtp2 = 140 - mtp3 = 141 - sccp = 142 - docsis = 143 - linux_irda = 144 - user0 = 147 - user1 = 148 - user2 = 149 - user3 = 150 - user4 = 151 - user5 = 152 - user6 = 153 - user7 = 154 - user8 = 155 - user9 = 156 - user10 = 157 - user11 = 158 - user12 = 159 - user13 = 160 - user14 = 161 - user15 = 162 - ieee802_11_avs = 163 - bacnet_ms_tp = 165 - ppp_pppd = 166 - gprs_llc = 169 - gpf_t = 170 - gpf_f = 171 - linux_lapd = 177 - bluetooth_hci_h4 = 187 - usb_linux = 189 - ppi = 192 - ieee802_15_4 = 195 - sita = 196 - erf = 197 - bluetooth_hci_h4_with_phdr = 201 - ax25_kiss = 202 - lapd = 203 - ppp_with_dir = 204 - c_hdlc_with_dir = 205 - frelay_with_dir = 206 - ipmb_linux = 209 - ieee802_15_4_nonask_phy = 215 - usb_linux_mmapped = 220 - fc_2 = 224 - fc_2_with_frame_delims = 225 - ipnet = 226 - can_socketcan = 227 - ipv4 = 228 - ipv6 = 229 - ieee802_15_4_nofcs = 230 - dbus = 231 - dvb_ci = 235 - mux27010 = 236 - stanag_5066_d_pdu = 237 - nflog = 239 - netanalyzer = 240 - netanalyzer_transparent = 241 - ipoib = 242 - mpeg_2_ts = 243 - ng40 = 244 - nfc_llcp = 245 - infiniband = 247 - sctp = 248 - usbpcap = 249 - rtac_serial = 250 - bluetooth_le_ll = 251 - netlink = 253 - bluetooth_linux_monitor = 254 - bluetooth_bredr_bb = 255 - bluetooth_le_ll_with_phdr = 256 - profibus_dl = 257 - pktap = 258 - epon = 259 - ipmi_hpm_2 = 260 - zwave_r1_r2 = 261 - zwave_r3 = 262 - wattstopper_dlm = 263 - iso_14443 = 264 - SEQ_FIELDS = ["signature", "version_minor", "version_major", "mac_type", "time_capture_start", "frame_table_ofs", "frame_table_len", "user_data_ofs", "user_data_len", "comment_ofs", "comment_len", "statistics_ofs", "statistics_len", "network_info_ofs", "network_info_len", "conversation_stats_ofs", "conversation_stats_len"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['signature']['start'] = self._io.pos() - self.signature = self._io.ensure_fixed_contents(b"\x47\x4D\x42\x55") - self._debug['signature']['end'] = self._io.pos() - self._debug['version_minor']['start'] = self._io.pos() - self.version_minor = self._io.read_u1() - self._debug['version_minor']['end'] = self._io.pos() - self._debug['version_major']['start'] = self._io.pos() - self.version_major = self._io.read_u1() - self._debug['version_major']['end'] = self._io.pos() - self._debug['mac_type']['start'] = self._io.pos() - self.mac_type = KaitaiStream.resolve_enum(self._root.Linktype, self._io.read_u2le()) - self._debug['mac_type']['end'] = self._io.pos() - self._debug['time_capture_start']['start'] = self._io.pos() - self.time_capture_start = WindowsSystemtime(self._io) - self.time_capture_start._read() - self._debug['time_capture_start']['end'] = self._io.pos() - self._debug['frame_table_ofs']['start'] = self._io.pos() - self.frame_table_ofs = self._io.read_u4le() - self._debug['frame_table_ofs']['end'] = self._io.pos() - self._debug['frame_table_len']['start'] = self._io.pos() - self.frame_table_len = self._io.read_u4le() - self._debug['frame_table_len']['end'] = self._io.pos() - self._debug['user_data_ofs']['start'] = self._io.pos() - self.user_data_ofs = self._io.read_u4le() - self._debug['user_data_ofs']['end'] = self._io.pos() - self._debug['user_data_len']['start'] = self._io.pos() - self.user_data_len = self._io.read_u4le() - self._debug['user_data_len']['end'] = self._io.pos() - self._debug['comment_ofs']['start'] = self._io.pos() - self.comment_ofs = self._io.read_u4le() - self._debug['comment_ofs']['end'] = self._io.pos() - self._debug['comment_len']['start'] = self._io.pos() - self.comment_len = self._io.read_u4le() - self._debug['comment_len']['end'] = self._io.pos() - self._debug['statistics_ofs']['start'] = self._io.pos() - self.statistics_ofs = self._io.read_u4le() - self._debug['statistics_ofs']['end'] = self._io.pos() - self._debug['statistics_len']['start'] = self._io.pos() - self.statistics_len = self._io.read_u4le() - self._debug['statistics_len']['end'] = self._io.pos() - self._debug['network_info_ofs']['start'] = self._io.pos() - self.network_info_ofs = self._io.read_u4le() - self._debug['network_info_ofs']['end'] = self._io.pos() - self._debug['network_info_len']['start'] = self._io.pos() - self.network_info_len = self._io.read_u4le() - self._debug['network_info_len']['end'] = self._io.pos() - self._debug['conversation_stats_ofs']['start'] = self._io.pos() - self.conversation_stats_ofs = self._io.read_u4le() - self._debug['conversation_stats_ofs']['end'] = self._io.pos() - self._debug['conversation_stats_len']['start'] = self._io.pos() - self.conversation_stats_len = self._io.read_u4le() - self._debug['conversation_stats_len']['end'] = self._io.pos() - - class FrameIndex(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.FrameIndexEntry(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 FrameIndexEntry(KaitaiStruct): - """Each index entry is just a pointer to where the frame data is - stored in the file. - """ - SEQ_FIELDS = ["ofs"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['ofs']['start'] = self._io.pos() - self.ofs = self._io.read_u4le() - self._debug['ofs']['end'] = self._io.pos() - - @property - def body(self): - """Frame body itself.""" - if hasattr(self, '_m_body'): - return self._m_body if hasattr(self, '_m_body') else None - - io = self._root._io - _pos = io.pos() - io.seek(self.ofs) - self._debug['_m_body']['start'] = io.pos() - self._m_body = self._root.Frame(io, self, self._root) - self._m_body._read() - self._debug['_m_body']['end'] = io.pos() - io.seek(_pos) - return self._m_body if hasattr(self, '_m_body') else None - - - class Frame(KaitaiStruct): - """A container for actually captured network data. Allow to - timestamp individual frames and designates how much data from - the original packet was actually written into the file. - - .. seealso:: - Source - https://msdn.microsoft.com/en-us/library/windows/desktop/ee831821.aspx - """ - SEQ_FIELDS = ["ts_delta", "orig_len", "inc_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['ts_delta']['start'] = self._io.pos() - self.ts_delta = self._io.read_u8le() - self._debug['ts_delta']['end'] = self._io.pos() - self._debug['orig_len']['start'] = self._io.pos() - self.orig_len = self._io.read_u4le() - self._debug['orig_len']['end'] = self._io.pos() - self._debug['inc_len']['start'] = self._io.pos() - self.inc_len = self._io.read_u4le() - self._debug['inc_len']['end'] = self._io.pos() - self._debug['body']['start'] = self._io.pos() - _on = self._root.mac_type - if _on == self._root.Linktype.ethernet: - self._raw_body = self._io.read_bytes(self.inc_len) - io = KaitaiStream(BytesIO(self._raw_body)) - self.body = EthernetFrame(io) - self.body._read() - else: - self.body = self._io.read_bytes(self.inc_len) - self._debug['body']['end'] = self._io.pos() - - - @property - def frame_table(self): - """Index that is used to access individual captured frames.""" - if hasattr(self, '_m_frame_table'): - return self._m_frame_table if hasattr(self, '_m_frame_table') else None - - _pos = self._io.pos() - self._io.seek(self.frame_table_ofs) - self._debug['_m_frame_table']['start'] = self._io.pos() - self._raw__m_frame_table = self._io.read_bytes(self.frame_table_len) - io = KaitaiStream(BytesIO(self._raw__m_frame_table)) - self._m_frame_table = self._root.FrameIndex(io, self, self._root) - self._m_frame_table._read() - self._debug['_m_frame_table']['end'] = self._io.pos() - self._io.seek(_pos) - return self._m_frame_table if hasattr(self, '_m_frame_table') else None - - diff --git a/python/examples/kaitai/kaitai_struct_formats/network/packet_ppi.py b/python/examples/kaitai/kaitai_struct_formats/network/packet_ppi.py deleted file mode 100644 index b213460a..00000000 --- a/python/examples/kaitai/kaitai_struct_formats/network/packet_ppi.py +++ /dev/null @@ -1,514 +0,0 @@ -from __future__ import absolute_import -# This is a generated file! Please edit source .ksy file and use kaitai-struct-compiler to rebuild - -from pkg_resources import parse_version -from ...kaitaistruct import __version__ as ks_version, KaitaiStruct, KaitaiStream, BytesIO -from enum import Enum -import collections - - -if parse_version(ks_version) < parse_version('0.7'): - raise Exception("Incompatible Kaitai Struct Python API: 0.7 or later is required, but you have %s" % (ks_version)) - -from ethernet_frame import EthernetFrame -class PacketPpi(KaitaiStruct): - """PPI is a standard for link layer packet encapsulation, proposed as - generic extensible container to store both captured in-band data and - out-of-band data. Originally it was developed to provide 802.11n - radio information, but can be used for other purposes as well. - - Sample capture: https://wiki.wireshark.org/SampleCaptures?action=AttachFile&do=get&target=Http.cap - - .. seealso:: - PPI header format spec, section 3 - https://www.cacetech.com/documents/PPI_Header_format_1.0.1.pdf - """ - - class PfhType(Enum): - radio_802_11_common = 2 - radio_802_11n_mac_ext = 3 - radio_802_11n_mac_phy_ext = 4 - spectrum_map = 5 - process_info = 6 - capture_info = 7 - - class Linktype(Enum): - null_linktype = 0 - ethernet = 1 - ax25 = 3 - ieee802_5 = 6 - arcnet_bsd = 7 - slip = 8 - ppp = 9 - fddi = 10 - ppp_hdlc = 50 - ppp_ether = 51 - atm_rfc1483 = 100 - raw = 101 - c_hdlc = 104 - ieee802_11 = 105 - frelay = 107 - loop = 108 - linux_sll = 113 - ltalk = 114 - pflog = 117 - ieee802_11_prism = 119 - ip_over_fc = 122 - sunatm = 123 - ieee802_11_radiotap = 127 - arcnet_linux = 129 - apple_ip_over_ieee1394 = 138 - mtp2_with_phdr = 139 - mtp2 = 140 - mtp3 = 141 - sccp = 142 - docsis = 143 - linux_irda = 144 - user0 = 147 - user1 = 148 - user2 = 149 - user3 = 150 - user4 = 151 - user5 = 152 - user6 = 153 - user7 = 154 - user8 = 155 - user9 = 156 - user10 = 157 - user11 = 158 - user12 = 159 - user13 = 160 - user14 = 161 - user15 = 162 - ieee802_11_avs = 163 - bacnet_ms_tp = 165 - ppp_pppd = 166 - gprs_llc = 169 - gpf_t = 170 - gpf_f = 171 - linux_lapd = 177 - bluetooth_hci_h4 = 187 - usb_linux = 189 - ppi = 192 - ieee802_15_4 = 195 - sita = 196 - erf = 197 - bluetooth_hci_h4_with_phdr = 201 - ax25_kiss = 202 - lapd = 203 - ppp_with_dir = 204 - c_hdlc_with_dir = 205 - frelay_with_dir = 206 - ipmb_linux = 209 - ieee802_15_4_nonask_phy = 215 - usb_linux_mmapped = 220 - fc_2 = 224 - fc_2_with_frame_delims = 225 - ipnet = 226 - can_socketcan = 227 - ipv4 = 228 - ipv6 = 229 - ieee802_15_4_nofcs = 230 - dbus = 231 - dvb_ci = 235 - mux27010 = 236 - stanag_5066_d_pdu = 237 - nflog = 239 - netanalyzer = 240 - netanalyzer_transparent = 241 - ipoib = 242 - mpeg_2_ts = 243 - ng40 = 244 - nfc_llcp = 245 - infiniband = 247 - sctp = 248 - usbpcap = 249 - rtac_serial = 250 - bluetooth_le_ll = 251 - netlink = 253 - bluetooth_linux_monitor = 254 - bluetooth_bredr_bb = 255 - bluetooth_le_ll_with_phdr = 256 - profibus_dl = 257 - pktap = 258 - epon = 259 - ipmi_hpm_2 = 260 - zwave_r1_r2 = 261 - zwave_r3 = 262 - wattstopper_dlm = 263 - iso_14443 = 264 - SEQ_FIELDS = ["header", "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['header']['start'] = self._io.pos() - self.header = self._root.PacketPpiHeader(self._io, self, self._root) - self.header._read() - self._debug['header']['end'] = self._io.pos() - self._debug['fields']['start'] = self._io.pos() - self._raw_fields = self._io.read_bytes((self.header.pph_len - 8)) - io = KaitaiStream(BytesIO(self._raw_fields)) - self.fields = self._root.PacketPpiFields(io, self, self._root) - self.fields._read() - self._debug['fields']['end'] = self._io.pos() - self._debug['body']['start'] = self._io.pos() - _on = self.header.pph_dlt - if _on == self._root.Linktype.ppi: - self._raw_body = self._io.read_bytes_full() - io = KaitaiStream(BytesIO(self._raw_body)) - self.body = PacketPpi(io) - self.body._read() - elif _on == self._root.Linktype.ethernet: - self._raw_body = self._io.read_bytes_full() - io = KaitaiStream(BytesIO(self._raw_body)) - self.body = EthernetFrame(io) - self.body._read() - else: - self.body = self._io.read_bytes_full() - self._debug['body']['end'] = self._io.pos() - - class PacketPpiFields(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.PacketPpiField(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 Radio80211nMacExtBody(KaitaiStruct): - """ - .. seealso:: - PPI header format spec, section 4.1.3 - https://www.cacetech.com/documents/PPI_Header_format_1.0.1.pdf - """ - SEQ_FIELDS = ["flags", "a_mpdu_id", "num_delimiters", "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['flags']['start'] = self._io.pos() - self.flags = self._root.MacFlags(self._io, self, self._root) - self.flags._read() - self._debug['flags']['end'] = self._io.pos() - self._debug['a_mpdu_id']['start'] = self._io.pos() - self.a_mpdu_id = self._io.read_u4le() - self._debug['a_mpdu_id']['end'] = self._io.pos() - self._debug['num_delimiters']['start'] = self._io.pos() - self.num_delimiters = self._io.read_u1() - self._debug['num_delimiters']['end'] = self._io.pos() - self._debug['reserved']['start'] = self._io.pos() - self.reserved = self._io.read_bytes(3) - self._debug['reserved']['end'] = self._io.pos() - - - class MacFlags(KaitaiStruct): - SEQ_FIELDS = ["unused1", "aggregate_delimiter", "more_aggregates", "aggregate", "dup_rx", "rx_short_guard", "is_ht_40", "greenfield", "unused2"] - 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.read_bits_int(1) != 0 - self._debug['unused1']['end'] = self._io.pos() - self._debug['aggregate_delimiter']['start'] = self._io.pos() - self.aggregate_delimiter = self._io.read_bits_int(1) != 0 - self._debug['aggregate_delimiter']['end'] = self._io.pos() - self._debug['more_aggregates']['start'] = self._io.pos() - self.more_aggregates = self._io.read_bits_int(1) != 0 - self._debug['more_aggregates']['end'] = self._io.pos() - self._debug['aggregate']['start'] = self._io.pos() - self.aggregate = self._io.read_bits_int(1) != 0 - self._debug['aggregate']['end'] = self._io.pos() - self._debug['dup_rx']['start'] = self._io.pos() - self.dup_rx = self._io.read_bits_int(1) != 0 - self._debug['dup_rx']['end'] = self._io.pos() - self._debug['rx_short_guard']['start'] = self._io.pos() - self.rx_short_guard = self._io.read_bits_int(1) != 0 - self._debug['rx_short_guard']['end'] = self._io.pos() - self._debug['is_ht_40']['start'] = self._io.pos() - self.is_ht_40 = self._io.read_bits_int(1) != 0 - self._debug['is_ht_40']['end'] = self._io.pos() - self._debug['greenfield']['start'] = self._io.pos() - self.greenfield = self._io.read_bits_int(1) != 0 - self._debug['greenfield']['end'] = self._io.pos() - self._io.align_to_byte() - self._debug['unused2']['start'] = self._io.pos() - self.unused2 = self._io.read_bytes(3) - self._debug['unused2']['end'] = self._io.pos() - - - class PacketPpiHeader(KaitaiStruct): - """ - .. seealso:: - PPI header format spec, section 3.1 - https://www.cacetech.com/documents/PPI_Header_format_1.0.1.pdf - """ - SEQ_FIELDS = ["pph_version", "pph_flags", "pph_len", "pph_dlt"] - 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['pph_version']['start'] = self._io.pos() - self.pph_version = self._io.read_u1() - self._debug['pph_version']['end'] = self._io.pos() - self._debug['pph_flags']['start'] = self._io.pos() - self.pph_flags = self._io.read_u1() - self._debug['pph_flags']['end'] = self._io.pos() - self._debug['pph_len']['start'] = self._io.pos() - self.pph_len = self._io.read_u2le() - self._debug['pph_len']['end'] = self._io.pos() - self._debug['pph_dlt']['start'] = self._io.pos() - self.pph_dlt = KaitaiStream.resolve_enum(self._root.Linktype, self._io.read_u4le()) - self._debug['pph_dlt']['end'] = self._io.pos() - - - class Radio80211CommonBody(KaitaiStruct): - """ - .. seealso:: - PPI header format spec, section 4.1.2 - https://www.cacetech.com/documents/PPI_Header_format_1.0.1.pdf - """ - SEQ_FIELDS = ["tsf_timer", "flags", "rate", "channel_freq", "channel_flags", "fhss_hopset", "fhss_pattern", "dbm_antsignal", "dbm_antnoise"] - 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['tsf_timer']['start'] = self._io.pos() - self.tsf_timer = self._io.read_u8le() - self._debug['tsf_timer']['end'] = self._io.pos() - self._debug['flags']['start'] = self._io.pos() - self.flags = self._io.read_u2le() - self._debug['flags']['end'] = self._io.pos() - self._debug['rate']['start'] = self._io.pos() - self.rate = self._io.read_u2le() - self._debug['rate']['end'] = self._io.pos() - self._debug['channel_freq']['start'] = self._io.pos() - self.channel_freq = self._io.read_u2le() - self._debug['channel_freq']['end'] = self._io.pos() - self._debug['channel_flags']['start'] = self._io.pos() - self.channel_flags = self._io.read_u2le() - self._debug['channel_flags']['end'] = self._io.pos() - self._debug['fhss_hopset']['start'] = self._io.pos() - self.fhss_hopset = self._io.read_u1() - self._debug['fhss_hopset']['end'] = self._io.pos() - self._debug['fhss_pattern']['start'] = self._io.pos() - self.fhss_pattern = self._io.read_u1() - self._debug['fhss_pattern']['end'] = self._io.pos() - self._debug['dbm_antsignal']['start'] = self._io.pos() - self.dbm_antsignal = self._io.read_s1() - self._debug['dbm_antsignal']['end'] = self._io.pos() - self._debug['dbm_antnoise']['start'] = self._io.pos() - self.dbm_antnoise = self._io.read_s1() - self._debug['dbm_antnoise']['end'] = self._io.pos() - - - class PacketPpiField(KaitaiStruct): - """ - .. seealso:: - PPI header format spec, section 3.1 - https://www.cacetech.com/documents/PPI_Header_format_1.0.1.pdf - """ - SEQ_FIELDS = ["pfh_type", "pfh_datalen", "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['pfh_type']['start'] = self._io.pos() - self.pfh_type = KaitaiStream.resolve_enum(self._root.PfhType, self._io.read_u2le()) - self._debug['pfh_type']['end'] = self._io.pos() - self._debug['pfh_datalen']['start'] = self._io.pos() - self.pfh_datalen = self._io.read_u2le() - self._debug['pfh_datalen']['end'] = self._io.pos() - self._debug['body']['start'] = self._io.pos() - _on = self.pfh_type - if _on == self._root.PfhType.radio_802_11_common: - self._raw_body = self._io.read_bytes(self.pfh_datalen) - io = KaitaiStream(BytesIO(self._raw_body)) - self.body = self._root.Radio80211CommonBody(io, self, self._root) - self.body._read() - elif _on == self._root.PfhType.radio_802_11n_mac_ext: - self._raw_body = self._io.read_bytes(self.pfh_datalen) - io = KaitaiStream(BytesIO(self._raw_body)) - self.body = self._root.Radio80211nMacExtBody(io, self, self._root) - self.body._read() - elif _on == self._root.PfhType.radio_802_11n_mac_phy_ext: - self._raw_body = self._io.read_bytes(self.pfh_datalen) - io = KaitaiStream(BytesIO(self._raw_body)) - self.body = self._root.Radio80211nMacPhyExtBody(io, self, self._root) - self.body._read() - else: - self.body = self._io.read_bytes(self.pfh_datalen) - self._debug['body']['end'] = self._io.pos() - - - class Radio80211nMacPhyExtBody(KaitaiStruct): - """ - .. seealso:: - PPI header format spec, section 4.1.4 - https://www.cacetech.com/documents/PPI_Header_format_1.0.1.pdf - """ - SEQ_FIELDS = ["flags", "a_mpdu_id", "num_delimiters", "mcs", "num_streams", "rssi_combined", "rssi_ant_ctl", "rssi_ant_ext", "ext_channel_freq", "ext_channel_flags", "rf_signal_noise", "evm"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['flags']['start'] = self._io.pos() - self.flags = self._root.MacFlags(self._io, self, self._root) - self.flags._read() - self._debug['flags']['end'] = self._io.pos() - self._debug['a_mpdu_id']['start'] = self._io.pos() - self.a_mpdu_id = self._io.read_u4le() - self._debug['a_mpdu_id']['end'] = self._io.pos() - self._debug['num_delimiters']['start'] = self._io.pos() - self.num_delimiters = self._io.read_u1() - self._debug['num_delimiters']['end'] = self._io.pos() - self._debug['mcs']['start'] = self._io.pos() - self.mcs = self._io.read_u1() - self._debug['mcs']['end'] = self._io.pos() - self._debug['num_streams']['start'] = self._io.pos() - self.num_streams = self._io.read_u1() - self._debug['num_streams']['end'] = self._io.pos() - self._debug['rssi_combined']['start'] = self._io.pos() - self.rssi_combined = self._io.read_u1() - self._debug['rssi_combined']['end'] = self._io.pos() - self._debug['rssi_ant_ctl']['start'] = self._io.pos() - self.rssi_ant_ctl = [None] * (4) - for i in range(4): - if not 'arr' in self._debug['rssi_ant_ctl']: - self._debug['rssi_ant_ctl']['arr'] = [] - self._debug['rssi_ant_ctl']['arr'].append({'start': self._io.pos()}) - self.rssi_ant_ctl[i] = self._io.read_u1() - self._debug['rssi_ant_ctl']['arr'][i]['end'] = self._io.pos() - - self._debug['rssi_ant_ctl']['end'] = self._io.pos() - self._debug['rssi_ant_ext']['start'] = self._io.pos() - self.rssi_ant_ext = [None] * (4) - for i in range(4): - if not 'arr' in self._debug['rssi_ant_ext']: - self._debug['rssi_ant_ext']['arr'] = [] - self._debug['rssi_ant_ext']['arr'].append({'start': self._io.pos()}) - self.rssi_ant_ext[i] = self._io.read_u1() - self._debug['rssi_ant_ext']['arr'][i]['end'] = self._io.pos() - - self._debug['rssi_ant_ext']['end'] = self._io.pos() - self._debug['ext_channel_freq']['start'] = self._io.pos() - self.ext_channel_freq = self._io.read_u2le() - self._debug['ext_channel_freq']['end'] = self._io.pos() - self._debug['ext_channel_flags']['start'] = self._io.pos() - self.ext_channel_flags = self._root.Radio80211nMacPhyExtBody.ChannelFlags(self._io, self, self._root) - self.ext_channel_flags._read() - self._debug['ext_channel_flags']['end'] = self._io.pos() - self._debug['rf_signal_noise']['start'] = self._io.pos() - self.rf_signal_noise = [None] * (4) - for i in range(4): - if not 'arr' in self._debug['rf_signal_noise']: - self._debug['rf_signal_noise']['arr'] = [] - self._debug['rf_signal_noise']['arr'].append({'start': self._io.pos()}) - _t_rf_signal_noise = self._root.Radio80211nMacPhyExtBody.SignalNoise(self._io, self, self._root) - _t_rf_signal_noise._read() - self.rf_signal_noise[i] = _t_rf_signal_noise - self._debug['rf_signal_noise']['arr'][i]['end'] = self._io.pos() - - self._debug['rf_signal_noise']['end'] = self._io.pos() - self._debug['evm']['start'] = self._io.pos() - self.evm = [None] * (4) - for i in range(4): - if not 'arr' in self._debug['evm']: - self._debug['evm']['arr'] = [] - self._debug['evm']['arr'].append({'start': self._io.pos()}) - self.evm[i] = self._io.read_u4le() - self._debug['evm']['arr'][i]['end'] = self._io.pos() - - self._debug['evm']['end'] = self._io.pos() - - class ChannelFlags(KaitaiStruct): - SEQ_FIELDS = ["spectrum_2ghz", "ofdm", "cck", "turbo", "unused", "gfsk", "dyn_cck_ofdm", "only_passive_scan", "spectrum_5ghz"] - 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['spectrum_2ghz']['start'] = self._io.pos() - self.spectrum_2ghz = self._io.read_bits_int(1) != 0 - self._debug['spectrum_2ghz']['end'] = self._io.pos() - self._debug['ofdm']['start'] = self._io.pos() - self.ofdm = self._io.read_bits_int(1) != 0 - self._debug['ofdm']['end'] = self._io.pos() - self._debug['cck']['start'] = self._io.pos() - self.cck = self._io.read_bits_int(1) != 0 - self._debug['cck']['end'] = self._io.pos() - self._debug['turbo']['start'] = self._io.pos() - self.turbo = self._io.read_bits_int(1) != 0 - self._debug['turbo']['end'] = self._io.pos() - self._debug['unused']['start'] = self._io.pos() - self.unused = self._io.read_bits_int(8) - self._debug['unused']['end'] = self._io.pos() - self._debug['gfsk']['start'] = self._io.pos() - self.gfsk = self._io.read_bits_int(1) != 0 - self._debug['gfsk']['end'] = self._io.pos() - self._debug['dyn_cck_ofdm']['start'] = self._io.pos() - self.dyn_cck_ofdm = self._io.read_bits_int(1) != 0 - self._debug['dyn_cck_ofdm']['end'] = self._io.pos() - self._debug['only_passive_scan']['start'] = self._io.pos() - self.only_passive_scan = self._io.read_bits_int(1) != 0 - self._debug['only_passive_scan']['end'] = self._io.pos() - self._debug['spectrum_5ghz']['start'] = self._io.pos() - self.spectrum_5ghz = self._io.read_bits_int(1) != 0 - self._debug['spectrum_5ghz']['end'] = self._io.pos() - - - class SignalNoise(KaitaiStruct): - """RF signal + noise pair at a single antenna.""" - SEQ_FIELDS = ["signal", "noise"] - 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['signal']['start'] = self._io.pos() - self.signal = self._io.read_s1() - self._debug['signal']['end'] = self._io.pos() - self._debug['noise']['start'] = self._io.pos() - self.noise = self._io.read_s1() - self._debug['noise']['end'] = self._io.pos() - - - - diff --git a/python/examples/kaitai/kaitai_struct_formats/network/pcap.py b/python/examples/kaitai/kaitai_struct_formats/network/pcap.py deleted file mode 100644 index 609a278d..00000000 --- a/python/examples/kaitai/kaitai_struct_formats/network/pcap.py +++ /dev/null @@ -1,235 +0,0 @@ -from __future__ import absolute_import -# This is a generated file! Please edit source .ksy file and use kaitai-struct-compiler to rebuild - -from pkg_resources import parse_version -from ...kaitaistruct import __version__ as ks_version, KaitaiStruct, KaitaiStream, BytesIO -from enum import Enum -import collections - - -if parse_version(ks_version) < parse_version('0.7'): - raise Exception("Incompatible Kaitai Struct Python API: 0.7 or later is required, but you have %s" % (ks_version)) - -from ethernet_frame import EthernetFrame -from packet_ppi import PacketPpi -class Pcap(KaitaiStruct): - """PCAP (named after libpcap / winpcap) is a popular format for saving - network traffic grabbed by network sniffers. It is typically - produced by tools like [tcpdump](https://www.tcpdump.org/) or - [Wireshark](https://www.wireshark.org/). - - .. seealso:: - Source - http://wiki.wireshark.org/Development/LibpcapFileFormat - """ - - class Linktype(Enum): - null_linktype = 0 - ethernet = 1 - ax25 = 3 - ieee802_5 = 6 - arcnet_bsd = 7 - slip = 8 - ppp = 9 - fddi = 10 - ppp_hdlc = 50 - ppp_ether = 51 - atm_rfc1483 = 100 - raw = 101 - c_hdlc = 104 - ieee802_11 = 105 - frelay = 107 - loop = 108 - linux_sll = 113 - ltalk = 114 - pflog = 117 - ieee802_11_prism = 119 - ip_over_fc = 122 - sunatm = 123 - ieee802_11_radiotap = 127 - arcnet_linux = 129 - apple_ip_over_ieee1394 = 138 - mtp2_with_phdr = 139 - mtp2 = 140 - mtp3 = 141 - sccp = 142 - docsis = 143 - linux_irda = 144 - user0 = 147 - user1 = 148 - user2 = 149 - user3 = 150 - user4 = 151 - user5 = 152 - user6 = 153 - user7 = 154 - user8 = 155 - user9 = 156 - user10 = 157 - user11 = 158 - user12 = 159 - user13 = 160 - user14 = 161 - user15 = 162 - ieee802_11_avs = 163 - bacnet_ms_tp = 165 - ppp_pppd = 166 - gprs_llc = 169 - gpf_t = 170 - gpf_f = 171 - linux_lapd = 177 - bluetooth_hci_h4 = 187 - usb_linux = 189 - ppi = 192 - ieee802_15_4 = 195 - sita = 196 - erf = 197 - bluetooth_hci_h4_with_phdr = 201 - ax25_kiss = 202 - lapd = 203 - ppp_with_dir = 204 - c_hdlc_with_dir = 205 - frelay_with_dir = 206 - ipmb_linux = 209 - ieee802_15_4_nonask_phy = 215 - usb_linux_mmapped = 220 - fc_2 = 224 - fc_2_with_frame_delims = 225 - ipnet = 226 - can_socketcan = 227 - ipv4 = 228 - ipv6 = 229 - ieee802_15_4_nofcs = 230 - dbus = 231 - dvb_ci = 235 - mux27010 = 236 - stanag_5066_d_pdu = 237 - nflog = 239 - netanalyzer = 240 - netanalyzer_transparent = 241 - ipoib = 242 - mpeg_2_ts = 243 - ng40 = 244 - nfc_llcp = 245 - infiniband = 247 - sctp = 248 - usbpcap = 249 - rtac_serial = 250 - bluetooth_le_ll = 251 - netlink = 253 - bluetooth_linux_monitor = 254 - bluetooth_bredr_bb = 255 - bluetooth_le_ll_with_phdr = 256 - profibus_dl = 257 - pktap = 258 - epon = 259 - ipmi_hpm_2 = 260 - zwave_r1_r2 = 261 - zwave_r3 = 262 - wattstopper_dlm = 263 - iso_14443 = 264 - SEQ_FIELDS = ["hdr", "packets"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['hdr']['start'] = self._io.pos() - self.hdr = self._root.Header(self._io, self, self._root) - self.hdr._read() - self._debug['hdr']['end'] = self._io.pos() - self._debug['packets']['start'] = self._io.pos() - self.packets = [] - i = 0 - while not self._io.is_eof(): - if not 'arr' in self._debug['packets']: - self._debug['packets']['arr'] = [] - self._debug['packets']['arr'].append({'start': self._io.pos()}) - _t_packets = self._root.Packet(self._io, self, self._root) - _t_packets._read() - self.packets.append(_t_packets) - self._debug['packets']['arr'][len(self.packets) - 1]['end'] = self._io.pos() - i += 1 - - self._debug['packets']['end'] = self._io.pos() - - class Header(KaitaiStruct): - """ - .. seealso:: - Source - https://wiki.wireshark.org/Development/LibpcapFileFormat#Global_Header - """ - SEQ_FIELDS = ["magic_number", "version_major", "version_minor", "thiszone", "sigfigs", "snaplen", "network"] - 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_number']['start'] = self._io.pos() - self.magic_number = self._io.ensure_fixed_contents(b"\xD4\xC3\xB2\xA1") - self._debug['magic_number']['end'] = self._io.pos() - self._debug['version_major']['start'] = self._io.pos() - self.version_major = self._io.read_u2le() - self._debug['version_major']['end'] = self._io.pos() - self._debug['version_minor']['start'] = self._io.pos() - self.version_minor = self._io.read_u2le() - self._debug['version_minor']['end'] = self._io.pos() - self._debug['thiszone']['start'] = self._io.pos() - self.thiszone = self._io.read_s4le() - self._debug['thiszone']['end'] = self._io.pos() - self._debug['sigfigs']['start'] = self._io.pos() - self.sigfigs = self._io.read_u4le() - self._debug['sigfigs']['end'] = self._io.pos() - self._debug['snaplen']['start'] = self._io.pos() - self.snaplen = self._io.read_u4le() - self._debug['snaplen']['end'] = self._io.pos() - self._debug['network']['start'] = self._io.pos() - self.network = KaitaiStream.resolve_enum(self._root.Linktype, self._io.read_u4le()) - self._debug['network']['end'] = self._io.pos() - - - class Packet(KaitaiStruct): - """ - .. seealso:: - Source - https://wiki.wireshark.org/Development/LibpcapFileFormat#Record_.28Packet.29_Header - """ - SEQ_FIELDS = ["ts_sec", "ts_usec", "incl_len", "orig_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['ts_sec']['start'] = self._io.pos() - self.ts_sec = self._io.read_u4le() - self._debug['ts_sec']['end'] = self._io.pos() - self._debug['ts_usec']['start'] = self._io.pos() - self.ts_usec = self._io.read_u4le() - self._debug['ts_usec']['end'] = self._io.pos() - self._debug['incl_len']['start'] = self._io.pos() - self.incl_len = self._io.read_u4le() - self._debug['incl_len']['end'] = self._io.pos() - self._debug['orig_len']['start'] = self._io.pos() - self.orig_len = self._io.read_u4le() - self._debug['orig_len']['end'] = self._io.pos() - self._debug['body']['start'] = self._io.pos() - _on = self._root.hdr.network - if _on == self._root.Linktype.ppi: - self._raw_body = self._io.read_bytes(self.incl_len) - io = KaitaiStream(BytesIO(self._raw_body)) - self.body = PacketPpi(io) - self.body._read() - elif _on == self._root.Linktype.ethernet: - self._raw_body = self._io.read_bytes(self.incl_len) - io = KaitaiStream(BytesIO(self._raw_body)) - self.body = EthernetFrame(io) - self.body._read() - else: - self.body = self._io.read_bytes(self.incl_len) - self._debug['body']['end'] = self._io.pos() - - - diff --git a/python/examples/kaitai/kaitai_struct_formats/network/protocol_body.py b/python/examples/kaitai/kaitai_struct_formats/network/protocol_body.py deleted file mode 100644 index 12d17565..00000000 --- a/python/examples/kaitai/kaitai_struct_formats/network/protocol_body.py +++ /dev/null @@ -1,260 +0,0 @@ -from __future__ import absolute_import -# This is a generated file! Please edit source .ksy file and use kaitai-struct-compiler to rebuild - -from pkg_resources import parse_version -from ...kaitaistruct import __version__ as ks_version, KaitaiStruct, KaitaiStream, BytesIO -from enum import Enum -import collections - - -if parse_version(ks_version) < parse_version('0.7'): - raise Exception("Incompatible Kaitai Struct Python API: 0.7 or later is required, but you have %s" % (ks_version)) - -from udp_datagram import UdpDatagram -from icmp_packet import IcmpPacket -from tcp_segment import TcpSegment -from ipv4_packet import Ipv4Packet -from ipv6_packet import Ipv6Packet -class ProtocolBody(KaitaiStruct): - """Protocol body represents particular payload on transport level (OSI - layer 4). - - Typically this payload in encapsulated into network level (OSI layer - 3) packet, which includes "protocol number" field that would be used - to decide what's inside the payload and how to parse it. Thanks to - IANA's standardization effort, multiple network level use the same - IDs for these payloads named "protocol numbers". - - This is effectively a "router" type: it expects to get protocol - number as a parameter, and then invokes relevant type parser based - on that parameter. - - .. seealso:: - Source - http://www.iana.org/assignments/protocol-numbers/protocol-numbers.xhtml - """ - - class ProtocolEnum(Enum): - hopopt = 0 - icmp = 1 - igmp = 2 - ggp = 3 - ipv4 = 4 - st = 5 - tcp = 6 - cbt = 7 - egp = 8 - igp = 9 - bbn_rcc_mon = 10 - nvp_ii = 11 - pup = 12 - argus = 13 - emcon = 14 - xnet = 15 - chaos = 16 - udp = 17 - mux = 18 - dcn_meas = 19 - hmp = 20 - prm = 21 - xns_idp = 22 - trunk_1 = 23 - trunk_2 = 24 - leaf_1 = 25 - leaf_2 = 26 - rdp = 27 - irtp = 28 - iso_tp4 = 29 - netblt = 30 - mfe_nsp = 31 - merit_inp = 32 - dccp = 33 - x_3pc = 34 - idpr = 35 - xtp = 36 - ddp = 37 - idpr_cmtp = 38 - tp_plus_plus = 39 - il = 40 - ipv6 = 41 - sdrp = 42 - ipv6_route = 43 - ipv6_frag = 44 - idrp = 45 - rsvp = 46 - gre = 47 - dsr = 48 - bna = 49 - esp = 50 - ah = 51 - i_nlsp = 52 - swipe = 53 - narp = 54 - mobile = 55 - tlsp = 56 - skip = 57 - ipv6_icmp = 58 - ipv6_nonxt = 59 - ipv6_opts = 60 - any_host_internal_protocol = 61 - cftp = 62 - any_local_network = 63 - sat_expak = 64 - kryptolan = 65 - rvd = 66 - ippc = 67 - any_distributed_file_system = 68 - sat_mon = 69 - visa = 70 - ipcv = 71 - cpnx = 72 - cphb = 73 - wsn = 74 - pvp = 75 - br_sat_mon = 76 - sun_nd = 77 - wb_mon = 78 - wb_expak = 79 - iso_ip = 80 - vmtp = 81 - secure_vmtp = 82 - vines = 83 - ttp_or_iptm = 84 - nsfnet_igp = 85 - dgp = 86 - tcf = 87 - eigrp = 88 - ospfigp = 89 - sprite_rpc = 90 - larp = 91 - mtp = 92 - ax_25 = 93 - ipip = 94 - micp = 95 - scc_sp = 96 - etherip = 97 - encap = 98 - any_private_encryption_scheme = 99 - gmtp = 100 - ifmp = 101 - pnni = 102 - pim = 103 - aris = 104 - scps = 105 - qnx = 106 - a_n = 107 - ipcomp = 108 - snp = 109 - compaq_peer = 110 - ipx_in_ip = 111 - vrrp = 112 - pgm = 113 - any_0_hop = 114 - l2tp = 115 - ddx = 116 - iatp = 117 - stp = 118 - srp = 119 - uti = 120 - smp = 121 - sm = 122 - ptp = 123 - isis_over_ipv4 = 124 - fire = 125 - crtp = 126 - crudp = 127 - sscopmce = 128 - iplt = 129 - sps = 130 - pipe = 131 - sctp = 132 - fc = 133 - rsvp_e2e_ignore = 134 - mobility_header = 135 - udplite = 136 - mpls_in_ip = 137 - manet = 138 - hip = 139 - shim6 = 140 - wesp = 141 - rohc = 142 - reserved_255 = 255 - SEQ_FIELDS = ["body"] - def __init__(self, protocol_num, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self.protocol_num = protocol_num - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['body']['start'] = self._io.pos() - _on = self.protocol - if _on == self._root.ProtocolEnum.ipv6_nonxt: - self.body = self._root.NoNextHeader(self._io, self, self._root) - self.body._read() - elif _on == self._root.ProtocolEnum.ipv4: - self.body = Ipv4Packet(self._io) - self.body._read() - elif _on == self._root.ProtocolEnum.udp: - self.body = UdpDatagram(self._io) - self.body._read() - elif _on == self._root.ProtocolEnum.icmp: - self.body = IcmpPacket(self._io) - self.body._read() - elif _on == self._root.ProtocolEnum.hopopt: - self.body = self._root.OptionHopByHop(self._io, self, self._root) - self.body._read() - elif _on == self._root.ProtocolEnum.ipv6: - self.body = Ipv6Packet(self._io) - self.body._read() - elif _on == self._root.ProtocolEnum.tcp: - self.body = TcpSegment(self._io) - self.body._read() - self._debug['body']['end'] = self._io.pos() - - class NoNextHeader(KaitaiStruct): - """Dummy type for IPv6 "no next header" type, which signifies end of headers chain.""" - 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 OptionHopByHop(KaitaiStruct): - SEQ_FIELDS = ["next_header_type", "hdr_ext_len", "body", "next_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['next_header_type']['start'] = self._io.pos() - self.next_header_type = self._io.read_u1() - self._debug['next_header_type']['end'] = self._io.pos() - self._debug['hdr_ext_len']['start'] = self._io.pos() - self.hdr_ext_len = self._io.read_u1() - self._debug['hdr_ext_len']['end'] = self._io.pos() - self._debug['body']['start'] = self._io.pos() - self.body = self._io.read_bytes((self.hdr_ext_len - 1)) - self._debug['body']['end'] = self._io.pos() - self._debug['next_header']['start'] = self._io.pos() - self.next_header = ProtocolBody(self.next_header_type, self._io) - self.next_header._read() - self._debug['next_header']['end'] = self._io.pos() - - - @property - def protocol(self): - if hasattr(self, '_m_protocol'): - return self._m_protocol if hasattr(self, '_m_protocol') else None - - self._m_protocol = KaitaiStream.resolve_enum(self._root.ProtocolEnum, self.protocol_num) - return self._m_protocol if hasattr(self, '_m_protocol') else None - - diff --git a/python/examples/kaitai/kaitai_struct_formats/network/rtcp_payload.py b/python/examples/kaitai/kaitai_struct_formats/network/rtcp_payload.py deleted file mode 100644 index 7450b3e7..00000000 --- a/python/examples/kaitai/kaitai_struct_formats/network/rtcp_payload.py +++ /dev/null @@ -1,579 +0,0 @@ -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 RtcpPayload(KaitaiStruct): - """RTCP is the Real-Time Control Protocol. - - .. seealso:: - Source - https://tools.ietf.org/html/rfc3550 - """ - - class PayloadType(Enum): - fir = 192 - nack = 193 - ij = 195 - sr = 200 - rr = 201 - sdes = 202 - bye = 203 - app = 204 - rtpfb = 205 - psfb = 206 - xr = 207 - avb = 208 - rsi = 209 - - class SdesSubtype(Enum): - pad = 0 - cname = 1 - name = 2 - email = 3 - phone = 4 - loc = 5 - tool = 6 - note = 7 - priv = 8 - - class PsfbSubtype(Enum): - pli = 1 - sli = 2 - rpsi = 3 - fir = 4 - tstr = 5 - tstn = 6 - vbcm = 7 - afb = 15 - - class RtpfbSubtype(Enum): - nack = 1 - tmmbr = 3 - tmmbn = 4 - rrr = 5 - transport_feedback = 15 - SEQ_FIELDS = ["rtcp_packets"] - 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['rtcp_packets']['start'] = self._io.pos() - self.rtcp_packets = [] - i = 0 - while not self._io.is_eof(): - if not 'arr' in self._debug['rtcp_packets']: - self._debug['rtcp_packets']['arr'] = [] - self._debug['rtcp_packets']['arr'].append({'start': self._io.pos()}) - _t_rtcp_packets = self._root.RtcpPacket(self._io, self, self._root) - _t_rtcp_packets._read() - self.rtcp_packets.append(_t_rtcp_packets) - self._debug['rtcp_packets']['arr'][len(self.rtcp_packets) - 1]['end'] = self._io.pos() - i += 1 - - self._debug['rtcp_packets']['end'] = self._io.pos() - - class PsfbAfbRembPacket(KaitaiStruct): - SEQ_FIELDS = ["num_ssrc", "br_exp", "br_mantissa", "ssrc_list"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['num_ssrc']['start'] = self._io.pos() - self.num_ssrc = self._io.read_u1() - self._debug['num_ssrc']['end'] = self._io.pos() - self._debug['br_exp']['start'] = self._io.pos() - self.br_exp = self._io.read_bits_int(6) - self._debug['br_exp']['end'] = self._io.pos() - self._debug['br_mantissa']['start'] = self._io.pos() - self.br_mantissa = self._io.read_bits_int(18) - self._debug['br_mantissa']['end'] = self._io.pos() - self._io.align_to_byte() - self._debug['ssrc_list']['start'] = self._io.pos() - self.ssrc_list = [None] * (self.num_ssrc) - for i in range(self.num_ssrc): - if not 'arr' in self._debug['ssrc_list']: - self._debug['ssrc_list']['arr'] = [] - self._debug['ssrc_list']['arr'].append({'start': self._io.pos()}) - self.ssrc_list[i] = self._io.read_u4be() - self._debug['ssrc_list']['arr'][i]['end'] = self._io.pos() - - self._debug['ssrc_list']['end'] = self._io.pos() - - @property - def max_total_bitrate(self): - if hasattr(self, '_m_max_total_bitrate'): - return self._m_max_total_bitrate if hasattr(self, '_m_max_total_bitrate') else None - - self._m_max_total_bitrate = (self.br_mantissa * (1 << self.br_exp)) - return self._m_max_total_bitrate if hasattr(self, '_m_max_total_bitrate') else None - - - class SrPacket(KaitaiStruct): - SEQ_FIELDS = ["ssrc", "ntp_msw", "ntp_lsw", "rtp_timestamp", "sender_packet_count", "sender_octet_count", "report_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['ssrc']['start'] = self._io.pos() - self.ssrc = self._io.read_u4be() - self._debug['ssrc']['end'] = self._io.pos() - self._debug['ntp_msw']['start'] = self._io.pos() - self.ntp_msw = self._io.read_u4be() - self._debug['ntp_msw']['end'] = self._io.pos() - self._debug['ntp_lsw']['start'] = self._io.pos() - self.ntp_lsw = self._io.read_u4be() - self._debug['ntp_lsw']['end'] = self._io.pos() - self._debug['rtp_timestamp']['start'] = self._io.pos() - self.rtp_timestamp = self._io.read_u4be() - self._debug['rtp_timestamp']['end'] = self._io.pos() - self._debug['sender_packet_count']['start'] = self._io.pos() - self.sender_packet_count = self._io.read_u4be() - self._debug['sender_packet_count']['end'] = self._io.pos() - self._debug['sender_octet_count']['start'] = self._io.pos() - self.sender_octet_count = self._io.read_u4be() - self._debug['sender_octet_count']['end'] = self._io.pos() - self._debug['report_block']['start'] = self._io.pos() - self.report_block = [None] * (self._parent.subtype) - for i in range(self._parent.subtype): - if not 'arr' in self._debug['report_block']: - self._debug['report_block']['arr'] = [] - self._debug['report_block']['arr'].append({'start': self._io.pos()}) - _t_report_block = self._root.ReportBlock(self._io, self, self._root) - _t_report_block._read() - self.report_block[i] = _t_report_block - self._debug['report_block']['arr'][i]['end'] = self._io.pos() - - self._debug['report_block']['end'] = self._io.pos() - - @property - def ntp(self): - if hasattr(self, '_m_ntp'): - return self._m_ntp if hasattr(self, '_m_ntp') else None - - self._m_ntp = ((self.ntp_msw << 32) & self.ntp_lsw) - return self._m_ntp if hasattr(self, '_m_ntp') else None - - - class RrPacket(KaitaiStruct): - SEQ_FIELDS = ["ssrc", "report_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['ssrc']['start'] = self._io.pos() - self.ssrc = self._io.read_u4be() - self._debug['ssrc']['end'] = self._io.pos() - self._debug['report_block']['start'] = self._io.pos() - self.report_block = [None] * (self._parent.subtype) - for i in range(self._parent.subtype): - if not 'arr' in self._debug['report_block']: - self._debug['report_block']['arr'] = [] - self._debug['report_block']['arr'].append({'start': self._io.pos()}) - _t_report_block = self._root.ReportBlock(self._io, self, self._root) - _t_report_block._read() - self.report_block[i] = _t_report_block - self._debug['report_block']['arr'][i]['end'] = self._io.pos() - - self._debug['report_block']['end'] = self._io.pos() - - - class RtcpPacket(KaitaiStruct): - SEQ_FIELDS = ["version", "padding", "subtype", "payload_type", "length", "body"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['version']['start'] = self._io.pos() - self.version = self._io.read_bits_int(2) - self._debug['version']['end'] = self._io.pos() - self._debug['padding']['start'] = self._io.pos() - self.padding = self._io.read_bits_int(1) != 0 - self._debug['padding']['end'] = self._io.pos() - self._debug['subtype']['start'] = self._io.pos() - self.subtype = self._io.read_bits_int(5) - self._debug['subtype']['end'] = self._io.pos() - self._io.align_to_byte() - self._debug['payload_type']['start'] = self._io.pos() - self.payload_type = KaitaiStream.resolve_enum(self._root.PayloadType, self._io.read_u1()) - self._debug['payload_type']['end'] = self._io.pos() - self._debug['length']['start'] = self._io.pos() - self.length = self._io.read_u2be() - self._debug['length']['end'] = self._io.pos() - self._debug['body']['start'] = self._io.pos() - _on = self.payload_type - if _on == self._root.PayloadType.sr: - self._raw_body = self._io.read_bytes((4 * self.length)) - io = KaitaiStream(BytesIO(self._raw_body)) - self.body = self._root.SrPacket(io, self, self._root) - self.body._read() - elif _on == self._root.PayloadType.psfb: - self._raw_body = self._io.read_bytes((4 * self.length)) - io = KaitaiStream(BytesIO(self._raw_body)) - self.body = self._root.PsfbPacket(io, self, self._root) - self.body._read() - elif _on == self._root.PayloadType.rr: - self._raw_body = self._io.read_bytes((4 * self.length)) - io = KaitaiStream(BytesIO(self._raw_body)) - self.body = self._root.RrPacket(io, self, self._root) - self.body._read() - elif _on == self._root.PayloadType.rtpfb: - self._raw_body = self._io.read_bytes((4 * self.length)) - io = KaitaiStream(BytesIO(self._raw_body)) - self.body = self._root.RtpfbPacket(io, self, self._root) - self.body._read() - elif _on == self._root.PayloadType.sdes: - self._raw_body = self._io.read_bytes((4 * self.length)) - io = KaitaiStream(BytesIO(self._raw_body)) - self.body = self._root.SdesPacket(io, self, self._root) - self.body._read() - else: - self.body = self._io.read_bytes((4 * self.length)) - self._debug['body']['end'] = self._io.pos() - - - class SdesTlv(KaitaiStruct): - SEQ_FIELDS = ["type", "length", "value"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['type']['start'] = self._io.pos() - self.type = KaitaiStream.resolve_enum(self._root.SdesSubtype, self._io.read_u1()) - self._debug['type']['end'] = self._io.pos() - if self.type != self._root.SdesSubtype.pad: - self._debug['length']['start'] = self._io.pos() - self.length = self._io.read_u1() - self._debug['length']['end'] = self._io.pos() - - if self.type != self._root.SdesSubtype.pad: - self._debug['value']['start'] = self._io.pos() - self.value = self._io.read_bytes(self.length) - self._debug['value']['end'] = self._io.pos() - - - - class ReportBlock(KaitaiStruct): - SEQ_FIELDS = ["ssrc_source", "lost_val", "highest_seq_num_received", "interarrival_jitter", "lsr", "dlsr"] - 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['ssrc_source']['start'] = self._io.pos() - self.ssrc_source = self._io.read_u4be() - self._debug['ssrc_source']['end'] = self._io.pos() - self._debug['lost_val']['start'] = self._io.pos() - self.lost_val = self._io.read_u1() - self._debug['lost_val']['end'] = self._io.pos() - self._debug['highest_seq_num_received']['start'] = self._io.pos() - self.highest_seq_num_received = self._io.read_u4be() - self._debug['highest_seq_num_received']['end'] = self._io.pos() - self._debug['interarrival_jitter']['start'] = self._io.pos() - self.interarrival_jitter = self._io.read_u4be() - self._debug['interarrival_jitter']['end'] = self._io.pos() - self._debug['lsr']['start'] = self._io.pos() - self.lsr = self._io.read_u4be() - self._debug['lsr']['end'] = self._io.pos() - self._debug['dlsr']['start'] = self._io.pos() - self.dlsr = self._io.read_u4be() - self._debug['dlsr']['end'] = self._io.pos() - - @property - def fraction_lost(self): - if hasattr(self, '_m_fraction_lost'): - return self._m_fraction_lost if hasattr(self, '_m_fraction_lost') else None - - self._m_fraction_lost = (self.lost_val >> 24) - return self._m_fraction_lost if hasattr(self, '_m_fraction_lost') else None - - @property - def cumulative_packets_lost(self): - if hasattr(self, '_m_cumulative_packets_lost'): - return self._m_cumulative_packets_lost if hasattr(self, '_m_cumulative_packets_lost') else None - - self._m_cumulative_packets_lost = (self.lost_val & 16777215) - return self._m_cumulative_packets_lost if hasattr(self, '_m_cumulative_packets_lost') else None - - - class RtpfbTransportFeedbackPacket(KaitaiStruct): - SEQ_FIELDS = ["base_sequence_number", "packet_status_count", "b4", "remaining"] - 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['base_sequence_number']['start'] = self._io.pos() - self.base_sequence_number = self._io.read_u2be() - self._debug['base_sequence_number']['end'] = self._io.pos() - self._debug['packet_status_count']['start'] = self._io.pos() - self.packet_status_count = self._io.read_u2be() - self._debug['packet_status_count']['end'] = self._io.pos() - self._debug['b4']['start'] = self._io.pos() - self.b4 = self._io.read_u4be() - self._debug['b4']['end'] = self._io.pos() - self._debug['remaining']['start'] = self._io.pos() - self.remaining = self._io.read_bytes_full() - self._debug['remaining']['end'] = self._io.pos() - - @property - def reference_time(self): - if hasattr(self, '_m_reference_time'): - return self._m_reference_time if hasattr(self, '_m_reference_time') else None - - self._m_reference_time = (self.b4 >> 8) - return self._m_reference_time if hasattr(self, '_m_reference_time') else None - - @property - def fb_pkt_count(self): - if hasattr(self, '_m_fb_pkt_count'): - return self._m_fb_pkt_count if hasattr(self, '_m_fb_pkt_count') else None - - self._m_fb_pkt_count = (self.b4 & 255) - return self._m_fb_pkt_count if hasattr(self, '_m_fb_pkt_count') else None - - @property - def packet_status(self): - if hasattr(self, '_m_packet_status'): - return self._m_packet_status if hasattr(self, '_m_packet_status') else None - - self._debug['_m_packet_status']['start'] = self._io.pos() - self._m_packet_status = self._io.read_bytes(0) - self._debug['_m_packet_status']['end'] = self._io.pos() - return self._m_packet_status if hasattr(self, '_m_packet_status') else None - - @property - def recv_delta(self): - if hasattr(self, '_m_recv_delta'): - return self._m_recv_delta if hasattr(self, '_m_recv_delta') else None - - self._debug['_m_recv_delta']['start'] = self._io.pos() - self._m_recv_delta = self._io.read_bytes(0) - self._debug['_m_recv_delta']['end'] = self._io.pos() - return self._m_recv_delta if hasattr(self, '_m_recv_delta') else None - - - class PsfbPacket(KaitaiStruct): - SEQ_FIELDS = ["ssrc", "ssrc_media_source", "fci_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['ssrc']['start'] = self._io.pos() - self.ssrc = self._io.read_u4be() - self._debug['ssrc']['end'] = self._io.pos() - self._debug['ssrc_media_source']['start'] = self._io.pos() - self.ssrc_media_source = self._io.read_u4be() - self._debug['ssrc_media_source']['end'] = self._io.pos() - self._debug['fci_block']['start'] = self._io.pos() - _on = self.fmt - if _on == self._root.PsfbSubtype.afb: - self._raw_fci_block = self._io.read_bytes_full() - io = KaitaiStream(BytesIO(self._raw_fci_block)) - self.fci_block = self._root.PsfbAfbPacket(io, self, self._root) - self.fci_block._read() - else: - self.fci_block = self._io.read_bytes_full() - self._debug['fci_block']['end'] = self._io.pos() - - @property - def fmt(self): - if hasattr(self, '_m_fmt'): - return self._m_fmt if hasattr(self, '_m_fmt') else None - - self._m_fmt = KaitaiStream.resolve_enum(self._root.PsfbSubtype, self._parent.subtype) - return self._m_fmt if hasattr(self, '_m_fmt') else None - - - class SourceChunk(KaitaiStruct): - SEQ_FIELDS = ["ssrc", "sdes_tlv"] - 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['ssrc']['start'] = self._io.pos() - self.ssrc = self._io.read_u4be() - self._debug['ssrc']['end'] = self._io.pos() - self._debug['sdes_tlv']['start'] = self._io.pos() - self.sdes_tlv = [] - i = 0 - while not self._io.is_eof(): - if not 'arr' in self._debug['sdes_tlv']: - self._debug['sdes_tlv']['arr'] = [] - self._debug['sdes_tlv']['arr'].append({'start': self._io.pos()}) - _t_sdes_tlv = self._root.SdesTlv(self._io, self, self._root) - _t_sdes_tlv._read() - self.sdes_tlv.append(_t_sdes_tlv) - self._debug['sdes_tlv']['arr'][len(self.sdes_tlv) - 1]['end'] = self._io.pos() - i += 1 - - self._debug['sdes_tlv']['end'] = self._io.pos() - - - class SdesPacket(KaitaiStruct): - SEQ_FIELDS = ["source_chunk"] - 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['source_chunk']['start'] = self._io.pos() - self.source_chunk = [None] * (self.source_count) - for i in range(self.source_count): - if not 'arr' in self._debug['source_chunk']: - self._debug['source_chunk']['arr'] = [] - self._debug['source_chunk']['arr'].append({'start': self._io.pos()}) - _t_source_chunk = self._root.SourceChunk(self._io, self, self._root) - _t_source_chunk._read() - self.source_chunk[i] = _t_source_chunk - self._debug['source_chunk']['arr'][i]['end'] = self._io.pos() - - self._debug['source_chunk']['end'] = self._io.pos() - - @property - def source_count(self): - if hasattr(self, '_m_source_count'): - return self._m_source_count if hasattr(self, '_m_source_count') else None - - self._m_source_count = self._parent.subtype - return self._m_source_count if hasattr(self, '_m_source_count') else None - - - class RtpfbPacket(KaitaiStruct): - SEQ_FIELDS = ["ssrc", "ssrc_media_source", "fci_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['ssrc']['start'] = self._io.pos() - self.ssrc = self._io.read_u4be() - self._debug['ssrc']['end'] = self._io.pos() - self._debug['ssrc_media_source']['start'] = self._io.pos() - self.ssrc_media_source = self._io.read_u4be() - self._debug['ssrc_media_source']['end'] = self._io.pos() - self._debug['fci_block']['start'] = self._io.pos() - _on = self.fmt - if _on == self._root.RtpfbSubtype.transport_feedback: - self._raw_fci_block = self._io.read_bytes_full() - io = KaitaiStream(BytesIO(self._raw_fci_block)) - self.fci_block = self._root.RtpfbTransportFeedbackPacket(io, self, self._root) - self.fci_block._read() - else: - self.fci_block = self._io.read_bytes_full() - self._debug['fci_block']['end'] = self._io.pos() - - @property - def fmt(self): - if hasattr(self, '_m_fmt'): - return self._m_fmt if hasattr(self, '_m_fmt') else None - - self._m_fmt = KaitaiStream.resolve_enum(self._root.RtpfbSubtype, self._parent.subtype) - return self._m_fmt if hasattr(self, '_m_fmt') else None - - - class PacketStatusChunk(KaitaiStruct): - SEQ_FIELDS = ["t", "s2", "s1", "rle", "symbol_list"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['t']['start'] = self._io.pos() - self.t = self._io.read_bits_int(1) != 0 - self._debug['t']['end'] = self._io.pos() - if int(self.t) == 0: - self._debug['s2']['start'] = self._io.pos() - self.s2 = self._io.read_bits_int(2) - self._debug['s2']['end'] = self._io.pos() - - if int(self.t) == 1: - self._debug['s1']['start'] = self._io.pos() - self.s1 = self._io.read_bits_int(1) != 0 - self._debug['s1']['end'] = self._io.pos() - - if int(self.t) == 0: - self._debug['rle']['start'] = self._io.pos() - self.rle = self._io.read_bits_int(13) - self._debug['rle']['end'] = self._io.pos() - - if int(self.t) == 1: - self._debug['symbol_list']['start'] = self._io.pos() - self.symbol_list = self._io.read_bits_int(14) - self._debug['symbol_list']['end'] = self._io.pos() - - - @property - def s(self): - if hasattr(self, '_m_s'): - return self._m_s if hasattr(self, '_m_s') else None - - self._m_s = (self.s2 if int(self.t) == 0 else (1 if int(self.s1) == 0 else 0)) - return self._m_s if hasattr(self, '_m_s') else None - - - class PsfbAfbPacket(KaitaiStruct): - SEQ_FIELDS = ["uid", "contents"] - 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['uid']['start'] = self._io.pos() - self.uid = self._io.read_u4be() - self._debug['uid']['end'] = self._io.pos() - self._debug['contents']['start'] = self._io.pos() - _on = self.uid - if _on == 1380273474: - self._raw_contents = self._io.read_bytes_full() - io = KaitaiStream(BytesIO(self._raw_contents)) - self.contents = self._root.PsfbAfbRembPacket(io, self, self._root) - self.contents._read() - else: - self.contents = self._io.read_bytes_full() - self._debug['contents']['end'] = self._io.pos() - - - diff --git a/python/examples/kaitai/kaitai_struct_formats/network/rtp_packet.py b/python/examples/kaitai/kaitai_struct_formats/network/rtp_packet.py deleted file mode 100644 index 64e50cba..00000000 --- a/python/examples/kaitai/kaitai_struct_formats/network/rtp_packet.py +++ /dev/null @@ -1,150 +0,0 @@ -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 RtpPacket(KaitaiStruct): - """The Real-time Transport Protocol (RTP) is a widely used network - protocol for transmitting audio or video. It usually works with the - RTP Control Protocol (RTCP). The transmission can be based on - Transmission Control Protocol (TCP) or User Datagram Protocol (UDP). - """ - - class PayloadTypeEnum(Enum): - pcmu = 0 - reserved1 = 1 - reserved2 = 2 - gsm = 3 - g723 = 4 - dvi4_1 = 5 - dvi4_2 = 6 - lpc = 7 - pama = 8 - g722 = 9 - l16_1 = 10 - l16_2 = 11 - qcelp = 12 - cn = 13 - mpa = 14 - g728 = 15 - dvi4_3 = 16 - dvi4_4 = 17 - g729 = 18 - reserved3 = 19 - unassigned1 = 20 - unassigned2 = 21 - unassigned3 = 22 - unassigned4 = 23 - unassigned5 = 24 - celb = 25 - jpeg = 26 - unassigned6 = 27 - nv = 28 - unassigned7 = 29 - unassigned8 = 30 - h261 = 31 - mpv = 32 - mp2t = 33 - h263 = 34 - mpeg_ps = 96 - SEQ_FIELDS = ["version", "has_padding", "has_extension", "csrc_count", "marker", "payload_type", "sequence_number", "timestamp", "ssrc", "header_extension", "data", "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['version']['start'] = self._io.pos() - self.version = self._io.read_bits_int(2) - self._debug['version']['end'] = self._io.pos() - self._debug['has_padding']['start'] = self._io.pos() - self.has_padding = self._io.read_bits_int(1) != 0 - self._debug['has_padding']['end'] = self._io.pos() - self._debug['has_extension']['start'] = self._io.pos() - self.has_extension = self._io.read_bits_int(1) != 0 - self._debug['has_extension']['end'] = self._io.pos() - self._debug['csrc_count']['start'] = self._io.pos() - self.csrc_count = self._io.read_bits_int(4) - self._debug['csrc_count']['end'] = self._io.pos() - self._debug['marker']['start'] = self._io.pos() - self.marker = self._io.read_bits_int(1) != 0 - self._debug['marker']['end'] = self._io.pos() - self._debug['payload_type']['start'] = self._io.pos() - self.payload_type = KaitaiStream.resolve_enum(self._root.PayloadTypeEnum, self._io.read_bits_int(7)) - self._debug['payload_type']['end'] = self._io.pos() - self._io.align_to_byte() - self._debug['sequence_number']['start'] = self._io.pos() - self.sequence_number = self._io.read_u2be() - self._debug['sequence_number']['end'] = self._io.pos() - self._debug['timestamp']['start'] = self._io.pos() - self.timestamp = self._io.read_u4be() - self._debug['timestamp']['end'] = self._io.pos() - self._debug['ssrc']['start'] = self._io.pos() - self.ssrc = self._io.read_u4be() - self._debug['ssrc']['end'] = self._io.pos() - if self.has_extension: - self._debug['header_extension']['start'] = self._io.pos() - self.header_extension = self._root.HeaderExtention(self._io, self, self._root) - self.header_extension._read() - self._debug['header_extension']['end'] = self._io.pos() - - self._debug['data']['start'] = self._io.pos() - self.data = self._io.read_bytes(((self._io.size() - self._io.pos()) - self.len_padding)) - self._debug['data']['end'] = self._io.pos() - self._debug['padding']['start'] = self._io.pos() - self.padding = self._io.read_bytes(self.len_padding) - self._debug['padding']['end'] = self._io.pos() - - class HeaderExtention(KaitaiStruct): - SEQ_FIELDS = ["id", "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['id']['start'] = self._io.pos() - self.id = self._io.read_u2be() - self._debug['id']['end'] = self._io.pos() - self._debug['length']['start'] = self._io.pos() - self.length = self._io.read_u2be() - self._debug['length']['end'] = self._io.pos() - - - @property - def len_padding_if_exists(self): - """If padding bit is enabled, last byte of data contains number of - bytes appended to the payload as padding. - """ - if hasattr(self, '_m_len_padding_if_exists'): - return self._m_len_padding_if_exists if hasattr(self, '_m_len_padding_if_exists') else None - - if self.has_padding: - _pos = self._io.pos() - self._io.seek((self._io.size() - 1)) - self._debug['_m_len_padding_if_exists']['start'] = self._io.pos() - self._m_len_padding_if_exists = self._io.read_u1() - self._debug['_m_len_padding_if_exists']['end'] = self._io.pos() - self._io.seek(_pos) - - return self._m_len_padding_if_exists if hasattr(self, '_m_len_padding_if_exists') else None - - @property - def len_padding(self): - """Always returns number of padding bytes to in the payload.""" - if hasattr(self, '_m_len_padding'): - return self._m_len_padding if hasattr(self, '_m_len_padding') else None - - self._m_len_padding = (self.len_padding_if_exists if self.has_padding else 0) - return self._m_len_padding if hasattr(self, '_m_len_padding') else None - - diff --git a/python/examples/kaitai/kaitai_struct_formats/network/tcp_segment.py b/python/examples/kaitai/kaitai_struct_formats/network/tcp_segment.py deleted file mode 100644 index 222f6794..00000000 --- a/python/examples/kaitai/kaitai_struct_formats/network/tcp_segment.py +++ /dev/null @@ -1,57 +0,0 @@ -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 TcpSegment(KaitaiStruct): - """TCP is one of the core Internet protocols on transport layer (AKA - OSI layer 4), providing stateful connections with error checking, - guarantees of delivery, order of segments and avoidance of duplicate - delivery. - """ - SEQ_FIELDS = ["src_port", "dst_port", "seq_num", "ack_num", "b12", "b13", "window_size", "checksum", "urgent_pointer", "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['src_port']['start'] = self._io.pos() - self.src_port = self._io.read_u2be() - self._debug['src_port']['end'] = self._io.pos() - self._debug['dst_port']['start'] = self._io.pos() - self.dst_port = self._io.read_u2be() - self._debug['dst_port']['end'] = self._io.pos() - self._debug['seq_num']['start'] = self._io.pos() - self.seq_num = self._io.read_u4be() - self._debug['seq_num']['end'] = self._io.pos() - self._debug['ack_num']['start'] = self._io.pos() - self.ack_num = self._io.read_u4be() - self._debug['ack_num']['end'] = self._io.pos() - self._debug['b12']['start'] = self._io.pos() - self.b12 = self._io.read_u1() - self._debug['b12']['end'] = self._io.pos() - self._debug['b13']['start'] = self._io.pos() - self.b13 = self._io.read_u1() - self._debug['b13']['end'] = self._io.pos() - self._debug['window_size']['start'] = self._io.pos() - self.window_size = self._io.read_u2be() - self._debug['window_size']['end'] = self._io.pos() - self._debug['checksum']['start'] = self._io.pos() - self.checksum = self._io.read_u2be() - self._debug['checksum']['end'] = self._io.pos() - self._debug['urgent_pointer']['start'] = self._io.pos() - self.urgent_pointer = self._io.read_u2be() - self._debug['urgent_pointer']['end'] = self._io.pos() - self._debug['body']['start'] = self._io.pos() - self.body = self._io.read_bytes_full() - self._debug['body']['end'] = self._io.pos() - - diff --git a/python/examples/kaitai/kaitai_struct_formats/network/tls_client_hello.py b/python/examples/kaitai/kaitai_struct_formats/network/tls_client_hello.py deleted file mode 100644 index 118eec3d..00000000 --- a/python/examples/kaitai/kaitai_struct_formats/network/tls_client_hello.py +++ /dev/null @@ -1,293 +0,0 @@ -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 TlsClientHello(KaitaiStruct): - SEQ_FIELDS = ["version", "random", "session_id", "cipher_suites", "compression_methods", "extensions"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['version']['start'] = self._io.pos() - self.version = self._root.Version(self._io, self, self._root) - self.version._read() - self._debug['version']['end'] = self._io.pos() - self._debug['random']['start'] = self._io.pos() - self.random = self._root.Random(self._io, self, self._root) - self.random._read() - self._debug['random']['end'] = self._io.pos() - self._debug['session_id']['start'] = self._io.pos() - self.session_id = self._root.SessionId(self._io, self, self._root) - self.session_id._read() - self._debug['session_id']['end'] = self._io.pos() - self._debug['cipher_suites']['start'] = self._io.pos() - self.cipher_suites = self._root.CipherSuites(self._io, self, self._root) - self.cipher_suites._read() - self._debug['cipher_suites']['end'] = self._io.pos() - self._debug['compression_methods']['start'] = self._io.pos() - self.compression_methods = self._root.CompressionMethods(self._io, self, self._root) - self.compression_methods._read() - self._debug['compression_methods']['end'] = self._io.pos() - if self._io.is_eof() == False: - self._debug['extensions']['start'] = self._io.pos() - self.extensions = self._root.Extensions(self._io, self, self._root) - self.extensions._read() - self._debug['extensions']['end'] = self._io.pos() - - - class ServerName(KaitaiStruct): - SEQ_FIELDS = ["name_type", "length", "host_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_type']['start'] = self._io.pos() - self.name_type = self._io.read_u1() - self._debug['name_type']['end'] = self._io.pos() - self._debug['length']['start'] = self._io.pos() - self.length = self._io.read_u2be() - self._debug['length']['end'] = self._io.pos() - self._debug['host_name']['start'] = self._io.pos() - self.host_name = self._io.read_bytes(self.length) - self._debug['host_name']['end'] = self._io.pos() - - - class Random(KaitaiStruct): - SEQ_FIELDS = ["gmt_unix_time", "random"] - 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['gmt_unix_time']['start'] = self._io.pos() - self.gmt_unix_time = self._io.read_u4be() - self._debug['gmt_unix_time']['end'] = self._io.pos() - self._debug['random']['start'] = self._io.pos() - self.random = self._io.read_bytes(28) - self._debug['random']['end'] = self._io.pos() - - - class SessionId(KaitaiStruct): - SEQ_FIELDS = ["len", "sid"] - 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() - self._debug['sid']['start'] = self._io.pos() - self.sid = self._io.read_bytes(self.len) - self._debug['sid']['end'] = self._io.pos() - - - class Sni(KaitaiStruct): - SEQ_FIELDS = ["list_length", "server_names"] - 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['list_length']['start'] = self._io.pos() - self.list_length = self._io.read_u2be() - self._debug['list_length']['end'] = self._io.pos() - self._debug['server_names']['start'] = self._io.pos() - self.server_names = [] - i = 0 - while not self._io.is_eof(): - if not 'arr' in self._debug['server_names']: - self._debug['server_names']['arr'] = [] - self._debug['server_names']['arr'].append({'start': self._io.pos()}) - _t_server_names = self._root.ServerName(self._io, self, self._root) - _t_server_names._read() - self.server_names.append(_t_server_names) - self._debug['server_names']['arr'][len(self.server_names) - 1]['end'] = self._io.pos() - i += 1 - - self._debug['server_names']['end'] = self._io.pos() - - - class CipherSuites(KaitaiStruct): - SEQ_FIELDS = ["len", "cipher_suites"] - 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_u2be() - self._debug['len']['end'] = self._io.pos() - self._debug['cipher_suites']['start'] = self._io.pos() - self.cipher_suites = [None] * (self.len // 2) - for i in range(self.len // 2): - if not 'arr' in self._debug['cipher_suites']: - self._debug['cipher_suites']['arr'] = [] - self._debug['cipher_suites']['arr'].append({'start': self._io.pos()}) - self.cipher_suites[i] = self._io.read_u2be() - self._debug['cipher_suites']['arr'][i]['end'] = self._io.pos() - - self._debug['cipher_suites']['end'] = self._io.pos() - - - class CompressionMethods(KaitaiStruct): - SEQ_FIELDS = ["len", "compression_methods"] - def __init__(self, _io, _parent=None, _root=None): - self._io = _io - self._parent = _parent - self._root = _root if _root else self - self._debug = collections.defaultdict(dict) - - def _read(self): - self._debug['len']['start'] = self._io.pos() - self.len = self._io.read_u1() - self._debug['len']['end'] = self._io.pos() - self._debug['compression_methods']['start'] = self._io.pos() - self.compression_methods = self._io.read_bytes(self.len) - self._debug['compression_methods']['end'] = self._io.pos() - - - class Alpn(KaitaiStruct): - SEQ_FIELDS = ["ext_len", "alpn_protocols"] - 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['ext_len']['start'] = self._io.pos() - self.ext_len = self._io.read_u2be() - self._debug['ext_len']['end'] = self._io.pos() - self._debug['alpn_protocols']['start'] = self._io.pos() - self.alpn_protocols = [] - i = 0 - while not self._io.is_eof(): - if not 'arr' in self._debug['alpn_protocols']: - self._debug['alpn_protocols']['arr'] = [] - self._debug['alpn_protocols']['arr'].append({'start': self._io.pos()}) - _t_alpn_protocols = self._root.Protocol(self._io, self, self._root) - _t_alpn_protocols._read() - self.alpn_protocols.append(_t_alpn_protocols) - self._debug['alpn_protocols']['arr'][len(self.alpn_protocols) - 1]['end'] = self._io.pos() - i += 1 - - self._debug['alpn_protocols']['end'] = self._io.pos() - - - class Extensions(KaitaiStruct): - SEQ_FIELDS = ["len", "extensions"] - 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_u2be() - self._debug['len']['end'] = self._io.pos() - self._debug['extensions']['start'] = self._io.pos() - self.extensions = [] - i = 0 - while not self._io.is_eof(): - if not 'arr' in self._debug['extensions']: - self._debug['extensions']['arr'] = [] - self._debug['extensions']['arr'].append({'start': self._io.pos()}) - _t_extensions = self._root.Extension(self._io, self, self._root) - _t_extensions._read() - self.extensions.append(_t_extensions) - self._debug['extensions']['arr'][len(self.extensions) - 1]['end'] = self._io.pos() - i += 1 - - self._debug['extensions']['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_u1() - self._debug['major']['end'] = self._io.pos() - self._debug['minor']['start'] = self._io.pos() - self.minor = self._io.read_u1() - self._debug['minor']['end'] = self._io.pos() - - - class Protocol(KaitaiStruct): - SEQ_FIELDS = ["strlen", "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['strlen']['start'] = self._io.pos() - self.strlen = self._io.read_u1() - self._debug['strlen']['end'] = self._io.pos() - self._debug['name']['start'] = self._io.pos() - self.name = self._io.read_bytes(self.strlen) - self._debug['name']['end'] = self._io.pos() - - - class Extension(KaitaiStruct): - SEQ_FIELDS = ["type", "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 = self._io.read_u2be() - self._debug['type']['end'] = self._io.pos() - self._debug['len']['start'] = self._io.pos() - self.len = self._io.read_u2be() - self._debug['len']['end'] = self._io.pos() - self._debug['body']['start'] = self._io.pos() - _on = self.type - if _on == 0: - self._raw_body = self._io.read_bytes(self.len) - io = KaitaiStream(BytesIO(self._raw_body)) - self.body = self._root.Sni(io, self, self._root) - self.body._read() - elif _on == 16: - self._raw_body = self._io.read_bytes(self.len) - io = KaitaiStream(BytesIO(self._raw_body)) - self.body = self._root.Alpn(io, self, self._root) - self.body._read() - else: - self.body = self._io.read_bytes(self.len) - self._debug['body']['end'] = self._io.pos() - - - diff --git a/python/examples/kaitai/kaitai_struct_formats/network/udp_datagram.py b/python/examples/kaitai/kaitai_struct_formats/network/udp_datagram.py deleted file mode 100644 index d3cac12f..00000000 --- a/python/examples/kaitai/kaitai_struct_formats/network/udp_datagram.py +++ /dev/null @@ -1,42 +0,0 @@ -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 UdpDatagram(KaitaiStruct): - """UDP is a simple stateless transport layer (AKA OSI layer 4) - protocol, one of the core Internet protocols. It provides source and - destination ports, basic checksumming, but provides not guarantees - of delivery, order of packets, or duplicate delivery. - """ - SEQ_FIELDS = ["src_port", "dst_port", "length", "checksum", "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['src_port']['start'] = self._io.pos() - self.src_port = self._io.read_u2be() - self._debug['src_port']['end'] = self._io.pos() - self._debug['dst_port']['start'] = self._io.pos() - self.dst_port = self._io.read_u2be() - self._debug['dst_port']['end'] = self._io.pos() - self._debug['length']['start'] = self._io.pos() - self.length = self._io.read_u2be() - self._debug['length']['end'] = self._io.pos() - self._debug['checksum']['start'] = self._io.pos() - self.checksum = self._io.read_u2be() - self._debug['checksum']['end'] = self._io.pos() - self._debug['body']['start'] = self._io.pos() - self.body = self._io.read_bytes_full() - self._debug['body']['end'] = self._io.pos() - - diff --git a/python/examples/kaitai/kaitai_struct_formats/network/windows_systemtime.py b/python/examples/kaitai/kaitai_struct_formats/network/windows_systemtime.py deleted file mode 100644 index 2f3bc5df..00000000 --- a/python/examples/kaitai/kaitai_struct_formats/network/windows_systemtime.py +++ /dev/null @@ -1,51 +0,0 @@ -# This is a generated file! Please edit source .ksy file and use kaitai-struct-compiler to rebuild - -from pkg_resources import parse_version -from kaitaistruct import __version__ as ks_version, KaitaiStruct, KaitaiStream, BytesIO -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 WindowsSystemtime(KaitaiStruct): - """Microsoft Windows SYSTEMTIME structure, stores individual components - of date and time as individual fields, up to millisecond precision. - - .. seealso:: - Source - https://msdn.microsoft.com/en-us/library/windows/desktop/ms724950.aspx - """ - SEQ_FIELDS = ["year", "month", "dow", "day", "hour", "min", "sec", "msec"] - 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_u2le() - self._debug['year']['end'] = self._io.pos() - self._debug['month']['start'] = self._io.pos() - self.month = self._io.read_u2le() - self._debug['month']['end'] = self._io.pos() - self._debug['dow']['start'] = self._io.pos() - self.dow = self._io.read_u2le() - self._debug['dow']['end'] = self._io.pos() - self._debug['day']['start'] = self._io.pos() - self.day = self._io.read_u2le() - self._debug['day']['end'] = self._io.pos() - self._debug['hour']['start'] = self._io.pos() - self.hour = self._io.read_u2le() - self._debug['hour']['end'] = self._io.pos() - self._debug['min']['start'] = self._io.pos() - self.min = self._io.read_u2le() - self._debug['min']['end'] = self._io.pos() - self._debug['sec']['start'] = self._io.pos() - self.sec = self._io.read_u2le() - self._debug['sec']['end'] = self._io.pos() - self._debug['msec']['start'] = self._io.pos() - self.msec = self._io.read_u2le() - self._debug['msec']['end'] = self._io.pos() - - |
