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: 4 additions & 0 deletions pcapkit/protocols/internet/hip.py
Original file line number Diff line number Diff line change
Expand Up @@ -128,6 +128,7 @@
from pcapkit.protocols.data.internet.hip import ViaRVSParameter as Data_ViaRVSParameter
from pcapkit.protocols.internet.internet import Internet
from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext
from pcapkit.protocols.misc.null import NoPayload
from pcapkit.protocols.schema.internet.hip import HIP as Schema_HIP
from pcapkit.protocols.schema.internet.hip import AckDataParameter as Schema_AckDataParameter
from pcapkit.protocols.schema.internet.hip import ACKParameter as Schema_ACKParameter
Expand Down Expand Up @@ -616,6 +617,9 @@ def read(self, length: 'Optional[int]' = None, *, extension: bool = False, **kwa
])

if extension:
# NOTE: As an IPv6 extension header the upper layer belongs to the enclosing
# IPv6 header, so hold NoPayload here so that membership does not raise (#2487).
self._next = NoPayload() # pylint: disable=attribute-defined-outside-init
return hip

# NOTE: ``_decode_next_layer`` builds the protocol chain from
Expand Down
4 changes: 4 additions & 0 deletions pcapkit/protocols/internet/hopopt.py
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,7 @@
TunnelEncapsulationLimitOption as Data_TunnelEncapsulationLimitOption
from pcapkit.protocols.data.internet.hopopt import UnassignedOption as Data_UnassignedOption
from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext
from pcapkit.protocols.misc.null import NoPayload
from pcapkit.protocols.schema.internet.hopopt import HOPOPT as Schema_HOPOPT
from pcapkit.protocols.schema.internet.hopopt import CALIPSOOption as Schema_CALIPSOOption
from pcapkit.protocols.schema.internet.hopopt import HomeAddressOption as Schema_HomeAddressOption
Expand Down Expand Up @@ -344,6 +345,9 @@ def read(self, length: 'Optional[int]' = None, *, extension: 'bool' = False, **k
)

if extension:
# NOTE: As an IPv6 extension header the upper layer belongs to the enclosing
# IPv6 header, so hold NoPayload here so that membership does not raise (#2487).
self._next = NoPayload() # pylint: disable=attribute-defined-outside-init
return hopopt
return self._decode_next_layer(hopopt, schema.next, length - hopopt.length)

Expand Down
4 changes: 4 additions & 0 deletions pcapkit/protocols/internet/ipv6_ext.py
Original file line number Diff line number Diff line change
Expand Up @@ -148,6 +148,7 @@ class for that code instead (see the bottom of the module), so dispatch
from pcapkit.const.reg.transtype import TransType as Enum_TransType
from pcapkit.protocols.data.internet.ipv6_ext import IPv6_Ext as Data_IPv6_Ext
from pcapkit.protocols.internet.internet import Internet
from pcapkit.protocols.misc.null import NoPayload
from pcapkit.protocols.protocol import _PT, _ST, ProtocolBase
from pcapkit.protocols.schema.internet.ipv6_ext import IPv6_Ext as Schema_IPv6_Ext
from pcapkit.protocols.schema.schema import Schema
Expand Down Expand Up @@ -486,6 +487,9 @@ def read(self, length: 'Optional[int]' = None, *, # pylint: disable=arguments-d
)

if extension:
# NOTE: As an IPv6 extension header the upper layer belongs to the enclosing
# IPv6 header, so hold NoPayload here so that membership does not raise (#2487).
self._next = NoPayload() # pylint: disable=attribute-defined-outside-init
return cast('_PT', generic_ext)
return self._decode_next_layer(cast('_PT', generic_ext), next_header, length - ext_len,
payload=rest[max(ext_len - 2, 0):])
Expand Down
4 changes: 4 additions & 0 deletions pcapkit/protocols/internet/ipv6_frag.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
from pcapkit.const.reg.transtype import TransType as Enum_TransType
from pcapkit.protocols.data.internet.ipv6_frag import IPv6_Frag as Data_IPv6_Frag
from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext
from pcapkit.protocols.misc.null import NoPayload
from pcapkit.protocols.schema.internet.ipv6_frag import IPv6_Frag as Schema_IPv6_Frag
from pcapkit.utilities.exceptions import UnsupportedCall

