From b54e4daa79561e95cfc640fc0153852ec1c58ed4 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Wed, 7 Oct 2026 23:20:16 -0400 Subject: [PATCH] fix(ipv6)!: a CALIPSO option loses its trailing pad and shrinks its length on rebuild (#1403) The octets Opt Data Len covers past the compartment bitmap were a zero-packing PaddingField with no data-class field, so from_data dropped them, wrote Opt Data Len as 8 + bitmap and padded the header with a PadN. - schema: pad is a BytesField sized by calipso_pad_len, kept as read - data: CALIPSOOption gains a required pad field (HOPOPT and IPv6-Opts) - make: _make_opt_calipso takes pad, writes it and counts it in the length --- pcapkit/protocols/data/internet/hopopt.py | 4 +- pcapkit/protocols/data/internet/ipv6_opts.py | 4 +- pcapkit/protocols/internet/hopopt.py | 7 +- pcapkit/protocols/internet/ipv6_opts.py | 7 +- pcapkit/protocols/schema/internet/hopopt.py | 7 +- .../protocols/schema/internet/ipv6_opts.py | 7 +- .../internet/test_ipv6_calipso_pad_unit.py | 86 +++++++++++++++++++ .../internet/test_ipv6_extension_unit.py | 2 +- 8 files changed, 113 insertions(+), 11 deletions(-) create mode 100644 tests/protocols/internet/test_ipv6_calipso_pad_unit.py diff --git a/pcapkit/protocols/data/internet/hopopt.py b/pcapkit/protocols/data/internet/hopopt.py index 975c8ac262..19885361aa 100644 --- a/pcapkit/protocols/data/internet/hopopt.py +++ b/pcapkit/protocols/data/internet/hopopt.py @@ -118,12 +118,14 @@ class CALIPSOOption(Option): level: 'int' #: Checksum. checksum: 'bytes' + #: Octets after the compartment bitmap that ``Opt Data Len`` covers. + pad: 'bytes' if TYPE_CHECKING: #: Compartment bitmap. cmpt_bitmap: 'bytes' - def __init__(self, type: 'Enum_Option', action: 'int', change: 'bool', length: 'int', domain: 'int', cmpt_len: 'int', level: 'int', checksum: 'bytes') -> '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', domain: 'int', cmpt_len: 'int', level: 'int', checksum: 'bytes', pad: 'bytes') -> 'None': ... # pylint: disable=super-init-not-called,unused-argument,redefined-builtin,multiple-statements,line-too-long class SMFDPDOption(Option): diff --git a/pcapkit/protocols/data/internet/ipv6_opts.py b/pcapkit/protocols/data/internet/ipv6_opts.py index ac6d684b48..007e09dfae 100644 --- a/pcapkit/protocols/data/internet/ipv6_opts.py +++ b/pcapkit/protocols/data/internet/ipv6_opts.py @@ -123,13 +123,15 @@ class CALIPSOOption(Option): level: 'int' #: Checksum. checksum: 'bytes' + #: Octets after the compartment bitmap that ``Opt Data Len`` covers. + pad: 'bytes' if TYPE_CHECKING: #: Compartment bitmap. cmpt_bitmap: 'bytes' def __init__(self, type: 'Enum_Option', action: 'int', change: 'bool', length: 'int', domain: 'int', cmpt_len: 'int', level: 'int', - checksum: 'bytes') -> 'None': ... # pylint: disable=super-init-not-called,unused-argument,redefined-builtin,multiple-statements,line-too-long + checksum: 'bytes', pad: 'bytes') -> 'None': ... # pylint: disable=super-init-not-called,unused-argument,redefined-builtin,multiple-statements,line-too-long class SMFDPDOption(Option): diff --git a/pcapkit/protocols/internet/hopopt.py b/pcapkit/protocols/internet/hopopt.py index b9cc1e6911..0e02e4b8cf 100644 --- a/pcapkit/protocols/internet/hopopt.py +++ b/pcapkit/protocols/internet/hopopt.py @@ -757,6 +757,7 @@ def _read_opt_calipso(self, schema: 'Schema_CALIPSOOption', *, options: 'Option' cmpt_len=schema.cmpt_len * 4, level=schema.level, checksum=schema.checksum, + pad=schema.pad, ) if schema.cmpt_len > 0: @@ -1563,6 +1564,7 @@ def _make_opt_calipso(self, code: 'Enum_Option', opt: 'Optional[Data_CALIPSOOpti level: 'int' = 0, checksum: 'bytes' = b'\x00\x00', bitmap: 'Optional[bytes | tuple[int, ...]]' = None, + pad: 'bytes' = b'', **kwargs: 'Any') -> 'Schema_CALIPSOOption': """Make HOPOPT calipso option. @@ -1575,6 +1577,7 @@ def _make_opt_calipso(self, code: 'Enum_Option', opt: 'Optional[Data_CALIPSOOpti bitmap: compartment bitmap, as :obj:`bytes` or as the :obj:`tuple` of octets the reader stores; its length must be a multiple of 4 octets + pad: octets after the compartment bitmap, counted in ``Opt Data Len`` **kwargs: arbitrary keyword arguments Returns: @@ -1593,6 +1596,7 @@ def _make_opt_calipso(self, code: 'Enum_Option', opt: 'Optional[Data_CALIPSOOpti level = opt.level checksum = opt.checksum bitmap = opt.cmpt_bitmap if hasattr(opt, 'cmpt_bitmap') else None + pad = opt.pad bitmap_bytes = bytes(bitmap) if bitmap is not None else b'' if len(bitmap_bytes) % 4 != 0: raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid CALIPSO bitmap length: ' @@ -1600,12 +1604,13 @@ def _make_opt_calipso(self, code: 'Enum_Option', opt: 'Optional[Data_CALIPSOOpti return Schema_CALIPSOOption( type=code, - len=8 + len(bitmap_bytes), + len=8 + len(bitmap_bytes) + len(pad), domain=domain, cmpt_len=len(bitmap_bytes) // 4, level=level, checksum=checksum, bitmap=bitmap_bytes if bitmap_bytes else None, + pad=pad, ) def _make_opt_smf_dpd(self, code: 'Enum_Option', opt: 'Optional[Data_SMFIdentificationBasedDPDOption | Data_SMFHashBasedDPDOption]' = None, *, diff --git a/pcapkit/protocols/internet/ipv6_opts.py b/pcapkit/protocols/internet/ipv6_opts.py index 998db90b02..6a6ac62dda 100644 --- a/pcapkit/protocols/internet/ipv6_opts.py +++ b/pcapkit/protocols/internet/ipv6_opts.py @@ -746,6 +746,7 @@ def _read_opt_calipso(self, schema: 'Schema_CALIPSOOption', *, options: 'Option' cmpt_len=schema.cmpt_len * 4, level=schema.level, checksum=schema.checksum, + pad=schema.pad, ) if schema.cmpt_len > 0: @@ -1552,6 +1553,7 @@ def _make_opt_calipso(self, code: 'Enum_Option', opt: 'Optional[Data_CALIPSOOpti level: 'int' = 0, checksum: 'bytes' = b'\x00\x00', bitmap: 'Optional[bytes | tuple[int, ...]]' = None, + pad: 'bytes' = b'', **kwargs: 'Any') -> 'Schema_CALIPSOOption': """Make IPv6-Opts calipso option. @@ -1564,6 +1566,7 @@ def _make_opt_calipso(self, code: 'Enum_Option', opt: 'Optional[Data_CALIPSOOpti bitmap: compartment bitmap, as :obj:`bytes` or as the :obj:`tuple` of octets the reader stores; its length must be a multiple of 4 octets + pad: octets after the compartment bitmap, counted in ``Opt Data Len`` **kwargs: arbitrary keyword arguments Returns: @@ -1582,6 +1585,7 @@ def _make_opt_calipso(self, code: 'Enum_Option', opt: 'Optional[Data_CALIPSOOpti level = opt.level checksum = opt.checksum bitmap = opt.cmpt_bitmap if hasattr(opt, 'cmpt_bitmap') else None + pad = opt.pad bitmap_bytes = bytes(bitmap) if bitmap is not None else b'' if len(bitmap_bytes) % 4 != 0: raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid CALIPSO bitmap length: ' @@ -1589,12 +1593,13 @@ def _make_opt_calipso(self, code: 'Enum_Option', opt: 'Optional[Data_CALIPSOOpti return Schema_CALIPSOOption( type=code, - len=8 + len(bitmap_bytes), + len=8 + len(bitmap_bytes) + len(pad), domain=domain, cmpt_len=len(bitmap_bytes) // 4, level=level, checksum=checksum, bitmap=bitmap_bytes if bitmap_bytes else None, + pad=pad, ) def _make_opt_smf_dpd(self, code: 'Enum_Option', opt: 'Optional[Data_SMFIdentificationBasedDPDOption | Data_SMFHashBasedDPDOption]' = None, *, diff --git a/pcapkit/protocols/schema/internet/hopopt.py b/pcapkit/protocols/schema/internet/hopopt.py index 8698333ef9..d8e7c03796 100644 --- a/pcapkit/protocols/schema/internet/hopopt.py +++ b/pcapkit/protocols/schema/internet/hopopt.py @@ -469,12 +469,13 @@ class CALIPSOOption(Option, code=Enum_Option.CALIPSO): BytesField(length=lambda pkt: pkt['cmpt_len'] * 4), lambda pkt: pkt['cmpt_len'] > 0, ) - #: Padding. - pad: 'bytes' = PaddingField(length=calipso_pad_len) + #: Octets that ``Opt Data Len`` covers past the compartment bitmap, kept as + #: read rather than packed as zeros. + pad: 'bytes' = BytesField(length=calipso_pad_len, default=b'') if TYPE_CHECKING: def __init__(self, type: 'Enum_Option', len: 'int', domain: 'int', cmpt_len: 'int', - level: 'int', checksum: 'bytes', bitmap: 'Optional[bytes]') -> 'None': ... + level: 'int', checksum: 'bytes', bitmap: 'Optional[bytes]', pad: 'bytes' = ...) -> 'None': ... @schema_final diff --git a/pcapkit/protocols/schema/internet/ipv6_opts.py b/pcapkit/protocols/schema/internet/ipv6_opts.py index 20ecaf669c..ba43afa4a3 100644 --- a/pcapkit/protocols/schema/internet/ipv6_opts.py +++ b/pcapkit/protocols/schema/internet/ipv6_opts.py @@ -469,12 +469,13 @@ class CALIPSOOption(Option, code=Enum_Option.CALIPSO): BytesField(length=lambda pkt: pkt['cmpt_len'] * 4), lambda pkt: pkt['cmpt_len'] > 0, ) - #: Padding. - pad: 'bytes' = PaddingField(length=calipso_pad_len) + #: Octets that ``Opt Data Len`` covers past the compartment bitmap, kept as + #: read rather than packed as zeros. + pad: 'bytes' = BytesField(length=calipso_pad_len, default=b'') if TYPE_CHECKING: def __init__(self, type: 'Enum_Option', len: 'int', domain: 'int', cmpt_len: 'int', - level: 'int', checksum: 'bytes', bitmap: 'Optional[bytes]') -> 'None': ... + level: 'int', checksum: 'bytes', bitmap: 'Optional[bytes]', pad: 'bytes' = ...) -> 'None': ... @schema_final diff --git a/tests/protocols/internet/test_ipv6_calipso_pad_unit.py b/tests/protocols/internet/test_ipv6_calipso_pad_unit.py new file mode 100644 index 0000000000..3fcdcb5b58 --- /dev/null +++ b/tests/protocols/internet/test_ipv6_calipso_pad_unit.py @@ -0,0 +1,86 @@ +# -*- coding: utf-8 -*- +"""A CALIPSO option keeps the octets ``Opt Data Len`` covers past its bitmap. + +GitHub issue #1403: those octets were a zero-packing padding field missing from +the data model, so :meth:`~pcapkit.protocols.protocol.ProtocolBase.from_data` +dropped them, shrank ``Opt Data Len`` and padded the header with a ``PadN`` +instead. They are now the ``pad`` data field, written back on rebuild. + +Every case builds its own octets in memory and reads no capture. The classes +are imported inside each test, after +:func:`~tests._support.reimport_once_per_class`, so that they belong to the +:mod:`pcapkit` import that is live when the test runs. + +""" + +import importlib +import io +import unittest + +from tests._support import reimport_once_per_class + +#: The protocol classes sharing the option code, by module and class name. +PROTOCOLS = ( + ('pcapkit.protocols.internet.hopopt', 'HOPOPT'), + ('pcapkit.protocols.internet.ipv6_opts', 'IPv6_Opts'), +) + +#: Parsed headers which have to rebuild unchanged, with the pad each carries. +HEADERS = { + 'cmpt_len 0, zero pad': ('3b01' '070c' '00000001' '0005' '0000' '00000000', b'\x00' * 4), + 'cmpt_len 0, non-zero pad': ('3b01' '070c' '00000001' '0005' '0000' 'aabbccdd', + b'\xaa\xbb\xcc\xdd'), + 'cmpt_len 2, zero pad': ('3b02' '0714' '00000001' '0205' '0000' '1122334455667788' '00000000', + b'\x00' * 4), + 'cmpt_len 2, non-zero pad': ('3b02' '0714' '00000001' '0205' '0000' '1122334455667788' 'aabbccdd', + b'\xaa\xbb\xcc\xdd'), + 'cmpt_len 0, no pad': ('3b01' '0708' '00000001' '0005' '0000' '01020000', b''), +} + + +class TestCALIPSOPad(unittest.TestCase): + """Pin the CALIPSO pad through read, make and rebuild.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _protocols(self) -> 'list[type]': + return [getattr(importlib.import_module(mod), name) for mod, name in PROTOCOLS] + + def _parse(self, cls: 'type', octets: 'bytes') -> 'object': + return cls(io.BytesIO(octets), len(octets), extension=True) + + def test_parsed_headers_rebuild_byte_for_byte(self) -> None: + from pcapkit.const.ipv6.option import Option + + for cls in self._protocols(): + for label, (hexstr, pad) in HEADERS.items(): + octets = bytes.fromhex(hexstr) + with self.subTest(protocol=cls.__name__, header=label): + parsed = self._parse(cls, octets) + self.assertEqual(parsed.info.options[Option.CALIPSO].pad, pad) # type: ignore[attr-defined] + self.assertEqual(cls.from_data(parsed.info).data, octets) + + def test_make_writes_the_pad_and_counts_it_in_the_length(self) -> None: + from pcapkit.const.ipv6.option import Option + + for cls in self._protocols(): + with self.subTest(protocol=cls.__name__, cmpt_len=0): + made = cls(next=59, options=[(Option.CALIPSO, { + 'domain': 1, 'level': 5, 'pad': b'\xaa\xbb\xcc\xdd', + })]) + self.assertEqual(made.data, bytes.fromhex(HEADERS['cmpt_len 0, non-zero pad'][0])) + with self.subTest(protocol=cls.__name__, cmpt_len=2): + made = cls(next=59, options=[(Option.CALIPSO, { + 'domain': 1, 'level': 5, 'bitmap': bytes.fromhex('1122334455667788'), + 'pad': b'\x00' * 4, + })]) + self.assertEqual(made.data, bytes.fromhex(HEADERS['cmpt_len 2, zero pad'][0])) + with self.subTest(protocol=cls.__name__, pad=None): + # without a pad the option is the fixed header alone, as before + made = cls(next=59, options=[(Option.CALIPSO, {'domain': 1, 'level': 5})]) + self.assertEqual(made.data, bytes.fromhex(HEADERS['cmpt_len 0, no pad'][0])) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/internet/test_ipv6_extension_unit.py b/tests/protocols/internet/test_ipv6_extension_unit.py index 5c09a687a1..49500bf91f 100644 --- a/tests/protocols/internet/test_ipv6_extension_unit.py +++ b/tests/protocols/internet/test_ipv6_extension_unit.py @@ -791,7 +791,7 @@ def _assert_option_data_model_constructor_edges(self, protocol_cls: type) -> Non calipso = data.CALIPSOOption(type=Option.CALIPSO, length=10, domain=7, cmpt_len=4, level=3, checksum=b'\x01\x02', - **base) + pad=b'', **base) calipso.__update__([('cmpt_bitmap', b'\xaa\xbb\xcc\xdd')]) self.assertEqual(proto._make_opt_calipso(Option.CALIPSO, calipso).bitmap, b'\xaa\xbb\xcc\xdd')