fix(protocols): hold a trial parse's warnings whenever beholder may replace it (#2516)

docs/source/pcapkit/utilities/warnings.rst

@@ -34,12 +34,13 @@ The counts are the same in development mode as outside it:
 there are. Constructing a warning class is not an act of reporting one -- it
 emits nothing and has no side effects.
 
-A next layer captured shorter than its length hint is parsed on trial, and
-replaced with :class:`~pcapkit.protocols.misc.raw.Raw` when its header was not
-captured. The trial's warnings are held back by
+A next layer is parsed on trial, and replaced with
+:class:`~pcapkit.protocols.misc.raw.Raw` when its header was not captured or its
+parse raises. The trial's warnings are held back by
 :class:`~pcapkit.utilities.warnings.hold_warnings`: a replaced layer's are dropped
 on each channel, since they describe a layer the result does not have, and a kept
-layer's are reported as above when the trial ends (:issue:`1580`).
+layer's are reported as above when the trial ends (:issue:`1580`, :issue:`2308`,
+:issue:`2516`).
 
 .. autofunction:: pcapkit.utilities.warnings.warn
 

pcapkit/protocols/protocol.py

@@ -2425,8 +2425,11 @@ def _parse_next_layer(klass: 'Type[ProtocolBase]', file_: 'bytes', length: 'int'
             too when :func:`~pcapkit.corekit.decorators.beholder` replaces the
             layer, as it does around :meth:`_import_next_layer`, since the
             replacement's ``error`` reports the failure (:issue:`2308`); outside
-            it, they are reported as the error leaves. A layer captured at least
-            as long as its hint cannot be replaced, and is parsed holding
+            it, they are reported as the error leaves. Under it, a layer is
+            held however much was captured, since one that raises is replaced
+            at any length (:issue:`2516`), so a kept layer's warnings follow
+            those of the layers it carries. Outside it, a layer captured at
+            least as long as its hint cannot be replaced, and is parsed holding
             nothing, even inside one that holds.
 
             A layer past :data:`~pcapkit.utilities.config.FRAME_LAYER_LIMIT` in its frame is not parsed:
@@ -2460,7 +2463,9 @@ def _parse_next_layer(klass: 'Type[ProtocolBase]', file_: 'bytes', length: 'int'
         captured = min(length, len(file_))
         token = _frame_depth.set(depth)
         try:
-            with hold_warnings(_short_of_hint(klass, captured)) as held:
+            # NOTE: Under ``beholder`` a layer that raises is replaced at any length,
+            # so its warnings are held however much was captured (#2516).
+            with hold_warnings(_beholding.get() or _short_of_hint(klass, captured)) as held:
                 try:
                     next_ = klass(file_, length, **kwargs)  # type: ignore[abstract]
                 except Exception:

tests/protocols/test_trial_parse_full_length_2516_unit.py

@@ -0,0 +1,153 @@
+# -*- coding: utf-8 -*-
+"""GitHub issue #2516: a trial parse that raises at full length and is kept as ``Raw`` warns nothing.
+
+:meth:`~pcapkit.protocols.protocol.ProtocolBase._parse_next_layer` held only a
+next layer captured shorter than its length hint (#1580), but
+:func:`~pcapkit.corekit.decorators.beholder` replaces a layer that raises at any
+length (#2308). Without the fix, a TCP segment to port 443 that starts mid-record
+warned ``packet length < 0`` twice, from TLS's record header and from the SSLv2
+fallback it tries, for a TLS layer the result keeps as ``Raw``. Under
+``beholder`` every layer is now held, and its warnings are dropped with it. A
+kept layer still warns, from the caller's line, after the layers it carries.
+
+:mod:`pcapkit` is imported inside each test, after
+:func:`~tests._support.reimport_once_per_class`.
+
+"""
+
+import inspect
+import struct
+import unittest
+import warnings
+
+from tests._support import reimport_once_per_class
+from tests.utilities._harness import capture
+
+#: Test tier (#1958): exercises the warning machinery, which free-threaded
+#: builds make context-aware.
+__tier__ = 'ALL'
+
+#: Ethernet destination and source addresses.
+MACS = bytes(6) + bytes(range(1, 7))
+#: A 20-octet TCP header, 40000 to 443, PSH|ACK.
+TCP_TO_443 = struct.pack('!HHIIBBHHH', 40000, 443, 1, 1, 5 << 4, 0x18, 65535, 0, 0)
+#: Octets from the middle of a TLS record, as the issue's segment carries.
+MID_RECORD = b'.android' + bytes(range(256)) * 4
+#: The ``error`` of the ``Raw`` that :data:`MID_RECORD`'s TLS is kept as.
+TLS_ERROR = 'TLS: invalid format: unrecognised content type 46'
+#: An SSLv2 CLIENT-HELLO, 2-octet header, cut 4 octets short of its record.
+CUT_HELLO = (b'\x80\x1c' + b'\x01\x00\x02\x00\x03\x00\x00\x00\x10'
+             + b'\x01\x00\x80' + bytes(12))
+#: A 24-octet TCP header whose one option runs past its end.
+TCP_CUT = struct.pack('!HHIIBBHHH', 12345, 80, 1, 0, 6 << 4, 0x02, 4095, 0, 0) + bytes.fromhex('b4060000')
+#: The message IPv4's Security option below reports, for a layer that is kept.
+OPTION_WARNING = 'IPv4: [OptNo 130] invalid format: remaining data'
+#: The warnings of IPv6 carrying a Shim6 header cut to 1 octet, as #1580 pins them.
+SHIM6_CUT_MESSAGES = ['packet length < 0: -1',
+                      'IPv6: extension header declares a length of 8 octet(s) with 1 octet(s) '
+                      'left in the chain; stopping the walk instead of skipping past it']
+
+
+def _ipv4(proto: 'int', body: 'bytes', options: 'bytes' = b'') -> 'bytes':
+    """An Ethernet frame carrying IPv4 with ``options``, then ``body`` as protocol ``proto``."""
+    header = struct.pack('!BBHHHBBH4s4s', 0x45 + len(options) // 4, 0, 20 + len(options) + len(body),
+                         0, 0, 64, proto, 0, bytes([127, 0, 0, 1]), bytes([127, 0, 0, 2]))
+    return MACS + b'\x08\x00' + header + options + body
+
+
+def _ipv6(next_header: 'int', body: 'bytes') -> 'bytes':
+    """IPv6 carrying ``body`` as ``next_header``."""
+    return (struct.pack('!IHBB', 0x60000000, len(body), next_header, 64)
+            + bytes(15) + b'\x01' + bytes(15) + b'\x02' + body)
+
+
+#: IPv4's Security option, its flags running on past the last octet, padded.
+SECURITY = bytes.fromhex('8205ab0000') + bytes(3)
+
+
+class TestTrialParseFullLength(unittest.TestCase):
+    """Pin that a full-length layer replaced by ``beholder`` reports nothing."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+
+    @staticmethod
+    def _parse(klass: 'str', data: 'bytes', action: 'str' = 'always') -> 'tuple[object, list[warnings.WarningMessage]]':
+        """Parse ``data`` as ``klass``, TCP or Ethernet, with the warnings it reports."""
+        from pcapkit.protocols.link.ethernet import Ethernet
+        from pcapkit.protocols.transport.tcp import TCP
+
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter(action)
+            parsed = {'TCP': TCP, 'Ethernet': Ethernet}[klass](data, len(data))
+        return parsed, caught
+
+    def test_mid_record_segment_is_raw_and_warns_nothing(self) -> None:
+        """The issue's segment: TLS and its SSLv2 fallback both reject it, on both channels."""
+        from pcapkit.protocols.misc.raw import Raw
+        from pcapkit.utilities.logging import logger
+
+        with capture(logger) as recorder:
+            parsed, caught = self._parse('TCP', TCP_TO_443 + MID_RECORD)
+        self.assertIs(type(parsed.payload), Raw)  # type: ignore[attr-defined]
+        self.assertEqual(parsed.payload.info.error, TLS_ERROR)  # type: ignore[attr-defined]
+        self.assertEqual([str(item.message) for item in caught], [])
+        self.assertEqual([record.getMessage() for record in recorder.records
+                          if record.getMessage().startswith('packet length')], [])
+
+    def test_error_filter_reports_the_tls_rejection(self) -> None:
+        """Under ``error``, the ``Raw`` names TLS's rejection, not a warning of the trial."""
+        parsed, caught = self._parse('TCP', TCP_TO_443 + MID_RECORD, 'error')
+        self.assertEqual(str(parsed.protochain), 'TCP:Raw')  # type: ignore[attr-defined]
+        self.assertEqual(parsed.payload.info.error, TLS_ERROR)  # type: ignore[attr-defined]
+        self.assertEqual(caught, [])
+
+    def test_rejected_sslv2_fallback_warns_nothing(self) -> None:
+        """A cut SSLv2 hello passes TLS's header read, then warns only in the fallback it fails."""
+        parsed, caught = self._parse('TCP', TCP_TO_443 + CUT_HELLO)
+        self.assertEqual(str(parsed.protochain), 'TCP:Raw')  # type: ignore[attr-defined]
+        self.assertEqual(parsed.payload.info.error,  # type: ignore[attr-defined]
+                         'TLS: invalid format: unrecognised content type 128')
+        self.assertEqual([str(item.message) for item in caught], [])
+
+    def test_full_length_layer_that_raises_warns_nothing(self) -> None:
+        """TCP past its hint raises under IPv4's and IPv6's own ``beholder`` alike."""
+        from pcapkit.protocols.misc.raw import Raw
+
+        for name, frame in (('IPv4', _ipv4(6, TCP_CUT)),
+                            ('IPv6', MACS + b'\x86\xdd' + _ipv6(6, TCP_CUT))):
+            with self.subTest(carrier=name):
+                parsed, caught = self._parse('Ethernet', frame)
+                self.assertEqual(str(parsed.protochain), f'Ethernet:{name}:TCP')  # type: ignore[attr-defined]
+                self.assertIs(type(parsed.payload.payload), Raw)  # type: ignore[attr-defined]
+                self.assertEqual([str(item.message) for item in caught], [])
+
+    def test_well_formed_tls_record_warns_nothing(self) -> None:
+        """Control: a whole TLS record on 443 parses as TLS, as before, reporting nothing."""
+        parsed, caught = self._parse('TCP', TCP_TO_443 + b'\x17\x03\x03\x00\x04abcd')
+        self.assertEqual(str(parsed.protochain), 'TCP:TLS 1.2')  # type: ignore[attr-defined]
+        self.assertEqual(caught, [])
+
+    def test_kept_layer_still_warns_from_the_callers_line(self) -> None:
+        """Control: a kept IPv4 is held now, and reports its warning as it did."""
+        from pcapkit.protocols.link.ethernet import Ethernet
+        from pcapkit.utilities.warnings import ProtocolWarning
+
+        frame = _ipv4(17, struct.pack('!HHHH', 40000, 40001, 12, 0) + b'abcd', SECURITY)
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter('always')
+            line = inspect.currentframe().f_lineno + 1  # type: ignore[union-attr]
+            parsed = Ethernet(frame, len(frame))
+        self.assertEqual(str(parsed.protochain), 'Ethernet:IPv4:UDP:Raw')
+        self.assertEqual([(item.category, str(item.message), item.filename, item.lineno) for item in caught],
+                         [(ProtocolWarning, OPTION_WARNING, __file__, line)])
+
+    def test_kept_layer_reports_after_the_layers_it_carries(self) -> None:
+        """Each held layer reports as its own parse ends, so IPv4's warning follows IPv6's."""
+        shim6_cut = _ipv6(140, b'\x00')
+        _, caught = self._parse('Ethernet', _ipv4(41, shim6_cut, SECURITY))
+        self.assertEqual([str(item.message) for item in caught], SHIM6_CUT_MESSAGES + [OPTION_WARNING])
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/protocols/test_trial_parse_warnings_1580_unit.py

@@ -16,9 +16,10 @@
 with the layer. A layer that is kept reports them from the caller's line when its
 trial ends, inside the same guard -- so under an ``error`` filter every chain is
 the one parsed holding nothing -- and a cut header that stays in the chain still
-warns (#1470). Any other layer is parsed holding nothing. Holding leaves the
-warnings filters and every ``__warningregistry__`` alone, lets a
-:exc:`BaseException` through untouched, and survives a block left out of order.
+warns (#1470). Under ``beholder`` every layer is held (#2516); any other layer is
+parsed holding nothing. Holding leaves the warnings filters and every
+``__warningregistry__`` alone, lets a :exc:`BaseException` through untouched, and
+survives a block left out of order.
 
 :mod:`pcapkit` is imported inside each test, after
 :func:`~tests._support.reimport_once_per_class`.
@@ -254,14 +255,17 @@ def test_holds_nest(self) -> None:
         self.assertEqual({item.filename for item in caught}, {__file__})
 
     def test_layer_that_cannot_be_replaced_holds_nothing(self) -> None:
-        """TCP of 24 octets is past its hint, so even a discarding block cannot drop its warnings.
+        """TCP of 24 octets, past its hint and outside ``beholder``, holds nothing.
 
-        It holds nothing, so they are reported as they are raised, and stay
-        reported when the parse then raises and is kept as ``Raw`` (#2308).
+        Nothing replaces it when it raises, so even a discarding block cannot
+        drop its warnings: they are reported as they are raised. Under
+        ``beholder`` the same cut is held and dropped with the layer (#2516).
         """
         from pcapkit.protocols.link.ethernet import Ethernet
         from pcapkit.protocols.misc.raw import Raw
         from pcapkit.protocols.protocol import ProtocolBase, _short_of_hint
+        from pcapkit.protocols.transport.tcp import TCP
+        from pcapkit.utilities.exceptions import ProtocolError
         from pcapkit.utilities.warnings import hold_warnings
 
         class NeedsAnInstance:
@@ -275,14 +279,21 @@ def __length_hint__(self) -> 'int':
                          [True, False, False, True])
 
         frame = FRAMES[0][:62]  # TCP cut to 24 octets, past its hint, raises
+        segment = frame[38:]
         with warnings.catch_warnings(record=True) as caught:
             warnings.simplefilter('always')
-            with hold_warnings() as outer:
-                parsed = ProtocolBase._parse_next_layer(Ethernet, frame, len(frame))  # pylint: disable=protected-access
-                outer.discard = True
+            with self.assertRaises(ProtocolError):
+                with hold_warnings() as outer:
+                    outer.discard = True
+                    ProtocolBase._parse_next_layer(TCP, segment, len(segment))  # pylint: disable=protected-access
+        self.assertEqual(self._schema(caught), ['packet length < 0: -95'])
+
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter('always')
+            parsed = ProtocolBase._parse_next_layer(Ethernet, frame, len(frame))  # pylint: disable=protected-access
         self.assertEqual(str(parsed.protochain), 'Ethernet:IPv4:TCP')
         self.assertIsInstance(parsed.payload.payload, Raw)
-        self.assertEqual(self._schema(caught), ['packet length < 0: -95'])
+        self.assertEqual(self._schema(caught), [])
 
     def test_error_filter_chains_match_holding_nothing(self) -> None:
         """Under ``error``, every cut of :data:`FRAMES` parses to the chain it has holding nothing.