diff --git a/pcapkit/protocols/internet/hopopt.py b/pcapkit/protocols/internet/hopopt.py index 0b7bff9a08..3060794bf0 100644 --- a/pcapkit/protocols/internet/hopopt.py +++ b/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 diff --git a/pcapkit/protocols/internet/ipv6_opts.py b/pcapkit/protocols/internet/ipv6_opts.py index 6aea7dea18..e841f9ac2a 100644 --- a/pcapkit/protocols/internet/ipv6_opts.py +++ b/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 diff --git a/pcapkit/protocols/internet/ipv6_route.py b/pcapkit/protocols/internet/ipv6_route.py index fdf4bd6383..d3ae36ad63 100644 --- a/pcapkit/protocols/internet/ipv6_route.py +++ b/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 diff --git a/pcapkit/protocols/internet/mh.py b/pcapkit/protocols/internet/mh.py index 516fa8328a..cd9e7da453 100644 --- a/pcapkit/protocols/internet/mh.py +++ b/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 diff --git a/tests/protocols/internet/test_ipv6_extension_unit.py b/tests/protocols/internet/test_ipv6_extension_unit.py index 4d52ca4d80..c95231c1be 100644 --- a/tests/protocols/internet/test_ipv6_extension_unit.py +++ b/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) diff --git a/tests/protocols/internet/test_length_hint_minimum_header_unit.py b/tests/protocols/internet/test_length_hint_minimum_header_unit.py new file mode 100644 index 0000000000..abf1dc9529 --- /dev/null +++ b/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 `. + +""" +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() diff --git a/tests/protocols/internet/test_mh_unit.py b/tests/protocols/internet/test_mh_unit.py index 51a1b82e1d..443df73584 100644 --- a/tests/protocols/internet/test_mh_unit.py +++ b/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)