Skip to content

fix(ipv6-opts): _make_opt_smf_dpd overwrites the parsed DPD mode, so H-DPD rebuilds as I-DPD #1192

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JarryShaw merged 1 commit into from
Oct 7, 2026
Merged

fix(ipv6-opts): _make_opt_smf_dpd overwrites the parsed DPD mode, so H-DPD rebuilds as I-DPD #1192

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
119 changes: 59 additions & 60 deletions pcapkit/protocols/internet/hopopt.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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'',
Expand All @@ -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)
Expand All @@ -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
Expand Down
119 changes: 59 additions & 60 deletions pcapkit/protocols/internet/ipv6_opts.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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'',
Expand All @@ -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)
Expand All @@ -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
Expand Down
91 changes: 91 additions & 0 deletions tests/protocols/internet/test_smf_dpd_mode_rebuild_unit.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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()
Loading