fix(protocols): a payload layer parsed from a short stream rebuilds default-filled octets

pcapkit/protocols/internet/internet.py

@@ -265,14 +265,13 @@ def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *,  #
         if length is None:
             length = len(file_)
 
-        if length == 0:
+        if min(length, len(file_)) == 0:
             from pcapkit.protocols.misc.null import NoPayload as protocol  # isort: skip # pylint: disable=import-outside-toplevel
         elif self._sigterm:
             from pcapkit.protocols.misc.raw import Raw as protocol  # isort: skip # pylint: disable=import-outside-toplevel
         else:
             protocol = self._lookup_next_layer(self.__proto__, proto)
 
-        next_ = protocol(file_, length, version=version, extension=extension,  # type: ignore[abstract]
-                         alias=proto, packet=packet, layer=self._exlayer, protocol=self._exproto,
-                         __context__=self._exctx)
-        return next_
+        return self._parse_next_layer(protocol, file_, length, version=version, extension=extension,
+                                      alias=proto, packet=packet, layer=self._exlayer,
+                                      protocol=self._exproto, __context__=self._exctx)

pcapkit/protocols/internet/ipv6.py

@@ -520,7 +520,7 @@ def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *,  #
         if length is None:
             length = len(file_)
 
-        if length == 0:
+        if min(length, len(file_)) == 0:
             from pcapkit.protocols.misc.null import \
                 NoPayload as protocol  # isort: skip # pylint: disable=import-outside-toplevel
         elif self._sigterm:
@@ -530,9 +530,9 @@ def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *,  #
             protocol = self._lookup_next_layer(self.__proto__, proto)
 
         try:
-            next_ = protocol(file_, length, version=version, extension=extension,  # type: ignore[abstract]
-                             alias=proto, packet=packet, layer=self._exlayer, protocol=self._exproto,
-                             __context__=self._exctx)
+            next_ = self._parse_next_layer(protocol, file_, length, version=version, extension=extension,
+                                           alias=proto, packet=packet, layer=self._exlayer,
+                                           protocol=self._exproto, __context__=self._exctx)
         except Exception as exc:
             from pcapkit.protocols.internet.ipv6_ext import \
                 IPv6_Ext  # isort: skip # pylint: disable=import-outside-toplevel

pcapkit/protocols/protocol.py

@@ -1820,16 +1820,59 @@ def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *,
         if length is None:
             length = len(file_)
 
-        if length == 0:
+        if min(length, len(file_)) == 0:
             from pcapkit.protocols.misc.null import NoPayload as protocol  # isort: skip # pylint: disable=import-outside-toplevel
         elif self._sigterm:
             from pcapkit.protocols.misc.raw import Raw as protocol  # isort: skip # pylint: disable=import-outside-toplevel
         else:
             protocol = self._lookup_next_layer(self.__proto__, proto)
 
