diff --git a/pcapkit/corekit/fields/collections.py b/pcapkit/corekit/fields/collections.py index cbe2e06def..4465eac4c2 100644 --- a/pcapkit/corekit/fields/collections.py +++ b/pcapkit/corekit/fields/collections.py @@ -126,6 +126,10 @@ def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> 'byte Returns: Unpacked field value. + Raises: + FieldValueError: If the items overrun the field, or if a schema item + consumes nothing from ``buffer`` -- see the note below. + """ length = self._length if isinstance(buffer, bytes): @@ -139,6 +143,24 @@ def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> 'byte from pcapkit.corekit.fields.misc import SchemaField is_schema = isinstance(self._item_type, SchemaField) + # NOTE: The item-typed branch below sizes each item by ``field.length``, + # which is what it read, but the schema branch sizes it by ``len(data)``, + # which is only what the schema *recorded*. A schema reading a stream that + # has already run out records nothing, so ``length -= len(data)`` makes no + # progress and the loop spins forever. Reachable from a TCP segment: a + # ``SACK`` option declaring more octets than the option area holds leaves + # ``sack``'s ``ListField`` reading ``SACKBlock`` off an exhausted stream. + # Remembering where the previous item ended is what bounds the iteration + # count, since it does not depend on what the schema reports. C.f. #431, + # which is the same defect in the ``OptionField`` subclass. + # + # ``start`` is where the field itself begins. The comparison needs stream + # positions, but the diagnostic wants an offset into the field, and the two + # only coincide when the field happens to be reading from the front of its + # stream -- which it does when handed a :obj:`bytes` buffer and does not + # when handed a live file. + start = offset = file.tell() + temp = [] # type: list[_TL] while length > 0: field = self._item_type(packet) @@ -146,6 +168,22 @@ def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> 'byte if is_schema: data = cast('SchemaField', self._item_type).unpack(file, packet) + end = file.tell() + if end <= offset: + # NOTE: ``len(temp)`` counts the items already parsed, so it + # names the failing one as a count rather than as an ordinal -- + # "after 2 item(s)" rather than "item 2", which would read as + # the second item when it is the third. The ``OptionField`` + # message below names the option code in this slot and so has + # no index to be read either way. + raise FieldValueError( + f'Field {self.name} has an item that consumed no data: ' + f'after {len(temp)} item(s), at offset {offset - start} of ' + f'{self._length}, with {length} octet(s) of the field ' + f'left to parse' + ) + offset = end + length -= len(data) if length < 0: raise FieldValueError(f'Field {self.name} has invalid length.') @@ -296,6 +334,10 @@ def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> 'list as the remaining length to the ``packet`` argument such that the next fields can be aware of such informations. + Raises: + FieldValueError: If an option consumes nothing from ``buffer``, since + the loop below has then no way to get past it. + """ length = self._length if isinstance(buffer, bytes): @@ -303,6 +345,30 @@ def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> 'list else: file = buffer + # NOTE: The loop below sizes each option by ``len(data)`` -- the size of + # the schema the option reported -- and that is not always the number of + # octets the option took from ``file``. The two part company for a schema + # whose ``post_process`` returns a *nested* schema, since ``len(data)`` + # then measures the nested schema rather than what the outer one read. So + # ``len(data)`` cannot be the loop's progress measure: an option that + # over-reads leaves ``length`` above zero with ``file`` already exhausted, + # every field of the next option reads ``b''``, and an option that read + # nothing reports ``len(data) == 0`` and leaves ``length`` untouched -- + # which spins forever, with no exception and no diagnostic. C.f. #431. + # + # Remembering where the previous option ended gives the loop a measure of + # progress that does not depend on what a schema reports, and one octet of + # it per iteration is what bounds the iteration count. ``length`` is still + # decremented by ``len(data)``, so that an option area which parses today + # parses identically. + # + # ``start`` is where the option area itself begins. The comparison needs + # stream positions, but the diagnostic wants an offset into the area, and + # the two only coincide when the field happens to be reading from the front + # of its stream -- which it does when handed a :obj:`bytes` buffer and does + # not when handed a live file. + start = offset = file.tell() + # make a copy of the ``packet`` dict so that we can include # parsed option schema in the ``packet`` dict new_packet = packet.copy() @@ -362,5 +428,32 @@ def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> 'list if code == self._eool: break + # NOTE: The progress check comes *after* the end-of-option-list break, + # and that order is not incidental. An area declared longer than the + # octets behind it -- an over-long ``ihl``, or a capture cut short by + # the snapshot length -- exhausts ``file`` early, and the exhausted + # read then decodes the type field as 0. For the IPv4, TCP and PCAP-NG + # registries 0 *is* the end-of-option-list code, so the break above has + # always absorbed that case and reported the rest of the area as + # padding. Checking progress first turned all of those into errors: + # measured on ``IPv4(bytes.fromhex('4a00001800010000400600000a0000010a000002'))``, + # 20 octets of options declared with none present, and on a TCP segment + # with a data offset of 10 and four option octets, both of which parse + # on ``main``. + # + # The registries that spin are the ones where 0 is *not* the + # end-of-option-list code, so they never reach the break: HOPOPT, + # IPv6-Opts and MH read 0 as ``Pad1``, HIP as an unassigned parameter, + # SCTP as a DATA chunk. Those are what this guards, and one octet of + # measured progress per surviving iteration is what bounds the loop. + end = file.tell() + if end <= offset: + raise FieldValueError( + f'Field {self.name} has an option that consumed no data: ' + f'{code!r} at offset {offset - start} of {self._length}, with ' + f'{length} octet(s) of the option area left to parse' + ) + offset = end + self._option_padding = length return temp diff --git a/pcapkit/protocols/schema/internet/hopopt.py b/pcapkit/protocols/schema/internet/hopopt.py index 6532eccab7..1e0ba35dc4 100644 --- a/pcapkit/protocols/schema/internet/hopopt.py +++ b/pcapkit/protocols/schema/internet/hopopt.py @@ -155,12 +155,22 @@ def smf_dpd_data_selector(pkt: 'dict[str, Any]') -> 'Field': wrapped :class:`~pcapkit.protocols.schema.internet.hopopt.SMFHashBasedDPDOption` instance. + Note: + The field is sized ``Opt Data Len + 2`` rather than ``Opt Data Len``. + ``Opt Data Len`` counts only what follows the option header + [:rfc:`8200#section-4.2`], while both schemas this may return inherit + :attr:`Option.type` and :attr:`Option.len` and so parse those two octets + themselves. Sizing the field at ``Opt Data Len`` handed them an area two + octets short of the option they read, which + :class:`~pcapkit.corekit.fields.collections.OptionField` then mis-counted + against the option area -- c.f. #431. + """ mode = Enum_SMFDPDMode.get(pkt['test']['mode']) schema = SMFDPDOption.registry[mode] if schema is None: raise FieldValueError(f'HOPOPT: invalid SMF DPD mode: {mode}') - return SchemaField(length=pkt['test']['len'], schema=schema) + return SchemaField(length=pkt['test']['len'] + 2, schema=schema) def smf_i_dpd_tid_selector(pkt: 'dict[str, Any]') -> 'Field': @@ -367,11 +377,21 @@ def __init__(self, type: 'Enum_Option', len: 'int', domain: 'int', cmpt_len: 'in @schema_final class _SMFDPDOption(Schema): - """Header schema for HOPOPT SMF DPD options with generic representation.""" + """Header schema for HOPOPT SMF DPD options with generic representation. + + The ``test`` field forward-matches the first three octets of the option -- + ``Option Type``, ``Opt Data Len`` and the octet carrying the DPD mode bit -- + without consuming them, so that :func:`smf_dpd_data_selector` can size and + choose the schema which then reads the option properly. Its ``namespace`` + offsets are therefore bit offsets into the *option*, not into any one field + of it: ``Opt Data Len`` is octet 1, i.e. bits 8 to 15, and the mode bit is + the first bit of octet 2 [:rfc:`6621#section-8.1`]. + + """ #: SMF DPD mode. test: 'SMFDPDTestFlag' = ForwardMatchField(BitField(length=3, namespace={ - 'len': (1, 8), + 'len': (8, 8), 'mode': (16, 1), })) #: SMF DPD data. diff --git a/pcapkit/protocols/schema/internet/ipv6_opts.py b/pcapkit/protocols/schema/internet/ipv6_opts.py index 6d2daae4c9..01ac119990 100644 --- a/pcapkit/protocols/schema/internet/ipv6_opts.py +++ b/pcapkit/protocols/schema/internet/ipv6_opts.py @@ -155,12 +155,22 @@ def smf_dpd_data_selector(pkt: 'dict[str, Any]') -> 'Field': wrapped :class:`~pcapkit.protocols.schema.internet.ipv6_opts.SMFHashBasedDPDOption` instance. + Note: + The field is sized ``Opt Data Len + 2`` rather than ``Opt Data Len``. + ``Opt Data Len`` counts only what follows the option header + [:rfc:`8200#section-4.2`], while both schemas this may return inherit + :attr:`Option.type` and :attr:`Option.len` and so parse those two octets + themselves. Sizing the field at ``Opt Data Len`` handed them an area two + octets short of the option they read, which + :class:`~pcapkit.corekit.fields.collections.OptionField` then mis-counted + against the option area -- c.f. #431. + """ mode = Enum_SMFDPDMode.get(pkt['test']['mode']) schema = SMFDPDOption.registry[mode] if schema is None: raise FieldValueError(f'IPv6-Opts: invalid SMF DPD mode: {mode}') - return SchemaField(length=pkt['test']['len'], schema=schema) + return SchemaField(length=pkt['test']['len'] + 2, schema=schema) def smf_i_dpd_tid_selector(pkt: 'dict[str, Any]') -> 'Field': @@ -367,11 +377,21 @@ def __init__(self, type: 'Enum_Option', len: 'int', domain: 'int', cmpt_len: 'in @schema_final class _SMFDPDOption(Schema): - """Header schema for IPv6-Opts SMF DPD options with generic representation.""" + """Header schema for IPv6-Opts SMF DPD options with generic representation. + + The ``test`` field forward-matches the first three octets of the option -- + ``Option Type``, ``Opt Data Len`` and the octet carrying the DPD mode bit -- + without consuming them, so that :func:`smf_dpd_data_selector` can size and + choose the schema which then reads the option properly. Its ``namespace`` + offsets are therefore bit offsets into the *option*, not into any one field + of it: ``Opt Data Len`` is octet 1, i.e. bits 8 to 15, and the mode bit is + the first bit of octet 2 [:rfc:`6621#section-8.1`]. + + """ #: SMF DPD mode. test: 'SMFDPDTestFlag' = ForwardMatchField(BitField(length=3, namespace={ - 'len': (1, 8), + 'len': (8, 8), 'mode': (16, 1), })) #: SMF DPD data. diff --git a/tests/_support.py b/tests/_support.py index a0e6d5082d..a431bf9564 100644 --- a/tests/_support.py +++ b/tests/_support.py @@ -2,17 +2,97 @@ import abc import collections.abc +import contextlib import importlib.util import inspect +import math import pathlib +import signal import sys +import time import types -from typing import Iterable +import unittest +from typing import Iterable, Iterator from tests._tiers import (ROOT, SAMPLE_ROOT, REGENERATE_SAMPLES_CMD, GeneratedFixtureInUnitTierError, check_unit_tier_read) +@contextlib.contextmanager +def time_limit(seconds: int = 5) -> Iterator[None]: + """Fail the calling test if its body has not finished in ``seconds`` seconds. + + A parser defect that degenerates into a loop making no progress -- GitHub + issue #431 is one -- offers a test nothing to assert on: the call under test + simply never returns. A test written for it without a deadline does not fail, + it *wedges*, taking the rest of the run with it, so the deadline is as much a + part of the regression test as the assertion is. + + :func:`signal.alarm` is what interrupts the body, rather than a watchdog + thread: the loops this guards are pure Python and hold the GIL for the whole + of an iteration, so nothing in another thread gets to run and stop them, + whereas a signal is delivered between bytecodes. That also rules out + :data:`signal.SIGTERM` from an outer :program:`timeout`, which such a loop + likewise never gets around to handling. + + There is only ever one pending alarm per process, so arming this one cancels + whatever was already scheduled -- an enclosing ``time_limit``, or a deadline the + test runner set for itself. Both the handler and that pending alarm are put back + on the way out, the alarm with the seconds spent in the body deducted, so an + enclosing deadline keeps counting down across the ``with`` rather than being + silently dropped. + + An enclosing deadline whose moment falls inside the body is not delivered on + time, and the reason is this helper rather than the body: arming an alarm + *replaces* the pending one, so the enclosing deadline was already cancelled + before the body began and there was nothing left to fire when it came due. It + is re-armed for one second on the way out rather than dropped -- honouring it + late is the lesser wrong, and dropping it is how an enclosing timeout goes + missing altogether. That one-second floor covers every case where the body ran + for longer than the enclosing deadline had left. + + Args: + seconds: Whole seconds to allow the body. :func:`signal.alarm` counts in + whole seconds, so this cannot usefully be fractional. + + Yields: + Nothing. The deadline applies to the body of the ``with`` statement. + + Raises: + TimeoutError: If the body has not finished within ``seconds`` seconds. + + """ + # An interval timer is a POSIX facility, and the deadline is the whole point + # of this helper: silently running the body without one would restore exactly + # the wedged run it exists to prevent, so the test is skipped instead. + if not hasattr(signal, 'SIGALRM'): + raise unittest.SkipTest('signal.alarm is unavailable on this platform') + + def expire(signum: int, frame: object) -> None: + raise TimeoutError(f'did not finish within {seconds}s') + + previous_handler = signal.signal(signal.SIGALRM, expire) + + # NOTE: ``signal.alarm`` returns the seconds left on the alarm it replaces, or + # zero when there was none. That return value is the only record of an + # enclosing deadline, so it is read here rather than discarded -- there is no + # way to ask for it again afterwards. + pending = signal.alarm(seconds) + started = time.monotonic() + try: + yield + finally: + # Cancel first, so that an alarm which fires between here and the handler + # being restored cannot be delivered to whatever handler was installed + # before -- and so that the alarm re-armed below belongs to that handler + # rather than to ``expire``. + signal.alarm(0) + signal.signal(signal.SIGALRM, previous_handler) + if pending: + left = pending - (time.monotonic() - started) + signal.alarm(max(1, math.ceil(left))) + + def sample_path(name: str) -> str: """Resolve a sample capture file name to its absolute path. diff --git a/tests/protocols/internet/test_ipv4_unit.py b/tests/protocols/internet/test_ipv4_unit.py index 59ed459c91..4e1ac9713d 100644 --- a/tests/protocols/internet/test_ipv4_unit.py +++ b/tests/protocols/internet/test_ipv4_unit.py @@ -833,6 +833,34 @@ 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. + + 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 + ` 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. + + """ + from pcapkit.const.ipv4.option_number import OptionNumber + from pcapkit.protocols.internet.ipv4 import IPv4 + from tests._support import time_limit + + raw = bytes.fromhex('4a00001800010000400600000a0000010a000002') + with time_limit(5): + proto = IPv4(raw, len(raw)) + + self.assertEqual(proto.info.hdr_len, 40) + self.assertEqual( + [(code, opt.length) for code, opt in proto.info.options.items(multi=True)], + [(OptionNumber.EOOL, 1)], + ) + if __name__ == '__main__': unittest.main() diff --git a/tests/protocols/internet/test_ipv6_extension_unit.py b/tests/protocols/internet/test_ipv6_extension_unit.py index cc1e3cb23f..0695f438b8 100644 --- a/tests/protocols/internet/test_ipv6_extension_unit.py +++ b/tests/protocols/internet/test_ipv6_extension_unit.py @@ -8,7 +8,7 @@ import unittest from unittest import mock -from tests._support import purge_modules +from tests._support import purge_modules, time_limit RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) @@ -1453,6 +1453,177 @@ def test_ipv6_opts_padding_options_parse_from_the_wire(self) -> None: self._assert_padding_options_parse_from_the_wire(IPv6_Opts) + def _assert_smf_dpd_options_parse_from_the_wire(self, protocol_cls: type) -> None: + """A hash-based ``SMF_DPD`` option must consume exactly what it declares. + + Per :rfc:`6621#section-8.1` the option is an ``Option Type`` octet, an + ``Opt Data Len`` octet, and ``Opt Data Len`` further octets whose first + bit is the DPD mode -- set, here, for hash-based DPD, so that the whole of + the hash assist value is the ``Opt Data Len`` octets. Two things have to + hold for that to read correctly, and each case below breaks if either + does not: the declared length has to be read from the ``Opt Data Len`` + octet, and the area handed to the mode schema has to be two octets wider + than it, since that schema parses the option header itself. + + The first case is #431's reproducer verbatim. On the pristine tree it does + not fail, it never returns -- 15 seconds of CPU with no exception and no + diagnostic -- so every case runs under a deadline: a regression here has + to fail the suite rather than wedge it. + + """ + from pcapkit.const.ipv6.option import Option + from pcapkit.const.ipv6.smf_dpd_mode import SMFDPDMode + + cases = ( + ('#431: hash-based DPD of length 3, then a pad1', + b'\x3b\x00' + b'\x08\x03\x81\x02\x03' + b'\x00', + [(Option.SMF_DPD, 5), (Option.Pad1, 1)], + [b'\x81\x02\x03']), + ('hash-based DPD filling the option area', + b'\x3b\x00' + b'\x08\x04\x81\x02\x03\x04', + [(Option.SMF_DPD, 6)], + [b'\x81\x02\x03\x04']), + ('two hash-based DPD options back to back', + b'\x3b\x01' + b'\x08\x02\x81\x01' + b'\x08\x02\x81\x02' + b'\x00' * 6, + [(Option.SMF_DPD, 4), (Option.SMF_DPD, 4)] + [(Option.Pad1, 1)] * 6, + [b'\x81\x01', b'\x81\x02']), + ) + + for name, raw, expected, expected_hav in cases: + with self.subTest(case=name): + with time_limit(5): + proto = protocol_cls(raw, extension=True) + + options = list(proto.info.options.items(multi=True)) + + self.assertEqual(proto.info.length, len(raw)) + self.assertEqual([(code, opt.length) for code, opt in options], expected) + + # every octet of the option area has to be accounted for by an + # option, which is what the mis-sized read got wrong + self.assertEqual(sum(length for _, length in expected), len(raw) - 2) + + self.assertEqual( + [opt.hav for code, opt in options if code == Option.SMF_DPD], + expected_hav, + ) + self.assertEqual( + [opt.dpd_type for code, opt in options if code == Option.SMF_DPD], + [SMFDPDMode.H_DPD] * len(expected_hav), + ) + + # the reader and the writer must agree: repacking the parsed + # schema has to give back the very bytes it was read from + self.assertEqual(bytes(proto.__header__), raw) + + def test_hopopt_smf_dpd_options_parse_from_the_wire(self) -> None: + from pcapkit.protocols.internet.hopopt import HOPOPT + + self._assert_smf_dpd_options_parse_from_the_wire(HOPOPT) + + def test_ipv6_opts_smf_dpd_options_parse_from_the_wire(self) -> None: + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + + self._assert_smf_dpd_options_parse_from_the_wire(IPv6_Opts) + + def _assert_identification_based_dpd_options_parse_from_the_wire(self, protocol_cls: type) -> None: + """The other ``SMF_DPD`` mode, which sizes itself from a TaggerID. + + Identification-based DPD reaches the same mis-sized read by a different + route: its length comes from ``Opt Data Len`` too, but the octets it + covers are a TaggerID whose own width comes from the ``TidLen`` nibble + [:rfc:`6621#section-8.1`], so a wrong ``Opt Data Len`` mis-sizes the + identifier rather than the whole option. Both cases below hang the + pristine tree exactly as the hash-based ones do. + + Only the option itself is asserted, not the padding after it: HOPOPT and + IPv6-Opts disagree on how much of the option area an + ``SMFIdentificationBasedDPDOption`` leaves over, because the IPv6-Opts + schema carries an extra forward-matched octet which + :attr:`Schema.__buffer__ ` + records although the stream never consumes it. That is its own + ``len(data)``-is-not-consumed defect, distinct from the one #431 is about, + and pinning either count here would bless one of the two. + + """ + from pcapkit.const.ipv6.option import Option + from pcapkit.const.ipv6.smf_dpd_mode import SMFDPDMode + from pcapkit.const.ipv6.tagger_id import TaggerID + + cases = ( + ('null taggerID, two-octet identifier', + b'\x3b\x00' + b'\x08\x03\x00\x01\x02' + b'\x00', + 5, TaggerID.NULL, 0, None, b'\x01\x02'), + ('four-octet taggerID, one-octet identifier', + b'\x3b\x01' + b'\x08\x06\x13\x0a\x00\x00\x01\xff' + b'\x00' * 6, + 8, TaggerID.DEFAULT, 3, b'\x0a\x00\x00\x01', b'\xff'), + ) + + for name, raw, length, tid_type, tid_len, tid, identifier in cases: + with self.subTest(case=name): + with time_limit(5): + proto = protocol_cls(raw, extension=True) + + option = next(opt for code, opt in proto.info.options.items(multi=True) + if code == Option.SMF_DPD) + + self.assertEqual(option.length, length) + self.assertEqual(option.dpd_type, SMFDPDMode.I_DPD) + self.assertEqual(option.tid_type, tid_type) + self.assertEqual(option.tid_len, tid_len) + self.assertEqual(option.tid, tid) + self.assertEqual(option.id, identifier) + + self.assertEqual(bytes(proto.__header__), raw) + + def test_hopopt_identification_based_dpd_options_parse_from_the_wire(self) -> None: + from pcapkit.protocols.internet.hopopt import HOPOPT + + self._assert_identification_based_dpd_options_parse_from_the_wire(HOPOPT) + + def test_ipv6_opts_identification_based_dpd_options_parse_from_the_wire(self) -> None: + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + + self._assert_identification_based_dpd_options_parse_from_the_wire(IPv6_Opts) + + def _assert_a_truncated_option_area_is_diagnosed(self, protocol_cls: type) -> None: + """An option area with nothing behind it is an error, not a hang. + + The simplest form of #431, and the one that needs no option schema at all: + a header extension length of 1 declares a 14-octet option area, and the + two octets given here are the whole header. The option loop reads ``b''`` + for every field, which decodes the type octet as 0 -- ``Pad1`` in this + registry, not an end-of-option-list -- so it appends a phantom one-octet + ``Pad1``, subtracts the zero octets it actually read, and goes round + again forever. + + Note that this is registry-dependent, and IPv4, TCP and PCAP-NG behave + differently on purpose: 0 is the end-of-option-list code there, so the + loop's ``eool`` break has always absorbed a truncated area and reported + the rest as padding. Their tolerance is pinned in their own tests. + + """ + from pcapkit.utilities.exceptions import FieldValueError + + for name, raw in ( + ('no option octets at all', b'\x3b\x01'), + ('four of fourteen octets', b'\x3b\x01' + b'\x01\x02\x00\x00'), + ): + with self.subTest(case=name): + with self.assertRaisesRegex(FieldValueError, 'consumed no data'): + with time_limit(5): + protocol_cls(raw, len(raw), extension=True) + + def test_hopopt_truncated_option_area_is_diagnosed(self) -> None: + from pcapkit.protocols.internet.hopopt import HOPOPT + + self._assert_a_truncated_option_area_is_diagnosed(HOPOPT) + + def test_ipv6_opts_truncated_option_area_is_diagnosed(self) -> None: + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + + self._assert_a_truncated_option_area_is_diagnosed(IPv6_Opts) + def _assert_padding_option_schema_sizes_itself(self, protocol_cls: type) -> None: """The padding option schema, on its own. diff --git a/tests/protocols/schema/test_schema_unit.py b/tests/protocols/schema/test_schema_unit.py index 1e6dc64761..9162714cd5 100644 --- a/tests/protocols/schema/test_schema_unit.py +++ b/tests/protocols/schema/test_schema_unit.py @@ -3,10 +3,11 @@ import collections import enum import importlib.util +import io import unittest from unittest import mock -from tests._support import purge_modules +from tests._support import purge_modules, time_limit RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) @@ -405,6 +406,146 @@ class TinyOptionsSchema(Schema): self.assertEqual(unpacked.pad, b'') self.assertEqual(observed_padding, [1]) + def test_schema_option_field_unpack_rejects_an_option_consuming_nothing(self) -> None: + """An option area that cannot be advanced past is an error, not a hang. + + :meth:`OptionField.unpack + ` sizes each + option by ``len(data)``, the size of the schema the option reported, + which is not the number of octets it took from the stream. ``Wrapper`` + below is the smallest thing that separates the two, and is the shape + every wrapper option schema in the package has: it reads two octets and + returns a nested schema that recorded one. So the first option leaves the + stream one octet ahead of where ``length`` thinks it is, the second + option consumes the last of it, and the third finds the stream exhausted, + reads ``b''`` for every field, and reports ``len(data) == 0`` -- against + which ``length -= len(data)`` makes no progress at all. C.f. #431, where + this spun forever on an eight-octet HOPOPT header. + + The deadline is part of the test: without it a regression here does not + fail, it hangs the run. + + """ + from pcapkit.utilities.exceptions import FieldValueError + + schema = self._make_wrapped_options_schema() + + with self.assertRaisesRegex(FieldValueError, 'consumed no data'): + with time_limit(5): + schema.unpack(b'\x01\xff\x00', 3, {}) + + def test_schema_option_field_unpack_reports_a_field_relative_offset(self) -> None: + """The offset in the diagnostic counts from the option area, not the stream. + + :meth:`OptionField.unpack + ` is handed a + :obj:`bytes` buffer by + :meth:`Schema.unpack `, so + its stream starts at zero and the two readings coincide -- but the method + is public and takes an ``IO[bytes]`` as well, and a + :class:`~pcapkit.corekit.fields.misc.SchemaField` hands the live file + straight down. Reporting the raw stream position then names an offset + outside the field: the same three-octet area that fails at offset 3 below + reported offset 5 when the stream was two octets in. + + """ + from pcapkit.utilities.exceptions import FieldValueError + + schema = self._make_wrapped_options_schema() + field = schema.__fields__['options'] + + stream = io.BytesIO(b'\xde\xad' + b'\x01\xff\x00') + stream.seek(2) + + with self.assertRaisesRegex(FieldValueError, r'at offset 3 of 3\b'): + with time_limit(5): + field.unpack(stream, {}) + + def test_schema_list_field_unpack_rejects_a_schema_item_consuming_nothing(self) -> None: + """A list of schema items must be advanced past too, or reported. + + :meth:`ListField.unpack + ` sizes a schema item + by ``len(data)`` in the same way, so a budget larger than the octets behind + it leaves the item schema reading an exhausted stream, recording nothing, + and subtracting nothing. Reachable from a TCP segment whose ``SACK`` option + declares more octets than the option area holds, which + :file:`tests/protocols/transport/test_tcp_udp_unit.py` pins; this covers + the field on its own, and pins the offset as a count into the field rather + than into the stream -- the same reading as its ``OptionField`` subclass. + + Two items parse off the two octets below and the third finds the stream + exhausted, so the diagnostic's ``after 2 item(s)`` is a count of what was + parsed rather than an ordinal naming the second item. + + """ + from pcapkit.corekit.fields.collections import ListField + from pcapkit.corekit.fields.misc import SchemaField + from pcapkit.corekit.fields.numbers import UInt8Field + from pcapkit.protocols.schema.schema import Schema, schema_final + from pcapkit.utilities.exceptions import FieldValueError + + @schema_final + class Marker(Schema): + type: int = UInt8Field(default=0) + + @schema_final + class MarkerListSchema(Schema): + #: Eight octets of budget, however few are really there. + markers: list[Marker] = ListField( + length=8, + item_type=SchemaField(length=2, schema=Marker), + ) + + field = MarkerListSchema.__fields__['markers'] + + stream = io.BytesIO(b'\xde\xad' + b'\x01\x02') + stream.seek(2) + + with self.assertRaisesRegex(FieldValueError, r'after 2 item\(s\), at offset 2 of 8\b'): + with time_limit(5): + field.unpack(stream, {}) + + def _make_wrapped_options_schema(self): + """A three-octet option area whose first option over-reads by one octet. + + Returns: + A :class:`~pcapkit.protocols.schema.schema.Schema` subclass carrying a + single :class:`~pcapkit.corekit.fields.collections.OptionField`. + + ``Wrapper`` is the smallest thing that separates the octets an option takes + from the stream from the ``len(data)`` it reports, and is the shape every + wrapper option schema in the package has: it reads two octets and returns a + nested schema that recorded one. + + """ + from pcapkit.corekit.fields.collections import OptionField + from pcapkit.corekit.fields.misc import SchemaField + from pcapkit.corekit.fields.numbers import UInt8Field + from pcapkit.protocols.schema.schema import Schema, schema_final + + @schema_final + class Marker(Schema): + type: int = UInt8Field(default=0) + + @schema_final + class Wrapper(Schema): + #: Two octets of stream, of which ``Marker`` records only the first. + body: Marker = SchemaField(length=2, schema=Marker) + + def post_process(self, packet: dict) -> Schema: + return self.body + + @schema_final + class WrappedOptionsSchema(Schema): + options: list[Marker] = OptionField( + length=3, + base_schema=Marker, + registry=collections.defaultdict(lambda: Marker, {1: Wrapper}), + ) + + return WrappedOptionsSchema + 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 4a4f28f251..2ec3e109f1 100644 --- a/tests/protocols/transport/test_tcp_udp_unit.py +++ b/tests/protocols/transport/test_tcp_udp_unit.py @@ -1175,6 +1175,88 @@ def test_construction_accepts_bare_integer_ports(self) -> None: self.assertEqual(unnamed.info.srcport.port, 53406) self.assertIs(unnamed.info.dstport, member) + def test_a_sack_option_overrunning_the_option_area_raises(self) -> None: + """An over-long ``SACK`` option must be an error, not a hang. + + ``SACK``'s blocks are a + :class:`~pcapkit.corekit.fields.collections.ListField` sized ``Length - 2`` + from the wire, with a + :class:`~pcapkit.corekit.fields.misc.SchemaField` item type -- and that + loop subtracts each item's *parsed* size from its budget. A ``Length`` + larger than the option area leaves it reading ``SACKBlock`` off an + exhausted stream, where every field reads ``b''``, the item parses to + nothing, and the budget never falls. This is the same defect as #431, in + ``OptionField``'s base class rather than in ``OptionField`` itself, and it + is reachable from a single TCP segment. + + The segment below is 24 octets: a data offset of 6, so a four-octet option + area, holding a ``SACK`` option that declares 22. The well-formed segment + is checked alongside it, because a guard that rejected real ``SACK`` + options would pass this test on its own. + + """ + import struct + + from pcapkit.const.tcp.option import Option + from pcapkit.protocols.transport.tcp import TCP + from pcapkit.utilities.exceptions import FieldValueError + from tests._support import time_limit + + def segment(data_offset: 'int', options: 'bytes') -> 'bytes': + return struct.pack('!HHIIBBHHH', 1, 2, 0, 0, data_offset << 4, + 0x10, 0, 0, 0) + options + + # one 8-octet block, declared as 10 octets, then two no-operations + good = segment(8, bytes([Option.SACK, 10]) + struct.pack('!II', 1, 2) + b'\x01\x01') + with time_limit(5): + proto = TCP(good, len(good)) + + self.assertEqual(proto.info.hdr_len, len(good)) + self.assertEqual( + [(code, opt.length) for code, opt in proto.info.options.items(multi=True)], + [(Option.SACK, 10), (Option.No_Operation, 1), (Option.No_Operation, 1)], + ) + sack = next(opt for code, opt in proto.info.options.items(multi=True) + if code == Option.SACK) + self.assertEqual([(block.left, block.right) for block in sack.sack], [(1, 2)]) + self.assertEqual(bytes(proto.__header__), good) + + bad = segment(6, bytes([Option.SACK, 22]) + b'\x36\xcc') + with self.assertRaisesRegex(FieldValueError, 'consumed no data'): + with time_limit(5): + 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. + + 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 + ` checks each + option's progress *after* its end-of-option-list break rather than before: + checking first turns every such segment into an error. + + """ + from pcapkit.const.tcp.option import Option + from pcapkit.protocols.transport.tcp import TCP + from tests._support import time_limit + + raw = bytes.fromhex('00501f900000000100000002a002ffff00000000020405b4') + with time_limit(5): + proto = TCP(raw, len(raw)) + + self.assertEqual(proto.info.hdr_len, 40) + self.assertEqual( + [(code, opt.length) for code, opt in proto.info.options.items(multi=True)], + [(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) + def test_unregistered_option_kind_does_not_mutate_the_class_registry(self) -> None: """Parsing must not write to the shared ``TCP.__option__``. diff --git a/tests/test_support_helpers.py b/tests/test_support_helpers.py index 4b45c0dc16..d159e5baae 100644 --- a/tests/test_support_helpers.py +++ b/tests/test_support_helpers.py @@ -1,5 +1,11 @@ # -*- coding: utf-8 -*- -"""Tests for :func:`tests._support.close_extractor`. +"""Tests for the helpers in :mod:`tests._support`. + +Two of them are pinned here, both for the same reason: they are *machinery* the +rest of the suite leans on, so a fault in either reports itself as a failure in +whichever test happened to be running rather than as a fault in the helper. +:func:`~tests._support.time_limit` is covered by :class:`TimeLimitTests` at the +end; the rest of the module is :func:`~tests._support.close_extractor`. That helper is teardown machinery: nearly every runtime and integration test hands it an extractor from ``addCleanup`` or a ``finally`` block. Teardown code @@ -24,9 +30,11 @@ """ from __future__ import annotations +import signal +import time import unittest -from tests._support import close_extractor +from tests._support import close_extractor, time_limit class Closeable: @@ -169,5 +177,80 @@ def test_base_exception_is_not_swallowed(self) -> None: close_extractor(Extractor(Closeable(KeyboardInterrupt()), Closeable())) +@unittest.skipUnless(hasattr(signal, 'SIGALRM'), 'signal.alarm is unavailable') +class TimeLimitTests(unittest.TestCase): + """A deadline that arrives, and an enclosing one that survives. + + A process has one pending alarm, so arming a deadline cancels whatever was + already scheduled. The helper reads what it displaced and puts it back; these + pin that, because an enclosing deadline going missing is invisible until the + run it should have bounded hangs instead. + + """ + + def setUp(self) -> None: + # Whatever a test leaves behind, the next one starts from nothing pending + # and from a handler this class owns rather than the helper's. + self.handled = [] # type: list[int] + previous = signal.signal(signal.SIGALRM, lambda signum, frame: self.handled.append(signum)) + self.addCleanup(signal.signal, signal.SIGALRM, previous) + self.addCleanup(signal.alarm, 0) + + def test_the_deadline_fires_on_a_body_that_overruns(self) -> None: + """The point of the helper, pinned so the rest cannot be met by disarming.""" + with self.assertRaises(TimeoutError): + with time_limit(1): + while True: + pass + + def test_an_enclosing_alarm_is_restored(self) -> None: + """An outer deadline keeps counting down across the ``with``. + + Before this was fixed the helper cancelled the pending alarm on the way out + and never re-armed it, so an outer ``signal.alarm(30)`` read back as ``0`` + afterwards: the enclosing deadline was gone, silently. + + """ + own_handler = signal.getsignal(signal.SIGALRM) + + signal.alarm(30) + with time_limit(5): + pass + + # Reading the remaining seconds cancels the alarm, which is the cleanup + # this test wanted anyway. + remaining = signal.alarm(0) + + self.assertGreater(remaining, 0) + self.assertLessEqual(remaining, 30) + self.assertIs(signal.getsignal(signal.SIGALRM), own_handler) + self.assertEqual(self.handled, []) + + def test_an_enclosing_alarm_that_expired_in_the_body_is_re_armed(self) -> None: + """An outer deadline overtaken by the body is honoured late, not dropped. + + The body holds the process past the moment the outer alarm was due, so it + cannot be delivered on time. The helper re-arms it for a second rather than + cancelling it, since cancelling is how an outer timeout goes missing + altogether. + + """ + signal.alarm(1) + with time_limit(5): + time.sleep(1.2) + + remaining = signal.alarm(0) + + self.assertGreaterEqual(remaining, 1) + self.assertEqual(self.handled, []) + + def test_nothing_is_re_armed_when_nothing_was_pending(self) -> None: + """The common case: no enclosing deadline, nothing left behind.""" + with time_limit(5): + pass + + self.assertEqual(signal.alarm(0), 0) + + if __name__ == '__main__': unittest.main()