diff --git a/docs/source/pcapkit/protocols/internet/hopopt.rst b/docs/source/pcapkit/protocols/internet/hopopt.rst index c87417efc..90200a420 100644 --- a/docs/source/pcapkit/protocols/internet/hopopt.rst +++ b/docs/source/pcapkit/protocols/internet/hopopt.rst @@ -218,6 +218,7 @@ Auxiliary Functions ~~~~~~~~~~~~~~~~~~~ .. autofunction:: pcapkit.protocols.schema.internet.hopopt.mpl_opt_seed_id_len +.. autofunction:: pcapkit.protocols.schema.internet.hopopt.pad_opt_data_len .. autofunction:: pcapkit.protocols.schema.internet.hopopt.smf_dpd_data_selector .. autofunction:: pcapkit.protocols.schema.internet.hopopt.smf_i_dpd_tid_selector diff --git a/docs/source/pcapkit/protocols/internet/ipv6_opts.rst b/docs/source/pcapkit/protocols/internet/ipv6_opts.rst index 1b734100a..5151c9167 100644 --- a/docs/source/pcapkit/protocols/internet/ipv6_opts.rst +++ b/docs/source/pcapkit/protocols/internet/ipv6_opts.rst @@ -218,6 +218,7 @@ Auxiliary Functions ~~~~~~~~~~~~~~~~~~~ .. autofunction:: pcapkit.protocols.schema.internet.ipv6_opts.mpl_opt_seed_id_len +.. autofunction:: pcapkit.protocols.schema.internet.ipv6_opts.pad_opt_data_len .. autofunction:: pcapkit.protocols.schema.internet.ipv6_opts.smf_dpd_data_selector .. autofunction:: pcapkit.protocols.schema.internet.ipv6_opts.smf_i_dpd_tid_selector diff --git a/docs/source/pcapkit/protocols/internet/mh.rst b/docs/source/pcapkit/protocols/internet/mh.rst index 0d238e25c..1f6545a95 100644 --- a/docs/source/pcapkit/protocols/internet/mh.rst +++ b/docs/source/pcapkit/protocols/internet/mh.rst @@ -386,6 +386,7 @@ Auxiliary Functions .. autofunction:: pcapkit.protocols.schema.internet.mh.mh_data_selector .. autofunction:: pcapkit.protocols.schema.internet.mh.mn_id_selector +.. autofunction:: pcapkit.protocols.schema.internet.mh.pad_opt_data_len Data Models ----------- diff --git a/pcapkit/protocols/internet/hopopt.py b/pcapkit/protocols/internet/hopopt.py index 6c0062361..fbea14130 100644 --- a/pcapkit/protocols/internet/hopopt.py +++ b/pcapkit/protocols/internet/hopopt.py @@ -334,11 +334,18 @@ def make(self, next_value = self._make_index(next, next_default, namespace=next_namespace, reversed=next_reversed, pack=False) - if options is not None: - options_value, total_length = self._make_hopopt_options(options) - length = math.ceil((total_length - 6) / 8) - else: - options_value, length = [], 0 + # NOTE: No options at all is not the same thing as no options area: the + # header is at least 8 octets, 2 of which are the fixed part, so the + # remaining 6 have to be padding options rather than nothing. Passing the + # empty list through the same path is what produces them -- returning + # ``[], 0`` here instead declared an 8-octet header and emitted 2. + options_value, total_length = self._make_hopopt_options( + options if options is not None else []) + + # NOTE: ``_make_hopopt_options`` has aligned the header, so this division + # is exact; rounding up here used to hide the 6-octet shortfall that the + # per-option alignment left behind. + length = (total_length - 6) // 8 return Schema_HOPOPT( next=next_value, # type: ignore[arg-type] @@ -538,6 +545,17 @@ def _read_opt_pad(self, schema: 'Schema_PadOption', *, options: 'Option') -> 'Da Raises: ProtocolError: If ``code`` is **NOT** ``0`` or ``1``. + Note: + A ``Pad1`` option occupies a single octet and carries no + ``Opt Data Len`` field, so its + :attr:`~pcapkit.protocols.data.internet.hopopt.PadOption.length` is + ``1`` rather than ``len + 2``. That one-octet wire shape is enforced + by :class:`~pcapkit.protocols.schema.internet.hopopt.PadOption` + itself, which sizes both the length octet and the padding data from + the option type; ``clen`` is therefore always ``0`` here for a + parsed ``Pad1``, and the check below only guards a schema built by + hand. + """ code, clen = schema.type, schema.len @@ -1175,6 +1193,47 @@ def _read_opt_ip_dff(self, schema: 'Schema_IPDFFOption', *, options: 'Option') - ) return opt + def _make_pad_options(self, offset: 'int') -> 'tuple[list[Schema_PadOption], int]': + """Make the padding options needed to align the header to 8 octets. + + Args: + offset: Number of octets emitted so far, **counting the two octets + of the fixed header** (``Next Header`` and ``Hdr Ext Len``) as + well as every option already in the list. + + Returns: + Tuple of the padding option schemas and the number of octets they + occupy. + + Note: + It is the *whole* extension header, fixed part included, that has to + be a multiple of 8 octets [:rfc:`8200#section-4.3`] -- which is why + the alignment is computed from ``offset`` rather than from the + length of the option that has just been emitted. Aligning each + option to 8 octets on its own leaves the options area a multiple of + 8 octets long, whereas :meth:`read` sizes it as + ``hdr_ext_len * 8 + 6`` -- six short of a multiple of 8 -- so the + constructed header declared 6 octets more than it actually carried. + + A ``PadN`` option spends two octets on its own type and + ``Opt Data Len`` fields before any padding data, so occupying + ``pad_len`` octets means an ``Opt Data Len`` of ``pad_len - 2``. + One or two octets are padded with ``Pad1`` options instead, that + being the only form which can occupy a single octet + [:rfc:`8200#section-4.2`]; two octets need two separate ``Pad1`` + schemas, since a schema instance is mutable and must not be shared + between two entries of the option list. + + """ + if offset % 8 == 0: + return [], 0 + + pad_len = 8 - (offset % 8) + if pad_len <= 2: + return [self._make_opt_pad(Enum_Option.Pad1, length=0) # type: ignore[arg-type] + for _ in range(pad_len)], pad_len + return [self._make_opt_pad(Enum_Option.PadN, length=pad_len - 2)], pad_len # type: ignore[arg-type] + def _make_hopopt_options(self, options: 'list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes] | Option') -> 'tuple[list[Schema_Option | bytes], int]': """Make options for HOPOPT. @@ -1184,6 +1243,14 @@ def _make_hopopt_options(self, options: 'list[Schema_Option | tuple[Enum_Option, Returns: Tuple of options and total length of options. + Note: + The returned length is that of the options area alone, so it is the + value :meth:`make` needs for ``hdr_ext_len``. Since the two octets + of the fixed header precede it, an options area is well formed only + when it is 6 octets short of a multiple of 8 -- including when it + holds no real options at all, in which case it is 6 octets of pure + padding rather than empty. + """ total_length = 0 if isinstance(options, list): @@ -1222,18 +1289,15 @@ def _make_hopopt_options(self, options: 'list[Schema_Option | tuple[Enum_Option, options_list.append(opt) total_length += opt_len - # force alignment to 8 octets - if opt_len % 8: - pad_len = 8 - (opt_len % 8) - if pad_len in (1, 2): - pad_opt = self._make_opt_pad(Enum_Option.Pad1, length=0) # type: ignore[arg-type] - else: - pad_opt = self._make_opt_pad(Enum_Option.PadN, length=pad_len) # type: ignore[arg-type] + # force alignment of the header (fixed part included) to 8 octets + pad_opts, pad_len = self._make_pad_options(total_length + 2) + options_list.extend(pad_opts) + total_length += pad_len - options_list.append(pad_opt) - if pad_len == 2: # need 2 Pad1 options - options_list.append(pad_opt) - total_length += pad_len + # an options area holding nothing at all is still 6 octets long + pad_opts, pad_len = self._make_pad_options(total_length + 2) + options_list.extend(pad_opts) + total_length += pad_len return options_list, total_length options_list = [] @@ -1256,18 +1320,15 @@ def _make_hopopt_options(self, options: 'list[Schema_Option | tuple[Enum_Option, options_list.append(opt) total_length += opt_len - # force alignment to 8 octets - if opt_len % 8: - pad_len = 8 - (opt_len % 8) - if pad_len in (1, 2): - pad_opt = self._make_opt_pad(Enum_Option.Pad1, length=0) # type: ignore[arg-type] - else: - pad_opt = self._make_opt_pad(Enum_Option.PadN, length=pad_len) # type: ignore[arg-type] + # force alignment of the header (fixed part included) to 8 octets + pad_opts, pad_len = self._make_pad_options(total_length + 2) + options_list.extend(pad_opts) + total_length += pad_len - options_list.append(pad_opt) - if pad_len == 2: # need 2 Pad1 options - options_list.append(pad_opt) - total_length += pad_len + # an options area holding nothing at all is still 6 octets long + pad_opts, pad_len = self._make_pad_options(total_length + 2) + options_list.extend(pad_opts) + total_length += pad_len return options_list, total_length def _make_opt_none(self, code: 'Enum_Option', opt: 'Optional[Data_UnassignedOption]' = None, *, @@ -1302,13 +1363,26 @@ def _make_opt_pad(self, code: 'Enum_Option', opt: 'Optional[Data_PadOption]' = N Args: code: option type value opt: option data - length: padding length + length: value of the ``Opt Data Len`` field, i.e. the number of + padding octets *after* the two octets of the option header **kwargs: arbitrary keyword arguments Returns: Constructured option schema. + Note: + :attr:`Data_PadOption.length + ` counts the + *whole* option, whereas :attr:`Schema_PadOption.len + ` is the + ``Opt Data Len`` field -- two octets fewer, and absent altogether for + a ``Pad1``. ``opt`` used to be ignored here, so re-making a parsed + padding option silently collapsed it to a single ``Pad1``. + """ + if opt is not None: + length = 0 if opt.type == Enum_Option.Pad1 else opt.length - 2 + if code == Enum_Option.Pad1 and length != 0: #raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid format') warn(f'{self.alias}: [OptNo {code}] invalid format', ProtocolWarning) diff --git a/pcapkit/protocols/internet/ipv6_opts.py b/pcapkit/protocols/internet/ipv6_opts.py index 7cb489067..cb1f55ace 100644 --- a/pcapkit/protocols/internet/ipv6_opts.py +++ b/pcapkit/protocols/internet/ipv6_opts.py @@ -345,11 +345,18 @@ def make(self, next_value = self._make_index(next, next_default, namespace=next_namespace, reversed=next_reversed, pack=False) - if options is not None: - options_value, total_length = self._make_ipv6_opts(options) - length = math.ceil((total_length - 6) / 8) - else: - options_value, length = [], 0 + # NOTE: No options at all is not the same thing as no options area: the + # header is at least 8 octets, 2 of which are the fixed part, so the + # remaining 6 have to be padding options rather than nothing. Passing the + # empty list through the same path is what produces them -- returning + # ``[], 0`` here instead declared an 8-octet header and emitted 2. + options_value, total_length = self._make_ipv6_opts( + options if options is not None else []) + + # NOTE: ``_make_ipv6_opts`` has aligned the header, so this division is + # exact; rounding up here used to hide the 6-octet shortfall that the + # per-option alignment left behind. + length = (total_length - 6) // 8 return Schema_IPv6_Opts( next=next_value, # type: ignore[arg-type] @@ -549,6 +556,18 @@ def _read_opt_pad(self, schema: 'Schema_PadOption', *, options: 'Option') -> 'Da Raises: ProtocolError: If ``code`` is **NOT** ``0`` or ``1``. + Note: + A ``Pad1`` option occupies a single octet and carries no + ``Opt Data Len`` field, so its + :attr:`~pcapkit.protocols.data.internet.ipv6_opts.PadOption.length` + is ``1`` rather than ``len + 2``. That one-octet wire shape is + enforced by + :class:`~pcapkit.protocols.schema.internet.ipv6_opts.PadOption` + itself, which sizes both the length octet and the padding data from + the option type; ``clen`` is therefore always ``0`` here for a + parsed ``Pad1``, and the check below only guards a schema built by + hand. + """ code, clen = schema.type, schema.len @@ -1186,6 +1205,47 @@ def _read_opt_ip_dff(self, schema: 'Schema_IPDFFOption', *, options: 'Option') - ) return opt + def _make_pad_options(self, offset: 'int') -> 'tuple[list[Schema_PadOption], int]': + """Make the padding options needed to align the header to 8 octets. + + Args: + offset: Number of octets emitted so far, **counting the two octets + of the fixed header** (``Next Header`` and ``Hdr Ext Len``) as + well as every option already in the list. + + Returns: + Tuple of the padding option schemas and the number of octets they + occupy. + + Note: + It is the *whole* extension header, fixed part included, that has to + be a multiple of 8 octets [:rfc:`8200#section-4.6`] -- which is why + the alignment is computed from ``offset`` rather than from the + length of the option that has just been emitted. Aligning each + option to 8 octets on its own leaves the options area a multiple of + 8 octets long, whereas :meth:`read` sizes it as + ``hdr_ext_len * 8 + 6`` -- six short of a multiple of 8 -- so the + constructed header declared 6 octets more than it actually carried. + + A ``PadN`` option spends two octets on its own type and + ``Opt Data Len`` fields before any padding data, so occupying + ``pad_len`` octets means an ``Opt Data Len`` of ``pad_len - 2``. + One or two octets are padded with ``Pad1`` options instead, that + being the only form which can occupy a single octet + [:rfc:`8200#section-4.2`]; two octets need two separate ``Pad1`` + schemas, since a schema instance is mutable and must not be shared + between two entries of the option list. + + """ + if offset % 8 == 0: + return [], 0 + + pad_len = 8 - (offset % 8) + if pad_len <= 2: + return [self._make_opt_pad(Enum_Option.Pad1, length=0) # type: ignore[arg-type] + for _ in range(pad_len)], pad_len + return [self._make_opt_pad(Enum_Option.PadN, length=pad_len - 2)], pad_len # type: ignore[arg-type] + def _make_ipv6_opts(self, options: 'list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes] | Option') -> 'tuple[list[Schema_Option | bytes], int]': """Make options for IPv6-Opts. @@ -1195,6 +1255,14 @@ def _make_ipv6_opts(self, options: 'list[Schema_Option | tuple[Enum_Option, dict Returns: Tuple of options and total length of options. + Note: + The returned length is that of the options area alone, so it is the + value :meth:`make` needs for ``hdr_ext_len``. Since the two octets + of the fixed header precede it, an options area is well formed only + when it is 6 octets short of a multiple of 8 -- including when it + holds no real options at all, in which case it is 6 octets of pure + padding rather than empty. + """ total_length = 0 if isinstance(options, list): @@ -1233,18 +1301,15 @@ def _make_ipv6_opts(self, options: 'list[Schema_Option | tuple[Enum_Option, dict options_list.append(opt) total_length += opt_len - # force alignment to 8 octets - if opt_len % 8: - pad_len = 8 - (opt_len % 8) - if pad_len in (1, 2): - pad_opt = self._make_opt_pad(Enum_Option.Pad1, length=0) # type: ignore[arg-type] - else: - pad_opt = self._make_opt_pad(Enum_Option.PadN, length=pad_len) # type: ignore[arg-type] + # force alignment of the header (fixed part included) to 8 octets + pad_opts, pad_len = self._make_pad_options(total_length + 2) + options_list.extend(pad_opts) + total_length += pad_len - options_list.append(pad_opt) - if pad_len == 2: # need 2 Pad1 options - options_list.append(pad_opt) - total_length += pad_len + # an options area holding nothing at all is still 6 octets long + pad_opts, pad_len = self._make_pad_options(total_length + 2) + options_list.extend(pad_opts) + total_length += pad_len return options_list, total_length options_list = [] @@ -1267,18 +1332,15 @@ def _make_ipv6_opts(self, options: 'list[Schema_Option | tuple[Enum_Option, dict options_list.append(opt) total_length += opt_len - # force alignment to 8 octets - if opt_len % 8: - pad_len = 8 - (opt_len % 8) - if pad_len in (1, 2): - pad_opt = self._make_opt_pad(Enum_Option.Pad1, length=0) # type: ignore[arg-type] - else: - pad_opt = self._make_opt_pad(Enum_Option.PadN, length=pad_len) # type: ignore[arg-type] + # force alignment of the header (fixed part included) to 8 octets + pad_opts, pad_len = self._make_pad_options(total_length + 2) + options_list.extend(pad_opts) + total_length += pad_len - options_list.append(pad_opt) - if pad_len == 2: # need 2 Pad1 options - options_list.append(pad_opt) - total_length += pad_len + # an options area holding nothing at all is still 6 octets long + pad_opts, pad_len = self._make_pad_options(total_length + 2) + options_list.extend(pad_opts) + total_length += pad_len return options_list, total_length def _make_opt_none(self, code: 'Enum_Option', opt: 'Optional[Data_UnassignedOption]' = None, *, @@ -1313,13 +1375,26 @@ def _make_opt_pad(self, code: 'Enum_Option', opt: 'Optional[Data_PadOption]' = N Args: code: option type value opt: option data - length: padding length + length: value of the ``Opt Data Len`` field, i.e. the number of + padding octets *after* the two octets of the option header **kwargs: arbitrary keyword arguments Returns: Constructured option schema. + Note: + :attr:`Data_PadOption.length + ` counts + the *whole* option, whereas :attr:`Schema_PadOption.len + ` is the + ``Opt Data Len`` field -- two octets fewer, and absent altogether for + a ``Pad1``. ``opt`` used to be ignored here, so re-making a parsed + padding option silently collapsed it to a single ``Pad1``. + """ + if opt is not None: + length = 0 if opt.type == Enum_Option.Pad1 else opt.length - 2 + if code == Enum_Option.Pad1 and length != 0: #raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid format') warn(f'{self.alias}: [OptNo {code}] invalid format', ProtocolWarning) diff --git a/pcapkit/protocols/internet/mh.py b/pcapkit/protocols/internet/mh.py index d82ec48cc..af426c80b 100644 --- a/pcapkit/protocols/internet/mh.py +++ b/pcapkit/protocols/internet/mh.py @@ -746,9 +746,16 @@ def make(self, else: raise ProtocolError(f'MH: [Type {type_val}] invalid format') + # NOTE: The header has to be a multiple of 8 octets, so the message data + # needs padding until ``len(data) + 6`` is aligned. Rounding ``length`` up + # without emitting that padding -- which is what ``math.ceil`` used to do + # here -- declares a header longer than the bytes that follow it, and the + # re-parse then reads whatever happens to be past the end of the buffer. + data_val = self._pad_mh_message(data_val) + return Schema_MH( next=next_val, - length=math.ceil((len(data_val) + 6) / 8) - 1, + length=(len(data_val) + 6) // 8 - 1, type=type_val, chksum=chksum, data=data_val, @@ -2978,6 +2985,111 @@ def _make_msg_hack(self, message: 'Optional[Data_HandoverAcknowledgeMessage]' = # TODO: Implement other message types. + def _make_pad_options(self, offset: 'int') -> 'tuple[list[Schema_PadOption], int]': + """Make the padding options needed to align the header to 8 octets. + + Args: + offset: Number of octets emitted so far, **counting the six octets + of the fixed header** (``Payload Proto``, ``Header Len``, + ``MH Type``, ``Reserved`` and ``Checksum``) as well as the whole + message data. + + Returns: + Tuple of the padding option schemas and the number of octets they + occupy. + + Note: + The Mobility Header is a multiple of 8 octets and + :attr:`~pcapkit.protocols.schema.internet.mh.MH.length` counts those + units less one [:rfc:`6275#section-6.1.1`]. Its fixed part is + **six** octets, not the two of an IPv6 options header, so it is + ``len(data) + 6`` that has to be aligned -- which means the message + data is well formed only when it is 2 octets past a multiple of 8, + a different modulus from the one the extension headers use. + + A ``PadN`` option spends two octets on its own type and + ``Option Length`` fields before any padding data, so occupying + ``pad_len`` octets means an ``Option Length`` of ``pad_len - 2``. + One or two octets are padded with ``Pad1`` options instead, that + being the only form which can occupy a single octet + [:rfc:`6275#section-6.2.5`]; two octets need two separate ``Pad1`` + schemas, since a schema instance is mutable and must not be shared + between two entries of the option list. + + """ + if offset % 8 == 0: + return [], 0 + + pad_len = 8 - (offset % 8) + if pad_len <= 2: + return [self._make_opt_pad(Enum_Option.Pad1, length=0) # type: ignore[arg-type] + for _ in range(pad_len)], pad_len + return [self._make_opt_pad(Enum_Option.PadN, length=pad_len - 2)], pad_len # type: ignore[arg-type] + + def _pad_mh_message(self, data: 'Schema_Packet | bytes') -> 'Schema_Packet | bytes': + """Pad an MH message so that the Mobility Header aligns to 8 octets. + + Args: + data: Constructed message data. + + Returns: + The message data, with padding options appended if any were needed. + + Note: + The padding goes into the message's mobility options, which is where + :rfc:`6275#section-6.2.5` puts it. A message whose body is opaque has + nowhere to put it -- + :class:`~pcapkit.protocols.schema.internet.mh.UnknownMessage` and + :class:`~pcapkit.protocols.schema.internet.mh.ExperimentalMessage` + carry raw bytes and no options, as does a ``data`` argument given + directly as :obj:`bytes` -- so for those the padding is appended to the + message body itself and a + :class:`~pcapkit.utilities.warnings.ProtocolWarning` says so. + + Appending is necessary rather than optional: ``length`` is + ``(len(data) + 6) // 8 - 1``, which floors, so leaving an opaque body + short emitted 10, 12 or 14 octets while declaring 8, and a parser reads + 8 and misinterprets the remainder. Since the caller asked for a packet + to be built and the shortfall is recoverable, completing it beats + refusing -- the warning is there because the emitted body is then not + byte-for-byte what was handed in. + + """ + pad_opts, pad_len = self._make_pad_options(len(data) + 6) + if pad_len == 0: + return data + + options = getattr(data, 'options', None) + # NOTE: The ``isinstance`` test comes first so that the type checker can + # narrow ``data`` for the assignment below; at runtime ``getattr`` has + # already covered the :obj:`bytes` case by returning :obj:`None`. + if isinstance(data, bytes) or options is None: + warn(f'{self.alias}: message data of {len(data)} octets carries no ' + f'mobility options to hold padding, so {pad_len} octet(s) were ' + 'appended to the message body to align the header', + ProtocolWarning) + if isinstance(data, bytes): + return data + b'\x00' * pad_len + + # An opaque schema body -- UnknownMessage, ExperimentalMessage -- keeps + # its content in ``data`` rather than in options, so that is where the + # octets go. Rebound rather than mutated in place so that ``len()`` and + # ``pack()`` see the change, exactly as for the options branch below. + body = getattr(data, 'data', None) + if not isinstance(body, bytes): + raise ProtocolError( + f'{self.alias}: message data of {len(data)} octets needs ' + f'{pad_len} octet(s) of padding, but the body is neither bytes ' + 'nor a schema carrying bytes, so there is nowhere to put it') + data.data = body + b'\x00' * pad_len + return data + + # NOTE: Rebinding the attribute rather than mutating the list in place is + # what marks the schema as updated, so that ``len()`` and ``pack()`` take + # the padding into account. + data.options = list(options) + pad_opts + return data + def _make_mh_options(self, options: 'Option | list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes]') -> 'list[Schema_Option | bytes]': """Make options for MH. @@ -3057,15 +3169,25 @@ def _make_opt_pad(self, type: 'Enum_Option', option: 'Optional[Data_PadOption]' Args: type: Option type. option: Option data model. - length: Padding length. + length: Value of the ``Option Length`` field, i.e. the number of + padding octets *after* the two octets of the option header. **kwargs: Arbitrary keyword arguments. Returns: Constructed option schema. + Note: + :attr:`Data_PadOption.length + ` counts the + *whole* option, whereas :attr:`Schema_PadOption.length + ` is the + ``Option Length`` field -- two octets fewer, and absent altogether + for a ``Pad1``. Copying one into the other unconverted is why + re-making a parsed ``PadN`` used to come back two octets too long. + """ - if option is not None: - length = option.length + if option is not None: + length = 0 if option.type == Enum_Option.Pad1 else option.length - 2 if type == Enum_Option.Pad1 and length != 0: # raise ProtocolError(f'{self.alias}: [OptNo {type}] invalid format') diff --git a/pcapkit/protocols/schema/internet/hopopt.py b/pcapkit/protocols/schema/internet/hopopt.py index e3d539a58..6532eccab 100644 --- a/pcapkit/protocols/schema/internet/hopopt.py +++ b/pcapkit/protocols/schema/internet/hopopt.py @@ -224,6 +224,39 @@ def quick_start_data_selector(pkt: 'dict[str, Any]') -> 'Field': return SchemaField(length=5, schema=schema) +def pad_opt_data_len(pkt: 'dict[str, Any]') -> 'int': + """Return the length of the padding data of a padding option. + + Args: + pkt: Padding option unpacked schema. + + Returns: + Number of padding octets carried after the option header, i.e. the + value of the ``Opt Data Len`` field of a ``PadN`` option, and zero for + a ``Pad1`` option, which carries no such field. + + Note: + A ``Pad1`` option is a single octet with neither an ``Opt Data Len`` + field nor any option data (c.f. :rfc:`8200#section-4.2`), which is why + :attr:`Option.len` is declared as a + :class:`~pcapkit.corekit.fields.misc.ConditionalField` and is skipped + for it. A skipped conditional field is *recorded* in the packet data as + :data:`~pcapkit.corekit.fields.field.NoValue`, rather than being left + out of it, so the test below has to be on the **value** and not on the + presence of the key: ``pkt.get('len', 0)`` on its own hands that + :obj:`~pcapkit.corekit.fields.field.NoValueType` straight to + :class:`~pcapkit.corekit.fields.strings.PaddingField`, where it becomes + an unusable :mod:`struct` template and surfaces much later as an opaque + :exc:`struct.error` -- which is exactly how a ``Pad1`` option used to + fail to parse. + + """ + length = pkt.get('len', 0) + if not isinstance(length, int): # ``NoValue`` (skipped) or :obj:`None` (unset) + return 0 + return length + + class Option(EnumSchema[Enum_Option]): """Header schema for HOPOPT options.""" @@ -267,10 +300,19 @@ def __init__(self, type: 'Enum_Option', len: 'int', data: 'bytes') -> 'None': .. @schema_final class PadOption(Option, code=[Enum_Option.Pad1, Enum_Option.PadN]): - """Header schema for HOPOPT padding options.""" + """Header schema for HOPOPT padding options. + + The two padding options do **not** share a wire shape: a ``Pad1`` option is + a lone type octet, whereas a ``PadN`` option is a type octet, an + ``Opt Data Len`` octet, and that many padding octets [:rfc:`8200#section-4.2`]. + Both the ``Opt Data Len`` octet (:attr:`Option.len`) and the padding data + (:attr:`self.pad `) are therefore sized from the option type, + so that a ``Pad1`` option consumes exactly one octet. + + """ #: Padding. - pad: 'bytes' = PaddingField(length=lambda pkt: pkt.get('len', 0)) + pad: 'bytes' = PaddingField(length=pad_opt_data_len) if TYPE_CHECKING: def __init__(self, type: 'Enum_Option', len: 'int') -> 'None': ... diff --git a/pcapkit/protocols/schema/internet/ipv6_opts.py b/pcapkit/protocols/schema/internet/ipv6_opts.py index ee8715a69..6d2daae4c 100644 --- a/pcapkit/protocols/schema/internet/ipv6_opts.py +++ b/pcapkit/protocols/schema/internet/ipv6_opts.py @@ -224,6 +224,39 @@ def quick_start_data_selector(pkt: 'dict[str, Any]') -> 'Field': return SchemaField(length=5, schema=schema) +def pad_opt_data_len(pkt: 'dict[str, Any]') -> 'int': + """Return the length of the padding data of a padding option. + + Args: + pkt: Padding option unpacked schema. + + Returns: + Number of padding octets carried after the option header, i.e. the + value of the ``Opt Data Len`` field of a ``PadN`` option, and zero for + a ``Pad1`` option, which carries no such field. + + Note: + A ``Pad1`` option is a single octet with neither an ``Opt Data Len`` + field nor any option data (c.f. :rfc:`8200#section-4.2`), which is why + :attr:`Option.len` is declared as a + :class:`~pcapkit.corekit.fields.misc.ConditionalField` and is skipped + for it. A skipped conditional field is *recorded* in the packet data as + :data:`~pcapkit.corekit.fields.field.NoValue`, rather than being left + out of it, so the test below has to be on the **value** and not on the + presence of the key: ``pkt.get('len', 0)`` on its own hands that + :obj:`~pcapkit.corekit.fields.field.NoValueType` straight to + :class:`~pcapkit.corekit.fields.strings.PaddingField`, where it becomes + an unusable :mod:`struct` template and surfaces much later as an opaque + :exc:`struct.error` -- which is exactly how a ``Pad1`` option used to + fail to parse. + + """ + length = pkt.get('len', 0) + if not isinstance(length, int): # ``NoValue`` (skipped) or :obj:`None` (unset) + return 0 + return length + + class Option(EnumSchema[Enum_Option]): """Header schema for IPv6-Opts options.""" @@ -267,10 +300,19 @@ def __init__(self, type: 'Enum_Option', len: 'int', data: 'bytes') -> 'None': .. @schema_final class PadOption(Option, code=[Enum_Option.Pad1, Enum_Option.PadN]): - """Header schema for IPv6-Opts padding options.""" + """Header schema for IPv6-Opts padding options. + + The two padding options do **not** share a wire shape: a ``Pad1`` option is + a lone type octet, whereas a ``PadN`` option is a type octet, an + ``Opt Data Len`` octet, and that many padding octets [:rfc:`8200#section-4.2`]. + Both the ``Opt Data Len`` octet (:attr:`Option.len`) and the padding data + (:attr:`self.pad `) are therefore sized from the option type, + so that a ``Pad1`` option consumes exactly one octet. + + """ #: Padding. - pad: 'bytes' = PaddingField(length=lambda pkt: pkt.get('len', 0)) + pad: 'bytes' = PaddingField(length=pad_opt_data_len) if TYPE_CHECKING: def __init__(self, type: 'Enum_Option', len: 'int') -> 'None': ... diff --git a/pcapkit/protocols/schema/internet/mh.py b/pcapkit/protocols/schema/internet/mh.py index d798e2c8c..4b4dea99e 100644 --- a/pcapkit/protocols/schema/internet/mh.py +++ b/pcapkit/protocols/schema/internet/mh.py @@ -188,6 +188,39 @@ def mn_id_selector(pkt: 'dict[str, Any]') -> 'Field': return BytesField(length=pkt['length'] - 1) +def pad_opt_data_len(pkt: 'dict[str, Any]') -> 'int': + """Return the length of the padding data of a padding option. + + Args: + pkt: Padding option unpacked schema. + + Returns: + Number of padding octets carried after the option header, i.e. the + value of the ``Option Length`` field of a ``PadN`` option, and zero for + a ``Pad1`` option, which carries no such field. + + Note: + A ``Pad1`` option is a single octet with neither an ``Option Length`` + field nor any option data (c.f. :rfc:`6275#section-6.2.5`), which is why + :attr:`Option.length` is declared as a + :class:`~pcapkit.corekit.fields.misc.ConditionalField` and is skipped + for it. A skipped conditional field is *recorded* in the packet data as + :data:`~pcapkit.corekit.fields.field.NoValue`, rather than being left + out of it, so the test below has to be on the **value** and not on the + presence of the key: ``pkt.get('length', 0)`` on its own hands that + :class:`~pcapkit.corekit.fields.field.NoValueType` straight to + :class:`~pcapkit.corekit.fields.strings.PaddingField`, where it becomes + an unusable :mod:`struct` template and surfaces much later as an opaque + :exc:`struct.error` -- which is exactly how a ``Pad1`` option used to + fail to parse. + + """ + length = pkt.get('length', 0) + if not isinstance(length, int): # ``NoValue`` (skipped) or :obj:`None` (unset) + return 0 + return length + + @schema_final class MH(Schema): """Header schema for MH packets.""" @@ -256,10 +289,20 @@ def __init__(self, type: 'Enum_Option', length: 'int', data: 'bytes') -> 'None': @schema_final class PadOption(Option, code=[Enum_Option.Pad1, Enum_Option.PadN]): - """Header schema for MH padding options.""" + """Header schema for MH padding options. + + The two padding options do **not** share a wire shape: a ``Pad1`` option is + a lone type octet, whereas a ``PadN`` option is a type octet, an + ``Option Length`` octet, and that many padding octets + [:rfc:`6275#section-6.2.5`]. Both the ``Option Length`` octet + (:attr:`Option.length`) and the padding data + (:attr:`self.data `) are therefore sized from the option + type, so that a ``Pad1`` option consumes exactly one octet. + + """ #: Option data. - data: 'bytes' = PaddingField(length=lambda pkt: pkt.get('length', 0)) + data: 'bytes' = PaddingField(length=pad_opt_data_len) if TYPE_CHECKING: def __init__(self, type: 'Enum_Option', length: 'int') -> 'None': ... diff --git a/tests/protocols/internet/test_ipv6_extension_unit.py b/tests/protocols/internet/test_ipv6_extension_unit.py index f99f635fb..22870d570 100644 --- a/tests/protocols/internet/test_ipv6_extension_unit.py +++ b/tests/protocols/internet/test_ipv6_extension_unit.py @@ -2,6 +2,7 @@ import datetime import importlib.util +import io from ipaddress import ip_address import types import unittest @@ -472,10 +473,16 @@ def _assert_option_header_read_make_and_registry_edges(self, protocol_cls: type) self.assertEqual(proto.read(length=8), 'decoded') decode.assert_called_once() + # NOTE: "No options" is not the same as "no options area". The header is at + # least (0 + 1) * 8 = 8 octets and 2 of those are the fixed part, so the + # remaining 6 have to be padding options -- an empty options list is not a + # thing that can go on the wire, and asserting for one here used to pass + # only because ``make`` emitted a 2-octet header while declaring 8. made_empty = proto.make(next=TransType.TCP) self.assertEqual(made_empty.next, TransType.TCP) self.assertEqual(made_empty.len, 0) - self.assertEqual(made_empty.options, []) + self.assertEqual([type(item).__name__ for item in made_empty.options], ['PadOption']) + self.assertEqual(len(bytes(made_empty)), 8) made_with_option = proto.make( next=TransType.UDP, @@ -825,11 +832,19 @@ def _assert_option_constructors_cover_branchy_values(self, protocol_cls: type) - seq=3, ).flags['dup']) + # NOTE: The lengths below are the length of the *options area*, and the two + # octets of the fixed header precede it, so a well-formed options area is 6 + # octets short of a multiple of 8 -- never a multiple of 8 itself. Each + # assertion is followed by the octet count that justifies it, since these + # numbers used to be 6 too large (or to account padding that was emitted two + # octets longer than it was counted as). + options, total_length = make_options([ b'\x00', bytes([Option.CALIPSO]) + b'123456789', ]) - self.assertEqual(total_length, 16) + # 10 octets of CALIPSO + a 4-octet PadN = 14; 2 + 14 = 16 = 2 * 8 + self.assertEqual(total_length, 14) self.assertEqual([type(item).__name__ if not isinstance(item, bytes) else 'bytes' for item in options], ['bytes', 'PadOption']) schema_options, schema_length = make_options([ @@ -837,21 +852,31 @@ def _assert_option_constructors_cover_branchy_values(self, protocol_cls: type) - schema.PadOption(type=Option.Pad1, len=0), (Option.PadN, {}), ]) - self.assertEqual(schema_length, 8) + # the 8-octet option + a 6-octet PadN = 14; 2 + 14 = 16 = 2 * 8 + self.assertEqual(schema_length, 14) self.assertEqual(type(schema_options[0]).__name__, 'UnassignedOption') + # NOTE: Two ``Pad1`` options are emitted when the shortfall is exactly 2 + # octets, i.e. when ``2 + opt_len`` is 6 past a multiple of 8. That is a + # 4-octet option, not the 6-octet one this case used to pass -- 2 + 6 is + # already aligned and now needs no padding at all. pad1_options, pad1_length = make_options([ - bytes([Option.CALIPSO]) + b'12345', + bytes([Option.CALIPSO]) + b'123', ]) - self.assertEqual(pad1_length, 8) + # 4 octets of CALIPSO + two 1-octet Pad1 = 6; 2 + 6 = 8 + self.assertEqual(pad1_length, 6) self.assertEqual([type(item).__name__ if not isinstance(item, bytes) else 'bytes' for item in pad1_options], ['bytes', 'PadOption', 'PadOption']) + # NOTE: Likewise, the option that needs no padding is the 6-octet one; the + # 8-octet option this case used to pass leaves 2 + 8 = 10 and now draws 6 + # octets of padding. no_pad_options, no_pad_length = make_options([ - bytes([Option.CALIPSO]) + b'1234567', + bytes([Option.CALIPSO]) + b'12345', ]) - self.assertEqual(no_pad_length, 8) - self.assertEqual(no_pad_options, [bytes([Option.CALIPSO]) + b'1234567']) + # 6 octets of CALIPSO, nothing else; 2 + 6 = 8 + self.assertEqual(no_pad_length, 6) + self.assertEqual(no_pad_options, [bytes([Option.CALIPSO]) + b'12345']) dict_options, dict_length = make_options(OrderedMultiDict([ (Option.Pad1, data.PadOption(type=Option.Pad1, length=1, **base)), @@ -863,10 +888,26 @@ def _assert_option_constructors_cover_branchy_values(self, protocol_cls: type) - (Option.get(254), data.UnassignedOption(type=Option.get(254), length=6, data=b'abcd', **base)), ])) - self.assertEqual(dict_length, 24) + # 3 (tun) + 3 (PadN) + 8 + 6 + 1 + 1 (two Pad1) = 22; 2 + 22 = 24 = 3 * 8 + self.assertEqual(dict_length, 22) self.assertEqual(type(dict_options[0]).__name__, 'TunnelEncapsulationLimitOption') self.assertEqual(type(dict_options[-3]).__name__, 'UnassignedOption') + # every options area the constructor produces has to leave the whole header + # a multiple of 8 octets long, and to account for exactly what it emitted + for label, (built, built_len) in { + 'bytes + padn': (options, total_length), + 'schema + padn': (schema_options, schema_length), + 'two pad1': (pad1_options, pad1_length), + 'no padding': (no_pad_options, no_pad_length), + 'from dict': (dict_options, dict_length), + }.items(): + with self.subTest(options=label): + emitted = b''.join(item if isinstance(item, bytes) else item.pack() + for item in built) + self.assertEqual(len(emitted), built_len) + self.assertEqual((built_len + 2) % 8, 0) + def test_hopopt_option_constructors_cover_branchy_values(self) -> None: from pcapkit.protocols.internet.hopopt import HOPOPT @@ -1284,6 +1325,197 @@ def test_ipv6_route_schema_selector_and_rpl_post_process_branches(self) -> None: with_dst.post_process({'dst': ip_address('2001:db8::ffff')}) self.assertEqual([str(item) for item in with_dst.ip], ['2001:db8::1', '2001:db8::2']) + def _assert_padding_options_parse_from_the_wire(self, protocol_cls: type) -> None: + """A ``Pad1`` option must consume exactly one octet, wherever it sits. + + Per :rfc:`8200#section-4.2` the two padding options do not share a wire + shape: ``Pad1`` is a lone type octet with no ``Opt Data Len`` field at + all, while ``PadN`` is a type octet, a length octet, and that many data + octets. Reading a length octet that is not there consumes the *next* + option's type byte instead, so every case below puts something after + the padding -- which is what made the defect visible. + + The header is ``next`` = 59 (``IPv6-NoNxt``) with a header extension + length of 0, i.e. an 8-octet header whose options area is the 6 octets + that follow. A parse that mis-sizes any one option therefore also fails + the ``counter != length`` check at the end of the option loop. + + """ + from pcapkit.const.ipv6.option import Option + + header = b'\x3b\x00' + cases = ( + ('pad1 then padn', + b'\x00' + b'\x01\x03\x00\x00\x00', + [(Option.Pad1, 1), (Option.PadN, 5)]), + ('pad1 then a real option then pad1', + b'\x00' + b'\x05\x02\x00\x00' + b'\x00', + [(Option.Pad1, 1), (Option.Router_Alert, 4), (Option.Pad1, 1)]), + ('six pad1 in a row', + b'\x00' * 6, + [(Option.Pad1, 1)] * 6), + ('pad1 as the final octet', + b'\x01\x03\x00\x00\x00' + b'\x00', + [(Option.PadN, 5), (Option.Pad1, 1)]), + ('padn is unaffected', + b'\x01\x04\x00\x00\x00\x00', + [(Option.PadN, 6)]), + ('pad1 run then a minimal padn', + b'\x00\x00\x00' + b'\x01\x01\x00', + [(Option.Pad1, 1), (Option.Pad1, 1), (Option.Pad1, 1), (Option.PadN, 3)]), + ) + + for name, options, expected in cases: + with self.subTest(case=name): + raw = header + options + proto = protocol_cls(raw, extension=True) + + self.assertEqual(proto.info.length, 8) + self.assertEqual( + [(code, opt.length) for code, opt in proto.info.options.items(multi=True)], + expected, + ) + + # 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_padding_options_parse_from_the_wire(self) -> None: + from pcapkit.protocols.internet.hopopt import HOPOPT + + self._assert_padding_options_parse_from_the_wire(HOPOPT) + + def test_ipv6_opts_padding_options_parse_from_the_wire(self) -> None: + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + + self._assert_padding_options_parse_from_the_wire(IPv6_Opts) + + def _assert_padding_option_schema_sizes_itself(self, protocol_cls: type) -> None: + """The padding option schema, on its own. + + :func:`~pcapkit.protocols.schema.internet.hopopt.pad_opt_data_len` is + the whole of the read-side fix, so it is tested directly as well as + through a parse: it has to read a skipped conditional field -- which is + recorded as :data:`~pcapkit.corekit.fields.field.NoValue`, not omitted -- + as zero padding octets rather than passing it on to + :class:`~pcapkit.corekit.fields.strings.PaddingField`. + + """ + from pcapkit.const.ipv6.option import Option + from pcapkit.corekit.fields.field import NoValue + from pcapkit.protocols.schema.internet import hopopt as hopopt_schema + from pcapkit.protocols.schema.internet import ipv6_opts as opts_schema + + schema = hopopt_schema if protocol_cls.__name__ == 'HOPOPT' else opts_schema + + self.assertEqual(schema.pad_opt_data_len({}), 0) + self.assertEqual(schema.pad_opt_data_len({'len': NoValue}), 0) + self.assertEqual(schema.pad_opt_data_len({'len': None}), 0) + self.assertEqual(schema.pad_opt_data_len({'len': 4}), 4) + + # a Pad1 option on its own: one octet read, and the octet after it left + # alone for whatever comes next + stream = io.BytesIO(b'\x00\x2a') + pad1 = schema.PadOption.unpack(stream, 2, {}) + self.assertEqual(pad1.type, Option.Pad1) + self.assertEqual(pad1.len, 0) + self.assertEqual(len(pad1), 1) + self.assertEqual(bytes(pad1), b'\x00') + self.assertEqual(stream.tell(), 1) + + # a PadN option still spans its length octet plus that many data octets + stream = io.BytesIO(b'\x01\x02\xaa\xbb\xcc') + padn = schema.PadOption.unpack(stream, 5, {}) + self.assertEqual(padn.type, Option.PadN) + self.assertEqual(padn.len, 2) + self.assertEqual(len(padn), 4) + self.assertEqual(bytes(padn), b'\x01\x02\xaa\xbb') + self.assertEqual(stream.tell(), 4) + + def _assert_constructed_header_round_trips(self, protocol_cls: type) -> None: + """Construction has to produce a header that parses back to itself. + + Constructing one of these protocols packs the header and immediately + re-parses the bytes, so any disagreement between the declared + ``Hdr Ext Len`` and the octets actually emitted surfaces here as a + :exc:`~pcapkit.utilities.exceptions.ProtocolError` from :meth:`read`. + Every case below therefore both exercises the padding branches and + proves that ``(hdr_ext_len + 1) * 8`` is the real length of the header. + + The option lengths are chosen for the padding they draw: the whole + header, its two fixed octets included, has to be a multiple of 8, so an + option of ``8n + 5`` octets needs a single ``Pad1``, one of ``8n + 4`` + needs two, and anything else needs a ``PadN``. + + """ + from pcapkit.const.ipv6.option import Option + from pcapkit.const.reg.transtype import TransType + + cases = ( + # a 5-octet option leaves one octet, which only a Pad1 can fill + ('one pad1', [(Option.get(250), {'data': b'abc'})], 8, + [(Option.get(250), 5), (Option.Pad1, 1)]), + # a 4-octet option leaves two, i.e. two Pad1 rather than a PadN + ('two pad1', [(Option.Router_Alert, {'alert': 0})], 8, + [(Option.Router_Alert, 4), (Option.Pad1, 1), (Option.Pad1, 1)]), + # three or more spare octets are a PadN, whose own two octets of + # overhead have to come out of the padding it is asked to occupy + ('padn', [(Option.Tunnel_Encapsulation_Limit, {'limit': 3})], 8, + [(Option.Tunnel_Encapsulation_Limit, 3), (Option.PadN, 3)]), + # an option that already lands on the boundary draws no padding + ('no padding', [(Option.Jumbo_Payload, {'jumbo_len': 70000})], 8, + [(Option.Jumbo_Payload, 6)]), + # more than one option, each aligned in turn + ('two options', [(Option.Tunnel_Encapsulation_Limit, {'limit': 3}), + (Option.Router_Alert, {'alert': 0})], 16, + [(Option.Tunnel_Encapsulation_Limit, 3), (Option.PadN, 3), + (Option.Router_Alert, 4), (Option.PadN, 4)]), + # an option needing more than 8 octets of its own + ('long option', [(Option.Home_Address, {'addr': '2001:db8::3'})], 24, + [(Option.Home_Address, 18), (Option.PadN, 4)]), + # no options at all is 6 octets of padding, not an empty header + ('no options', [], 8, [(Option.PadN, 6)]), + ) + + for name, options, size, expected in cases: + with self.subTest(case=name): + built = protocol_cls(next=TransType.IPv6_NoNxt, options=list(options)) + raw = bytes(built) + + self.assertEqual(len(raw), size) + self.assertEqual(len(raw) % 8, 0) + # the declared length and the emitted length must be the same + self.assertEqual((raw[1] + 1) * 8, len(raw)) + + parsed = protocol_cls(raw, extension=True) + self.assertEqual(parsed.info.length, size) + self.assertEqual( + [(code, opt.length) + for code, opt in parsed.info.options.items(multi=True)], + expected, + ) + self.assertEqual(bytes(parsed.__header__), raw) + + def test_hopopt_padding_option_schema_sizes_itself(self) -> None: + from pcapkit.protocols.internet.hopopt import HOPOPT + + self._assert_padding_option_schema_sizes_itself(HOPOPT) + + def test_ipv6_opts_padding_option_schema_sizes_itself(self) -> None: + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + + self._assert_padding_option_schema_sizes_itself(IPv6_Opts) + + def test_hopopt_constructed_header_round_trips(self) -> None: + from pcapkit.protocols.internet.hopopt import HOPOPT + + self._assert_constructed_header_round_trips(HOPOPT) + + def test_ipv6_opts_constructed_header_round_trips(self) -> None: + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + + self._assert_constructed_header_round_trips(IPv6_Opts) + def test_option_registries_are_not_clobbered_by_a_nested_enum_registry(self) -> None: """Option type 0 must resolve to the padding option in every module. diff --git a/tests/protocols/internet/test_mh_unit.py b/tests/protocols/internet/test_mh_unit.py index f8f424bbb..70c75a8b8 100644 --- a/tests/protocols/internet/test_mh_unit.py +++ b/tests/protocols/internet/test_mh_unit.py @@ -590,7 +590,11 @@ def make_extension(code: CGAExtension, extension: data.UnknownExtension | None = data=SimpleNamespace(data=b'read'), ) self.assertEqual(proto.read(extension=True).data, b'read') - self.assertEqual(proto.make(type=custom_packet, data={'value': b'dict'}).data, b'dict') + # this constructor returns raw bytes, and 6 + 4 octets is 6 short of + # 16, so the body is padded to align the header; before that it emitted + # 10 octets while declaring 8 + self.assertEqual(proto.make(type=custom_packet, data={'value': b'dict'}).data, + b'dict' + b'\x00' * 6) unknown_message = data.UnknownMessage( next=TransType.UDP, length=2, @@ -598,7 +602,10 @@ def make_extension(code: CGAExtension, extension: data.UnknownExtension | None = chksum=b'', data=b'model', ) - self.assertEqual(proto.make(type=custom_packet, data=unknown_message).data, b'model') + # likewise raw bytes out of the constructor: 6 + 5 octets is 5 short of + # 16, so the body is padded rather than the header misdeclared + self.assertEqual(proto.make(type=custom_packet, data=unknown_message).data, + b'model' + b'\x00' * 5) parsed_options = proto._read_mh_options([ SimpleNamespace(type=custom_option, data=b'opt'), @@ -666,10 +673,17 @@ def make_extension(code: CGAExtension, extension: data.UnknownExtension | None = finally: schema.CGAParameter.public_key_test = public_key_test - self.assertEqual(proto._make_opt_pad( + # NOTE: The data model's ``length`` is the whole option, the schema's is + # its ``Option Length`` field. A 4-octet ``PadN`` spends 2 octets on its + # type and length fields, so ``Option Length`` is 2 and the schema packs + # back to the 4 octets the data model described. Asserting 4 here was + # asserting the two-octet overshoot. + remade = proto._make_opt_pad( Option.Pad1, data.PadOption(type=Option.PadN, length=4), - ).length, 4) + ) + self.assertEqual(remade.length, 2) + self.assertEqual(len(remade.pack()), 4) self.assertEqual(str(proto._make_opt_aca( Option.Alternate_Care_of_Address, data.AlternateCareofAddressOption(type=Option.Alternate_Care_of_Address, @@ -1172,10 +1186,12 @@ def test_mh_fmipv6_round_trip_is_byte_identical(self) -> None: mhlla_type = Option.Mobility_Header_Link_Layer_Address_option exp_type = Option.Experimental_Mobility_Option - # NOTE: every case below is chosen to be 8-octet aligned without a padding - # option, since ``_make_opt_pad`` round-trips a ``PadN`` data model two bytes - # too long -- a pre-existing defect of the RFC 6275 padding option, outside - # the FMIPv6 message and option types under test here. + # NOTE: Every case below is 8-octet aligned and so draws no padding option, + # which keeps these cases about the FMIPv6 message and option types. It used + # to be a workaround as well: ``_make_opt_pad`` round-tripped a ``PadN`` data + # model two octets too long, because it copied the data model's whole-option + # ``length`` into the schema's ``Option Length`` field. That is fixed, and + # ``test_mh_padding_option_data_model_round_trips`` covers it. cases = [ ('FBU', Packet.Fast_Binding_Update, { 'seq': 0x1234, 'ack': True, 'home': True, 'lla_compat': False, @@ -1434,6 +1450,197 @@ def test_mh_message_flags_pack_each_bit_independently(self) -> None: 'options': []})) self.assertEqual(raw[8], octet) + def test_mh_padding_options_parse_from_the_wire(self) -> None: + """A ``Pad1`` option must consume exactly one octet, wherever it sits. + + Per :rfc:`6275#section-6.2.5` the two padding options do not share a wire + shape: ``Pad1`` is a lone type octet with no ``Option Length`` field at + all, while ``PadN`` is a type octet, a length octet, and that many data + octets. Reading a length octet that is not there consumes the *next* + option's type byte instead, so every case below puts something after the + padding. + + The header is a Binding Refresh Request with ``Header Len`` 1, i.e. 16 + octets: 6 of fixed header, 2 of the message's own reserved field, and 8 + of mobility options. + """ + from pcapkit.const.mh.option import Option + from pcapkit.const.mh.packet import Packet + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.mh import MH + + header = b'\x3b\x01\x00\x00\x00\x00' + b'\x00\x00' + cases = ( + ('pad1 then padn', b'\x00' + b'\x01\x05\x00\x00\x00\x00\x00', + [(Option.Pad1, 1), (Option.PadN, 7)]), + ('pad1 then a real option then pad1', b'\x00' + b'\x02\x02\x00\x0a' + b'\x00\x00\x00', + [(Option.Pad1, 1), (Option.Binding_Refresh_Advice, 4), + (Option.Pad1, 1), (Option.Pad1, 1), (Option.Pad1, 1)]), + ('eight pad1 in a row', b'\x00' * 8, [(Option.Pad1, 1)] * 8), + ('pad1 as the final octet', b'\x01\x05\x00\x00\x00\x00\x00' + b'\x00', + [(Option.PadN, 7), (Option.Pad1, 1)]), + ('padn is unaffected', b'\x01\x06\x00\x00\x00\x00\x00\x00', + [(Option.PadN, 8)]), + ) + + for name, options, expected in cases: + with self.subTest(case=name): + raw = header + options + self.assertEqual(len(raw), 16) + + info = MH(raw, extension=True).info + self.assertEqual(info.length, 16) + self.assertEqual( + [(code, opt.length) for code, opt in info.options.items(multi=True)], + expected, + ) + self.assertEqual(bytes(MH(next=info.next, type=info.type, + chksum=info.chksum, data=info)), raw) + + def test_mh_padding_option_schema_sizes_itself(self) -> None: + """The padding option schema on its own, and the helper behind it. + + :func:`~pcapkit.protocols.schema.internet.mh.pad_opt_data_len` has to read + a skipped conditional field -- which is recorded as + :data:`~pcapkit.corekit.fields.field.NoValue`, not omitted -- as zero + padding octets, rather than handing that singleton to + :class:`~pcapkit.corekit.fields.strings.PaddingField` where it becomes an + unusable :mod:`struct` template. + """ + from pcapkit.const.mh.option import Option + from pcapkit.corekit.fields.field import NoValue + from pcapkit.protocols.schema.internet import mh as schema + + self.assertEqual(schema.pad_opt_data_len({}), 0) + self.assertEqual(schema.pad_opt_data_len({'length': NoValue}), 0) + self.assertEqual(schema.pad_opt_data_len({'length': None}), 0) + self.assertEqual(schema.pad_opt_data_len({'length': 4}), 4) + + # a Pad1 option on its own: one octet read, the next left for whatever follows + stream = io.BytesIO(b'\x00\x2a') + pad1 = schema.PadOption.unpack(stream, 2, {}) + self.assertEqual(pad1.type, Option.Pad1) + self.assertEqual(pad1.length, 0) + self.assertEqual(len(pad1), 1) + self.assertEqual(bytes(pad1), b'\x00') + self.assertEqual(stream.tell(), 1) + + # a PadN option still spans its length octet plus that many data octets + stream = io.BytesIO(b'\x01\x02\xaa\xbb\xcc') + padn = schema.PadOption.unpack(stream, 5, {}) + self.assertEqual(padn.type, Option.PadN) + self.assertEqual(padn.length, 2) + self.assertEqual(len(padn), 4) + self.assertEqual(bytes(padn), b'\x01\x02\xaa\xbb') + self.assertEqual(stream.tell(), 4) + + def test_mh_padding_option_data_model_round_trips(self) -> None: + """Re-making a parsed padding option must give back the same octets. + + The data model's ``length`` counts the whole option while the schema's is + the ``Option Length`` field, so copying one into the other unconverted + produced a ``PadN`` two octets longer than the one that was parsed. + """ + from pcapkit.const.mh.option import Option + from pcapkit.protocols.data.internet.mh import PadOption + from pcapkit.protocols.internet.mh import MH + + proto = object.__new__(MH) + for code, total in ((Option.Pad1, 1), (Option.PadN, 3), (Option.PadN, 6)): + with self.subTest(option=code.name, length=total): + made = proto._make_opt_pad(code, PadOption(type=code, length=total)) + self.assertEqual(made.type, code) + self.assertEqual(len(made.pack()), total) + + def test_mh_constructed_header_round_trips(self) -> None: + """Construction must declare the number of octets it actually emits. + + The Mobility Header is a multiple of 8 octets and ``Header Len`` counts + those units less one [:rfc:`6275#section-6.1.1`]. Its fixed part is six + octets, so it is ``len(message data) + 6`` that has to be aligned -- a + different modulus from the 2-octet fixed part of an IPv6 options header. + Rounding ``Header Len`` up without emitting the padding to match sent the + re-parse off the end of the buffer. + """ + from pcapkit.const.mh.option import Option + from pcapkit.const.mh.packet import Packet + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.mh import MH + + brr = Packet.Binding_Refresh_Request + cases = ( + # 2 octets of reserved + a 6-octet PadN = 8, so 6 + 8 = 14 needs 2 more + ('padn then two pad1', [(Option.PadN, {'length': 4})], 16), + # 2 + 8 = 10, and 6 + 10 = 16 is already aligned + ('padn, already aligned', [(Option.PadN, {'length': 6})], 16), + # 2 + 4 = 6, and 6 + 6 = 12 needs 4 more octets + ('real option then padn', [(Option.Binding_Refresh_Advice, {'interval': 10})], 16), + # no options at all: 6 + 2 = 8 is aligned, so MH needs no padding here + # where an IPv6 options header would need 6 octets of it + ('no options', [], 8), + ) + + for name, options, size in cases: + with self.subTest(case=name): + built = MH(next=TransType.IPv6_NoNxt, type=brr, chksum=b'\x00\x00', + data={'options': list(options)}) + raw = bytes(built) + + self.assertEqual(len(raw), size) + self.assertEqual(len(raw) % 8, 0) + # the declared length and the emitted length must be the same + self.assertEqual((raw[1] + 1) * 8, len(raw)) + + parsed = MH(raw, extension=True).info + self.assertEqual(parsed.length, size) + self.assertEqual(bytes(MH(next=parsed.next, type=parsed.type, + chksum=parsed.chksum, data=parsed)), raw) + + def test_mh_opaque_message_body_cannot_be_padded(self) -> None: + """An opaque message body is padded to alignment, with a warning. + + A message whose body is raw bytes carries no mobility options, so the + padding goes into the body itself. Warning and returning it unchanged was + not enough: ``length`` is ``(len(data) + 6) // 8 - 1``, which floors, so the + header shipped declaring 8 octets while emitting 10, 12 or 14, and a parser + reads 8 and misinterprets the rest. + + """ + from pcapkit.const.mh.packet import Packet + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.mh import MH + from pcapkit.utilities.warnings import ProtocolWarning + + proto = object.__new__(MH) + + # each of these previously emitted 6 + n octets while declaring only 8 + for n in (4, 6, 8): + with self.subTest(body=n): + with mock.patch('pcapkit.protocols.internet.mh.warn') as warned: + padded = proto._pad_mh_message(b'\x00' * n) + self.assertEqual(warned.call_count, 1) + self.assertIs(warned.call_args.args[1], ProtocolWarning) + + # the emitted header is now exactly what it declares + total = 6 + len(padded) + self.assertEqual(total % 8, 0) + self.assertEqual(((len(padded) + 6) // 8) * 8, total) + + # an already-aligned body needs no padding, is returned untouched, and says + # nothing -- the warning is only for a body the caller will not get back + # byte-for-byte + for n in (2, 10): + with self.subTest(body=n): + body = b'\x00' * n + with mock.patch('pcapkit.protocols.internet.mh.warn') as warned: + self.assertIs(proto._pad_mh_message(body), body) + warned.assert_not_called() + + # and the whole header still comes out 8-octet aligned when it can be padded + raw = bytes(MH(next=TransType.IPv6_NoNxt, type=Packet.Binding_Refresh_Request, + chksum=b'\x00\x00', data={'options': []})) + self.assertEqual(len(raw) % 8, 0) + if __name__ == '__main__': unittest.main()