diff options
Diffstat (limited to 'python/examples/kaitai/kaitai_struct_formats/network')
19 files changed, 3670 insertions, 0 deletions
diff --git a/python/examples/kaitai/kaitai_struct_formats/network/__init__.py b/python/examples/kaitai/kaitai_struct_formats/network/__init__.py new file mode 100644 index 00000000..e69de29b --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/network/__init__.py diff --git a/python/examples/kaitai/kaitai_struct_formats/network/bitcoin_transaction.py b/python/examples/kaitai/kaitai_struct_formats/network/bitcoin_transaction.py new file mode 100644 index 00000000..41d05158 --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/network/bitcoin_transaction.py @@ -0,0 +1,202 @@ +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 new file mode 100644 index 00000000..c90b66da --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/network/dns_packet.py @@ -0,0 +1,356 @@ +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 new file mode 100644 index 00000000..e6baaccf --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/network/ethernet_frame.py @@ -0,0 +1,66 @@ +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 new file mode 100644 index 00000000..db02ef14 --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/network/hccap.py @@ -0,0 +1,110 @@ +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 new file mode 100644 index 00000000..439db7d4 --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/network/hccapx.py @@ -0,0 +1,101 @@ +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 new file mode 100644 index 00000000..45de7aae --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/network/icmp_packet.py @@ -0,0 +1,134 @@ +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 new file mode 100644 index 00000000..0bc7bb95 --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/network/ipv4_packet.py @@ -0,0 +1,158 @@ +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 new file mode 100644 index 00000000..c108d643 --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/network/ipv6_packet.py @@ -0,0 +1,55 @@ +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 new file mode 100644 index 00000000..7fec8613 --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/network/microsoft_network_monitor_v2.py @@ -0,0 +1,307 @@ +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 new file mode 100644 index 00000000..b213460a --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/network/packet_ppi.py @@ -0,0 +1,514 @@ +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 new file mode 100644 index 00000000..609a278d --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/network/pcap.py @@ -0,0 +1,235 @@ +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 new file mode 100644 index 00000000..12d17565 --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/network/protocol_body.py @@ -0,0 +1,260 @@ +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 new file mode 100644 index 00000000..7450b3e7 --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/network/rtcp_payload.py @@ -0,0 +1,579 @@ +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 new file mode 100644 index 00000000..64e50cba --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/network/rtp_packet.py @@ -0,0 +1,150 @@ +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 new file mode 100644 index 00000000..222f6794 --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/network/tcp_segment.py @@ -0,0 +1,57 @@ +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 new file mode 100644 index 00000000..118eec3d --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/network/tls_client_hello.py @@ -0,0 +1,293 @@ +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 new file mode 100644 index 00000000..d3cac12f --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/network/udp_datagram.py @@ -0,0 +1,42 @@ +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 new file mode 100644 index 00000000..2f3bc5df --- /dev/null +++ b/python/examples/kaitai/kaitai_struct_formats/network/windows_systemtime.py @@ -0,0 +1,51 @@ +# 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() + + |