diff --git a/pcapkit/protocols/schema/schema.py b/pcapkit/protocols/schema/schema.py index e323289b97..a54f1ff789 100644 --- a/pcapkit/protocols/schema/schema.py +++ b/pcapkit/protocols/schema/schema.py @@ -979,7 +979,13 @@ def unpack(cls, data: 'bytes | IO[bytes]', # ``len(self)`` -- :class:`~pcapkit.corekit.fields.collections.OptionField` # and :class:`~pcapkit.corekit.fields.collections.ListField` both # do this -- see the octets actually consumed. - data.seek(-length, io.SEEK_CUR) + # + # The rewind is by the octets actually read, not by ``length``: + # near the end of the stream the read comes back short, and + # rewinding the full width would move the stream *before* the + # forward match, so the fields after it would re-read the + # preceding octet as their first. + data.seek(-len(byte), io.SEEK_CUR) self.__buffer__[field.name] = b'' elif isinstance(field, OptionField) and field.option_padding > 0: # the option list ended before the declared field length was diff --git a/tests/protocols/internet/test_smf_dpd_truncated_len_unit.py b/tests/protocols/internet/test_smf_dpd_truncated_len_unit.py new file mode 100644 index 0000000000..25cb96c53a --- /dev/null +++ b/tests/protocols/internet/test_smf_dpd_truncated_len_unit.py @@ -0,0 +1,80 @@ +# -*- coding: utf-8 -*- +"""An option in the last two octets of a HOPOPT or IPv6-Opts header fails in-library. + +GitHub issue #1195: an ``SMF_DPD`` option of ``Opt Data Len`` zero, placed in +the last two octets of the header (``3b00010200000800``), raised a bare +:exc:`TypeError` from +:func:`~pcapkit.protocols.schema.internet.hopopt.smf_i_dpd_id_len`. The option +starts with a :class:`~pcapkit.corekit.fields.misc.ForwardMatchField` three +octets wide, which came back one octet short at the end of the stream, yet +:meth:`Schema.unpack ` rewound +the full three. The option was therefore re-read from the preceding ``PadN`` +padding octet, as a ``Pad1`` option, so ``len`` was skipped and left as +:data:`~pcapkit.corekit.fields.field.NO_VALUE`. + +Every case builds its own octets in memory and reads no capture. The classes +are imported inside each test, after +:func:`~tests._support.reimport_once_per_class`, so that they belong to the +:mod:`pcapkit` import that is live when the test runs. + +""" + +import importlib +import io +import unittest +import warnings + +from tests._support import reimport_once_per_class + +#: Next Header ``No Next Header``, ``Hdr Ext Len`` zero, then a ``PadN`` option +#: carrying two octets, leaving the last two octets of the header for the option +#: under test. +PREFIX = bytes.fromhex('3b00' '01020000') + +#: The protocol classes sharing the option schema, by module and class name. +PROTOCOLS = ( + ('pcapkit.protocols.internet.hopopt', 'HOPOPT'), + ('pcapkit.protocols.internet.ipv6_opts', 'IPv6_Opts'), +) + +#: ``RPL_Option_0x23`` fails here for an unrelated reason, tracked in #1196. +EXCLUDED = {0x23} + + +class TestSMFDPDTruncatedLength(unittest.TestCase): + """Pin the error an option in the last two octets of the header raises.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _protocols(self) -> 'list[type]': + return [getattr(importlib.import_module(mod), name) for mod, name in PROTOCOLS] + + def test_smf_dpd_raises_field_value_error(self) -> None: + from pcapkit.utilities.exceptions import FieldValueError + + for cls in self._protocols(): + with self.subTest(protocol=cls.__name__): + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + with self.assertRaisesRegex(FieldValueError, 'invalid SMF I-DPD option length: 0'): + cls(io.BytesIO(PREFIX + bytes.fromhex('0800'))) + + def test_no_option_type_raises_a_non_library_error(self) -> None: + from pcapkit.utilities.exceptions import BaseError + + for cls in self._protocols(): + for code in range(256): + if code in EXCLUDED: + continue + with self.subTest(protocol=cls.__name__, type=f'0x{code:02x}'): + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + try: + cls(io.BytesIO(PREFIX + bytes([code, 0]))) + except BaseError: + pass + + +if __name__ == '__main__': + unittest.main()