fix(internet): read a checksum behind a cut TSO IPv4 as truncated (#2689)

docs/source/contributing/pep.rst

@@ -959,13 +959,17 @@ offload heuristic on :issue:`2470`.
   :rfc:`3376#section-4.1.2`, :rfc:`2784#section-2.5`). The IGMPv1 rule follows
   :rfc:`2236`, which updates :rfc:`1112`'s 8-octet one. A GRE header with ``C``
   clear is :attr:`~pcapkit.corekit.checksum.internet.ChecksumStatus.NOT_COMPUTED`,
-  and ``make()`` computes one when ``checksum_present`` asks without giving it.
-  Neither GRE nor the Linux cooked-capture header carries a length, so a GRE frame
-  the capture cut short there is
-  :attr:`~pcapkit.corekit.checksum.internet.ChecksumStatus.INVALID`, settled at
-  implementation (:issue:`2618`); behind IPv4 or IPv6, which state the payload
-  length, it is :attr:`~pcapkit.corekit.checksum.internet.ChecksumStatus.TRUNCATED`
-  (:issue:`2649`).
+  and ``make()`` computes one when ``checksum_present`` asks without giving it
+  (:issue:`2618`). Neither GRE nor the Linux cooked-capture header carries a
+  length, so a GRE message the capture cut short there is
+  :attr:`~pcapkit.corekit.checksum.internet.ChecksumStatus.TRUNCATED` by the
+  original length its PCAP record or PCAP-NG packet block states (:issue:`2653`);
+  behind IPv4 or IPv6, by the payload length they state (:issue:`2649`). An IPv4
+  Total Length of 0 states none, so behind one, and through an AH after it, GRE,
+  ICMP and IGMP go by the original length as well (:issue:`2689`). A GRE or IGMP
+  header the data ends inside is
+  :attr:`~pcapkit.corekit.checksum.internet.ChecksumStatus.TRUNCATED` too
+  (:issue:`2689`).
 * **DCCP covers what its Checksum Coverage gives** (:rfc:`4340#section-9.2`), over
   a pseudo-header that carries the whole packet's length, as TCP's does, so a
   fragment is

pcapkit/corekit/checksum/internet.py

@@ -31,7 +31,7 @@
 from typing import TYPE_CHECKING
 
 from pcapkit.corekit.enum import EnumLookup
-from pcapkit.utilities.contextvars import rebuilding
+from pcapkit.utilities.contextvars import frame_length, rebuilding
 from pcapkit.utilities.exceptions import stacklevel
 from pcapkit.utilities.warnings import ChecksumWarning, warn
 
