diff --git a/pcapkit/protocols/application/http.py b/pcapkit/protocols/application/http.py index 7e7ce42e81..db985e69bf 100644 --- a/pcapkit/protocols/application/http.py +++ b/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 diff --git a/pcapkit/protocols/internet/ipv6_route.py b/pcapkit/protocols/internet/ipv6_route.py index 71a26b1065..3abda59c76 100644 --- a/pcapkit/protocols/internet/ipv6_route.py +++ b/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, diff --git a/tests/protocols/application/test_http_unit.py b/tests/protocols/application/test_http_unit.py index 05c5ddec6c..99d3431ec0 100644 --- a/tests/protocols/application/test_http_unit.py +++ b/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 diff --git a/tests/protocols/internet/test_ipv6_extension_unit.py b/tests/protocols/internet/test_ipv6_extension_unit.py index 79835897b4..231601fd38 100644 --- a/tests/protocols/internet/test_ipv6_extension_unit.py +++ b/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 ]`` 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("", 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