diff --git a/docs/source/contributing/conventions/protocol-family-grouping.rst b/docs/source/contributing/conventions/protocol-family-grouping.rst index 353916255a..e3b12e143c 100644 --- a/docs/source/contributing/conventions/protocol-family-grouping.rst +++ b/docs/source/contributing/conventions/protocol-family-grouping.rst @@ -96,13 +96,72 @@ that member alone, so no registry binds :class:`~pcapkit.protocols.internet.ip.I the registry only; parsing or building through a base directly is the rule below. **A versioned family's base dispatches on both paths** (:issue:`2524`). Parsing -octets through it, constructing it from fields and its ``from_data`` give the member -the octets, fields or data model describe, never an instance of the base. In -:class:`~pcapkit.protocols.application.http.HTTP`, -:meth:`~pcapkit.protocols.application.http.HTTP._dispatch_version` picks the member -from the octets, for a payload a transport port binds to HTTP as well, and -:meth:`~pcapkit.protocols.application.http.HTTP._dispatch_fields` from the fields. -The other versioned families are tracked on :issue:`2574`. +octets through it, constructing it from fields and its ``from_data`` give a member, +never an instance of the base, and a subclass of the base is constructed as itself. + +**The member is the one the** ``version`` **keyword names, or else the one the octets +or the data model carry, and never one a field picks.** ``version`` names the member +on either path, unchecked against the octets it is given; where their format carries +a version, the member's own parser refuses another's. Without it, octets give the +member the version they carry, where the format has one, and ``from_data`` reads the +data model's class or its ``version``, as the table below gives. No field picks a +member, not even a field only one member takes, and no member is a default: fields +without ``version`` raise :exc:`~pcapkit.utilities.exceptions.ProtocolError`, and a +field the member does not take fails that member's own keyword check. Ruled on +:issue:`2574`. A class built directly takes ``version`` as well, which must then be +its own version: a member's, or, for a subclass outside the members, the one an IGMP +class states in its ``version``, which a subclass of the other bases does not have. +While parsing, the keyword is handed on unread, since an IP layer passes its own +version under that name to the layer it carries. + +.. list-table:: + :header-rows: 1 + :widths: 13 15 42 30 + + * - Base + - ``version`` + - Octets without ``version`` + - ``from_data`` + * - :class:`~pcapkit.protocols.internet.igmp.IGMP` + - ``1``, ``2``, ``3`` + - :meth:`~pcapkit.protocols.internet.igmp.IGMP._dispatch`: type, max response + time and length, placed by + :meth:`~pcapkit.protocols.internet.igmp.IGMP._classify` + - ``version``, which each IGMP data model carries though no octet does, + ruled on :issue:`2574` + * - :class:`~pcapkit.protocols.internet.ip.IP` + - ``4``, ``6`` + - the Version nibble + - ``version`` + * - :class:`~pcapkit.protocols.internet.icmp.ICMP` + - ``4``, ``6`` + - refused, below + - the data model; a :obj:`dict` is refused + * - :class:`~pcapkit.protocols.link.l2tp.L2TP` + - ``2`` + - the version nibble + - ``version`` + * - :class:`~pcapkit.protocols.link.linux_sll.LinuxSLL` + - ``1``, ``2`` + - refused, below + - the data model; a :obj:`dict` is refused + +:class:`~pcapkit.protocols.application.http.HTTP` dispatches otherwise. Its octets +go to :meth:`~pcapkit.protocols.application.http.HTTP._dispatch_version`, which reads +the connection preface, a start line or a frame header, for a payload a transport +port binds to HTTP as well. Without ``version``, its +:meth:`~pcapkit.protocols.application.http.HTTP._dispatch_fields` picks by each +member's own fields and builds ``HTTPv1`` from none, and its ``from_data`` reads the +data model, or a :obj:`dict` by its ``receipt`` or ``sid``. + +**Octets that carry no version are refused through the base without** ``version`` +(:issue:`2574`). An ICMP message and a Linux cooked-capture header have no version +field: the enclosing layer names it, by IP protocol number 1 or 58 and by link type +113 or 276, and those registries bind the members. So ``ICMP(...)`` and +``LinuxSLL(...)`` on octets raise :exc:`~pcapkit.utilities.exceptions.ProtocolError` +rather than guess, as does their ``from_data`` on a :obj:`dict`, whose keys pick no +member. A version a family has no member for, such as L2TP's ``1`` (L2F) and ``3``, +raises the same way. **Octets neither version takes**, a frameless HTTP/2 connection preface among them, are :class:`~pcapkit.protocols.misc.raw.Raw` with no ``error`` on the transport path, diff --git a/docs/source/pcapkit/protocols/internet/icmp.rst b/docs/source/pcapkit/protocols/internet/icmp.rst index 652e18f238..6a0316e14f 100644 --- a/docs/source/pcapkit/protocols/internet/icmp.rst +++ b/docs/source/pcapkit/protocols/internet/icmp.rst @@ -8,13 +8,23 @@ ICMP - Internet Control Message Protocol which is a base class for Internet Control Message Protocol (ICMP) protocol family [*]_, e.g. :class:`~pcapkit.protocols.internet.icmpv4.ICMPv4` and -:class:`~pcapkit.protocols.internet.icmpv6.ICMPv6`. +:class:`~pcapkit.protocols.internet.icmpv6.ICMPv6`. Constructing the base gives the +one its ``version`` keyword names, from octets or from fields, and is refused +without it, since an ICMP message carries no version of its own. Its +:meth:`~pcapkit.protocols.internet.icmp.ICMP.from_data` gives the one whose data +model it is handed (:issue:`2574`). .. autoclass:: pcapkit.protocols.internet.icmp.ICMP :no-members: :show-inheritance: .. automethod:: id + .. automethod:: from_data + + .. automethod:: __init__ + .. automethod:: __new__ + + .. automethod:: _member_of_version .. rubric:: Footnotes diff --git a/docs/source/pcapkit/protocols/internet/igmp.rst b/docs/source/pcapkit/protocols/internet/igmp.rst index 876b476189..a533fafdbb 100644 --- a/docs/source/pcapkit/protocols/internet/igmp.rst +++ b/docs/source/pcapkit/protocols/internet/igmp.rst @@ -15,7 +15,13 @@ data model as-is; IGMPv3's wire format differs materially by message, so instead -- see :doc:`igmpv3`. :attr:`~pcapkit.protocols.internet.igmp.IGMP.alias` names the protocol each message is read as: ``'IGMP'`` for IGMPv1, the name :rfc:`1112` gives it, and -``'IGMPv2'`` or ``'IGMPv3'`` for the others. +``'IGMPv2'`` or ``'IGMPv3'`` for the others. Constructing the base gives the +version class its ``version`` keyword names; without it, octets give the class +the message classifies as, and fields are refused. Its +:meth:`~pcapkit.protocols.internet.igmp.IGMP.from_data` gives the class the +``version`` of its data names: every IGMP data model carries one, ``1``, ``2`` or +``3``, set by the class that read or built the message and on no octet, so an +IGMPv1 and an IGMPv2 message read alike apart from it (:issue:`2574`). The structure is keyed as for IGMPv1; an IGMPv2 message's keys start ``igmpv2`` instead: @@ -42,11 +48,16 @@ Octets Bits Name Description .. automethod:: read .. automethod:: make + .. automethod:: from_data .. automethod:: __index__ + .. automethod:: __init__ + .. automethod:: __new__ - .. automethod:: _classify .. automethod:: _make_data + .. automethod:: _classify + .. automethod:: _dispatch + .. automethod:: _member_of_version Header Schemas -------------- diff --git a/docs/source/pcapkit/protocols/internet/ip.rst b/docs/source/pcapkit/protocols/internet/ip.rst index 09ef95acf9..18402337c5 100644 --- a/docs/source/pcapkit/protocols/internet/ip.rst +++ b/docs/source/pcapkit/protocols/internet/ip.rst @@ -8,13 +8,23 @@ IP - Internet Protocol which is a base class for Internet Protocol (IP) protocol family [*]_, e.g. :class:`~pcapkit.protocols.internet.ipv4.IPv4` and -:class:`~pcapkit.protocols.internet.ipv6.IPv6`. +:class:`~pcapkit.protocols.internet.ipv6.IPv6`. Constructing the base gives the one +its ``version`` keyword names; without it, octets give the one their Version field +names, and fields are refused. Its +:meth:`~pcapkit.protocols.internet.ip.IP.from_data` reads the ``version`` of its +data (:issue:`2574`). .. autoclass:: pcapkit.protocols.internet.ip.IP :no-members: :show-inheritance: .. automethod:: id + .. automethod:: from_data + + .. automethod:: __init__ + .. automethod:: __new__ + + .. automethod:: _member_of_version .. rubric:: Footnotes diff --git a/docs/source/pcapkit/protocols/internet/ipv4.rst b/docs/source/pcapkit/protocols/internet/ipv4.rst index ac58a5f32d..a53a975d52 100644 --- a/docs/source/pcapkit/protocols/internet/ipv4.rst +++ b/docs/source/pcapkit/protocols/internet/ipv4.rst @@ -48,6 +48,7 @@ Octets Bits Name Description .. automethod:: read .. automethod:: make + .. automethod:: unpack .. automethod:: _make_data diff --git a/docs/source/pcapkit/protocols/internet/ipv6.rst b/docs/source/pcapkit/protocols/internet/ipv6.rst index 546e06f6ae..fe2e3d89c6 100644 --- a/docs/source/pcapkit/protocols/internet/ipv6.rst +++ b/docs/source/pcapkit/protocols/internet/ipv6.rst @@ -37,6 +37,7 @@ Octets Bits Name Description .. automethod:: read .. automethod:: make + .. automethod:: unpack .. automethod:: _make_data diff --git a/docs/source/pcapkit/protocols/link/l2tp.rst b/docs/source/pcapkit/protocols/link/l2tp.rst index 3d3787d9f6..ce9a117a2f 100644 --- a/docs/source/pcapkit/protocols/link/l2tp.rst +++ b/docs/source/pcapkit/protocols/link/l2tp.rst @@ -33,8 +33,12 @@ version's layout for every version. .. autoproperty:: info_name .. automethod:: id + .. automethod:: from_data .. automethod:: __index__ + .. automethod:: __new__ + + .. automethod:: _member_of_version Versions Not Implemented ------------------------ @@ -60,15 +64,21 @@ since callers take element zero as canonical. Version Dispatch ---------------- -Nothing dispatches on the version nibble, because L2TPv2 is the only version -implemented. Another would follow the precedent of -:class:`~pcapkit.protocols.application.http.HTTP`, which reads a version and -delegates to a per-version class. L2TP is the easier case: -:meth:`HTTP._guess_version ` -has to *infer* the version -- recognising the HTTP/2 preface or an HTTP/1 start -line, else trial-parsing each candidate -- because HTTP's wire format carries no -version field, whereas L2TP states its version explicitly in those four bits. A -deterministic switch on ``Ver`` is enough, and no new registry is needed. +Constructing the base gives the member its ``version`` field names, from octets or +from fields; without it, octets give the one their version nibble names, and fields +are refused, since no field picks a member (:issue:`2574`). +:meth:`~pcapkit.protocols.link.l2tp.L2TP._member_of_version` maps ``2`` to +:class:`~pcapkit.protocols.link.l2tpv2.L2TPv2` and refuses any other version, for +that field, for the nibble and for the data +:meth:`~pcapkit.protocols.link.l2tp.L2TP.from_data` is handed. UDP port 1701 binds +L2TPv2 itself, the only member. The switch is where a further member goes, and +binding the port to the base through a hook making the same switch, as +:meth:`IGMP._dispatch ` does for +IP protocol 2, needs no new registry. L2TP is the easier case beside +:class:`~pcapkit.protocols.application.http.HTTP`, whose +:meth:`~pcapkit.protocols.application.http.HTTP._guess_version` has to *infer* +the version, because HTTP's wire format carries no version field, whereas L2TP +states its version explicitly in those four bits. .. rubric:: Footnotes diff --git a/docs/source/pcapkit/protocols/link/linux_sll.rst b/docs/source/pcapkit/protocols/link/linux_sll.rst index e3d4a2bede..c9da4d945c 100644 --- a/docs/source/pcapkit/protocols/link/linux_sll.rst +++ b/docs/source/pcapkit/protocols/link/linux_sll.rst @@ -19,7 +19,12 @@ Linux "cooked" capture encapsulation family: The versions differ in field order and width, so each parses its own header. The base holds what they share: the next-layer dispatch, and link-layer address -validation. It has no registry index, so its ``__index__`` raises. +validation. It has no registry index, so its ``__index__`` raises. Constructing +the base gives the version its ``version`` keyword names, from octets or from +fields, and is refused without it, since the link type rather than the header +names the version. Its +:meth:`~pcapkit.protocols.link.linux_sll.LinuxSLL.from_data` gives the version +whose data model it is handed (:issue:`2574`). .. autoclass:: pcapkit.protocols.link.linux_sll.LinuxSLL :no-members: @@ -27,8 +32,12 @@ validation. It has no registry index, so its ``__index__`` raises. .. automethod:: id .. automethod:: register + .. automethod:: from_data .. automethod:: __index__ + .. automethod:: __init__ + .. automethod:: __new__ .. automethod:: _import_next_layer + .. automethod:: _member_of_version .. autoattribute:: __proto__ :no-value: diff --git a/pcapkit/protocols/data/internet/igmp.py b/pcapkit/protocols/data/internet/igmp.py index dd63734655..8a2a9e324e 100644 --- a/pcapkit/protocols/data/internet/igmp.py +++ b/pcapkit/protocols/data/internet/igmp.py @@ -9,6 +9,8 @@ if TYPE_CHECKING: from ipaddress import IPv4Address + from typing_extensions import Literal + from pcapkit.const.igmp.type import Type as Enum_Type __all__ = ['IGMP'] @@ -27,6 +29,13 @@ class IGMP(Protocol): """ + #: Version of IGMP the message was read as, ``1`` or ``2``: the + #: :attr:`~pcapkit.protocols.internet.igmp.IGMP.version` of the class that + #: read or built it. It is on no octet, since the versions share this + #: layout, and :meth:`IGMP.from_data + #: ` dispatches on it + #: (:issue:`2574`). + version: 'Literal[1, 2]' #: Message type. type: 'Enum_Type' #: Max response time, in units of 1/10 second. Meaningful only for a @@ -41,5 +50,5 @@ class IGMP(Protocol): group: 'IPv4Address' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Type', max_resp_time: 'int', checksum: 'bytes', # pylint: disable=unused-argument,super-init-not-called,multiple-statements,redefined-builtin - group: 'IPv4Address') -> 'None': ... + def __init__(self, version: 'Literal[1, 2]', type: 'Enum_Type', max_resp_time: 'int', # pylint: disable=unused-argument,super-init-not-called,multiple-statements,redefined-builtin + checksum: 'bytes', group: 'IPv4Address') -> 'None': ... diff --git a/pcapkit/protocols/data/internet/igmpv3.py b/pcapkit/protocols/data/internet/igmpv3.py index 09baaaf843..b69a5cfb4a 100644 --- a/pcapkit/protocols/data/internet/igmpv3.py +++ b/pcapkit/protocols/data/internet/igmpv3.py @@ -10,6 +10,8 @@ if TYPE_CHECKING: from ipaddress import IPv4Address + from typing_extensions import Literal + from pcapkit.const.igmp.record_type import RecordType as Enum_RecordType from pcapkit.const.igmp.type import Type as Enum_Type @@ -27,6 +29,11 @@ class IGMPv3(Protocol): """ + #: Version of IGMP, always ``3``. It is on no octet: it names the class + #: that read or built the message, as on the IGMPv1/v2 data model, and + #: :meth:`IGMP.from_data ` + #: dispatches on it (:issue:`2574`). + version: 'Literal[3]' #: Message type. type: 'Enum_Type' #: Header length, in octets. IGMP has no header field of its own that @@ -79,7 +86,7 @@ class Query(IGMPv3): sources: 'tuple[IPv4Address, ...]' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Type', length: 'int', max_resp_code: 'int', + def __init__(self, version: 'Literal[3]', type: 'Enum_Type', length: 'int', max_resp_code: 'int', max_resp_time: 'int', checksum: 'bytes', group: 'IPv4Address', resv: 'int', s_flag: 'bool', qrv: 'int', qqic: 'int', qqi: 'int', sources: 'tuple[IPv4Address, ...]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,redefined-builtin,line-too-long @@ -128,6 +135,7 @@ class Report(IGMPv3): records: 'tuple[GroupRecord, ...]' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Type', length: 'int', reserved_1: 'int', checksum: 'bytes', + def __init__(self, version: 'Literal[3]', type: 'Enum_Type', length: 'int', reserved_1: 'int', + checksum: 'bytes', reserved_2: 'bytes', num_group_records: 'int', records: 'tuple[GroupRecord, ...]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,redefined-builtin diff --git a/pcapkit/protocols/internet/icmp.py b/pcapkit/protocols/internet/icmp.py index 02baf39666..13a71cd535 100644 --- a/pcapkit/protocols/internet/icmp.py +++ b/pcapkit/protocols/internet/icmp.py @@ -9,18 +9,25 @@ which is a base class for Internet Control Message Protocol (ICMP) protocol family [*]_, e.g. :class:`~pcapkit.protocols.internet.icmpv4.ICMPv4` and -:class:`~pcapkit.protocols.internet.icmpv6.ICMPv6`. +:class:`~pcapkit.protocols.internet.icmpv6.ICMPv6`. Constructing the base gives the +one its ``version`` keyword names, from octets or from fields, and is refused +without it, since an ICMP message carries no version of its own. Its +:meth:`~pcapkit.protocols.internet.icmp.ICMP.from_data` gives the one whose data +model it is handed (:issue:`2574`). .. [*] https://en.wikipedia.org/wiki/Internet_Control_Message_Protocol """ -from typing import TYPE_CHECKING, Generic +from typing import TYPE_CHECKING, Generic, cast from pcapkit.protocols.internet.internet import Internet from pcapkit.protocols.protocol import _PT, _ST +from pcapkit.utilities.exceptions import ProtocolError if TYPE_CHECKING: - from typing_extensions import Literal + from typing import IO, Any, Optional, Type + + from typing_extensions import Literal, Self __all__ = ['ICMP'] @@ -31,6 +38,13 @@ class ICMP(Internet[_PT, _ST], Generic[_PT, _ST]): # pylint: disable=abstract-m - Internet Control Message Protocol (:class:`~pcapkit.protocols.internet.icmpv4.ICMPv4`) [:rfc:`792`] - Internet Control Message Protocol for IPv6 (:class:`~pcapkit.protocols.internet.icmpv6.ICMPv6`) [:rfc:`4443`] + It dispatches where the version is known, never giving an instance of this + class (:issue:`2574`). ``version`` names the member, from octets or from + fields, and nothing else does: an ICMP message carries no version of its + own, and no field picks a member, so either is refused without it. + :meth:`from_data` gives the member whose data model it is handed, and + refuses a :obj:`dict`. See :meth:`__new__`. + """ ########################################################################## @@ -46,3 +60,139 @@ def id(cls) -> 'tuple[Literal["ICMP"], Literal["ICMPv6"]]': """ return ('ICMP', 'ICMPv6') + + @classmethod + def from_data(cls, data: '_PT | dict[str, Any]', **kwargs: 'Any') -> 'Self': + """Create protocol instance from data. + + Args: + data: Protocol data. + **kwargs: As :meth:`Protocol.from_data + `. + + Returns: + Protocol instance; on this class, an instance of the member whose data + model ``data`` is. + + Raises: + ProtocolError: If ``data`` is neither member's data model, on this + class. A :obj:`dict` is refused: it carries no version, and its + keys need not tell the members apart, as an Echo message's do not. + + """ + if cls is not ICMP: + return super().from_data(data, **kwargs) + + from pcapkit.protocols.data.internet.icmpv4 import ICMPv4 as Data_ICMPv4 # isort: skip # pylint: disable=line-too-long,import-outside-toplevel + from pcapkit.protocols.data.internet.icmpv6 import ICMPv6 as Data_ICMPv6 # isort: skip # pylint: disable=line-too-long,import-outside-toplevel + from pcapkit.protocols.internet.icmpv4 import ICMPv4 # isort: skip # pylint: disable=import-outside-toplevel + from pcapkit.protocols.internet.icmpv6 import ICMPv6 # isort: skip # pylint: disable=import-outside-toplevel + + if isinstance(data, Data_ICMPv4): + return cast('Self', ICMPv4.from_data(data, **kwargs)) + if isinstance(data, Data_ICMPv6): + return cast('Self', ICMPv6.from_data(data, **kwargs)) + raise ProtocolError('ICMP: invalid data: must be an ICMPv4 or ICMPv6 data model') + + ########################################################################## + # Data models. + ########################################################################## + + def __init__(self, file: 'Optional[IO[bytes] | bytes]' = None, + length: 'Optional[int]' = None, *, + version: 'Optional[Literal[4, 6]]' = None, **kwargs: 'Any') -> 'None': + """Initialisation. + + Args: + file: Source packet stream. + length: Length of packet data. + version: Version of ICMP, ``4`` or ``6``: the member :meth:`__new__` + picks. Building from fields, it must be this class's own. Parsing, + it is handed on unread, as the enclosing IP layer passes its own + version under this name to the layer it carries. + **kwargs: As :meth:`Protocol.__init__ + `. + + Raises: + ProtocolError: If, building from fields, ``version`` names neither + member, or is not this class's own version: that of the member + it is or derives from, which a subclass outside the members + does not have. + + """ + if file is not None: + if version is not None: + kwargs['version'] = version + elif version is not None and not isinstance(self, self._member_of_version(version)): + from pcapkit.protocols.internet.icmpv4 import ICMPv4 # isort: skip # pylint: disable=import-outside-toplevel + from pcapkit.protocols.internet.icmpv6 import ICMPv6 # isort: skip # pylint: disable=import-outside-toplevel + + if not isinstance(self, (ICMPv4, ICMPv6)): + raise ProtocolError(f'{type(self).__name__}: no version of its own to match ' + f'version={version!r}') + raise ProtocolError(f'{type(self).__name__}: version={version!r} names another member') + super().__init__(file, length, **kwargs) # type: ignore[arg-type] + + def __new__(cls, file: 'Optional[IO[bytes] | bytes]' = None, + length: 'Optional[int]' = None, **kwargs: 'Any') -> 'ICMP': + """Create the member a construction of this class describes. + + The ``version`` keyword names the member, on either path, and nothing + else does: an ICMP message carries no version of its own, and no field + picks a member. That member's ``__init__`` then runs with the same + arguments. A subclass of this class is created as itself. + + Args: + file: Source packet stream. + length: Length of packet data. + **kwargs: Arbitrary keyword arguments. + + Returns: + An uninitialised instance of the member. + + Raises: + ProtocolError: If ``version`` is missing, or names neither member. + Without it, octets are refused as fields are: the IP layer names + the version, by protocol number 1 or 58. + + """ + if cls is not ICMP: + return super().__new__(cls, file, length, **kwargs) + version = kwargs.get('version') + if version is None: + if file is not None: + raise ProtocolError('ICMP: no version field; pass version=4 or version=6, ' + 'or parse with ICMPv4 or ICMPv6') + raise ProtocolError('ICMP: no version to build with; pass version=4 or version=6') + member = cls._member_of_version(version) + return member.__new__(member, file, length, **kwargs) + + ########################################################################## + # Utilities. + ########################################################################## + + @staticmethod + def _member_of_version(version: 'Any') -> 'Type[ICMP]': + """The member a ``version`` of ``4`` or ``6`` names. + + Args: + version: Version of ICMP. + + Returns: + :class:`~pcapkit.protocols.internet.icmpv4.ICMPv4` or + :class:`~pcapkit.protocols.internet.icmpv6.ICMPv6`. + + Raises: + ProtocolError: If ``version`` is neither, or is no :obj:`int`: a + :obj:`bool` or a :obj:`float` is refused even where it equals one. + + """ + from pcapkit.protocols.internet.icmpv4 import ICMPv4 # isort: skip # pylint: disable=import-outside-toplevel + from pcapkit.protocols.internet.icmpv6 import ICMPv6 # isort: skip # pylint: disable=import-outside-toplevel + + if isinstance(version, int) and not isinstance(version, bool): + if version == 4: + return cast('Type[ICMP]', ICMPv4) + if version == 6: + return cast('Type[ICMP]', ICMPv6) + raise ProtocolError(f'ICMP: invalid version: {version!r}') diff --git a/pcapkit/protocols/internet/igmp.py b/pcapkit/protocols/internet/igmp.py index b8c6b5203a..73e5ec12e0 100644 --- a/pcapkit/protocols/internet/igmp.py +++ b/pcapkit/protocols/internet/igmp.py @@ -15,6 +15,11 @@ group-record list), so it carries its own schema and data model instead. The registry entry for IP protocol 2 is this base, and :meth:`IGMP._dispatch ` picks the version class. +Constructing the base gives the class its ``version`` keyword names; without it, +octets give the class ``_dispatch`` picks, and fields are refused. +:meth:`~pcapkit.protocols.internet.igmp.IGMP.from_data` gives the class the +``version`` of its data names, which each data model carries and no octet does +(:issue:`2574`). .. versionadded:: 1.6.0 @@ -34,9 +39,10 @@ if TYPE_CHECKING: from enum import IntEnum as StdlibEnum from ipaddress import IPv4Address - from typing import Any, Literal, Optional, Type + from typing import IO, Any, Literal, Optional, Type from aenum import IntEnum as AenumEnum + from typing_extensions import Self from pcapkit.protocols.protocol import ProtocolBase from pcapkit.protocols.schema.schema import Schema @@ -52,6 +58,12 @@ class IGMP(Internet[_PT, _ST], Generic[_PT, _ST], IGMPv2 and IGMPv3 classes. :meth:`read` refuses a message that :meth:`_classify` does not place in :attr:`version`. + It dispatches on both paths, never giving an instance of this class + (:issue:`2574`). ``version`` names the version class; without it, octets + give the one :meth:`_dispatch` picks, and fields are refused, since no field + picks a class. :meth:`from_data` reads the ``version`` of its data. See + :meth:`__new__`. + """ ########################################################################## @@ -128,6 +140,7 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> '_PT': f'message ({length} octets)') igmp = Data_IGMP( + version=self.version, # type: ignore[arg-type] type=schema.type, max_resp_time=schema.max_resp_time, checksum=schema.checksum, @@ -188,6 +201,33 @@ def make(self, payload=payload, ) # type: ignore[return-value] + @classmethod + def from_data(cls, data: '_PT | dict[str, Any]', **kwargs: 'Any') -> 'Self': + """Create protocol instance from data. + + Args: + data: Protocol data. + **kwargs: As :meth:`Protocol.from_data + `. + + Returns: + Protocol instance; on this class, an instance of the version class + the ``version`` of ``data`` names, which every IGMP data model + carries (:issue:`2574`). + + Raises: + ProtocolError: If ``data`` carries no ``version``, or one that names + no version class, on this class. + + """ + if cls is not IGMP: + return super().from_data(data, **kwargs) + version = data.get('version') + if version is None: + raise ProtocolError('IGMP: invalid data: must carry a version of 1, 2 or 3') + member = cls._member_of_version(version) + return cast('Self', member.from_data(data, **kwargs)) + ########################################################################## # Data models. ########################################################################## @@ -208,6 +248,84 @@ def __length_hint__(self) -> 'Literal[8]': """Return an estimated length for the object.""" return 8 + def __init__(self, file: 'Optional[IO[bytes] | bytes]' = None, + length: 'Optional[int]' = None, *, + version: 'Optional[Literal[1, 2, 3]]' = None, **kwargs: 'Any') -> 'None': + """Initialisation. + + Args: + file: Source packet stream. + length: Length of packet data. + version: Version of IGMP, ``1``, ``2`` or ``3``: the class + :meth:`__new__` picks. Building from fields, it must be this + class's :attr:`version`. Parsing, it is handed on unread, as the + enclosing IP layer passes its own version under this name to the + layer it carries. + **kwargs: As :meth:`Protocol.__init__ + `. + + Raises: + ProtocolError: If, building from fields, ``version`` names no version + class, or is not this class's own :attr:`version`, which every + IGMP class states, a subclass outside the members included. + + """ + if file is not None: + if version is not None: + kwargs['version'] = version + elif version is not None: + self._member_of_version(version) + if version != self.version: + raise ProtocolError(f'{type(self).__name__}: version={version!r} ' + 'names another member') + super().__init__(file, length, **kwargs) # type: ignore[arg-type] + + def __new__(cls, file: 'Optional[IO[bytes] | bytes]' = None, + length: 'Optional[int]' = None, **kwargs: 'Any') -> 'IGMP': + """Create the version class a construction of this class describes. + + The ``version`` keyword names the class, on either path. Without it, + parsing octets gives the class :meth:`_dispatch` picks, and building from + fields is refused, since no field picks a class. That class's + ``__init__`` then runs with the same arguments. A subclass of this class + is created as itself. + + Args: + file: Source packet stream, read here without being consumed. + length: Length of packet data. + **kwargs: Arbitrary keyword arguments. + + Returns: + An uninitialised instance of the version class. + + Raises: + ProtocolError: If ``version`` names no version class; or, without it, + if building from fields, or if the octets are no version's + message. + + """ + if cls is not IGMP: + return super().__new__(cls, file, length, **kwargs) + version = kwargs.get('version') + if version is not None: + member = cls._member_of_version(version) + elif file is None: + raise ProtocolError('IGMP: no version to build with; pass version=1, 2 or 3') + else: + if isinstance(file, bytes): + data = file[:length] + else: + offset = file.tell() + data = file.read(length) # type: ignore[arg-type] + file.seek(offset) + from pcapkit.protocols.misc.raw import Raw # isort: skip # pylint: disable=import-outside-toplevel + + picked = cls._dispatch(data, len(data)) + if picked is Raw: + raise ProtocolError('IGMP: unknown version') + member = cast('Type[IGMP]', picked) + return member.__new__(member, file, length, **kwargs) + ########################################################################## # Utilities. ########################################################################## @@ -224,6 +342,7 @@ def _make_data(cls, data: 'Data_IGMP') -> 'dict[str, Any]': # type: ignore[over """ return { + 'version': data.get('version'), 'type': data.type, 'max_resp_time': data.max_resp_time, 'checksum': data.checksum, @@ -287,12 +406,38 @@ def _dispatch(cls, data: 'bytes', length: 'int') -> 'Type[ProtocolBase] | Type[R for any other message, which then keeps its octets undissected. """ - from pcapkit.protocols.internet.igmpv1 import IGMPv1 # isort: skip # pylint: disable=import-outside-toplevel - from pcapkit.protocols.internet.igmpv2 import IGMPv2 # isort: skip # pylint: disable=import-outside-toplevel - from pcapkit.protocols.internet.igmpv3 import IGMPv3 # isort: skip # pylint: disable=import-outside-toplevel from pcapkit.protocols.misc.raw import Raw # isort: skip # pylint: disable=import-outside-toplevel if len(data) < 2: return Raw version = cls._classify(data[0], data[1], length) - return {1: IGMPv1, 2: IGMPv2, 3: IGMPv3}.get(version, Raw) + return cls._member_of_version(version) if version else Raw + + @staticmethod + def _member_of_version(version: 'Any') -> 'Type[IGMP]': + """The version class a ``version`` of ``1``, ``2`` or ``3`` names. + + Args: + version: Version of IGMP, as :meth:`_classify` returns it or the + ``version`` keyword gives it. + + Returns: + :class:`~pcapkit.protocols.internet.igmpv1.IGMPv1`, + :class:`~pcapkit.protocols.internet.igmpv2.IGMPv2` or + :class:`~pcapkit.protocols.internet.igmpv3.IGMPv3`. + + Raises: + ProtocolError: If ``version`` names none of them, or is no + :obj:`int`: a :obj:`bool` or a :obj:`float` is refused even + where it equals one. + + """ + from pcapkit.protocols.internet.igmpv1 import IGMPv1 # isort: skip # pylint: disable=import-outside-toplevel + from pcapkit.protocols.internet.igmpv2 import IGMPv2 # isort: skip # pylint: disable=import-outside-toplevel + from pcapkit.protocols.internet.igmpv3 import IGMPv3 # isort: skip # pylint: disable=import-outside-toplevel + + if isinstance(version, int) and not isinstance(version, bool): + for number, member in ((1, IGMPv1), (2, IGMPv2), (3, IGMPv3)): + if version == number: + return cast('Type[IGMP]', member) + raise ProtocolError(f'IGMP: invalid version: {version!r}') diff --git a/pcapkit/protocols/internet/igmpv3.py b/pcapkit/protocols/internet/igmpv3.py index 5fdef5d440..c295f32a75 100644 --- a/pcapkit/protocols/internet/igmpv3.py +++ b/pcapkit/protocols/internet/igmpv3.py @@ -238,6 +238,7 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_IGMPv3' trailing = ext.get_payload() igmp = Data_Query( + version=self.version, type=schema.type, length=1 + len(body) - len(trailing), max_resp_code=body.max_resp_code, @@ -264,6 +265,7 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_IGMPv3' ) trailing = body.payload igmp = Data_Report( + version=self.version, type=schema.type, length=1 + len(body) - len(trailing), reserved_1=body.reserved_1, @@ -434,6 +436,7 @@ def _make_data(cls, data: 'Data_IGMPv3') -> 'dict[str, Any]': # type: ignore[ov if data.type == Enum_Type.IGMPv3_Membership_Report: report = cast('Data_Report', data) return { + 'version': report.get('version'), 'type': report.type, 'reserved_1': report.reserved_1, 'checksum': report.checksum, @@ -444,6 +447,7 @@ def _make_data(cls, data: 'Data_IGMPv3') -> 'dict[str, Any]': # type: ignore[ov query = cast('Data_Query', data) return { + 'version': query.get('version'), 'type': query.type, 'max_resp_code': query.max_resp_code, 'checksum': query.checksum, diff --git a/pcapkit/protocols/internet/ip.py b/pcapkit/protocols/internet/ip.py index a1661b9e47..712efd06a8 100644 --- a/pcapkit/protocols/internet/ip.py +++ b/pcapkit/protocols/internet/ip.py @@ -13,14 +13,27 @@ .. [*] https://en.wikipedia.org/wiki/Internet_Protocol +.. versionchanged:: 1.6.0 + + Constructing :class:`~pcapkit.protocols.internet.ip.IP` and its + :meth:`~pcapkit.protocols.internet.ip.IP.from_data` give + :class:`~pcapkit.protocols.internet.ipv4.IPv4` or + :class:`~pcapkit.protocols.internet.ipv6.IPv6`, where they raised + :exc:`TypeError`. The ``version`` keyword names the member; without it, + octets give the one their Version field names, and fields are refused + (:issue:`2574`). + """ -from typing import TYPE_CHECKING, Generic +from typing import TYPE_CHECKING, Generic, cast from pcapkit.protocols.internet.internet import Internet from pcapkit.protocols.protocol import _PT, _ST +from pcapkit.utilities.exceptions import ProtocolError if TYPE_CHECKING: - from typing_extensions import Literal + from typing import IO, Any, Optional, Type + + from typing_extensions import Literal, Self __all__ = ['IP'] @@ -35,6 +48,12 @@ class IP(Internet[_PT, _ST], Generic[_PT, _ST]): # pylint: disable=abstract-met :class:`~pcapkit.protocols.internet.esp.ESP`) derive from :class:`~pcapkit.protocols.internet.ipsec.IPsec` instead. + It dispatches on both paths, never giving an instance of this class + (:issue:`2574`). ``version`` names the member; without it, octets give the + one their Version field names, and fields are refused, since no field + picks a member. :meth:`from_data` reads the ``version`` of its data. See + :meth:`__new__`. + """ ########################################################################## @@ -50,3 +69,137 @@ def id(cls) -> 'tuple[Literal["IPv4"], Literal["IPv6"]]': """ return ('IPv4', 'IPv6') + + @classmethod + def from_data(cls, data: '_PT | dict[str, Any]', **kwargs: 'Any') -> 'Self': + """Create protocol instance from data. + + Args: + data: Protocol data. + **kwargs: As :meth:`Protocol.from_data + `. + + Returns: + Protocol instance; on this class, an instance of the member the + ``version`` of ``data`` names, which both data models carry. + + Raises: + ProtocolError: If that ``version`` is neither ``4`` nor ``6``, on this + class. + + """ + if cls is not IP: + return super().from_data(data, **kwargs) + member = cls._member_of_version(data.get('version')) + return cast('Self', member.from_data(data, **kwargs)) + + ########################################################################## + # Data models. + ########################################################################## + + def __init__(self, file: 'Optional[IO[bytes] | bytes]' = None, + length: 'Optional[int]' = None, *, + version: 'Optional[Literal[4, 6]]' = None, **kwargs: 'Any') -> 'None': + """Initialisation. + + Args: + file: Source packet stream. + length: Length of packet data. + version: Version of IP, ``4`` or ``6``: the member :meth:`__new__` + picks. Building from fields, it must be this class's own. Parsing, + it is handed on unread, as the enclosing IP layer passes its own + version under this name to the layer it carries. + **kwargs: As :meth:`Protocol.__init__ + `. + + Raises: + ProtocolError: If, building from fields, ``version`` names neither + member, or is not this class's own version: that of the member + it is or derives from, which a subclass outside the members + does not have. + + """ + if file is not None: + if version is not None: + kwargs['version'] = version + elif version is not None and not isinstance(self, self._member_of_version(version)): + from pcapkit.protocols.internet.ipv4 import IPv4 # isort: skip # pylint: disable=import-outside-toplevel + from pcapkit.protocols.internet.ipv6 import IPv6 # isort: skip # pylint: disable=import-outside-toplevel + + if not isinstance(self, (IPv4, IPv6)): + raise ProtocolError(f'{type(self).__name__}: no version of its own to match ' + f'version={version!r}') + raise ProtocolError(f'{type(self).__name__}: version={version!r} names another member') + super().__init__(file, length, **kwargs) # type: ignore[arg-type] + + def __new__(cls, file: 'Optional[IO[bytes] | bytes]' = None, + length: 'Optional[int]' = None, **kwargs: 'Any') -> 'IP': + """Create the member a construction of this class describes. + + The ``version`` keyword names the member, on either path. Without it, + parsing octets gives the member the Version field in their first four + bits names, and building from fields is refused, since no field picks a + member. That member's ``__init__`` then runs with the same arguments. A + subclass of this class is created as itself. + + Args: + file: Source packet stream, read here without being consumed. + length: Length of packet data. + **kwargs: Arbitrary keyword arguments. + + Returns: + An uninitialised instance of the member. + + Raises: + ProtocolError: If ``version`` names neither member; or, without it, + if building from fields, or if the octets are empty or name + neither version. + + """ + if cls is not IP: + return super().__new__(cls, file, length, **kwargs) + version = kwargs.get('version') + if version is None: + if file is None: + raise ProtocolError('IP: no version to build with; pass version=4 or version=6') + if isinstance(file, bytes): + head = file[:1] + else: + offset = file.tell() + head = file.read(1) + file.seek(offset) + if not head: + raise ProtocolError('IP: no version in an empty packet') + version = head[0] >> 4 + member = cls._member_of_version(version) + return member.__new__(member, file, length, **kwargs) + + ########################################################################## + # Utilities. + ########################################################################## + + @staticmethod + def _member_of_version(version: 'Any') -> 'Type[IP]': + """The member a ``version`` of ``4`` or ``6`` names. + + Args: + version: Version of IP. + + Returns: + :class:`~pcapkit.protocols.internet.ipv4.IPv4` or + :class:`~pcapkit.protocols.internet.ipv6.IPv6`. + + Raises: + ProtocolError: If ``version`` is neither, or is no :obj:`int`: a + :obj:`bool` or a :obj:`float` is refused even where it equals one. + + """ + from pcapkit.protocols.internet.ipv4 import IPv4 # isort: skip # pylint: disable=import-outside-toplevel + from pcapkit.protocols.internet.ipv6 import IPv6 # isort: skip # pylint: disable=import-outside-toplevel + + if isinstance(version, int) and not isinstance(version, bool): + if version == 4: + return cast('Type[IP]', IPv4) + if version == 6: + return cast('Type[IP]', IPv6) + raise ProtocolError(f'IP: invalid version: {version!r}') diff --git a/pcapkit/protocols/internet/ipv4.py b/pcapkit/protocols/internet/ipv4.py index 9547c3b02c..8febc0a4ec 100644 --- a/pcapkit/protocols/internet/ipv4.py +++ b/pcapkit/protocols/internet/ipv4.py @@ -623,6 +623,30 @@ def make(self, schema.payload = upper.from_data(upper.info, checksum=None, pseudo=pseudo).data return schema + def unpack(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_IPv4': + """Unpack (parse) packet data. + + The Version field is checked before the header schema is parsed, so + octets of another version are refused as such, rather than by whichever + field of this header they break first, as an IPv6 header does with the IHL of ``0`` to ``4`` it reads (:issue:`2574`). + + Args: + length: Length of packet data. + **kwargs: Arbitrary keyword arguments. + + Returns: + Parsed packet data. + + Raises: + ProtocolError: If the Version field is not ``4``. + + """ + if cast('Optional[Schema_IPv4]', self.__header__) is None and self._data: + version = self._data[0] >> 4 + if version != 4: + raise ProtocolError(f'[IPv4] invalid version: {version}') + return super().unpack(length, **kwargs) + @classmethod def id(cls) -> 'tuple[Literal["IPv4"]]': # type: ignore[override] """Index ID of the protocol. diff --git a/pcapkit/protocols/internet/ipv6.py b/pcapkit/protocols/internet/ipv6.py index 8e6c243b5a..63d5c47554 100644 --- a/pcapkit/protocols/internet/ipv6.py +++ b/pcapkit/protocols/internet/ipv6.py @@ -340,6 +340,30 @@ def make(self, schema.payload = upper.from_data(upper.info, checksum=None, pseudo=pseudo).data return schema + def unpack(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_IPv6': + """Unpack (parse) packet data. + + The Version field is checked before the header schema is parsed, so + octets of another version are refused as such, rather than by whichever + field of this header they break first (:issue:`2574`). + + Args: + length: Length of packet data. + **kwargs: Arbitrary keyword arguments. + + Returns: + Parsed packet data. + + Raises: + ProtocolError: If the Version field is not ``6``. + + """ + if cast('Optional[Schema_IPv6]', self.__header__) is None and self._data: + version = self._data[0] >> 4 + if version != 6: + raise ProtocolError(f'[IPv6] invalid version: {version}') + return super().unpack(length, **kwargs) + @classmethod def id(cls) -> 'tuple[Literal["IPv6"]]': # type: ignore[override] """Index ID of the protocol. diff --git a/pcapkit/protocols/link/l2tp.py b/pcapkit/protocols/link/l2tp.py index 755fa91af8..0c0ea62932 100644 --- a/pcapkit/protocols/link/l2tp.py +++ b/pcapkit/protocols/link/l2tp.py @@ -70,42 +70,44 @@ Selecting a Version ------------------- -Nothing *dispatches* on the version nibble, because L2TPv2 is the only version -implemented -- but -:meth:`L2TPv2.read ` does **check** -it, and refuses anything other than ``2``. That is the half of the mechanism -which is useful with no other version implemented: it keeps v3 traffic on port -1701 (:rfc:`3931` §4.7.2 shares the port, so this is ordinary capture +Constructing the base gives the member its ``version`` field names, from octets +or from fields; without it, octets give the one their version nibble names, and +fields are refused, since no field picks a member (:issue:`2574`). +:meth:`L2TP._member_of_version ` +maps ``2`` to :class:`~pcapkit.protocols.link.l2tpv2.L2TPv2` and refuses any +other version, for that field, for the nibble and for the data +:meth:`L2TP.from_data ` is handed. +:meth:`L2TPv2.read ` **checks** the +nibble as well and refuses anything other than ``2``. That keeps v3 traffic on +port 1701 (:rfc:`3931` §4.7.2 shares the port, so this is ordinary capture traffic rather than a corner case) from being reported as v2 with a tunnel and session ID read out of v3's Control Connection ID. -When another version lands, the remaining half -- delegation rather than refusal --- already has a precedent in -:class:`~pcapkit.protocols.application.http.HTTP`, which reads a version and -delegates to a per-version class. L2TP is the easier case: +UDP port 1701 binds L2TPv2 itself, the only member. +:meth:`~pcapkit.protocols.link.l2tp.L2TP._member_of_version` is where a further +member goes, and binding the port to this base through a hook making the same +switch, as :meth:`IGMP._dispatch ` +does for IP protocol 2, needs no new registry. L2TP is the easier case beside +:class:`~pcapkit.protocols.application.http.HTTP`, whose :meth:`~pcapkit.protocols.application.http.HTTP._guess_version` has to *infer* -the version -- recognising the HTTP/2 preface or an HTTP/1 start line, else -trial-parsing each candidate -- because HTTP's wire format carries no version -field, whereas L2TP states its version explicitly in those four bits. So a -deterministic switch on ``Ver`` is enough, and no new registry is needed -- the -class bound at UDP 1701 reads two octets, masks out the nibble, and hands the -datagram to the matching class. Note the switch belongs on -the **UDP** path only: over IP protocol 115 there is no nibble to switch on, per -the §4.1.1.1 note above. +the version because HTTP's wire format carries no version field, whereas L2TP +states its version explicitly in those four bits. The switch belongs on the +**UDP** path only: over IP protocol 115 there is no nibble to switch on, per the +§4.1.1.1 note above. .. [*] https://en.wikipedia.org/wiki/Layer_2_Tunneling_Protocol """ -from typing import TYPE_CHECKING, Generic +from typing import TYPE_CHECKING, Generic, cast from pcapkit.protocols.link.link import Link from pcapkit.protocols.protocol import _PT, _ST -from pcapkit.utilities.exceptions import UnsupportedCall +from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall if TYPE_CHECKING: - from typing import NoReturn + from typing import IO, Any, NoReturn, Optional, Type - from typing_extensions import Literal + from typing_extensions import Literal, Self __all__ = ['L2TP'] @@ -121,14 +123,26 @@ class L2TP(Link[_PT, _ST], Generic[_PT, _ST]): # pylint: disable=abstract-metho :attr:`~pcapkit.protocols.protocol.Protocol.name` and :meth:`~pcapkit.protocols.protocol.Protocol.read` are both declared abstract by :class:`~pcapkit.protocols.protocol.ProtocolBase` and neither is - defined here, so the class cannot be instantiated. Bind a version, never this - class. + defined here, so no instance of it exists. Bind a version, never this class. + + It dispatches on both paths, never giving an instance of this class + (:issue:`2574`). The ``version`` field names the member; without it, octets + give the one their version nibble names, and fields are refused. + :meth:`from_data` reads the ``version`` of its data. See :meth:`__new__`. .. versionchanged:: 1.5.0 Became an abstract base for the family; its version-2 parsing moved to :class:`~pcapkit.protocols.link.l2tpv2.L2TPv2`. + .. versionchanged:: 1.6.0 + + Constructing it and :meth:`from_data` give + :class:`~pcapkit.protocols.link.l2tpv2.L2TPv2`, where they raised + :exc:`TypeError`. The ``version`` field names the member; without it, + octets give the one their version nibble names, and fields are refused + (:issue:`2574`). + """ ########################################################################## @@ -163,6 +177,28 @@ def id(cls) -> 'tuple[Literal["L2TP"], Literal["L2TPv2"]]': """ return ('L2TP', 'L2TPv2') + @classmethod + def from_data(cls, data: '_PT | dict[str, Any]', **kwargs: 'Any') -> 'Self': + """Create protocol instance from data. + + Args: + data: Protocol data. + **kwargs: As :meth:`Protocol.from_data + `. + + Returns: + Protocol instance; on this class, an instance of the member the + ``version`` of ``data`` names. + + Raises: + ProtocolError: If no member reads that ``version``, on this class. + + """ + if cls is not L2TP: + return super().from_data(data, **kwargs) + member = cls._member_of_version(data.get('version')) + return cast('Self', member.from_data(data, **kwargs)) + ########################################################################## # Data models. ########################################################################## @@ -180,3 +216,72 @@ def __index__(cls) -> 'NoReturn': # pylint: disable=invalid-index-returned """ raise UnsupportedCall(f'{cls.__name__!r} object cannot be interpreted as an integer') + + def __new__(cls, file: 'Optional[IO[bytes] | bytes]' = None, + length: 'Optional[int]' = None, **kwargs: 'Any') -> 'L2TP': + """Create the member a construction of this class describes. + + The ``version`` field names the member, on either path. Without it, + parsing octets gives the member the version nibble, bits 12 to 15 of + their first word, names, and building from fields is refused, since no + field picks a member. That member's ``__init__`` then runs with the same + arguments. A subclass of this class is created as itself. + + Args: + file: Source packet stream, read here without being consumed. + length: Length of packet data. + **kwargs: Arbitrary keyword arguments. + + Returns: + An uninitialised instance of the member. + + Raises: + ProtocolError: If no member reads ``version``; or, without it, if + building from fields, or if the octets are shorter than the first + word or no member reads their version. + + """ + if cls is not L2TP: + return super().__new__(cls, file, length, **kwargs) + version = kwargs.get('version') + if version is None: + if file is None: + raise ProtocolError('L2TP: no version to build with; pass version=2') + if isinstance(file, bytes): + head = file[:2] + else: + offset = file.tell() + head = file.read(2) + file.seek(offset) + if len(head) < 2: + raise ProtocolError('L2TP: no version in a packet shorter than 2 octets') + version = head[1] & 0x0F + member = cls._member_of_version(version) + return member.__new__(member, file, length, **kwargs) + + ########################################################################## + # Utilities. + ########################################################################## + + @staticmethod + def _member_of_version(version: 'Any') -> 'Type[L2TP]': + """The member that reads a ``version`` of the version nibble. + + Args: + version: Version of L2TP. + + Returns: + :class:`~pcapkit.protocols.link.l2tpv2.L2TPv2` for ``2``. + + Raises: + ProtocolError: For any other ``version``, or one that is no + :obj:`int`, as a :obj:`bool` or a :obj:`float` is even where it + equals ``2``: ``1`` is L2F and ``3`` is L2TPv3, neither + implemented. + + """ + from pcapkit.protocols.link.l2tpv2 import L2TPv2 # isort: skip # pylint: disable=import-outside-toplevel + + if isinstance(version, int) and not isinstance(version, bool) and version == 2: + return cast('Type[L2TP]', L2TPv2) + raise ProtocolError(f'L2TP: invalid version: {version!r}') diff --git a/pcapkit/protocols/link/linux_sll.py b/pcapkit/protocols/link/linux_sll.py index ee7d8b5c0a..bef19a8b21 100644 --- a/pcapkit/protocols/link/linux_sll.py +++ b/pcapkit/protocols/link/linux_sll.py @@ -7,11 +7,15 @@ :mod:`pcapkit.protocols.link.linux_sll` contains :class:`~pcapkit.protocols.link.linux_sll.LinuxSLL` only, the base class of :class:`~pcapkit.protocols.link.linux_sll1.LinuxSLL1` and -:class:`~pcapkit.protocols.link.linux_sll2.LinuxSLL2`. +:class:`~pcapkit.protocols.link.linux_sll2.LinuxSLL2`. Constructing the base +gives the one its ``version`` keyword names, from octets or from fields, and is +refused without it, since the link type rather than the header names the +version. Its :meth:`~pcapkit.protocols.link.linux_sll.LinuxSLL.from_data` gives +the one whose data model it is handed (:issue:`2574`). """ import collections -from typing import TYPE_CHECKING, Generic +from typing import TYPE_CHECKING, Generic, cast from pcapkit.const.linux.arphrd import ARPHRD as Enum_ARPHRD from pcapkit.const.reg.ethertype import EtherType as Enum_EtherType @@ -24,9 +28,9 @@ from pcapkit.utilities.warnings import RegistryWarning, warn if TYPE_CHECKING: - from typing import Any, ClassVar, DefaultDict, NoReturn, Optional, Type + from typing import IO, Any, ClassVar, DefaultDict, NoReturn, Optional, Type - from typing_extensions import Literal + from typing_extensions import Literal, Self from pcapkit.corekit.enum import EnumRegistry from pcapkit.protocols.protocol import ProtocolBase @@ -123,6 +127,13 @@ class LinuxSLL(Link[_PT, _ST], Generic[_PT, _ST]): # pylint: disable=abstract-m dispatches it through :attr:`Link.__proto__ `, as an Ethernet frame does. + It dispatches where the version is known, never giving an instance of this + class (:issue:`2574`). ``version`` names the member, from octets or from + fields, and nothing else does: the header carries no version of its own, + and no field picks a member, so either is refused without it. + :meth:`from_data` gives the member whose data model it is handed, and + refuses a :obj:`dict`. See :meth:`__new__`. + """ ########################################################################## @@ -203,6 +214,39 @@ def register(cls, code: 'Enum_ARPHRD', protocol: 'ModuleDescriptor[ProtocolBase] f'{incumbent!r} with {entry!r}', RegistryWarning) cls.__proto__[code] = entry + @classmethod + def from_data(cls, data: '_PT | dict[str, Any]', **kwargs: 'Any') -> 'Self': + """Create protocol instance from data. + + Args: + data: Protocol data. + **kwargs: As :meth:`Protocol.from_data + `. + + Returns: + Protocol instance; on this class, an instance of the member whose data + model ``data`` is. + + Raises: + ProtocolError: If ``data`` is neither member's data model, on this + class. A :obj:`dict` is refused: it carries no version, and no + key picks a member. + + """ + if cls is not LinuxSLL: + return super().from_data(data, **kwargs) + + from pcapkit.protocols.data.link.linux_sll1 import LinuxSLL1 as Data_LinuxSLL1 # isort: skip # pylint: disable=line-too-long,import-outside-toplevel + from pcapkit.protocols.data.link.linux_sll2 import LinuxSLL2 as Data_LinuxSLL2 # isort: skip # pylint: disable=line-too-long,import-outside-toplevel + from pcapkit.protocols.link.linux_sll1 import LinuxSLL1 # isort: skip # pylint: disable=import-outside-toplevel + from pcapkit.protocols.link.linux_sll2 import LinuxSLL2 # isort: skip # pylint: disable=import-outside-toplevel + + if isinstance(data, Data_LinuxSLL1): + return cast('Self', LinuxSLL1.from_data(data, **kwargs)) + if isinstance(data, Data_LinuxSLL2): + return cast('Self', LinuxSLL2.from_data(data, **kwargs)) + raise ProtocolError('LinuxSLL: invalid data: must be a LinuxSLL1 or LinuxSLL2 data model') + ########################################################################## # Data models. ########################################################################## @@ -218,6 +262,71 @@ def __index__(cls) -> 'NoReturn': # pylint: disable=invalid-index-returned """ raise UnsupportedCall(f'{cls.__name__!r} object cannot be interpreted as an integer') + def __init__(self, file: 'Optional[IO[bytes] | bytes]' = None, + length: 'Optional[int]' = None, *, + version: 'Optional[Literal[1, 2]]' = None, **kwargs: 'Any') -> 'None': + """Initialisation. + + Args: + file: Source packet stream. + length: Length of packet data. + version: Version of the header, ``1`` or ``2``: the member + :meth:`__new__` picks. Building from fields, it must be this + class's own. Parsing, it is handed on unread. + **kwargs: As :meth:`Protocol.__init__ + `. + + Raises: + ProtocolError: If, building from fields, ``version`` names neither + member, or is not this class's own version: that of the member + it is or derives from, which a subclass outside the members + does not have. + + """ + if file is not None: + if version is not None: + kwargs['version'] = version + elif version is not None and not isinstance(self, self._member_of_version(version)): + if not any(isinstance(self, self._member_of_version(own)) for own in (1, 2)): + raise ProtocolError(f'{type(self).__name__}: no version of its own to match ' + f'version={version!r}') + raise ProtocolError(f'{type(self).__name__}: version={version!r} names another member') + super().__init__(file, length, **kwargs) # type: ignore[arg-type] + + def __new__(cls, file: 'Optional[IO[bytes] | bytes]' = None, + length: 'Optional[int]' = None, **kwargs: 'Any') -> 'LinuxSLL': + """Create the member a construction of this class describes. + + The ``version`` keyword names the member, on either path, and nothing + else does: the header carries no version of its own, and no field picks + a member. That member's ``__init__`` then runs with the same arguments. A + subclass of this class is created as itself. + + Args: + file: Source packet stream. + length: Length of packet data. + **kwargs: Arbitrary keyword arguments. + + Returns: + An uninitialised instance of the member. + + Raises: + ProtocolError: If ``version`` is missing, or names neither member. + Without it, octets are refused as fields are: the link type names + the version, ``LINUX_SLL`` (113) or ``LINUX_SLL2`` (276). + + """ + if cls is not LinuxSLL: + return super().__new__(cls, file, length, **kwargs) + version = kwargs.get('version') + if version is None: + if file is not None: + raise ProtocolError('LinuxSLL: no version field; pass version=1 or version=2, ' + 'or parse with LinuxSLL1 or LinuxSLL2') + raise ProtocolError('LinuxSLL: no version to build with; pass version=1 or version=2') + member = cls._member_of_version(version) + return member.__new__(member, file, length, **kwargs) + ########################################################################## # Utilities. ########################################################################## @@ -273,3 +382,29 @@ def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *, # return self._parse_next_layer(protocol, file_, length, alias=proto, packet=packet, layer=self._exlayer, protocol=self._exproto, __context__=self._exctx) + + @staticmethod + def _member_of_version(version: 'Any') -> 'Type[LinuxSLL]': + """The member a ``version`` of ``1`` or ``2`` names. + + Args: + version: Version of the header. + + Returns: + :class:`~pcapkit.protocols.link.linux_sll1.LinuxSLL1` or + :class:`~pcapkit.protocols.link.linux_sll2.LinuxSLL2`. + + Raises: + ProtocolError: If ``version`` is neither, or is no :obj:`int`: a + :obj:`bool` or a :obj:`float` is refused even where it equals one. + + """ + from pcapkit.protocols.link.linux_sll1 import LinuxSLL1 # isort: skip # pylint: disable=import-outside-toplevel + from pcapkit.protocols.link.linux_sll2 import LinuxSLL2 # isort: skip # pylint: disable=import-outside-toplevel + + if isinstance(version, int) and not isinstance(version, bool): + if version == 1: + return cast('Type[LinuxSLL]', LinuxSLL1) + if version == 2: + return cast('Type[LinuxSLL]', LinuxSLL2) + raise ProtocolError(f'LinuxSLL: invalid version: {version!r}') diff --git a/tests/project/test_rfc_constants_one_home_1739_unit.py b/tests/project/test_rfc_constants_one_home_1739_unit.py index 45eaf48827..2f9f92bba0 100644 --- a/tests/project/test_rfc_constants_one_home_1739_unit.py +++ b/tests/project/test_rfc_constants_one_home_1739_unit.py @@ -236,6 +236,12 @@ 'the IP version number, not the Next Header offset', ('pcapkit/protocols/internet/ipv6.py', "{'version': 6}"): 'the IP version number, not the Next Header offset', + ('pcapkit/protocols/internet/ipv6.py', 'version != 6'): + 'the IP version number, not the Next Header offset', + ('pcapkit/protocols/internet/ip.py', 'version == 6'): + 'the IP version number, not the Next Header offset', + ('pcapkit/protocols/internet/icmp.py', 'version == 6'): + 'ICMPv6, the ICMP of IP version 6 (:rfc:`4443`), not the Next Header offset', ('pcapkit/protocols/schema/internet/ipv6.py', 'ptr + 6'): 'the end of a Jumbo Payload option (2 octets of header, 4 of value)', ('pcapkit/protocols/link/linux_sll1.py', 'return 16'): diff --git a/tests/protocols/internet/test_icmp_base_dispatch_2574_unit.py b/tests/protocols/internet/test_icmp_base_dispatch_2574_unit.py new file mode 100644 index 0000000000..dc29c52ef1 --- /dev/null +++ b/tests/protocols/internet/test_icmp_base_dispatch_2574_unit.py @@ -0,0 +1,202 @@ +# -*- coding: utf-8 -*- +"""The ICMP family base dispatches where the version is known. C.f. #2574. + +A versioned family's base never yields an instance of itself. The ``version`` +keyword names the member :class:`~pcapkit.protocols.internet.icmp.ICMP` gives, +from octets or from fields, and nothing else does: an ICMP message has no +version field -- the IP layer names it, by protocol number 1 (RFC 792) or 58 +(RFC 4443, section 1) -- and no field picks a member, so both are refused +without it, ``pseudo`` included. ``from_data`` gives the member whose data model +it is handed, and refuses a ``dict``, whose keys need not tell the two apart: an +Echo message's do not. + +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 struct +import unittest + +from tests._support import reimport_once_per_class + +#: An Echo Request of each version, identifier 7, sequence 2, checksum zero: +#: type 8 in ICMPv4 (RFC 792) and 128 in ICMPv6 (RFC 4443, section 4.1). +ECHO = { + 4: struct.pack('>BBHHH', 8, 0, 0, 7, 2) + b'odd', + 6: struct.pack('>BBHHH', 128, 0, 0, 7, 2) + b'odd', +} + +#: The refusals given no ``version``. +NO_VERSION_OCTETS = '^ICMP: no version field; pass version=4 or version=6, or parse with ICMPv4 or ICMPv6$' +NO_VERSION_FIELDS = '^ICMP: no version to build with; pass version=4 or version=6$' + + +def pseudo() -> 'object': + """An IPv6 pseudo-header, which only ICMPv6 takes.""" + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + + return IP_Pseudo(src='2001:db8::1', dst='2001:db8::2', protocol='IPv6_ICMP', upper_length=0) + + +class ICMPBaseDispatchTests(unittest.TestCase): + """``ICMP(...)`` and ``ICMP.from_data`` give the member, never the base.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + @staticmethod + def member(version: 'int') -> 'type': + from pcapkit.protocols.internet.icmpv4 import ICMPv4 + from pcapkit.protocols.internet.icmpv6 import ICMPv6 + + return {4: ICMPv4, 6: ICMPv6}[version] + + def test_octets_without_version_raise(self) -> None: + from pcapkit.protocols.internet.icmp import ICMP + from pcapkit.utilities.exceptions import ProtocolError + + for version, octets in ECHO.items(): + for source in ('bytes', 'stream'): + with self.subTest(octets=version, source=source): + with self.assertRaisesRegex(ProtocolError, NO_VERSION_OCTETS): + if source == 'bytes': + ICMP(octets) + else: + ICMP(io.BytesIO(octets), len(octets)) + + def test_version_picks_the_member_from_octets(self) -> None: + """Even octets of the other version: the format carries none to object.""" + from pcapkit.protocols.internet.icmp import ICMP + + for version in (4, 6): + for octets_of, octets in ECHO.items(): + for source in ('bytes', 'stream'): + with self.subTest(version=version, octets=octets_of, source=source): + layer = (ICMP(octets, version=version) if source == 'bytes' + else ICMP(io.BytesIO(octets), len(octets), version=version)) + self.assertIs(type(layer), self.member(version)) + self.assertIsInstance(layer, ICMP) + self.assertEqual(layer.data, octets) + self.assertEqual(layer.data, self.member(version)(octets).data) + + def test_version_picks_the_member_from_fields(self) -> None: + """Each build is byte-exact against the member's own and parses back unchanged.""" + from pcapkit.protocols.internet.icmp import ICMP + + cases = { + 'ICMPv4 defaults': ({}, 4), + 'ICMPv4 echo': ({'type': 8, 'body': ECHO[4][4:]}, 4), + 'ICMPv6 defaults': ({}, 6), + 'ICMPv6 echo': ({'type': 128, 'body': ECHO[6][4:], 'pseudo': pseudo()}, 6), + } + for label, (fields, version) in cases.items(): + with self.subTest(fields=label): + layer = ICMP(version=version, **fields) + self.assertIs(type(layer), self.member(version)) + self.assertEqual(layer.data, self.member(version)(**fields).data) + parsed = ICMP(layer.data, version=version) + self.assertEqual(parsed.data, layer.data) + self.assertEqual(ICMP.from_data(parsed.info).data, layer.data) + self.assertEqual(ICMP(version=4, type=8, body=ECHO[4][4:], checksum=b'\x00\x00').data, ECHO[4]) + + def test_fields_without_version_raise(self) -> None: + """Even ``pseudo``, the field only ICMPv6 takes, picks nothing.""" + from pcapkit.protocols.internet.icmp import ICMP + from pcapkit.utilities.exceptions import ProtocolError + + for label, fields in (('no fields', {}), ('shared fields', {'type': 8, 'body': ECHO[4][4:]}), + ('pseudo', {'type': 128, 'pseudo': pseudo()})): + with self.subTest(fields=label): + with self.assertRaisesRegex(ProtocolError, NO_VERSION_FIELDS): + ICMP(**fields) + + def test_version_is_the_only_check(self) -> None: + """A bad ``version`` raises; a field the member lacks fails the member's check.""" + from pcapkit.protocols.internet.icmp import ICMP + from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall + + for version in (1, True, '6', 4.0, 6.0): + with self.subTest(version=version): + with self.assertRaisesRegex(ProtocolError, f'^ICMP: invalid version: {version!r}$'): + ICMP(version=version) + with self.assertRaisesRegex(UnsupportedCall, "^ICMPv4: unexpected keyword\\(s\\): 'pseudo'"): + ICMP(version=4, pseudo=pseudo()) + + def test_a_member_takes_version_only_as_its_own(self) -> None: + """Building, ``version`` must name the member; parsing, it is handed on unread.""" + from pcapkit.protocols.internet.icmpv4 import ICMPv4 + from pcapkit.protocols.internet.icmpv6 import ICMPv6 + from pcapkit.utilities.exceptions import ProtocolError + + self.assertEqual(ICMPv4(version=4).data, ICMPv4().data) + self.assertEqual(ICMPv6(version=6).data, ICMPv6().data) + with self.assertRaisesRegex(ProtocolError, '^ICMPv4: version=6 names another member$'): + ICMPv4(version=6) + with self.assertRaisesRegex(ProtocolError, '^ICMPv6: version=4 names another member$'): + ICMPv6(version=4) + self.assertEqual(ICMPv6(ECHO[6], version=4).data, ECHO[6]) + + def test_a_subclass_outside_the_members_has_no_version_of_its_own(self) -> None: + """So it takes no ``version``, as a member takes no other's.""" + from pcapkit.protocols.internet.icmp import ICMP + from pcapkit.utilities.exceptions import ProtocolError + + class Outside(ICMP): # type: ignore[type-arg] + """A concrete subclass of the base that is no member.""" + + name = length = None # type: ignore[assignment] + @classmethod + def __index__(cls) -> 'int': # type: ignore[override] + raise NotImplementedError + + def read(self, length: 'object' = None, **kwargs: 'object') -> 'object': + raise NotImplementedError + + def make(self, **kwargs: 'object') -> 'object': + raise NotImplementedError + + with self.assertRaisesRegex(ProtocolError, '^Outside: no version of its own to match version=6$'): + Outside(version=6) + + def test_from_data_on_the_base_gives_the_member(self) -> None: + from pcapkit.protocols.internet.icmp import ICMP + from pcapkit.utilities.exceptions import ProtocolError + + for version, octets in ECHO.items(): + with self.subTest(octets=version): + parsed = self.member(version)(octets) + rebuilt = ICMP.from_data(parsed.info) + self.assertIs(type(rebuilt), self.member(version)) + self.assertEqual(rebuilt.data, octets) + with self.assertRaisesRegex(ProtocolError, + '^ICMP: invalid data: must be an ICMPv4 or ICMPv6 data model$'): + ICMP.from_data(parsed.info.to_dict()) + with self.assertRaisesRegex(ProtocolError, '^ICMPv4: version=6 names another member$'): + ICMP.from_data(self.member(4)(ECHO[4]).info, version=6) + + def test_only_the_base_dispatches(self) -> None: + """A subclass of the base is constructed as itself, abstract or not.""" + from pcapkit.protocols.data.internet.icmpv4 import ICMPv4 as Data_ICMPv4 + from pcapkit.protocols.internet.icmp import ICMP + from pcapkit.protocols.internet.icmpv4 import ICMPv4 + from pcapkit.protocols.schema.internet.icmpv4 import ICMPv4 as Schema_ICMPv4 + + class Abstract(ICMP): # type: ignore[type-arg] # pylint: disable=abstract-method + """A direct subclass of the base, still abstract.""" + + class Custom(ICMPv4, schema=Schema_ICMPv4, data=Data_ICMPv4): + """A concrete subclass of the base, through a member.""" + + octets = ECHO[4] + self.assertIs(type(ICMP(type=8, version=4)), ICMPv4) + self.assertIs(type(Custom(octets)), Custom) + self.assertIs(type(Custom(type=8, version=4)), Custom) + self.assertIs(type(Custom.from_data(Custom(octets).info)), Custom) + with self.assertRaisesRegex(TypeError, 'abstract'): + Abstract(octets) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/internet/test_igmp_unit.py b/tests/protocols/internet/test_igmp_unit.py index 8443a2df80..88cf5032c5 100644 --- a/tests/protocols/internet/test_igmp_unit.py +++ b/tests/protocols/internet/test_igmp_unit.py @@ -12,10 +12,11 @@ group-record list, both materially different shapes of their own. Each case parses its octets with the concrete class for the version it -names; the abstract base cannot be built. A type none of the three versions -define falls back to :class:`~pcapkit.protocols.misc.raw.Raw`, as does a -Membership Query of 9 to 11 octets, which is neither IGMPv1/v2 (8) nor -IGMPv3 (12 or more) (:rfc:`3376#section-7.1`). +names; constructing the abstract base gives that class (#2574). A type none +of the three versions define falls back to +:class:`~pcapkit.protocols.misc.raw.Raw`, as does a Membership Query of 9 to +11 octets, which is neither IGMPv1/v2 (8) nor IGMPv3 (12 or more) +(:rfc:`3376#section-7.1`). 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`, @@ -410,12 +411,15 @@ def setUp(self) -> None: reimport_once_per_class(self) def test_base_is_abstract(self) -> None: + """The base stays abstract; constructing it gives the version class (#2574).""" from pcapkit.protocols.internet.igmp import IGMP + from pcapkit.protocols.internet.igmpv2 import IGMPv2 self.assertIn('version', IGMP.__abstractmethods__) - packet = IGMP_PACKETS['v2-report'][0] with self.assertRaises(TypeError): - IGMP(io.BytesIO(packet), len(packet)) # type: ignore[abstract] + object.__new__(IGMP) + packet = IGMP_PACKETS['v2-report'][0] + self.assertIs(type(IGMP(io.BytesIO(packet), len(packet))), IGMPv2) def test_id_and_version(self) -> None: from pcapkit.protocols.internet.igmpv1 import IGMPv1 @@ -1155,12 +1159,204 @@ def test_dispatch_selects_version_per_message(self) -> None: self.assertIs(IGMP._dispatch(IGMP_QUERY_10_OCTETS, len(IGMP_QUERY_10_OCTETS)), Raw) self.assertIs(IGMP._dispatch(b'', 0), Raw) - def test_base_refuses_instantiation(self) -> None: + def test_parsing_through_the_base_gives_the_version_class(self) -> None: + """``IGMP(octets)`` is the class ``_dispatch`` picks, and keeps the octets (#2574).""" from pcapkit.protocols.internet.igmp import IGMP - self.assertIn('version', IGMP.__abstractmethods__) - with self.assertRaises(TypeError): - IGMP(io.BytesIO(IGMP_PACKETS['v2-report'][0]), 8) # type: ignore[abstract] + for name, (packet, version) in {**IGMP_PACKETS, **IGMPV3_PACKETS}.items(): + for source in ('bytes', 'stream'): + with self.subTest(packet=name, source=source): + layer = (IGMP(packet) if source == 'bytes' + else IGMP(io.BytesIO(packet), len(packet))) + self.assertIs(type(layer), igmp_class(version)) + self.assertIsInstance(layer, IGMP) + self.assertEqual(layer.data, packet) + self.assertEqual(IGMP.from_data(layer.info).data, packet) + + def test_octets_no_version_reads_raise(self) -> None: + from pcapkit.protocols.internet.igmp import IGMP + from pcapkit.utilities.exceptions import ProtocolError + + for label, octets in (('10-octet-query', IGMP_QUERY_10_OCTETS), + ('unknown type', igmp_packet(0x30, 0, '0.0.0.0')), + ('empty', b'')): + with self.subTest(octets=label): + with self.assertRaisesRegex(ProtocolError, '^IGMP: unknown version$'): + IGMP(octets) + + def test_version_picks_the_class_from_octets(self) -> None: + """``IGMP(octets, version=N)`` parses as that class, and only as it (#2574).""" + from pcapkit.protocols.internet.igmp import IGMP + from pcapkit.utilities.exceptions import ProtocolError + + packets = {**IGMP_PACKETS, **IGMPV3_PACKETS} + for name, (packet, version) in packets.items(): + for source in ('bytes', 'stream'): + with self.subTest(packet=name, source=source): + number = int(version[-1]) + layer = (IGMP(packet, version=number) if source == 'bytes' + else IGMP(io.BytesIO(packet), len(packet), version=number)) + self.assertIs(type(layer), igmp_class(version)) + self.assertEqual(layer.data, packet) + # NOTE: No cross-check against the octets: the class ``version`` + # names parses them, and its own ``read`` refuses another version's. + for other in (1, 2, 3): + if other == int(version[-1]): + continue + with self.subTest(packet=name, version=other): + with self.assertRaisesRegex(ProtocolError, f'IGMPv{other}'): + IGMP(packet, version=other) + + def test_version_picks_the_class_from_fields(self) -> None: + """Each build is byte-exact against the class's own and parses back unchanged (#2574).""" + from pcapkit.const.igmp.type import Type + from pcapkit.protocols.internet.igmp import IGMP + + cases = { + 'v1 defaults': ({}, 'IGMPv1'), + 'v1 report': ({'type': Type.IGMPv1_Membership_Report, 'group': '224.0.0.1'}, 'IGMPv1'), + 'v2 query': ({'type': Type.IGMP_Membership_Query, 'max_resp_time': 100}, 'IGMPv2'), + 'v2 leave by name': ({'type': 'IGMPv2_Leave_Group', 'group': '239.1.2.3'}, 'IGMPv2'), + 'v3 report by type': ({'type': Type.IGMPv3_Membership_Report}, 'IGMPv3'), + 'v3 query by field': ({'type': Type.IGMP_Membership_Query, 'qrv': 2, 'qqic': 125}, 'IGMPv3'), + } + for label, (fields, version) in cases.items(): + with self.subTest(fields=label): + layer = IGMP(version=int(version[-1]), **fields) + klass = igmp_class(version) + self.assertIs(type(layer), klass) + self.assertEqual(layer.data, klass(**fields).data) + parsed = IGMP(layer.data) + self.assertIs(type(parsed), klass) + self.assertEqual(parsed.data, layer.data) + + def test_fields_without_version_raise(self) -> None: + """Even a field only IGMPv3 takes picks nothing (#2574).""" + from pcapkit.const.igmp.type import Type + from pcapkit.protocols.internet.igmp import IGMP + from pcapkit.utilities.exceptions import ProtocolError + + cases = { + 'no fields': {}, + 'v2 query': {'type': Type.IGMP_Membership_Query, 'max_resp_time': 100}, + 'v3 report by type': {'type': Type.IGMPv3_Membership_Report}, + 'v3 query by field': {'type': Type.IGMP_Membership_Query, 'qrv': 2, 'qqic': 125}, + 'unknown type': {'type': 0x30}, + 'unknown field': {'no_such_field': 1}, + } + for label, fields in cases.items(): + with self.subTest(fields=label): + with self.assertRaisesRegex(ProtocolError, + r'^IGMP: no version to build with; pass version=1, 2 or 3$'): + IGMP(**fields) + + def test_version_is_the_only_check(self) -> None: + """The class ``version`` names refuses what it does not build, by its own checks.""" + from pcapkit.protocols.internet.igmp import IGMP + from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall + + for version in (0, 4, True, '2', 1.0, 3.0): + with self.subTest(version=version): + with self.assertRaisesRegex(ProtocolError, f'^IGMP: invalid version: {version!r}$'): + IGMP(version=version) + with self.assertRaisesRegex(ProtocolError, 'invalid type'): + IGMP(version=2, type=300) + with self.assertRaisesRegex(ProtocolError, 'is not an IGMPv2 message'): + IGMP(version=2) + with self.assertRaises(UnsupportedCall): + IGMP(version=1, no_such_field=1) + with self.assertRaisesRegex(UnsupportedCall, "^IGMPv1: unexpected keyword\\(s\\): 'qqic', 'qrv'"): + IGMP(version=1, qrv=2, qqic=125) + + def test_a_version_class_takes_version_only_as_its_own(self) -> None: + """Building, ``version`` must be the class's; parsing, it is handed on unread.""" + from pcapkit.protocols.internet.igmpv1 import IGMPv1 + from pcapkit.protocols.internet.igmpv2 import IGMPv2 + from pcapkit.utilities.exceptions import ProtocolError + + self.assertEqual(IGMPv1(version=1).data, IGMPv1().data) + with self.assertRaisesRegex(ProtocolError, '^IGMPv1: version=2 names another member$'): + IGMPv1(version=2) + with self.assertRaisesRegex(ProtocolError, '^IGMPv2: version=3 names another member$'): + IGMPv2(version=3, max_resp_time=100) + # NOTE: The enclosing IP layer passes its own version, ``4``, under the + # same name, as ``test_ipv4_dispatches_igmp_to_each_version`` relies on. + packet = IGMP_PACKETS['v2-report'][0] + self.assertEqual(IGMPv2(packet, version=4).data, packet) + + def test_version_is_in_the_data_model_only(self) -> None: + """Each data model carries ``version``; no octet does, and every round trip keeps it (#2574).""" + from pcapkit.protocols.internet.igmp import IGMP + + for name, (packet, version) in {**IGMP_PACKETS, **IGMPV3_PACKETS}.items(): + with self.subTest(packet=name): + klass, number = igmp_class(version), int(version[-1]) + parsed = IGMP(packet) + self.assertEqual(parsed.info.version, number) + self.assertEqual(list(parsed.info.to_dict())[0], 'version') + self.assertEqual(parsed.data, packet) + rebuilt = klass.from_data(parsed.info) + self.assertEqual(rebuilt.info.version, number) + self.assertEqual(rebuilt.data, packet) + fields = klass._make_data(parsed.info) + self.assertEqual(fields['version'], number) + made = IGMP(**fields) + self.assertEqual(made.info.version, number) + self.assertEqual(made.data, packet) + self.assertEqual(IGMP(made.data).info, parsed.info) + + def test_from_data_on_the_base_dispatches_on_the_data_s_version(self) -> None: + """Not on its type, max response time or length (#2574).""" + from pcapkit.protocols.internet.igmp import IGMP + from pcapkit.utilities.exceptions import ProtocolError + + for name, (packet, version) in {**IGMP_PACKETS, **IGMPV3_PACKETS}.items(): + parsed = igmp_class(version)(packet) + for form in ('info', 'dict'): + with self.subTest(packet=name, form=form): + data = parsed.info if form == 'info' else parsed.info.to_dict() + rebuilt = IGMP.from_data(data) + self.assertIs(type(rebuilt), igmp_class(version)) + self.assertEqual(rebuilt.data, packet) + report = igmp_class('IGMPv2')(IGMP_PACKETS['v2-report'][0]).info.to_dict() + # NOTE: An IGMPv2 Report relabelled ``1`` goes to IGMPv1, whose own read + # then refuses the message its type describes. + with self.assertRaisesRegex(ProtocolError, 'is not an IGMPv1 message'): + IGMP.from_data({**report, 'version': 1}) + for data in ({'type': 0x30}, {key: val for key, val in report.items() if key != 'version'}): + with self.subTest(data=sorted(data)): + with self.assertRaisesRegex(ProtocolError, + '^IGMP: invalid data: must carry a version of 1, 2 or 3$'): + IGMP.from_data(data) + with self.assertRaisesRegex(ProtocolError, '^IGMP: invalid version: 4$'): + IGMP.from_data({**report, 'version': 4}) + with self.assertRaisesRegex(ProtocolError, '^IGMPv2: version=1 names another member$'): + IGMP.from_data(igmp_class('IGMPv2')(IGMP_PACKETS['v2-report'][0]).info, version=1) + + def test_only_the_base_dispatches(self) -> None: + """A subclass of the base is constructed as itself, on every path.""" + from pcapkit.protocols.data.internet.igmp import IGMP as Data_IGMP + from pcapkit.protocols.internet.igmp import IGMP + from pcapkit.protocols.internet.igmpv2 import IGMPv2 + from pcapkit.protocols.schema.internet.igmp import IGMP as Schema_IGMP + from pcapkit.utilities.exceptions import ProtocolError + + class Custom(IGMP, schema=Schema_IGMP, data=Data_IGMP): # type: ignore[type-arg] + """An IGMPv2-shaped subclass of the base, outside the family.""" + + @property + def version(self) -> 'int': + return 2 + + packet = IGMP_PACKETS['v2-report'][0] + self.assertIs(type(IGMP(packet)), IGMPv2) + self.assertIs(type(Custom(packet)), Custom) + self.assertIs(type(Custom.from_data(Custom(packet).info)), Custom) + self.assertIs(type(Custom(type=0x16, group='239.1.2.3', version=2)), Custom) + # NOTE: A subclass outside the members states a version of its own, as + # every IGMP class does, and takes no other in ``version``. + with self.assertRaisesRegex(ProtocolError, '^Custom: version=1 names another member$'): + Custom(type=0x16, group='239.1.2.3', version=1) def test_ipv4_dispatches_igmp_to_each_version(self) -> None: from pcapkit.protocols.internet.igmpv1 import IGMPv1 diff --git a/tests/protocols/internet/test_ip_base_dispatch_2574_unit.py b/tests/protocols/internet/test_ip_base_dispatch_2574_unit.py new file mode 100644 index 0000000000..01d07429f9 --- /dev/null +++ b/tests/protocols/internet/test_ip_base_dispatch_2574_unit.py @@ -0,0 +1,272 @@ +# -*- coding: utf-8 -*- +"""The IP family base dispatches on both paths. C.f. #2574. + +A versioned family's base never yields an instance of itself: parsing octets +through :class:`~pcapkit.protocols.internet.ip.IP`, constructing it from fields, +and its ``from_data`` all give ``IPv4`` or ``IPv6``. The ``version`` keyword +names the member on either path, and it is the only thing that does: fields +without it are refused, whichever member takes them. Octets without it are read +by their Version field (RFC 791, section 3.1; RFC 8200, section 3), and data by +its ``version``. + +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 enum +import io +import struct +import unittest + +from tests._support import reimport_once_per_class + +#: A UDP datagram to the discard port, carrying ``ab``. +UDP = struct.pack('>HHHH', 40000, 9, 10, 0) + b'ab' + + +def ipv4(body: 'bytes', proto: 'int', version: 'int' = 4) -> 'bytes': + """An IPv4 header of ``version``, checksum zero, carrying ``body`` as ``proto``.""" + return struct.pack('>BBHHHBBH4s4s', version << 4 | 5, 0, 20 + len(body), 1, 0x4000, 64, proto, 0, + b'\x0a\x00\x00\x01', b'\x0a\x00\x00\x02') + body + + +def ipv6(body: 'bytes', nxt: 'int') -> 'bytes': + """An IPv6 header, ``::1`` to ``::2``, carrying ``body`` as ``nxt``.""" + return (struct.pack('>IHBB', 6 << 28, len(body), nxt, 64) + + bytes(15) + b'\x01' + bytes(15) + b'\x02' + body) + + +#: Octets, with the member that parses them. +OCTETS = { + 'IPv4 UDP': (ipv4(UDP, 17), 'IPv4'), + 'IPv6 UDP': (ipv6(UDP, 17), 'IPv6'), + 'IPv6 no next header': (ipv6(b'abcd', 59), 'IPv6'), +} + +#: Fields, with the ``version`` that builds them. +FIELDS = { + 'IPv4 defaults': ({}, 4), + 'IPv4 shared fields': ({'src': '10.0.0.1', 'dst': '10.0.0.2', 'payload': UDP}, 4), + 'IPv4 own fields': ({'ttl': 3, 'protocol': 17, 'payload': UDP}, 4), + 'IPv6 defaults': ({}, 6), + 'IPv6 shared fields': ({'src': '::1', 'dst': '::2', 'payload': UDP}, 6), + 'IPv6 own fields': ({'hop_limit': 3, 'next': 17, 'src': '::1', 'dst': '::2', 'payload': UDP}, 6), +} + +#: The refusal of fields given no ``version``. +NO_VERSION = r'^IP: no version to build with; pass version=4 or version=6$' + + +class IPBaseDispatchTests(unittest.TestCase): + """``IP(...)`` and ``IP.from_data`` give the member, never the base.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + @staticmethod + def member(name: 'str | int') -> 'type': + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.internet.ipv6 import IPv6 + + return {'IPv4': IPv4, 'IPv6': IPv6, 4: IPv4, 6: IPv6}[name] + + def test_octets_without_version_give_the_member_their_version_names(self) -> None: + from pcapkit.protocols.internet.ip import IP + + for label, (octets, name) in OCTETS.items(): + for source in ('bytes', 'stream'): + with self.subTest(octets=label, source=source): + layer = IP(octets) if source == 'bytes' else IP(io.BytesIO(octets), len(octets)) + direct = self.member(name)(octets) + self.assertIs(type(layer), self.member(name)) + self.assertIsInstance(layer, IP) + self.assertEqual(layer.info.version, octets[0] >> 4) + self.assertEqual(layer.data, octets) + self.assertEqual(str(layer.protochain), str(direct.protochain)) + + def test_octets_naming_no_version_raise(self) -> None: + from pcapkit.protocols.internet.ip import IP + from pcapkit.utilities.exceptions import ProtocolError + + for version in (0, 5, 15): + with self.subTest(version=version): + with self.assertRaisesRegex(ProtocolError, f'^IP: invalid version: {version}$'): + IP(ipv4(UDP, 17, version)) + with self.assertRaisesRegex(ProtocolError, '^IP: no version in an empty packet$'): + IP(b'') + + def test_version_picks_the_member_from_fields(self) -> None: + """Each build is byte-exact against the member's own and parses back unchanged.""" + from pcapkit.protocols.internet.ip import IP + + for label, (fields, version) in FIELDS.items(): + with self.subTest(fields=label): + layer = IP(version=version, **fields) + self.assertIs(type(layer), self.member(version)) + self.assertEqual(layer.data, self.member(version)(**fields).data) + parsed = IP(layer.data) + self.assertIs(type(parsed), type(layer)) + self.assertEqual(parsed.data, layer.data) + + def test_version_picks_the_member_from_octets(self) -> None: + """A ``version`` that matches the octets parses them as the member would.""" + from pcapkit.protocols.internet.ip import IP + + for label, (octets, name) in OCTETS.items(): + for source in ('bytes', 'stream'): + with self.subTest(octets=label, source=source): + version = octets[0] >> 4 + layer = (IP(octets, version=version) if source == 'bytes' + else IP(io.BytesIO(octets), len(octets), version=version)) + self.assertIs(type(layer), self.member(name)) + self.assertEqual(layer.data, octets) + + def test_version_over_octets_of_the_other_member_is_its_parser_s_error(self) -> None: + """No cross-check: the member ``version`` names parses, and refuses the Version field.""" + from pcapkit.protocols.internet.ip import IP + from pcapkit.utilities.exceptions import ProtocolError + + cases = { + 'IPv4 octets, version=6': (OCTETS['IPv4 UDP'][0], 6, r'^\[IPv6\] invalid version: 4$'), + 'IPv6 octets, version=4': (OCTETS['IPv6 UDP'][0], 4, r'^\[IPv4\] invalid version: 6$'), + } + for label, (octets, version, message) in cases.items(): + with self.subTest(case=label): + with self.assertRaisesRegex(ProtocolError, message): + IP(octets, version=version) + + def test_a_member_checks_the_version_field_before_its_header(self) -> None: + """IPv6's first octet reads to IPv4 as an IHL of 0 to 4; the Version field is checked first.""" + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.utilities.exceptions import ProtocolError + + six = OCTETS['IPv6 UDP'][0] + for first in range(0x60, 0x65): + with self.subTest(first=hex(first)): + with self.assertRaisesRegex(ProtocolError, r'^\[IPv4\] invalid version: 6$'): + IPv4(bytes([first]) + six[1:]) + with self.assertRaisesRegex(ProtocolError, r'^\[IPv6\] invalid version: 4$'): + IPv6(OCTETS['IPv4 UDP'][0]) + + def test_fields_without_version_raise(self) -> None: + """Even a field only one member takes picks nothing.""" + from pcapkit.protocols.internet.ip import IP + from pcapkit.utilities.exceptions import ProtocolError + + cases = { + 'no fields': {}, + 'shared fields': {'src': '10.0.0.1', 'dst': '10.0.0.2'}, + 'IPv4 only': {'ttl': 3}, + 'IPv6 only': {'hop_limit': 3}, + 'flow label': {'flow_label': 1, 'src': '::1', 'dst': '::2'}, + 'both': {'ttl': 3, 'hop_limit': 3}, + } + for label, fields in cases.items(): + with self.subTest(fields=label): + with self.assertRaisesRegex(ProtocolError, NO_VERSION): + IP(**fields) + + def test_version_is_the_only_check(self) -> None: + """A bad ``version`` raises; a field the member lacks fails the member's check.""" + from pcapkit.protocols.internet.ip import IP + from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall + + for version in (5, True, '4', 4.0, 6.0): + with self.subTest(version=version): + with self.assertRaisesRegex(ProtocolError, f'^IP: invalid version: {version!r}$'): + IP(version=version) + + class Version(enum.IntEnum): + """An :class:`~enum.IntEnum` member is an :obj:`int`, and accepted.""" + + IPV6 = 6 + + self.assertIs(type(IP(version=Version.IPV6)), self.member(6)) + with self.assertRaisesRegex(UnsupportedCall, "^IPv4: unexpected keyword\\(s\\): 'hop_limit'"): + IP(version=4, hop_limit=3) + with self.assertRaisesRegex(UnsupportedCall, "^IPv6: unexpected keyword\\(s\\): 'ttl'"): + IP(version=6, ttl=3) + + def test_a_member_takes_version_only_as_its_own(self) -> None: + """Building, ``version`` must name the member; parsing, it is handed on unread.""" + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.utilities.exceptions import ProtocolError + + self.assertEqual(IPv4(version=4).data, IPv4().data) + self.assertEqual(IPv6(version=6).data, IPv6().data) + with self.assertRaisesRegex(ProtocolError, '^IPv4: version=6 names another member$'): + IPv4(version=6) + with self.assertRaisesRegex(ProtocolError, '^IPv6: version=4 names another member$'): + IPv6(version=4) + # NOTE: The enclosing IP layer hands its own version to the one it + # carries under this name, so IPv6 in IPv4 is parsed with ``version=4``. + six_in_four = ipv4(OCTETS['IPv6 UDP'][0], 41) + self.assertEqual(str(IPv4(six_in_four).protochain), 'IPv4:IPv6:UDP:Raw') + self.assertIs(type(IPv6(OCTETS['IPv6 UDP'][0], version=4)), IPv6) + + def test_a_subclass_outside_the_members_has_no_version_of_its_own(self) -> None: + """So it takes no ``version``, as a member takes no other's.""" + from pcapkit.protocols.internet.ip import IP + from pcapkit.utilities.exceptions import ProtocolError + + class Outside(IP): # type: ignore[type-arg] + """A concrete subclass of the base that is no member.""" + + name = length = None # type: ignore[assignment] + @classmethod + def __index__(cls) -> 'int': # type: ignore[override] + raise NotImplementedError + + def read(self, length: 'object' = None, **kwargs: 'object') -> 'object': + raise NotImplementedError + + def make(self, **kwargs: 'object') -> 'object': + raise NotImplementedError + + with self.assertRaisesRegex(ProtocolError, '^Outside: no version of its own to match version=4$'): + Outside(version=4) + + def test_from_data_on_the_base_gives_the_member(self) -> None: + from pcapkit.protocols.internet.ip import IP + from pcapkit.utilities.exceptions import ProtocolError + + for label, (octets, name) in OCTETS.items(): + parsed = self.member(name)(octets) + for form in ('info', 'dict'): + with self.subTest(octets=label, form=form): + data = parsed.info if form == 'info' else parsed.info.to_dict() + rebuilt = IP.from_data(data) + self.assertIs(type(rebuilt), self.member(name)) + self.assertEqual(rebuilt.data, octets) + with self.assertRaisesRegex(ProtocolError, '^IP: invalid version: None$'): + IP.from_data({}) + with self.assertRaisesRegex(ProtocolError, '^IPv4: version=6 names another member$'): + IP.from_data(self.member('IPv4')(OCTETS['IPv4 UDP'][0]).info, version=6) + + def test_only_the_base_dispatches(self) -> None: + """A subclass of the base is constructed as itself, abstract or not.""" + from pcapkit.protocols.data.internet.ipv4 import IPv4 as Data_IPv4 + from pcapkit.protocols.internet.ip import IP + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.schema.internet.ipv4 import IPv4 as Schema_IPv4 + + class Abstract(IP): # type: ignore[type-arg] # pylint: disable=abstract-method + """A direct subclass of the base, still abstract.""" + + class Custom(IPv4, schema=Schema_IPv4, data=Data_IPv4): + """A concrete subclass of the base, through a member.""" + + octets = OCTETS['IPv4 UDP'][0] + self.assertIs(type(IP(octets)), IPv4) + self.assertIs(type(Custom(octets)), Custom) + self.assertIs(type(Custom(ttl=3, version=4)), Custom) + self.assertIs(type(Custom.from_data(Custom(octets).info)), Custom) + with self.assertRaisesRegex(TypeError, 'abstract'): + Abstract(octets) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/link/test_l2tp_base_dispatch_2574_unit.py b/tests/protocols/link/test_l2tp_base_dispatch_2574_unit.py new file mode 100644 index 0000000000..7c383fdef2 --- /dev/null +++ b/tests/protocols/link/test_l2tp_base_dispatch_2574_unit.py @@ -0,0 +1,173 @@ +# -*- coding: utf-8 -*- +"""The L2TP family base dispatches on both paths. C.f. #2574. + +A versioned family's base never yields an instance of itself: parsing octets +through :class:`~pcapkit.protocols.link.l2tp.L2TP`, constructing it from fields, +and its ``from_data`` all give the member the version names. The ``version`` +field names it on either path, and it is the only thing that does from fields: +fields without it are refused. Octets without it are read by the version +nibble, bits 12 to 15 of the first word (RFC 2661, section 3.1), and data by its +``version``. ``2`` is ``L2TPv2``; no other member exists, so ``1`` (L2F) and +``3`` (L2TPv3) raise. + +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 struct +import unittest + +from tests._support import reimport_once_per_class + + +def l2tp_data(version: 'int' = 2) -> 'bytes': + """A data message, tunnel ``0x1234``, session ``0x5678``, with the nibble ``version``.""" + return struct.pack('!HHH', version, 0x1234, 0x5678) + b'\xff\x03\x00\x21PPP' + + +#: A control message (RFC 2661, section 3.1): T, L and S set, Length 12, Ns and Nr zero. +CONTROL = struct.pack('!HHHHHH', 0xC802, 12, 1, 0, 0, 0) + +#: Octets ``L2TPv2`` parses. +OCTETS = {'data message': l2tp_data(), 'control message': CONTROL} + +#: The refusal of fields given no ``version``. +NO_VERSION = '^L2TP: no version to build with; pass version=2$' + + +class L2TPBaseDispatchTests(unittest.TestCase): + """``L2TP(...)`` and ``L2TP.from_data`` give the member, never the base.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_octets_without_version_give_the_member_their_nibble_names(self) -> None: + from pcapkit.protocols.link.l2tp import L2TP + from pcapkit.protocols.link.l2tpv2 import L2TPv2 + + for label, octets in OCTETS.items(): + for source in ('bytes', 'stream'): + with self.subTest(octets=label, source=source): + layer = L2TP(octets) if source == 'bytes' else L2TP(io.BytesIO(octets), len(octets)) + self.assertIs(type(layer), L2TPv2) + self.assertIsInstance(layer, L2TP) + self.assertEqual(layer.info.version, octets[1] & 0x0F) + self.assertEqual(layer.data, octets) + self.assertEqual(str(layer.protochain), str(L2TPv2(octets).protochain)) + + def test_octets_no_member_reads_raise(self) -> None: + from pcapkit.protocols.link.l2tp import L2TP + from pcapkit.utilities.exceptions import ProtocolError + + for version in (1, 3, 15): + with self.subTest(version=version): + with self.assertRaisesRegex(ProtocolError, f'^L2TP: invalid version: {version}$'): + L2TP(l2tp_data(version)) + with self.assertRaisesRegex(ProtocolError, '^L2TP: no version in a packet shorter than 2 octets$'): + L2TP(b'\x00') + + def test_version_picks_the_member_from_octets(self) -> None: + from pcapkit.protocols.link.l2tp import L2TP + from pcapkit.protocols.link.l2tpv2 import L2TPv2 + + for label, octets in OCTETS.items(): + for source in ('bytes', 'stream'): + with self.subTest(octets=label, source=source): + layer = (L2TP(octets, version=2) if source == 'bytes' + else L2TP(io.BytesIO(octets), len(octets), version=2)) + self.assertIs(type(layer), L2TPv2) + self.assertEqual(layer.data, octets) + + def test_version_over_octets_of_another_version_is_the_parser_s_error(self) -> None: + """No cross-check: the member ``version`` names parses, and refuses the nibble.""" + from pcapkit.protocols.link.l2tp import L2TP + from pcapkit.utilities.exceptions import ProtocolError + + for nibble in (1, 3): + with self.subTest(nibble=nibble): + with self.assertRaisesRegex(ProtocolError, f'^L2TPv2: invalid version: {nibble}$'): + L2TP(l2tp_data(nibble), version=2) + with self.assertRaisesRegex(ProtocolError, '^L2TP: invalid version: 3$'): + L2TP(l2tp_data(), version=3) + + def test_version_picks_the_member_from_fields(self) -> None: + """Each build is byte-exact against the member's own and parses back unchanged.""" + from pcapkit.protocols.link.l2tp import L2TP + from pcapkit.protocols.link.l2tpv2 import L2TPv2 + + for fields in ({}, {'tunnel_id': 0x1234, 'session_id': 0x5678}): + with self.subTest(fields=fields): + layer = L2TP(version=2, **fields) + self.assertIs(type(layer), L2TPv2) + self.assertEqual(layer.data, L2TPv2(version=2, **fields).data) + self.assertEqual(layer.data, L2TPv2(**fields).data) + parsed = L2TP(layer.data) + self.assertIs(type(parsed), L2TPv2) + self.assertEqual(parsed.data, layer.data) + + def test_fields_without_version_raise(self) -> None: + """Even fields only L2TPv2 takes pick nothing.""" + from pcapkit.protocols.link.l2tp import L2TP + from pcapkit.utilities.exceptions import ProtocolError + + for fields in ({}, {'tunnel_id': 0x1234, 'session_id': 0x5678}): + with self.subTest(fields=fields): + with self.assertRaisesRegex(ProtocolError, NO_VERSION): + L2TP(**fields) + + def test_version_is_the_only_check(self) -> None: + """A version no member reads raises; a field the member lacks fails its check.""" + from pcapkit.protocols.link.l2tp import L2TP + from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall + + for version in (1, 3, True, '2', 2.0): + with self.subTest(version=version): + with self.assertRaisesRegex(ProtocolError, f'^L2TP: invalid version: {version!r}$'): + L2TP(version=version) + with self.assertRaisesRegex(UnsupportedCall, "^L2TPv2: unexpected keyword\\(s\\): 'ttl'"): + L2TP(version=2, ttl=3) + with self.assertRaisesRegex(ProtocolError, '^L2TP: invalid version: 3$'): + L2TP.from_data({'version': 3}) + with self.assertRaisesRegex(ProtocolError, '^L2TP: invalid version: None$'): + L2TP.from_data({}) + + def test_from_data_on_the_base_gives_the_member(self) -> None: + from pcapkit.protocols.link.l2tp import L2TP + from pcapkit.protocols.link.l2tpv2 import L2TPv2 + + for label, octets in OCTETS.items(): + parsed = L2TPv2(octets) + for form in ('info', 'dict'): + with self.subTest(octets=label, form=form): + data = parsed.info if form == 'info' else parsed.info.to_dict() + rebuilt = L2TP.from_data(data) + self.assertIs(type(rebuilt), L2TPv2) + self.assertEqual(rebuilt.data, octets) + + def test_only_the_base_dispatches(self) -> None: + """A subclass of the base is constructed as itself, abstract or not.""" + from pcapkit.protocols.data.link.l2tp import L2TP as Data_L2TP + from pcapkit.protocols.link.l2tp import L2TP + from pcapkit.protocols.link.l2tpv2 import L2TPv2 + from pcapkit.protocols.schema.link.l2tp import L2TP as Schema_L2TP + + class Abstract(L2TP): # type: ignore[type-arg] # pylint: disable=abstract-method + """A direct subclass of the base, still abstract.""" + + class Custom(L2TPv2, schema=Schema_L2TP, data=Data_L2TP): + """A concrete subclass of the base, through a member.""" + + octets = OCTETS['data message'] + self.assertIs(type(L2TP(octets)), L2TPv2) + self.assertIs(type(L2TP(version=2, tunnel_id=1)), L2TPv2) + self.assertIs(type(Custom(octets)), Custom) + self.assertIs(type(Custom(tunnel_id=1)), Custom) + self.assertIs(type(Custom.from_data(Custom(octets).info)), Custom) + with self.assertRaisesRegex(TypeError, 'abstract'): + Abstract(octets) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/link/test_linux_sll_base_dispatch_2574_unit.py b/tests/protocols/link/test_linux_sll_base_dispatch_2574_unit.py new file mode 100644 index 0000000000..27eed2ed8e --- /dev/null +++ b/tests/protocols/link/test_linux_sll_base_dispatch_2574_unit.py @@ -0,0 +1,207 @@ +# -*- coding: utf-8 -*- +"""The LinuxSLL family base dispatches where the version is known. C.f. #2574. + +A versioned family's base never yields an instance of itself. The ``version`` +keyword names the member :class:`~pcapkit.protocols.link.linux_sll.LinuxSLL` +gives, from octets or from fields, and nothing else does: neither header has a +version field -- the link type names it, ``LINUX_SLL`` (113) or ``LINUX_SLL2`` +(276), per tcpdump.org's pages for the two -- and no field picks a member, so +both are refused without it, ``interface_index`` included. ``from_data`` gives +the member whose data model it is handed, and refuses a ``dict``. + +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 struct +import unittest + +from tests._support import reimport_once_per_class + +#: An IPv4 header (DF clear, protocol UDP) and an 8-octet UDP header. +IPV4 = (bytes.fromhex('4500001c00010000401100000a0000010a000002') + + bytes.fromhex('3039003500080000')) + +#: A link-layer address, padded to the 8 octets both headers keep. +ADDR = bytes.fromhex('001122334455') + b'\x00\x00' + +#: Each version's header carrying ``IPV4`` from an Ethernet interface: v1 is +#: packet type, ARPHRD type, address length, address and protocol type; v2 is +#: protocol type, reserved, interface index, ARPHRD type, packet type, address +#: length and address. +OCTETS = { + 1: struct.pack('>HHH', 0, 1, 6) + ADDR + struct.pack('>H', 0x0800) + IPV4, + 2: struct.pack('>HHIHBB', 0x0800, 0, 3, 1, 0, 6) + ADDR + IPV4, +} + +#: Fields both members take, describing ``OCTETS``. +SHARED = {'packet_type': 0, 'arphrd_type': 1, 'link_addr': ADDR[:6], + 'protocol_type': 0x0800, 'payload': IPV4} + +#: The refusals given no ``version``. +NO_VERSION_OCTETS = ('^LinuxSLL: no version field; pass version=1 or version=2, ' + 'or parse with LinuxSLL1 or LinuxSLL2$') +NO_VERSION_FIELDS = '^LinuxSLL: no version to build with; pass version=1 or version=2$' + + +class LinuxSLLBaseDispatchTests(unittest.TestCase): + """``LinuxSLL(...)`` and ``LinuxSLL.from_data`` give the member, never the base.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + @staticmethod + def member(version: 'int') -> 'type': + from pcapkit.protocols.link.linux_sll1 import LinuxSLL1 + from pcapkit.protocols.link.linux_sll2 import LinuxSLL2 + + return {1: LinuxSLL1, 2: LinuxSLL2}[version] + + def test_octets_without_version_raise(self) -> None: + from pcapkit.protocols.link.linux_sll import LinuxSLL + from pcapkit.utilities.exceptions import ProtocolError + + for version, octets in OCTETS.items(): + for source in ('bytes', 'stream'): + with self.subTest(octets=version, source=source): + with self.assertRaisesRegex(ProtocolError, NO_VERSION_OCTETS): + if source == 'bytes': + LinuxSLL(octets) + else: + LinuxSLL(io.BytesIO(octets), len(octets)) + + def test_version_picks_the_member_from_octets(self) -> None: + """Even octets of the other version: the format carries none to object.""" + from pcapkit.protocols.link.linux_sll import LinuxSLL + + for version in (1, 2): + for octets_of, octets in OCTETS.items(): + for source in ('bytes', 'stream'): + with self.subTest(version=version, octets=octets_of, source=source): + layer = (LinuxSLL(octets, version=version) if source == 'bytes' + else LinuxSLL(io.BytesIO(octets), len(octets), version=version)) + self.assertIs(type(layer), self.member(version)) + self.assertIsInstance(layer, LinuxSLL) + self.assertEqual(str(layer.protochain), str(self.member(version)(octets).protochain)) + self.assertEqual(str(LinuxSLL(OCTETS[2], version=2).protochain), 'LINUX_SLL2:IPv4:UDP') + + def test_version_picks_the_member_from_fields(self) -> None: + """Each build is byte-exact against the member's own and parses back unchanged.""" + from pcapkit.protocols.link.linux_sll import LinuxSLL + + cases = { + 'LinuxSLL1 defaults': ({}, 1), + 'LinuxSLL1 shared fields': (SHARED, 1), + 'LinuxSLL2 defaults': ({}, 2), + 'LinuxSLL2 own fields': ({**SHARED, 'interface_index': 3, 'reserved': 0}, 2), + } + for label, (fields, version) in cases.items(): + with self.subTest(fields=label): + layer = LinuxSLL(version=version, **fields) + self.assertIs(type(layer), self.member(version)) + self.assertEqual(layer.data, self.member(version)(**fields).data) + parsed = LinuxSLL(layer.data, version=version) + self.assertEqual(parsed.data, layer.data) + self.assertEqual(LinuxSLL.from_data(parsed.info).data, layer.data) + self.assertEqual(LinuxSLL(version=1, **SHARED).data, OCTETS[1]) + self.assertEqual(LinuxSLL(version=2, **SHARED, interface_index=3).data, OCTETS[2]) + + def test_fields_without_version_raise(self) -> None: + """Even ``reserved`` or ``interface_index``, which only LinuxSLL2 takes, pick nothing.""" + from pcapkit.protocols.link.linux_sll import LinuxSLL + from pcapkit.utilities.exceptions import ProtocolError + + for label, fields in (('no fields', {}), ('shared fields', SHARED), + ('interface index', {**SHARED, 'interface_index': 3}), + ('reserved', {'reserved': 0})): + with self.subTest(fields=label): + with self.assertRaisesRegex(ProtocolError, NO_VERSION_FIELDS): + LinuxSLL(**fields) + + def test_version_is_the_only_check(self) -> None: + """A bad ``version`` raises; a field the member lacks fails the member's check.""" + from pcapkit.protocols.link.linux_sll import LinuxSLL + from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall + + for version in (0, 3, True, '2', 1.0, 2.0): + with self.subTest(version=version): + with self.assertRaisesRegex(ProtocolError, f'^LinuxSLL: invalid version: {version!r}$'): + LinuxSLL(version=version) + with self.assertRaisesRegex(UnsupportedCall, "^LinuxSLL1: unexpected keyword\\(s\\): 'interface_index'"): + LinuxSLL(version=1, interface_index=3) + + def test_a_member_takes_version_only_as_its_own(self) -> None: + """Building, ``version`` must name the member; parsing, it is handed on unread.""" + from pcapkit.protocols.link.linux_sll1 import LinuxSLL1 + from pcapkit.protocols.link.linux_sll2 import LinuxSLL2 + from pcapkit.utilities.exceptions import ProtocolError + + self.assertEqual(LinuxSLL1(version=1).data, LinuxSLL1().data) + self.assertEqual(LinuxSLL2(version=2).data, LinuxSLL2().data) + with self.assertRaisesRegex(ProtocolError, '^LinuxSLL1: version=2 names another member$'): + LinuxSLL1(version=2) + with self.assertRaisesRegex(ProtocolError, '^LinuxSLL2: version=1 names another member$'): + LinuxSLL2(version=1) + self.assertEqual(LinuxSLL2(OCTETS[2], version=1).data, OCTETS[2]) + + def test_a_subclass_outside_the_members_has_no_version_of_its_own(self) -> None: + """So it takes no ``version``, as a member takes no other's.""" + from pcapkit.protocols.link.linux_sll import LinuxSLL + from pcapkit.utilities.exceptions import ProtocolError + + class Outside(LinuxSLL): # type: ignore[type-arg] + """A concrete subclass of the base that is no member.""" + + name = length = None # type: ignore[assignment] + + def read(self, length: 'object' = None, **kwargs: 'object') -> 'object': + raise NotImplementedError + + def make(self, **kwargs: 'object') -> 'object': + raise NotImplementedError + + with self.assertRaisesRegex(ProtocolError, '^Outside: no version of its own to match version=2$'): + Outside(version=2) + + def test_from_data_on_the_base_gives_the_member(self) -> None: + """By the data model's class; a ``dict`` carries none, and is refused.""" + from pcapkit.protocols.link.linux_sll import LinuxSLL + from pcapkit.utilities.exceptions import ProtocolError + + for version, octets in OCTETS.items(): + with self.subTest(octets=version): + parsed = self.member(version)(octets) + rebuilt = LinuxSLL.from_data(parsed.info) + self.assertIs(type(rebuilt), self.member(version)) + self.assertEqual(rebuilt.data, octets) + self.assertEqual(self.member(version).from_data(parsed.info.to_dict()).data, octets) + with self.assertRaisesRegex( + ProtocolError, '^LinuxSLL: invalid data: must be a LinuxSLL1 or LinuxSLL2 data model$'): + LinuxSLL.from_data(parsed.info.to_dict()) + + def test_only_the_base_dispatches(self) -> None: + """A subclass of the base is constructed as itself, abstract or not.""" + from pcapkit.protocols.data.link.linux_sll2 import LinuxSLL2 as Data_LinuxSLL2 + from pcapkit.protocols.link.linux_sll import LinuxSLL + from pcapkit.protocols.link.linux_sll2 import LinuxSLL2 + from pcapkit.protocols.schema.link.linux_sll2 import LinuxSLL2 as Schema_LinuxSLL2 + + class Abstract(LinuxSLL): # type: ignore[type-arg] # pylint: disable=abstract-method + """A direct subclass of the base, still abstract.""" + + class Custom(LinuxSLL2, schema=Schema_LinuxSLL2, data=Data_LinuxSLL2): + """A concrete subclass of the base, through a member.""" + + octets = OCTETS[2] + self.assertIs(type(LinuxSLL(interface_index=3, version=2)), LinuxSLL2) + self.assertIs(type(Custom(octets)), Custom) + self.assertIs(type(Custom(packet_type=0, version=2)), Custom) + self.assertIs(type(Custom.from_data(Custom(octets).info)), Custom) + with self.assertRaisesRegex(TypeError, 'abstract'): + Abstract(octets) + + +if __name__ == '__main__': + unittest.main()