protocols: real routing type in IPv6-Route diagnostics, and fix HTTP's drained explicit version= buffer

pcapkit/protocols/application/http.py

@@ -100,7 +100,12 @@ def read(self, length: 'Optional[int]' = None, *,
             else:
                 raise ProtocolError(f"invalid HTTP version: {version}")
 
-            http = protocol(self._file, length, **kwargs)
+            try:
+                http = protocol(self._data, length, **kwargs)
+            except ProtocolError:
+                raise
+            except ValueError as error:
+                raise ProtocolError(f'HTTP/{version}: invalid format') from error
 
         self._version = http.version
         self._length = http.length

pcapkit/protocols/internet/ipv6_route.py

@@ -459,7 +459,7 @@ def _read_data_type_src(self, schema: 'Schema_SourceRoute', *, header: 'Schema_I
 
         """
         if (header.length - 8) % 16 != 0:
-            raise ProtocolError(f'{self.alias} [TypeNo {type}]: invalid format')
+            raise ProtocolError(f'{self.alias}: [TypeNo {header.type}] invalid format')
 
         ipv6_route = Data_SourceRoute(
             next=header.next,
@@ -503,7 +503,7 @@ def _read_data_type_2(self, schema: 'Schema_Type2', *, header: 'Schema_IPv6_Rout
 
         """
         if header.length != 24:
-            raise ProtocolError(f'{self.alias}: [TypeNo {type}] invalid format')
+            raise ProtocolError(f'{self.alias}: [TypeNo {header.type}] invalid format')
 
         ipv6_route = Data_Type2(
             next=header.next,
@@ -543,7 +543,7 @@ def _read_data_type_rpl(self, schema: 'Schema_RPL', *, header: 'Schema_IPv6_Rout
 
         """
         if header.length % 16 != 0:
-            raise ProtocolError(f'{self.alias}: [TypeNo {type}] invalid format')
+            raise ProtocolError(f'{self.alias}: [TypeNo {header.type}] invalid format')
 
         ipv6_route = Data_RPL(
             next=header.next,

tests/protocols/application/test_http_unit.py

@@ -144,6 +144,74 @@ def make(**kwargs):
             with self.assertRaises(ProtocolError):
                 HTTP._guess_version(http, 9)
 
+    def test_http_read_explicit_version_uses_same_buffer_as_guess(self) -> None:
+        """Regression test for GH-447.
+
+        The explicit ``version=`` path passed ``self._file`` -- already
+        drained by the outer read -- instead of ``self._data``, so it built
+        the sub-protocol from a short or empty buffer while
+        ``_guess_version``, built from the same bytes, worked fine. This pins
+        the fix by asserting *which* buffer object reaches the sub-protocol
+        constructor, for both ``version=1`` and ``version=2``, rather than
+        merely that a call succeeds.
+        """
+        from pcapkit.protocols.application.http import HTTP
+
+        class RecordingHTTP:
+            version = '1.1'
+            length = 5
+            info = 'recorded-info'
+
+            def __init__(self, file: object, length: int, **kwargs: object) -> None:
+                self.received = file
+
+        http = object.__new__(HTTP)
+        http._data = b'GET / HTTP/1.1\r\n\r\n'
+        # Stands in for a stream already advanced by the outer read; passing
+        # this instead of ``_data`` is exactly the GH-447 defect.
+        http._file = object()
+        http.__cached__ = {}
+
+        with mock.patch('pcapkit.protocols.application.httpv1.HTTP', RecordingHTTP):
+            http.read(version=1)
+        self.assertIs(http._http.received, http._data)
+
+        with mock.patch('pcapkit.protocols.application.httpv2.HTTP', RecordingHTTP):
+            http.read(version=2)
+        self.assertIs(http._http.received, http._data)
+
+    def test_http_read_explicit_version_1_matches_guess_on_real_bytes(self) -> None:
+        """The literal GH-447 reproduction: a real HTTP/1.1 request parses
+        identically whether the version is guessed or given explicitly."""
+        import io
+
+        from pcapkit.protocols.application.http import HTTP
+
+        raw = b'GET /index.html HTTP/1.1\r\nHost: example.com\r\n\r\n'
+
+        guessed = HTTP(io.BytesIO(raw), len(raw))
+        explicit = HTTP(io.BytesIO(raw), len(raw), version=1)
+
+        self.assertEqual(guessed.alias, 'HTTP/1.1')
+        self.assertEqual(explicit.alias, 'HTTP/1.1')
+        self.assertEqual(explicit.length, guessed.length)
+        self.assertEqual(explicit.info, guessed.info)
+
+    def test_http_read_explicit_version_wraps_malformed_payload(self) -> None:
+        """The second half of GH-447: a payload that fails to parse on the
+        explicit path must surface as a chained :class:`ProtocolError`, not a
+        bare :class:`ValueError` a caller cannot catch as a protocol error."""
+        import io
+
+        from pcapkit.protocols.application.http import HTTP
+        from pcapkit.utilities.exceptions import ProtocolError
+
+        bad = b'not a valid http request at all'
+
+        with self.assertRaises(ProtocolError) as ctx:
+            HTTP(io.BytesIO(bad), len(bad), version=1)
+        self.assertIsInstance(ctx.exception.__cause__, ValueError)
+
     def test_http_make_data_delegates_to_httpv1(self) -> None:
         from pcapkit.const.http.method import Method
         from pcapkit.corekit.multidict import OrderedMultiDict

tests/protocols/internet/test_ipv6_extension_unit.py

@@ -375,6 +375,51 @@ def test_ipv6_route_readers_and_constructors_cover_registered_types(self) -> Non
         )
         self.assertEqual(rpl_from_data.cmpr_i, 1)
 
+    def test_ipv6_route_read_data_type_errors_report_real_routing_type(self) -> None:
+        """Regression test for GH-442.
+
+        The three ``_read_data_type_*`` diagnostics interpolated a bare
+        ``{type}`` -- which resolves to the *builtin* ``type``, since none of
+        these methods bind a ``type`` parameter -- so every message read
+        ``[TypeNo <class 'type'>]`` instead of the routing type number that
+        ``header.type`` already carries. Asserting only ``ProtocolError`` was
+        raised (as the pre-existing coverage does) would not have caught
+        this, so this checks the rendered message text directly.
+        """
+        from pcapkit.const.ipv6.routing import Routing
+        from pcapkit.const.reg.transtype import TransType
+        from pcapkit.protocols.internet.ipv6_route import IPv6_Route
+        from pcapkit.protocols.schema.internet import ipv6_route as route_schema
+        from pcapkit.utilities.exceptions import ProtocolError
+
+        proto = object.__new__(IPv6_Route)
+        route_type = Routing.get(250)
+        header = types.SimpleNamespace(next=TransType.TCP, length=1, type=route_type, seg_left=0)
+        expected = f'{proto.alias}: [TypeNo {route_type}] invalid format'
+
+        with self.assertRaises(ProtocolError) as src_ctx:
+            proto._read_data_type_src(route_schema.SourceRoute(ip=[]), header=header)
+        self.assertEqual(str(src_ctx.exception), expected)
+
+        with self.assertRaises(ProtocolError) as type2_ctx:
+            proto._read_data_type_2(route_schema.Type2(ip='2001:db8::2'), header=header)
+        self.assertEqual(str(type2_ctx.exception), expected)
+
+        rpl_schema = route_schema.RPL(cmpr_i=0, cmpr_e=0, pad={'pad_len': 0}, addresses=[])
+        with self.assertRaises(ProtocolError) as rpl_ctx:
+            proto._read_data_type_rpl(rpl_schema, header=header)
+        self.assertEqual(str(rpl_ctx.exception), expected)
+
+        # The three messages must keep carrying the substrings PR #440's
+        # round-trip table (tests/protocols/test_option_roundtrip_unit.py)
+        # matches on: the alias, the bracket, and "invalid format".
+        for message in (str(src_ctx.exception), str(type2_ctx.exception), str(rpl_ctx.exception)):
+            self.assertIn('IPv6-Route', message)
+            self.assertIn('[TypeNo', message)
+            self.assertIn('invalid format', message)
+            self.assertIn(str(route_type), message)
+            self.assertNotIn("<class 'type'>", message)
+
     def test_ipv6_route_read_make_registry_and_property_edges(self) -> None:
         from pcapkit.const.ipv6.routing import Routing
         from pcapkit.const.reg.transtype import TransType