From 6011af7d7e6798adbd46850561e9e963f2af49a2 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Tue, 6 Oct 2026 15:17:09 -0400 Subject: [PATCH] fix(tcp): Quick-Start Response reader reads a stray octet; maker fails below 40 kbps - _read_mode_qs read one octet past the option before its length check, so every Quick-Start Response failed with "StructError: unpack: empty buffer". - _make_mode_qs took floor(log2(rate / 40)), negative for rates under 40 kbps, which cannot pack into the 4-bit field. Clamp to N = 0: RFC 4782 Section 3.1 defines N = 0 as a rate of zero, and Section 4.3 lets a response report a lower rate than requested. - Drop the tcp-option/Quick_Start_Response EXPECTED_FAILURES entry, now closed. - Add tests/protocols/transport/test_tcp_quick_start_response_unit.py. Closes #1119 --- pcapkit/protocols/transport/tcp.py | 6 +- tests/protocols/test_option_roundtrip_unit.py | 11 --- .../test_tcp_quick_start_response_unit.py | 84 +++++++++++++++++++ 3 files changed, 88 insertions(+), 13 deletions(-) create mode 100644 tests/protocols/transport/test_tcp_quick_start_response_unit.py diff --git a/pcapkit/protocols/transport/tcp.py b/pcapkit/protocols/transport/tcp.py index 1f9d12f368..c16fb7cd4d 100644 --- a/pcapkit/protocols/transport/tcp.py +++ b/pcapkit/protocols/transport/tcp.py @@ -1348,7 +1348,6 @@ def _read_mode_qs(self, schema: 'Schema_QuickStartResponse', *, options: 'Option ProtocolError: If length is **NOT** ``8``. """ - size = self._read_unpack(1) if schema.length != 8: raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format') @@ -2576,7 +2575,10 @@ def _make_mode_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartRespo diff = opt.ttl_diff nonce = opt.nonce - rate_val = math.floor(math.log2(rate * 1000 / 40000)) if rate > 0 else 0 + # :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 diff_val = diff if isinstance(diff, int) else math.floor(diff.total_seconds()) return Schema_QuickStartResponse( diff --git a/tests/protocols/test_option_roundtrip_unit.py b/tests/protocols/test_option_roundtrip_unit.py index 1c0973cb0f..4206f87be0 100644 --- a/tests/protocols/test_option_roundtrip_unit.py +++ b/tests/protocols/test_option_roundtrip_unit.py @@ -137,17 +137,6 @@ class Gap(NamedTuple): # -- TCP ------------------------------------------------------------------ - # ``_make_mode_qs`` computes ``rate_val`` as a floor of a logarithm that is - # negative for any rate under 40 kbps, and a negative value then fails to - # pack into the 4-bit field with ``ValueError: invalid literal for int() - # with base 2: b'0000-110'``. At the default ``rate=0`` it is guarded, and - # the option constructs -- and then will not parse back. - 'tcp-option/Quick_Start_Response': Gap( - 'PARSE', 'StructError', - 'pcapkit/protocols/transport/tcp.py:2465 -- _make_mode_qs; rate_val is ' - 'negative below 40 kbps, and the option it emits at rate=0 does not ' - 'parse back'), - # #541 declared real ``kind``/``length`` fields on ``MPTCP``, which got six # of these seven far enough to construct and pack, and #566/#567 closed # the rest of the chain: #566 gave ``MPTCP.subtype`` the same treatment diff --git a/tests/protocols/transport/test_tcp_quick_start_response_unit.py b/tests/protocols/transport/test_tcp_quick_start_response_unit.py new file mode 100644 index 0000000000..378acaae8b --- /dev/null +++ b/tests/protocols/transport/test_tcp_quick_start_response_unit.py @@ -0,0 +1,84 @@ +# -*- coding: utf-8 -*- +"""The TCP Quick-Start Response option (kind 27), read and constructed. + +GitHub issue #1119 found two defects in this option: + +* :meth:`TCP._read_mode_qs ` + read one octet past the option before checking its length, so every Quick-Start + Response failed with ``StructError: unpack: empty buffer``, whether or not a + payload followed. +* :meth:`TCP._make_mode_qs ` + computed the rate code as ``floor(log2(rate / 40))``. That is negative for + ``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. + +Every case builds its own octets in memory, so this belongs to the unit tier. + +""" +from __future__ import annotations + +import unittest + +from pcapkit.const.tcp.option import Option +from pcapkit.protocols.transport.tcp import TCP + +#: Quick-Start Response option kind, per :rfc:`4782`. +QS = Option.Quick_Start_Response + + +def build(rate: 'int', payload: 'bytes' = b'') -> 'bytes': + """Construct a segment that carries one Quick-Start Response option.""" + return TCP(srcport=1, dstport=2, payload=payload, + options=[(QS, {'rate': rate, 'diff': 3, 'nonce': 0x1234})]).data + + +def parse(raw: 'bytes') -> 'TCP': + """Parse ``raw`` as a TCP segment.""" + return TCP(raw, len(raw)) + + +class QuickStartResponseTests(unittest.TestCase): + """Quick-Start Response against :rfc:`4782` Sections 3.1 and 4.2.""" + + def test_wire_octets_parse(self) -> 'None': + """Hand-built option octets parse, with and without a payload.""" + option = bytes([27, 8, 0x04, 3]) + (0x1234 << 2).to_bytes(4, 'big') + header = bytes.fromhex('0001000200000000000000007010ffff00000000') + for payload in (b'', b'xyz'): + with self.subTest(payload=payload): + tcp = parse(header + option + payload) + opt = tcp.info.options[QS] + self.assertEqual(opt.length, 8) + self.assertEqual(opt.req_rate, 640) + self.assertEqual(opt.ttl_diff, 3) + self.assertEqual(opt.nonce, 0x1234) + self.assertEqual(bytes(tcp.payload.data), payload) + + def test_round_trip(self) -> 'None': + """Construct, parse and construct again gives identical octets.""" + for rate in (0, 80, 640, 1_310_720): + for payload in (b'', b'xyz'): + with self.subTest(rate=rate, payload=payload): + raw = build(rate, payload) + tcp = parse(raw) + self.assertEqual(tcp.info.options[QS].req_rate, rate) + again = TCP(srcport=1, dstport=2, payload=payload, + options=tcp.info.options).data + self.assertEqual(again, raw) + self.assertEqual(bytes(tcp.payload.data), payload) + + 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``.""" + for rate, code in ((1, 0), (20, 0), (39, 0), (40, 0), (79, 0), (80, 1), (159, 1), (160, 2)): + 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) + + +if __name__ == '__main__': + unittest.main()