diff --git a/pcapkit/protocols/schema/internet/hopopt.py b/pcapkit/protocols/schema/internet/hopopt.py index ee822d9c4..baef4a051 100644 --- a/pcapkit/protocols/schema/internet/hopopt.py +++ b/pcapkit/protocols/schema/internet/hopopt.py @@ -574,7 +574,7 @@ class MPLOption(Option, code=Enum_Option.MPL_Option): #: MPL sequence number. seq: 'int' = UInt8Field() #: MPL Seed-ID. - seed: 'int' = ConditionalField( + seed: 'int | IPv6Address' = ConditionalField( NumberField(length=mpl_opt_seed_id_len, signed=False), lambda pkt: pkt['flags']['type'] != Enum_SeedID.IPV6_SOURCE_ADDRESS, ) @@ -597,7 +597,7 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': if TYPE_CHECKING: def __init__(self, type: 'Enum_Option', len: 'int', flags: 'MPLFlags', seq: 'int', - seed: 'Optional[int]') -> 'None': ... + seed: 'Optional[int | IPv6Address]') -> 'None': ... @schema_final diff --git a/pcapkit/protocols/schema/internet/ipv6_opts.py b/pcapkit/protocols/schema/internet/ipv6_opts.py index b62e6b088..ebe33da85 100644 --- a/pcapkit/protocols/schema/internet/ipv6_opts.py +++ b/pcapkit/protocols/schema/internet/ipv6_opts.py @@ -586,7 +586,7 @@ class MPLOption(Option, code=Enum_Option.MPL_Option): #: MPL sequence number. seq: 'int' = UInt8Field() #: MPL Seed-ID. - seed: 'int' = ConditionalField( + seed: 'int | IPv6Address' = ConditionalField( NumberField(length=mpl_opt_seed_id_len, signed=False), lambda pkt: pkt['flags']['type'] != Enum_SeedID.IPV6_SOURCE_ADDRESS, ) @@ -609,7 +609,7 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': if TYPE_CHECKING: def __init__(self, type: 'Enum_Option', len: 'int', flags: 'MPLFlags', seq: 'int', - seed: 'Optional[int]') -> 'None': ... + seed: 'Optional[int | IPv6Address]') -> 'None': ... @schema_final diff --git a/tests/protocols/internet/test_mpl_seed_annotation_1840_unit.py b/tests/protocols/internet/test_mpl_seed_annotation_1840_unit.py new file mode 100644 index 000000000..14692b696 --- /dev/null +++ b/tests/protocols/internet/test_mpl_seed_annotation_1840_unit.py @@ -0,0 +1,111 @@ +# -*- coding: utf-8 -*- +"""The MPL option schemas annotate ``seed`` as the values it actually holds. + +GitHub issue #1840: ``MPLOption.seed`` in the HOPOPT and IPv6-Opts schemas was +annotated ``'int'``, and its ``__init__`` stub's ``seed`` ``'Optional[int]'``. +When ``S`` is 0 (:rfc:`7731#section-6.1`) the seed is the IPv6 source address, +which ``post_process`` stores as an :class:`~ipaddress.IPv6Address`; the data +models already say ``Optional[int | IPv6Address]``. + +The annotations are string literals naming ``TYPE_CHECKING``-only imports, so +they are checked by the names they use rather than evaluated. The ``__init__`` +stub exists only under ``TYPE_CHECKING``, so it is read from the module source. +The runtime cases build their octets in memory and read no capture, as in +:mod:`tests.protocols.internet.test_mpl_seed_rebuild_unit`. + +""" + +import ast +import ipaddress +import pathlib +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, so only the ``S`` field sets their width. +HEADERS = { + 0: bytes.fromhex('3b006d0220070000'), + 1: bytes.fromhex('3b006d046007beef'), + 2: bytes.fromhex('3b016d0aa007' '0000000000000005' '0000'), + 3: bytes.fromhex('3b026d12e007' '00000000000000000000000000000005' '0000'), +} + +#: Integer seed each non-address kind carries in :data:`HEADERS`. +INT_SEEDS = {1: 0xBEEF, 2: 5, 3: 5} + +SRC = ipaddress.IPv6Address('2001:db8::1') +DST = ipaddress.IPv6Address('2001:db8::2') + + +def _names(annotation: 'str') -> 'set[str]': + """Return every name a string annotation uses.""" + tree = ast.parse(annotation, mode='eval') + return {node.id for node in ast.walk(tree) if isinstance(node, ast.Name)} + + +def _schemas() -> 'dict[str, tuple[object, int]]': + from pcapkit.const.ipv6.extension_header import ExtensionHeader + from pcapkit.protocols.schema.internet import hopopt, ipv6_opts + return { + 'HOPOPT': (hopopt, ExtensionHeader.HOPOPT), + 'IPv6_Opts': (ipv6_opts, ExtensionHeader.IPv6_Opts), + } + + +def _init_stub_seed(module: 'object') -> 'str': + """Return the ``seed`` annotation of ``MPLOption.__init__`` in ``module``'s source.""" + tree = ast.parse(pathlib.Path(module.__file__).read_text(encoding='utf-8')) # type: ignore[attr-defined] + for cls in tree.body: + if isinstance(cls, ast.ClassDef) and cls.name == 'MPLOption': + for node in ast.walk(cls): + if isinstance(node, ast.FunctionDef) and node.name == '__init__': + for arg in node.args.args: + if arg.arg == 'seed': + return ast.literal_eval(arg.annotation) # type: ignore[arg-type] + raise AssertionError('MPLOption.__init__ stub has no seed parameter') + + +class TestMPLSeedAnnotation(unittest.TestCase): + """Pin the ``seed`` annotations and the runtime types they describe.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_seed_field_annotation_names_ipv6address(self) -> None: + for name, (module, _) in _schemas().items(): + with self.subTest(proto=name): + annotation = module.MPLOption.__annotations__['seed'] # type: ignore[attr-defined] + self.assertEqual(_names(annotation), {'int', 'IPv6Address'}) + + def test_init_stub_seed_annotation_names_ipv6address(self) -> None: + for name, (module, _) in _schemas().items(): + with self.subTest(proto=name): + self.assertEqual(_names(_init_stub_seed(module)), {'Optional', 'int', 'IPv6Address'}) + + def test_seed_types_inside_ipv6(self) -> None: + from pcapkit.protocols.internet.ipv6 import IPv6 + + for name, (module, code) in _schemas().items(): + for kind, octets in HEADERS.items(): + with self.subTest(proto=name, S=kind): + packet = (struct.pack('!IHBB', 6 << 28, len(octets), int(code), 64) + + SRC.packed + DST.packed + octets) + exthdrs = IPv6(packet, len(packet)).extension_headers + self.assertEqual(list(exthdrs.keys()), [code]) + options = list(exthdrs.values())[0].schema.options + mpl = [opt for opt in options if isinstance(opt, module.MPLOption)] # type: ignore[attr-defined] + self.assertEqual(len(mpl), 1) + seed = mpl[0].seed + if kind == 0: + self.assertIsInstance(seed, ipaddress.IPv6Address) + self.assertEqual(seed, SRC) + else: + self.assertIs(type(seed), int) + self.assertEqual(seed, INT_SEEDS[kind]) + + +if __name__ == '__main__': + unittest.main()