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: 3 additions & 1 deletion pcapkit/protocols/data/internet/hopopt.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down
4 changes: 3 additions & 1 deletion pcapkit/protocols/data/internet/ipv6_opts.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down
7 changes: 6 additions & 1 deletion pcapkit/protocols/internet/hopopt.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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.

Expand All @@ -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:
Expand All @@ -1593,19 +1596,21 @@ 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: '
f'{len(bitmap_bytes)}')

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, *,
Expand Down
7 changes: 6 additions & 1 deletion pcapkit/protocols/internet/ipv6_opts.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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.

Expand All @@ -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:
Expand All @@ -1582,19 +1585,21 @@ 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: '
f'{len(bitmap_bytes)}')

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, *,
Expand Down
7 changes: 4 additions & 3 deletions pcapkit/protocols/schema/internet/hopopt.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
7 changes: 4 additions & 3 deletions pcapkit/protocols/schema/internet/ipv6_opts.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
86 changes: 86 additions & 0 deletions tests/protocols/internet/test_ipv6_calipso_pad_unit.py
Original file line number Diff line number Diff line change
@@ -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()
2 changes: 1 addition & 1 deletion tests/protocols/internet/test_ipv6_extension_unit.py
Original file line number Diff line number Diff line change
Expand Up @@ -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')

Expand Down
Loading