fix(protocols)!: a declared length running past the data is accepted and zero-filled on rebuild

OptionField.unpack reads each option through a proxy that notes short
reads, and raises ProtocolError when an option that began with octets
present then reads past the end of its data. An option starting at the
end of the data keeps the end-of-option-list handling. Inside a parent
layer the octets are kept as captured and rebuild byte for byte.

BREAKING CHANGE: a truncated IPv4 or TCP option no longer parses with
zero-filled data; the layer raises ProtocolError, and a parent keeps it
as raw octets. The #431/#572 pins and the #1394 raw-item cases are
rewritten to match. PCAP-NG is a follow-up after #1378.

Refs #1325

Author: JarryShaw

pcapkit/corekit/fields/collections.py

@@ -9,7 +9,7 @@
 from pcapkit.corekit.fields.numbers import NumberField
 from pcapkit.corekit.multidict import OrderedMultiDict
 from pcapkit.utilities.compat import List
-from pcapkit.utilities.exceptions import FieldValueError
+from pcapkit.utilities.exceptions import FieldValueError, ProtocolError
 
 __all__ = [
     'ListField', 'OptionField',
@@ -29,6 +29,44 @@
 _TS = TypeVar('_TS', bound='Schema')
 
 
+class _ShortReadTracker:
+    """Stream proxy recording whether a read came back short.
+
+    Args:
+        file: Stream to read from.
+
+    Only the ``read``, ``seek`` and ``tell`` calls that
+    :meth:`Schema.unpack <pcapkit.protocols.schema.schema.Schema.unpack>` and
+    :meth:`OptionField.unpack` make are delegated.
+
+    """
+
+    __slots__ = ('_file', 'short')
+
+    def __init__(self, file: 'IO[bytes]') -> 'None':
+        self._file = file
+        #: bool: Whether a read since the last reset returned fewer octets than
+        #: it asked for.
+        self.short = False
+
+    def read(self, size: 'Optional[int]' = -1) -> 'bytes':
+        """Read ``size`` octets, noting a short read."""
+        if size is None:
+            size = -1
+        data = self._file.read(size)
+        if 0 <= size and len(data) < size:
+            self.short = True
+        return data
+
+    def seek(self, offset: 'int', whence: 'int' = io.SEEK_SET) -> 'int':
+        """Seek the underlying stream."""
+        return self._file.seek(offset, whence)
+
+    def tell(self) -> 'int':
+        """Tell the underlying stream's position."""
+        return self._file.tell()
+
+
 class ListField(FieldBase[List[_TL]], Generic[_TL]):
     """Field list for protocol fields.
 
@@ -362,6 +400,8 @@ def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> 'list
         Raises:
             FieldValueError: If an option consumes nothing from ``buffer``, since
                 the loop below has then no way to get past it.
+            ProtocolError: If an option that begins with octets present reads
+                past the end of the data.
 
         """
         length = self._length
@@ -421,8 +461,24 @@ def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> 'list
         # number of octets to rewind to get back to the front of the option.
         type_field = self._type_field
 
+        # NOTE: An option whose declared length runs past the octets behind it
+        # reads short, and every field reading off the end pads the missing
+        # octets with zeros, so the option would rebuild octets that were never
+        # there: a HIP ``ACK`` declaring 256 octets with 4 present, or an MH
+        # ``PadN`` declaring 3 with none. Nothing downstream can tell, because
+        # the schema records the zeros as values. C.f. #1325.
+        #
+        # So the stream is read through a proxy that notes any short read, and
+        # an option that began with at least one octet present and then read
+        # short is rejected. An option that begins at the end of the data reads
+        # nothing at all and is left to the end-of-option-list and progress
+        # handling below, which is what an option area declared longer than its
+        # octets relies on.
+        tracker = file = _ShortReadTracker(file)  # type: ignore[assignment]
+
         temp = []  # type: list[_TS]
         while length > 0:
+            tracker.short = False
             if type_field is None:
                 # unpack option type using base schema
                 meta = self._base_schema.unpack(file, length, packet)  # type: ignore[call-arg,misc,var-annotated]
@@ -443,6 +499,13 @@ def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> 'list
 
             # unpack option using option schema
             data = schema.unpack(file, length, packet)  # type: ignore[call-arg,misc,var-annotated]
+            if tracker.short and consumed > 0:
+                raise ProtocolError(
+                    f'Field {self.name} has an option that runs past the end of '
+                    f'the data: {code!r} at offset {offset - start} of '
+                    f'{self._length}, with only {file.tell() - offset} octet(s) '
+                    f'of it present'
+                )
             new_packet[self.name].add(code, data)
             temp.append(data)
 

tests/protocols/internet/test_ipv4_unit.py

@@ -1992,82 +1992,49 @@ def test_an_option_area_longer_than_the_datagram_still_parses(self) -> None:
         )
 
     def test_a_truncated_option_still_parses_its_declared_length(self) -> None:
-        """A capture cut short mid-option is tolerated, not just mid-header. C.f. #431, #572.
+        """A capture cut short mid-option is refused alone, and kept as captured in a frame. C.f. #431, #572, #1325.
 
         :meth:`test_an_option_area_longer_than_the_datagram_still_parses` above
         pins the *empty*-tail half of the #431 accommodation: an option area
-        that runs out before it starts. Nothing pinned the other half -- an
-        option that *does* start, declares more data than the capture actually
-        holds, and runs out partway through, the shape of a datagram cut short
-        by the snapshot length rather than one with no options at all. A
-        candidate fix for #554 (PR #571) turned that into an unwrapped
-        ``FieldValueError`` while the rest of the suite stayed green, because
-        nothing exercised it. See
-        ``TCPUDPUnitTests.test_a_truncated_option_still_parses_its_declared_length``
-        for the same case on TCP, where the shortfall is simpler to reach.
+        that runs out before it starts. This is the other half -- an option
+        that *does* start, declares more data than the capture holds, and runs
+        out partway through.
 
         The header below sets ``ihl`` to 9 -- 16 declared octets of options --
-        for an unassigned option (code 31) declaring ``length=12``, which asks
-        :class:`~pcapkit.protocols.schema.internet.ipv4.UnassignedOption`'s
-        ``data`` field (``BytesField(length=lambda pkt: pkt['length'] - 2)``,
-        10 octets here) for more than the 6 octets actually behind it, and
-        :meth:`FieldBase.unpack <pcapkit.corekit.fields.field.FieldBase.unpack>`
-        pads the short read with zero octets rather than raising -- the
-        same accommodation as the TCP case, reached the same way. The zeros go
-        on the *tail*, where the octets that were never read would have been;
-        before #604 they were placed at the front, which for a numeric field
-        corrupted the value outright.
-
-        ``ihl`` has to declare *more* than the 8 octets actually present,
-        though, which the TCP case does not need. Unlike TCP,
-        :meth:`~pcapkit.protocols.internet.ipv4.IPv4._read_ipv4_options` sums
-        each option's self-*declared* ``length`` -- not what it actually
-        consumed -- and raises ``IPv4: invalid format`` once its loop over the
-        parsed options finishes, if that sum exceeds the declared option
-        area. Declaring exactly 8 does not skip the accommodation -- the short
-        ``data`` field is still read and tail-padded as above -- it just trips
-        that check afterwards, discarding the result before the test can
-        assert on it. Declaring 16 leaves headroom, at the cost of a second
-        effect: once the option loop's 16-octet budget outlives the 8 octets
-        its one real option consumed, the loop reads one further, fully
-        exhausted phantom option, decodes it as end-of-option-list (the same
-        mechanism the test above pins), and :meth:`Schema.unpack
-        <pcapkit.protocols.schema.schema.Schema.unpack>` hands the same 8
-        octets to the schema a second time as padding, via the
-        ``option_padding`` rewind that #371 added -- code that predates #431
-        and lives outside :meth:`OptionField.unpack
-        <pcapkit.corekit.fields.collections.OptionField.unpack>`. That is why
-        ``bytes(proto.__header__)`` does not round-trip to ``raw`` here and is
-        not asserted -- immaterial to what this test pins: the option's
-        declared ``length``, the full parsed options list, and the ``data``
-        field's short-read reconstruction.
+        for an unassigned option (code 31) declaring ``length=12``, with only
+        6 of its 10 data octets behind it. Those 4 missing octets used to be
+        read as zeros and rebuilt as if captured. :meth:`OptionField.unpack
+        <pcapkit.corekit.fields.collections.OptionField.unpack>` now refuses
+        the option with :exc:`~pcapkit.utilities.exceptions.ProtocolError`
+        instead (#1325). Inside a frame the datagram is kept as the octets
+        captured, a :class:`~pcapkit.protocols.misc.raw.Raw` payload carrying
+        that error, and rebuilds byte for byte.
 
         """
         from pcapkit.const.ipv4.option_number import OptionNumber
         from pcapkit.protocols.internet.ipv4 import IPv4
+        from pcapkit.protocols.link.ethernet import Ethernet
+        from pcapkit.protocols.misc.raw import Raw
+        from pcapkit.utilities.exceptions import ProtocolError
         from tests._support import time_limit
 
         custom = OptionNumber.get(31)
         trailing = bytes.fromhex('aabbccddeeff')
         raw = (bytes.fromhex('4900001c00010000400600000a0000010a000002') +
                bytes([custom, 12]) + trailing)
-        with time_limit():
-            proto = IPv4(raw, len(raw))
+        with time_limit(), warnings.catch_warnings():
+            warnings.simplefilter('ignore')
+            with self.assertRaisesRegex(ProtocolError, 'runs past the end of the data'):
+                IPv4(raw, len(raw))
 
-        self.assertEqual(proto.info.hdr_len, 36)
-        self.assertEqual(
-            [(code, opt.length) for code, opt in proto.info.options.items(multi=True)],
-            [(custom, 12), (OptionNumber.EOOL, 1)],
-        )
-        unassigned = next(opt for code, opt in proto.info.options.items(multi=True)
-                           if code == custom)
-        # the six octets actually behind the option come first, where they were
-        # read, and the four synthesised for the ones that were not follow them.
-        # Before #604 they arrived the other way round. See #604 and
-        # ``FieldBaseShortReadPaddingSideTests`` in
-        # ``tests/corekit/test_fields_field.py``.
-        self.assertEqual(unassigned.data, trailing + b'\x00\x00\x00\x00')
+            frame = bytes.fromhex('0123456789ab' 'fedcba987654' '0800') + raw
+            proto = Ethernet(frame, len(frame))
 
+        self.assertIsInstance(proto.payload, Raw)
+        self.assertIn('runs past the end of the data', str(proto.payload.info.error))
+        # the option octets are kept as captured, with nothing synthesised
+        self.assertEqual(proto.payload.data, raw)
+        self.assertEqual(Ethernet.from_data(proto.info).data, frame)
 
 if __name__ == '__main__':
     unittest.main()

tests/protocols/test_declared_length_overrun_unit.py

@@ -0,0 +1,174 @@
+# -*- coding: utf-8 -*-
+"""An option declaring more octets than are present is rejected, not zero-filled.
+
+GitHub issue #1325: :meth:`OptionField.unpack
+<pcapkit.corekit.fields.collections.OptionField.unpack>` let an option schema
+read past the end of its data, and every field reading off the end padded the
+missing octets with zeros. A HIP ``ACK`` declaring 256 octets with 4 present
+parsed as 64 update IDs and rebuilt to 304 octets; an MH ``PadN`` declaring 3
+octets with none present rebuilt 24 octets from 16.
+
+Such an option now raises :exc:`~pcapkit.utilities.exceptions.ProtocolError`.
+Inside a parent layer the octets are kept as captured -- raw extension data
+under IPv6, a :class:`~pcapkit.protocols.misc.raw.Raw` payload elsewhere -- and
+the parent rebuilds byte for byte.
+
+An option that begins at the end of the data reads nothing at all, and is still
+left to the end-of-option-list handling an over-long option area relies on.
+
+Every case builds its own octets in memory and reads no capture.
+
+"""
+
+import struct
+import unittest
+import warnings
+
+from tests._support import reimport_once_per_class
+
+
+def _hip(params: bytes) -> bytes:
+    """A HIP ``I1`` header (40 octets) carrying ``params``."""
+    return (bytes([59, (40 + len(params)) // 8 - 1, 1, 0x11, 0, 0, 0, 0])
+            + b'\x11' * 16 + b'\x22' * 16 + params)
+
+
+def _ext6(opts: bytes) -> bytes:
+    """An IPv6 options extension header carrying ``opts``."""
+    return bytes([59, (2 + len(opts)) // 8 - 1]) + opts
+
+
+def _sctp(chunks: bytes) -> bytes:
+    """An SCTP common header carrying ``chunks``."""
+    return struct.pack('>HHII', 1, 2, 0, 0) + chunks
+
+
+def _tcp(opts: bytes) -> bytes:
+    """A TCP header carrying ``opts``."""
+    return struct.pack('>HHIIBBHHH', 1, 2, 0, 0, (5 + len(opts) // 4) << 4, 0x10, 0, 0, 0) + opts
+
+
+def _ipv4(opts: bytes, ihl: int, proto: int = 253, payload: bytes = b'') -> bytes:
+    """An IPv4 header declaring ``ihl`` and carrying ``opts`` and ``payload``."""
+    return (bytes([0x40 | ihl, 0]) + struct.pack('>H', 20 + len(opts) + len(payload))
+            + bytes.fromhex('00010000 40') + bytes([proto]) + bytes(2)
+            + bytes.fromhex('0a000001 0a000002') + opts + payload)
+
+
+def _ipv6(proto: int, payload: bytes) -> bytes:
+    """An IPv6 header carrying ``payload`` as next header ``proto``."""
+    return (bytes.fromhex('60000000') + struct.pack('>H', len(payload)) + bytes([proto, 64])
+            + b'\x11' * 16 + b'\x22' * 16 + payload)
+
+
+#: Option areas whose last option declares more octets than the area holds.
+OVERRUNS = {
+    # issue #1325: ACK declaring 0x0100 octets, 4 present
+    'HIP ACK': ('pcapkit.protocols.internet.hip', 'HIP', _hip(bytes.fromhex('01c10100 00000000'))),
+    # issue #1325: PadN declaring 3 octets, none present
+    'MH PadN': ('pcapkit.protocols.internet.mh', 'MH',
+                bytes.fromhex('3b010000 00000000 010400000000 0103')),
+    'HOPOPT PadN': ('pcapkit.protocols.internet.hopopt', 'HOPOPT', _ext6(bytes.fromhex('010700000000'))),
+    'IPv6-Opts PadN': ('pcapkit.protocols.internet.ipv6_opts', 'IPv6_Opts', _ext6(bytes.fromhex('010700000000'))),
+    'SCTP INIT parameter': ('pcapkit.protocols.transport.sctp', 'SCTP', _sctp(
+        bytes.fromhex('01000020 00000001 00010000 00010001 00000001 000c0010 00050000'))),
+    # an unassigned kind declaring 12 octets, 8 present
+    'TCP unassigned': ('pcapkit.protocols.transport.tcp', 'TCP', _tcp(bytes.fromhex('4f0caabb ccddeeff'))),
+    # an unassigned option declaring 12 octets, 8 present in a 16-octet area
+    'IPv4 unassigned': ('pcapkit.protocols.internet.ipv4', 'IPv4',
+                        _ipv4(bytes.fromhex('1f0caabb ccddeeff'), ihl=9)),
+}
+
+#: The same option areas with every option complete.
+CONTROLS = {
+    'HIP ACK': ('pcapkit.protocols.internet.hip', 'HIP', _hip(bytes.fromhex('01c10004 00000001'))),
+    'MH PadN': ('pcapkit.protocols.internet.mh', 'MH',
+                bytes.fromhex('3b010000 00000000 0106000000000000')),
+    'HOPOPT PadN': ('pcapkit.protocols.internet.hopopt', 'HOPOPT', _ext6(bytes.fromhex('010400000000'))),
+    'IPv6-Opts PadN': ('pcapkit.protocols.internet.ipv6_opts', 'IPv6_Opts', _ext6(bytes.fromhex('010400000000'))),
+    'SCTP INIT parameter': ('pcapkit.protocols.transport.sctp', 'SCTP', _sctp(
+        bytes.fromhex('0100001c 00000001 00010000 00010001 00000001 000c0008 00050006'))),
+    'TCP unassigned': ('pcapkit.protocols.transport.tcp', 'TCP', _tcp(bytes.fromhex('4f08aabb ccddeeff'))),
+    'IPv4 unassigned': ('pcapkit.protocols.internet.ipv4', 'IPv4',
+                        _ipv4(bytes.fromhex('1f08aabb ccddeeff'), ihl=7)),
+}
+
+#: Each overrun inside a parent layer: (parent module, parent class, frame).
+IN_PARENT = {
+    'HIP ACK': ('pcapkit.protocols.internet.ipv6', 'IPv6', _ipv6(139, OVERRUNS['HIP ACK'][2])),
+    'MH PadN': ('pcapkit.protocols.internet.ipv6', 'IPv6', _ipv6(135, OVERRUNS['MH PadN'][2])),
+    'HOPOPT PadN': ('pcapkit.protocols.internet.ipv6', 'IPv6', _ipv6(0, OVERRUNS['HOPOPT PadN'][2])),
+    'IPv6-Opts PadN': ('pcapkit.protocols.internet.ipv6', 'IPv6', _ipv6(60, OVERRUNS['IPv6-Opts PadN'][2])),
+    'SCTP INIT parameter': ('pcapkit.protocols.internet.ipv4', 'IPv4',
+                            _ipv4(b'', ihl=5, proto=132, payload=OVERRUNS['SCTP INIT parameter'][2])),
+    'TCP unassigned': ('pcapkit.protocols.internet.ipv4', 'IPv4',
+                       _ipv4(b'', ihl=5, proto=6, payload=OVERRUNS['TCP unassigned'][2])),
+    'IPv4 unassigned': ('pcapkit.protocols.link.ethernet', 'Ethernet',
+                        bytes.fromhex('0123456789ab fedcba987654 0800') + OVERRUNS['IPv4 unassigned'][2]),
+}
+
+
+class TestDeclaredLengthOverrun(unittest.TestCase):
+    """Pin the rejection of an option running past the end of its data."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+
+    @staticmethod
+    def _protocol(module: str, name: str) -> type:
+        import importlib
+
+        return getattr(importlib.import_module(module), name)
+
+    def test_an_overrunning_option_raises_protocol_error(self) -> None:
+        from pcapkit.utilities.exceptions import ProtocolError
+
+        for case, (module, name, data) in OVERRUNS.items():
+            with self.subTest(case=case):
+                protocol = self._protocol(module, name)
+                with warnings.catch_warnings():
+                    warnings.simplefilter('ignore')
+                    with self.assertRaisesRegex(ProtocolError, 'runs past the end of the data'):
+                        protocol(data, len(data))
+
+    def test_a_complete_option_still_rebuilds_byte_for_byte(self) -> None:
+        for case, (module, name, data) in CONTROLS.items():
+            with self.subTest(case=case):
+                protocol = self._protocol(module, name)
+                parsed = protocol(data, len(data))
+                self.assertEqual(protocol.from_data(parsed.info).data, data)
+
+    def test_an_overrunning_layer_is_kept_as_captured_by_its_parent(self) -> None:
+        from pcapkit.protocols.misc.raw import Raw
+
+        for case, (module, name, frame) in IN_PARENT.items():
+            with self.subTest(case=case):
+                protocol = self._protocol(module, name)
+                with warnings.catch_warnings():
+                    warnings.simplefilter('ignore')
+                    parsed = protocol(frame, len(frame))
+
+                if name == 'IPv6':
+                    # an extension header is kept as raw extension data
+                    self.assertIn('runs past the end of the data', str(parsed.info.ext.error))
+                else:
+                    # any other layer is kept as a raw payload, as captured
+                    self.assertIsInstance(parsed.payload, Raw)
+                    self.assertIn('runs past the end of the data', str(parsed.payload.info.error))
+                    self.assertEqual(parsed.payload.data, OVERRUNS[case][2])
+                self.assertEqual(protocol.from_data(parsed.info).data, frame)
+
+    def test_an_option_area_declared_past_the_data_still_ends_at_eool(self) -> None:
+        from pcapkit.protocols.internet.ipv4 import IPv4
+
+        # ``ihl`` declares 20 octets of options with none present: the first
+        # option begins at the end of the data, so it is end-of-option-list.
+        data = bytes.fromhex('4a00001800010000400600000a0000010a000002')
+        with warnings.catch_warnings():
+            warnings.simplefilter('ignore')
+            parsed = IPv4(data, len(data))
+        self.assertEqual(parsed.info.src.compressed, '10.0.0.1')
+
+
+if __name__ == '__main__':
+    unittest.main()