Expand Down Expand Up @@ -158,6 +159,9 @@ def read(self, length: 'Optional[int]' = None, *, extension: 'bool' = False, #
)

if extension:
# NOTE: As an IPv6 extension header the upper layer belongs to the enclosing
# IPv6 header, so hold NoPayload here so that membership does not raise (#2487).
self._next = NoPayload() # pylint: disable=attribute-defined-outside-init
return ipv6_frag
return self._decode_next_layer(ipv6_frag, schema.next, length - self.length)

Expand Down
4 changes: 4 additions & 0 deletions pcapkit/protocols/internet/ipv6_opts.py
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,7 @@
TunnelEncapsulationLimitOption as Data_TunnelEncapsulationLimitOption
from pcapkit.protocols.data.internet.ipv6_opts import UnassignedOption as Data_UnassignedOption
from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext
from pcapkit.protocols.misc.null import NoPayload
from pcapkit.protocols.schema.internet.ipv6_opts import CALIPSOOption as Schema_CALIPSOOption
from pcapkit.protocols.schema.internet.ipv6_opts import \
HomeAddressOption as Schema_HomeAddressOption
Expand Down Expand Up @@ -336,6 +337,9 @@ def read(self, length: 'Optional[int]' = None, *, extension: 'bool' = False, #
)

if extension:
# NOTE: As an IPv6 extension header the upper layer belongs to the enclosing
# IPv6 header, so hold NoPayload here so that membership does not raise (#2487).
self._next = NoPayload() # pylint: disable=attribute-defined-outside-init
return ipv6_opts
return self._decode_next_layer(ipv6_opts, schema.next, length - ipv6_opts.length)

