fix(protocols): understated __length_hint__ and an unguarded next-layer import still fabricate octets

pcapkit/protocols/internet/hopopt.py

@@ -415,9 +415,14 @@ def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None, length: 'Opt
         # call super __post_init__
         super().__post_init__(file, length, extension=extension, **kwargs)  # type: ignore[arg-type]
 
-    def __length_hint__(self) -> 'Literal[2]':
-        """Return an estimated length for the object."""
-        return 2
+    def __length_hint__(self) -> 'Literal[8]':
+        """Return an estimated length for the object.
+
+        Eight octets: the two fixed ones and the options that
+        :rfc:`8200#section-4.3` counts in the first 8-octet unit.
+
+        """
+        return 8
 
     @classmethod
     def __index__(cls) -> 'Enum_TransType':  # pylint: disable=invalid-index-returned

pcapkit/protocols/internet/ipv6_opts.py

@@ -407,9 +407,14 @@ def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None, length: 'Opt
         # call super __post_init__
         super().__post_init__(file, length, extension=extension, **kwargs)  # type: ignore[arg-type]
 
-    def __length_hint__(self) -> 'Literal[2]':
-        """Return an estimated length for the object."""
-        return 2
+    def __length_hint__(self) -> 'Literal[8]':
+        """Return an estimated length for the object.
+
+        Eight octets: the two fixed ones and the options that
+        :rfc:`8200#section-4.6` counts in the first 8-octet unit.
+
+        """
+        return 8
 
     @classmethod
     def __index__(cls) -> 'Enum_TransType':  # pylint: disable=invalid-index-returned

pcapkit/protocols/internet/ipv6_route.py

@@ -389,9 +389,14 @@ def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None, length: 'Opt
         # call super __post_init__
         super().__post_init__(file, length, extension=extension, **kwargs)  # type: ignore[arg-type]
 
-    def __length_hint__(self) -> 'Literal[4]':
-        """Return an estimated length for the object."""
-        return 4
+    def __length_hint__(self) -> 'Literal[8]':
+        """Return an estimated length for the object.
+
+        Eight octets: the four fixed ones and the type-specific data that
+        :rfc:`8200#section-4.4` counts in the first 8-octet unit.
+
+        """
+        return 8
 
     @classmethod
     def __index__(cls) -> 'Enum_TransType':  # pylint: disable=invalid-index-returned

pcapkit/protocols/internet/mh.py

@@ -1579,9 +1579,14 @@ def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None, length: 'Opt
         # call super __post_init__
         super().__post_init__(file, length, extension=extension, **kwargs)  # type: ignore[arg-type]
 
-    def __length_hint__(self) -> 'Literal[6]':
-        """Return an estimated length for the object."""
-        return 6
+    def __length_hint__(self) -> 'Literal[8]':
+        """Return an estimated length for the object.
+
+        Eight octets: the six fixed ones and the message data that
+        :rfc:`6275#section-6.1.1` counts in the first 8-octet unit.
+
+        """
+        return 8
 
     @classmethod
     def __index__(cls) -> 'Enum_TransType':  # pylint: disable=invalid-index-returned

tests/protocols/internet/test_ipv6_extension_unit.py

@@ -191,7 +191,7 @@ def test_hopopt_index_length_and_make_data(self) -> None:
         proto = object.__new__(HOPOPT)
 
         self.assertEqual(HOPOPT.__index__(), TransType.HOPOPT)
-        self.assertEqual(proto.__length_hint__(), 2)
+        self.assertEqual(proto.__length_hint__(), 8)
         values = HOPOPT._make_data(data)
         self.assertEqual(values['next'], TransType.UDP)
         self.assertEqual(values['options'], options)
@@ -206,7 +206,7 @@ def test_ipv6_opts_index_length_and_make_data(self) -> None:
         proto = object.__new__(IPv6_Opts)
 
         self.assertEqual(IPv6_Opts.__index__(), TransType.IPv6_Opts)
-        self.assertEqual(proto.__length_hint__(), 2)
+        self.assertEqual(proto.__length_hint__(), 8)
         values = IPv6_Opts._make_data(data)
         self.assertEqual(values['next'], TransType.UDP)
         self.assertEqual(values['options'], options)
@@ -228,7 +228,7 @@ def test_ipv6_route_index_length_and_make_data(self) -> None:
         proto._info = data
 
         self.assertEqual(IPv6_Route.__index__(), TransType.IPv6_Route)
-        self.assertEqual(proto.__length_hint__(), 4)
+        self.assertEqual(proto.__length_hint__(), 8)
         self.assertEqual(proto.length, 24)
         values = IPv6_Route._make_data(data)
         self.assertEqual(values['next'], TransType.UDP)

tests/protocols/internet/test_length_hint_minimum_header_unit.py

