diff --git a/pcapkit/protocols/internet/ipv4.py b/pcapkit/protocols/internet/ipv4.py index d634ab111c..0aec17fb53 100644 --- a/pcapkit/protocols/internet/ipv4.py +++ b/pcapkit/protocols/internet/ipv4.py @@ -56,6 +56,7 @@ from pcapkit.const.ipv4.tos_thr import ToSThroughput as Enum_ToSThroughput from pcapkit.const.ipv4.ts_flag import TSFlag as Enum_TSFlag from pcapkit.const.reg.transtype import TransType as Enum_TransType +from pcapkit.corekit.fields.ipaddress import parse_ip_address from pcapkit.corekit.multidict import OrderedMultiDict from pcapkit.protocols.data.internet.ipv4 import EOOLOption as Data_EOOLOption from pcapkit.protocols.data.internet.ipv4 import ESECOption as Data_ESECOption @@ -283,6 +284,10 @@ def read(self, length: 'Optional[int]' = None, *, # pylint: disable=unused-argu Returns: Parsed packet data. + Raises: + ProtocolError: If the version is not 4, or if ``ihl`` declares + more header than the data holds. + """ if length is None: length = len(self) @@ -316,6 +321,13 @@ def read(self, length: 'Optional[int]' = None, *, # pylint: disable=unused-argu dst=schema.dst, ) + # NOTE: An ``ihl`` declaring more header than was captured leaves the + # missing octets to read as zeros, which ``from_data`` then rebuilds; + # rejecting it keeps the octets as captured inside the parent (:issue:`1404`). + if len(self) < ipv4.hdr_len: + raise ProtocolError(f'{self.alias}: header length {ipv4.hdr_len} runs past the end of ' + f'the data ({len(self)} octets)') + _optl = ipv4.hdr_len - 20 if _optl: options = self._read_ipv4_options(_optl) @@ -1606,7 +1618,7 @@ def _make_opt_lsr(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_LSROpt def _make_opt_ts(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_TSOption]' = None, *, counts: 'int' = 5, overflow: 'int' = 0, - timestamp: 'Optional[list[int | timedelta] | dict[IPv4Address, int | timedelta]]' = None, + timestamp: 'Optional[list[int | timedelta] | dict[IPv4Address | str | int, int | timedelta]]' = None, remaining: 'Optional[list[int]]' = None, **kwargs: 'Any') -> 'Schema_TSOption': """Make IPv4 Timestamp (``TS``) option. @@ -1615,7 +1627,10 @@ def _make_opt_ts(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_TSOptio kind: option type code option: option data counts: maximum number of timestamps to record - timestamp: list of timestamps, i.e. the slots before the pointer + timestamp: list of timestamps, i.e. the slots before the pointer, or + a mapping of the addresses to their timestamps; an address may + be anything :func:`~pcapkit.corekit.fields.ipaddress.parse_ip_address` + accepts for IPv4 remaining: raw 32-bit words for the slots at or beyond the pointer; the slots left over up to ``counts`` are zero **kwargs: arbitrary keyword arguments @@ -1623,6 +1638,10 @@ def _make_opt_ts(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_TSOptio Returns: Constructed option schema. + Raises: + FieldValueError: If an address is a :obj:`bool` or is not a valid + IPv4 address. + """ if option is not None: ts_list = [] # type: list[int] @@ -1638,7 +1657,8 @@ def _make_opt_ts(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_TSOptio ts_list.append(ts) else: for ip, ts in option.timestamp.items(True): - ts_list.append(int(ip)) + ts_list.append(int(parse_ip_address(ip, f'{self.alias}: [OptNo {kind}] timestamp address', + version=4))) if not isinstance(ts, int): ts = ts // datetime.timedelta(milliseconds=1) if ts.bit_length() > 31: @@ -1676,12 +1696,13 @@ def _make_opt_ts(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_TSOptio counts = min(4, counts) # 4 is the maximum number of timestamps length = 4 + counts * 8 - for index, (ip, ts) in enumerate(timestamp.items()): + for index, (addr, ts) in enumerate(timestamp.items()): if index >= counts: warn(f'{self.alias}: [OptNo {kind}] too many timestamps: {len(timestamp)}', ProtocolWarning) break - ts_list.append(int(ip)) + ts_list.append(int(parse_ip_address(addr, f'{self.alias}: [OptNo {kind}] timestamp address', + version=4))) if not isinstance(ts, int): ts = ts // datetime.timedelta(milliseconds=1) if ts.bit_length() > 31: diff --git a/pcapkit/protocols/transport/tcp.py b/pcapkit/protocols/transport/tcp.py index 2fe44ecd35..6e0345a56c 100644 --- a/pcapkit/protocols/transport/tcp.py +++ b/pcapkit/protocols/transport/tcp.py @@ -459,6 +459,10 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_TCP': Returns: Parsed packet data. + Raises: + ProtocolError: If the Data Offset declares more header than the + data holds. + """ if length is None: length = len(self) @@ -508,6 +512,13 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_TCP': 'connection': self._flags, }) + # NOTE: A Data Offset declaring more header than was captured leaves the + # missing octets to read as zeros, which ``from_data`` then rebuilds; + # rejecting it keeps the octets as captured inside the parent (:issue:`1404`). + if len(self) < tcp.hdr_len: + raise ProtocolError(f'{self.alias}: header length {tcp.hdr_len} runs past the end of ' + f'the data ({len(self)} octets)') + _optl = tcp.hdr_len - 20 if _optl: # NOTE: the octets after an End of Option List option are kept as diff --git a/tests/protocols/internet/test_ipv4_unit.py b/tests/protocols/internet/test_ipv4_unit.py index bd3f45585f..73d9de850e 100644 --- a/tests/protocols/internet/test_ipv4_unit.py +++ b/tests/protocols/internet/test_ipv4_unit.py @@ -1981,52 +1981,60 @@ def test_ipv4_schema_helpers_and_post_process_branches(self) -> None: self.assertEqual(tuple(unknown.timestamp), (1,)) warn.assert_called_once() - def test_an_option_area_longer_than_the_datagram_still_parses(self) -> None: - """An ``ihl`` promising more options than are there is tolerated. + def test_an_option_area_longer_than_the_datagram_is_refused(self) -> None: + """An ``ihl`` promising more options than are there is refused. C.f. #431, #1404. The header below sets ``ihl`` to 10 -- a 20-octet option area -- and stops - after the fixed 20 octets, so there are no option octets at all. Reading - past them yields ``b''``, which decodes the option number as 0, and 0 is - IPv4's end-of-option-list, so the option loop breaks there and reports the - whole area as padding. That is how a datagram cut short by the snapshot - length parses at all, and it is why :meth:`OptionField.unpack + after the fixed 20 octets, so there are no option octets at all. Read by + the schema, the option area yields ``b''``, which decodes the option + number as 0, and 0 is IPv4's end-of-option-list, so the option loop breaks + there. That is why :meth:`OptionField.unpack ` checks each option's progress *after* its end-of-option-list break rather than before: checking first turns every such header into an error. C.f. #431. + :class:`~pcapkit.protocols.internet.ipv4.IPv4` itself refuses the header + with :exc:`~pcapkit.utilities.exceptions.ProtocolError`, since the 20 + missing octets would otherwise be rebuilt as zeros (#1404). + """ from pcapkit.const.ipv4.option_number import OptionNumber from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.schema.internet.ipv4 import IPv4 as Schema_IPv4 + from pcapkit.utilities.exceptions import ProtocolError from tests._support import time_limit raw = bytes.fromhex('4a00001800010000400600000a0000010a000002') with time_limit(): - proto = IPv4(raw, len(raw)) - - self.assertEqual(proto.info.hdr_len, 40) + schema = Schema_IPv4.unpack(raw) self.assertEqual( - [(code, opt.length) for code, opt in proto.info.options.items(multi=True)], + [(opt.type, opt.length) for opt in schema.options], [(OptionNumber.EOOL, 1)], ) + with self.assertRaisesRegex(ProtocolError, 'header length 40 runs past the end of the data'): + IPv4(raw, len(raw)) + def test_a_truncated_option_still_parses_its_declared_length(self) -> None: """A capture cut short mid-option is refused alone, and kept as captured in a frame. C.f. #431, #572, #1325. - :meth:`test_an_option_area_longer_than_the_datagram_still_parses` above + :meth:`test_an_option_area_longer_than_the_datagram_is_refused` above pins the *empty*-tail half of the #431 accommodation: an option area that runs out before it starts. This is the other half -- an option that *does* start, declares more data than the capture holds, and runs out partway through. - The header below sets ``ihl`` to 9 -- 16 declared octets of options -- - for an unassigned option (code 31) declaring ``length=12``, with only - 6 of its 10 data octets behind it. Those 4 missing octets used to be - read as zeros and rebuilt as if captured. :meth:`OptionField.unpack - ` now refuses - the option with :exc:`~pcapkit.utilities.exceptions.ProtocolError` - instead (#1325). Inside a frame the datagram is kept as the octets - captured, a :class:`~pcapkit.protocols.misc.raw.Raw` payload carrying - that error, and rebuilds byte for byte. + The header below sets ``ihl`` to 7 -- 8 octets of options, all present, + with 8 octets of payload behind them -- for an unassigned option (code + 31) declaring ``length=12``, with only 6 of its 10 data octets inside + the option area. Those 4 missing octets used to be read as zeros and + rebuilt as if captured. The header itself is complete, so the + header-length check (#1404) passes, and :meth:`OptionField.unpack + ` refuses the + option with :exc:`~pcapkit.utilities.exceptions.ProtocolError` (#1325). + Inside a frame the datagram is kept as the octets captured, a + :class:`~pcapkit.protocols.misc.raw.Raw` payload carrying that error, + and rebuilds byte for byte. """ from pcapkit.const.ipv4.option_number import OptionNumber @@ -2038,18 +2046,18 @@ def test_a_truncated_option_still_parses_its_declared_length(self) -> None: custom = OptionNumber.get(31) trailing = bytes.fromhex('aabbccddeeff') - raw = (bytes.fromhex('4900001c00010000400600000a0000010a000002') + - bytes([custom, 12]) + trailing) + raw = (bytes.fromhex('4700002400010000400600000a0000010a000002') + + bytes([custom, 12]) + trailing + bytes(8)) with time_limit(), warnings.catch_warnings(): warnings.simplefilter('ignore') - with self.assertRaisesRegex(ProtocolError, 'runs past the end of the data'): + with self.assertRaisesRegex(ProtocolError, 'has an option that runs past the end of the data'): IPv4(raw, len(raw)) frame = bytes.fromhex('0123456789ab' 'fedcba987654' '0800') + raw proto = Ethernet(frame, len(frame)) self.assertIsInstance(proto.payload, Raw) - self.assertIn('runs past the end of the data', str(proto.payload.info.error)) + self.assertIn('has an option that runs past the end of the data', str(proto.payload.info.error)) # the option octets are kept as captured, with nothing synthesised self.assertEqual(proto.payload.data, raw) self.assertEqual(Ethernet.from_data(proto.info).data, frame) diff --git a/tests/protocols/test_declared_length_overrun_unit.py b/tests/protocols/test_declared_length_overrun_unit.py index 37ea41a87b..2055b83ac3 100644 --- a/tests/protocols/test_declared_length_overrun_unit.py +++ b/tests/protocols/test_declared_length_overrun_unit.py @@ -14,7 +14,10 @@ the parent rebuilds byte for byte. An option that begins at the end of the data reads nothing at all, and is still -left to the end-of-option-list handling an over-long option area relies on. +left to the end-of-option-list handling an over-long option area relies on. An +IPv4 or TCP header declaring an option area past the data is rejected before its +options are read (GitHub issue #1404), and its parent likewise keeps it as +captured. Every case builds its own octets in memory and reads no capture. @@ -74,9 +77,10 @@ def _ipv6(proto: int, payload: bytes) -> bytes: bytes.fromhex('01000020 00000001 00010000 00010001 00000001 000c0010 00050000'))), # an unassigned kind declaring 12 octets, 8 present 'TCP unassigned': ('pcapkit.protocols.transport.tcp', 'TCP', _tcp(bytes.fromhex('4f0caabb ccddeeff'))), - # an unassigned option declaring 12 octets, 8 present in a 16-octet area + # an unassigned option declaring 12 octets, 8 present in a complete + # 8-octet area, with 8 octets of payload behind it 'IPv4 unassigned': ('pcapkit.protocols.internet.ipv4', 'IPv4', - _ipv4(bytes.fromhex('1f0caabb ccddeeff'), ihl=9)), + _ipv4(bytes.fromhex('1f0caabb ccddeeff'), ihl=7, payload=bytes(8))), } #: The same option areas with every option complete. @@ -128,7 +132,7 @@ def test_an_overrunning_option_raises_protocol_error(self) -> None: protocol = self._protocol(module, name) with warnings.catch_warnings(): warnings.simplefilter('ignore') - with self.assertRaisesRegex(ProtocolError, 'runs past the end of the data'): + with self.assertRaisesRegex(ProtocolError, 'has an option that runs past the end of the data'): protocol(data, len(data)) def test_a_complete_option_still_rebuilds_byte_for_byte(self) -> None: @@ -150,24 +154,33 @@ def test_an_overrunning_layer_is_kept_as_captured_by_its_parent(self) -> None: if name == 'IPv6': # an extension header is kept as raw extension data - self.assertIn('runs past the end of the data', str(parsed.info.ext.error)) + self.assertIn('has an option that runs past the end of the data', str(parsed.info.ext.error)) else: # any other layer is kept as a raw payload, as captured self.assertIsInstance(parsed.payload, Raw) - self.assertIn('runs past the end of the data', str(parsed.payload.info.error)) + self.assertIn('has an option that runs past the end of the data', str(parsed.payload.info.error)) self.assertEqual(parsed.payload.data, OVERRUNS[case][2]) self.assertEqual(protocol.from_data(parsed.info).data, frame) - def test_an_option_area_declared_past_the_data_still_ends_at_eool(self) -> None: + def test_an_option_area_declared_past_the_data_is_kept_as_captured(self) -> None: from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.link.ethernet import Ethernet + from pcapkit.protocols.misc.raw import Raw + from pcapkit.utilities.exceptions import ProtocolError - # ``ihl`` declares 20 octets of options with none present: the first - # option begins at the end of the data, so it is end-of-option-list. + # ``ihl`` declares 20 octets of options with none present: reading the + # option area would zero-fill all 20 of them on rebuild (issue #1404). data = bytes.fromhex('4a00001800010000400600000a0000010a000002') + with self.assertRaisesRegex(ProtocolError, 'header length 40 runs past the end of the data'): + IPv4(data, len(data)) + + frame = bytes.fromhex('0123456789ab fedcba987654 0800') + data with warnings.catch_warnings(): warnings.simplefilter('ignore') - parsed = IPv4(data, len(data)) - self.assertEqual(parsed.info.src.compressed, '10.0.0.1') + parsed = Ethernet(frame, len(frame)) + self.assertIsInstance(parsed.payload, Raw) + self.assertEqual(parsed.payload.data, data) + self.assertEqual(Ethernet.from_data(parsed.info).data, frame) if __name__ == '__main__': diff --git a/tests/protocols/test_ipv4_tcp_header_overrun_unit.py b/tests/protocols/test_ipv4_tcp_header_overrun_unit.py new file mode 100644 index 0000000000..c2d32a052c --- /dev/null +++ b/tests/protocols/test_ipv4_tcp_header_overrun_unit.py @@ -0,0 +1,161 @@ +# -*- coding: utf-8 -*- +"""A header length declared past the data, and TS address keys given as strings. + +GitHub issue #1404: an IPv4 ``ihl`` or a TCP Data Offset declaring more header +than the data holds read the missing octets as zeros, and ``from_data`` rebuilt +them -- 34 octets in and 38 out for an Ethernet frame whose IPv4 ``ihl`` is 6 +with only 20 header octets present. Such a header now raises +:exc:`~pcapkit.utilities.exceptions.ProtocolError`, so its parent keeps the +octets as a :class:`~pcapkit.protocols.misc.raw.Raw` payload and rebuilds them +byte for byte. + +GitHub issue #1402: ``IPv4.make()`` called :func:`int` on each key of a TS +option's ``timestamp`` mapping, so a dotted-quad string key raised a bare +:exc:`ValueError`. The keys are now converted as addresses, and a :obj:`bool` +key is rejected with :exc:`~pcapkit.utilities.exceptions.FieldValueError`. + +Every case builds its own octets in memory and reads no capture. + +""" + +import struct +import unittest +import warnings + +from tests._support import reimport_once_per_class + +_ETH = bytes.fromhex('0123456789ab fedcba987654 0800') + + +def _ipv4(ihl: int, proto: int, body: bytes) -> bytes: + """An IPv4 header declaring ``ihl``, followed by ``body``.""" + return (bytes([0x40 | ihl, 0]) + struct.pack('>H', 20 + len(body)) + + bytes.fromhex('00010000 40') + bytes([proto]) + bytes(2) + + bytes.fromhex('0a000001 0a000002') + body) + + +def _tcp(offset: int, opts: bytes = b'') -> bytes: + """A TCP header declaring ``offset``, followed by ``opts``.""" + return struct.pack('>HHIIBBHHH', 1, 2, 0, 0, offset << 4, 0x10, 0, 0, 0) + opts + + +#: Layers whose declared header length runs past the data: (module, class, octets). +SHORT = { + # issue #1404: options declared, none present + 'IPv4 ihl=6, 20 octets': ('pcapkit.protocols.internet.ipv4', 'IPv4', _ipv4(6, 253, b'')), + 'IPv4 ihl=7, 24 octets': ('pcapkit.protocols.internet.ipv4', 'IPv4', _ipv4(7, 253, bytes.fromhex('01010101'))), + 'TCP offset=6, 20 octets': ('pcapkit.protocols.transport.tcp', 'TCP', _tcp(6)), + 'TCP offset=10, 28 octets': ('pcapkit.protocols.transport.tcp', 'TCP', _tcp(10, bytes.fromhex('0101010101010101'))), +} + +#: Each short layer inside its parent: (parent module, parent class, frame). +IN_PARENT = { + 'IPv4 ihl=6, 20 octets': ('pcapkit.protocols.link.ethernet', 'Ethernet', _ETH + SHORT['IPv4 ihl=6, 20 octets'][2]), + 'IPv4 ihl=7, 24 octets': ('pcapkit.protocols.link.ethernet', 'Ethernet', _ETH + SHORT['IPv4 ihl=7, 24 octets'][2]), + 'TCP offset=6, 20 octets': ('pcapkit.protocols.internet.ipv4', 'IPv4', + _ipv4(5, 6, SHORT['TCP offset=6, 20 octets'][2])), + 'TCP offset=10, 28 octets': ('pcapkit.protocols.internet.ipv4', 'IPv4', + _ipv4(5, 6, SHORT['TCP offset=10, 28 octets'][2])), +} + + +class TestHeaderLengthOverrun(unittest.TestCase): + """Pin the rejection of a header length running past the end of its data.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + @staticmethod + def _protocol(module: str, name: str) -> type: + import importlib + + return getattr(importlib.import_module(module), name) + + def test_a_short_header_raises_protocol_error(self) -> None: + from pcapkit.utilities.exceptions import ProtocolError + + for case, (module, name, data) in SHORT.items(): + with self.subTest(case=case): + protocol = self._protocol(module, name) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + with self.assertRaisesRegex(ProtocolError, 'header length .* runs past the end of the data'): + protocol(data, len(data)) + + def test_a_short_header_is_kept_as_captured_by_its_parent(self) -> None: + from pcapkit.protocols.misc.raw import Raw + + for case, (module, name, frame) in IN_PARENT.items(): + with self.subTest(case=case): + protocol = self._protocol(module, name) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + parsed = protocol(frame, len(frame)) + self.assertIsInstance(parsed.payload, Raw) + self.assertEqual(parsed.payload.data, SHORT[case][2]) + self.assertEqual(protocol.from_data(parsed.info).data, frame) + + def test_a_complete_header_still_rebuilds_byte_for_byte(self) -> None: + # the same option areas, every declared octet present + for module, name, data in (('pcapkit.protocols.internet.ipv4', 'IPv4', _ipv4(6, 253, bytes(4))), + ('pcapkit.protocols.transport.tcp', 'TCP', _tcp(6, bytes(4)))): + with self.subTest(name=name): + protocol = self._protocol(module, name) + parsed = protocol(data, len(data)) + self.assertEqual(protocol.from_data(parsed.info).data, data) + + +class TestTimestampAddressKeys(unittest.TestCase): + """Pin the conversion of a TS option's address keys on ``make()``.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _make(self, key: object) -> bytes: + from datetime import timedelta + + from pcapkit.const.ipv4.option_number import OptionNumber + from pcapkit.protocols.internet.ipv4 import IPv4 + + return IPv4(options=[(OptionNumber.TS, {'timestamp': {key: timedelta(milliseconds=1)}})], + protocol=253).data + + def test_a_string_key_builds_like_an_address_key(self) -> None: + from ipaddress import IPv4Address + + from pcapkit.protocols.internet.ipv4 import IPv4 + + expected = self._make(IPv4Address('1.2.3.4')) + data = self._make('1.2.3.4') + self.assertEqual(data, expected) + self.assertEqual(data[20:32], bytes.fromhex('44240d01 01020304 00000001')) + self.assertEqual(IPv4(data, len(data)).data, data) + + def test_a_string_key_in_parsed_data_rebuilds(self) -> None: + from datetime import timedelta + from ipaddress import IPv4Address + + from pcapkit.corekit.multidict import OrderedMultiDict + from pcapkit.protocols.internet.ipv4 import IPv4 + + data = self._make(IPv4Address('1.2.3.4')) + parsed = IPv4(data, len(data)) + option = next(iter(parsed.info.options.values())) + option.__update__(timestamp=OrderedMultiDict([('1.2.3.4', timedelta(milliseconds=1))])) + self.assertEqual(IPv4.from_data(parsed.info).data, data) + + def test_a_bool_key_raises_field_value_error(self) -> None: + from pcapkit.utilities.exceptions import FieldValueError + + with self.assertRaisesRegex(FieldValueError, 'timestamp address: must not be a bool'): + self._make(True) + + def test_an_invalid_key_raises_field_value_error(self) -> None: + from pcapkit.utilities.exceptions import FieldValueError + + with self.assertRaisesRegex(FieldValueError, 'timestamp address'): + self._make('1.2.3') + + +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 cbf6a40e51..cae9d97a34 100644 --- a/tests/protocols/transport/test_tcp_udp_unit.py +++ b/tests/protocols/transport/test_tcp_udp_unit.py @@ -1350,41 +1350,44 @@ def segment(data_offset: 'int', options: 'bytes') -> 'bytes': with time_limit(): TCP(bad, len(bad)) - def test_an_option_area_longer_than_the_segment_still_parses(self) -> None: - """A data offset promising more options than are there is tolerated. + def test_an_option_area_longer_than_the_segment_is_refused(self) -> None: + """A data offset promising more options than are there is refused. C.f. #431, #1404. The segment below declares a data offset of 10 -- a 20-octet option area -- - and carries four option octets. Reading past them yields ``b''``, which - decodes the option kind as 0, and 0 is TCP's end-of-option-list, so the - option loop breaks there and reports the remaining 16 octets as padding. - That is how a capture cut short by the snapshot length parses at all, and - it is why :meth:`OptionField.unpack + and carries four option octets. Read by the schema, the option area yields + ``b''`` past them, which decodes the option kind as 0, and 0 is TCP's + end-of-option-list, so the option loop breaks there. That is why + :meth:`OptionField.unpack ` checks each option's progress *after* its end-of-option-list break rather than before: checking first turns every such segment into an error. + :class:`~pcapkit.protocols.transport.tcp.TCP` itself refuses the segment + with :exc:`~pcapkit.utilities.exceptions.ProtocolError`, since the 16 + missing octets would otherwise be rebuilt as zeros (#1404). + """ from pcapkit.const.tcp.option import Option + from pcapkit.protocols.schema.transport.tcp import TCP as Schema_TCP from pcapkit.protocols.transport.tcp import TCP + from pcapkit.utilities.exceptions import ProtocolError from tests._support import time_limit raw = bytes.fromhex('00501f900000000100000002a002ffff00000000020405b4') with time_limit(): - proto = TCP(raw, len(raw)) - - self.assertEqual(proto.info.hdr_len, 40) + schema = Schema_TCP.unpack(raw) self.assertEqual( - [(code, opt.length) for code, opt in proto.info.options.items(multi=True)], + [(opt.kind, opt.length) for opt in schema.options], [(Option.Maximum_Segment_Size, 4), (Option.End_of_Option_List, 1)], ) - mss = next(opt for code, opt in proto.info.options.items(multi=True) - if code == Option.Maximum_Segment_Size) - self.assertEqual(mss.mss, 1460) + + with self.assertRaisesRegex(ProtocolError, 'header length 40 runs past the end of the data'): + TCP(raw, len(raw)) def test_a_truncated_option_still_parses_its_declared_length(self) -> None: """An option declaring more data than its option area holds is refused alone, and kept in a datagram. C.f. #431, #572, #1325. - :meth:`test_an_option_area_longer_than_the_segment_still_parses` above + :meth:`test_an_option_area_longer_than_the_segment_is_refused` above pins the *empty*-tail half of the #431 accommodation: an option area that runs out before it starts, so the type byte decodes as 0 and the loop reads end-of-option-list. This is the other half -- an option that @@ -1426,12 +1429,12 @@ def segment(data_offset: 'int', options: 'bytes') -> 'bytes': + bytes.fromhex('000100004006' '0000' '0a000001' '0a000002') + raw) with time_limit(), warnings.catch_warnings(): warnings.simplefilter('ignore') - with self.assertRaisesRegex(ProtocolError, 'runs past the end of the data'): + with self.assertRaisesRegex(ProtocolError, 'has an option that runs past the end of the data'): TCP(raw, len(raw)) proto = IPv4(datagram, len(datagram)) self.assertIsInstance(proto.payload, Raw) - self.assertIn('runs past the end of the data', str(proto.payload.info.error)) + self.assertIn('has an option that runs past the end of the data', str(proto.payload.info.error)) # the option octets are kept as captured, with nothing synthesised self.assertEqual(proto.payload.data, raw) self.assertEqual(IPv4.from_data(proto.info).data, datagram)