-        next_ = protocol(file_, length, alias=proto, packet=packet,
-                         layer=self._exlayer, protocol=self._exproto,
-                         __context__=self._exctx)  # type: ignore[abstract]
+        return self._parse_next_layer(protocol, file_, length, alias=proto, packet=packet,
+                                      layer=self._exlayer, protocol=self._exproto,
+                                      __context__=self._exctx)
+
+    @staticmethod
+    def _parse_next_layer(klass: 'Type[ProtocolBase]', file_: 'bytes', length: 'int',
+                          **kwargs: 'Any') -> 'ProtocolBase':
+        r"""Parse the next layer, or keep it as raw octets if its header was not captured.
+
+        Arguments:
+            klass: next layer protocol class
+            file\_: payload of the current layer
+            length: valid (*non-padding*) length
+            **kwargs: keyword arguments for the next layer's constructor
+
+        Returns:
+            Instance of ``klass``, or of :class:`~pcapkit.protocols.misc.raw.Raw`
+            when fewer octets were captured than its header needs.
+
+        Note:
+            :meth:`Schema.unpack <pcapkit.protocols.schema.schema.Schema.unpack>`
+            zero-fills a field the stream ends inside, so a header parsed from
+            too few octets is filled with values that were never on the wire, and
+            :meth:`from_data <ProtocolBase.from_data>` rebuilds them (:issue:`1170`).
+            Such a payload is kept as :class:`~pcapkit.protocols.misc.raw.Raw`
+            instead, as an unparseable one is.
+
+            The captured octets are what ``file_`` holds, which may be fewer than
+            the ``length`` the current header declares. They are short of the
+            header when there are fewer than the next layer's
+            :meth:`__length_hint__ <ProtocolBase.__length_hint__>`, unless the
+            header it parsed, :attr:`length <ProtocolBase.length>` octets long, fits
+            in them anyway: for a variable-length header such as
+            :class:`~pcapkit.protocols.link.l2tpv2.L2TPv2` the hint is the longest
+            form rather than the shortest. A next layer that reports no hint is
+            kept as parsed.
+
+        """
+        next_ = klass(file_, length, **kwargs)  # type: ignore[abstract]
+
+        captured = min(length, len(file_))
+        length_hint = getattr(next_, '__length_hint__', None)
+        hint = None if length_hint is None else length_hint()
+        if hint is not None and captured < hint and captured < next_.length:
+            from pcapkit.protocols.misc.raw import Raw  # isort: skip # pylint: disable=import-outside-toplevel
+            next_ = Raw(file_, length, **kwargs)
         return next_
 
     def _check_term_threshold(self) -> bool:

tests/protocols/internet/test_ipv4_header_truncated_payload_runtime.py

@@ -3,11 +3,12 @@
 
 GitHub issue #1167. :meth:`IPv4.read <pcapkit.protocols.internet.ipv4.IPv4.read>`
 handed the next layer the length the header declared even when the capture held
-none of its octets. ``test.pcapng`` frame 3 is cut inside the IPv4 header (18 of
-20 octets), so its UDP layer was parsed from an empty stream into an all-zero
-header, and :meth:`~pcapkit.protocols.protocol.ProtocolBase.from_data` rebuilt
-those 8 octets as if they had been on the wire. Such a payload is now a
-:class:`~pcapkit.protocols.misc.null.NoPayload`.
+none of its octets, so its UDP layer was parsed from an empty stream into an
+all-zero header, and :meth:`~pcapkit.protocols.protocol.ProtocolBase.from_data`
+rebuilt those 8 octets as if they had been on the wire. Such a payload is now a
+:class:`~pcapkit.protocols.misc.null.NoPayload`. ``test.pcapng`` frame 3, cut
+inside the IPv4 header itself (18 of 20 octets), is kept as raw octets
+(:issue:`1170`).
 
 The capture case reads a generated sample, so this module belongs to the
 fixture-dependent tier. The in-memory cases need no capture.
@@ -36,24 +37,20 @@ def setUp(self) -> None:
 
     def test_capture_cut_inside_the_header(self) -> None:
         from pcapkit.interface import extract
-        from pcapkit.protocols.internet.ipv4 import IPv4
-        from pcapkit.protocols.misc.null import NoPayload
+        from pcapkit.protocols.link.ethernet import Ethernet
+        from pcapkit.protocols.misc.raw import Raw
 
         extractor = extract(fin=sample_path('test.pcapng'), fout='/tmp/out', format='tree',
                             store=True, nofile=True)
         self.addCleanup(close_extractor, extractor)
-        parsed = list(extractor.frame)[2][IPv4]
-
-        self.assertEqual(parsed.data, bytes.fromhex('4500012ca8370000fa11178a00000000ffff'))
-        self.assertIsInstance(parsed.payload, NoPayload)
-        self.assertEqual(str(parsed.protochain), 'IPv4')
-        self.assertNotIn('udp', parsed.info)
-
-        # The rebuild is the header alone: the 18 captured octets, then the two
-        # the parse already filled into ``dst``, and no UDP header after them.
-        rebuilt = IPv4.from_data(parsed.info)
-        self.assertEqual(rebuilt.data, parsed.data + b'\x00\x00')
-        self.assertEqual(rebuilt.info.len, 300)
+        parsed = list(extractor.frame)[2][Ethernet]
+
+        # 18 octets are short of the 20-octet IPv4 header, so they are kept as
+        # raw octets (:issue:`1170`) rather than parsed into a ``dst`` whose low
+        # half is zero-filled, and the rebuild is exactly what was captured.
+        self.assertIsInstance(parsed.payload, Raw)
+        self.assertEqual(parsed.payload.data, bytes.fromhex('4500012ca8370000fa11178a00000000ffff'))
+        self.assertEqual(Ethernet.from_data(parsed.info).data, parsed.data)
 
     def test_header_without_payload_octets(self) -> None:
         from pcapkit.protocols.internet.ipv4 import IPv4