Expand Down
4 changes: 4 additions & 0 deletions pcapkit/protocols/internet/ipv6_route.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
from pcapkit.protocols.data.internet.ipv6_route import Type2 as Data_Type2
from pcapkit.protocols.data.internet.ipv6_route import UnknownType as Data_UnknownType
from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext
from pcapkit.protocols.misc.null import NoPayload
from pcapkit.protocols.schema.internet.ipv6_route import RPL as Schema_RPL
from pcapkit.protocols.schema.internet.ipv6_route import IPv6_Route as Schema_IPv6_Route
from pcapkit.protocols.schema.internet.ipv6_route import RoutingType as Schema_RoutingType
Expand Down Expand Up @@ -234,6 +235,9 @@ def read(self, length: 'Optional[int]' = None, *, extension: 'bool' = False, #
ipv6_route = meth(schema.data, header=schema)

if extension:
# NOTE: As an IPv6 extension header the upper layer belongs to the enclosing
# IPv6 header, so hold NoPayload here so that membership does not raise (#2487).
self._next = NoPayload() # pylint: disable=attribute-defined-outside-init
return ipv6_route
return self._decode_next_layer(ipv6_route, schema.next, length - ipv6_route.length)

Expand Down
4 changes: 4 additions & 0 deletions pcapkit/protocols/internet/mh.py
Original file line number Diff line number Diff line change
Expand Up @@ -285,6 +285,7 @@
UpdateNotificationMessage as Data_UpdateNotificationMessage
from pcapkit.protocols.data.internet.mh import VendorSpecificOption as Data_VendorSpecificOption
from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext
from pcapkit.protocols.misc.null import NoPayload
from pcapkit.protocols.schema.internet.mh import MH as Schema_MH
from pcapkit.protocols.schema.internet.mh import \
AccessNetworkIdentifierOption as Schema_AccessNetworkIdentifierOption
Expand Down Expand Up @@ -1393,6 +1394,9 @@ def read(self, length: 'Optional[int]' = None, *, version: 'Literal[4, 6]' = 4,
mh = meth(schema.data, header=schema)

if extension:
# NOTE: As an IPv6 extension header the upper layer belongs to the enclosing
# IPv6 header, so hold NoPayload here so that membership does not raise (#2487).
self._next = NoPayload() # pylint: disable=attribute-defined-outside-init
return mh
return self._decode_next_layer(mh, schema.next, length - mh.length)

Expand Down
32 changes: 32 additions & 0 deletions tests/protocols/internet/test_hip_unit.py
Original file line number Diff line number Diff line change
Expand Up @@ -3556,5 +3556,37 @@ def test_hip_echo_response_unsigned_maker_builds_the_response_schema(self) -> No
self.assertEqual(hints['return'], 'Schema_EchoResponseUnsignedParameter')


@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed')
class HIPExtensionMembershipTests(unittest.TestCase):
"""``in`` on a terminal HIP layer parsed from an IPv6 chain (#2487).

The HIP layer is taken from ``IPv6.extension_headers`` in extension mode.
``in`` on it answers for its own name and does not raise. The frame is
built in memory; no capture is read.
"""

def setUp(self) -> None:
reimport_once_per_class(self)

def test_terminal_hip_membership_does_not_raise(self) -> None:
import io # pylint: disable=import-outside-toplevel
import struct # pylint: disable=import-outside-toplevel
import warnings # pylint: disable=import-outside-toplevel

from pcapkit.protocols.internet.hip import HIP
from pcapkit.protocols.internet.ipv6 import IPv6

# HIP: next=59, Hdr Len 4 (40 octets), packet type 1, version 1, checksum, controls
hip_body = struct.pack('!BBBBHH', 59, 4, 0x01, 0x10, 0, 0) + bytes(32)
data = struct.pack('!IHBB', 6 << 28, len(hip_body), 139, 64) + bytes(32) + hip_body
with warnings.catch_warnings():
warnings.simplefilter('ignore')
ipv6 = IPv6(io.BytesIO(data), len(data))
(hip,) = [ext for ext in ipv6.extension_headers.values() if isinstance(ext, HIP)]

self.assertIn('HIP', hip)
self.assertNotIn('ESP', hip)


if __name__ == '__main__':
unittest.main()
79 changes: 79 additions & 0 deletions tests/protocols/internet/test_ipv6_ext_unit.py
Original file line number Diff line number Diff line change
Expand Up @@ -946,3 +946,82 @@ def test_next_is_shared_by_every_subclass(self) -> None:
self.assertEqual(
IPv6_Frag(io.BytesIO(frag_wire), len(frag_wire), version=6,
extension=True).next, TransType.UDP)


@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed')
class ExtensionLayerMembershipTests(unittest.TestCase):
"""``in`` on a terminal extension-header layer parsed from an IPv6 chain (#2487).

A layer taken from ``IPv6.extension_headers`` in extension mode holds
``NoPayload``, so ``in`` answers for its own name and does not raise. Frames
are built in memory; no capture is read.
"""

def setUp(self) -> None:
reimport_once_per_class(self)

@staticmethod
def _ipv6(first: int, body: bytes) -> bytes:
return struct.pack('!IHBB', 6 << 28, len(body), first, 64) + bytes(32) + body

def _parse(self, first: int, body: bytes): # type: ignore[no-untyped-def]
from pcapkit.protocols.internet.ipv6 import IPv6

data = self._ipv6(first, body)
with warnings.catch_warnings():
warnings.simplefilter('ignore')
return IPv6(io.BytesIO(data), len(data))

def _layer(self, first: int, body: bytes, cls): # type: ignore[no-untyped-def]
ipv6 = self._parse(first, body)
return next(ext for ext in ipv6.extension_headers.values() if isinstance(ext, cls))

def test_terminal_hopopt_membership_does_not_raise(self) -> None:
from pcapkit.protocols.internet.hopopt import HOPOPT

# next=59 (No Next Header), Hdr Ext Len 0, PadN(4)
body = struct.pack('!BB', 59, 0) + bytes([1, 4, 0, 0, 0, 0])
hopopt = self._layer(0, body, HOPOPT)

self.assertIn('HOPOPT', hopopt)
self.assertNotIn('ESP', hopopt)

def test_terminal_ipv6_opts_membership_does_not_raise(self) -> None:
from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts

body = struct.pack('!BB', 59, 0) + bytes([1, 4, 0, 0, 0, 0])
opts = self._layer(60, body, IPv6_Opts)

self.assertIn('IPv6_Opts', opts)
self.assertNotIn('ESP', opts)

def test_terminal_ipv6_route_membership_does_not_raise(self) -> None:
from pcapkit.protocols.internet.ipv6_route import IPv6_Route

# next=59, Hdr Ext Len 0, Routing Type 0, Segments Left 0, reserved
body = struct.pack('!BBBBI', 59, 0, 0, 0, 0)
route = self._layer(43, body, IPv6_Route)

self.assertIn('IPv6_Route', route)
self.assertNotIn('ESP', route)

def test_terminal_ipv6_frag_membership_does_not_raise(self) -> None:
from pcapkit.protocols.internet.ipv6_frag import IPv6_Frag

body = struct.pack('!BBHI', 59, 0, 0, 0x1234)
frag = self._layer(44, body, IPv6_Frag)

self.assertIn('IPv6_Frag', frag)
self.assertNotIn('ESP', frag)

def test_terminal_shim6_membership_does_not_raise(self) -> None:
from pcapkit.const.ipv6.extension_header import ExtensionHeader
from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext

# Shim6 (140) has no dedicated parser: the generic fallback handles it.
# next=59 (No Next Header), Hdr Ext Len 0, padding to 8 octets
body = struct.pack('!BB', 59, 0) + bytes(6)
shim6 = self._layer(int(ExtensionHeader.Shim6), body, IPv6_Ext)

self.assertIn('IPv6_Ext', shim6)
self.assertNotIn('TCP', shim6)
31 changes: 31 additions & 0 deletions tests/protocols/internet/test_mh_unit.py
Original file line number Diff line number Diff line change
Expand Up @@ -3647,5 +3647,36 @@ def test_mh_cga_parameters_option_parses(self) -> None:
self.assertEqual(option.parameters[0].prefix, 0x20010db8)


@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed')
class MHExtensionMembershipTests(unittest.TestCase):
"""``in`` on a terminal MH layer parsed from an IPv6 chain (#2487).

The MH layer is taken from ``IPv6.extension_headers`` in extension mode.
``in`` on it answers for its own name and does not raise. The frame is
built in memory; no capture is read.
"""

def setUp(self) -> None:
reimport_once_per_class(self)

def test_terminal_mh_membership_does_not_raise(self) -> None:
import struct # pylint: disable=import-outside-toplevel
import warnings # pylint: disable=import-outside-toplevel

from pcapkit.protocols.internet.ipv6 import IPv6
from pcapkit.protocols.internet.mh import MH

# MH: next=59, Hdr Len 0 (8 octets), MH type 0, reserved, checksum
mh_body = struct.pack('!BBBBH', 59, 0, 0, 0, 0) + bytes(4)
data = struct.pack('!IHBB', 6 << 28, len(mh_body), 135, 64) + bytes(32) + mh_body
with warnings.catch_warnings():
warnings.simplefilter('ignore')
ipv6 = IPv6(io.BytesIO(data), len(data))
(mh,) = [ext for ext in ipv6.extension_headers.values() if isinstance(ext, MH)]

self.assertIn('MH', mh)
self.assertNotIn('ESP', mh)


if __name__ == '__main__':
unittest.main()
Loading