From 45d815f7047c32c27ee75bf51d6b8745d2c03447 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Wed, 7 Oct 2026 10:15:19 -0400 Subject: [PATCH] fix(ipv6-opts): _make_opt_smf_dpd overwrites the parsed DPD mode, so H-DPD rebuilds as I-DPD The parsed DPD mode is now fed through the mode keyword rather than overwritten by it, so an H-DPD option rebuilds as H-DPD. The parsed TaggerID type wins the same way, via a new tid_type keyword, so a 4- or 16-octet DEFAULT Tagger ID no longer rebuilds as IPv4 or IPv6. A Tagger ID that does not fit its type, or is over 16 octets, raises ProtocolError. Same change in HOPOPT and IPv6-Opts. Closes #1190 --- pcapkit/protocols/internet/hopopt.py | 119 +++++++++--------- pcapkit/protocols/internet/ipv6_opts.py | 119 +++++++++--------- .../test_smf_dpd_mode_rebuild_unit.py | 91 ++++++++++++++ 3 files changed, 209 insertions(+), 120 deletions(-) create mode 100644 tests/protocols/internet/test_smf_dpd_mode_rebuild_unit.py diff --git a/pcapkit/protocols/internet/hopopt.py b/pcapkit/protocols/internet/hopopt.py index 9ed3ba561c..1ef8bae70a 100644 --- a/pcapkit/protocols/internet/hopopt.py +++ b/pcapkit/protocols/internet/hopopt.py @@ -1580,6 +1580,10 @@ def _make_opt_smf_dpd(self, code: 'Enum_Option', opt: 'Optional[Data_SMFIdentifi mode_default: 'Optional[int]' = None, mode_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long mode_reversed: 'bool' = False, + tid_type: 'Optional[Enum_TaggerID | StdlibEnum | AenumEnum | str | int]' = None, + tid_type_default: 'Optional[int]' = None, + tid_type_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long + tid_type_reversed: 'bool' = False, tid: 'Optional[bytes | IPv4Address | IPv6Address]' = None, id: 'bytes' = b'', hav: 'bytes' = b'', @@ -1593,6 +1597,11 @@ def _make_opt_smf_dpd(self, code: 'Enum_Option', opt: 'Optional[Data_SMFIdentifi mode_default: default value of DPD mode mode_namespace: namespace of DPD mode mode_reversed: reversed flag of DPD mode + tid_type: TaggerID type (``TidTy`` field); inferred from the + length of ``tid`` if not given + tid_type_default: default value of TaggerID type + tid_type_namespace: namespace of TaggerID type + tid_type_reversed: reversed flag of TaggerID type tid: Tagger ID id: identifier hav: hash assist value (HAV) @@ -1601,78 +1610,68 @@ def _make_opt_smf_dpd(self, code: 'Enum_Option', opt: 'Optional[Data_SMFIdentifi Returns: Constructed option schema. + Raises: + ProtocolError: If the DPD mode is invalid, or the Tagger ID does + not fit the TaggerID type. + """ if opt is not None: - dpd_type = opt.dpd_type + mode = opt.dpd_type + tid_type = getattr(opt, 'tid_type', None) tid = getattr(opt, 'tid', None) id = getattr(opt, 'id', b'') hav = getattr(opt, 'hav', b'') - dpd_type = self._make_index(mode, mode_default, namespace=mode_namespace, # type: ignore[assignment] + dpd_type = self._make_index(mode, mode_default, namespace=mode_namespace, reversed=mode_reversed, pack=False) if dpd_type == Enum_SMFDPDMode.I_DPD: if tid is None: - schema = Schema_SMFIdentificationBasedDPDOption( - type=code, - len=1 + len(id), - info={ - 'mode': 0, - 'type': Enum_TaggerID.NULL, - 'len': 0, - }, - tid=None, - id=id, - ) # type: Schema_SMFDPDOption + tid_bytes = b'' elif isinstance(tid, bytes): - tid_len = len(tid) - if tid_len == 0: - tid_type = Enum_TaggerID.NULL - else: - try: - tid_ip_ver = ipaddress.ip_address(tid).version - if tid_ip_ver == 4: - tid_type = Enum_TaggerID.IPv4 - elif tid_ip_ver == 6: - tid_type = Enum_TaggerID.IPv6 - else: - tid_type = Enum_TaggerID.DEFAULT # type: ignore[unreachable] - except ValueError: - tid_type = Enum_TaggerID.DEFAULT - - schema = Schema_SMFIdentificationBasedDPDOption( - type=code, - len=1 + tid_len + len(id), - info={ - 'mode': 0, - 'type': tid_type, - 'len': tid_len - 1, - }, - tid=tid, - id=id, - ) + tid_bytes = tid + elif tid.version in (4, 6): + tid_bytes = tid.packed else: - tid_ver = tid.version - if tid_ver == 4: - tid_type = Enum_TaggerID.IPv4 - tid_len = 4 - elif tid_ver == 6: - tid_type = Enum_TaggerID.IPv6 - tid_len = 16 - else: - raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid TaggerID version: {tid_ver}') - - schema = Schema_SMFIdentificationBasedDPDOption( - type=code, - len=1 + tid_len + len(id), - info={ - 'mode': 0, - 'type': tid_type, - 'len': tid_len - 1, - }, - tid=tid.packed, - id=id, - ) + raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid TaggerID version: {tid.version}') + tid_len = len(tid_bytes) + + if tid_type is not None: + tid_kind = self._make_index(tid_type, tid_type_default, namespace=tid_type_namespace, + reversed=tid_type_reversed, pack=False) + elif tid_len == 0: + tid_kind = Enum_TaggerID.NULL + elif tid_len == 4: + tid_kind = Enum_TaggerID.IPv4 + elif tid_len == 16: + tid_kind = Enum_TaggerID.IPv6 + else: + tid_kind = Enum_TaggerID.DEFAULT + + if tid_kind == Enum_TaggerID.NULL: + tid_ok = tid_len == 0 + elif tid_kind == Enum_TaggerID.IPv4: + tid_ok = tid_len == 4 + elif tid_kind == Enum_TaggerID.IPv6: + tid_ok = tid_len == 16 + else: + # TidLen is 4 bits and holds the length minus one. + tid_ok = 1 <= tid_len <= 16 + if not tid_ok: + raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid TaggerID length {tid_len} ' + f'for type {tid_kind}') + + schema = Schema_SMFIdentificationBasedDPDOption( + type=code, + len=1 + tid_len + len(id), + info={ + 'mode': 0, + 'type': tid_kind, + 'len': max(tid_len - 1, 0), + }, + tid=tid_bytes if tid_len else None, + id=id, + ) # type: Schema_SMFDPDOption elif dpd_type == Enum_SMFDPDMode.H_DPD: hav_ba = bytearray(hav) hav_ba[0] = hav[0] | 0x80 diff --git a/pcapkit/protocols/internet/ipv6_opts.py b/pcapkit/protocols/internet/ipv6_opts.py index 7cb66b5515..22c53e8961 100644 --- a/pcapkit/protocols/internet/ipv6_opts.py +++ b/pcapkit/protocols/internet/ipv6_opts.py @@ -1569,6 +1569,10 @@ def _make_opt_smf_dpd(self, code: 'Enum_Option', opt: 'Optional[Data_SMFIdentifi mode_default: 'Optional[int]' = None, mode_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long mode_reversed: 'bool' = False, + tid_type: 'Optional[Enum_TaggerID | StdlibEnum | AenumEnum | str | int]' = None, + tid_type_default: 'Optional[int]' = None, + tid_type_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long + tid_type_reversed: 'bool' = False, tid: 'Optional[bytes | IPv4Address | IPv6Address]' = None, id: 'bytes' = b'', hav: 'bytes' = b'', @@ -1582,6 +1586,11 @@ def _make_opt_smf_dpd(self, code: 'Enum_Option', opt: 'Optional[Data_SMFIdentifi mode_default: default value of DPD mode mode_namespace: namespace of DPD mode mode_reversed: reversed flag of DPD mode + tid_type: TaggerID type (``TidTy`` field); inferred from the + length of ``tid`` if not given + tid_type_default: default value of TaggerID type + tid_type_namespace: namespace of TaggerID type + tid_type_reversed: reversed flag of TaggerID type tid: Tagger ID id: identifier hav: hash assist value (HAV) @@ -1590,78 +1599,68 @@ def _make_opt_smf_dpd(self, code: 'Enum_Option', opt: 'Optional[Data_SMFIdentifi Returns: Constructed option schema. + Raises: + ProtocolError: If the DPD mode is invalid, or the Tagger ID does + not fit the TaggerID type. + """ if opt is not None: - dpd_type = opt.dpd_type + mode = opt.dpd_type + tid_type = getattr(opt, 'tid_type', None) tid = getattr(opt, 'tid', None) id = getattr(opt, 'id', b'') hav = getattr(opt, 'hav', b'') - dpd_type = self._make_index(mode, mode_default, namespace=mode_namespace, # type: ignore[assignment] + dpd_type = self._make_index(mode, mode_default, namespace=mode_namespace, reversed=mode_reversed, pack=False) if dpd_type == Enum_SMFDPDMode.I_DPD: if tid is None: - schema = Schema_SMFIdentificationBasedDPDOption( - type=code, - len=1 + len(id), - info={ - 'mode': 0, - 'type': Enum_TaggerID.NULL, - 'len': 0, - }, - tid=None, - id=id, - ) # type: Schema_SMFDPDOption + tid_bytes = b'' elif isinstance(tid, bytes): - tid_len = len(tid) - if tid_len == 0: - tid_type = Enum_TaggerID.NULL - else: - try: - tid_ip_ver = ipaddress.ip_address(tid).version - if tid_ip_ver == 4: - tid_type = Enum_TaggerID.IPv4 - elif tid_ip_ver == 6: - tid_type = Enum_TaggerID.IPv6 - else: - tid_type = Enum_TaggerID.DEFAULT # type: ignore[unreachable] - except ValueError: - tid_type = Enum_TaggerID.DEFAULT - - schema = Schema_SMFIdentificationBasedDPDOption( - type=code, - len=1 + tid_len + len(id), - info={ - 'mode': 0, - 'type': tid_type, - 'len': tid_len - 1, - }, - tid=tid, - id=id, - ) + tid_bytes = tid + elif tid.version in (4, 6): + tid_bytes = tid.packed else: - tid_ver = tid.version - if tid_ver == 4: - tid_type = Enum_TaggerID.IPv4 - tid_len = 4 - elif tid_ver == 6: - tid_type = Enum_TaggerID.IPv6 - tid_len = 16 - else: - raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid TaggerID version: {tid_ver}') - - schema = Schema_SMFIdentificationBasedDPDOption( - type=code, - len=1 + tid_len + len(id), - info={ - 'mode': 0, - 'type': tid_type, - 'len': tid_len - 1, - }, - tid=tid.packed, - id=id, - ) + raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid TaggerID version: {tid.version}') + tid_len = len(tid_bytes) + + if tid_type is not None: + tid_kind = self._make_index(tid_type, tid_type_default, namespace=tid_type_namespace, + reversed=tid_type_reversed, pack=False) + elif tid_len == 0: + tid_kind = Enum_TaggerID.NULL + elif tid_len == 4: + tid_kind = Enum_TaggerID.IPv4 + elif tid_len == 16: + tid_kind = Enum_TaggerID.IPv6 + else: + tid_kind = Enum_TaggerID.DEFAULT + + if tid_kind == Enum_TaggerID.NULL: + tid_ok = tid_len == 0 + elif tid_kind == Enum_TaggerID.IPv4: + tid_ok = tid_len == 4 + elif tid_kind == Enum_TaggerID.IPv6: + tid_ok = tid_len == 16 + else: + # TidLen is 4 bits and holds the length minus one. + tid_ok = 1 <= tid_len <= 16 + if not tid_ok: + raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid TaggerID length {tid_len} ' + f'for type {tid_kind}') + + schema = Schema_SMFIdentificationBasedDPDOption( + type=code, + len=1 + tid_len + len(id), + info={ + 'mode': 0, + 'type': tid_kind, + 'len': max(tid_len - 1, 0), + }, + tid=tid_bytes if tid_len else None, + id=id, + ) # type: Schema_SMFDPDOption elif dpd_type == Enum_SMFDPDMode.H_DPD: hav_ba = bytearray(hav) hav_ba[0] = hav[0] | 0x80 diff --git a/tests/protocols/internet/test_smf_dpd_mode_rebuild_unit.py b/tests/protocols/internet/test_smf_dpd_mode_rebuild_unit.py new file mode 100644 index 0000000000..ccf560d140 --- /dev/null +++ b/tests/protocols/internet/test_smf_dpd_mode_rebuild_unit.py @@ -0,0 +1,91 @@ +# -*- coding: utf-8 -*- +"""HOPOPT and IPv6-Opts rebuild every SMF_DPD option byte for byte. + +GitHub issue #1190: ``_make_opt_smf_dpd`` read the DPD mode from the parsed +option and then overwrote it from the ``mode`` keyword, whose default is +I-DPD, so an H-DPD option rebuilt as I-DPD. The TaggerID type was likewise +inferred from the Tagger ID length, so a 4-octet ``DEFAULT`` Tagger ID rebuilt +as IPv4. + +Every case builds its own octets in memory and reads no capture. The protocol +classes are imported inside each test, after +:func:`~tests._support.reimport_once_per_class`, as in +:mod:`tests.protocols.link.test_ethernet_mac_roundtrip_unit`. + +""" + +import unittest + +from tests._support import reimport_once_per_class + +#: Extension headers carrying one SMF_DPD option (type 0x08), next header 59, +#: padded to 8 octets with Pad1 or PadN. +HEADERS = { + 'H-DPD': bytes.fromhex('3b00080481020304'), + 'I-DPD NULL': bytes.fromhex('3b00080300010200'), + 'I-DPD IPv4': bytes.fromhex('3b0108072301020304aabb0103000000'), + 'I-DPD DEFAULT 4 octets': bytes.fromhex('3b010807130a0b0c0daabb0103000000'), + 'I-DPD DEFAULT 2 octets': bytes.fromhex('3b010805110a0baabb01050000000000'), +} + + +def _classes() -> 'dict[str, type]': + from pcapkit.protocols.internet.hopopt import HOPOPT + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + return {'HOPOPT': HOPOPT, 'IPv6_Opts': IPv6_Opts} + + +class TestSMFDPDRebuild(unittest.TestCase): + """Pin the SMF_DPD option rebuild for each DPD mode and TaggerID type.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_parsed_option_rebuilds_byte_for_byte(self) -> None: + for name, cls in _classes().items(): + for case, octets in HEADERS.items(): + with self.subTest(proto=name, case=case): + parsed = cls(octets, len(octets), extension=True) + self.assertEqual(cls.from_data(parsed.info).data, octets) + + def test_mode_keyword_selects_h_dpd(self) -> None: + from pcapkit.const.ipv6.option import Option + from pcapkit.const.ipv6.smf_dpd_mode import SMFDPDMode + from pcapkit.const.reg.transtype import TransType + + for name, cls in _classes().items(): + with self.subTest(proto=name): + built = cls(options=[(Option.SMF_DPD, {'mode': SMFDPDMode.H_DPD, 'hav': b'\x01\x02\x03\x04'})], + next=TransType.IPv6_NoNxt, payload=b'') + self.assertEqual(built.data, HEADERS['H-DPD']) + + def test_tid_type_keyword_selects_default(self) -> None: + from pcapkit.const.ipv6.option import Option + from pcapkit.const.ipv6.tagger_id import TaggerID + from pcapkit.const.reg.transtype import TransType + + for name, cls in _classes().items(): + with self.subTest(proto=name): + built = cls(options=[(Option.SMF_DPD, {'tid_type': TaggerID.DEFAULT, 'tid': b'\x0a\x0b\x0c\x0d', + 'id': b'\xaa\xbb'})], + next=TransType.IPv6_NoNxt, payload=b'') + self.assertEqual(built.data, HEADERS['I-DPD DEFAULT 4 octets']) + + def test_tagger_id_that_does_not_fit_its_type_is_rejected(self) -> None: + from pcapkit.const.ipv6.option import Option + from pcapkit.const.ipv6.tagger_id import TaggerID + from pcapkit.const.reg.transtype import TransType + from pcapkit.utilities.exceptions import ProtocolError + + for name, cls in _classes().items(): + for kwargs in ({'tid_type': TaggerID.IPv4, 'tid': b'\x0a\x0b'}, + {'tid_type': TaggerID.NULL, 'tid': b'\x0a'}, + {'tid_type': TaggerID.DEFAULT}, + {'tid': bytes(17)}): + with self.subTest(proto=name, kwargs=kwargs): + with self.assertRaises(ProtocolError): + cls(options=[(Option.SMF_DPD, kwargs)], next=TransType.IPv6_NoNxt, payload=b'') + + +if __name__ == '__main__': + unittest.main()