From b90b10cb7b2969d49b66b3a96d541edaed6ea977 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Wed, 7 Oct 2026 13:18:11 -0400 Subject: [PATCH] fix(ipv6-opts): RPL option drops reserved flag bits on rebuild and rejects RFC 6553 sub-TLVs - Schema: map the five reserved flag bits, and read the octets past the fixed fields as `sub_tlvs`. - Data: add `RPLFlags.reserved` and `RPLOption.sub_tlvs`. - Reader: accept any Opt Data Len >= 4; below 4 still raises ProtocolError. - Maker: new `reserved=0` and `sub_tlvs=b''` keywords, taken from `opt` when given; Opt Data Len is now 4 + len(sub_tlvs). - Both changes apply to HOPOPT and IPv6-Opts, types 0x23 and 0x63. Closes #1199 --- pcapkit/protocols/data/internet/hopopt.py | 8 +- pcapkit/protocols/data/internet/ipv6_opts.py | 8 +- pcapkit/protocols/internet/hopopt.py | 28 ++++- pcapkit/protocols/internet/ipv6_opts.py | 28 ++++- pcapkit/protocols/schema/internet/hopopt.py | 22 +++- .../protocols/schema/internet/ipv6_opts.py | 22 +++- .../internet/test_ipv6_extension_unit.py | 9 +- .../test_rpl_option_reserved_subtlv_unit.py | 107 ++++++++++++++++++ 8 files changed, 216 insertions(+), 16 deletions(-) create mode 100644 tests/protocols/internet/test_rpl_option_reserved_subtlv_unit.py diff --git a/pcapkit/protocols/data/internet/hopopt.py b/pcapkit/protocols/data/internet/hopopt.py index 1f04dab048..a63656b86d 100644 --- a/pcapkit/protocols/data/internet/hopopt.py +++ b/pcapkit/protocols/data/internet/hopopt.py @@ -226,9 +226,11 @@ class RPLFlags(Data): rank_err: 'bool' #: Forwarding error flag. fwd_err: 'bool' + #: Reserved flag bits, carried verbatim so that a rebuild reproduces them. + reserved: 'int' if TYPE_CHECKING: - def __init__(self, down: 'bool', rank_err: 'bool', fwd_err: 'bool') -> 'None': ... # pylint: disable=super-init-not-called,unused-argument,redefined-builtin,multiple-statements,line-too-long + def __init__(self, down: 'bool', rank_err: 'bool', fwd_err: 'bool', reserved: 'int') -> 'None': ... # pylint: disable=super-init-not-called,unused-argument,redefined-builtin,multiple-statements,line-too-long @info_final @@ -241,9 +243,11 @@ class RPLOption(Option): id: 'int' #: Sender rank. rank: 'int' + #: Sub-TLVs [:rfc:`6553#section-3`], kept as raw octets. + sub_tlvs: 'bytes' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Option', action: 'int', change: 'bool', length: 'int', flags: 'RPLFlags', id: 'int', rank: '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', flags: 'RPLFlags', id: 'int', rank: 'int', sub_tlvs: 'bytes') -> '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 598ad41795..30dfeb02eb 100644 --- a/pcapkit/protocols/data/internet/ipv6_opts.py +++ b/pcapkit/protocols/data/internet/ipv6_opts.py @@ -236,10 +236,12 @@ class RPLFlags(Data): rank_err: 'bool' #: Forwarding error flag. fwd_err: 'bool' + #: Reserved flag bits, carried verbatim so that a rebuild reproduces them. + reserved: 'int' if TYPE_CHECKING: def __init__(self, down: 'bool', rank_err: 'bool', - fwd_err: 'bool') -> 'None': ... # pylint: disable=super-init-not-called,unused-argument,redefined-builtin,multiple-statements,line-too-long + fwd_err: 'bool', reserved: 'int') -> 'None': ... # pylint: disable=super-init-not-called,unused-argument,redefined-builtin,multiple-statements,line-too-long @info_final @@ -252,10 +254,12 @@ class RPLOption(Option): id: 'int' #: Sender rank. rank: 'int' + #: Sub-TLVs [:rfc:`6553#section-3`], kept as raw octets. + sub_tlvs: 'bytes' if TYPE_CHECKING: def __init__(self, type: 'Enum_Option', action: 'int', change: 'bool', length: 'int', flags: 'RPLFlags', id: 'int', - rank: 'int') -> 'None': ... # pylint: disable=super-init-not-called,unused-argument,redefined-builtin,multiple-statements,line-too-long + rank: 'int', sub_tlvs: 'bytes') -> '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 6d23eff4e1..9311155113 100644 --- a/pcapkit/protocols/internet/hopopt.py +++ b/pcapkit/protocols/internet/hopopt.py @@ -992,6 +992,9 @@ def _read_opt_rpl(self, schema: 'Schema_RPLOption', *, options: 'Option') -> 'Da | (sub-TLVs) | +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + The sub-TLVs, which :rfc:`6553#section-3` says MUST be skipped when + unrecognised, are kept as raw octets in ``sub_tlvs``. + Args: schema: parsed parameter schema options: extracted HOPOPT options @@ -1000,10 +1003,10 @@ def _read_opt_rpl(self, schema: 'Schema_RPLOption', *, options: 'Option') -> 'Da Parsed option data. Raises: - ProtocolError: If ``schema.len`` is **NOT** ``4``. + ProtocolError: If ``schema.len`` is less than ``4``. """ - if schema.len != 4: + if schema.len < 4: raise ProtocolError(f'{self.alias}: [OptNo {schema.type}] invalid format') opt = Data_RPLOption( @@ -1015,9 +1018,11 @@ def _read_opt_rpl(self, schema: 'Schema_RPLOption', *, options: 'Option') -> 'Da down=bool(schema.flags['down']), rank_err=bool(schema.flags['rank_err']), fwd_err=bool(schema.flags['fwd_err']), + reserved=schema.flags['reserved'], ), id=schema.id, rank=schema.rank, + sub_tlvs=schema.sub_tlvs, ) return opt @@ -1815,6 +1820,8 @@ def _make_opt_rpl(self, code: 'Enum_Option', opt: 'Optional[Data_RPLOption]' = N fwd_err: 'bool' = False, id: 'int' = 0, rank: 'int' = 0, + reserved: 'int' = 0, + sub_tlvs: 'bytes' = b'', **kwargs: 'Any') -> 'Schema_RPLOption': """Make HOPOPT RPL option. @@ -1826,29 +1833,44 @@ def _make_opt_rpl(self, code: 'Enum_Option', opt: 'Optional[Data_RPLOption]' = N fwd_err: forwarding error flag id: RPL instance ID rank: sender rank + reserved: reserved flag bits, i.e. the low five bits of the flags octet + sub_tlvs: sub-TLVs, as raw octets **kwargs: arbitrary keyword arguments Returns: Constructed option schema. + Raises: + ProtocolError: If ``reserved`` does not fit in five bits, or + ``sub_tlvs`` does not fit in the option. + """ if opt is not None: down = opt.flags.down rank_err = opt.flags.rank_err fwd_err = opt.flags.fwd_err + reserved = opt.flags.reserved id = opt.id rank = opt.rank + sub_tlvs = opt.sub_tlvs + + if not 0 <= reserved < 32: + raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid RPL reserved flags: {reserved}') + if len(sub_tlvs) > 251: + raise ProtocolError(f'{self.alias}: [OptNo {code}] too long RPL sub-TLVs: {len(sub_tlvs)}') return Schema_RPLOption( type=code, - len=4, + len=4 + len(sub_tlvs), flags={ 'down': down, 'rank_err': rank_err, 'fwd_err': fwd_err, + 'reserved': reserved, }, id=id, rank=rank, + sub_tlvs=sub_tlvs, ) def _make_opt_mpl(self, code: 'Enum_Option', opt: 'Optional[Data_MPLOption]' = None, *, diff --git a/pcapkit/protocols/internet/ipv6_opts.py b/pcapkit/protocols/internet/ipv6_opts.py index 8a3983fd3e..2011d37fa5 100644 --- a/pcapkit/protocols/internet/ipv6_opts.py +++ b/pcapkit/protocols/internet/ipv6_opts.py @@ -981,6 +981,9 @@ def _read_opt_rpl(self, schema: 'Schema_RPLOption', *, options: 'Option') -> 'Da | (sub-TLVs) | +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + The sub-TLVs, which :rfc:`6553#section-3` says MUST be skipped when + unrecognised, are kept as raw octets in ``sub_tlvs``. + Args: schema: parsed parameter schema options: extracted IPv6-Opts options @@ -989,10 +992,10 @@ def _read_opt_rpl(self, schema: 'Schema_RPLOption', *, options: 'Option') -> 'Da Parsed option data. Raises: - ProtocolError: If ``schema.len`` is **NOT** ``4``. + ProtocolError: If ``schema.len`` is less than ``4``. """ - if schema.len != 4: + if schema.len < 4: raise ProtocolError(f'{self.alias}: [OptNo {schema.type}] invalid format') opt = Data_RPLOption( @@ -1004,9 +1007,11 @@ def _read_opt_rpl(self, schema: 'Schema_RPLOption', *, options: 'Option') -> 'Da down=bool(schema.flags['down']), rank_err=bool(schema.flags['rank_err']), fwd_err=bool(schema.flags['fwd_err']), + reserved=schema.flags['reserved'], ), id=schema.id, rank=schema.rank, + sub_tlvs=schema.sub_tlvs, ) return opt @@ -1804,6 +1809,8 @@ def _make_opt_rpl(self, code: 'Enum_Option', opt: 'Optional[Data_RPLOption]' = N fwd_err: 'bool' = False, id: 'int' = 0, rank: 'int' = 0, + reserved: 'int' = 0, + sub_tlvs: 'bytes' = b'', **kwargs: 'Any') -> 'Schema_RPLOption': """Make IPv6-Opts RPL option. @@ -1815,29 +1822,44 @@ def _make_opt_rpl(self, code: 'Enum_Option', opt: 'Optional[Data_RPLOption]' = N fwd_err: forwarding error flag id: RPL instance ID rank: sender rank + reserved: reserved flag bits, i.e. the low five bits of the flags octet + sub_tlvs: sub-TLVs, as raw octets **kwargs: arbitrary keyword arguments Returns: Constructed option schema. + Raises: + ProtocolError: If ``reserved`` does not fit in five bits, or + ``sub_tlvs`` does not fit in the option. + """ if opt is not None: down = opt.flags.down rank_err = opt.flags.rank_err fwd_err = opt.flags.fwd_err + reserved = opt.flags.reserved id = opt.id rank = opt.rank + sub_tlvs = opt.sub_tlvs + + if not 0 <= reserved < 32: + raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid RPL reserved flags: {reserved}') + if len(sub_tlvs) > 251: + raise ProtocolError(f'{self.alias}: [OptNo {code}] too long RPL sub-TLVs: {len(sub_tlvs)}') return Schema_RPLOption( type=code, - len=4, + len=4 + len(sub_tlvs), flags={ 'down': down, 'rank_err': rank_err, 'fwd_err': fwd_err, + 'reserved': reserved, }, id=id, rank=rank, + sub_tlvs=sub_tlvs, ) def _make_opt_mpl(self, code: 'Enum_Option', opt: 'Optional[Data_MPLOption]' = None, *, diff --git a/pcapkit/protocols/schema/internet/hopopt.py b/pcapkit/protocols/schema/internet/hopopt.py index 06226068b6..5045f1e08c 100644 --- a/pcapkit/protocols/schema/internet/hopopt.py +++ b/pcapkit/protocols/schema/internet/hopopt.py @@ -73,6 +73,8 @@ class RPLFlags(TypedDict): rank_err: int #: Forwarding error flag. fwd_err: int + #: Reserved flag bits. + reserved: int class MPLFlags(TypedDict): """MPL flags.""" @@ -119,6 +121,21 @@ class QSNonce(TypedDict): nonce: int +def rpl_opt_sub_tlv_len(pkt: 'dict[str, Any]') -> 'int': + """Return RPL option sub-TLV length. + + Args: + pkt: RPL option unpacked schema. + + Returns: + RPL option sub-TLV length, i.e. ``Opt Data Len`` less the four octets + of the fixed fields. An ``Opt Data Len`` below ``4`` gives ``0``, and + the reader then rejects the option. + + """ + return max(pkt['len'] - 4, 0) + + def mpl_opt_seed_id_len(pkt: 'dict[str, Any]') -> 'int': """Return MPL Seed-ID length. @@ -667,15 +684,18 @@ class RPLOption(Option, code=[Enum_Option.RPL_Option_0x23, 'down': (0, 1), 'rank_err': (1, 1), 'fwd_err': (2, 1), + 'reserved': (3, 5), }) #: RPL instance ID. id: 'int' = UInt8Field() #: Sender rank. rank: 'int' = UInt16Field() + #: Sub-TLVs, kept as raw octets. + sub_tlvs: 'bytes' = BytesField(length=rpl_opt_sub_tlv_len) if TYPE_CHECKING: def __init__(self, type: 'Enum_Option', len: 'int', flags: 'RPLFlags', id: 'int', - rank: 'int') -> 'None': ... + rank: 'int', sub_tlvs: 'bytes') -> 'None': ... @schema_final diff --git a/pcapkit/protocols/schema/internet/ipv6_opts.py b/pcapkit/protocols/schema/internet/ipv6_opts.py index e4b57be19e..615a41b7e3 100644 --- a/pcapkit/protocols/schema/internet/ipv6_opts.py +++ b/pcapkit/protocols/schema/internet/ipv6_opts.py @@ -73,6 +73,8 @@ class RPLFlags(TypedDict): rank_err: int #: Forwarding error flag. fwd_err: int + #: Reserved flag bits. + reserved: int class MPLFlags(TypedDict): """MPL flags.""" @@ -119,6 +121,21 @@ class QSNonce(TypedDict): nonce: int +def rpl_opt_sub_tlv_len(pkt: 'dict[str, Any]') -> 'int': + """Return RPL option sub-TLV length. + + Args: + pkt: RPL option unpacked schema. + + Returns: + RPL option sub-TLV length, i.e. ``Opt Data Len`` less the four octets + of the fixed fields. An ``Opt Data Len`` below ``4`` gives ``0``, and + the reader then rejects the option. + + """ + return max(pkt['len'] - 4, 0) + + def mpl_opt_seed_id_len(pkt: 'dict[str, Any]') -> 'int': """Return MPL Seed-ID length. @@ -672,15 +689,18 @@ class RPLOption(Option, code=[Enum_Option.RPL_Option_0x23, 'down': (0, 1), 'rank_err': (1, 1), 'fwd_err': (2, 1), + 'reserved': (3, 5), }) #: RPL instance ID. id: 'int' = UInt8Field() #: Sender rank. rank: 'int' = UInt16Field() + #: Sub-TLVs, kept as raw octets. + sub_tlvs: 'bytes' = BytesField(length=rpl_opt_sub_tlv_len) if TYPE_CHECKING: def __init__(self, type: 'Enum_Option', len: 'int', flags: 'RPLFlags', id: 'int', - rank: 'int') -> 'None': ... + rank: 'int', sub_tlvs: 'bytes') -> 'None': ... @schema_final diff --git a/tests/protocols/internet/test_ipv6_extension_unit.py b/tests/protocols/internet/test_ipv6_extension_unit.py index c95231c1be..520a8601c5 100644 --- a/tests/protocols/internet/test_ipv6_extension_unit.py +++ b/tests/protocols/internet/test_ipv6_extension_unit.py @@ -854,8 +854,9 @@ def _assert_option_data_model_constructor_edges(self, protocol_cls: type) -> Non rpl = proto._make_opt_rpl( Option.RPL_Option_0x63, data.RPLOption(type=Option.RPL_Option_0x63, length=6, - flags=data.RPLFlags(down=True, rank_err=True, fwd_err=False), - id=5, rank=6, **base), + flags=data.RPLFlags(down=True, rank_err=True, fwd_err=False, + reserved=0), + id=5, rank=6, sub_tlvs=b'', **base), ) self.assertTrue(rpl.flags['down']) mpl = proto._make_opt_mpl( @@ -1222,8 +1223,8 @@ def assert_bad(reader, option_schema) -> None: rpl = proto._read_opt_rpl( schema.RPLOption(type=Option.RPL_Option_0x63, len=4, - flags={'down': 1, 'rank_err': 0, 'fwd_err': 1}, - id=9, rank=10), + flags={'down': 1, 'rank_err': 0, 'fwd_err': 1, 'reserved': 0}, + id=9, rank=10, sub_tlvs=b''), options=options, ) self.assertTrue(rpl.flags.down) diff --git a/tests/protocols/internet/test_rpl_option_reserved_subtlv_unit.py b/tests/protocols/internet/test_rpl_option_reserved_subtlv_unit.py new file mode 100644 index 0000000000..4a9e094fc3 --- /dev/null +++ b/tests/protocols/internet/test_rpl_option_reserved_subtlv_unit.py @@ -0,0 +1,107 @@ +# -*- coding: utf-8 -*- +"""HOPOPT and IPv6-Opts keep the RPL option's reserved flag bits and sub-TLVs. + +GitHub issue #1199: ``_read_opt_rpl`` and ``_make_opt_rpl`` kept only the +``O``, ``R`` and ``F`` flags, so the five reserved flag bits were rebuilt as +zero. The reader also rejected any ``Opt Data Len`` other than 4, although +:rfc:`6553#section-3` makes the length variable and says unrecognised sub-TLVs +MUST be skipped. + +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 + +#: One-option extension headers, next header 59, for each RPL option type. +#: ``reserved``: flags octet 0x01, so only a reserved bit is set; RPLInstanceID +#: 0x02, SenderRank 0x0000. ``sub_tlv``: Opt Data Len 12, i.e. eight octets of +#: sub-TLVs after the fixed fields, with flags 0xE5 mixing O/R/F and reserved +#: bits. ``sub_tlv_padn``: Opt Data Len 6, two sub-TLV octets, then a PadN. +HEADERS = { + 'reserved': ('3b00{t}0401020000', 0x01, b''), + 'sub_tlv': ('3b01{t}0ce52a010200aa0102030405ff', 0x05, bytes.fromhex('00aa0102030405ff')), + 'sub_tlv_padn': ('3b01{t}06e52a010200aa010400000000', 0x05, bytes.fromhex('00aa')), +} + + +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} + + +def _cases() -> 'list[tuple[str, int, bytes, int, bytes]]': + cases = [] + for label, (template, reserved, sub_tlvs) in HEADERS.items(): + for code in (0x23, 0x63): + octets = bytes.fromhex(template.format(t=f'{code:02x}')) + cases.append((label, code, octets, reserved, sub_tlvs)) + return cases + + +class TestRPLOptionReservedAndSubTLVs(unittest.TestCase): + """Pin reserved flag bits and sub-TLVs of the RPL option.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_fixtures_are_whole_headers(self) -> None: + for label, code, octets, _, _ in _cases(): + with self.subTest(case=label, type=hex(code)): + self.assertEqual(len(octets), (octets[1] + 1) * 8) + + def test_reserved_bits_and_sub_tlvs_are_read(self) -> None: + for name, cls in _classes().items(): + for label, code, octets, reserved, sub_tlvs in _cases(): + with self.subTest(proto=name, case=label, type=hex(code)): + parsed = cls(octets, len(octets), extension=True) + opt = list(parsed.info.options.values())[0] + self.assertEqual(opt.type, code) + self.assertEqual(opt.flags.reserved, reserved) + self.assertEqual(opt.sub_tlvs, sub_tlvs) + self.assertEqual(opt.length, 6 + len(sub_tlvs)) + + def test_rebuild_is_byte_exact(self) -> None: + for name, cls in _classes().items(): + for label, code, octets, _, _ in _cases(): + with self.subTest(proto=name, case=label, type=hex(code)): + parsed = cls(octets, len(octets), extension=True) + self.assertEqual(cls.from_data(parsed.info).data.hex(), octets.hex()) + + def test_maker_defaults_and_keywords(self) -> None: + from pcapkit.const.ipv6.option import Option + from pcapkit.utilities.exceptions import ProtocolError + + for name, cls in _classes().items(): + proto = object.__new__(cls) + with self.subTest(proto=name, kind='defaults'): + made = proto._make_opt_rpl(Option.RPL_Option_0x63, down=True, id=2, rank=3) + self.assertEqual(bytes(made).hex(), '630480020003') + with self.subTest(proto=name, kind='keywords'): + made = proto._make_opt_rpl(Option.RPL_Option_0x23, fwd_err=True, reserved=0x1F, + id=2, rank=3, sub_tlvs=b'\x00\xaa') + self.assertEqual(bytes(made).hex(), '23063f02000300aa') + for kwargs in ({'reserved': 32}, {'reserved': -1}, {'sub_tlvs': bytes(252)}): + with self.subTest(proto=name, kind=f'invalid {sorted(kwargs)}'): + with self.assertRaises(ProtocolError): + proto._make_opt_rpl(Option.RPL_Option_0x23, **kwargs) + + def test_short_rpl_option_still_raises_protocol_error(self) -> None: + from pcapkit.utilities.exceptions import ProtocolError + + # Opt Data Len 3, one short of the fixed fields. + octets = bytes.fromhex('3b00230301020001') + for name, cls in _classes().items(): + with self.subTest(proto=name): + with self.assertRaises(ProtocolError): + cls(octets, len(octets), extension=True) + + +if __name__ == '__main__': + unittest.main()