diff --git a/pcapkit/protocols/data/transport/tcp.py b/pcapkit/protocols/data/transport/tcp.py index 4edfd4ae88..3d20b99911 100644 --- a/pcapkit/protocols/data/transport/tcp.py +++ b/pcapkit/protocols/data/transport/tcp.py @@ -66,10 +66,12 @@ class Flags(Data): syn: 'bool' #: Last packet from sender. fin: 'bool' + #: Reserved bits 4-6 of octet 12, between the Data Offset and ``NS``. + reserved: 'int' if TYPE_CHECKING: def __init__(self, ns: 'bool', cwr: 'bool', ece: 'bool', urg: 'bool', ack: 'bool', - psh: 'bool', rst: 'bool', syn: 'bool', fin: 'bool') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + psh: 'bool', rst: 'bool', syn: 'bool', fin: 'bool', reserved: 'int') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final @@ -98,6 +100,8 @@ class TCP(Protocol): if TYPE_CHECKING: #: TCP options. options: 'OrderedMultiDict[OptionNumber, Option]' + #: Octets of the option area after the End of Option List option. + padding: 'bytes' #: Connection control flags. connection: 'TCP_Flags' @@ -246,9 +250,11 @@ class PartialOrderServiceProfile(Option): start: 'bool' #: End flag. end: 'bool' + #: Reserved bits (``Filler``). + reserved: 'int' if TYPE_CHECKING: - def __init__(self, kind: 'OptionNumber', length: 'int', start: 'bool', end: 'bool') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, kind: 'OptionNumber', length: 'int', start: 'bool', end: 'bool', reserved: 'int') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final @@ -327,9 +333,13 @@ class QuickStartResponse(Option): ttl_diff: 'int' #: QS nonce. nonce: 'int' + #: Reserved bits before the rate request (``Resv.``). + reserved: 'int' + #: Reserved bits after the QS nonce (``R``). + nonce_reserved: 'int' if TYPE_CHECKING: - def __init__(self, kind: 'OptionNumber', length: 'int', req_rate: 'int', ttl_diff: 'int', nonce: 'int') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, kind: 'OptionNumber', length: 'int', req_rate: 'int', ttl_diff: 'int', nonce: 'int', reserved: 'int', nonce_reserved: 'int') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final @@ -338,9 +348,11 @@ class UserTimeout(Option): #: User timeout. timeout: 'timedelta' + #: Granularity (``G``): the timeout is in minutes if set, in seconds if not. + granularity: 'bool' if TYPE_CHECKING: - def __init__(self, kind: 'OptionNumber', length: 'int', timeout: 'timedelta') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, kind: 'OptionNumber', length: 'int', timeout: 'timedelta', granularity: 'bool') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final diff --git a/pcapkit/protocols/schema/transport/tcp.py b/pcapkit/protocols/schema/transport/tcp.py index 4fffa71fe6..576d0a7cbe 100644 --- a/pcapkit/protocols/schema/transport/tcp.py +++ b/pcapkit/protocols/schema/transport/tcp.py @@ -54,6 +54,8 @@ class OffsetFlag(TypedDict): #: Data offset. offset: int + #: Reserved bits. + reserved: int #: ECN-nonce concealment protection. ns: int @@ -84,10 +86,14 @@ class POCProfile(TypedDict): start: int #: End flag. end: int + #: Reserved bits (``Filler``). + reserved: int class QuickStartFlags(TypedDict): """TCP quick start flags.""" + #: Reserved bits. + reserved: int #: Rate request. rate: int @@ -96,6 +102,8 @@ class QuickStartNonce(TypedDict): #: Nonce. nonce: int + #: Reserved bits (``R``). + reserved: int class TimeoutInfo(TypedDict): """User timeout information.""" @@ -643,6 +651,7 @@ class PartialOrderServiceProfile(Option, code=Enum_Option.Partial_Order_Service_ profile: 'POCProfile' = BitField(length=1, namespace={ 'start': (0, 1), 'end': (1, 1), + 'reserved': (2, 6), }) if TYPE_CHECKING: @@ -721,6 +730,7 @@ class QuickStartResponse(Option, code=Enum_Option.Quick_Start_Response): #: Flags. flags: 'QuickStartFlags' = BitField(length=1, namespace={ + 'reserved': (0, 4), 'rate': (4, 4), }) #: TTL difference. @@ -728,6 +738,7 @@ class QuickStartResponse(Option, code=Enum_Option.Quick_Start_Response): #: QS nonce. nonce: 'QuickStartNonce' = BitField(length=4, namespace={ 'nonce': (0, 30), + 'reserved': (30, 2), }) if TYPE_CHECKING: @@ -1176,6 +1187,7 @@ class TCP(Schema): #: Data offset. offset: 'OffsetFlag' = BitField(length=1, namespace={ 'offset': (0, 4), + 'reserved': (4, 3), 'ns': (7, 1), }) #: TCP flags. diff --git a/pcapkit/protocols/transport/tcp.py b/pcapkit/protocols/transport/tcp.py index ad02495054..4c89d8d271 100644 --- a/pcapkit/protocols/transport/tcp.py +++ b/pcapkit/protocols/transport/tcp.py @@ -19,7 +19,7 @@ 4 32 ``tcp.seq`` Sequence Number 8 64 ``tcp.ack`` Acknowledgement Number (if ACK set) 12 96 ``tcp.hdr_len`` Data Offset - 12 100 Reserved (must be ``\\x00``) + 12 100 ``tcp.flags.reserved`` Reserved (must be ``\\x00``) 12 103 ``tcp.flags.ns`` ECN Concealment Protection (NS) 13 104 ``tcp.flags.cwr`` Congestion Window Reduced (CWR) 13 105 ``tcp.flags.ece`` ECN-Echo (ECE) @@ -480,6 +480,7 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_TCP': rst=bool(schema.flags['rst']), syn=bool(schema.flags['syn']), fin=bool(schema.flags['fin']), + reserved=schema.offset['reserved'], ), window_size=schema.window, checksum=schema.checksum, @@ -509,8 +510,13 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_TCP': _optl = tcp.hdr_len - 20 if _optl: + # NOTE: the octets after an End of Option List option are kept as + # read, so that :meth:`make` rebuilds the option area -- and the + # Data Offset computed from it -- at its original length. A schema + # built by hand rather than unpacked carries no ``padding``. tcp.__update__({ 'options': self._read_tcp_options(_optl), + 'padding': schema.get('padding', b''), }) return self._decode_next_layer(tcp, (tcp.srcport.port, tcp.dstport.port), length - tcp.hdr_len) @@ -534,6 +540,8 @@ def make(self, urgent: 'int' = 0, options: 'Optional[list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes] | Option]' = None, # pylint: disable=line-too-long payload: 'bytes | ProtocolBase | Schema' = b'', + padding: 'bytes' = b'', + reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_TCP': """Make (construct) packet data. @@ -556,6 +564,9 @@ def make(self, urgent: Urgent pointer. options: TCP options. payload: Payload of the packet. + padding: Octets after the End of Option List option; dropped + unless ``options`` end with one. + reserved: Reserved bits 4-6 of octet 12. **kwargs: Arbitrary keyword arguments. Returns: @@ -603,7 +614,7 @@ def make(self, self._flags = _flag if options is not None: - options_value, total_length = self._make_tcp_options(options) + options_value, total_length = self._make_tcp_options(options, padding) else: options_value, total_length = [], 0 @@ -616,6 +627,7 @@ def make(self, ack=ack_no, offset={ 'offset': offset, + 'reserved': reserved, 'ns': int(ns), }, flags=flags, @@ -705,6 +717,8 @@ def _make_data(cls, data: 'Data_TCP') -> 'dict[str, Any]': # type: ignore[overr 'checksum': data.checksum, 'urgent': data.urgent_pointer, 'options': getattr(data, 'options', None), + 'padding': getattr(data, 'padding', b''), + 'reserved': data.flags.reserved, 'payload': cls._make_payload(data), } @@ -1121,6 +1135,7 @@ def _read_mode_pocsp(self, schema: 'Schema_PartialOrderServiceProfile', *, optio length=schema.length, start=bool(schema.profile['start']), end=bool(schema.profile['end']), + reserved=schema.profile['reserved'], ) return data @@ -1358,6 +1373,8 @@ def _read_mode_qs(self, schema: 'Schema_QuickStartResponse', *, options: 'Option req_rate=40000 * (2 ** rate) / 1000 if rate > 0 else 0, ttl_diff=schema.diff, nonce=schema.nonce['nonce'], + reserved=schema.flags['reserved'], + nonce_reserved=schema.nonce['reserved'], ) return data @@ -1397,6 +1414,7 @@ def _read_mode_timeout(self, schema: 'Schema_UserTimeout', *, options: 'Option') kind=schema.kind, length=schema.length, timeout=time, + granularity=bool(schema.info['granularity']), ) return data @@ -2056,11 +2074,13 @@ def _read_mode_fastopen(self, schema: 'Schema_FastOpenCookie', *, options: 'Opti ) return data - def _make_tcp_options(self, options: 'list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes] | Option') -> 'tuple[list[Schema_Option | bytes], int]': + def _make_tcp_options(self, options: 'list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes] | Option', + padding: 'bytes' = b'') -> 'tuple[list[Schema_Option | bytes], int]': """Make options for TCP. Args: options: TCP options + padding: octets after the End of Option List option Returns: Tuple of options and total length of options. @@ -2068,9 +2088,11 @@ def _make_tcp_options(self, options: 'list[Schema_Option | tuple[Enum_Option, di Note: The options are emitted in the given order, ``NOP`` and ``EOOL`` included, so that a parsed option list re-emits the same octets. - Padding is added only once, after the last option, when the list - does not end on a 32-bit boundary: zero octets if the list already - ends with ``EOOL``, otherwise ``NOP`` options followed by an ``EOOL``. + ``padding`` follows them as given if they end with ``EOOL``, and is + dropped otherwise. Alignment is added only once, after that, when + the list does not end on a 32-bit boundary: zero octets if the list + already ends with ``EOOL``, otherwise ``NOP`` options followed by an + ``EOOL``. """ total_length = 0 @@ -2117,6 +2139,13 @@ def _make_tcp_options(self, options: 'list[Schema_Option | tuple[Enum_Option, di options_list.append(data) total_length += data_len + # NOTE: ``padding`` belongs after an End of Option List option; with + # that gone -- e.g. options edited after parsing -- it is dropped, and + # the Data Offset is computed from the options alone. + if padding and code == Enum_Option.End_of_Option_List: + options_list.append(padding) + total_length += len(padding) + # force alignment to 32-bit boundary if total_length % 4: pad_len = 4 - (total_length % 4) @@ -2388,6 +2417,7 @@ def _make_mode_poc(self, code: 'Enum_Option', opt: 'Optional[Data_PartialOrderCo def _make_mode_pocsp(self, code: 'Enum_Option', opt: 'Optional[Data_PartialOrderServiceProfile]' = None, *, start: 'bool' = False, end: 'bool' = False, + reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_PartialOrderServiceProfile': """Make TCP partial order connection service profile option. @@ -2396,6 +2426,7 @@ def _make_mode_pocsp(self, code: 'Enum_Option', opt: 'Optional[Data_PartialOrder opt: option data start: start partial order connection end: end partial order connection + reserved: reserved bits (``Filler``, 6 bits) **kwargs: arbitrary keyword arguments Returns: @@ -2405,6 +2436,7 @@ def _make_mode_pocsp(self, code: 'Enum_Option', opt: 'Optional[Data_PartialOrder if opt is not None: start = opt.start end = opt.end + reserved = opt.reserved return Schema_PartialOrderServiceProfile( kind=code, @@ -2412,6 +2444,7 @@ def _make_mode_pocsp(self, code: 'Enum_Option', opt: 'Optional[Data_PartialOrder profile={ 'start': start, 'end': end, + 'reserved': reserved, }, ) @@ -2572,6 +2605,8 @@ def _make_mode_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartRespo rate: 'int' = 0, diff: 'timedelta | int' = 0, nonce: 'int' = 0, + reserved: 'int' = 0, + nonce_reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_QuickStartResponse': """Make TCP quick start response option. @@ -2581,6 +2616,8 @@ def _make_mode_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartRespo rate: rate (in kbps) diff: time to live (in seconds) difference nonce: nonce value + reserved: reserved bits before the rate request (``Resv.``, 4 bits) + nonce_reserved: reserved bits after the nonce (``R``, 2 bits) **kwargs: arbitrary keyword arguments Returns: @@ -2591,6 +2628,8 @@ def _make_mode_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartRespo rate = opt.req_rate diff = opt.ttl_diff nonce = opt.nonce + 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 @@ -2602,39 +2641,57 @@ def _make_mode_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartRespo kind=code, length=8, flags={ + 'reserved': reserved, 'rate': rate_val, }, diff=diff_val, nonce={ 'nonce': nonce, + 'reserved': nonce_reserved, }, ) def _make_mode_timeout(self, code: 'Enum_Option', opt: 'Optional[Data_UserTimeout]' = None, *, timeout: 'timedelta | int' = 0, + granularity: 'Optional[bool]' = None, **kwargs: 'Any') -> 'Schema_UserTimeout': """Make TCP user timeout option. Args: code: option code opt: option data - timeout: timeout value + timeout: timeout value (in seconds, if an :obj:`int`) + granularity: preferred unit, minutes (``G`` set) or seconds; it is + kept only while the value still fits in it -- seconds within 15 + bits, minutes as a whole number of them. Otherwise, and if not + given, seconds are used when they fit in 15 bits, and minutes if + not. **kwargs: arbitrary keyword arguments Returns: Constructed option schema. + Raises: + ProtocolError: If the timeout does not fit in 15 bits of minutes. + """ if opt is not None: - timeout_val = math.floor(opt.timeout.total_seconds()) + seconds = math.floor(opt.timeout.total_seconds()) + granularity = opt.granularity else: - timeout_val = timeout if isinstance(timeout, int) else math.floor(timeout.total_seconds()) + seconds = timeout if isinstance(timeout, int) else math.floor(timeout.total_seconds()) - granularity = timeout_val.bit_length() > 15 - timeout_val = math.floor(timeout_val / 60) if granularity else timeout_val + # NOTE: a parsed (or requested) unit that the value no longer fits in -- + # e.g. a parsed seconds option edited past 15 bits, or a minutes option + # edited to a value that is not a whole number of minutes -- falls back + # to the size-based choice, as when no unit is given at all. + fits_seconds = seconds.bit_length() <= 15 + if granularity is None or (not granularity and not fits_seconds) or (granularity and seconds % 60): + granularity = not fits_seconds + timeout_val = math.floor(seconds / 60) if granularity else seconds if timeout_val.bit_length() > 15: - raise ProtocolError(f'TCP: [OptNo {code}] timeout value too large: {timeout}') + raise ProtocolError(f'TCP: [OptNo {code}] timeout value too large: {seconds} seconds') return Schema_UserTimeout( kind=code, diff --git a/tests/protocols/transport/test_tcp_mptcp_addaddr_unit.py b/tests/protocols/transport/test_tcp_mptcp_addaddr_unit.py index 5afb122b54..f04fbe8ef7 100644 --- a/tests/protocols/transport/test_tcp_mptcp_addaddr_unit.py +++ b/tests/protocols/transport/test_tcp_mptcp_addaddr_unit.py @@ -108,7 +108,7 @@ def build_tcp_segment(option_octets: 'bytes') -> 'bytes': schema = Schema_TCP( srcport=50000, dstport=80, seq=1, ack=0, - offset={'offset': 5 + len(option_octets) // 4, 'ns': 0}, + offset={'offset': 5 + len(option_octets) // 4, 'reserved': 0, 'ns': 0}, flags={'cwr': 0, 'ece': 0, 'urg': 0, 'ack': 0, 'psh': 0, 'rst': 0, 'syn': 1, 'fin': 0}, window=8192, checksum=b'\x00\x00', urgent=0, options=option_octets, diff --git a/tests/protocols/transport/test_tcp_mptcp_capable_length_unit.py b/tests/protocols/transport/test_tcp_mptcp_capable_length_unit.py index 61d76b8292..93125d2bf8 100644 --- a/tests/protocols/transport/test_tcp_mptcp_capable_length_unit.py +++ b/tests/protocols/transport/test_tcp_mptcp_capable_length_unit.py @@ -103,7 +103,7 @@ def build_tcp_segment(option_octets: 'bytes') -> 'bytes': schema = Schema_TCP( srcport=50000, dstport=80, seq=1, ack=0, - offset={'offset': 5 + len(option_octets) // 4, 'ns': 0}, + offset={'offset': 5 + len(option_octets) // 4, 'reserved': 0, 'ns': 0}, flags={'cwr': 0, 'ece': 0, 'urg': 0, 'ack': 0, 'psh': 0, 'rst': 0, 'syn': 1, 'fin': 0}, window=8192, checksum=b'\x00\x00', urgent=0, options=option_octets, diff --git a/tests/protocols/transport/test_tcp_mptcp_join_flag_ordering_unit.py b/tests/protocols/transport/test_tcp_mptcp_join_flag_ordering_unit.py index 6ab3faa1e7..34403ccc84 100644 --- a/tests/protocols/transport/test_tcp_mptcp_join_flag_ordering_unit.py +++ b/tests/protocols/transport/test_tcp_mptcp_join_flag_ordering_unit.py @@ -239,7 +239,7 @@ def build_tcp_segment(option_octets: 'bytes', *, syn: 'bool', ack: 'bool') -> 'b schema = Schema_TCP( srcport=50000, dstport=80, seq=1, ack=0, - offset={'offset': 5 + len(option_octets) // 4, 'ns': 0}, + offset={'offset': 5 + len(option_octets) // 4, 'reserved': 0, 'ns': 0}, flags={'cwr': 0, 'ece': 0, 'urg': 0, 'ack': int(ack), 'psh': 0, 'rst': 0, 'syn': int(syn), 'fin': 0}, window=8192, checksum=b'\x00\x00', urgent=0, @@ -593,7 +593,7 @@ def test_flags_are_assigned_before_the_options_are_built(self) -> None: source = inspect.getsource(TCP.make) assignment = source.index('self._flags = _flag') - build = source.index('self._make_tcp_options(options)') + build = source.index('self._make_tcp_options(options') self.assertLess( assignment, build, @@ -612,7 +612,7 @@ def test_the_offset_still_follows_the_option_build(self) -> None: from pcapkit.protocols.transport.tcp import TCP source = inspect.getsource(TCP.make) - build = source.index('self._make_tcp_options(options)') + build = source.index('self._make_tcp_options(options') offset = source.index('offset = math.ceil') self.assertLess(build, offset, diff --git a/tests/protocols/transport/test_tcp_mptcp_length_arithmetic_unit.py b/tests/protocols/transport/test_tcp_mptcp_length_arithmetic_unit.py index 62fa45058e..402b3dc739 100644 --- a/tests/protocols/transport/test_tcp_mptcp_length_arithmetic_unit.py +++ b/tests/protocols/transport/test_tcp_mptcp_length_arithmetic_unit.py @@ -208,7 +208,7 @@ def build_tcp_segment(option_octets: 'bytes', *, syn: 'bool' = True, schema = Schema_TCP( srcport=50000, dstport=80, seq=1, ack=0, - offset={'offset': 5 + len(option_octets) // 4, 'ns': 0}, + offset={'offset': 5 + len(option_octets) // 4, 'reserved': 0, 'ns': 0}, flags={'cwr': 0, 'ece': 0, 'urg': 0, 'ack': int(ack_flag), 'psh': 0, 'rst': 0, 'syn': int(syn), 'fin': 0}, window=8192, checksum=b'\x00\x00', urgent=0, diff --git a/tests/protocols/transport/test_tcp_ns_flag_roundtrip_unit.py b/tests/protocols/transport/test_tcp_ns_flag_roundtrip_unit.py index ee77af3a03..c3acd07994 100644 --- a/tests/protocols/transport/test_tcp_ns_flag_roundtrip_unit.py +++ b/tests/protocols/transport/test_tcp_ns_flag_roundtrip_unit.py @@ -53,12 +53,12 @@ def test_from_data_keeps_ns(self) -> None: self.assertIs(rebuilt.info.flags.ns, ns) def test_flags_field_order(self) -> None: - """``ns`` leads the flags, in wire order, ahead of ``cwr``.""" + """``ns`` leads the flags, in wire order, ahead of ``cwr``; ``reserved`` trails them.""" from pcapkit.protocols.transport.tcp import TCP flags = TCP(TCP(srcport=1, dstport=2).data).info.flags self.assertEqual(list(flags.keys()), - ['ns', 'cwr', 'ece', 'urg', 'ack', 'psh', 'rst', 'syn', 'fin']) + ['ns', 'cwr', 'ece', 'urg', 'ack', 'psh', 'rst', 'syn', 'fin', 'reserved']) if __name__ == '__main__': diff --git a/tests/protocols/transport/test_tcp_roundtrip_unit.py b/tests/protocols/transport/test_tcp_roundtrip_unit.py new file mode 100644 index 0000000000..48ac84a870 --- /dev/null +++ b/tests/protocols/transport/test_tcp_roundtrip_unit.py @@ -0,0 +1,236 @@ +# -*- coding: utf-8 -*- +"""TCP headers and options rebuild byte for byte. + +GitHub issues #1211, #1218 and #1219, all found by the transport round-trip +audit (#1202). Each case is a whole TCP segment, and +:meth:`~pcapkit.protocols.protocol.ProtocolBase.from_data` over the parsed +segment has to reproduce it: + +* #1211 -- the reserved bits of the header (octet 12, bits 4-6), of the POC-SP + option (``Filler``) and of the Quick-Start Response option (``Resv.`` and + ``R``) were written back as zero; +* #1218 -- the octets after an End of Option List option were dropped, and the + Data Offset was recomputed from what was left, shrinking the header; they + are kept now, but only while the options still end with that option, so an + edited option list still builds; +* #1219 -- the User Timeout option lost its ``G`` bit, the maker choosing the + granularity by the size of the value instead; the parsed unit is kept while + the value still fits in it. + +Every case builds its own octets in memory and reads no capture. :class:`TCP` +is imported inside each test, after +:func:`~tests._support.reimport_once_per_class`, not at module load. + +""" + +import datetime +import unittest + +from tests._support import reimport_once_per_class + +#: TCP header without its offset/flags octets: ports 1234 and 80, sequence 1, +#: acknowledgement 2; the window, checksum and urgent pointer follow them. +HEAD = bytes.fromhex('04d2' '0050' '00000001' '00000002') +TAIL = bytes.fromhex('03e8' 'abcd' '0000') + +#: ACK flag, the only one set. +ACK = 0x10 + +#: ``name -> (octet 12, option octets)``; ``None`` for octet 12 means a Data +#: Offset matching the options, with the reserved bits and ``NS`` clear. +SEGMENTS = { + # 1211 + 'header-reserved': (0x5e, ''), + 'header-reserved-ns': (0x5f, ''), + 'pocsp-reserved': (None, '0a033f00'), + 'qs-reserved': (None, '1b08834012345679'), + 'qs-reserved-all': (None, '1b08f340fffffffc'), + # 1218 + 'eool-then-options': (None, '00010101 020405b4'), + 'eool-then-zeros': (None, '020405b4 00000000'), + 'eool-then-octets': (None, '020405b4 00aabbcc'), + 'mss-nop-eool-padded': (None, '020405b4 0100 000000000000'), + # 1219 + 'uto-minutes': (None, '1c048005'), + 'uto-seconds': (None, '1c04012c'), + 'uto-minutes-small': (None, '1c048001'), +} + + +def segment(octet12: 'int | None', option: str, payload: bytes = b'PAY') -> bytes: + """Build a TCP segment carrying ``option`` and ``payload``.""" + opt = bytes.fromhex(option.replace(' ', '')) + assert len(opt) % 4 == 0, option + if octet12 is None: + octet12 = (5 + len(opt) // 4) << 4 + return HEAD + bytes([octet12, ACK]) + TAIL + opt + payload + + +class TestTCPRoundTrip(unittest.TestCase): + """Pin the wire form of the TCP header fields and options the audit flagged.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def parse(self, name: str) -> 'tuple[bytes, object]': + from pcapkit.protocols.transport.tcp import TCP + + raw = segment(*SEGMENTS[name]) + return raw, TCP(raw, len(raw)) + + def option(self, name: str, kind: str) -> 'object': + from pcapkit.const.tcp.option import Option + + _, tcp = self.parse(name) + return tcp.info.options[getattr(Option, kind)] + + def test_from_data_rebuilds_byte_for_byte(self) -> None: + from pcapkit.protocols.transport.tcp import TCP + + for name in SEGMENTS: + with self.subTest(segment=name): + raw, tcp = self.parse(name) + self.assertEqual(TCP.from_data(tcp.info).data.hex(), raw.hex()) + + def test_header_reserved_bits_are_kept(self) -> None: + for name, reserved in (('header-reserved', 0b111), ('header-reserved-ns', 0b111)): + with self.subTest(segment=name): + _, tcp = self.parse(name) + self.assertEqual(tcp.info.flags.reserved, reserved) + _, tcp = self.parse('header-reserved-ns') + self.assertTrue(tcp.info.flags.ns) + + def test_make_writes_the_header_reserved_bits(self) -> None: + from pcapkit.protocols.transport.tcp import TCP + + data = object.__new__(TCP).make(reserved=0b101, ns=True).pack() + self.assertEqual(data[12], 0x5b) + + def test_option_reserved_bits_are_kept(self) -> None: + pocsp = self.option('pocsp-reserved', 'Partial_Order_Service_Profile') + self.assertEqual((pocsp.start, pocsp.end, pocsp.reserved), (False, False, 0x3f)) + + qs = self.option('qs-reserved', 'Quick_Start_Response') + self.assertEqual((qs.reserved, qs.nonce_reserved), (0x8, 0b01)) + self.assertEqual(qs.nonce, 0x12345679 >> 2) + + qs = self.option('qs-reserved-all', 'Quick_Start_Response') + self.assertEqual((qs.reserved, qs.nonce_reserved), (0xf, 0b00)) + + def test_octets_after_eool_are_kept(self) -> None: + cases = { + 'eool-then-options': bytes.fromhex('010101020405b4'), + 'eool-then-zeros': bytes(3), + 'eool-then-octets': bytes.fromhex('aabbcc'), + } + for name, padding in cases.items(): + with self.subTest(segment=name): + raw, tcp = self.parse(name) + self.assertEqual(tcp.info.padding, padding) + self.assertEqual(tcp.info.hdr_len, raw[12] >> 4 << 2) + + def test_header_keeps_its_length_after_eool(self) -> None: + from pcapkit.protocols.transport.tcp import TCP + + raw, tcp = self.parse('eool-then-options') + rebuilt = TCP.from_data(tcp.info).data + self.assertEqual(rebuilt[12] >> 4, 7) + self.assertEqual(rebuilt[28:], b'PAY') + + def test_padding_is_dropped_once_the_eool_is_edited_out(self) -> None: + from pcapkit.const.tcp.option import Option + from pcapkit.corekit.multidict import OrderedMultiDict + from pcapkit.protocols.transport.tcp import TCP + + _, tcp = self.parse('mss-nop-eool-padded') + self.assertEqual(tcp.info.padding, bytes(6)) + mss, nop, eool = Option.Maximum_Segment_Size, Option.No_Operation, Option.End_of_Option_List + + #: ``removed kinds -> (Data Offset, option kinds read back)``; the + #: options left over are aligned with NOPs and an EOOL as for any list. + cases = { + 'eool': ({eool}, 7, [mss, nop, nop, nop, eool]), + 'nop-and-eool': ({nop, eool}, 6, [mss]), + 'every-option': ({mss, nop, eool}, 5, None), + } + for name, (removed, offset, kinds) in cases.items(): + with self.subTest(removed=name): + kept = OrderedMultiDict() + for kind, opt in tcp.info.options.items(multi=True): + if kind not in removed: + kept.add(kind, opt) + kwargs = TCP._make_data(tcp.info) + kwargs['options'] = kept + + rebuilt = TCP(**kwargs).data + self.assertEqual(rebuilt[12] >> 4, offset) + self.assertEqual(rebuilt[offset * 4:], b'PAY') + + info = TCP(rebuilt, len(rebuilt)).info + if kinds is None: + self.assertNotIn('options', info) + else: + self.assertEqual([kind for kind, _ in info.options.items(multi=True)], kinds) + + def test_make_drops_padding_without_eool(self) -> None: + from pcapkit.const.tcp.option import Option + from pcapkit.protocols.transport.tcp import TCP + + proto = object.__new__(TCP) + self.assertEqual(proto.make(padding=b'\x01').pack()[12] >> 4, 5) + made = proto.make(options=[(Option.Maximum_Segment_Size, {'mss': 1460})], padding=b'\x01') + self.assertEqual(made.pack()[12] >> 4, 6) + + def test_user_timeout_keeps_its_granularity(self) -> None: + cases = { + 'uto-minutes': (True, datetime.timedelta(minutes=5)), + 'uto-seconds': (False, datetime.timedelta(seconds=300)), + 'uto-minutes-small': (True, datetime.timedelta(minutes=1)), + } + for name, (granularity, timeout) in cases.items(): + with self.subTest(segment=name): + opt = self.option(name, 'User_Timeout_Option') + self.assertEqual((opt.granularity, opt.timeout), (granularity, timeout)) + + def test_make_honours_an_explicit_granularity(self) -> None: + from pcapkit.const.tcp.option import Option + from pcapkit.protocols.transport.tcp import TCP + + proto = object.__new__(TCP) + code = Option.User_Timeout_Option + minutes = proto._make_mode_timeout(code, timeout=300, granularity=True) + self.assertEqual(minutes.pack().hex(), '1c048005') + seconds = proto._make_mode_timeout(code, timeout=300) + self.assertEqual(seconds.pack().hex(), '1c04012c') + # 40000 seconds does not fit in 15 bits, so minutes are chosen: 666. + large = proto._make_mode_timeout(code, timeout=40000) + self.assertEqual(large.pack().hex(), '1c04829a') + large = proto._make_mode_timeout(code, timeout=40000, granularity=False) + self.assertEqual(large.pack().hex(), '1c04829a') + + def test_edited_user_timeout_switches_unit_when_it_no_longer_fits(self) -> None: + from pcapkit.protocols.data.transport.tcp import UserTimeout + from pcapkit.protocols.transport.tcp import TCP + from pcapkit.utilities.exceptions import ProtocolError + + proto = object.__new__(TCP) + cases = { + # parsed in seconds, edited past 15 bits of seconds -> minutes + 'uto-seconds': (datetime.timedelta(seconds=40000), '1c04829a'), + # parsed in minutes, edited to a part of a minute -> seconds + 'uto-minutes': (datetime.timedelta(seconds=90), '1c04005a'), + } + for name, (timeout, wire) in cases.items(): + with self.subTest(segment=name): + opt = self.option(name, 'User_Timeout_Option') + edited = UserTimeout(kind=opt.kind, length=opt.length, timeout=timeout, + granularity=opt.granularity) + self.assertEqual(proto._make_mode_timeout(opt.kind, edited).pack().hex(), wire) + + opt = self.option('uto-seconds', 'User_Timeout_Option') + with self.assertRaisesRegex(ProtocolError, r'too large: 1966080 seconds'): + proto._make_mode_timeout(opt.kind, timeout=0x8000 * 60) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/transport/test_tcp_udp_unit.py b/tests/protocols/transport/test_tcp_udp_unit.py index b88036d38f..4337e07c30 100644 --- a/tests/protocols/transport/test_tcp_udp_unit.py +++ b/tests/protocols/transport/test_tcp_udp_unit.py @@ -121,7 +121,8 @@ def test_tcp_index_and_make_data_preserve_flags_and_ports(self) -> None: from pcapkit.const.reg.transtype import TransType from pcapkit.protocols.transport.tcp import TCP - flags = DummyData(ns=False, cwr=False, ece=False, urg=False, ack=True, psh=True, rst=False, syn=False, fin=True) + flags = DummyData(ns=False, cwr=False, ece=False, urg=False, ack=True, psh=True, rst=False, syn=False, fin=True, + reserved=0) data = DummyData( srcport=443, dstport=59000, @@ -132,6 +133,7 @@ def test_tcp_index_and_make_data_preserve_flags_and_ports(self) -> None: checksum=b'\xfe\xed', urgent_pointer=0, options=None, + padding=b'', __next_type__=None, ) @@ -179,7 +181,7 @@ def test_tcp_properties_read_make_and_callable_registry_edges(self) -> None: dstport=AppType.get(59000, proto=TransportProtocol.tcp), seq=1, ack=2, - offset={'offset': 6, 'ns': 0}, + offset={'offset': 6, 'reserved': 0, 'ns': 0}, flags={'cwr': 0, 'ece': 0, 'urg': 0, 'ack': 1, 'psh': 0, 'rst': 0, 'syn': 1, 'fin': 0}, window=4096, @@ -205,7 +207,7 @@ def test_tcp_properties_read_make_and_callable_registry_edges(self) -> None: dstport=AppType.get(5353, proto=TransportProtocol.tcp), seq=3, ack=4, - offset={'offset': 5, 'ns': 0}, + offset={'offset': 5, 'reserved': 0, 'ns': 0}, flags={'cwr': 0, 'ece': 0, 'urg': 0, 'ack': 0, 'psh': 0, 'rst': 0, 'syn': 0, 'fin': 1}, window=1024, @@ -456,6 +458,7 @@ def test_tcp_option_constructors_cover_data_model_and_mapping_paths(self) -> Non length=3, start=True, end=True, + reserved=0, ), ).profile['end'], True) self.assertEqual(proto._make_mode_cc( @@ -492,13 +495,15 @@ def test_tcp_option_constructors_cover_data_model_and_mapping_paths(self) -> Non Option.Quick_Start_Response, tcp_data.QuickStartResponse(kind=Option.Quick_Start_Response, length=8, req_rate=80, - ttl_diff=7, nonce=9), + ttl_diff=7, nonce=9, + reserved=0, nonce_reserved=0), ).diff, 7) self.assertEqual(proto._make_mode_timeout( Option.User_Timeout_Option, tcp_data.UserTimeout(kind=Option.User_Timeout_Option, length=4, - timeout=datetime.timedelta(seconds=120)), + timeout=datetime.timedelta(seconds=120), + granularity=False), ).info['timeout'], 120) self.assertEqual(proto._make_mode_ao( Option.TCP_Authentication_Option, @@ -755,7 +760,7 @@ def assert_bad(reader, schema) -> None: ) profile = proto._read_mode_pocsp( PartialOrderServiceProfile(kind=Option.Partial_Order_Service_Profile, length=3, - profile={'start': 1, 'end': 0}), + profile={'start': 1, 'end': 0, 'reserved': 0}), options=options, ) self.assertTrue(profile.start) @@ -781,7 +786,8 @@ def assert_bad(reader, schema) -> None: ).digest, b'0' * 16) self.assertEqual(proto._read_mode_qs( QuickStartResponse(kind=Option.Quick_Start_Response, length=8, - flags={'rate': 1}, diff=4, nonce={'nonce': 5}), + flags={'reserved': 0, 'rate': 1}, diff=4, + nonce={'nonce': 5, 'reserved': 0}), options=options, ).req_rate, 80.0) self.assertEqual(proto._read_mode_timeout(