diff --git a/docs/source/contributing/conventions/index.rst b/docs/source/contributing/conventions/index.rst index 8febc6895d..b8a006fc69 100644 --- a/docs/source/contributing/conventions/index.rst +++ b/docs/source/contributing/conventions/index.rst @@ -30,8 +30,8 @@ House Conventions * :ref:`test-coverage` -- what the test suite has to prove; * :ref:`utilities-or-corekit` -- which of :mod:`pcapkit.utilities` and :mod:`pcapkit.corekit` a support module belongs in, where a context - variable lives, and where a schema helper used outside the schema layer or - outside its own protocol lives; + variable lives, where a schema helper used outside the schema layer or + outside its own protocol lives, and what reads a capture cut; * :ref:`reassembly-integration` -- :mod:`pcapkit.foundation.reassembly` and what a new engine there needs from :mod:`pcapkit.toolkit` and :mod:`pcapkit.foundation.analysis`; diff --git a/docs/source/contributing/conventions/utilities-or-corekit.rst b/docs/source/contributing/conventions/utilities-or-corekit.rst index 485fa232d2..341ed2c144 100644 --- a/docs/source/contributing/conventions/utilities-or-corekit.rst +++ b/docs/source/contributing/conventions/utilities-or-corekit.rst @@ -72,6 +72,18 @@ header schema framework, which stays :mod:`pcapkit.protocols.schema` and its :class:`~pcapkit.protocols.schema.schema.Schema`; it holds only the helpers shared across the layer. +**A capture cut belongs to the built-in short read of** +:class:`~pcapkit.protocols.schema.schema.Schema`. :meth:`Schema.unpack +` records the field the data ends +inside, through ``_note_short_read``, as ``__short_read__``; +:meth:`~pcapkit.protocols.schema.schema.Schema._cut_short_read` packs back only the +octets read; and :func:`~pcapkit.corekit.schema.shortread.keep_short_read` and +:func:`~pcapkit.corekit.schema.shortread.replay_short_read` carry the record through +``info`` and cut a rebuild back by it. A protocol reads and rebuilds a cut header +through them, never through an ``unpack`` override of its own or a length callable +clamped to the octets present, which hides the cut from the record. Ruled in review +of :pr:`2682` (:issue:`2680`), and applied to option areas on :issue:`2694`. + **A helper in one protocol's own schema module moves there too once code outside that protocol uses it and it is generic enough to reuse**, so that another implementor can find it; a helper only its own protocol calls stays beside the schema. Ruled on diff --git a/docs/source/pcapkit/protocols/internet/mh.rst b/docs/source/pcapkit/protocols/internet/mh.rst index f14da8075e..5bef68ea02 100644 --- a/docs/source/pcapkit/protocols/internet/mh.rst +++ b/docs/source/pcapkit/protocols/internet/mh.rst @@ -988,6 +988,7 @@ Auxiliary Functions .. autofunction:: pcapkit.protocols.schema.internet.mh.bid_address_selector .. autofunction:: pcapkit.protocols.schema.internet.mh.br_code_selector .. autofunction:: pcapkit.protocols.schema.internet.mh.cga_public_key_length +.. autofunction:: pcapkit.protocols.schema.internet.mh.item_list_length .. autofunction:: pcapkit.protocols.schema.internet.mh.dmnp_prefix_selector .. autofunction:: pcapkit.protocols.schema.internet.mh.fb_code_selector .. autofunction:: pcapkit.protocols.schema.internet.mh.lma_address_selector diff --git a/pcapkit/corekit/fields/collections.py b/pcapkit/corekit/fields/collections.py index 1a8ddb4ba1..0d0a787a63 100644 --- a/pcapkit/corekit/fields/collections.py +++ b/pcapkit/corekit/fields/collections.py @@ -445,12 +445,23 @@ def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> 'list ``packet`` as ``__option_padding__`` so that the next fields can be aware of it. + A ``__capture_cut__`` key in ``packet``, false, says that the data + ends at the length the enclosing layer declares for it, so that an + area the data ends inside runs past that length rather than being + cut short by the capture, as + :class:`~pcapkit.protocols.internet.ipv6.IPv6` tells the extension + headers it walks (:issue:`2694`). Absent, the data may end at a cut. + Raises: FieldValueError: If an option consumes nothing from ``buffer``, since the loop below has then no way to get past it. ProtocolError: If an option that begins with octets present reads - past the end of the data, unless :meth:`_accepts_short_read` - accepts it. + past the end of the data, unless the data ends inside the area, + the option's fields fit it and :meth:`_accepts_cut` keeps it, or + :meth:`_accepts_short_read` accepts it. Or if the area runs past + the end of the data where ``packet`` gives ``__capture_cut__`` + as false, the data ending at the length the enclosing layer + declares rather than at a cut. .. versionchanged:: 1.5.0 @@ -458,6 +469,16 @@ def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> 'list the data raises :exc:`~pcapkit.utilities.exceptions.ProtocolError`, where it was parsed from zero-filled octets (:issue:`1325`). + .. versionchanged:: 1.6.0 + + Where the data ends inside the area, as a capture cut short leaves + it, the list ends with the data, and an option it ends inside is + kept if its fields fit the area, so :meth:`Schema.unpack + ` records the cut as + a short read rather than either error being raised + (:issue:`2694`). See :meth:`_accepts_cut`, and ``__capture_cut__`` + above. + """ length = self._length if isinstance(buffer, bytes): @@ -529,6 +550,24 @@ def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> 'list # nothing at all and is left to the end-of-option-list and progress # handling below, which is what an option area declared longer than its # octets relies on. + # + # Unless the data ends inside the area: fewer octets are left than it + # declares, as a capture cut short leaves them. An option the cut lands + # inside, or in front of, is not one running past its area, and is read + # as the short read :meth:`Schema.unpack` records (#2694). An enclosing + # layer that knows the data ends at the length it declares, not at a cut, + # says so with ``__capture_cut__`` false in ``packet``, and the area then + # runs past that length. + if isinstance(buffer, bytes): + present = len(buffer) + else: + try: + present = file.seek(0, io.SEEK_END) - start + file.seek(start, io.SEEK_SET) + except (OSError, ValueError): # a stream that cannot seek to its end counts as whole + present = self._length + short = present < self._length + cut = short and packet.get('__capture_cut__', True) tracker = file = _ShortReadTracker(file) # type: ignore[assignment] temp = [] # type: list[_TS] @@ -547,6 +586,20 @@ def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> 'list byte = file.read(field.length) code = field.unpack(byte, packet.copy()) consumed = len(byte) + + # NOTE: The data ends where this option would begin, inside the area: the + # cut leaves no option to read, so the list ends with the data, and + # :meth:`Schema.unpack` records the cut from the rest of the area left + # as ``option_padding`` (#2694). Where no cut ends it, the area runs + # past the data. A registry whose end-of-option-list code is 0 reads its + # exhausted type as that code, and stops at it below. + if not consumed and short and code != self._eool: + if cut: + break + raise ProtocolError( + f'Field {self.name} has an option area that runs past the end of ' + f'the data: {self._length} octet(s) declared, with {present} present' + ) schema = self._registry[code] # rewind to the beginning of the option @@ -554,7 +607,14 @@ def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> 'list # unpack option using option schema data = schema.unpack(file, length, packet) # type: ignore[call-arg,misc,var-annotated] - if tracker.short and consumed > 0 and not self._accepts_short_read(data): + + # NOTE: An option the cut lands inside stands when the fields it + # declares fit the area, as ``__length__`` left at zero or more shows, + # and it read the octets there (#2694). One whose fields reach past + # the area runs past both (#1325). + stands = (cut and packet['__length__'] >= 0 and file.tell() > offset + and self._accepts_cut(data)) + if tracker.short and consumed > 0 and not stands and not self._accepts_short_read(data): raise ProtocolError( f'Field {self.name} has an option that runs past the end of ' f'the data: {code!r} at offset {offset - start} of ' @@ -604,7 +664,8 @@ def _accepts_short_read(self, option: '_TS') -> 'bool': # pylint: disable=unuse """Whether an option that read past the end of the data may stand. Args: - option: Option just unpacked, which read short. + option: Option just unpacked, which read short and ran past its area + or past the data at the end of the area. Returns: :data:`False`, so that :meth:`unpack` refuses the option @@ -617,3 +678,20 @@ def _accepts_short_read(self, option: '_TS') -> 'bool': # pylint: disable=unuse """ return False + + def _accepts_cut(self, option: '_TS') -> 'bool': # pylint: disable=unused-argument + """Whether an option the data ends inside, inside its area, may stand. + + Args: + option: Option just unpacked, which read short where the data ends + short of the area declared, and whose fields fit the area. + + Returns: + :data:`True`, so that :meth:`unpack` keeps the option, as the short + read :meth:`Schema.unpack ` + records (:issue:`2694`). A list that keeps such an area as the octets + captured instead may override this, as + :class:`~pcapkit.protocols.schema.misc.pcapng.OptionAreaField` does. + + """ + return True diff --git a/pcapkit/protocols/internet/hopopt.py b/pcapkit/protocols/internet/hopopt.py index 9af668fbeb..cc7c059e07 100644 --- a/pcapkit/protocols/internet/hopopt.py +++ b/pcapkit/protocols/internet/hopopt.py @@ -92,7 +92,7 @@ TunnelEncapsulationLimitOption as Schema_TunnelEncapsulationLimitOption from pcapkit.protocols.schema.internet.hopopt import UnassignedOption as Schema_UnassignedOption from pcapkit.protocols.schema.schema import Schema -from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall +from pcapkit.utilities.exceptions import BaseError, CaughtByCaller, ProtocolError, UnsupportedCall from pcapkit.utilities.warnings import ProtocolWarning, RegistryWarning, warn if TYPE_CHECKING: @@ -395,7 +395,8 @@ def read(self, length: 'Optional[int]' = None, *, extension: 'bool' = False, **k def make(self, next: 'Enum_TransType | int' = Enum_TransType.UDP, options: 'Optional[list[Schema_Option | tuple[Enum_Option | int, dict[str, Any]] | bytes] | Option]' = None, # pylint: disable=line-too-long - payload: 'bytes | ProtocolBase | Schema' = b'', + payload: 'bytes | ProtocolBase | Schema' = b'', *, + len: 'Optional[int]' = None, # pylint: disable=redefined-builtin **kwargs: 'Any') -> 'Schema_HOPOPT': """Make (construct) packet data. @@ -403,6 +404,12 @@ def make(self, next: Next header type. options: Hop-by-Hop Options. payload: Payload of current protocol. + len: ``Hdr Ext Len`` to declare, rather than the one the options + fill. A header the capture cut short declares one its parsed + options do not fill, which :meth:`from_data` passes on, so that + the rebuild declares it too (:issue:`2694`). Options that fill + less than it declares are not padded out to it, so the header + then declares more octets than it packs, as a cut one does. **kwargs: Arbitrary keyword arguments. Returns: @@ -412,7 +419,12 @@ def make(self, ProtocolError: If ``next``, or the type of an option, is a name, a :obj:`bool`, or a member or code that does not fit in one octet; or if the options make a header too long for its - one-octet ``Hdr Ext Len``. + one-octet ``Hdr Ext Len``, or longer than ``len`` declares. + + .. versionchanged:: 1.6.0 + + Takes ``len``, the ``Hdr Ext Len`` of a header the capture cut short + (:issue:`2694`). """ next_value = self._make_index(next, Enum_TransType, bits=8, name='next header') @@ -428,6 +440,11 @@ def make(self, # NOTE: ``_make_hopopt_options`` has aligned the header, so this division # is exact and rounding up is unnecessary. length = self._make_uint((total_length - 6) // 8, bits=8, name='Hdr Ext Len') + if len is not None: + length = self._make_uint(len, bits=8, name='Hdr Ext Len') + if length * 8 + 6 < total_length: + raise ProtocolError(f'{self.alias}: Hdr Ext Len {length} declares fewer octets ' + f'than the options take ({total_length})') return Schema_HOPOPT( next=next_value, @@ -538,10 +555,13 @@ def _make_data(cls, data: 'Data_HOPOPT') -> 'dict[str, Any]': # type: ignore[ov Key-value pairs for protocol construction. """ + # NOTE: A header the capture cut short declares an area its parsed options + # do not fill, and is rebuilt with the Hdr Ext Len it declared (#2694). return { 'next': data.next, 'options': data.options, 'payload': cls._make_payload(data), + 'len': data.length // 8 - 1 if data.get('__short_read__') is not None else None, } def _read_opt_type(self, kind: 'int') -> 'tuple[int, bool]': @@ -609,7 +629,16 @@ def _read_hopopt_options(self, length: 'int') -> 'Option': Raises: ProtocolError: If the parsed options do not total exactly ``length`` - octets. + octets, unless the capture cut the header short, as its short + read records. Or if the capture cut an option short and its + parser refuses the zeros it lacks, or its maker does not write + the octets captured back. + + .. versionchanged:: 1.6.0 + + A header the capture cut short reads as the short read it is, and an + option it cuts as its parser reads the zeros it lacks, rather than + being refused (:issue:`2694`). """ counter = 0 # length of read options @@ -623,16 +652,34 @@ def _read_hopopt_options(self, length: 'int') -> 'Option': meth_name = f'_read_opt_{name}' meth = cast('OptionParser', getattr(self, meth_name, self._read_opt_none)) + maker = cast('OptionConstructor', + getattr(self, f'_make_opt_{name}', self._make_opt_none)) + else: + meth, maker = name + + # NOTE: The capture cut this option short, as its short read records, so + # the octets it lacks read as zeros. It stands if its parser takes those + # zeros and its maker writes the octets captured back (#2694); otherwise + # it is refused, as an option running past the data is (#1325). + if '__short_read__' in schema.__dict__: + try: + with CaughtByCaller(BaseError): + data = meth(schema, options=options) + made = maker(dscp, data).pack()[:len(schema)] + except BaseError as error: + raise ProtocolError(f'{self.alias}: [OptNo {dscp}] option cut short: {error}') from error + if made != bytes(schema): + raise ProtocolError(f'{self.alias}: [OptNo {dscp}] option cut short does not ' + 'rebuild as captured') else: - meth = name[0] - data = meth(schema, options=options) + data = meth(schema, options=options) # record option data options.add(dscp, data) counter += len(schema) - # check threshold - if counter != length: + # check threshold, which a header the capture cut short falls short of (#2694) + if counter != length and '__short_read__' not in self.__header__.__dict__: raise ProtocolError('HOPOPT: invalid format') return options diff --git a/pcapkit/protocols/internet/ipv6.py b/pcapkit/protocols/internet/ipv6.py index b409e21742..b3f1a418c2 100644 --- a/pcapkit/protocols/internet/ipv6.py +++ b/pcapkit/protocols/internet/ipv6.py @@ -582,13 +582,18 @@ def _decode_next_layer(self, ipv6: 'Data_IPv6', proto: 'Optional[int]' = None, # proto = None # break + # NOTE: The Payload Length left bounds this header. With every octet of + # it present, a header longer than that runs past it, malformed and + # not cut by the capture, which its option area is told (#2694). + context = dict(packet or {}, __capture_cut__=len(payload) < raw_len) + # make protocol name # NOTE: Counted after every header walked, as it would be were each # dissected by the one before it. ESP dissects its decrypted # payload even when walked, and those layers count on from here. token = frame_depth.set(depth + len(_protos)) try: - next_ = self._import_next_layer(proto, packet=packet, version=6, extension=True, + next_ = self._import_next_layer(proto, packet=context, version=6, extension=True, payload=payload) # type: ignore[misc,call-arg] finally: frame_depth.reset(token) diff --git a/pcapkit/protocols/internet/ipv6_opts.py b/pcapkit/protocols/internet/ipv6_opts.py index f3cfa7d0ec..1af53b9398 100644 --- a/pcapkit/protocols/internet/ipv6_opts.py +++ b/pcapkit/protocols/internet/ipv6_opts.py @@ -95,7 +95,7 @@ TunnelEncapsulationLimitOption as Schema_TunnelEncapsulationLimitOption from pcapkit.protocols.schema.internet.ipv6_opts import UnassignedOption as Schema_UnassignedOption from pcapkit.protocols.schema.schema import Schema -from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall +from pcapkit.utilities.exceptions import BaseError, CaughtByCaller, ProtocolError, UnsupportedCall from pcapkit.utilities.warnings import ProtocolWarning, RegistryWarning, warn if TYPE_CHECKING: @@ -387,7 +387,8 @@ def read(self, length: 'Optional[int]' = None, *, extension: 'bool' = False, # def make(self, next: 'Enum_TransType | int' = Enum_TransType.UDP, options: 'Optional[list[Schema_Option | tuple[Enum_Option | int, dict[str, Any]] | bytes] | Option]' = None, # pylint: disable=line-too-long - payload: 'bytes | ProtocolBase | Schema' = b'', + payload: 'bytes | ProtocolBase | Schema' = b'', *, + len: 'Optional[int]' = None, # pylint: disable=redefined-builtin **kwargs: 'Any') -> 'Schema_IPv6_Opts': """Make (construct) packet data. @@ -395,6 +396,12 @@ def make(self, next: Next header type. options: Destination Options. payload: Payload of current protocol. + len: ``Hdr Ext Len`` to declare, rather than the one the options + fill. A header the capture cut short declares one its parsed + options do not fill, which :meth:`from_data` passes on, so that + the rebuild declares it too (:issue:`2694`). Options that fill + less than it declares are not padded out to it, so the header + then declares more octets than it packs, as a cut one does. **kwargs: Arbitrary keyword arguments. Returns: @@ -404,7 +411,12 @@ def make(self, ProtocolError: If ``next``, or the type of an option, is a name, a :obj:`bool`, or a member or code that does not fit in one octet; or if the options make a header too long for its - one-octet ``Hdr Ext Len``. + one-octet ``Hdr Ext Len``, or longer than ``len`` declares. + + .. versionchanged:: 1.6.0 + + Takes ``len``, the ``Hdr Ext Len`` of a header the capture cut short + (:issue:`2694`). """ next_value = self._make_index(next, Enum_TransType, bits=8, name='next header') @@ -420,6 +432,11 @@ def make(self, # NOTE: ``_make_ipv6_opts`` has aligned the header, so this division is # exact and rounding up is unnecessary. length = self._make_uint((total_length - 6) // 8, bits=8, name='Hdr Ext Len') + if len is not None: + length = self._make_uint(len, bits=8, name='Hdr Ext Len') + if length * 8 + 6 < total_length: + raise ProtocolError(f'{self.alias}: Hdr Ext Len {length} declares fewer octets ' + f'than the options take ({total_length})') return Schema_IPv6_Opts( next=next_value, @@ -530,10 +547,13 @@ def _make_data(cls, data: 'Data_IPv6_Opts') -> 'dict[str, Any]': # type: ignore Key-value pairs for protocol construction. """ + # NOTE: A header the capture cut short declares an area its parsed options + # do not fill, and is rebuilt with the Hdr Ext Len it declared (#2694). return { 'next': data.next, 'options': data.options, 'payload': cls._make_payload(data), + 'len': data.length // 8 - 1 if data.get('__short_read__') is not None else None, } def _read_opt_type(self, kind: 'int') -> 'tuple[int, bool]': @@ -597,7 +617,16 @@ def _read_ipv6_opts(self, length: 'int') -> 'Option': Raises: ProtocolError: If the parsed options do not total exactly ``length`` - octets. + octets, unless the capture cut the header short, as its short + read records. Or if the capture cut an option short and its + parser refuses the zeros it lacks, or its maker does not write + the octets captured back. + + .. versionchanged:: 1.6.0 + + A header the capture cut short reads as the short read it is, and an + option it cuts as its parser reads the zeros it lacks, rather than + being refused (:issue:`2694`). """ counter = 0 # length of read options @@ -611,16 +640,34 @@ def _read_ipv6_opts(self, length: 'int') -> 'Option': meth_name = f'_read_opt_{name}' meth = cast('OptionParser', getattr(self, meth_name, self._read_opt_none)) + maker = cast('OptionConstructor', + getattr(self, f'_make_opt_{name}', self._make_opt_none)) + else: + meth, maker = name + + # NOTE: The capture cut this option short, as its short read records, so + # the octets it lacks read as zeros. It stands if its parser takes those + # zeros and its maker writes the octets captured back (#2694); otherwise + # it is refused, as an option running past the data is (#1325). + if '__short_read__' in schema.__dict__: + try: + with CaughtByCaller(BaseError): + data = meth(schema, options=options) + made = maker(dscp, data).pack()[:len(schema)] + except BaseError as error: + raise ProtocolError(f'{self.alias}: [OptNo {dscp}] option cut short: {error}') from error + if made != bytes(schema): + raise ProtocolError(f'{self.alias}: [OptNo {dscp}] option cut short does not ' + 'rebuild as captured') else: - meth = name[0] - data = meth(schema, options=options) + data = meth(schema, options=options) # record option data options.add(dscp, data) counter += len(schema) - # check threshold - if counter != length: + # check threshold, which a header the capture cut short falls short of (#2694) + if counter != length and '__short_read__' not in self.__header__.__dict__: raise ProtocolError('IPv6-Opts: invalid format') return options diff --git a/pcapkit/protocols/internet/mh.py b/pcapkit/protocols/internet/mh.py index 87c1fa4e10..b7d77ea0c3 100644 --- a/pcapkit/protocols/internet/mh.py +++ b/pcapkit/protocols/internet/mh.py @@ -499,7 +499,8 @@ from pcapkit.protocols.schema.internet.mh import VendorSpecificOption as Schema_VendorSpecificOption from pcapkit.protocols.schema.schema import Schema from pcapkit.utilities.datetime import fromtimestamp, totimestamp -from pcapkit.utilities.exceptions import EnumValueError, ProtocolError, UnsupportedCall +from pcapkit.utilities.exceptions import (BaseError, CaughtByCaller, EnumValueError, ProtocolError, + UnsupportedCall) from pcapkit.utilities.warnings import ProtocolWarning, RegistryWarning, warn if TYPE_CHECKING: @@ -3055,12 +3056,17 @@ def _read_mh_options(self, options_schema: 'list[Schema_Option]') -> 'Option': ProtocolError: If an option read by a ``_read_opt_*`` method has fields that do not take exactly the octets its ``Option Length`` declares [:rfc:`6275#section-6.2.1`]. A parser registered as a callable - reports its own length, and is not checked. + reports its own length, and is not checked. Or if the capture cut + an option short and its parser refuses the zeros it lacks, or its + maker does not write the octets captured back. .. versionchanged:: 1.6.0 - Such an option is refused, rather than read from octets outside it and - rebuilt with the length its fields give (:issue:`2684`). + * Such an option is refused, rather than read from octets outside it + and rebuilt with the length its fields give (:issue:`2684`). + * An option the capture cut short is read, its missing octets as + zeros, if its maker writes the octets captured back + (:issue:`2694`). """ options = OrderedMultiDict() # type: Option @@ -3073,14 +3079,34 @@ def _read_mh_options(self, options_schema: 'list[Schema_Option]') -> 'Option': meth_name = f'_read_opt_{name}' meth = cast('OptionParser', getattr(self, meth_name, self._read_opt_none)) + maker = cast('OptionConstructor', + getattr(self, f'_make_opt_{name}', self._make_opt_none)) else: - meth = name[0] - data = meth(schema, options=options) + meth, maker = name + + # NOTE: The capture cut this option short, as its short read records, so + # the octets it lacks read as zeros and it is short of its length. It + # stands if its parser takes those zeros and its maker writes the octets + # captured back (#2694); otherwise it is refused, as an option running + # past the data is (#1325). + if '__short_read__' in schema.__dict__: + try: + with CaughtByCaller(BaseError): + data = meth(schema, options=options) + made = maker(type, data).pack()[:len(schema)] + except BaseError as error: + raise ProtocolError(f'{self.alias}: [OptNo {type}] option cut short: {error}') from error + if made != bytes(schema): + raise ProtocolError(f'{self.alias}: [OptNo {type}] option cut short does not ' + 'rebuild as captured') + else: + data = meth(schema, options=options) - # NOTE: Fields are read at their own widths, so a length they do not fill - # exactly reads octets not the option's, and rebuilds as another (#2684). - if isinstance(name, str) and len(schema) != data.length: - raise ProtocolError(f'{self.alias}: [OptNo {type}] invalid option length: {schema.length}') + # NOTE: Fields are read at their own widths, so a length they do not + # fill exactly reads octets not the option's, and rebuilds as another + # (#2684). + if isinstance(name, str) and len(schema) != data.length: + raise ProtocolError(f'{self.alias}: [OptNo {type}] invalid option length: {schema.length}') # record option data options.add(type, data) diff --git a/pcapkit/protocols/misc/checksum/ip_pseudo.py b/pcapkit/protocols/misc/checksum/ip_pseudo.py index 2e2b31a5f3..a34904457b 100644 --- a/pcapkit/protocols/misc/checksum/ip_pseudo.py +++ b/pcapkit/protocols/misc/checksum/ip_pseudo.py @@ -425,13 +425,17 @@ def _walk(ip: 'IPv6') -> 'tuple[Enum_TransType, int]': The code the walk stopped on, and the octets of the headers ahead of it. The code is an extension header's if ``ip`` did not parse that header: a malformed one (:issue:`2477`), one past the header limit, - or one not captured. + or one not captured, or cut short by the capture (:issue:`2694`). """ code, before = ip.info.next, 0 for _, header in ip.extension_headers.items(multi=True): if code in _TERMINAL_HEADERS or code == Enum_TransType.ESP: break + # NOTE: A header the capture cut short is its schema's short read + # (#2694), and as unwalked as one not captured at all. + if header.info.get('__short_read__') is not None: + break before += header.length code = header.info.next return code, before diff --git a/pcapkit/protocols/schema/internet/mh.py b/pcapkit/protocols/schema/internet/mh.py index ad4fac4bb3..bd6533783d 100644 --- a/pcapkit/protocols/schema/internet/mh.py +++ b/pcapkit/protocols/schema/internet/mh.py @@ -132,7 +132,7 @@ if TYPE_CHECKING: from datetime import datetime as dt_type from ipaddress import IPv4Address, IPv6Address - from typing import Any, Optional + from typing import Any, Callable, Optional from pcapkit.corekit.fields.field import FieldBase as Field from pcapkit.protocols.protocol import ProtocolBase @@ -648,6 +648,37 @@ def cga_public_key_length(pkt: 'dict[str, Any]') -> 'int': return length +def item_list_length(width: 'int', label: 'str', offset: 'int' = 0) -> 'Callable[[dict[str, Any]], int]': + """Return a callback sizing a list of fixed-width items from a ``length`` field. + + Args: + width: Octets of each item. + label: Name of the structure the list is in, for the error message, + as ``'OptNo'`` for a mobility option. + offset: Octets the ``length`` counts ahead of the list. + + Returns: + A callback returning the octets of the list. + + The callback raises :exc:`~pcapkit.utilities.exceptions.ProtocolError` for a + length that leaves part of an item, which + :class:`~pcapkit.corekit.fields.collections.ListField` would otherwise refuse + with a :exc:`~pcapkit.utilities.exceptions.FieldValueError`, logged even + where the refusal is handled (:issue:`2694`). A length short of ``offset`` + is left to the parser of the structure, which refuses it. + + .. versionadded:: 1.6.0 + + """ + def callback(pkt: 'dict[str, Any]') -> 'int': + length = pkt['length'] - offset + if length > 0 and length % width: + raise ProtocolError(f'MH: [{label} {pkt["type"]}] invalid length: {pkt["length"]}, ' + f'not a whole number of {width}-octet items') + return length + return callback + + @schema_final class MH(Schema): """Header schema for MH packets. @@ -959,7 +990,7 @@ class MultiPrefixExtension(CGAExtension, code=Enum_CGAExtension.Multi_Prefix): }) #: Prefixes. prefixes: 'list[int]' = ListField( - length=lambda pkt: pkt['length'] - 4, + length=item_list_length(8, 'CGA Ext', offset=4), item_type=UInt64Field(), ) @@ -1618,7 +1649,7 @@ class FlowSummaryOption(Option, code=Enum_Option.Flow_Summary_Mobility_Option): #: Flow identifiers being refreshed. fid: 'list[int]' = ListField( - length=lambda pkt: pkt['length'], + length=item_list_length(2, 'OptNo'), item_type=UInt16Field(), ) @@ -1692,7 +1723,7 @@ class BIDReferenceSuboption(FlowIdentificationSuboption, #: Referenced binding identifiers. bid: 'list[int]' = ListField( - length=lambda pkt: pkt['length'], + length=item_list_length(2, 'FID SubOpt'), item_type=UInt16Field(), ) diff --git a/pcapkit/protocols/schema/misc/pcapng.py b/pcapkit/protocols/schema/misc/pcapng.py index d3cf50e94c..2c33b7b7f5 100644 --- a/pcapkit/protocols/schema/misc/pcapng.py +++ b/pcapkit/protocols/schema/misc/pcapng.py @@ -459,6 +459,20 @@ def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> 'list self._option_padding = 0 return [raw] + def _accepts_cut(self, option: 'Schema') -> 'bool': # pylint: disable=unused-argument + """Whether an option the data ends inside, inside its area, may stand. + + Args: + option: Option just unpacked, which read short. + + Returns: + :data:`False`: a block keeps an area the data ends inside as the + octets captured, as it keeps one an option runs past + (:issue:`1325`), rather than read its missing octets as zeros. + + """ + return False + def bounded_area(length: 'Callable[[dict[str, Any]], int]') -> 'Callable[[dict[str, Any]], int]': """Clamp a packet block's option area to the octets the block itself holds. diff --git a/tests/protocols/internet/test_ipv6_ext_unit.py b/tests/protocols/internet/test_ipv6_ext_unit.py index 1dd15abd3e..f30bb650dc 100644 --- a/tests/protocols/internet/test_ipv6_ext_unit.py +++ b/tests/protocols/internet/test_ipv6_ext_unit.py @@ -250,12 +250,12 @@ def test_failed_dedicated_parse_runs_once(self) -> None: """GitHub issue #2477: the ``Raw`` the walk ends on is the upper layer, not parsed again, so the header's parser runs once and no ``SchemaWarning`` is left from the header it replaced. The refusal is - handled, so it logs nothing (#2588). + handled, so it logs nothing (#2588). A header the capture cut short is not + refused at all, since it parses as the schema's short read (#2694). """ from unittest import mock from pcapkit.const.ipv6.extension_header import ExtensionHeader as EH - from pcapkit.const.ipv6.option import Option from pcapkit.const.reg.transtype import TransType from pcapkit.protocols.internet.hopopt import HOPOPT from pcapkit.protocols.internet.ipv6 import IPv6 @@ -263,12 +263,9 @@ def test_failed_dedicated_parse_runs_once(self) -> None: from pcapkit.utilities.logging import logger udp = int(TransType.UDP) - truncated = (f'Field options has an option that runs past the end of the data: ' - f'{Option.PadN!r} at offset 0 of 6, with only 1 octet(s) of it present') for form, hopopt, error in ( ('malformed', bytes([udp, 0, 0x05, 10]) + bytes(4) + _udp_bytes(b'ok'), 'HOPOPT: [OptNo 5] invalid format'), - ('truncated', bytes([udp, 0, 1]), truncated), ): raw = _ipv6_bytes(int(EH.HOPOPT), hopopt) with self.subTest(form=form): diff --git a/tests/protocols/internet/test_ipv6_extension_unit.py b/tests/protocols/internet/test_ipv6_extension_unit.py index 8ae06a77be..0e30ed84b2 100644 --- a/tests/protocols/internet/test_ipv6_extension_unit.py +++ b/tests/protocols/internet/test_ipv6_extension_unit.py @@ -2202,15 +2202,15 @@ def test_ipv6_opts_mpl_option_rejects_underflowing_length(self) -> None: self._assert_mpl_option_rejects_underflowing_length(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. + def _assert_a_truncated_option_area_ends_with_the_data(self, protocol_cls: type) -> None: + """An option area with nothing behind it ends with the data, and is no 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 + registry, not an end-of-option-list -- so it appended a phantom one-octet + ``Pad1``, subtracted the zero octets it actually read, and went round again forever. Note that this is registry-dependent, and IPv4, TCP and PCAP-NG behave @@ -2218,27 +2218,29 @@ def _assert_a_truncated_option_area_is_diagnosed(self, protocol_cls: type) -> No 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 + The data ends inside the area, so the loop ends with the data, and the + header reads as the short read its schema records (#2694). + """ 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(): - protocol_cls(raw, len(raw), extension=True) + with time_limit(): + header = protocol_cls(raw, len(raw), extension=True) + self.assertEqual(header.info.__short_read__, ('options', len(raw) - 2)) + self.assertEqual(protocol_cls.from_data(header.info).data, raw) - def test_hopopt_truncated_option_area_is_diagnosed(self) -> None: + def test_hopopt_truncated_option_area_ends_with_the_data(self) -> None: from pcapkit.protocols.internet.hopopt import HOPOPT - self._assert_a_truncated_option_area_is_diagnosed(HOPOPT) + self._assert_a_truncated_option_area_ends_with_the_data(HOPOPT) - def test_ipv6_opts_truncated_option_area_is_diagnosed(self) -> None: + def test_ipv6_opts_truncated_option_area_ends_with_the_data(self) -> None: from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts - self._assert_a_truncated_option_area_is_diagnosed(IPv6_Opts) + self._assert_a_truncated_option_area_ends_with_the_data(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/internet/test_ipv6_exthdr_cut_rebuild_unit.py b/tests/protocols/internet/test_ipv6_exthdr_cut_rebuild_unit.py index 5b20a31c93..4927d2391a 100644 --- a/tests/protocols/internet/test_ipv6_exthdr_cut_rebuild_unit.py +++ b/tests/protocols/internet/test_ipv6_exthdr_cut_rebuild_unit.py @@ -8,8 +8,9 @@ Options, TCP cut at 57 octets rebuilt 65. GitHub issue #2477: a header whose dedicated parser raises is kept as -:class:`~pcapkit.protocols.misc.raw.Raw`, and the walk ends there, so a cut one -rebuilds from that :class:`~pcapkit.protocols.misc.raw.Raw`. +:class:`~pcapkit.protocols.misc.raw.Raw`, and the walk ends there. GitHub issue +#2694: a capture cut is no refusal but the schema's short read, so a cut header +is an extension header of its own and rebuilds from its own info. Every case builds its own octets and reads no capture. Classes are imported inside each test, after :func:`~tests._support.reimport_once_per_class`. @@ -101,18 +102,23 @@ def test_raw_upper_layer_starts_past_the_extension_headers(self) -> None: self.assertEqual(ip6.payload.data, data[48:]) self.assertEqual(len(IPv6.from_data(ip6.info).data), 57) - def test_cut_header_is_raw_and_ends_the_walk(self) -> None: + def test_cut_header_reads_short(self) -> None: + # #2477 kept this cut header as Raw, since its parser refused it; a capture + # cut belongs to the schema's short read instead (#2694), so the header is + # parsed, and records where the data ended + from pcapkit.const.ipv6.extension_header import ExtensionHeader + from pcapkit.protocols.internet.hopopt import HOPOPT from pcapkit.protocols.internet.ipv6 import IPv6 - from pcapkit.protocols.misc.raw import Raw for cut in (48, 50, 52, 55): data = packet('hop-nonxt')[:cut] with self.subTest(cut=cut): ip6 = IPv6(data, len(data)) - self.assertEqual(list(ip6.extension_headers.items(multi=True)), []) - self.assertIs(type(ip6.payload), Raw) - self.assertEqual(ip6.payload.data, data[40:]) - self.assertEqual(ip6.info.to_dict()['__exthdr__'], ()) + hop = ip6.extension_headers.get(ExtensionHeader.HOPOPT) + self.assertIs(type(hop), HOPOPT) + self.assertEqual(hop.data, data[40:]) + self.assertEqual(hop.info.__short_read__, ('options', cut - 42)) + self.assertEqual([item[0] for item in ip6.info.to_dict()['__exthdr__']], [HOPOPT]) self.assertEqual(IPv6.from_data(ip6.info.to_dict()).data, data) diff --git a/tests/protocols/internet/test_option_area_cut_2694_unit.py b/tests/protocols/internet/test_option_area_cut_2694_unit.py new file mode 100644 index 0000000000..0527c09f4f --- /dev/null +++ b/tests/protocols/internet/test_option_area_cut_2694_unit.py @@ -0,0 +1,450 @@ +# -*- coding: utf-8 -*- +"""A capture cut inside an option area reads as a short read, quietly (#2694). + +Under the #1325 guards, ``OptionField.unpack`` refused a capture cut that lands +before or inside an option area: an option the data ends inside "runs past the end +of the data", and one the data ends in front of "consumed no data", a +``FieldValueError`` that logged at CRITICAL, and again from ``beholder`` inside +IPv6. So an MH message cut inside its options raised on its own, as did an LRI or +UPA message cut inside its reserved octets, and a HOPOPT header cut to 2 or 3 +octets logged inside IPv6. + +Where the data ends inside the area, the list ends with the data, and an option +it ends inside stands if its fields fit the area, so ``Schema.unpack`` records the +cut as the short read #2682 extended to option areas. A HOPOPT or IPv6-Opts header +cut short reads as that short read too, rather than failing its check that its +options fill the area, and rebuilds with the ``Hdr Ext Len`` it declared. An option +whose maker would not write the octets captured back is refused, and so is an +option whose fields run past its area (#1325). + +A header IPv6 hands every octet its Payload Length leaves, and which is longer +than that, runs past it: malformed, not cut by the capture, and kept as ``Raw`` +as #2477 has it. IPv6 tells its option area so, and the area is refused. + +The issue's second comment is malformed data rather than a cut: an MH Flow Summary +option of an odd length raised a ``FieldValueError`` from the list field reading its +16-bit identifiers, logged even inside IPv6. A list of fixed-width items that a +length leaves part of one is refused with a ``ProtocolError``, which a handled +refusal keeps quiet (#2588). + +Every case builds its own octets and reads no capture. Classes are imported inside +each test, after :func:`~tests._support.reimport_once_per_class`. + +""" + +import contextlib +import struct +import unittest +import warnings +from typing import TYPE_CHECKING + +from tests._support import reimport_once_per_class, time_limit +from tests.utilities._harness import capture + +if TYPE_CHECKING: + from typing import Any, Callable, Iterator, Optional + +#: Test tier (#1958): a sweep that cuts every MH message type at every offset. +__tier__ = 'LATEST' + +#: Header fields shared by every MH built here; Next Header is IPv6-NoNxt. +MH_BASE = {'next': 59, 'chksum': b'\x00\x00', 'payload': b''} + +#: The octets of a Mobility Header IPv6 keeps as its own layer: its +#: ``__length_hint__``, under which it is kept as raw octets (#1170). +MH_HINT = 8 + +#: A Hop-by-Hop or Destination Options header of 16 octets: a Pad1, then a PadN of +#: four octets and one of five, so a cut lands in front of, and inside, each. +OPTIONS = bytes([59, 1]) + bytes([0]) + bytes([1, 4]) + bytes(4) + bytes([1, 5]) + bytes(5) + +#: The octets of a HOPOPT or IPv6-Opts header IPv6 keeps as its own layer: its +#: ``__length_hint__``, under which it is kept as raw octets (#1170). +OPTIONS_HINT = 8 + +#: An options header carrying a Line Identification option (RFC 6788) of the line +#: ID ``line-1``, and a PadN. +LINE_ID = bytes.fromhex('1101' '8c07066c696e652d31' '0103000000') + +#: A Binding Refresh Request carrying a Binding Refresh Advice and a PadN +#: (RFC 6275, sections 6.1.1 and 6.2.4). +BRR_BRA = bytes.fromhex('3b01000000000000' '02020007' '01020000') + +#: A Binding Refresh Request carrying a Flow Identification option (RFC 6089, +#: section 4.2), whose one sub-option is a Flow Binding Action (RFC 7109), and a PadN. +FID = bytes.fromhex('3b02000000000000' '2d0a000100010000' '0402000b' '01020000') + +#: The SchemaWarning a layer kept with its header cut short gives, which +#: ``Protocol._parse_next_layer`` documents and #2526 pins: the header's fields run +#: past the length it was handed. It is not a refusal. +CUT_HEADER_WARNING = 'packet length < 0' + + +def ipv6(code: int, body: bytes, length: 'Optional[int]' = None) -> bytes: + """An IPv6 packet carrying ``body`` as next header ``code``, its Payload Length + ``length``, or that of ``body`` if not given.""" + declared = len(body) if length is None else length + return struct.pack('!IHBB', 6 << 28, declared, code, 64) + bytes(32) + body + + +def brr(option: bytes) -> bytes: + """A Binding Refresh Request carrying ``option``, padded to a multiple of 8 octets.""" + body = bytes(2) + option + pad = -(len(body) + 6) % 8 + if pad == 1: + body += bytes(1) + elif pad: + body += bytes([1, pad - 2]) + bytes(pad - 2) + return bytes([59, (len(body) + 6) // 8 - 1, 0, 0, 0, 0]) + body + + +@contextlib.contextmanager +def reported() -> 'Iterator[list[str]]': + """What the block reports, on the ``pcapkit`` logger and on :mod:`warnings`.""" + from pcapkit.utilities.logging import logger + + messages = [] # type: list[str] + with capture(logger) as logs, warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + yield messages + messages.extend(f'{level}: {text}' for level, text in logs.messages) + messages.extend(f'warned: {item.message}' for item in caught) + + +def refusals(messages: 'list[str]') -> 'list[str]': + """``messages`` but the warnings a header cut short gives.""" + return [text for text in messages if CUT_HEADER_WARNING not in text] + + +class OptionAreaCutTests(unittest.TestCase): + """Pin a cut inside or in front of an option area as a quiet short read.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def messages(self) -> 'dict[str, bytes]': + """Every registered MH message type, built with its defaults.""" + from pcapkit.protocols.internet.mh import MH + + return {getattr(code, 'name', str(code)): bytes(MH(type=code, data={}, **MH_BASE)) + for code in MH.__message__} + + def assertRebuildsQuietly(self, cls: 'Any', layer: 'Any', raw: bytes, # pylint: disable=invalid-name + reparsed: bool = False) -> None: + """``cls`` rebuilt from ``layer``'s info and its dict gives ``raw``, reporting nothing. + + A ``reparsed`` rebuild parses a header kept cut short again, as an IPv6 + packet does its extension headers, which then warns as its parse did. + + """ + for path, data in (('info', layer.info), ('dict', layer.info.to_dict())): + with self.subTest(path=path): + with reported() as messages: + rebuilt = cls.from_data(data) + if reparsed: + messages = refusals(messages) + self.assertEqual((rebuilt.data.hex(), messages), (raw.hex(), [])) + + def parse(self, build: 'Callable[[], Any]') -> 'tuple[Any, list[str]]': + """What ``build`` returns, and what it reported.""" + with time_limit(), reported() as messages: + layer = build() + return layer, messages + + def test_every_message_cut_at_every_offset_reads_short(self) -> None: + from pcapkit.protocols.internet.mh import MH + + for name, whole in self.messages().items(): + for cut in range(1, len(whole)): + raw = whole[:cut] + with self.subTest(message=name, cut=cut): + mh, messages = self.parse(lambda: MH(raw)) # pylint: disable=cell-var-from-loop + self.assertEqual((mh.data, refusals(messages)), (raw, [])) + self.assertIsNotNone(mh.info.get('__short_read__')) + self.assertRebuildsQuietly(MH, mh, raw) + + def test_every_message_cut_inside_ipv6_rebuilds_quietly(self) -> None: + from pcapkit.const.ipv6.extension_header import ExtensionHeader + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.misc.raw import Raw + + for name, whole in self.messages().items(): + for cut in range(1, len(whole)): + raw = ipv6(135, whole)[:40 + cut] + with self.subTest(message=name, cut=cut): + ip6, messages = self.parse(lambda: IPv6(raw)) # pylint: disable=cell-var-from-loop + self.assertEqual((ip6.data, refusals(messages)), (raw, [])) + if cut < MH_HINT: + self.assertIs(type(ip6.payload), Raw) + else: + mh = ip6.extension_headers.get(ExtensionHeader.Mobility_Header) + self.assertIsNotNone(mh) + self.assertIsNotNone(mh.info.get('__short_read__')) + self.assertRebuildsQuietly(IPv6, ip6, raw, reparsed=True) + + def test_reserved_octets_cut_reads_short(self) -> None: + # the LRI's two reserved octets follow its sequence number, and the UPA's + # three its sequence number and status (RFC 6705, section 10.1; RFC 7077, + # section 4.2), behind the six octets of the Mobility Header + from pcapkit.const.mh.packet import Packet + from pcapkit.protocols.internet.mh import MH + + for code, reserved in ((Packet.Localized_Routing_Initiation, range(8, 10)), + (Packet.Update_Notification_Acknowledgement, range(9, 12))): + whole = bytes(MH(type=code, data={}, **MH_BASE)) + for cut in reserved: + raw = whole[:cut] + with self.subTest(message=code.name, cut=cut): + mh, messages = self.parse(lambda: MH(raw)) # pylint: disable=cell-var-from-loop + self.assertEqual(refusals(messages), []) + self.assertEqual(mh.__header__.data.__dict__['__short_read__'], + ('reserved', cut - reserved.start)) + self.assertEqual(mh.info.msg_reserved, raw[reserved.start:]) + self.assertRebuildsQuietly(MH, mh, raw) + + def test_options_header_schema_cut_reads_short(self) -> None: + from pcapkit.protocols.schema.internet.hopopt import HOPOPT + from pcapkit.protocols.schema.internet.ipv6_opts import IPv6_Opts + + for schema in (HOPOPT, IPv6_Opts): + for cut in range(1, len(OPTIONS)): + raw = OPTIONS[:cut] + with self.subTest(schema=schema.__name__, cut=cut): + with time_limit(), warnings.catch_warnings(): + warnings.simplefilter('ignore') + header = schema.unpack(raw, len(raw)) + self.assertEqual(header.__dict__['__short_read__'], + ('len', 0) if cut == 1 else ('options', cut - 2)) + self.assertEqual((bytes(header), header.pack()), (raw, raw)) + + def test_options_header_cut_at_every_offset_reads_short(self) -> None: + from pcapkit.protocols.internet.hopopt import HOPOPT + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + + for cls in (HOPOPT, IPv6_Opts): + for cut in range(1, len(OPTIONS)): + raw = OPTIONS[:cut] + with self.subTest(header=cls.__name__, cut=cut): + header, messages = self.parse(lambda: cls(raw)) # pylint: disable=cell-var-from-loop + self.assertEqual((header.data, refusals(messages)), (raw, [])) + self.assertIsNotNone(header.info.get('__short_read__')) + self.assertRebuildsQuietly(cls, header, raw) + + def test_options_header_cut_inside_ipv6_reads_short(self) -> None: + from pcapkit.const.ipv6.extension_header import ExtensionHeader + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.misc.raw import Raw + + for code, key in ((0, ExtensionHeader.HOPOPT), (60, ExtensionHeader.IPv6_Opts)): + for cut in range(1, len(OPTIONS)): + raw = ipv6(code, OPTIONS)[:40 + cut] + with self.subTest(next_header=code, cut=cut): + ip6, messages = self.parse(lambda: IPv6(raw)) # pylint: disable=cell-var-from-loop + self.assertEqual((ip6.data, refusals(messages)), (raw, [])) + if cut < OPTIONS_HINT: + self.assertIs(type(ip6.payload), Raw) + else: + header = ip6.extension_headers.get(key) + self.assertIsNotNone(header) + self.assertIsNotNone(header.info.get('__short_read__')) + self.assertRebuildsQuietly(IPv6, ip6, raw, reparsed=True) + + def test_options_header_make_takes_the_length_a_cut_declared(self) -> None: + # a whole header is made with the Hdr Ext Len its options fill, and one the + # capture cut short with the one it declared + from pcapkit.protocols.internet.hopopt import HOPOPT + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + from pcapkit.utilities.exceptions import ProtocolError + + # pylint: disable=protected-access + for cls in (HOPOPT, IPv6_Opts): + with self.subTest(header=cls.__name__): + self.assertIsNone(cls._make_data(cls(OPTIONS).info)['len']) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + self.assertEqual(cls._make_data(cls(OPTIONS[:5]).info)['len'], 1) + self.assertEqual(bytes(cls(next=59, options=[], len=0)), bytes([59, 0, 1, 4]) + bytes(4)) + with reported(): + for value in (-1, 256, True): + with self.assertRaises(ProtocolError): + cls(next=59, options=[], len=value) + + def test_options_header_cut_option_that_does_not_rebuild_is_refused_quietly(self) -> None: + # a Router Alert cut after its type reads a length of zero, which its parser + # refuses; a Line Identification cut before its line ID length is made with + # a length that leaves the line ID out + from pcapkit.protocols.internet.hopopt import HOPOPT + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + from pcapkit.protocols.misc.raw import Raw + from pcapkit.utilities.exceptions import ProtocolError + + for cls, code in ((HOPOPT, 0), (IPv6_Opts, 60)): + for raw, error in ( + (bytes([59, 1, 5]), r'option cut short: .*\[OptNo 5\] invalid format'), + (LINE_ID[:4], 'option cut short does not rebuild as captured'), + ): + with self.subTest(header=cls.__name__, raw=raw.hex()): + with reported(), self.assertRaisesRegex(ProtocolError, error): + cls(raw) + + packet = ipv6(code, raw) + ip6, messages = self.parse(lambda: IPv6(packet)) # pylint: disable=cell-var-from-loop + self.assertEqual(messages, []) + self.assertIs(type(ip6.payload), Raw) + self.assertRebuildsQuietly(IPv6, ip6, packet) + + def test_option_past_its_area_is_still_refused(self) -> None: + # #1325: a PadN declaring three octets past a whole header, and one + # declaring twenty-two in an area of sixteen the data also ends inside + from pcapkit.protocols.internet.mh import MH + from pcapkit.utilities.exceptions import ProtocolError + + for raw in (bytes.fromhex('3b010000000000000104000000000103'), + bytes.fromhex('3b020000000000000114000000000000')): + with self.subTest(raw=raw.hex()): + with reported(), self.assertRaisesRegex(ProtocolError, 'runs past the end of the data'): + MH(raw) + + def test_cut_option_that_does_not_rebuild_is_refused_quietly(self) -> None: + # a Binding Refresh Advice cut after its type reads a length of zero, which + # its parser refuses; an IPv6 Address/Prefix cut before its code reads a + # code of zero, which its closed enumeration refuses; a Flow Identification + # cut where its sub-options start is made with a length that leaves them out + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.internet.mh import MH + from pcapkit.protocols.misc.raw import Raw + from pcapkit.utilities.exceptions import ProtocolError + + for raw, error in ( + (BRR_BRA[:9], r'option cut short: MH: \[Opt 2\] invalid format'), + (bytes.fromhex('3b03000000000000' '2212'), + 'option cut short: 0 is not a valid IPv6AddressPrefixCode'), + (FID[:16], 'option cut short does not rebuild as captured'), + ): + with self.subTest(raw=raw.hex()): + with reported(), self.assertRaisesRegex(ProtocolError, error): + MH(raw) + + packet = ipv6(135, raw) + ip6, messages = self.parse(lambda: IPv6(packet)) # pylint: disable=cell-var-from-loop + self.assertEqual(messages, []) + self.assertIs(type(ip6.payload), Raw) + self.assertRebuildsQuietly(IPv6, ip6, packet) + + def test_cut_option_that_rebuilds_stands(self) -> None: + # the Binding Refresh Advice cut inside its interval, and the PadN after it + # cut inside its padding + from pcapkit.const.mh.option import Option + from pcapkit.protocols.internet.mh import MH + + for cut in (10, 11, 13, 14, 15): + raw = BRR_BRA[:cut] + with self.subTest(cut=cut): + mh, messages = self.parse(lambda: MH(raw)) # pylint: disable=cell-var-from-loop + self.assertEqual(refusals(messages), []) + self.assertEqual(list(mh.info.options), + [Option.Binding_Refresh_Advice] + [Option.PadN] * (cut > 12)) + self.assertEqual(mh.__header__.data.__dict__['__short_read__'], ('options', cut - 8)) + self.assertRebuildsQuietly(MH, mh, raw) + + + def test_list_short_of_an_item_is_refused_quietly(self) -> None: + # a Flow Summary option, or a Flow Identification option's BID Reference + # sub-option, of an odd length (RFC 6089, sections 4.2.2 and 4.2.1), and a + # Multi-Prefix CGA extension a length leaves part of a prefix (RFC 5535); + # a Load Information option of an odd length (RFC 6463, section 4.3) was + # refused quietly, and is the control + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.internet.mh import MH + from pcapkit.protocols.misc.raw import Raw + from pcapkit.utilities.exceptions import ProtocolError + + cases = [(brr(bytes([44, n]) + bytes(range(1, n + 1))), 'OptNo 44') for n in (1, 3, 5)] + cases += [(brr(bytes([45, 11]) + bytes([0, 1, 0, 1, 0, 0]) + bytes([2, 3, 0, 1, 2])), 'FID SubOpt 2')] + cases += [(bytes.fromhex('3b060000000000000c2d086fca5e10b200c99c8ce00164277c0820010db8' + '00000000003002aabb0012000b0000000020010db80000000100'), 'CGA Ext 18')] + cases += [(brr(bytes([48, n]) + bytes(range(1, n + 1))), None) for n in (17, 19)] + for raw, where in cases: + with self.subTest(raw=raw.hex()): + with reported(), self.assertRaises(ProtocolError) as refusal: + MH(raw) + if where is not None: + self.assertRegex(str(refusal.exception), rf'\[{where}\] invalid length: \d+, ' + r'not a whole number of \d+-octet items') + + packet = ipv6(135, raw) + ip6, messages = self.parse(lambda: IPv6(packet)) # pylint: disable=cell-var-from-loop + self.assertEqual(refusals(messages), []) + self.assertIs(type(ip6.payload), Raw) + self.assertRebuildsQuietly(IPv6, ip6, packet, reparsed=True) + + def test_list_of_whole_items_reads(self) -> None: + # the same option and extension, of a length the items fill + from pcapkit.protocols.internet.mh import MH + + for raw in (brr(bytes([44, 4, 0, 1, 0, 2])), + bytes.fromhex('3b060000000000000c2d086fca5e10b200c99c8ce00164277c0820010db8' + '00000000003002aabb0012000c0000000020010db80000000100')): + with self.subTest(raw=raw.hex()): + mh, messages = self.parse(lambda: MH(raw)) # pylint: disable=cell-var-from-loop + self.assertEqual((mh.data, messages), (raw, [])) + self.assertRebuildsQuietly(MH, mh, raw) + + def test_header_past_the_payload_length_is_raw_quietly(self) -> None: + # each header is handed every octet its Payload Length declares, short of + # the header's own length: on the Payload Length's side of the line, where + # a cut is handed fewer octets than it declares + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.misc.raw import Raw + + headers = ((0, OPTIONS), (60, OPTIONS), (135, BRR_BRA), + (135, bytes(self.messages()['Localized_Routing_Initiation']))) + for code, whole in headers: + for length in range(1, len(whole)): + raw = ipv6(code, whole[:length], length) + with self.subTest(next_header=code, header=whole.hex(), payload_length=length): + ip6, messages = self.parse(lambda: IPv6(raw)) # pylint: disable=cell-var-from-loop + self.assertEqual((ip6.data, messages), (raw, [])) + self.assertEqual(list(ip6.extension_headers.items(multi=True)), []) + self.assertIs(type(ip6.payload), Raw) + self.assertRebuildsQuietly(IPv6, ip6, raw) + + def test_payload_length_tells_a_cut_from_an_overrun(self) -> None: + # the review's case: a 16-octet header under a Payload Length of 8 with + # those 8 octets, then the same header cut to 10 octets of the 16 declared + from pcapkit.const.ipv6.extension_header import ExtensionHeader + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.misc.raw import Raw + + for code, key, whole in ((0, ExtensionHeader.HOPOPT, OPTIONS), + (60, ExtensionHeader.IPv6_Opts, OPTIONS), + (135, ExtensionHeader.Mobility_Header, BRR_BRA)): + with self.subTest(next_header=code): + ip6, _ = self.parse(lambda: IPv6(ipv6(code, whole[:8], 8))) # pylint: disable=cell-var-from-loop + self.assertIs(type(ip6.payload), Raw) + self.assertIsNone(ip6.extension_headers.get(key)) + + ip6, _ = self.parse(lambda: IPv6(ipv6(code, whole[:10], len(whole)))) # pylint: disable=cell-var-from-loop + header = ip6.extension_headers.get(key) + self.assertIsNotNone(header) + self.assertIsNotNone(header.info.get('__short_read__')) + + def test_option_area_told_its_data_is_bounded_runs_past_it(self) -> None: + # the field on its own: ``__capture_cut__`` false where the data ends + # inside the area, at an option boundary and inside an option + from pcapkit.protocols.schema.internet.hopopt import HOPOPT + from pcapkit.utilities.exceptions import ProtocolError + + field = HOPOPT.__fields__['options'] + for raw, error in ((OPTIONS[2:3], 'option area that runs past the end of the data'), + (OPTIONS[2:5], 'option that runs past the end of the data')): + with self.subTest(raw=raw.hex()): + packet = {'len': 1, '__capture_cut__': False} + with reported(), self.assertRaisesRegex(ProtocolError, error): + field(packet).unpack(raw, packet) + packet = {'len': 1} + self.assertEqual(sum(map(len, field(packet).unpack(raw, packet))), len(raw)) + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/test_maker_octet_widths_2676_unit.py b/tests/protocols/test_maker_octet_widths_2676_unit.py index 90c82fb141..093b0efffd 100644 --- a/tests/protocols/test_maker_octet_widths_2676_unit.py +++ b/tests/protocols/test_maker_octet_widths_2676_unit.py @@ -539,12 +539,6 @@ class CapturedOctetTests(_Case): #: past them records a short read: the options area after them records none. NOTHING_AFTER = {'_make_msg_fna', '_make_msg_has', '_make_msg_sq'} - #: MH messages whose cut inside the reserved octets does not parse, and why. - UNPARSED = { - '_make_msg_lri': 'a cut message reads its options area as zeros, which raises a FieldValueError', - '_make_msg_upa': 'a cut message reads its options area as zeros, which raises a FieldValueError', - } - def _mh_messages(self) -> 'list[tuple[str, object, int]]': from pcapkit.protocols.internet.mh import MH @@ -606,30 +600,18 @@ def assertEditRefused(self, cls: 'type', data: 'bytes', mark: 'bytes', edit: 'ob return True def test_mh_message_cut_inside_reserved_octets(self) -> None: + # the LRI and UPA cuts parse as the short read their options area records + # where the data ends inside it (#2694) from pcapkit.protocols.internet.mh import MH - from pcapkit.utilities.exceptions import FieldValueError swept = [] for name, code, size in self._mh_messages(): swept.append(name) mark = bytes(range(0xA1, 0xA1 + size)) data = bytes(MH(next=59, type=code, data={'msg_reserved': mark})) - if name in self.UNPARSED: - # So the short read is recorded on a model parsed whole, as a - # cut there would record it: the field ``data`` follows the - # six octets of the MH header. - start = data.index(mark) - for count in range(size): - with self.subTest(maker=name, cut=start + count): - with self.assertRaises(FieldValueError): - MH(io.BytesIO(data[:start + count]), start + count, extension=True) - info = MH(io.BytesIO(data), len(data), extension=True).info - info.__update__({'msg_reserved': mark[:count], '__short_read__': ('data', start - 6 + count)}) - self.assertEqual(MH.from_data(info).data, data[:start + count]) - continue with self.subTest(maker=name): self.assertCutRebuilds(MH, data, mark, extension=True) - for name in ('_make_msg_fna', '_make_msg_has', '_make_msg_sq', *self.UNPARSED): + for name in ('_make_msg_fna', '_make_msg_has', '_make_msg_sq', '_make_msg_lri', '_make_msg_upa'): self.assertIn(name, swept) def test_ipv6_frag_and_route_cut_inside_reserved_octets(self) -> None: @@ -676,8 +658,6 @@ def update(key: 'str') -> 'object': return lambda info, value: info.__update__({key: value}) for name, code, size in self._mh_messages(): - if name in self.UNPARSED: - continue mark = bytes(range(0xA1, 0xA1 + size)) unit = 'octet' if size == 1 else 'octets' with self.subTest(maker=name):