fix(tcp): Quick-Start Response reader reads a stray octet; maker fails below 40 kbps

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(

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

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 <pcapkit.protocols.transport.tcp.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 <pcapkit.protocols.transport.tcp.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()