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
3 changes: 3 additions & 0 deletions docs/source/pcapkit/protocols/internet/ipx.rst
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,9 @@ Octets Bits Name Description

.. automethod:: __index__

.. autoattribute:: __proto__
:no-value:

Header Schemas
--------------

Expand Down
45 changes: 42 additions & 3 deletions pcapkit/protocols/internet/ipx.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,20 +24,22 @@
.. [*] https://en.wikipedia.org/wiki/Internetwork_Packet_Exchange

"""
import collections
import textwrap
from typing import TYPE_CHECKING

from pcapkit.const.ipx.packet import Packet as Enum_Packet
from pcapkit.const.ipx.socket import Socket as Enum_Socket
from pcapkit.const.reg.transtype import TransType as Enum_TransType
from pcapkit.corekit.module import ModuleDescriptor
from pcapkit.protocols.data.internet.ipx import IPX as Data_IPX
from pcapkit.protocols.data.internet.ipx import Address as Data_Address
from pcapkit.protocols.internet.internet import Internet
from pcapkit.protocols.schema.internet.ipx import IPX as Schema_IPX

if TYPE_CHECKING:
from enum import IntEnum as StdlibEnum
from typing import Any, Optional, Type
from typing import Any, DefaultDict, Optional, Type

from aenum import IntEnum as AenumEnum
from typing_extensions import Literal
Expand All @@ -61,8 +63,31 @@ class IPX(Internet[Data_IPX, Schema_IPX],
:attr:`~pcapkit.protocols.schema.internet.ipx.IPX.trailer`
(:issue:`1433`).

The payload is dissected by IPX Packet Type from a registry of its own,
:attr:`__proto__`, rather than looked up as an IP protocol number
(:issue:`2169`).

"""

##########################################################################
# Defaults.
##########################################################################

#: Protocol index mapping for decoding next layer, see
#: :meth:`self._decode_next_layer
#: <pcapkit.protocols.internet.internet.Internet._decode_next_layer>` &
#: :meth:`self._import_next_layer
#: <pcapkit.protocols.internet.internet.Internet._import_next_layer>`. Keyed by
#: IPX Packet Type, as in :class:`~pcapkit.const.ipx.packet.Packet`, rather
#: than by IP protocol number, so this class keeps a registry of its own
#: instead of sharing :attr:`Internet.__proto__
#: <pcapkit.protocols.internet.internet.Internet.__proto__>` (:issue:`2169`).
#: No IPX-native protocol is shipped, so the payload is
#: :class:`~pcapkit.protocols.misc.raw.Raw` unless one is registered.
__proto__ = collections.defaultdict(
lambda: ModuleDescriptor('pcapkit.protocols.misc.raw', 'Raw'),
) # type: DefaultDict[int, ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]]

##########################################################################
# Properties.
##########################################################################
Expand All @@ -78,8 +103,8 @@ def length(self) -> 'Literal[30]':
return 30

@property
def protocol(self) -> 'Enum_TransType':
"""Name of next layer protocol."""
def protocol(self) -> 'Enum_Packet':
"""IPX Packet Type of next layer protocol."""
return self._info.type

@property
Expand Down Expand Up @@ -234,6 +259,20 @@ def _make_data(cls, data: 'Data_IPX') -> 'dict[str, Any]': # type: ignore[overr
'trailer': data.get('trailer', b''),
}

def _next_exthdr_depth(self, proto: 'Optional[int]', extension: 'bool' = False) -> 'int': # pylint: disable=unused-argument
"""Place the next layer in its chain of IPv6 extension headers.

Arguments:
proto: next layer IPX Packet Type
extension: if the next layer is parsed as an extension header

Returns:
Always ``0``: an IPX Packet Type is not an IP protocol number, so it
never names an IPv6 extension header (:issue:`2169`).

"""
return 0

@staticmethod
def _make_ipx_address(addr: 'Data_Address') -> 'bytes':
"""Make IPX address field.
Expand Down
110 changes: 110 additions & 0 deletions tests/protocols/internet/test_ipx_next_layer_2169_unit.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,110 @@
# -*- coding: utf-8 -*-
"""An IPX payload is dissected by IPX Packet Type, not as an IP protocol number.

GitHub issue #2169: :meth:`IPX.read <pcapkit.protocols.internet.ipx.IPX.read>`
handed the IPX Packet Type to the next-layer lookup it inherits from
:class:`~pcapkit.protocols.internet.internet.Internet`, whose registry is keyed
by IP protocol number. So Packet Type 17 (NCP) parsed as UDP, 4 (PEP) as IPv4
and 6 as TCP; and Packet Type 0 counted as an IPv6 Hop-by-Hop Options header
towards :data:`~pcapkit.utilities.config.EXTENSION_HEADER_LIMIT`.

