From 48c3e9ea32e372387ff2bb900b6c4645dd0f7ed7 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Sun, 1 Nov 2026 22:45:38 -0500 Subject: [PATCH] fix(protocols): refuse Quick-Start values their fields cannot carry (#2617) - Move the RFC 4782 rate codec into a ProtocolBase reader/maker pair, _read_qs_rate and _make_qs_rate, called by IPv4, HOPOPT, IPv6-Opts and TCP in place of their own copies. - Raise ProtocolError for a rate other than 0 or 80..2621439 kbps, and, via _make_uint, for a TTL, nonce or reserved value outside its bits, where the makers clamped or wrapped them, wrote reserved bits into the nonce, or raised ValueError, OverflowError or FieldValueError. - HOPOPT and IPv6-Opts make an unassigned QS function as an unassigned option, the form their readers keep one in, as IPv4 does. - Tests: the codec, every refusal, and make -> parse and parse -> rebuild for every rate code; TCP's clamp pin becomes a refusal. --- pcapkit/protocols/internet/hopopt.py | 46 ++- pcapkit/protocols/internet/ipv4.py | 32 +- pcapkit/protocols/internet/ipv6_opts.py | 46 ++- pcapkit/protocols/protocol.py | 49 +++ pcapkit/protocols/transport/tcp.py | 32 +- .../internet/test_ipv6_extension_unit.py | 3 +- .../test_quick_start_rate_codec_2617_unit.py | 318 ++++++++++++++++++ .../test_tcp_quick_start_response_unit.py | 43 ++- 8 files changed, 513 insertions(+), 56 deletions(-) create mode 100644 tests/protocols/test_quick_start_rate_codec_2617_unit.py diff --git a/pcapkit/protocols/internet/hopopt.py b/pcapkit/protocols/internet/hopopt.py index eca9e0dbf6..572d3b936f 100644 --- a/pcapkit/protocols/internet/hopopt.py +++ b/pcapkit/protocols/internet/hopopt.py @@ -203,6 +203,17 @@ class HOPOPT(IPv6_Ext[Data_HOPOPT, Schema_HOPOPT], * A PadN option's data and the "Not Used" octet of a Quick-Start report are kept as captured rather than zeroed (:issue:`1223`). + .. versionchanged:: 1.6.0 + + * A Quick-Start rate the four-bit rate code cannot carry, or a TTL, + nonce or reserved value outside its field, raises + :exc:`~pcapkit.utilities.exceptions.ProtocolError`, rather than a bare + :exc:`ValueError`, an :exc:`OverflowError` or a + :exc:`~pcapkit.utilities.exceptions.FieldValueError`, or a different + value written in its place (:issue:`2617`). + * The Quick-Start maker builds an unassigned function code as an + unassigned option, the form the reader keeps one in (:issue:`2617`). + """ ########################################################################## @@ -1101,14 +1112,13 @@ def _read_opt_qs(self, schema: 'Schema_QuickStartOption', *, options: 'Option') if func == Enum_QSFunction.Quick_Start_Request: schema_req = cast('Schema_QuickStartRequestOption', schema) - rate = schema_req.flags['rate'] opt = Data_QuickStartRequestOption( type=schema.type, action=Enum_OptionAction.get(schema.type >> 6), change=bool(schema.type & 0b00100000), length=self._hopopt_option_length(schema_req.len), func=func, - rate=40000 * (2 ** rate) // 1000 if rate > 0 else 0, + rate=self._read_qs_rate(schema_req.flags['rate']), ttl=datetime.timedelta(seconds=schema_req.ttl), nonce=schema_req.nonce['nonce'], reserved=schema_req.nonce['reserved'], @@ -1116,14 +1126,13 @@ def _read_opt_qs(self, schema: 'Schema_QuickStartOption', *, options: 'Option') elif func == Enum_QSFunction.Report_of_Approved_Rate: schema_rep = cast('Schema_QuickStartReportOption', schema) - rate = schema_rep.flags['rate'] opt = Data_QuickStartReportOption( type=schema.type, action=Enum_OptionAction.get(schema.type >> 6), change=bool(schema.type & 0b00100000), length=self._hopopt_option_length(schema_rep.len), func=func, - rate=40000 * (2 ** rate) // 1000 if rate > 0 else 0, + rate=self._read_qs_rate(schema_rep.flags['rate']), nonce=schema_rep.nonce['nonce'], reserved=schema_rep.nonce['reserved'], unused=schema_rep.reserved, @@ -1909,7 +1918,8 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption func_default: default value for QS function type func_namespace: namespace for QS function type func_reversed: reversed flag for QS function type - rate: rate (in kbps) + rate: rate (in kbps), rounded down to the nearest rate the + four-bit rate code carries (:rfc:`4782#section-3.1`) ttl: time to live (in seconds) nonce: nonce value reserved: reserved bits (``R``) after the nonce @@ -1917,12 +1927,15 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption **kwargs: arbitrary keyword arguments Returns: - Constructed option schema; an unassigned option schema if ``opt`` - is a Quick-Start option parsed as unassigned, see :meth:`_read_opt_qs`. + Constructed option schema. An unassigned option schema if ``opt`` + is a Quick-Start option parsed as unassigned, see :meth:`_read_opt_qs`, + or if ``func`` is neither of the assigned functions, in which case + the option data takes the layout of a Quick-Start Request. Raises: - ProtocolError: If the QS function is neither a Quick-Start Request - nor a Report of Approved Rate. + ProtocolError: If ``func`` does not fit in four bits, the rate code + cannot carry ``rate``, or ``ttl``, ``nonce`` or ``reserved`` does + not fit its field. """ if isinstance(opt, Data_UnassignedOption): @@ -1937,11 +1950,12 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption else: func_enum = self._make_index(func, func_default, namespace=func_namespace, # type: ignore[assignment] reversed=func_reversed, pack=False) - rate_val = math.floor(math.log2(rate * 1000 / 40000)) if rate > 0 else 0 + rate_val = self._make_qs_rate(rate) + ttl_value = ttl if isinstance(ttl, int) else math.floor(ttl.total_seconds()) + nonce = self._make_uint(nonce, bits=30, name='Quick-Start nonce') + reserved = self._make_uint(reserved, bits=2, name='Quick-Start reserved') if func_enum == Enum_QSFunction.Quick_Start_Request: - ttl_value = ttl if isinstance(ttl, int) else math.floor(ttl.total_seconds()) - return Schema_QuickStartRequestOption( type=code, len=6, @@ -1949,7 +1963,7 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption 'func': func_enum, 'rate': rate_val, }, - ttl=ttl_value, + ttl=self._make_uint(ttl_value, bits=8, name='Quick-Start TTL'), nonce={ 'nonce': nonce, 'reserved': reserved, @@ -1969,7 +1983,11 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption 'reserved': reserved, }, ) - raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid QS function: {func_enum}') + if not 0 <= func_enum < 16: + raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid QS function: {func_enum}') + ttl_value = self._make_uint(ttl_value, bits=8, name='Quick-Start TTL') + data = bytes([(func_enum << 4) | rate_val, ttl_value]) + ((nonce << 2) | reserved).to_bytes(4, 'big') + return self._make_opt_none(code, data=data) def _make_opt_rpl(self, code: 'Enum_Option', opt: 'Optional[Data_RPLOption]' = None, *, down: 'bool' = False, diff --git a/pcapkit/protocols/internet/ipv4.py b/pcapkit/protocols/internet/ipv4.py index 99c76ddd87..5eaaaaaeb2 100644 --- a/pcapkit/protocols/internet/ipv4.py +++ b/pcapkit/protocols/internet/ipv4.py @@ -212,9 +212,15 @@ class IPv4(IP[Data_IPv4, Schema_IPv4], .. versionchanged:: 1.6.0 - The header checksum is computed when :meth:`make` is given none, and - verified on read under :data:`~pcapkit.utilities.config.CHECKSUM_VERIFY` - (:issue:`2472`). + * The header checksum is computed when :meth:`make` is given none, and + verified on read under :data:`~pcapkit.utilities.config.CHECKSUM_VERIFY` + (:issue:`2472`). + * A Quick-Start rate the four-bit rate code cannot carry, or a TTL, + nonce or reserved value outside its field, raises + :exc:`~pcapkit.utilities.exceptions.ProtocolError`, rather than a bare + :exc:`ValueError`, an :exc:`OverflowError` or a + :exc:`~pcapkit.utilities.exceptions.FieldValueError`, or a different + value written in its place (:issue:`2617`). """ @@ -1517,13 +1523,12 @@ def _read_opt_qs(self, schema: 'Schema_QSOption', *, options: 'Option') -> 'Data if func == Enum_QSFunction.Quick_Start_Request: schema_req = cast('Schema_QuickStartRequestOption', schema) - rate = schema_req.flags['rate'] opt = Data_QuickStartRequestOption( code=schema.type, type=self._read_ipv4_opt_type(schema.type), length=schema_req.length, func=func, - rate=40000 * (2 ** rate) // 1000 if rate > 0 else 0, + rate=self._read_qs_rate(schema_req.flags['rate']), ttl=datetime.timedelta(seconds=schema_req.ttl), nonce=schema_req.nonce['nonce'], reserved=schema_req.nonce['reserved'], @@ -1531,13 +1536,12 @@ def _read_opt_qs(self, schema: 'Schema_QSOption', *, options: 'Option') -> 'Data elif func == Enum_QSFunction.Report_of_Approved_Rate: schema_rep = cast('Schema_QuickStartReportOption', schema) - rate = schema_rep.flags['rate'] opt = Data_QuickStartReportOption( code=schema.type, type=self._read_ipv4_opt_type(schema.type), length=schema_rep.length, func=func, - rate=40000 * (2 ** rate) // 1000 if rate > 0 else 0, + rate=self._read_qs_rate(schema_rep.flags['rate']), unused=schema_rep.reserved, nonce=schema_rep.nonce['nonce'], reserved=schema_rep.nonce['reserved'], @@ -2170,7 +2174,8 @@ def _make_opt_qs(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_QuickSt func_default: default value for QS function type func_namespace: namespace for QS function type func_reversed: reversed flag for QS function type - rate: rate (in kbps) + rate: rate (in kbps), rounded down to the nearest rate the + four-bit rate code carries (:rfc:`4782#section-3.1`) ttl: time to live (in seconds) unused: the not-used fourth octet of a Report of Approved Rate nonce: nonce value @@ -2184,7 +2189,9 @@ def _make_opt_qs(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_QuickSt the option takes the eight-octet layout of a Quick-Start Request. Raises: - ProtocolError: If ``func`` does not fit in four bits. + ProtocolError: If ``func`` does not fit in four bits, the rate code + cannot carry ``rate``, or ``ttl``, ``nonce`` or ``reserved`` does + not fit its field. """ if isinstance(option, Data_UnassignedOption): @@ -2199,8 +2206,10 @@ def _make_opt_qs(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_QuickSt else: func_enum = self._make_index(func, func_default, namespace=func_namespace, # type: ignore[assignment] reversed=func_reversed, pack=False) - rate_val = math.floor(math.log2(rate * 1000 / 40000)) if rate > 0 else 0 + rate_val = self._make_qs_rate(rate) ttl_value = ttl if isinstance(ttl, int) else math.floor(ttl.total_seconds()) + nonce = self._make_uint(nonce, bits=30, name='Quick-Start nonce') + reserved = self._make_uint(reserved, bits=2, name='Quick-Start reserved') if func_enum == Enum_QSFunction.Quick_Start_Request: return Schema_QuickStartRequestOption( @@ -2210,7 +2219,7 @@ def _make_opt_qs(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_QuickSt 'func': func_enum, 'rate': rate_val, }, - ttl=ttl_value, + ttl=self._make_uint(ttl_value, bits=8, name='Quick-Start TTL'), nonce={ 'nonce': nonce, 'reserved': reserved, @@ -2232,5 +2241,6 @@ def _make_opt_qs(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_QuickSt ) if not 0 <= func_enum < 16: raise ProtocolError(f'{self.alias}: [OptNo {kind}] invalid QS function: {func_enum}') + ttl_value = self._make_uint(ttl_value, bits=8, name='Quick-Start TTL') data = bytes([(func_enum << 4) | rate_val, ttl_value]) + ((nonce << 2) | reserved).to_bytes(4, 'big') return cast('Schema_UnassignedOption', self._make_opt_unassigned(kind, data=data)) diff --git a/pcapkit/protocols/internet/ipv6_opts.py b/pcapkit/protocols/internet/ipv6_opts.py index 75d6628ebb..27b48d263a 100644 --- a/pcapkit/protocols/internet/ipv6_opts.py +++ b/pcapkit/protocols/internet/ipv6_opts.py @@ -207,6 +207,17 @@ class IPv6_Opts(IPv6_Ext[Data_IPv6_Opts, Schema_IPv6_Opts], * A PadN option's data and the "Not Used" octet of a Quick-Start report are kept as captured rather than zeroed (:issue:`1223`). + .. versionchanged:: 1.6.0 + + * A Quick-Start rate the four-bit rate code cannot carry, or a TTL, + nonce or reserved value outside its field, raises + :exc:`~pcapkit.utilities.exceptions.ProtocolError`, rather than a bare + :exc:`ValueError`, an :exc:`OverflowError` or a + :exc:`~pcapkit.utilities.exceptions.FieldValueError`, or a different + value written in its place (:issue:`2617`). + * The Quick-Start maker builds an unassigned function code as an + unassigned option, the form the reader keeps one in (:issue:`2617`). + """ ########################################################################## @@ -1089,14 +1100,13 @@ def _read_opt_qs(self, schema: 'Schema_QuickStartOption', *, options: 'Option') if func == Enum_QSFunction.Quick_Start_Request: schema_req = cast('Schema_QuickStartRequestOption', schema) - rate = schema_req.flags['rate'] opt = Data_QuickStartRequestOption( type=schema.type, action=Enum_OptionAction.get(schema.type >> 6), change=bool(schema.type & 0b00100000), length=self._ipv6_opts_option_length(schema_req.len), func=func, - rate=40000 * (2 ** rate) // 1000 if rate > 0 else 0, + rate=self._read_qs_rate(schema_req.flags['rate']), ttl=datetime.timedelta(seconds=schema_req.ttl), nonce=schema_req.nonce['nonce'], reserved=schema_req.nonce['reserved'], @@ -1104,14 +1114,13 @@ def _read_opt_qs(self, schema: 'Schema_QuickStartOption', *, options: 'Option') elif func == Enum_QSFunction.Report_of_Approved_Rate: schema_rep = cast('Schema_QuickStartReportOption', schema) - rate = schema_rep.flags['rate'] opt = Data_QuickStartReportOption( type=schema.type, action=Enum_OptionAction.get(schema.type >> 6), change=bool(schema.type & 0b00100000), length=self._ipv6_opts_option_length(schema_rep.len), func=func, - rate=40000 * (2 ** rate) // 1000 if rate > 0 else 0, + rate=self._read_qs_rate(schema_rep.flags['rate']), nonce=schema_rep.nonce['nonce'], reserved=schema_rep.nonce['reserved'], unused=schema_rep.reserved, @@ -1897,7 +1906,8 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption func_default: default value for QS function type func_namespace: namespace for QS function type func_reversed: reversed flag for QS function type - rate: rate (in kbps) + rate: rate (in kbps), rounded down to the nearest rate the + four-bit rate code carries (:rfc:`4782#section-3.1`) ttl: time to live (in seconds) nonce: nonce value reserved: reserved bits (``R``) after the nonce @@ -1905,12 +1915,15 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption **kwargs: arbitrary keyword arguments Returns: - Constructed option schema; an unassigned option schema if ``opt`` - is a Quick-Start option parsed as unassigned, see :meth:`_read_opt_qs`. + Constructed option schema. An unassigned option schema if ``opt`` + is a Quick-Start option parsed as unassigned, see :meth:`_read_opt_qs`, + or if ``func`` is neither of the assigned functions, in which case + the option data takes the layout of a Quick-Start Request. Raises: - ProtocolError: If the QS function is neither a Quick-Start Request - nor a Report of Approved Rate. + ProtocolError: If ``func`` does not fit in four bits, the rate code + cannot carry ``rate``, or ``ttl``, ``nonce`` or ``reserved`` does + not fit its field. """ if isinstance(opt, Data_UnassignedOption): @@ -1925,11 +1938,12 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption else: func_enum = self._make_index(func, func_default, namespace=func_namespace, # type: ignore[assignment] reversed=func_reversed, pack=False) - rate_val = math.floor(math.log2(rate * 1000 / 40000)) if rate > 0 else 0 + rate_val = self._make_qs_rate(rate) + ttl_value = ttl if isinstance(ttl, int) else math.floor(ttl.total_seconds()) + nonce = self._make_uint(nonce, bits=30, name='Quick-Start nonce') + reserved = self._make_uint(reserved, bits=2, name='Quick-Start reserved') if func_enum == Enum_QSFunction.Quick_Start_Request: - ttl_value = ttl if isinstance(ttl, int) else math.floor(ttl.total_seconds()) - return Schema_QuickStartRequestOption( type=code, len=6, @@ -1937,7 +1951,7 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption 'func': func_enum, 'rate': rate_val, }, - ttl=ttl_value, + ttl=self._make_uint(ttl_value, bits=8, name='Quick-Start TTL'), nonce={ 'nonce': nonce, 'reserved': reserved, @@ -1957,7 +1971,11 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption 'reserved': reserved, }, ) - raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid QS function: {func_enum}') + if not 0 <= func_enum < 16: + raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid QS function: {func_enum}') + ttl_value = self._make_uint(ttl_value, bits=8, name='Quick-Start TTL') + data = bytes([(func_enum << 4) | rate_val, ttl_value]) + ((nonce << 2) | reserved).to_bytes(4, 'big') + return self._make_opt_none(code, data=data) def _make_opt_rpl(self, code: 'Enum_Option', opt: 'Optional[Data_RPLOption]' = None, *, down: 'bool' = False, diff --git a/pcapkit/protocols/protocol.py b/pcapkit/protocols/protocol.py index 219fb8d34f..34047bf68c 100644 --- a/pcapkit/protocols/protocol.py +++ b/pcapkit/protocols/protocol.py @@ -2654,6 +2654,55 @@ def _check_term_threshold(self) -> bool: return layer_match or protocol_match + def _read_qs_rate(self, code: 'int') -> 'int': + """Read a Quick-Start rate code as a rate in kbps. + + :rfc:`4782#section-3.1` encodes the Rate Request, and the Rate Report + formatted exactly as it, as a four-bit ``N`` that stands for + ``K * 2 ** N`` bps with ``K`` set to 40,000, and ``N = 0`` for a rate of + zero. The IPv6 option keeps the IPv4 format (:rfc:`4782#section-3.2`), + and the TCP Quick-Start Response formats its Rate Request the same way + (:rfc:`4782#section-4.2`). + + Args: + code: the four-bit rate field, ``N`` + + Returns: + The rate in kbps: ``0`` for ``N = 0``, else ``40 * 2 ** N``. + + .. versionadded:: 1.6.0 + + """ + return 40 * 2 ** code if code else 0 + + def _make_qs_rate(self, rate: 'Any') -> 'int': + """Make the Quick-Start rate code for a rate in kbps. + + The inverse of :meth:`_read_qs_rate`. A rate between two codes is + rounded down to the lower one, so the option never carries more than + the rate given. + + Args: + rate: the rate, in kbps + + Returns: + The four-bit rate field, ``N``. + + Raises: + ProtocolError: If ``rate`` is not an :class:`int`, is a + :obj:`bool`, or is one the field cannot carry: negative, above + zero but below the 80 kbps of ``N = 1``, or at or above the + ``40 * 2 ** 16`` kbps that needs a fifth bit. The message names + :attr:`alias`. + + .. versionadded:: 1.6.0 + + """ + if (not isinstance(rate, int) or isinstance(rate, bool) + or (rate and not 80 <= rate < 40 * 2 ** 16)): + raise ProtocolError(f'{self.alias}: invalid Quick-Start rate: {rate!r}') + return (rate // 40).bit_length() - 1 if rate else 0 + class Protocol(ProtocolBase, Generic[_PT, _ST]): """User-facing abstract base class for third-party protocols. diff --git a/pcapkit/protocols/transport/tcp.py b/pcapkit/protocols/transport/tcp.py index fc429949d8..7f8a48b0e8 100644 --- a/pcapkit/protocols/transport/tcp.py +++ b/pcapkit/protocols/transport/tcp.py @@ -332,10 +332,16 @@ class TCP(Transport[Data_TCP, Schema_TCP], .. versionchanged:: 1.6.0 - The checksum is computed when :meth:`make` is given none, over the - ``pseudo`` it is given or by the enclosing IP layer, and verified by that - layer under :data:`~pcapkit.utilities.config.CHECKSUM_VERIFY` - (:issue:`2473`). + * The checksum is computed when :meth:`make` is given none, over the + ``pseudo`` it is given or by the enclosing IP layer, and verified by + that layer under :data:`~pcapkit.utilities.config.CHECKSUM_VERIFY` + (:issue:`2473`). + * A Quick-Start rate the four-bit rate code cannot carry, or a nonce or + reserved value outside its field, raises + :exc:`~pcapkit.utilities.exceptions.ProtocolError`, rather than a bare + :exc:`ValueError`, an :exc:`OverflowError` or a + :exc:`~pcapkit.utilities.exceptions.FieldValueError`, or a rate of + zero written in its place (:issue:`2617`). """ @@ -1602,11 +1608,10 @@ def _read_mode_qs(self, schema: 'Schema_QuickStartResponse', *, options: 'Option if schema.length != 8: raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format') - rate = schema.flags['rate'] data = Data_QuickStartResponse( kind=schema.kind, length=schema.length, - req_rate=40000 * (2 ** rate) // 1000 if rate > 0 else 0, + req_rate=self._read_qs_rate(schema.flags['rate']), ttl_diff=schema.diff, nonce=schema.nonce['nonce'], reserved=schema.flags['reserved'], @@ -2765,7 +2770,8 @@ def _make_mode_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartRespo Args: code: option code opt: option data - rate: rate (in kbps) + rate: rate (in kbps), rounded down to the nearest rate the + four-bit rate code carries (:rfc:`4782#section-4.2`) diff: TTL Diff, the IP TTL minus the QS TTL, modulo 256 (:rfc:`4782` Section 4.2); a :class:`~datetime.timedelta` is taken as its whole seconds @@ -2777,6 +2783,10 @@ def _make_mode_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartRespo Returns: Constructed option schema. + Raises: + ProtocolError: If the rate code cannot carry ``rate``, or ``nonce``, + ``reserved`` or ``nonce_reserved`` does not fit its field. + """ if opt is not None: rate = opt.req_rate @@ -2785,11 +2795,11 @@ def _make_mode_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartRespo reserved = opt.reserved nonce_reserved = opt.nonce_reserved - # :rfc:`4782` Section 3.1 encodes the rate as 40 kbps * 2 ** N, with - # N = 0 meaning a rate of zero; a response may report a lower rate than - # requested (Section 4.3), so a rate under 80 kbps rounds down to N = 0. - rate_val = max(math.floor(math.log2(rate * 1000 / 40000)), 0) if rate > 0 else 0 + rate_val = self._make_qs_rate(rate) diff_val = diff if isinstance(diff, int) else math.floor(diff.total_seconds()) + nonce = self._make_uint(nonce, bits=30, name='Quick-Start nonce') + reserved = self._make_uint(reserved, bits=4, name='Quick-Start reserved') + nonce_reserved = self._make_uint(nonce_reserved, bits=2, name='Quick-Start nonce reserved') return Schema_QuickStartResponse( kind=code, diff --git a/tests/protocols/internet/test_ipv6_extension_unit.py b/tests/protocols/internet/test_ipv6_extension_unit.py index c4b5be5c71..d340bce438 100644 --- a/tests/protocols/internet/test_ipv6_extension_unit.py +++ b/tests/protocols/internet/test_ipv6_extension_unit.py @@ -858,8 +858,9 @@ def _assert_option_data_model_constructor_edges(self, protocol_cls: type) -> Non unused=b'\x00', **base), ).nonce['nonce'], 4) + # An unassigned function builds (#2617); one past four bits does not. with self.assertRaises(ProtocolError): - proto._make_opt_qs(Option.Quick_Start, func=1) + proto._make_opt_qs(Option.Quick_Start, func=16) rpl = proto._make_opt_rpl( Option.RPL_Option_0x63, diff --git a/tests/protocols/test_quick_start_rate_codec_2617_unit.py b/tests/protocols/test_quick_start_rate_codec_2617_unit.py new file mode 100644 index 0000000000..c53f05630c --- /dev/null +++ b/tests/protocols/test_quick_start_rate_codec_2617_unit.py @@ -0,0 +1,318 @@ +# -*- coding: utf-8 -*- +"""The Quick-Start rate codec, and the Quick-Start option round trips (GitHub issue #2617). + +:rfc:`4782` Section 3.1 encodes a Quick-Start rate as a four-bit ``N`` that +stands for ``40 kbps * 2 ** N``, with ``N = 0`` for a rate of zero. The IPv6 +option keeps the IPv4 format (Section 3.2), and the TCP Quick-Start Response +formats its Rate Request the same way (Section 4.2). IPv4, HOPOPT, IPv6-Opts and +TCP each held a copy of the encoder and the decoder, and the copies disagreed +outside the field: the IP families raised a bare ``ValueError`` for +``0 < rate < 40``, TCP wrote ``N = 0`` there, and all four wrote ``N = 0`` for a +negative rate or one under 80 kbps. The owner ruled on #2617 that a rate the +field cannot carry raises ``ProtocolError`` everywhere, and that every reader +and maker pairs up and round-trips both ways. + +* :class:`CodecTests` -- ``_read_qs_rate`` and ``_make_qs_rate``, the pair on the + protocol base that every family calls. +* :class:`RefusalTests` -- each IP family's ``make()`` refuses a rate the field + cannot carry; TCP's case is in + ``tests/protocols/transport/test_tcp_quick_start_response_unit.py``. +* :class:`FieldRefusalTests` -- the same for the TTL, nonce and ``R`` bits, which + the IP makers wrapped, wrote into the nonce, or refused with a bare + ``ValueError``, ``OverflowError`` or ``FieldValueError``. +* :class:`RoundTripTests` -- for every rate code, family and function, make -> + parse gives the values back and parse -> rebuild gives the octets back, both + checked against the RFC's octet layout. +* :class:`UnassignedFunctionTests` -- HOPOPT and IPv6-Opts read an unassigned + function code as an unassigned option but had no maker for one, which IPv4 had. + +Every case builds its own octets, so this module is unit-tier. Classes are +imported inside each test, after :func:`~tests._support.reimport_once_per_class`. + +""" + +from __future__ import annotations + +import datetime +import io +import re +import unittest +import warnings +from typing import TYPE_CHECKING + +from tests._support import reimport_once_per_class + +if TYPE_CHECKING: + from typing import Any, Callable + +#: Test tier (#1958): hand-picked round-trip cases of one field, not a sweep. +__tier__ = 'ALL' + +#: A 30-bit nonce and the two ``R`` bits after it, as a wire word. +NONCE, RESERVED = 0xDEADBEE0 >> 2, 1 +NONCE_WORD = ((NONCE << 2) | RESERVED).to_bytes(4, 'big') + +#: Every rate a four-bit code carries, by code: ``40 * 2 ** N`` kbps, zero for ``N = 0``. +RATES = {code: 40 * 2 ** code if code else 0 for code in range(16)} + +#: Rates the field cannot carry: negative, above zero but under ``N = 1``, past ``N = 15``. +UNCARRIABLE = (-80, -1, 1, 10, 39, 40, 79, 40 * 2 ** 16, 10 ** 12) + +#: The unassigned Quick-Start function codes (:rfc:`4782` assigns ``0`` and ``8``). +UNASSIGNED = tuple(func for func in range(16) if func not in (0, 8)) + + +def ip_families() -> 'list[tuple[str, Callable[..., bytes], Callable[[bytes], Any], Any, bytes]]': + """``(name, make, parse, option code, option prefix)`` for each IP family. + + ``make`` builds a packet from a list of options and ``parse`` reads its octets; + the prefix is the option's type and length octets, eight for IPv4's option + length and six for the IPv6 ``Opt Data Len`` (:rfc:`4782` Section 3.2). + """ + from pcapkit.const.ipv4.option_number import OptionNumber + from pcapkit.const.ipv6.option import Option + 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 + + def ipv4(options: 'list') -> 'bytes': + return IPv4(src='192.0.2.1', dst='192.0.2.2', protocol=TransType.UDP, options=options).data + + def ipv6(cls: 'type') -> 'Callable[[list], bytes]': + return lambda options: cls(next=TransType.UDP, options=options).data + + def parse6(cls: 'type') -> 'Callable[[bytes], Any]': + return lambda octets: cls(io.BytesIO(octets), len(octets), extension=True) + + return [ + ('IPv4', ipv4, lambda octets: IPv4(io.BytesIO(octets), len(octets)), OptionNumber.QS, b'\x19\x08'), + ('HOPOPT', ipv6(HOPOPT), parse6(HOPOPT), Option.Quick_Start, b'\x26\x06'), + ('IPv6-Opts', ipv6(IPv6_Opts), parse6(IPv6_Opts), Option.Quick_Start, b'\x26\x06'), + ] + + +def option_octets(name: 'str', data: 'bytes') -> 'bytes': + """The eight octets of the first option in a packet built by :func:`ip_families`.""" + return data[20:28] if name == 'IPv4' else data[2:10] + + +def ip_wire(name: 'str', option: 'bytes') -> 'bytes': + """A header carrying ``option`` and nothing else, padded to a whole header.""" + if name == 'IPv4': + # IHL 8: a 20-octet header, the option and four octets of EOOL. + return (bytes.fromhex('48000020' '00000000' '40110000') + bytes([192, 0, 2, 1, 192, 0, 2, 2]) + + option + bytes(4)) + # next UDP, Hdr Ext Len 1: the option, then a PadN of four octets. + return bytes([17, 1]) + option + bytes.fromhex('010400000000') + + +def rebuild(parsed: 'Any') -> 'bytes': + """``from_data`` of a parsed layer, as octets.""" + return type(parsed).from_data(parsed.info).data + + +class _QSTestCase(unittest.TestCase): + """A fresh :mod:`pcapkit` per class.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + +class CodecTests(_QSTestCase): + """``_read_qs_rate`` and ``_make_qs_rate`` against :rfc:`4782` Section 3.1.""" + + def setUp(self) -> None: + super().setUp() + from pcapkit.protocols.misc.raw import Raw + + self.proto = Raw(b'') + + def test_read_gives_k_times_two_to_the_n(self) -> None: + # K * 2 ** N bps with K = 40,000, zero for N = 0; "80 Kbps to 1.3 Gbps". + self.assertEqual(self.proto._read_qs_rate(0), 0) + self.assertEqual(self.proto._read_qs_rate(1), 80) + self.assertEqual(self.proto._read_qs_rate(15), 1_310_720) + for code in range(1, 16): + with self.subTest(code=code): + self.assertEqual(self.proto._read_qs_rate(code) * 1000, 40_000 * 2 ** code) + + def test_make_inverts_read(self) -> None: + for code, rate in RATES.items(): + with self.subTest(code=code): + self.assertEqual(self.proto._make_qs_rate(rate), code) + + def test_make_rounds_down_between_codes(self) -> None: + for code in range(1, 16): + for rate in (RATES[code] + 1, 2 * RATES[code] - 1): + with self.subTest(rate=rate): + self.assertEqual(self.proto._make_qs_rate(rate), code) + + def test_make_refuses_a_rate_the_field_cannot_carry(self) -> None: + from pcapkit.utilities.exceptions import ProtocolError + + for rate in UNCARRIABLE + (80.0, True, False, None, '80'): + with self.subTest(rate=rate): + with self.assertRaisesRegex(ProtocolError, f'^Raw: invalid Quick-Start rate: {re.escape(repr(rate))}$'): + self.proto._make_qs_rate(rate) + + +class RefusalTests(_QSTestCase): + """The IP families raise ``ProtocolError`` for a rate the field cannot carry.""" + + def test_each_family_and_function_refuses(self) -> None: + from pcapkit.const.ipv6.qs_function import QSFunction + from pcapkit.utilities.exceptions import ProtocolError + + for name, make, _, code, _ in ip_families(): + for func in (QSFunction.Quick_Start_Request, QSFunction.Report_of_Approved_Rate, 5): + for rate in UNCARRIABLE: + with self.subTest(family=name, func=func, rate=rate): + with self.assertRaisesRegex(ProtocolError, f': invalid Quick-Start rate: {rate}$'): + make([(code, dict(func=func, rate=rate))]) + + +class FieldRefusalTests(_QSTestCase): + """The IP families raise ``ProtocolError`` for a TTL, nonce or ``R`` value outside its field.""" + + #: ``(keyword, message name, values outside the field)``: 8, 30 and 2 bits. + FIELDS = ( + ('ttl', 'TTL', (-1, 256, 300, True)), + ('nonce', 'nonce', (-1, 1 << 30, 1 << 32, True)), + ('reserved', 'reserved', (-1, 4, 7, True)), + ) + + def test_each_family_and_function_refuses(self) -> None: + from pcapkit.const.ipv6.qs_function import QSFunction + from pcapkit.utilities.exceptions import ProtocolError + + for name, make, _, code, _ in ip_families(): + for func in (QSFunction.Quick_Start_Request, QSFunction.Report_of_Approved_Rate, 5): + for keyword, what, values in self.FIELDS: + if keyword == 'ttl' and func == QSFunction.Report_of_Approved_Rate: + continue + for value in values: + with self.subTest(family=name, func=func, field=keyword, value=value): + pattern = f': invalid Quick-Start {what}: {re.escape(repr(value))}$' + with self.assertRaisesRegex(ProtocolError, pattern): + make([(code, {'func': func, 'rate': 80, keyword: value})]) + + def test_the_bounds_of_each_field_build(self) -> None: + from pcapkit.const.ipv6.qs_function import QSFunction + + for name, make, parse, code, prefix in ip_families(): + for ttl, nonce, reserved in ((0, 0, 0), (255, (1 << 30) - 1, 3)): + with self.subTest(family=name, ttl=ttl): + data = make([(code, dict(func=QSFunction.Quick_Start_Request, rate=80, ttl=ttl, + nonce=nonce, reserved=reserved))]) + word = ((nonce << 2) | reserved).to_bytes(4, 'big') + self.assertEqual(option_octets(name, data), prefix + bytes([0x01, ttl]) + word) + option = parse(data).info.options[code] + self.assertEqual((option.ttl.seconds, option.nonce, option.reserved), (ttl, nonce, reserved)) + + def test_a_report_ignores_the_ttl_it_does_not_carry(self) -> None: + from pcapkit.const.ipv6.qs_function import QSFunction + + for name, make, _, code, _ in ip_families(): + with self.subTest(family=name): + args = dict(func=QSFunction.Report_of_Approved_Rate, rate=80) + self.assertEqual(make([(code, dict(args, ttl=300))]), make([(code, args)])) + + +class RoundTripTests(_QSTestCase): + """Every rate code round-trips both ways through each family's reader and maker.""" + + def test_ip_make_then_parse_gives_the_values_back(self) -> None: + from pcapkit.const.ipv6.qs_function import QSFunction + + request, report = QSFunction.Quick_Start_Request, QSFunction.Report_of_Approved_Rate + for name, make, parse, code, prefix in ip_families(): + unused = 7 if name == 'IPv4' else b'\x07' + for func, fourth in ((request, dict(ttl=5)), (report, dict(unused=unused))): + for n, rate in RATES.items(): + with self.subTest(family=name, func=func.name, rate=rate): + data = make([(code, dict(func=func, rate=rate, nonce=NONCE, reserved=RESERVED, **fourth))]) + # Figures 3 and 4: function and rate nibbles, the fourth octet, then the nonce word. + self.assertEqual(option_octets(name, data), + prefix + bytes([(func << 4) | n, 5 if func == request else 7]) + NONCE_WORD) + + option = parse(data).info.options[code] + self.assertEqual((option.func, option.rate, option.nonce, option.reserved), + (func, rate, NONCE, RESERVED)) + if func == request: + self.assertEqual(option.ttl, datetime.timedelta(seconds=5)) + else: + self.assertEqual(option.unused, unused) + + def test_ip_parse_then_rebuild_gives_the_octets_back(self) -> None: + from pcapkit.const.ipv6.qs_function import QSFunction + + for name, _, parse, code, prefix in ip_families(): + for func in (QSFunction.Quick_Start_Request, QSFunction.Report_of_Approved_Rate): + for n, rate in RATES.items(): + with self.subTest(family=name, func=func.name, rate=rate): + wire = ip_wire(name, prefix + bytes([(func << 4) | n, 9]) + NONCE_WORD) + with warnings.catch_warnings(): + warnings.simplefilter('error') + parsed = parse(wire) + self.assertEqual(parsed.info.options[code].rate, rate) + self.assertEqual(rebuild(parsed), wire) + + def test_tcp_make_then_parse_gives_the_values_back(self) -> None: + from pcapkit.const.tcp.option import Option + from pcapkit.protocols.transport.tcp import TCP + + code = Option.Quick_Start_Response + for n, rate in RATES.items(): + with self.subTest(rate=rate): + data = TCP(srcport=1, dstport=2, options=[(code, dict(rate=rate, diff=3, nonce=NONCE, reserved=0xa, + nonce_reserved=RESERVED))]).data + # Section 4.2: Kind 27, Length 8, Resv. and Rate Request, TTL Diff, then the nonce word. + self.assertEqual(data[20:28], bytes([27, 8, 0xa0 | n, 3]) + NONCE_WORD) + + option = TCP(data, len(data)).info.options[code] + self.assertEqual((option.req_rate, option.ttl_diff, option.nonce, option.reserved, + option.nonce_reserved), (rate, 3, NONCE, 0xa, RESERVED)) + + def test_tcp_parse_then_rebuild_gives_the_octets_back(self) -> None: + from pcapkit.const.tcp.option import Option + from pcapkit.protocols.transport.tcp import TCP + + header = bytes.fromhex('0001000200000000000000007000ffff00000000') + for n, rate in RATES.items(): + with self.subTest(rate=rate): + wire = header + bytes([27, 8, 0x50 | n, 3]) + NONCE_WORD + parsed = TCP(wire, len(wire)) + self.assertEqual(parsed.info.options[Option.Quick_Start_Response].req_rate, rate) + self.assertEqual(rebuild(parsed), wire) + + +class UnassignedFunctionTests(_QSTestCase): + """Each IP family makes an unassigned function code, as its reader reads one.""" + + def test_make_then_parse_keeps_the_option_data(self) -> None: + for name, make, parse, code, prefix in ip_families(): + for func in UNASSIGNED: + with self.subTest(family=name, func=func): + data = make([(code, dict(func=func, rate=160, ttl=5, nonce=NONCE, reserved=RESERVED))]) + # The layout of a Quick-Start Request, under the unassigned code. + body = bytes([(func << 4) | 2, 5]) + NONCE_WORD + self.assertEqual(option_octets(name, data), prefix + body) + + parsed = parse(data) + option = parsed.info.options[code] + self.assertEqual(type(option).__name__, 'UnassignedOption') + self.assertEqual(option.data, body) + self.assertEqual(rebuild(parsed), data) + + def test_a_function_past_four_bits_is_refused(self) -> None: + from pcapkit.utilities.exceptions import ProtocolError + + for name, make, _, code, _ in ip_families(): + with self.subTest(family=name): + with self.assertRaisesRegex(ProtocolError, 'invalid QS function: 16'): + make([(code, dict(func=16))]) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/transport/test_tcp_quick_start_response_unit.py b/tests/protocols/transport/test_tcp_quick_start_response_unit.py index 61d0d6db92..94c4144aa1 100644 --- a/tests/protocols/transport/test_tcp_quick_start_response_unit.py +++ b/tests/protocols/transport/test_tcp_quick_start_response_unit.py @@ -12,9 +12,11 @@ ``0 < rate < 40`` kbps, and a negative code cannot pack into the 4-bit field. :rfc:`4782` Section 4.2 gives the option layout, and Section 3.1 the rate -encoding: ``40 kbps * 2 ** N``, where ``N = 0`` means a rate of zero. Section 4.3 -lets a response report a lower rate than the one requested, so a rate below the -smallest non-zero code (80 kbps) is rounded down to ``N = 0`` rather than refused. +encoding: ``40 kbps * 2 ** N``, where ``N = 0`` means a rate of zero. A rate +between two codes rounds down to the lower one. The owner ruled on #2617 that a +rate the field cannot carry -- above zero but under the 80 kbps of ``N = 1``, or +past ``N = 15`` -- raises ``ProtocolError`` rather than being written as zero, +which is what #1119's fix had done. Every case builds its own octets in memory, so this belongs to the unit tier. @@ -104,14 +106,45 @@ def test_rate_parses_as_an_int(self) -> 'None': self.assertEqual(req_rate, rate) def test_rate_rounds_down_to_a_representable_code(self) -> 'None': - """A rate is floored to ``40 * 2 ** N``, and below 80 kbps to ``N = 0``.""" + """A rate is floored to ``40 * 2 ** N``.""" QS = qs() - for rate, code in ((1, 0), (20, 0), (39, 0), (40, 0), (79, 0), (80, 1), (159, 1), (160, 2)): + for rate, code in ((0, 0), (80, 1), (159, 1), (160, 2), (2_621_439, 15)): with self.subTest(rate=rate): raw = build(rate) self.assertEqual(raw[20:24], bytes([27, 8, code, 3])) self.assertEqual(parse(raw).info.options[QS].req_rate, 40 * 2 ** code if code else 0) + def test_a_rate_the_field_cannot_carry_is_refused(self) -> 'None': + """Under 80 kbps but not zero, negative, or past ``N = 15``: refused, not zeroed (#2617).""" + from pcapkit.utilities.exceptions import ProtocolError + + for rate in (-1, 1, 20, 39, 40, 79, 2_621_440): + with self.subTest(rate=rate): + with self.assertRaisesRegex(ProtocolError, rf'^TCP: invalid Quick-Start rate: {rate}$'): + build(rate) + + def test_a_field_value_outside_its_bits_is_refused(self) -> 'None': + """The nonce (30 bits), ``Resv.`` (4) and ``R`` (2): refused, not wrapped or leaked (#2617).""" + from pcapkit.protocols.transport.tcp import TCP + from pcapkit.utilities.exceptions import ProtocolError + + for keyword, what, values in (('nonce', 'nonce', (-1, 1 << 30, True)), + ('reserved', 'reserved', (-1, 16, True)), + ('nonce_reserved', 'nonce reserved', (-1, 4, True))): + for value in values: + with self.subTest(field=keyword, value=value): + with self.assertRaisesRegex(ProtocolError, rf'^TCP: invalid Quick-Start {what}: {value!r}$'): + TCP(srcport=1, dstport=2, options=[(qs(), {'rate': 80, keyword: value})]) + + def test_ttl_diff_is_taken_modulo_256(self) -> 'None': + """:rfc:`4782` Section 3.1 defines TTL Diff modulo 256, so its octet wraps.""" + from pcapkit.protocols.transport.tcp import TCP + + for diff, octet in ((255, 0xff), (256, 0x00), (300, 0x2c), (-1, 0xff)): + with self.subTest(diff=diff): + raw = TCP(srcport=1, dstport=2, options=[(qs(), {'rate': 80, 'diff': diff})]).data + self.assertEqual(raw[23], octet) + if __name__ == '__main__': unittest.main()