@@ -204,6 +204,11 @@ def verify_message(alias: 'str', data: 'bytes', field: 'bytes', *,
     verified as the message is parsed, unless an IP fragment carries it; see
     :func:`fragment_payload`.
 
+    Where no layer before the message declares its length, the message runs to
+    the end of its frame, and the octets the capture cut off the frame, as
+    :data:`~pcapkit.utilities.contextvars.frame_length` gives them, are
+    covered too (:issue:`2653`, :issue:`2689`).
+
     Args:
         alias: Protocol alias, to name in the warning.
         data: Octets the checksum covers, the checksum field included.
@@ -221,6 +226,9 @@ def verify_message(alias: 'str', data: 'bytes', field: 'bytes', *,
     """
     if _in_fragment.get():
         return ChecksumStatus.NO_PSEUDO_HEADER
+    lengths = frame_length.get()
+    if lengths is not None:
+        length = (len(data) if length is None else length) + max(lengths[1] - lengths[0], 0)
     return verify_checksum(alias, data, field, length=length)
 
 

pcapkit/protocols/internet/ah.py

@@ -195,7 +195,9 @@ def read(self, length: 'Optional[int]' = None, *, version: 'Literal[4, 6]' = 4,
             # IPv6 header, so hold NoPayload here so that membership does not raise (#2484).
             self._next = NoPayload()  # pylint: disable=attribute-defined-outside-init
             return ah
-        return self._decode_next_layer(ah, schema.next, length - ah.length)
+        # NOTE: The payload is the rest of the parent's, so a frame's cut the
+        # parent handed on is handed on in turn (#2689).
+        return self._decode_next_layer(ah, schema.next, length - ah.length, declared=False)
 
     def make(self,
              next: 'Enum_TransType | StdlibEnum | AenumEnum | str | int' = Enum_TransType.UDP,

pcapkit/protocols/internet/igmp.py

@@ -412,7 +412,10 @@ def _verify_checksum(self, igmp: '_PT', checksum: 'bytes', length: 'int') -> 'No
         :rfc:`3376#section-4.1.2`, :rfc:`3376#section-4.2.2`). That holds for
         IGMPv1 as well: :rfc:`1112` defines an 8-octet message, and :rfc:`2236`,
         which updates it, sums a longer one whole (:rfc:`2236#section-2.5`). See
-        :func:`~pcapkit.corekit.checksum.internet.verify_message`.
+        :func:`~pcapkit.corekit.checksum.internet.verify_message`. A message the
+        data ends inside, as its header's ``__short_read__`` records, is
+        :attr:`~pcapkit.corekit.checksum.internet.ChecksumStatus.TRUNCATED`
+        (:issue:`2689`).
 
         Args:
             igmp: Parsed message data.
@@ -421,9 +424,16 @@ def _verify_checksum(self, igmp: '_PT', checksum: 'bytes', length: 'int') -> 'No
 
         """
         if config.CHECKSUM_VERIFY:
-            from pcapkit.corekit.checksum.internet import verify_message  # isort: skip # pylint: disable=import-outside-toplevel
+            from pcapkit.corekit.checksum.internet import (  # isort: skip # pylint: disable=import-outside-toplevel
+                ChecksumStatus, verify_message)
+            # NOTE: The checksum covers the whole message, so a message the data
+            # ends inside leaves part of what it covers uncaptured (#2689).
+            if '__short_read__' in self.__header__.__dict__:
+                status = ChecksumStatus.TRUNCATED
+            else:
+                status = verify_message(self.alias, self._data, checksum, length=length)
             igmp.__update__([
-                ('checksum_status', verify_message(self.alias, self._data, checksum, length=length)),
+                ('checksum_status', status),
             ])
 
     @classmethod

pcapkit/protocols/internet/internet.py

@@ -449,7 +449,7 @@ class registered for ``proto``. Paired with it, the keywords that
     def _decode_next_layer(self, dict_: '_PT', proto: 'Optional[int]' = None,  # pylint: disable=arguments-differ
                            length: 'Optional[int]' = None, *, packet: 'Optional[dict[str, Any]]' = None,
                            version: 'Literal[4, 6]' = 4, ipv6_exthdr: 'Optional[ProtoChain]' = None,
-                           payload: 'Optional[bytes]' = None) -> '_PT':
+                           payload: 'Optional[bytes]' = None, declared: 'bool' = True) -> '_PT':
         r"""Decode next layer extractor.
 
         Arguments:
@@ -461,6 +461,7 @@ def _decode_next_layer(self, dict_: '_PT', proto: 'Optional[int]' = None,  # pyl
             ipv6_exthdr: protocol chain of IPv6 extension headers
             payload: payload from packet. If not provided, will extract from
                 :meth:`self.__header__.get_payload <pcapkit.protocols.schema.schema.Schema.get_payload>`
+            declared: whether this layer declares the length of its payload
 
         Returns:
             Current protocol with next layer extracted.
@@ -474,18 +475,22 @@ def _decode_next_layer(self, dict_: '_PT', proto: 'Optional[int]' = None,  # pyl
             The next layer is parsed with
             :data:`~pcapkit.utilities.contextvars.frame_length` cleared, since
             this layer, not the frame, bounds its payload (:issue:`2653`).
+            With ``declared`` false it is left as it is, for a payload that
+            runs to the end of the frame, as behind an IPv4 Total Length of 0,
+            or to the end of the parent's (:issue:`2689`).
 
         """
         # NOTE: The length this layer hands on bounds the payload, so octets the
         # capture cut off the frame past it are none of the next layer's (#2653).
-        # An IPv4 Total Length of 0, read as the rest of the frame, loses the cut.
-        token = frame_length.set(None)
+        # A layer declaring none hands the frame's cut on instead (#2689).
+        token = frame_length.set(None) if declared else None
         try:
             next_ = cast('ProtocolBase',  # type: ignore[redundant-cast]
                          self._import_next_layer(proto, length, packet=packet, version=version,
                                                  payload=payload))  # type: ignore[misc,call-arg]
         finally:
-            frame_length.reset(token)
+            if token is not None:
+                frame_length.reset(token)
         info, chain = next_.info, next_.protochain
 
         # make next layer protocol name

pcapkit/protocols/internet/ipv4.py

@@ -442,13 +442,16 @@ def read(self, length: 'Optional[int]' = None, *,  # pylint: disable=unused-argu
         # (#1553). Reassembly takes ``packet.payload``, the octets as
         # captured, which this leaves as they were.
         proto = None if ipv4.offset else ipv4.protocol
+        # NOTE: A Total Length of 0 declares none, so the frame's lengths are the
+        # only record of a cut and the payload keeps them (#2689).
+        declared = bool(ipv4.len)
         if config.CHECKSUM_VERIFY:
             # NOTE: An upper layer checksummed alone, as ICMP is, is not verified
             # in a fragment; see pcapkit.corekit.checksum.internet (#2470).
             from pcapkit.corekit.checksum.internet import fragment_payload  # isort: skip # pylint: disable=import-outside-toplevel
             with fragment_payload(bool(ipv4.offset or ipv4.flags.mf)):
-                return self._decode_next_layer(ipv4, proto, _plen, packet=__packet__)
-        return self._decode_next_layer(ipv4, proto, _plen, packet=__packet__)
+                return self._decode_next_layer(ipv4, proto, _plen, packet=__packet__, declared=declared)
+        return self._decode_next_layer(ipv4, proto, _plen, packet=__packet__, declared=declared)
 
     def make(self,
              tos_pre: 'Enum_ToSPrecedence | StdlibEnum | AenumEnum | int | str' = Enum_ToSPrecedence.Routine,

pcapkit/protocols/link/gre.py

@@ -78,7 +78,6 @@
 from pcapkit.protocols.link.link import Link
 from pcapkit.protocols.schema.link.gre import GRE as Schema_GRE
 from pcapkit.utilities import config
-from pcapkit.utilities.contextvars import frame_length
 from pcapkit.utilities.exceptions import ProtocolError
 
 if TYPE_CHECKING:
@@ -113,7 +112,8 @@ class GRE(Link[Data_GRE, Schema_GRE],
     capture cut short, holding fewer octets than its original length, reads
     as :attr:`~pcapkit.corekit.checksum.internet.ChecksumStatus.TRUNCATED`,
     as :data:`~pcapkit.utilities.contextvars.frame_length` tells it
-    (:issue:`2653`).
+    (:issue:`2653`), and so does a header the data ends inside
+    (:issue:`2689`).
 
     """
 
@@ -223,14 +223,15 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_GRE':
                 ChecksumStatus, verify_message)
             if _checksum is None:
                 status = ChecksumStatus.NOT_COMPUTED
+            elif '__short_read__' in schema.__dict__:
+                # NOTE: The checksum covers the whole message, so a header the data
+                # ends inside leaves part of what it covers uncaptured (#2689).
+                status = ChecksumStatus.TRUNCATED
             else:
                 # NOTE: Behind LinuxSLL the message runs to the end of its frame, so
-                # what the capture cut off is covered too. Behind IP the value is None:
-                # Internet clears it, as the length IP declares bounds GRE (#2653).
-                lengths = frame_length.get()
-                missing = 0 if lengths is None else max(lengths[1] - lengths[0], 0)
+                # verify_message covers what the capture cut off too (#2653).
                 status = verify_message(self.alias, self._data, _checksum.to_bytes(2, 'big'),
-                                        length=length + missing)
+                                        length=length)
             gre.__update__([
                 ('checksum_status', status),
             ])

pcapkit/utilities/contextvars.py

@@ -38,11 +38,12 @@
 #: contextvars.ContextVar[tuple[int, int] | None]: The captured and the original
 #: length of the frame being dissected, in octets, as its PCAP record or PCAP-NG
 #: packet block states them; :data:`None` outside a frame, and inside the payload
-#: of an :class:`~pcapkit.protocols.internet.internet.Internet` layer, which
-#: bounds it. A frame the capture cut short holds fewer octets than were on the
-#: wire, which a layer stating no length of its own, as
-#: :class:`~pcapkit.protocols.link.gre.GRE`, cannot tell from the octets it is
-#: handed (:issue:`2653`).
+#: of an :class:`~pcapkit.protocols.internet.internet.Internet` layer that
+#: bounds it, which neither an IPv4 Total Length of 0 nor
+#: :class:`~pcapkit.protocols.internet.ah.AH` does (:issue:`2689`). A frame
+#: the capture cut short holds fewer octets than were on the wire, which a layer
+#: stating no length of its own, as :class:`~pcapkit.protocols.link.gre.GRE`,
+#: cannot tell from the octets it is handed (:issue:`2653`).
 frame_length = contextvars.ContextVar(
     'pcapkit.utilities.contextvars.frame_length', default=None,
 )  # type: contextvars.ContextVar[Optional[tuple[int, int]]]

tests/protocols/internet/test_igmp_header_cut_2689_unit.py

@@ -0,0 +1,121 @@
+# -*- coding: utf-8 -*-
+"""An IGMP message the data ends inside reads ``TRUNCATED``, not ``INVALID``.
+
+GitHub issue #2689: built directly on octets that end inside it, with no enclosing
+length, an IGMPv1, IGMPv2 or IGMPv3 message reads the fields the data never held as
+zeros, and verifying that zero-filled checksum would read ``INVALID`` with a
+``ChecksumWarning``. Every version's checksum covers the whole message
+(:rfc:`2236#section-2.3`, :rfc:`3376#section-4.1.2`), so
+:meth:`IGMP._verify_checksum <pcapkit.protocols.internet.igmp.IGMP._verify_checksum>`
+reads the header's own ``__short_read__`` record as ``TRUNCATED``, as GRE does. A
+whole message is verified.
+
+The messages are built here; this module is unit-tier and reads no capture file.
+
+"""
+from __future__ import annotations
+
+import unittest
+import warnings
+from typing import TYPE_CHECKING
+from unittest import mock
+
+from tests._support import reimport_once_per_class
+from tests.foundation import _roundtrip as wire
+
+if TYPE_CHECKING:
+    from typing import Any
+
+#: Test tier (#1958): hand-picked parse -> rebuild round trips.
+__tier__ = 'ALL'
+
+
+def message(head: bytes, rest: bytes) -> bytes:
+    """``head`` (type and Max Resp Code), the checksum over the whole message, then ``rest``."""
+    return head + wire.checksum(head + b'\x00\x00' + rest).to_bytes(2, 'big') + rest
+
+
+#: A group record of type MODE_IS_INCLUDE for 239.1.1.1 with one source, 10.0.0.1.
+RECORD = bytes.fromhex('01000001ef0101010a000001')
+
+#: Each message: the class it is read as, its octets, and the lengths of the cuts
+#: that leave a short read on its header. Shorter than eight octets, an IGMPv2 or
+#: IGMPv3 Query is refused by its class, which classifies a Query by its length.
+#: Eight octets of the Report are a whole header whose record count no octet
+#: backs, which records no short read.
+MESSAGES = {
+    'v1-report': ('IGMPv1', message(b'\x12\x00', bytes([224, 0, 0, 1])), range(1, 8)),
+    'v2-report': ('IGMPv2', message(b'\x16\x00', bytes([239, 1, 2, 3])), range(1, 8)),
+    'v2-leave': ('IGMPv2', message(b'\x17\x00', bytes([239, 1, 2, 3])), range(1, 8)),
+    'v3-report': ('IGMPv3', message(b'\x22\x00', bytes.fromhex('00000001') + RECORD),
+                  [*range(1, 8), *range(9, 20)]),
+    # A General Query with QRV 2, QQIC 125 and one source, 10.0.0.1.
+    'v3-query': ('IGMPv3', message(b'\x11\x64', bytes(4) + bytes.fromhex('027d00010a000001')),
+                 range(12, 16)),
+}
+
+
+def wrong(msg: bytes) -> bytes:
+    """``msg`` with its checksum flipped in both octets."""
+    return msg[:2] + (int.from_bytes(msg[2:4], 'big') ^ 0x0101).to_bytes(2, 'big') + msg[4:]
+
+
+class IGMPHeaderCutTests(unittest.TestCase):
+    """An IGMP message the data ends inside holds no checksum to verify."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+        from pcapkit.corekit.checksum.internet import ChecksumStatus
+        from pcapkit.protocols.internet.igmpv1 import IGMPv1
+        from pcapkit.protocols.internet.igmpv2 import IGMPv2
+        from pcapkit.protocols.internet.igmpv3 import IGMPv3
+        from pcapkit.utilities import config
+        from pcapkit.utilities.warnings import ChecksumWarning
+
+        self.classes = {'IGMPv1': IGMPv1, 'IGMPv2': IGMPv2, 'IGMPv3': IGMPv3}
+        self.Status = ChecksumStatus
+        self.config = config
+        self.ChecksumWarning = ChecksumWarning
+
+    def read(self, name: str, data: bytes) -> 'tuple[Any, list[str]]':
+        """``data`` read as class ``name`` with verification on, and the checksum warnings."""
+        with mock.patch.object(self.config, 'CHECKSUM_VERIFY', True), \
+                warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter('always')
+            layer = self.classes[name](data, len(data))
+        return layer, [str(item.message) for item in caught
+                       if issubclass(item.category, self.ChecksumWarning)]
+
+    def test_every_cut_reads_truncated(self) -> None:
+        for label, (name, msg, cuts) in MESSAGES.items():
+            for bad in (False, True):
+                data = wrong(msg) if bad else msg
+                for count in cuts:
+                    with self.subTest(message=label, bad=bad, captured=count):
+                        layer, caught = self.read(name, data[:count])
+                        self.assertIsNotNone(layer.info.get('__short_read__'))
+                        self.assertEqual((layer.info.checksum_status, caught), (self.Status.TRUNCATED, []))
+
+    def test_a_whole_message_is_verified(self) -> None:
+        for label, (name, msg, _) in MESSAGES.items():
+            with self.subTest(message=label):
+                layer, caught = self.read(name, msg)
+                self.assertIsNone(layer.info.get('__short_read__'))
+                self.assertEqual((layer.info.checksum_status, caught), (self.Status.VALID, []))
+                layer, caught = self.read(name, wrong(msg))
+                self.assertIs(layer.info.checksum_status, self.Status.INVALID)
+                self.assertEqual(len(caught), 1)
+                self.assertRegex(caught[0], r'^IGMP(v[23])?: checksum 0x[0-9a-f]{4} is wrong;')
+
+    def test_parse_rebuild_is_exact(self) -> None:
+        for label, (name, msg, cuts) in MESSAGES.items():
+            for count in cuts:
+                with self.subTest(message=label, captured=count):
+                    layer, _ = self.read(name, msg[:count])
+                    with warnings.catch_warnings():
+                        warnings.simplefilter('ignore')
+                        self.assertEqual(self.classes[name].from_data(layer.info).data, msg[:count])
+
+
+if __name__ == '__main__':
+    unittest.main()