Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
4 changes: 2 additions & 2 deletions pcapkit/protocols/data/internet/hopopt.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
4 changes: 2 additions & 2 deletions pcapkit/protocols/data/internet/ipv6_opts.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
47 changes: 39 additions & 8 deletions pcapkit/protocols/internet/hopopt.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -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, *,
Expand Down
47 changes: 39 additions & 8 deletions pcapkit/protocols/internet/ipv6_opts.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -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, *,
Expand Down
94 changes: 94 additions & 0 deletions tests/protocols/internet/test_mpl_seed_rebuild_unit.py
Original file line number Diff line number Diff line change
@@ -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()
Loading