tests/protocols/test_declared_length_roundtrip_runtime.py

@@ -98,20 +98,22 @@ def test_complete_frames_still_round_trip(self) -> None:
     def test_header_truncated_frame_keeps_its_total_length(self) -> None:
         """``test.pcapng`` frame 3 is cut inside the IPv4 header itself.
 
-        Only 18 of the 20 header octets were recorded, so the rebuild cannot be
-        byte for byte -- it writes a whole header. What it must not do is replace
-        the declared 300 with the 18 that were kept.
+        Only 18 of the 20 header octets were recorded, so they are kept as raw
+        octets rather than parsed into a header (:issue:`1170`). The rebuild is
+        then byte for byte, and keeps the declared 300 rather than the 18 that
+        were kept.
 
         """
-        from pcapkit.protocols.internet.ipv4 import IPv4
+        from pcapkit.protocols.link.ethernet import Ethernet
+        from pcapkit.protocols.misc.raw import Raw
 
-        parsed = self._frames('test.pcapng')[2][IPv4]
-        self.assertEqual(parsed.info.len, 300)
-        self.assertEqual(len(parsed.data), 18)
+        parsed = self._frames('test.pcapng')[2][Ethernet]
+        self.assertIsInstance(parsed.payload, Raw)
+        self.assertEqual(len(parsed.payload.data), 18)
+        self.assertEqual(parsed.payload.data[2:4], (300).to_bytes(2, 'big'))
 
-        rebuilt = IPv4.from_data(parsed.info)
-        self.assertEqual(rebuilt.info.len, 300)
-        self.assertEqual(rebuilt.data[:18], parsed.data)
+        rebuilt = Ethernet.from_data(parsed.info)
+        self.assertEqual(rebuilt.data, parsed.data)
 
 
 @unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed')

tests/protocols/test_next_layer_short_header_unit.py

