From c2f9288791cf0ecf04a4ae7ad452010a89a75699 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Tue, 6 Oct 2026 15:03:00 -0400 Subject: [PATCH] fix(protocols): Quick-Start option readers use schema.func, which exists only under TYPE_CHECKING - _read_opt_qs in IPv4, HOPOPT and IPv6-Opts now reads the function from flags['func'], as it already reads rate. schema.func is set only by _QSOption.post_process, so a schema built by _make_opt_qs raised AttributeError when re-read. - quick_start_data_selector in schema/internet/hopopt.py and ipv6_opts.py sized the nested suboption as 5 octets against the 8-octet option, which mis-decoded the nonce; it now uses 8. - Drop the three Quick-Start EXPECTED_FAILURES entries; two reader tests that injected schema.func now set flags['func'] instead. Closes #1091 --- pcapkit/protocols/internet/hopopt.py | 2 +- pcapkit/protocols/internet/ipv4.py | 2 +- pcapkit/protocols/internet/ipv6_opts.py | 2 +- pcapkit/protocols/schema/internet/hopopt.py | 6 +- .../protocols/schema/internet/ipv6_opts.py | 6 +- tests/protocols/internet/test_ipv4_unit.py | 3 +- .../internet/test_ipv6_extension_unit.py | 3 +- .../internet/test_quick_start_func_unit.py | 105 ++++++++++++++++++ tests/protocols/test_option_roundtrip_unit.py | 56 ++-------- 9 files changed, 130 insertions(+), 55 deletions(-) create mode 100644 tests/protocols/internet/test_quick_start_func_unit.py diff --git a/pcapkit/protocols/internet/hopopt.py b/pcapkit/protocols/internet/hopopt.py index 0cb6b5a847..0b7bff9a08 100644 --- a/pcapkit/protocols/internet/hopopt.py +++ b/pcapkit/protocols/internet/hopopt.py @@ -934,7 +934,7 @@ def _read_opt_qs(self, schema: 'Schema_QuickStartOption', *, options: 'Option') if schema.len != 6: raise ProtocolError(f'{self.alias}: [OptNo {schema.type}] invalid format') - func = schema.func + func = Enum_QSFunction.get(schema.flags['func']) if func == Enum_QSFunction.Quick_Start_Request: schema_req = cast('Schema_QuickStartRequestOption', schema) diff --git a/pcapkit/protocols/internet/ipv4.py b/pcapkit/protocols/internet/ipv4.py index 4d7b703e3b..7ce2dc9c52 100644 --- a/pcapkit/protocols/internet/ipv4.py +++ b/pcapkit/protocols/internet/ipv4.py @@ -1178,7 +1178,7 @@ def _read_opt_qs(self, schema: 'Schema_QSOption', *, options: 'Option') -> 'Data if schema.length != 8: raise ProtocolError(f'{self.alias}: [OptNo {schema.type}] invalid format') - func = schema.func + func = Enum_QSFunction.get(schema.flags['func']) if func == Enum_QSFunction.Quick_Start_Request: schema_req = cast('Schema_QuickStartRequestOption', schema) diff --git a/pcapkit/protocols/internet/ipv6_opts.py b/pcapkit/protocols/internet/ipv6_opts.py index e951287638..6aea7dea18 100644 --- a/pcapkit/protocols/internet/ipv6_opts.py +++ b/pcapkit/protocols/internet/ipv6_opts.py @@ -923,7 +923,7 @@ def _read_opt_qs(self, schema: 'Schema_QuickStartOption', *, options: 'Option') if schema.len != 6: raise ProtocolError(f'{self.alias}: [OptNo {schema.type}] invalid format') - func = schema.func + func = Enum_QSFunction.get(schema.flags['func']) if func == Enum_QSFunction.Quick_Start_Request: schema_req = cast('Schema_QuickStartRequestOption', schema) diff --git a/pcapkit/protocols/schema/internet/hopopt.py b/pcapkit/protocols/schema/internet/hopopt.py index 16832867fe..06226068b6 100644 --- a/pcapkit/protocols/schema/internet/hopopt.py +++ b/pcapkit/protocols/schema/internet/hopopt.py @@ -252,7 +252,11 @@ def quick_start_data_selector(pkt: 'dict[str, Any]') -> 'Field': schema = QuickStartOption.registry[func] if schema is None: raise FieldValueError(f'HOPOPT: invalid QS function: {func}') - return SchemaField(length=5, schema=schema) + + # NOTE: The suboption schemas carry the option's own ``type`` and ``len`` + # octets, so this is the width of the whole option: the two header octets + # plus the six data octets :rfc:`4782#section-3.2` gives both functions. + return SchemaField(length=8, schema=schema) def pad_opt_data_len(pkt: 'dict[str, Any]') -> 'int': diff --git a/pcapkit/protocols/schema/internet/ipv6_opts.py b/pcapkit/protocols/schema/internet/ipv6_opts.py index 6ae3e68bbb..e4b57be19e 100644 --- a/pcapkit/protocols/schema/internet/ipv6_opts.py +++ b/pcapkit/protocols/schema/internet/ipv6_opts.py @@ -252,7 +252,11 @@ def quick_start_data_selector(pkt: 'dict[str, Any]') -> 'Field': schema = QuickStartOption.registry[func] if schema is None: raise FieldValueError(f'IPv6-Opts: invalid QS function: {func}') - return SchemaField(length=5, schema=schema) + + # NOTE: The suboption schemas carry the option's own ``type`` and ``len`` + # octets, so this is the width of the whole option: the two header octets + # plus the six data octets :rfc:`4782#section-3.2` gives both functions. + return SchemaField(length=8, schema=schema) def pad_opt_data_len(pkt: 'dict[str, Any]') -> 'int': diff --git a/tests/protocols/internet/test_ipv4_unit.py b/tests/protocols/internet/test_ipv4_unit.py index e5f7f4dedd..a1c1292ae2 100644 --- a/tests/protocols/internet/test_ipv4_unit.py +++ b/tests/protocols/internet/test_ipv4_unit.py @@ -1859,11 +1859,10 @@ def assert_bad(reader, schema) -> None: qs_unknown = QuickStartRequestOption( type=OptionNumber.QS, length=8, - flags={'func': QSFunction.Quick_Start_Request, 'rate': 1}, + flags={'func': QSFunction.get(1), 'rate': 1}, ttl=7, nonce={'nonce': 3}, ) - object.__setattr__(qs_unknown, 'func', QSFunction.get(1)) assert_bad(proto._read_opt_qs, qs_unknown) def test_ipv4_schema_helpers_and_post_process_branches(self) -> None: diff --git a/tests/protocols/internet/test_ipv6_extension_unit.py b/tests/protocols/internet/test_ipv6_extension_unit.py index 2dfa7b2587..4d52ca4d80 100644 --- a/tests/protocols/internet/test_ipv6_extension_unit.py +++ b/tests/protocols/internet/test_ipv6_extension_unit.py @@ -1330,11 +1330,10 @@ def assert_bad(reader, option_schema) -> None: qs_unknown = schema.QuickStartRequestOption( type=Option.Quick_Start, len=6, - flags={'func': QSFunction.Quick_Start_Request, 'rate': 1}, + flags={'func': QSFunction.get(1), 'rate': 1}, ttl=7, nonce={'nonce': 3}, ) - object.__setattr__(qs_unknown, 'func', QSFunction.get(1)) assert_bad(proto._read_opt_qs, qs_unknown) assert_bad(proto._read_opt_rpl, schema.RPLOption( type=Option.RPL_Option_0x63, len=3, diff --git a/tests/protocols/internet/test_quick_start_func_unit.py b/tests/protocols/internet/test_quick_start_func_unit.py new file mode 100644 index 0000000000..b74530b11d --- /dev/null +++ b/tests/protocols/internet/test_quick_start_func_unit.py @@ -0,0 +1,105 @@ +# -*- coding: utf-8 -*- +"""Quick-Start options in IPv4, HOPOPT and IPv6-Opts build, parse and round-trip. + +GitHub issue #1091. ``_read_opt_qs`` read ``schema.func``, which the schema +declares only under ``TYPE_CHECKING`` and which only ``_QSOption.post_process`` +sets on the parse path. A schema built by ``_make_opt_qs`` never passes through +that hook, so constructing any Quick-Start option raised ``AttributeError: +'QuickStartRequestOption' object has no attribute 'func'``. The readers now +take the function from ``flags['func']``, the bit-field they already read +``rate`` from. + +HOPOPT and IPv6-Opts also sized the nested suboption with +``SchemaField(length=5)``, three octets short of the eight-octet option, so a +well-formed option decoded its nonce wrongly and left octets behind; +:meth:`QuickStartWireTests.test_wire_option_decodes_the_full_nonce` covers that. + +""" + +from __future__ import annotations + +import datetime +import importlib.util +import unittest +import warnings + +from tests._support import reimport_once_per_class + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + +#: A 30-bit nonce whose wire form, shifted left two bits, is ``deadbee0``. +NONCE = 0xDEADBEE0 >> 2 + + +def _families() -> 'list[tuple]': + from pcapkit.const.ipv4.option_number import OptionNumber as Enum_IPv4Option + from pcapkit.const.ipv6.option import Option as Enum_IPv6Option + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.hopopt import HOPOPT + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + + return [ + ('IPv4', IPv4, dict(src='192.0.2.1', dst='192.0.2.2', protocol=TransType.UDP), + Enum_IPv4Option.QS), + ('HOPOPT', HOPOPT, dict(next=TransType.UDP), Enum_IPv6Option.Quick_Start), + ('IPv6-Opts', IPv6_Opts, dict(next=TransType.UDP), Enum_IPv6Option.Quick_Start), + ] + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class QuickStartRoundTripTests(unittest.TestCase): + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_quick_start_option_round_trips(self) -> None: + from pcapkit.const.ipv6.qs_function import QSFunction as Enum_QSFunction + + for name, cls, base, code in _families(): + for func in (Enum_QSFunction.Quick_Start_Request, + Enum_QSFunction.Report_of_Approved_Rate): + with self.subTest(family=name, func=func.name): + with warnings.catch_warnings(): + warnings.simplefilter('error') + built = cls(options=[(code, dict(func=func, rate=80, ttl=5, nonce=NONCE))], + **base) + again = cls(built.data) + self.assertEqual(again.data, built.data) + + option = list(again.info.options.values())[0] + self.assertEqual(option.func, func) + self.assertEqual(option.rate, 80) + self.assertEqual(option.nonce, NONCE) + if func == Enum_QSFunction.Quick_Start_Request: + self.assertEqual(option.ttl, datetime.timedelta(seconds=5)) + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class QuickStartWireTests(unittest.TestCase): + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_wire_option_decodes_the_full_nonce(self) -> None: + from pcapkit.const.ipv6.qs_function import QSFunction as Enum_QSFunction + from pcapkit.protocols.internet.hopopt import HOPOPT + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + + # next=UDP, len=1 (16 octets); Quick-Start Request type 0x26 len 6, + # func 0 rate 1, ttl 5, nonce; then PadN of four octets. + wire = bytes.fromhex('1101' '26060105deadbee0' '010400000000') + for cls in (HOPOPT, IPv6_Opts): + with self.subTest(family=cls.__name__): + with warnings.catch_warnings(): + warnings.simplefilter('error') + parsed = cls(wire) + + options = list(parsed.info.options.values()) + self.assertEqual(options[0].func, Enum_QSFunction.Quick_Start_Request) + self.assertEqual(options[0].nonce, NONCE) + self.assertEqual(options[0].ttl, datetime.timedelta(seconds=5)) + self.assertEqual(len(options), 2) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/test_option_roundtrip_unit.py b/tests/protocols/test_option_roundtrip_unit.py index ad9f6dcc0c..393c780a44 100644 --- a/tests/protocols/test_option_roundtrip_unit.py +++ b/tests/protocols/test_option_roundtrip_unit.py @@ -246,52 +246,16 @@ class Gap(NamedTuple): # -- Quick-Start, in all three protocols that carry it -------------------- - # ``_make_opt_qs`` returns a bare nested schema, but ``func`` is set only by - # ``_QSOption.post_process``, which runs on the parse path. Since - # ``__post_init__`` packs and then re-reads, construction fails. - # - # There was a second, independent defect in the same option that these cases - # never reach, and it was the more serious of the two: - # ``quick_start_data_selector`` handed the nested schema a hardcoded - # ``SchemaField(length=5)`` where a Quick-Start Request needs eight octets - # (type 1 + length 1 + flags 1 + ttl 1 + nonce 4). Measured on a - # hand-built, well-formed 8-octet IPv4 Quick-Start option - # ``1908002adeadbee0``: it parsed "successfully" with - # ``SchemaWarning: packet length < 0: -3`` and decoded ``nonce`` as **55** - # instead of 933982136 -- silent corruption rather than a failure -- and the - # three unconsumed octets were then read as a further, fabricated option, - # which made the enclosing IPv4 datagram fail with ``ProtocolError: IPv4: - # invalid format``. - # - # #552 fixed IPv4's copy: the length now comes from - # ``quick_start_option_length(schema)``, i.e. from the suboption the selector - # resolved, and ``QuickStartReportOption`` gained the ``Not Used`` octet - # :rfc:`4782#section-3.1` gives it and it was missing -- it packed seven - # octets against the ``length=8`` that ``_make_opt_qs`` writes and - # ``_read_opt_qs`` demands, so a spec-correct Report of Approved Rate read off - # the wire decoded its nonce one octet early. ``ipv4-option/QS`` is still - # recorded below because the ``func`` defect is untouched and fails first. - # - # The identical ``SchemaField(length=5)`` is still at - # ``schema/internet/hopopt.py:255`` and ``schema/internet/ipv6_opts.py:255``, - # with the same measured nonce of 55; #552 was scoped to IPv4's copy. Note a - # fix there is not a copy of this one: :rfc:`4782#section-3.2` sets the IPv6 - # option's ``length`` field to 6 rather than 8, since it excludes the common - # type and length octets the extension header already carries. Fixing - # the ``func`` defect alone will not make any of these three cases pass. - 'ipv4-option/QS': Gap( - 'CONSTRUCT', "no attribute 'func'", - 'pcapkit/protocols/internet/ipv4.py:1178 -- func is set only by ' - 'post_process'), - 'hopopt-option/Quick_Start': Gap( - 'CONSTRUCT', "no attribute 'func'", - 'pcapkit/protocols/internet/hopopt.py:918; and separately ' - 'pcapkit/protocols/schema/internet/hopopt.py:255 -- SchemaField(length=5)'), - 'ipv6-opts-option/Quick_Start': Gap( - 'CONSTRUCT', "no attribute 'func'", - 'pcapkit/protocols/internet/ipv6_opts.py:921; and separately ' - 'pcapkit/protocols/schema/internet/ipv6_opts.py:255 -- ' - 'SchemaField(length=5)'), + # ``ipv4-option/QS``, ``hopopt-option/Quick_Start`` and + # ``ipv6-opts-option/Quick_Start`` used to be recorded here. ``_read_opt_qs`` + # read ``schema.func``, which is declared only under ``TYPE_CHECKING`` and set + # only by ``_QSOption.post_process`` on the parse path, so the schema + # ``_make_opt_qs`` builds raised ``AttributeError`` when ``__post_init__`` + # re-read it. The readers now take the function from ``flags['func']``, as + # they already took ``rate`` (#1091). HOPOPT and IPv6-Opts also sized the + # nested schema with ``SchemaField(length=5)`` against an eight-octet option, + # which decoded the nonce wrongly; #552 had fixed only IPv4's copy, and #1091 + # fixes the other two. All three now round-trip. # -- The non-progress loop, now fully fixed -------------------------------