fix(protocols): rebuild a cut standalone header with the values its parse read (#2761)

docs/source/contributing/conventions/test-coverage.rst

@@ -18,7 +18,8 @@ Round-Trip Invariants
 
 **A round trip either reproduces its input exactly or is a recorded defect.** There is
 no tolerance in between: a cycle that comes back with different octets is a bug, even
-when both halves are internally consistent.
+when both halves are internally consistent, and so is one whose parsed values differ:
+``from_data(info).info`` equals ``info``, for a cut header too (:issue:`2761`).
 
 Coverage is **enumerated from the registries rather than hand-picked**, so a code
 registered tomorrow gets a case tomorrow. The registry-driven suites that carry a

pcapkit/protocols/protocol.py

@@ -850,7 +850,10 @@ def pack(self, **kwargs: 'Any') -> 'bytes':
             short, cuts the header back to the octets captured, through its own
             ``__short_read__`` and :meth:`Schema._cut_short_read
             <pcapkit.protocols.schema.Schema._cut_short_read>`, and
-            returns only those (:issue:`2703`).
+            returns only those (:issue:`2703`). :meth:`__post_init__` then
+            drops that header, so that :meth:`unpack` parses the octets back
+            and the rebuild holds the values its parse read, not those
+            :meth:`make` filled in past the cut (:issue:`2761`).
 
         .. versionchanged:: 1.5.0
 
@@ -901,6 +904,12 @@ def unpack(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> '_PT':
             cannot consume. The header is then replaced by the parsed one
             (:issue:`1312`).
 
+            A header :meth:`from_data` rebuilds from a short read is parsed
+            from the octets captured instead, as quietly as the header made
+            was read, and given the ``__short_read__`` of its data model before
+            :meth:`read` reads it, as the record of a cut those octets may not
+            show (:issue:`2761`).
+
         Raises:
             ProtocolError: If the made header holds raw items and its packed
                 octets do not parse, either as a schema or by :meth:`read`.
@@ -915,7 +924,15 @@ def unpack(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> '_PT':
 
         """
         packet = kwargs.get('__packet__', {})  # packet data
-        if cast('Optional[_ST]', self.__header__) is None:
+        if cast('Optional[_ST]', self.__header__) is None and self.__short_read__ is not None:
+            # NOTE: A rebuilt short read is parsed as quietly as its made header was read,
+            # and keeps the record of a cut its octets may not show (#2711, #2761).
+            with hold_warnings() as held:
+                held.discard = True
+                header = cast('_ST', self.__schema__.unpack(self._file, length, packet))  # type: ignore[call-arg,misc]
+            header.__dict__['__short_read__'] = self.__short_read__
+            self.__header__ = header
+        elif cast('Optional[_ST]', self.__header__) is None:
             self.__header__ = cast('_ST', self.__schema__.unpack(self._file, length, packet))  # type: ignore[call-arg,misc]
         elif self._make_raw and _holds_raw(self.__header__):
             # NOTE: A schema may accept octets that :meth:`read` then rejects,
@@ -1202,7 +1219,10 @@ def from_data(cls, data: '_PT | dict[str, Any]', **kwargs: 'Any') -> 'Self':
         A header the capture cut short, as ``__short_read__`` records, is cut
         back before it is read, and read at the width it declares, so the
         rebuild parses and verifies the octets captured as the capture did
-        (:issue:`2703`). It keeps the ``Raw`` next layer its data model holds,
+        (:issue:`2703`), holding the values its parse read from them, not those
+        :meth:`make` fills in past the cut (:issue:`2761`). Inside a payload
+        :meth:`_make_payload` rebuilds, which gives only its octets, the header
+        made is read instead. It keeps the ``Raw`` next layer its data model holds,
         which a parse reading the header short of that width leaves it
         (:issue:`2717`).
 
@@ -1490,6 +1510,10 @@ def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None,
             uncaptured = vars(self).pop('_uncaptured', None)
             if uncaptured is not None and length is None:
                 length = len(_data) + uncaptured
+            # NOTE: Parsed back from the octets captured, not read from what ``make`` filled
+            # in past the cut (#2761); a payload ``_make_payload`` rebuilds gives its octets.
+            if uncaptured is not None and not rebuilding.get():
+                self.__header__ = None  # type: ignore[assignment]
         else:
             _data = file if isinstance(file, bytes) else file.read(length)  # type: ignore[arg-type]
 

tests/protocols/internet/test_cut_ext_header_values_2761_unit.py

@@ -0,0 +1,101 @@
+# -*- coding: utf-8 -*-
+"""A standalone IPv6 extension header cut short rebuilds with the values its parse read.
+
+GitHub issue #2761: built on its own octets and cut short, an extension header rebuilt
+to the octets captured, but ``from_data(info).info`` differed from ``info``. The rebuild
+was read from the header ``make()`` built, whose Reserved fields are zero-filled past the
+cut and whose option area a ``PadN`` pads out to the length the header declares; the
+parse read the octets captured. Each field the issue names is checked against those
+octets, in the parse and in its rebuild from the data model and from the :obj:`dict`,
+but ``IPv6_Ext``'s ``length``, which :file:`tests/protocols/test_cut_header_values_2761_unit.py`
+records as still divergent.
+
+The octets are built here; this module is unit-tier and reads no capture file.
+
+"""
+
+import unittest
+import warnings
+
+from tests._support import reimport_once_per_class
+
+#: Test tier (#1958): hand-picked parse -> rebuild round trips.
+__tier__ = 'ALL'
+
+#: A Type 2 Routing header (RFC 6275, section 6.4) with a Reserved field of 01020304
+#: and a Home Address of ::1, the issue's own.
+ROUTE = bytes([59, 2, 2, 1, 1, 2, 3, 4]) + bytes(15) + b'\x01'
+
+#: A Fragment header (RFC 8200, section 4.5), its Reserved octet the second.
+FRAG = bytes([17, 0, 0, 8, 0x12, 0x34, 0x56, 0x78])
+
+#: A 16-octet Hop-by-Hop or Destination Options header: Router Alert, then a PadN of
+#: eight, whose Option Type is its seventh octet.
+OPTIONS = bytes([17, 1, 5, 2, 0, 0, 1, 8]) + bytes(8)
+
+#: The cuts of :data:`OPTIONS` the issue names; cut inside its Opt Data Len, the Router
+#: Alert is refused.
+OPTION_CUTS = (1, 2, 4, 5, 6)
+
+
+class CutExtensionHeaderValuesTests(unittest.TestCase):
+    """The issue's fields read the octets captured, parsed and rebuilt."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+        # a cut header warns that its length runs past the data, which is the point
+        catcher = warnings.catch_warnings()
+        catcher.__enter__()  # pylint: disable=unnecessary-dunder-call
+        self.addCleanup(catcher.__exit__, None, None, None)
+        warnings.simplefilter('ignore')
+
+    def rebuilds(self, layer: 'object', raw: bytes) -> 'list[object]':
+        """``layer`` rebuilt from its data model and its dict, each giving ``raw`` and its info."""
+        out = []
+        for data in (layer.info, layer.info.to_dict()):  # type: ignore[attr-defined]
+            rebuilt = type(layer).from_data(data)  # type: ignore[attr-defined]
+            self.assertEqual(rebuilt.data, raw)
+            self.assertEqual(rebuilt.info, layer.info)  # type: ignore[attr-defined]
+            out.append(rebuilt)
+        return out
+
+    def test_ipv6_route_reserved(self) -> None:
+        from pcapkit.protocols.internet.ipv6_route.protocol import IPv6_Route
+
+        for count in range(1, 8):
+            with self.subTest(captured=count):
+                raw = ROUTE[:count]
+                layer = IPv6_Route(raw, count)
+                self.assertEqual(layer.info.reserved, ROUTE[4:count])
+                for rebuilt in self.rebuilds(layer, raw):
+                    self.assertEqual(rebuilt.info.reserved, ROUTE[4:count])
+
+    def test_ipv6_frag_reserved_octet(self) -> None:
+        from pcapkit.protocols.internet.ipv6_frag.protocol import IPv6_Frag
+
+        for count in range(1, 3):
+            with self.subTest(captured=count):
+                raw = FRAG[:count]
+                layer = IPv6_Frag(raw, count)
+                self.assertEqual(layer.info.reserved_octet, FRAG[1:count])
+                for rebuilt in self.rebuilds(layer, raw):
+                    self.assertEqual(rebuilt.info.reserved_octet, FRAG[1:count])
+
+    def test_hopopt_and_ipv6_opts_options(self) -> None:
+        # No octet before the seventh is a PadN, so neither parse nor rebuild holds one.
+        from pcapkit.const.ipv6.option import Option
+        from pcapkit.protocols.internet.hopopt.protocol import HOPOPT
+        from pcapkit.protocols.internet.ipv6_opts.protocol import IPv6_Opts
+
+        for klass in (HOPOPT, IPv6_Opts):
+            for count in OPTION_CUTS:
+                with self.subTest(protocol=klass.__name__, captured=count):
+                    raw = OPTIONS[:count]
+                    layer = klass(raw, count)
+                    self.assertNotIn(Option.PadN, list(layer.info.options))
+                    for rebuilt in self.rebuilds(layer, raw):
+                        self.assertEqual(list(rebuilt.info.options), list(layer.info.options))
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/protocols/test_cut_header_values_2761_unit.py

@@ -0,0 +1,190 @@
+# -*- coding: utf-8 -*-
+"""A header the capture cut short rebuilds with the values its parse read.
+
+GitHub issue #2761: a header cut short rebuilt to the octets captured, but
+``from_data(info).info`` differed from ``info``. The rebuild was read from the header
+``make()`` built, which holds what it filled in past the cut: the zeros
+:meth:`Protocol._fill_short_read <pcapkit.protocols.protocol.Protocol._fill_short_read>`
+gives a reserved field, a ``PadN`` padding an option area out, a nested schema at its
+full width. The parse read the octets captured. The rebuild now parses those octets
+back, so a round trip reproduces the parsed values as well as the octets.
+
+Every concrete protocol class, found by walking the protocol classes rather than listed,
+is built from its ``make()`` defaults and cut at every octet of that header, and so are
+the octets below, which carry the values its defaults leave out. The parse and its
+rebuild from the data model and from the :obj:`dict` give the same octets and the same
+data model.
+
+The octets are built here; this module is unit-tier and reads no capture file.
+
+"""
+from __future__ import annotations
+
+import importlib
+import inspect
+import pkgutil
+import unittest
+import warnings
+from typing import TYPE_CHECKING
+
+from tests._support import reimport_once_per_class
+
+if TYPE_CHECKING:
+    from typing import Any, Iterator
+
+#: Test tier (#1958): a sweep that cuts every protocol class at every offset.
+__tier__ = 'LATEST'
+
+#: ``make()`` keywords for a class whose defaults build nothing it reads back: a Query
+#: with a Max Resp Time of zero is IGMPv1's (RFC 3376, section 7.1), and HTTP builds
+#: through the member its ``version`` names.
+MAKE = {
+    'pcapkit.protocols.application.http.protocol.HTTP': {'version': 2},
+    'pcapkit.protocols.internet.igmpv2.protocol.IGMPv2': {'max_resp_time': 100},
+}
+
+#: Whole packets cut at every octet besides those ``make()`` builds, for a class whose
+#: defaults build nothing, or leave empty the fields a cut reads short.
+OCTETS = {
+    'pcapkit.protocols.application.ftp.protocol.FTP': (b'XYZW arg\r\n',),
+    'pcapkit.protocols.application.httpv1.protocol.HTTPv1': (b'GET / HTTP/1.1\r\n\r\n',),
+    # an NGSetupRequest in aligned PER, which no cut of leaves decodable
+    'pcapkit.protocols.application.ngap.protocol.NGAP': (
+        bytes.fromhex('00150036000004001b00080002f839100001020052400d0500706361706b69742d676e62')
+        + bytes.fromhex('0066000d00000000010002f839000000080015400140'),),
+    # Hop-by-Hop and Destination Options: Router Alert, then a PadN of eight
+    'pcapkit.protocols.internet.hopopt.protocol.HOPOPT': (bytes([17, 1, 5, 2, 0, 0, 1, 8]) + bytes(8),),
+    'pcapkit.protocols.internet.ipv6_opts.protocol.IPv6_Opts': (bytes([17, 1, 5, 2, 0, 0, 1, 8]) + bytes(8),),
+    # a Type 2 Routing header with a non-zero Reserved field
+    'pcapkit.protocols.internet.ipv6_route.protocol.IPv6_Route': (
+        bytes([59, 2, 2, 1, 1, 2, 3, 4]) + bytes(15) + b'\x01',),
+    # a Report of two group records
+    'pcapkit.protocols.internet.igmpv3.protocol.IGMPv3': (
+        bytes.fromhex('2200c7e100000002') + bytes.fromhex('01000002ef0101010a0000010a000002')
+        + bytes.fromhex('02000003ef0202020a0000030a0000040a000005'),),
+    # a Binding Refresh Request carrying a PadN
+    'pcapkit.protocols.internet.mh.protocol.MH': (
+        bytes([59, 1, 5, 0, 0, 0, 0, 1, 0x80, 0, 0, 0x10, 1, 2, 0, 0]),),
+}
+
+#: Classes no octets are cut for, and why.
+NOT_CUT = {
+    'pcapkit.protocols.misc.checksum.ip_pseudo.protocol.IP_Pseudo':
+        'is built from an IP layer, and is no header on the wire',
+    'pcapkit.protocols.misc.pcap.frame.protocol.Frame':
+        "reads its record by the global header of its savefile",
+    'pcapkit.protocols.misc.pcapng.protocol.PCAPNG':
+        'reads its block by the section and interfaces before it',
+}
+
+#: Cuts whose rebuild still reads a value its parse did not, by class and the octets
+#: captured, in hex: the keys that differ, and why. An entry whose cut stops differing
+#: so fails, as does one the sweep never reaches.
+KNOWN = {
+    ('pcapkit.protocols.internet.ipv6_ext.protocol.IPv6_Ext', '11'): (
+        ('length',),
+        # Protocol.__post_init__ reads the rebuild at the width make() packs (#2703),
+        # past the octet IPv6_Ext.read was given, and an overrun is that long (#2761).
+        'IPv6_Ext.read takes an overrunning header for the width it is given'),
+}
+
+
+def _concrete() -> 'Iterator[tuple[str, type]]':
+    """Every concrete protocol class, by qualified name."""
+    import pcapkit.protocols as package
+    from pcapkit.protocols.protocol import ProtocolBase
+
+    for info in pkgutil.walk_packages(package.__path__, 'pcapkit.protocols.'):
+        if 'NotImplemented' not in info.name:
+            importlib.import_module(info.name)
+    seen = set()  # type: set[type]
+    pending = [ProtocolBase]  # type: list[type]
+    found = {}  # type: dict[str, type]
+    while pending:
+        klass = pending.pop()
+        if klass in seen:
+            continue
+        seen.add(klass)
+        pending.extend(klass.__subclasses__())
+        if klass.__module__.startswith('pcapkit.protocols.') and not inspect.isabstract(klass):
+            found[f'{klass.__module__}.{klass.__qualname__}'] = klass
+    for name in sorted(found):
+        yield name, found[name]
+
+
+def _differing(parsed: 'Any', rebuilt: 'Any') -> 'tuple[str, ...]':
+    """The keys whose values ``parsed`` and ``rebuilt`` do not share, in order."""
+    keys = list(dict.fromkeys([*parsed, *rebuilt]))
+    return tuple(key for key in keys if key not in parsed or key not in rebuilt or parsed[key] != rebuilt[key])
+
+
+class CutHeaderValuesTests(unittest.TestCase):
+    """Every protocol class rebuilds the values its parse read, at every cut."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+        # a cut header warns that its length runs past the data, which is the point
+        catcher = warnings.catch_warnings()
+        catcher.__enter__()  # pylint: disable=unnecessary-dunder-call
+        self.addCleanup(catcher.__exit__, None, None, None)
+        warnings.simplefilter('ignore')
+
+    def packets(self, name: str, klass: type) -> 'Iterator[bytes]':
+        """The whole packets of ``klass`` to cut: its ``make()`` header, if it builds, and its octets."""
+        from pcapkit.utilities.exceptions import BaseError, CaughtByCaller, ProtocolError
+
+        try:
+            with CaughtByCaller(ProtocolError):
+                yield klass(**MAKE.get(name, {})).data
+        except BaseError:
+            pass  # its defaults build nothing, so its octets are below
+        yield from OCTETS.get(name, ())
+
+    def cuts(self) -> 'Iterator[tuple[str, bytes, Any]]':
+        """Every class, the octets captured, and their parse, for each cut a class reads."""
+        from pcapkit.utilities.exceptions import BaseError, CaughtByCaller, ProtocolError
+
+        for name, klass in _concrete():
+            if name in NOT_CUT:
+                continue
+            for whole in self.packets(name, klass):
+                for count in range(1, len(whole)):
+                    raw = whole[:count]
+                    try:
+                        with CaughtByCaller(ProtocolError):
+                            layer = klass(raw, count)
+                    except BaseError:
+                        continue  # refused, as a cut header may be
+                    yield name, raw, layer
+
+    def test_every_class_is_cut(self) -> None:
+        # a class with neither a header nor octets to cut is named, with its reason
+        from pcapkit.utilities.exceptions import BaseError, CaughtByCaller, ProtocolError
+
+        names = dict(_concrete())
+        self.assertLessEqual(set(NOT_CUT) | set(MAKE) | set(OCTETS), set(names))
+        for name, klass in names.items():
+            with self.subTest(protocol=name):
+                try:
+                    with CaughtByCaller(ProtocolError):
+                        built = klass(**MAKE.get(name, {})) is not None
+                except (BaseError, TypeError):
+                    built = False
+                self.assertEqual(name in NOT_CUT, not built and name not in OCTETS)
+
+    def test_every_cut_rebuilds_its_values(self) -> None:
+        reached = set()  # type: set[tuple[str, str]]
+        for name, raw, layer in self.cuts():
+            key = (name, raw.hex())
+            fields = KNOWN.get(key, ((),))[0]
+            reached.update({key} & set(KNOWN))
+            for path, data in (('model', layer.info), ('dict', layer.info.to_dict())):
+                with self.subTest(protocol=name, captured=raw.hex(), path=path):
+                    rebuilt = type(layer).from_data(data)
+                    self.assertEqual(rebuilt.data, raw)
+                    self.assertEqual(_differing(layer.info, rebuilt.info), fields)
+        self.assertEqual(reached, set(KNOWN))
+
+
+if __name__ == '__main__':
+    unittest.main()