:class:`~pcapkit.protocols.internet.ipx.IPX` keeps a registry of its own, keyed
by Packet Type, and its payload is Raw unless an IPX-native protocol is
registered there.

Every case builds its own octets in memory and reads no capture. Classes are
imported inside each test, after :func:`~tests._support.reimport_once_per_class`.

"""

import io
import unittest
import warnings
from unittest import mock

from tests._support import reimport_once_per_class

#: Destination and source IPX addresses.
ADDRS = bytes(range(1, 25))

#: An IPv4 header carrying nothing, which parses as IPv4 when looked up by IP protocol number 4.
IPV4 = bytes.fromhex('4500' '0014' '0000' '0000' '4011' '0000' '0a000001' '0a000002')
#: A TCP header carrying nothing, which parses as TCP when looked up by IP protocol number 6.
TCP = bytes.fromhex('0050' '1f90' '00000001' '00000000' '5002' 'ffff' '0000' '0000')
#: An 8-octet payload, which parses as UDP when looked up by IP protocol number 17.
UDP = bytes.fromhex('0035' '0035' '0008' '0000')


def _ipx(packet_type: int, payload: bytes) -> bytes:
"""Return an IPX packet of ``packet_type`` carrying ``payload``."""
length = 30 + len(payload)
return b'\xff\xff' + length.to_bytes(2, 'big') + b'\x00' + bytes([packet_type]) + ADDRS + payload


#: Packet Types that share a number with a dissected IP protocol, and a payload
#: that protocol would parse.
PACKETS = {
'NCP-17': _ipx(17, UDP),
'PEP-4': _ipx(4, IPV4),
'type-6': _ipx(6, TCP),
}


class TestIPXNextLayer(unittest.TestCase):
"""Pin the IPX next-layer dispatch to its own Packet Type registry."""

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

def test_payload_is_raw_and_rebuilds_byte_for_byte(self) -> None:
from pcapkit.protocols.internet.ipx import IPX
from pcapkit.protocols.misc.raw import Raw

for name, packet in PACKETS.items():
with self.subTest(packet=name), warnings.catch_warnings():
warnings.simplefilter('ignore')
parsed = IPX(io.BytesIO(packet), len(packet))
self.assertIs(type(parsed.payload), Raw)
self.assertEqual(IPX.from_data(parsed.info).data, packet)

def test_registered_ipx_protocol_is_dissected(self) -> None:
from pcapkit.const.ipx.packet import Packet
from pcapkit.protocols.internet.internet import Internet
from pcapkit.protocols.internet.ipx import IPX
from pcapkit.protocols.transport.udp import UDP as Proto_UDP

self.assertIsNot(IPX.__proto__, Internet.__proto__)
packet = PACKETS['NCP-17']
with mock.patch.dict(IPX.__proto__, {Packet.NCP: Proto_UDP}), warnings.catch_warnings():
warnings.simplefilter('ignore')
parsed = IPX(io.BytesIO(packet), len(packet))
self.assertIs(type(parsed.payload), Proto_UDP)

def test_packet_type_zero_is_no_extension_header(self) -> None:
from pcapkit.protocols.internet.ipx import IPX
from pcapkit.protocols.misc.raw import Raw
from pcapkit.utilities import config
from pcapkit.utilities.warnings import ProtocolWarning

packet = _ipx(0, b'abcd')
with mock.patch.object(config, 'EXTENSION_HEADER_LIMIT', 0), warnings.catch_warnings(record=True) as caught:
warnings.simplefilter('always')
parsed = IPX(io.BytesIO(packet), len(packet))
self.assertIs(type(parsed.payload), Raw)
self.assertIsNone(parsed.payload.info.error)
self.assertEqual([w for w in caught if issubclass(w.category, ProtocolWarning)], [])

def test_protocol_is_the_packet_type(self) -> None:
from pcapkit.const.ipx.packet import Packet
from pcapkit.protocols.internet.ipx import IPX

packet = PACKETS['NCP-17']
with warnings.catch_warnings():
warnings.simplefilter('ignore')
parsed = IPX(io.BytesIO(packet), len(packet))
self.assertIs(parsed.protocol, Packet.NCP)
self.assertEqual(IPX.protocol.fget.__annotations__['return'], 'Enum_Packet')


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