diff --git a/pcapkit/protocols/data/internet/hopopt.py b/pcapkit/protocols/data/internet/hopopt.py index 69f7247017..1f04dab048 100644 --- a/pcapkit/protocols/data/internet/hopopt.py +++ b/pcapkit/protocols/data/internet/hopopt.py @@ -270,10 +270,10 @@ class MPLOption(Option): #: Sequence number. seq: 'int' #: Seed ID. - seed_id: 'Optional[int]' + seed_id: 'Optional[int | IPv6Address]' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Option', action: 'int', change: 'bool', length: 'int', seed_type: 'int', flags: 'MPLFlags', seq: 'int', seed_id: 'Optional[int]') -> 'None': ... # pylint: disable=super-init-not-called,unused-argument,redefined-builtin,multiple-statements,line-too-long + def __init__(self, type: 'Enum_Option', action: 'int', change: 'bool', length: 'int', seed_type: 'int', flags: 'MPLFlags', seq: 'int', seed_id: 'Optional[int | IPv6Address]') -> 'None': ... # pylint: disable=super-init-not-called,unused-argument,redefined-builtin,multiple-statements,line-too-long @info_final diff --git a/pcapkit/protocols/data/internet/ipv6_opts.py b/pcapkit/protocols/data/internet/ipv6_opts.py index 78fbeb60ca..598ad41795 100644 --- a/pcapkit/protocols/data/internet/ipv6_opts.py +++ b/pcapkit/protocols/data/internet/ipv6_opts.py @@ -282,11 +282,11 @@ class MPLOption(Option): #: Sequence number. seq: 'int' #: Seed ID. - seed_id: 'Optional[int]' + seed_id: 'Optional[int | IPv6Address]' if TYPE_CHECKING: def __init__(self, type: 'Enum_Option', action: 'int', change: 'bool', length: 'int', seed_type: 'int', flags: 'MPLFlags', seq: 'int', - seed_id: 'Optional[int]') -> 'None': ... # pylint: disable=super-init-not-called,unused-argument,redefined-builtin,multiple-statements,line-too-long + seed_id: 'Optional[int | IPv6Address]') -> 'None': ... # pylint: disable=super-init-not-called,unused-argument,redefined-builtin,multiple-statements,line-too-long @info_final diff --git a/pcapkit/protocols/internet/hopopt.py b/pcapkit/protocols/internet/hopopt.py index 3060794bf0..9ed3ba561c 100644 --- a/pcapkit/protocols/internet/hopopt.py +++ b/pcapkit/protocols/internet/hopopt.py @@ -1845,47 +1845,78 @@ def _make_opt_rpl(self, code: 'Enum_Option', opt: 'Optional[Data_RPLOption]' = N ) def _make_opt_mpl(self, code: 'Enum_Option', opt: 'Optional[Data_MPLOption]' = None, *, + seed_type: 'Optional[Enum_SeedID | StdlibEnum | AenumEnum | str | int]' = None, + seed_type_default: 'Optional[int]' = None, + seed_type_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long + seed_type_reversed: 'bool' = False, max: 'bool' = False, drop: 'bool' = False, seq: 'int' = 0, - seed: 'Optional[int]' = None, + seed: 'Optional[int | IPv6Address]' = None, **kwargs: 'Any') -> 'Schema_MPLOption': """Make HOPOPT MPL option. Args: code: option type value opt: option data + seed_type: seed-id type (``S`` field); inferred from ``seed`` if not given + seed_type_default: default value of seed-id type + seed_type_namespace: namespace of seed-id type + seed_type_reversed: reversed flag of seed-id type max: maximum sequence number flag drop: drop packet flag seq: MPL sequence number - seed: MPL seed ID + seed: MPL seed ID, ignored when the seed-id type is + :attr:`~pcapkit.const.ipv6.seed_id.SeedID.IPV6_SOURCE_ADDRESS`, + as the seed is then the IPv6 source address **kwargs: arbitrary keyword arguments Returns: Constructed option schema. + Raises: + ProtocolError: If the seed ID is missing or does not fit the seed-id type. + """ if opt is not None: + seed_type = opt.seed_type max = opt.flags.max drop = opt.flags.drop seq = opt.seq seed = opt.seed_id - if seed is None: + if seed_type is not None: + kind = self._make_index(seed_type, seed_type_default, namespace=seed_type_namespace, + reversed=seed_type_reversed, pack=False) + elif seed is None or isinstance(seed, ipaddress.IPv6Address): kind = Enum_SeedID.IPV6_SOURCE_ADDRESS - clen = 2 else: seed_bl = seed.bit_length() if seed_bl <= 16: kind = Enum_SeedID.SEEDID_16_BIT_UNSIGNED_INTEGER - clen = 4 elif seed_bl <= 64: kind = Enum_SeedID.SEEDID_64_BIT_UNSIGNED_INTEGER - clen = 10 - elif seed_bl <= 128: + else: kind = Enum_SeedID.SEEDID_128_BIT_UNSIGNED_INTEGER + + if kind == Enum_SeedID.IPV6_SOURCE_ADDRESS: + # The seed is the IPv6 source address, which is not on the wire. + seed_id = None # type: Optional[int] + clen = 2 + else: + if kind == Enum_SeedID.SEEDID_16_BIT_UNSIGNED_INTEGER: + clen = 4 + elif kind == Enum_SeedID.SEEDID_64_BIT_UNSIGNED_INTEGER: + clen = 10 + elif kind == Enum_SeedID.SEEDID_128_BIT_UNSIGNED_INTEGER: clen = 18 else: + raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid MPL seed-id type: {kind}') + + if seed is None: + raise ProtocolError(f'{self.alias}: [OptNo {code}] missing MPL seed ID for type {kind}') + seed_id = int(seed) + if seed_id.bit_length() > (clen - 2) * 8: raise ProtocolError(f'{self.alias}: [OptNo {code}] too large MPL seed ID: {seed}') return Schema_MPLOption( @@ -1897,7 +1928,7 @@ def _make_opt_mpl(self, code: 'Enum_Option', opt: 'Optional[Data_MPLOption]' = N 'drop': drop, }, seq=seq, - seed=seed, + seed=seed_id, ) def _make_opt_ilnp(self, code: 'Enum_Option', opt: 'Optional[Data_ILNPOption]' = None, *, diff --git a/pcapkit/protocols/internet/ipv6_opts.py b/pcapkit/protocols/internet/ipv6_opts.py index e841f9ac2a..7cb66b5515 100644 --- a/pcapkit/protocols/internet/ipv6_opts.py +++ b/pcapkit/protocols/internet/ipv6_opts.py @@ -1834,47 +1834,78 @@ def _make_opt_rpl(self, code: 'Enum_Option', opt: 'Optional[Data_RPLOption]' = N ) def _make_opt_mpl(self, code: 'Enum_Option', opt: 'Optional[Data_MPLOption]' = None, *, + seed_type: 'Optional[Enum_SeedID | StdlibEnum | AenumEnum | str | int]' = None, + seed_type_default: 'Optional[int]' = None, + seed_type_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long + seed_type_reversed: 'bool' = False, max: 'bool' = False, drop: 'bool' = False, seq: 'int' = 0, - seed: 'Optional[int]' = None, + seed: 'Optional[int | IPv6Address]' = None, **kwargs: 'Any') -> 'Schema_MPLOption': """Make IPv6-Opts MPL option. Args: code: option type value opt: option data + seed_type: seed-id type (``S`` field); inferred from ``seed`` if not given + seed_type_default: default value of seed-id type + seed_type_namespace: namespace of seed-id type + seed_type_reversed: reversed flag of seed-id type max: maximum sequence number flag drop: drop packet flag seq: MPL sequence number - seed: MPL seed ID + seed: MPL seed ID, ignored when the seed-id type is + :attr:`~pcapkit.const.ipv6.seed_id.SeedID.IPV6_SOURCE_ADDRESS`, + as the seed is then the IPv6 source address **kwargs: arbitrary keyword arguments Returns: Constructed option schema. + Raises: + ProtocolError: If the seed ID is missing or does not fit the seed-id type. + """ if opt is not None: + seed_type = opt.seed_type max = opt.flags.max drop = opt.flags.drop seq = opt.seq seed = opt.seed_id - if seed is None: + if seed_type is not None: + kind = self._make_index(seed_type, seed_type_default, namespace=seed_type_namespace, + reversed=seed_type_reversed, pack=False) + elif seed is None or isinstance(seed, ipaddress.IPv6Address): kind = Enum_SeedID.IPV6_SOURCE_ADDRESS - clen = 2 else: seed_bl = seed.bit_length() if seed_bl <= 16: kind = Enum_SeedID.SEEDID_16_BIT_UNSIGNED_INTEGER - clen = 4 elif seed_bl <= 64: kind = Enum_SeedID.SEEDID_64_BIT_UNSIGNED_INTEGER - clen = 10 - elif seed_bl <= 128: + else: kind = Enum_SeedID.SEEDID_128_BIT_UNSIGNED_INTEGER + + if kind == Enum_SeedID.IPV6_SOURCE_ADDRESS: + # The seed is the IPv6 source address, which is not on the wire. + seed_id = None # type: Optional[int] + clen = 2 + else: + if kind == Enum_SeedID.SEEDID_16_BIT_UNSIGNED_INTEGER: + clen = 4 + elif kind == Enum_SeedID.SEEDID_64_BIT_UNSIGNED_INTEGER: + clen = 10 + elif kind == Enum_SeedID.SEEDID_128_BIT_UNSIGNED_INTEGER: clen = 18 else: + raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid MPL seed-id type: {kind}') + + if seed is None: + raise ProtocolError(f'{self.alias}: [OptNo {code}] missing MPL seed ID for type {kind}') + seed_id = int(seed) + if seed_id.bit_length() > (clen - 2) * 8: raise ProtocolError(f'{self.alias}: [OptNo {code}] too large MPL seed ID: {seed}') return Schema_MPLOption( @@ -1886,7 +1917,7 @@ def _make_opt_mpl(self, code: 'Enum_Option', opt: 'Optional[Data_MPLOption]' = N 'drop': drop, }, seq=seq, - seed=seed, + seed=seed_id, ) def _make_opt_ilnp(self, code: 'Enum_Option', opt: 'Optional[Data_ILNPOption]' = None, *, diff --git a/tests/protocols/internet/test_mpl_seed_rebuild_unit.py b/tests/protocols/internet/test_mpl_seed_rebuild_unit.py new file mode 100644 index 0000000000..a7d6c1c48f --- /dev/null +++ b/tests/protocols/internet/test_mpl_seed_rebuild_unit.py @@ -0,0 +1,94 @@ +# -*- coding: utf-8 -*- +"""HOPOPT and IPv6-Opts rebuild every MPL seed-id kind byte for byte. + +GitHub issue #1185: ``_make_opt_mpl`` inferred the ``S`` field from +``seed.bit_length()``. The reader stores the IPv6 source address as the seed +when ``S`` is 0 (:rfc:`7731#section-6.1`), so rebuilding such an option raised +:exc:`AttributeError`. And a 64- or 128-bit seed with a small value was +rebuilt with a narrower ``S``, changing the octets. + +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 ipaddress +import struct +import unittest + +from tests._support import reimport_once_per_class + +#: Extension headers carrying one MPL option (type 0x6D), next header 59, one +#: per seed-id kind. Flags ``M`` set, sequence 7. The 64- and 128-bit seeds are +#: 5, which fits in 16 bits, so the ``S`` field alone sets their width. +HEADERS = { + 0: bytes.fromhex('3b006d0220070000'), + 1: bytes.fromhex('3b006d046007beef'), + 2: bytes.fromhex('3b016d0aa007' '0000000000000005' '0000'), + 3: bytes.fromhex('3b026d12e007' '00000000000000000000000000000005' '0000'), +} + +SRC = ipaddress.IPv6Address('2001:db8::1') +DST = ipaddress.IPv6Address('2001:db8::2') + + +def _classes() -> 'dict[str, tuple[type, int, str]]': + from pcapkit.protocols.internet.hopopt import HOPOPT + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + return {'HOPOPT': (HOPOPT, 0, 'hopopt'), 'IPv6_Opts': (IPv6_Opts, 60, 'opts')} + + +class TestMPLSeedRebuild(unittest.TestCase): + """Pin the MPL option rebuild for each ``S`` value.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_standalone_header_rebuilds_byte_for_byte(self) -> None: + for name, (cls, _, _) in _classes().items(): + for kind, octets in HEADERS.items(): + with self.subTest(proto=name, S=kind): + parsed = cls(octets, len(octets), extension=True) + opt = list(parsed.info.options.values())[0] + self.assertEqual(opt.seed_type, kind) + self.assertEqual(cls.from_data(parsed.info).data, octets) + + def test_source_address_seed_rebuilds_inside_ipv6(self) -> None: + from pcapkit.protocols.internet.ipv6 import IPv6 + + for name, (cls, nh, key) in _classes().items(): + with self.subTest(proto=name): + octets = HEADERS[0] + packet = (struct.pack('!IHBB', 6 << 28, len(octets), nh, 64) + + SRC.packed + DST.packed + octets) + info = IPv6(packet, len(packet)).info[key] + opt = list(info.options.values())[0] + self.assertEqual(opt.seed_id, SRC) + self.assertEqual(cls.from_data(info).data, octets) + + def test_address_seed_keyword_selects_source_address_kind(self) -> None: + from pcapkit.const.ipv6.option import Option + from pcapkit.const.reg.transtype import TransType + + for name, (cls, _, _) in _classes().items(): + with self.subTest(proto=name): + built = cls(options=[(Option.MPL_Option, {'seed': SRC, 'seq': 7, 'max': True})], + next=TransType.IPv6_NoNxt, payload=b'') + self.assertEqual(built.data, HEADERS[0]) + + def test_seed_that_does_not_fit_its_kind_is_rejected(self) -> None: + from pcapkit.const.ipv6.option import Option + from pcapkit.const.reg.transtype import TransType + from pcapkit.utilities.exceptions import ProtocolError + + for name, (cls, _, _) in _classes().items(): + for kwargs in ({'seed_type': 1, 'seed': 0x10000}, {'seed_type': 2}, {'seed': 1 << 128}): + with self.subTest(proto=name, kwargs=kwargs): + with self.assertRaises(ProtocolError): + cls(options=[(Option.MPL_Option, kwargs)], next=TransType.IPv6_NoNxt, payload=b'') + + +if __name__ == '__main__': + unittest.main()