diff --git a/pcapkit/toolkit/dpkt.py b/pcapkit/toolkit/dpkt.py index a0e9a8d61e..25c5a191e7 100644 --- a/pcapkit/toolkit/dpkt.py +++ b/pcapkit/toolkit/dpkt.py @@ -582,6 +582,70 @@ def wrapper(packet: 'Packet') -> 'dict[str, Any]': } +def _first_ipv4(packet: 'Packet') -> 'tuple[Optional[IP], Optional[int]]': + """Fetch the first IPv4 layer of a DPKT packet, as the default engine dissects it. + + Args: + packet: DPKT packet, i.e. the outermost layer. + + Returns: + The network layer (see :func:`_network`) if it is IPv4, else the first + IPv4 packet its IPv6 packet carries (:rfc:`2473`) -- directly, behind + extension headers, or inside more IPv6. Beside it, :data:`None` for the + former, and for the latter how many of its octets, from its first on, + lie inside the Payload Length of every IPv6 packet carrying it. Both are + :data:`None` if there is no such IPv4 packet, or the default engine + dissects nothing past an IPv6 packet on the way: a later fragment + (:rfc:`8200#section-4.5`), one with more than + :data:`~pcapkit.protocols.internet.internet.EXTENSION_HEADER_LIMIT` + extension headers (:issue:`1604`), or one whose Payload Length is zero + with no Jumbo Payload option to stand in for it (:rfc:`2675`). + + Note: + The default engine takes ``frame['IPv4']``, the first IPv4 layer however + deep (:func:`pcapkit.toolkit.pcap.ipv4_reassembly`); TCP is keyed by the + innermost instead (see :func:`_tcp_carrier`). It hands that layer only + the octets inside each Payload Length above it, so they end there, short + of any trailer (:issue:`1615`). `DPKT`_ decodes the IPv4 packet in all + three cases above -- a later fragment unless its Fragment header comes + first, any number of extension headers, and a zero Payload Length as the + rest of the frame -- so each is checked here. + + .. _DPKT: https://dpkt.readthedocs.io + + """ + import dpkt # isort:skip + + from pcapkit.protocols.internet.internet import EXTENSION_HEADER_LIMIT # isort:skip + from pcapkit.protocols.schema.internet.ipv6 import jumbo_payload_length # isort:skip + + ip = _network(packet) + extent = None # type: Optional[int] # octets of ``ip`` the packets above hold + while isinstance(ip, dpkt.ip6.IP6): + ext_hdrs = _ipv6_ext_hdrs(ip) + if len(ext_hdrs) > EXTENSION_HEADER_LIMIT or any( + isinstance(ext_hdr, dpkt.ip6.IP6FragmentHeader) and ext_hdr.frag_off for ext_hdr in ext_hdrs + ): + return None, None + if getattr(ip, 'p', None) not in IP_TUNNELS or not isinstance(ip.data, (dpkt.ip.IP, dpkt.ip6.IP6)): + return None, None + + plen = ip.plen + if not plen: + plen = jumbo_payload_length(ip.nxt, _layer2bytes(packet, ip)[0][ip.__hdr_len__:]) + if plen is None: + return None, None + end = ip.__hdr_len__ + plen + if extent is not None: + end = min(end, extent) + extent = max(end - _wire_hdr_len(ip, ip.data), 0) + ip = ip.data + + if not isinstance(ip, dpkt.ip.IP): + return None, None + return ip, extent + + def ipv4_reassembly(packet: 'Packet', timestamp: 'float', *, count: 'int' = -1) -> 'IP_Packet[IPv4Address] | None': """Make data for IPv4 reassembly. @@ -602,9 +666,11 @@ def ipv4_reassembly(packet: 'Packet', timestamp: 'float', *, Data for IPv4 reassembly. * If the ``packet`` can be used for IPv4 reassembly. A packet can be reassembled - if it contains IPv4 layer (:class:`dpkt.ip.IP`) whose header the default engine - accepts (see :func:`_ipv4_accepted`), and the **DF** (:attr:`dpkt.ip.IP.df`) - flag is :data:`False`. + if it contains IPv4 layer (:class:`dpkt.ip.IP`) -- the first of the frame, as + for the default engine, so one carried in IPv6 counts (see + :func:`_first_ipv4`) -- whose header the default engine accepts (see + :func:`_ipv4_accepted`), and the **DF** (:attr:`dpkt.ip.IP.df`) flag is + :data:`False`. * If the ``packet`` can be reassembled, then the :obj:`dict` mapping of data for IPv4 reassembly (:term:`reasm.ipv4.packet`) will be returned; otherwise, returns :data:`None`. @@ -617,13 +683,17 @@ def ipv4_reassembly(packet: 'Packet', timestamp: 'float', *, :func:`attach_buffer` is what makes them the wire's again. The packet itself is left untouched either way. + The header is the IHL's octets and the payload the rest, to the end of + the frame or of the IPv6 packets carrying it, as the default engine + hands them over; the reassembler bounds the payload by the Total Length + itself. Neither is cut at the Total Length, which may be shorter than + the header (:issue:`1616`). + See Also: :class:`pcapkit.foundation.reassembly.ipv4.IPv4` """ - network = _network(packet) - # an IPv6 layer is the one with extension headers, as in ``_wire_hdr_len`` - ipv4 = None if hasattr(network, 'extension_hdrs') else network + ipv4, extent = _first_ipv4(packet) if ipv4 is not None and _ipv4_accepted(ipv4): if ipv4.df: # dismiss not fragmented packet return None @@ -631,11 +701,13 @@ def ipv4_reassembly(packet: 'Packet', timestamp: 'float', *, # covers any IP options, whereas ``IP.__hdr_len__`` is the fixed 20-octet # struct size and would leave option octets at the head of the payload ihl = ipv4.hl * 4 - wire, captured = _layer2bytes(packet, ipv4) - # a zero Total Length is what TCP segmentation offload leaves behind, in - # which case the datagram runs to the end of the frame, as with DPKT - if captured and ipv4.len: - wire = wire[:ipv4.len] + wire, _ = _layer2bytes(packet, ipv4) + if extent is not None: + wire = wire[:extent] + # a Jumbo Payload Length may leave less than DPKT decoded from the + # rest of the frame, and the default engine then dissects no IPv4 + if len(wire) < ihl: + return None data = IP_Packet( bufid=( diff --git a/tests/toolkit/test_dpkt_ipv4_reassembly_input_1615_1616_unit.py b/tests/toolkit/test_dpkt_ipv4_reassembly_input_1615_1616_unit.py new file mode 100644 index 0000000000..9808fca249 --- /dev/null +++ b/tests/toolkit/test_dpkt_ipv4_reassembly_input_1615_1616_unit.py @@ -0,0 +1,257 @@ +# -*- coding: utf-8 -*- +"""dpkt hands IPv4 reassembly what the default engine hands it. C.f. #1615, #1616. + +The default engine hands IPv4 reassembly the first IPv4 layer of a frame however +deep, so IPv4 carried in IPv6 (RFC 2473) is an input, with the octets inside the +Payload Length of each IPv6 packet above it. dpkt took the network layer only, +so gave no input for 4in6 (#1615). It also cut the input at the Total Length, +which the reassembler bounds the payload by itself: a Total Length shorter than +the header cut the header, and a padded last fragment lost its padding (#1616). +The default engine's header is the IHL's octets and its payload the rest. + +Every capture is built here, so the module is in the unit tier. + +""" +from __future__ import annotations + +import importlib.util +import os +import struct +import tempfile +import unittest +import warnings +from typing import TYPE_CHECKING +from unittest import mock + +from tests._support import close_extractor, reimport_once_per_class +from tests.foundation import _roundtrip as wire + +if TYPE_CHECKING: + from typing import Any + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) +HAS_DPKT = importlib.util.find_spec('dpkt') is not None + + +def ipv4(payload: bytes, *, ident: int, options: bytes = b'', length: 'int | None' = None, + mf: bool = False, offset: int = 0, proto: int = 253) -> bytes: + """An IPv4 fragment carrying ``payload``; ``offset`` is in octets, and the Total Length as given.""" + if length is None: + length = 20 + len(options) + len(payload) + flags = (0x2000 if mf else 0) | (offset // 8) + header = bytearray(struct.pack('!BBHHHBBH', 0x45 + len(options) // 4, 0, length, ident, flags, 64, proto, 0) + + wire.IPV4_A + wire.IPV4_B + options) + header[10:12] = wire.checksum(bytes(header)).to_bytes(2, 'big') + return bytes(header) + payload + + +def ipv6(payload: bytes, nxt: int, *, plen: 'int | None' = None) -> bytes: + """An IPv6 packet carrying ``payload``, its Payload Length as given.""" + packet = wire.ipv6(payload, nxt=nxt) + if plen is None: + return packet + return packet[:4] + plen.to_bytes(2, 'big') + packet[6:] + + +def options(nxt: int) -> bytes: + """An 8-octet Hop-by-Hop or Destination Options header, one PadN option in it.""" + return bytes([nxt, 0, 1, 4, 0, 0, 0, 0]) + + +def fragment(nxt: int, offset: int, mf: bool) -> bytes: + """An IPv6 Fragment header; ``offset`` is in octets.""" + return bytes([nxt, 0]) + (offset | mf).to_bytes(2, 'big') + (7).to_bytes(4, 'big') + + +def jumbo(nxt: int, length: int) -> bytes: + """A Hop-by-Hop Options header carrying a Jumbo Payload option (RFC 2675).""" + return bytes([nxt, 0, 0xC2, 4]) + length.to_bytes(4, 'big') + + +def ethernet(packet: bytes, *, pad_to: int = 0, trailer: bytes = b'') -> bytes: + """An Ethernet frame carrying ``packet``, padded to ``pad_to`` octets, then ``trailer``.""" + frame = wire.ethernet(packet, 0x86DD if packet[0] >> 4 == 6 else 0x0800) + return frame + bytes(max(pad_to - len(frame), 0)) + trailer + + +def cases() -> 'dict[str, list[bytes]]': + """Frames by case, each case's IPv4 datagram with an Identification of its own. + + Each is the first fragment of a 24-octet datagram, then for a pair its last. + + """ + built = {} # type: dict[str, list[bytes]] + + def pair(label: str, carry: 'Any', ident: int) -> None: + first = ipv4(b'A' * 16, ident=ident, mf=True) + last = ipv4(b'B' * 8, ident=ident, offset=16) + built[label] = [carry(first), carry(last)] + + def one(label: str, carry: 'Any', ident: int) -> None: + built[label] = [carry(ipv4(b'A' * 16, ident=ident, mf=True))] + + # #1615: the first IPv4 layer, inside IPv6, behind extension headers, or in 6in6 + pair('4in6', lambda p: ethernet(ipv6(p, 4)), 1) + pair('4in6 behind Destination Options', lambda p: ethernet(ipv6(options(4) + p, 60)), 2) + pair('4in6 in 6in6', lambda p: ethernet(ipv6(ipv6(p, 4), 41)), 3) + one('4in6 in a first IPv6 fragment', lambda p: ethernet(ipv6(fragment(4, 0, True) + p, 44)), 4) + one('4in6 behind 32 extension headers', lambda p: ethernet(ipv6(options(60) * 31 + options(4) + p, 60)), 5) + # whose octets end at each Payload Length above, short of a trailer + pair('4in6 with a trailer', lambda p: ethernet(ipv6(p, 4), trailer=b'\xee' * 6), 6) + one('4in6 cut by the Payload Length', lambda p: ethernet(ipv6(p, 4, plen=30)), 7) + one('4in6 cut by the outer Payload Length of 6in6', + lambda p: ethernet(ipv6(ipv6(p, 4, plen=200), 41, plen=70)), 8) + one('4in6 jumbogram', lambda p: ethernet(ipv6(jumbo(4, 8 + 30) + p, 0, plen=0), trailer=b'\xee' * 4), 9) + # where the default engine dissects no IPv4 inside the IPv6 packet + one('4in6 in a later IPv6 fragment', lambda p: ethernet(ipv6(options(44) + fragment(4, 8, False) + p, 0)), 10) + one('4in6 behind 33 extension headers', lambda p: ethernet(ipv6(options(60) * 32 + options(4) + p, 60)), 11) + one('4in6 with a zero Payload Length', lambda p: ethernet(ipv6(p, 4, plen=0)), 12) + one('4in6 jumbogram shorter than the IPv4 header', lambda p: ethernet(ipv6(jumbo(4, 8 + 10) + p, 0, plen=0)), 13) + # #1616: the header is the IHL's octets and the payload the rest of the frame + built['Total Length below the IHL'] = [ethernet(ipv4(b'C' * 16, ident=14, options=b'\x94\x04\x00\x00', + length=10, mf=True))] + pair('padded to the Ethernet minimum', lambda p: ethernet(p, pad_to=60), 15) + one('Total Length 0 in 4in6, with a trailer', + lambda p: ethernet(ipv6(p[:2] + b'\0\0' + p[4:], 4), trailer=b'\xee' * 4), 16) + # controls: plain IPv4, the outer IPv4 of 4in4 and of 4in6in4, and no IPv4 at all + pair('plain', ethernet, 17) + one('4in4', lambda p: ethernet(ipv4(p, ident=18, proto=4)), 19) + one('4in6in4', lambda p: ethernet(ipv4(ipv6(p, 4), ident=20, proto=41)), 21) + built['IPv6 carrying UDP'] = [ethernet(ipv6(wire.udp(b'data'), 17))] + built['ARP'] = [wire.ethernet(bytes.fromhex('0001080006040001') + wire.MAC_A + wire.IPV4_A + + bytes(6) + wire.IPV4_B, 0x0806)] + return built + + +def read(test: unittest.TestCase, engine: str, frames: 'list[bytes]') -> 'dict[str, Any]': + """Read ``frames`` with ``engine``, with IPv4 reassembly. + + Returns: + Every input the engine hands IPv4 reassembly, every field of it, keyed + by frame number, and the reassembled datagrams. + + """ + from pcapkit import extract + from pcapkit.foundation.reassembly.ipv4 import IPv4 + + seen = {} # type: dict[int, dict[str, Any]] + reassemble = IPv4.__call__ + + def spy(this: 'Any', packet: 'Any') -> 'Any': + seen[packet.num] = {key: bytes(value) if isinstance(value, bytearray) else value + for key, value in packet.items()} + return reassemble(this, packet) + + with tempfile.TemporaryDirectory() as tmp, warnings.catch_warnings(), \ + mock.patch.object(IPv4, '__call__', spy): + path = os.path.join(tmp, 'ipv4.pcap') + with open(path, 'wb') as file: + file.write(wire.pcap([(1, number, octets) for number, octets in enumerate(frames)])) + warnings.simplefilter('ignore') + extractor = extract(fin=path, nofile=True, engine=engine, reassembly=True, ipv4=True) + try: + if extractor._exnam != engine: # pylint: disable=protected-access + test.skipTest(f'{engine} did not run') + datagrams = [(datagram.completed, datagram.id, datagram.index, datagram.header, + datagram.payload) for datagram in extractor.reassembly.ipv4] + finally: + close_extractor(extractor) + return {'inputs': seen, 'datagrams': datagrams} + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +@unittest.skipUnless(HAS_DPKT, 'dpkt not installed') +class DPKTIPv4ReassemblyInputTests(unittest.TestCase): + """dpkt's IPv4 reassembly inputs are the default engine's, octet for octet.""" + + @classmethod + def setUpClass(cls) -> None: + cls.cases = cases() + cls.frames = [frame for frames in cls.cases.values() for frame in frames] + # one-based frame numbers of each case + cls.numbers = {} # type: dict[str, list[int]] + number = 0 + for label, frames in cls.cases.items(): + cls.numbers[label] = list(range(number + 1, number + 1 + len(frames))) + number += len(frames) + + def setUp(self) -> None: + reimport_once_per_class(self) + + def read(self) -> 'tuple[dict[str, Any], dict[str, Any]]': + """The default engine's reading of every case, and dpkt's.""" + if not hasattr(type(self), 'seen'): + type(self).seen = (read(self, 'default', self.frames), read(self, 'dpkt', self.frames)) + return type(self).seen + + def inputs(self, seen: 'dict[str, Any]', label: str) -> 'list[dict[str, Any]]': + return [seen['inputs'][number] for number in self.numbers[label] if number in seen['inputs']] + + def test_every_input_agrees_with_the_default_engine(self) -> None: + base, dpkt = self.read() + for label in self.cases: + with self.subTest(case=label): + self.assertEqual(self.inputs(dpkt, label), self.inputs(base, label)) + self.assertEqual(dpkt['datagrams'], base['datagrams']) + + def test_ipv4_in_ipv6_is_an_input(self) -> None: + # #1615: two inputs of the datagram tunnelled in IPv6, which completes + base, dpkt = self.read() + for seen in (base, dpkt): + got = self.inputs(seen, '4in6') + self.assertEqual([(len(packet['header']), len(packet['payload'])) for packet in got], + [(20, 16), (20, 8)]) + self.assertIn((self.numbers['4in6'][0], self.numbers['4in6'][1]), + [datagram[2] for datagram in seen['datagrams'] if datagram[0]]) + + def test_each_payload_length_above_bounds_the_input(self) -> None: + base, dpkt = self.read() + for label, payloads in ( + ('4in6 with a trailer', [16, 8]), + ('4in6 cut by the Payload Length', [10]), + ('4in6 cut by the outer Payload Length of 6in6', [10]), + ('4in6 jumbogram', [10]), + ('Total Length 0 in 4in6, with a trailer', [16]), + ): + for seen in (base, dpkt): + with self.subTest(case=label): + self.assertEqual([len(packet['payload']) for packet in self.inputs(seen, label)], payloads) + + def test_no_input_where_the_default_engine_dissects_no_ipv4(self) -> None: + base, dpkt = self.read() + self.assertEqual(len(self.inputs(base, '4in6 behind 32 extension headers')), 1) + for label in ('4in6 in a later IPv6 fragment', '4in6 behind 33 extension headers', + '4in6 with a zero Payload Length', '4in6 jumbogram shorter than the IPv4 header', + 'IPv6 carrying UDP', 'ARP'): + for seen in (base, dpkt): + with self.subTest(case=label): + self.assertEqual(self.inputs(seen, label), []) + + def test_input_is_not_cut_at_the_total_length(self) -> None: + # #1616: the IHL's octets, then the rest of the frame, padding included + base, dpkt = self.read() + for seen in (base, dpkt): + short, = self.inputs(seen, 'Total Length below the IHL') + self.assertEqual((short['ihl'], short['tl']), (24, 10)) + self.assertEqual((len(short['header']), bytes(short['payload'])), (24, b'C' * 16)) + padded = self.inputs(seen, 'padded to the Ethernet minimum') + self.assertEqual([len(packet['payload']) for packet in padded], [26, 26]) + self.assertEqual(bytes(padded[1]['payload']), b'B' * 8 + bytes(18)) + + def test_a_packet_built_by_hand_is_an_input_too(self) -> None: + # with no record buffer attached, the octets are dpkt's own serialisation + import dpkt + + from pcapkit.toolkit import dpkt as toolkit + + for label, payload in (('4in6', b'A' * 16), ('4in6 cut by the Payload Length', b'A' * 10)): + with self.subTest(case=label): + data = toolkit.ipv4_reassembly(dpkt.ethernet.Ethernet(self.cases[label][0]), 1.0, count=1) + self.assertIsNotNone(data) + self.assertEqual((bytes(data.header), bytes(data.payload)), + (self.cases[label][0][54:74], payload)) + + +if __name__ == '__main__': + unittest.main()