diff --git a/pcapkit/protocols/data/internet/ipv4.py b/pcapkit/protocols/data/internet/ipv4.py index 715e0cdbf8..f05b5d4524 100644 --- a/pcapkit/protocols/data/internet/ipv4.py +++ b/pcapkit/protocols/data/internet/ipv4.py @@ -124,6 +124,8 @@ class IPv4(Protocol): options: 'OrderedMultiDict[OptionNumber, Option]' #: Octets after the End of Option List, present along with ``options``. padding: 'bytes' + #: Octets captured past the Total Length, present only when there are any. + trailer: 'bytes' def __init__(self, version: 'Literal[4]', hdr_len: 'int', tos: 'ToSField', len: 'int', id: 'int', flags: 'Flags', offset: 'int', ttl: 'timedelta', protocol: 'TransType', checksum: 'bytes', src: 'IPv4Address', dst: 'IPv4Address') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin diff --git a/pcapkit/protocols/data/internet/ipv6.py b/pcapkit/protocols/data/internet/ipv6.py index a9d78d1268..93b2504c48 100644 --- a/pcapkit/protocols/data/internet/ipv6.py +++ b/pcapkit/protocols/data/internet/ipv6.py @@ -60,6 +60,9 @@ class IPv6(Protocol): hdr_len: 'int' #: Raw payload length (excluding extension headers). raw_len: 'int' + #: Octets captured past the Payload Length, present only when there + #: are any. + trailer: 'bytes' #: Extension header chain, in wire order, as pairs of the parser #: class and its info. Kept apart from the per-header keys above, #: which a repeated header overwrites. diff --git a/pcapkit/protocols/internet/ipv4.py b/pcapkit/protocols/internet/ipv4.py index 0aec17fb53..3464a3cab5 100644 --- a/pcapkit/protocols/internet/ipv4.py +++ b/pcapkit/protocols/internet/ipv4.py @@ -338,6 +338,15 @@ def read(self, length: 'Optional[int]' = None, *, # pylint: disable=unused-argu ('padding', self._data[20 + _used:ipv4.hdr_len]), ]) + # NOTE: Octets captured past the Total Length belong to this layer, + # since its own length field is what leaves them out of the payload: an + # Ethernet carrier hands the whole rest of its frame here, padding + # included, and the rebuild keeps them as captured (:issue:`1209`). + if schema.trailer: + ipv4.__update__([ + ('trailer', schema.trailer), + ]) + # update packet info if __packet__ is None: __packet__ = {} @@ -395,6 +404,7 @@ def make(self, total_length: 'Optional[int]' = None, padding: 'Optional[bytes]' = None, payload: 'bytes | ProtocolBase | Schema' = b'', + trailer: 'bytes' = b'', **kwargs: 'Any') -> 'Schema_IPv4': """Make (construct) packet data. @@ -439,6 +449,8 @@ def make(self, ``options`` is a container that carries an ``EOOL``, c.f. :meth:`_make_ipv4_options`. payload: Payload of the packet. + trailer: Octets after the packet, outside the Total Length, such as + link-layer padding; written as is. **kwargs: Arbitrary keyword arguments. Returns: @@ -502,6 +514,7 @@ def make(self, dst=dst, options=options_value, payload=payload, + trailer=trailer, ) @classmethod @@ -605,6 +618,8 @@ def _make_data(cls, data: 'Data_IPv4') -> 'dict[str, Any]': # type: ignore[over # options and these octets, so edited options recompute it. 'padding': getattr(data, 'padding', None), 'payload': cls._make_payload(data), + # NOTE: And the octets captured past the Total Length (:issue:`1209`). + 'trailer': data.get('trailer', b''), } def _read_ipv4_addr(self) -> 'IPv4Address': diff --git a/pcapkit/protocols/internet/ipv6.py b/pcapkit/protocols/internet/ipv6.py index 24fb58dad2..3717012e4b 100644 --- a/pcapkit/protocols/internet/ipv6.py +++ b/pcapkit/protocols/internet/ipv6.py @@ -207,6 +207,14 @@ def read(self, length: 'Optional[int]' = None, *, 'dst': schema.dst, }) # type: Data_IPv6 + # NOTE: Octets captured past the Payload Length are kept as captured, + # c.f. :meth:`IPv4.read ` + # (:issue:`1209`). + if schema.trailer: + ipv6.__update__({ + 'trailer': schema.trailer, + }) + # update packet info if __packet__ is None: __packet__ = {} @@ -229,6 +237,7 @@ def make(self, dst: 'IPv6Address | str | bytes | int' = '::', payload_length: 'Optional[int]' = None, payload: 'bytes | ProtocolBase | Schema' = b'', + trailer: 'bytes' = b'', **kwargs: 'Any') -> 'Schema_IPv6': """Make (construct) packet data. @@ -245,6 +254,8 @@ def make(self, payload_length: Length of the payload, extension headers included; computed from the payload when omitted. payload: Payload data. + trailer: Octets after the packet, outside the Payload Length, such + as link-layer padding; written as is. **kwargs: Arbitrary keyword arguments. Returns: @@ -274,6 +285,7 @@ def make(self, src=src, dst=dst, payload=payload, + trailer=trailer, ) @classmethod @@ -343,6 +355,8 @@ def _make_data(cls, data: 'Data_IPv6') -> 'dict[str, Any]': # type: ignore[over # (:issue:`1155`). 'payload_length': data.payload, 'payload': payload, + # NOTE: And the octets captured past the Payload Length (:issue:`1209`). + 'trailer': data.get('trailer', b''), } def _read_ip_hextet(self) -> 'tuple[int, int, int]': diff --git a/pcapkit/protocols/schema/internet/ipv4.py b/pcapkit/protocols/schema/internet/ipv4.py index 2ac7b5eb14..5c1a4b3df3 100644 --- a/pcapkit/protocols/schema/internet/ipv4.py +++ b/pcapkit/protocols/schema/internet/ipv4.py @@ -734,9 +734,15 @@ class IPv4(Schema): padding: 'bytes' = PaddingField(length=lambda pkt: pkt.get('__option_padding__', 0)) #: Payload. payload: 'bytes' = PayloadField(length=lambda pkt: pkt['length'] - pkt['vihl']['ihl'] * 4) + #: Octets captured past the Total Length, such as Ethernet minimum-frame + #: padding. A :class:`~pcapkit.corekit.fields.misc.PayloadField` with no + #: length takes whatever the payload leaves on unpack and writes its value + #: verbatim on pack, so the octets survive whatever they hold. + trailer: 'bytes' = PayloadField(default=b'') if TYPE_CHECKING: def __init__(self, vihl: 'VerIHLField', tos: 'ToSField', length: 'int', id: 'int', flags: 'Flags', ttl: 'int', proto: 'Enum_TransType', chksum: 'bytes', src: 'IPv4Address | str | bytes | int', dst: 'IPv4Address | str | bytes | int', - options: 'list[Option | bytes] | bytes', payload: 'bytes | ProtocolBase | Schema') -> 'None': ... + options: 'list[Option | bytes] | bytes', payload: 'bytes | ProtocolBase | Schema', + trailer: 'bytes' = b'') -> 'None': ... diff --git a/pcapkit/protocols/schema/internet/ipv6.py b/pcapkit/protocols/schema/internet/ipv6.py index 507080319d..e54e272514 100644 --- a/pcapkit/protocols/schema/internet/ipv6.py +++ b/pcapkit/protocols/schema/internet/ipv6.py @@ -63,9 +63,13 @@ class IPv6(Schema): dst: 'IPv6Address' = IPv6AddressField() #: Payload. payload: 'bytes' = PayloadField(length=lambda pkt: pkt['length']) + #: Octets captured past the Payload Length, such as Ethernet minimum-frame + #: padding, c.f. :attr:`IPv4.trailer `. + trailer: 'bytes' = PayloadField(default=b'') if TYPE_CHECKING: def __init__(self, hextet: 'IPv6Hextet', length: 'int', next: 'Enum_TransType', limit: 'int', src: 'IPv6Address | bytes | str | int', dst: 'IPv6Address | bytes | str | int', - payload: 'bytes | ProtocolBase | Schema') -> None: ... + payload: 'bytes | ProtocolBase | Schema', + trailer: 'bytes' = b'') -> None: ... diff --git a/tests/protocols/link/test_link_trailer_kept_unit.py b/tests/protocols/link/test_link_trailer_kept_unit.py new file mode 100644 index 0000000000..ac32cceab7 --- /dev/null +++ b/tests/protocols/link/test_link_trailer_kept_unit.py @@ -0,0 +1,136 @@ +# -*- coding: utf-8 -*- +"""Octets captured past an IP packet's own length survive a rebuild. + +GitHub issue #1209: :class:`~pcapkit.protocols.link.ethernet.Ethernet` and the +VLAN tags hand the whole rest of the frame to the next layer, but +:class:`~pcapkit.protocols.internet.ipv4.IPv4` reads only its Total Length and +:class:`~pcapkit.protocols.internet.ipv6.IPv6` only its Payload Length. The +octets after that -- most often the padding of a 60-octet minimum Ethernet +frame -- were kept by no ``info``, so +:meth:`~pcapkit.protocols.protocol.ProtocolBase.from_data` dropped them: the +frame below rebuilt as 42 octets of its 60. + +The IP layer owns them as ``trailer``, since its own length field is what +leaves them out of the payload, and so the IP packet rebuilds byte for byte on +its own as well as inside a frame. + +Every case builds its own octets in memory and reads no capture. Classes are +imported inside each test, after :func:`~tests._support.reimport_once_per_class`, +c.f. ``tests/protocols/link/test_ethernet_mac_roundtrip_unit.py``. + +""" + +import importlib +import unittest + +from tests._support import reimport_once_per_class + +#: Destination and source MAC addresses. +MACS = bytes.fromhex('001122334455' '66778899aabb') +#: An IPv4/UDP packet, Total Length 28. +UDP4 = bytes.fromhex('4500001c000100004011f97d7f0000017f000001' '0035003500080000') +#: The two IPv6 addresses, ``::1`` twice. +ADDRS6 = (bytes(15) + b'\x01') * 2 +#: An IPv6 header, Payload Length 0 and No Next Header. +NONXT6 = bytes.fromhex('60000000' '0000' '3b' '40') + ADDRS6 + +#: Frames whose IP packet is shorter than what the link layer carries. +FRAMES = { + 'ethernet': MACS + b'\x08\x00' + UDP4 + bytes(18), + 'ethernet-junk': MACS + b'\x08\x00' + UDP4 + b'\x01\x02\x03' * 6, + 'c-tag': MACS + bytes.fromhex('8100' '0064' '0800') + UDP4 + bytes(14), + 'q-in-q': MACS + bytes.fromhex('88a8' '00c8' '8100' '0064' '0800') + UDP4 + bytes(10), + 'ethernet-ipv6': MACS + b'\x86\xdd' + NONXT6 + b'\xaa\xbb', +} + +#: Bare IP packets with octets past their own length, and those octets. +PACKETS = { + 'ipv4': ('pcapkit.protocols.internet.ipv4', 'IPv4', + bytes.fromhex('450000180001000040fd0000' '7f000001' '7f000001') + b'abcd' + bytes(6), + bytes(6)), + 'ipv4-options': ('pcapkit.protocols.internet.ipv4', 'IPv4', + bytes.fromhex('4600001c0001000040fd0000' '7f000001' '7f000001' '01010100') + + b'abcd' + b'\xee' * 5, + b'\xee' * 5), + 'ipv4-header-only': ('pcapkit.protocols.internet.ipv4', 'IPv4', + bytes.fromhex('450000140001000040fd0000' '7f000001' '7f000001') + b'wxyz', + b'wxyz'), + 'ipv6': ('pcapkit.protocols.internet.ipv6', 'IPv6', NONXT6 + b'\xaa\xbb', b'\xaa\xbb'), + 'ipv6-hopopt': ('pcapkit.protocols.internet.ipv6', 'IPv6', + bytes.fromhex('60000000' '0010' '00' '40') + ADDRS6 + + bytes.fromhex('1100010400000000' '0035003500080000') + bytes(6), + bytes(6)), +} + +#: Bare IP packets that end at or before their own length. +NO_TRAILER = { + 'ipv4-exact': ('pcapkit.protocols.internet.ipv4', 'IPv4', + bytes.fromhex('450000180001000040fd0000' '7f000001' '7f000001') + b'abcd'), + 'ipv4-truncated': ('pcapkit.protocols.internet.ipv4', 'IPv4', + bytes.fromhex('450000280001000040fd0000' '7f000001' '7f000001') + b'abcdefghij'), + 'ipv6-exact': ('pcapkit.protocols.internet.ipv6', 'IPv6', NONXT6), + 'ipv6-truncated': ('pcapkit.protocols.internet.ipv6', 'IPv6', + bytes.fromhex('60000000' '0010' '3b' '40') + ADDRS6 + b'\xaa' * 4), +} + + +def _klass(module: str, name: str) -> type: + return getattr(importlib.import_module(module), name) + + +class TestLinkTrailerKept(unittest.TestCase): + """Pin the octets past an IP packet's own length through ``from_data``.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_frame_rebuilds_byte_for_byte(self) -> None: + from pcapkit.protocols.link.ethernet import Ethernet + + for name, frame in FRAMES.items(): + with self.subTest(frame=name): + parsed = Ethernet(frame) + self.assertEqual(len(parsed.data), 60 if name != 'ethernet-ipv6' else 56) + self.assertEqual(Ethernet.from_data(parsed.info).data, frame) + + def test_packet_rebuilds_byte_for_byte(self) -> None: + for name, (module, klass, packet, trailer) in PACKETS.items(): + with self.subTest(packet=name): + proto = _klass(module, klass) + parsed = proto(packet) + self.assertEqual(parsed.info.trailer, trailer) + self.assertEqual(proto.from_data(parsed.info).data, packet) + + def test_no_trailer_key_without_excess(self) -> None: + for name, (module, klass, packet) in NO_TRAILER.items(): + with self.subTest(packet=name): + proto = _klass(module, klass) + parsed = proto(packet) + self.assertNotIn('trailer', parsed.info) + self.assertEqual(proto.from_data(parsed.info).data, packet) + + def test_padded_frame_records_padding_on_ipv4(self) -> None: + from pcapkit.protocols.link.ethernet import Ethernet + + info = Ethernet(FRAMES['ethernet']).info + self.assertEqual(info.ipv4.len, 28) + self.assertEqual(info.ipv4.trailer, bytes(18)) + self.assertNotIn('trailer', info) + + def test_make_writes_trailer_outside_the_length(self) -> None: + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.internet.ipv6 import IPv6 + + ipv4 = IPv4(protocol=253, payload=b'abcd', trailer=b'\x00\x01') + self.assertEqual(ipv4.info.len, 24) + self.assertEqual(ipv4.data[-6:], b'abcd\x00\x01') + self.assertEqual(ipv4.info.trailer, b'\x00\x01') + + ipv6 = IPv6(next=59, payload=b'abcd', trailer=b'\x01') + self.assertEqual(ipv6.info.payload, 4) + self.assertEqual(ipv6.data[-5:], b'abcd\x01') + self.assertEqual(ipv6.info.trailer, b'\x01') + + +if __name__ == '__main__': + unittest.main()