fix(ipv4,ipv6)!: keep the octets past Total Length or Payload Length as captured (#1209)

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
 

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.

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':

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 <pcapkit.protocols.internet.ipv4.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]':

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': ...

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 <pcapkit.protocols.schema.internet.ipv4.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: ...

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()