@@ -0,0 +1,91 @@
+# -*- coding: utf-8 -*-
+"""Every ``__length_hint__`` is its protocol's minimum header length.
+
+GitHub issue #1173. :meth:`~pcapkit.protocols.protocol.ProtocolBase._parse_next_layer`
+keeps a next layer as :class:`~pcapkit.protocols.misc.raw.Raw` when fewer octets
+were captured than its ``__length_hint__`` (:issue:`1170`). ``IPv6_Route`` hinted
+4 octets and ``MH`` 6, though each header is at least 8, so a stream of 4--7 or
+6--7 octets was parsed and rebuilt as a full 8-octet header. ``HOPOPT`` and
+``IPv6_Opts`` hinted 2 for the same 8-octet minimum. An application layer's
+payload reaches the same guard through
+:meth:`ProtocolBase._import_next_layer <pcapkit.protocols.protocol.ProtocolBase._import_next_layer>`.
+
+"""
+from __future__ import annotations
+
+import importlib.util
+import unittest
+import unittest.mock
+
+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)
+
+
+def ipv4(proto: int) -> bytes:
+    """A complete 20-octet IPv4 header declaring a 300-octet ``proto`` payload."""
+    return bytes.fromhex('4500012ca8370000fa') + bytes([proto]) + bytes.fromhex('178a00000000ffffffff')
+
+
+#: An OSPFv2 Hello header with no authentication, declaring a 27-octet packet.
+OSPF_HEADER = bytes.fromhex('0201001b010203040000000000000000') + bytes(8)
+
+
+@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed')
+class LengthHintMinimumHeaderTests(unittest.TestCase):
+    """A stream short of the minimum header rebuilds exactly what was captured."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+
+    def test_hints_are_the_eight_octet_minimum(self) -> None:
+        from pcapkit.protocols.internet.hopopt import HOPOPT
+        from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts
+        from pcapkit.protocols.internet.ipv6_route import IPv6_Route
+        from pcapkit.protocols.internet.mh import MH
+
+        for klass in (IPv6_Route, MH, HOPOPT, IPv6_Opts):
+            with self.subTest(klass=klass.__name__):
+                self.assertEqual(object.__new__(klass).__length_hint__(), 8)
+
+    def test_short_routing_and_mobility_headers_are_raw(self) -> None:
+        from pcapkit.protocols.internet.ipv4 import IPv4
+        from pcapkit.protocols.misc.raw import Raw
+
+        for proto, sizes in ((43, range(4, 8)), (135, range(6, 8))):
+            for size in sizes:
+                with self.subTest(proto=proto, size=size):
+                    packet = ipv4(proto) + bytes(size)
+                    parsed = IPv4(packet, len(packet))
+                    self.assertIsInstance(parsed.payload, Raw)
+                    self.assertEqual(IPv4.from_data(parsed.info).data, packet)
+
+    def test_complete_headers_still_parse(self) -> None:
+        from pcapkit.protocols.internet.ipv4 import IPv4
+        from pcapkit.protocols.internet.ipv6_route import IPv6_Route
+        from pcapkit.protocols.internet.mh import MH
+
+        for proto, klass in ((43, IPv6_Route), (135, MH)):
+            with self.subTest(klass=klass.__name__):
+                packet = ipv4(proto) + bytes([59]) + bytes(7)
+                parsed = IPv4(packet, len(packet))
+                self.assertIsInstance(parsed.payload, klass)
+                self.assertEqual(parsed.payload.length, 8)
+
+    def test_application_payload_goes_through_the_guard(self) -> None:
+        from pcapkit.protocols.application.ospf import OSPF
+        from pcapkit.protocols.misc.raw import Raw
+        from pcapkit.protocols.protocol import ProtocolBase
+
+        packet = OSPF_HEADER + b'\xab\xcd\xef'
+        guard = ProtocolBase._parse_next_layer
+        with unittest.mock.patch.object(ProtocolBase, '_parse_next_layer', side_effect=guard) as spy:
+            parsed = OSPF(packet, len(packet))
+        self.assertIsInstance(parsed.payload, Raw)
+        self.assertEqual(parsed.payload.data, b'\xab\xcd\xef')
+        self.assertEqual([call.args[0] for call in spy.call_args_list], [Raw])
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/protocols/internet/test_mh_unit.py

@@ -37,7 +37,7 @@ def test_mh_index_length_and_make_data(self) -> None:
         proto = object.__new__(MH)
 
         self.assertEqual(MH.__index__(), TransType.Mobility_Header)
-        self.assertEqual(proto.__length_hint__(), 6)
+        self.assertEqual(proto.__length_hint__(), 8)
         values = MH._make_data(data)
         self.assertEqual(values['next'], TransType.UDP)
         self.assertEqual(values['type'], Packet.Binding_Refresh_Request)