@@ -0,0 +1,116 @@
+# -*- coding: utf-8 -*-
+"""A payload captured short of the next layer's header is kept as raw octets.
+
+GitHub issue #1170. Every ``_import_next_layer`` parsed the next layer from
+whatever octets were captured, however few, and only picked
+:class:`~pcapkit.protocols.misc.null.NoPayload` for a declared length of zero.
+:meth:`Schema.unpack <pcapkit.protocols.schema.schema.Schema.unpack>` zero-fills
+the fields the stream ends inside, so
+:meth:`~pcapkit.protocols.protocol.ProtocolBase.from_data` rebuilt a header that
+was never captured. A payload with no captured octets is now a ``NoPayload``,
+and one short of the next layer's header a
+:class:`~pcapkit.protocols.misc.raw.Raw`.
+
+"""
+from __future__ import annotations
+
+import importlib.util
+import unittest
+
+from tests._support import reimport_once_per_class
+
+RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper')
+HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS)
+
+#: A complete 20-octet IPv4 header declaring a 300-octet UDP datagram.
+IPV4 = bytes.fromhex('4500012ca8370000fa11178a00000000ffffffff')
+#: A complete 40-octet IPv6 header declaring a 168-octet UDP payload.
+IPV6 = bytes.fromhex('6000000000a81140') + bytes(15) + b'\x01' + bytes(15) + b'\x02'
+#: A UDP header (port 53 to 53, length 32) and the first of its payload octets.
+UDP = bytes.fromhex('0035003500200000ab')
+#: An Ethernet header carrying IPv4.
+ETHERNET = bytes.fromhex('ffffffffffff0000000000010800')
+
+
+@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed')
+class ShortNextLayerHeaderTests(unittest.TestCase):
+    """The rebuild of a short payload is the octets that were captured."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+
+    def test_ipv6_header_without_payload_octets(self) -> None:
+        from pcapkit.protocols.internet.ipv6 import IPv6
+        from pcapkit.protocols.misc.null import NoPayload
+
+        parsed = IPv6(IPV6, len(IPV6))
+        self.assertIsInstance(parsed.payload, NoPayload)
+        self.assertEqual(IPv6.from_data(parsed.info).data, IPV6)
+
+    def test_ipv6_partial_udp_header(self) -> None:
+        from pcapkit.protocols.internet.ipv6 import IPv6
+        from pcapkit.protocols.misc.raw import Raw
+
+        packet = IPV6 + UDP[:4]
+        parsed = IPv6(packet, len(packet))
+        self.assertIsInstance(parsed.payload, Raw)
+        self.assertEqual(IPv6.from_data(parsed.info).data, packet)
+
+    def test_ipv4_partial_udp_header(self) -> None:
+        from pcapkit.protocols.internet.ipv4 import IPv4
+        from pcapkit.protocols.misc.raw import Raw
+
+        for size in (1, 4, 7):
+            with self.subTest(size=size):
+                packet = IPV4 + UDP[:size]
+                parsed = IPv4(packet, len(packet))
+                self.assertIsInstance(parsed.payload, Raw)
+                self.assertEqual(parsed.payload.data, UDP[:size])
+                self.assertEqual(IPv4.from_data(parsed.info).data, packet)
+
+    def test_udp_header_without_payload_octets(self) -> None:
+        from pcapkit.protocols.internet.ipv4 import IPv4
+        from pcapkit.protocols.misc.null import NoPayload
+        from pcapkit.protocols.transport.udp import UDP as UDP_
+
+        packet = IPV4 + UDP[:8]
+        parsed = IPv4(packet, len(packet))
+        self.assertIsInstance(parsed.payload, UDP_)
+        self.assertIsInstance(parsed.payload.payload, NoPayload)
+        self.assertEqual(IPv4.from_data(parsed.info).data, packet)
+
+    def test_ipv4_header_cut_short(self) -> None:
+        from pcapkit.protocols.link.ethernet import Ethernet
+        from pcapkit.protocols.misc.raw import Raw
+
+        frame = ETHERNET + IPV4[:18]
+        parsed = Ethernet(frame, len(frame))
+        self.assertIsInstance(parsed.payload, Raw)
+        self.assertEqual(Ethernet.from_data(parsed.info).data, frame)
+
+    def test_header_shorter_than_its_length_hint_still_parses(self) -> None:
+        # A 12-octet L2TPv2 ZLB acknowledgement (flags T, L and S) is a whole
+        # header, though ``L2TPv2.__length_hint__`` reports its 16-octet form.
+        from pcapkit.protocols.internet.ipv4 import IPv4
+        from pcapkit.protocols.link.l2tpv2 import L2TPv2
+
+        zlb = bytes.fromhex('c802000c000100000001' '0002')
+        udp = bytes.fromhex('06a506a5') + (8 + len(zlb)).to_bytes(2, 'big') + b'\x00\x00' + zlb
+        ipv4 = IPV4[:2] + (20 + len(udp)).to_bytes(2, 'big') + IPV4[4:]
+        packet = ipv4 + udp
+        parsed = IPv4(packet, len(packet))
+        self.assertIsInstance(parsed.payload.payload, L2TPv2)
+        self.assertEqual(parsed.payload.payload.info.length, 12)
+
+    def test_captured_header_is_still_parsed(self) -> None:
+        from pcapkit.protocols.internet.ipv4 import IPv4
+        from pcapkit.protocols.transport.udp import UDP as UDP_
+
+        packet = IPV4 + UDP
+        parsed = IPv4(packet, len(packet))
+        self.assertIsInstance(parsed.payload, UDP_)
+        self.assertEqual(parsed.payload.info.srcport.port, 53)
+
+
+if __name__ == '__main__':
+    unittest.main()