diff --git a/docs/source/contributing/pep.rst b/docs/source/contributing/pep.rst index 53e1cee330..eed7379040 100644 --- a/docs/source/contributing/pep.rst +++ b/docs/source/contributing/pep.rst @@ -889,9 +889,9 @@ from *wrong*, or it will cry wolf on the commonest capture there is — model for how to spell "expected, not an error". Sequenced for **wave 2 or 3**. The cryptographic half could be done sooner since -ESP has already laid the groundwork, but the checksum half needs the IP -pseudo-header class defined first, and that is worth doing deliberately rather -than as a side effect of a checksum patch. +ESP has already laid the groundwork. The checksum half builds on the IP +pseudo-header class, :class:`~pcapkit.protocols.misc.checksum.ip_pseudo.IP_Pseudo` +(:issue:`2471`). Deterministic Construction Arguments ------------------------------------ @@ -1277,12 +1277,14 @@ equally blocked: the cryptographic half (HIP's ``HIP_MAC``/``RVS_HMAC``/ ``RELAY_HMAC`` and its signature parameters) could start whenever, since :class:`~pcapkit.protocols.internet.esp.ESP` has already established both the keying-context channel and the report-rather-than-raise policy. The -one's-complement half (IPv4, TCP, UDP, ICMP/ICMPv6, OSPF) needs the IP -pseudo-header, and the owner has settled how: a **pseudo-protocol class** for the -IP family, serving both the parsing and the constructing path, rather than a -back-reference from a parsed protocol to its parent. So that half is not blocked -on an open design question -- only on someone defining that class, which is a -bounded piece of work and the natural first step. +one's-complement half (IPv4, TCP, UDP, ICMP/ICMPv6, OSPF) is :issue:`2470` and +:issue:`2472` to :issue:`2475`. + +**IP pseudo-header — done** (:issue:`2471`). +:class:`~pcapkit.protocols.misc.checksum.ip_pseudo.IP_Pseudo` is the +**pseudo-protocol class** for the IP family the owner settled on, serving both +the parsing and the constructing path, rather than a back-reference from a +parsed protocol to its parent. **Queued for wave 2 or 3 — deterministic construction arguments.** See `Deterministic Construction Arguments`_ above: dropping ``make()``'s implicit diff --git a/docs/source/pcapkit/protocols/index.rst b/docs/source/pcapkit/protocols/index.rst index 4244985bb9..d456dbbe90 100644 --- a/docs/source/pcapkit/protocols/index.rst +++ b/docs/source/pcapkit/protocols/index.rst @@ -38,6 +38,10 @@ from given information. The class hierarchy of :mod:`pcapkit.protocols`: PCAPNG end + subgraph checksum [Checksum Pseudo-Headers] + IP_Pseudo + end + Raw & NoPayload end @@ -112,7 +116,7 @@ from given information. The class hierarchy of :mod:`pcapkit.protocols`: B --> Link & Internet & Transport & Application - B --> Header & Frame & PCAPNG & Raw & NoPayload + B --> Header & Frame & PCAPNG & Raw & NoPayload & IP_Pseudo Raw --> FTP_DATA ARP --> RARP IPv6_Ext --> AH & ESP @@ -158,6 +162,7 @@ from given information. The class hierarchy of :mod:`pcapkit.protocols`: click IPv6-Frag "internet/ipv6_frag.html#pcapkit.protocols.internet.ipv6_frag.IPv6_Frag" click IPv6-Opts "internet/ipv6_opts.html#pcapkit.protocols.internet.ipv6_opts.IPv6_Opts" click IPv6-Route "internet/ipv6_route.html#pcapkit.protocols.internet.ipv6_route.IPv6_Route" + click IP_Pseudo "misc/checksum.html#pcapkit.protocols.misc.checksum.ip_pseudo.IP_Pseudo" click IPX "internet/ipx.html#pcapkit.protocols.internet.ipx.IPX" click MH "internet/mh.html#pcapkit.protocols.internet.mh.MH" diff --git a/docs/source/pcapkit/protocols/misc/checksum.rst b/docs/source/pcapkit/protocols/misc/checksum.rst new file mode 100644 index 0000000000..d580a05175 --- /dev/null +++ b/docs/source/pcapkit/protocols/misc/checksum.rst @@ -0,0 +1,85 @@ +IP Pseudo-Header +================ + +.. module:: pcapkit.protocols.misc.checksum.ip_pseudo + +:mod:`pcapkit.protocols.misc.checksum.ip_pseudo` contains +:class:`~pcapkit.protocols.misc.checksum.ip_pseudo.IP_Pseudo` only, +which implements the pseudo-header of the IP protocol family +that the upper-layer checksums cover, whose structure is +described as below: + +IPv4 (:rfc:`793#section-3.1`, :rfc:`768`): + +======= ========= ========================= =============================== +Octets Bits Name Description +======= ========= ========================= =============================== + 0 0 ``ip_pseudo.src`` Source Address + 4 32 ``ip_pseudo.dst`` Destination Address + 8 64 ``ip_pseudo.zero`` Reserved + 9 72 ``ip_pseudo.protocol`` Protocol + 10 80 ``ip_pseudo.len`` Upper-Layer Packet Length +======= ========= ========================= =============================== + +IPv6 (:rfc:`8200#section-8.1`): + +======= ========= ========================= =============================== +Octets Bits Name Description +======= ========= ========================= =============================== + 0 0 ``ip_pseudo.src`` Source Address + 16 128 ``ip_pseudo.dst`` Destination Address + 32 256 ``ip_pseudo.len`` Upper-Layer Packet Length + 36 288 ``ip_pseudo.zero`` Reserved + 39 312 ``ip_pseudo.protocol`` Next Header +======= ========= ========================= =============================== + +.. autoclass:: pcapkit.protocols.misc.checksum.ip_pseudo.IP_Pseudo + :no-members: + :show-inheritance: + + .. autoproperty:: name + .. autoproperty:: length + .. autoproperty:: protocol + .. autoproperty:: version + .. autoproperty:: src + .. autoproperty:: dst + + .. automethod:: read + .. automethod:: make + + .. automethod:: _make_data + .. automethod:: _source + .. automethod:: _destination + .. automethod:: _upper_protocol + .. automethod:: _upper_length + + .. automethod:: __post_init__ + .. automethod:: __index__ + +Header Schemas +-------------- + +.. module:: pcapkit.protocols.schema.misc.checksum.ip_pseudo + +.. autodata:: pcapkit.protocols.schema.misc.checksum.ip_pseudo.IPV4_PSEUDO_LEN +.. autodata:: pcapkit.protocols.schema.misc.checksum.ip_pseudo.IPV6_PSEUDO_LEN + +.. autoclass:: pcapkit.protocols.schema.misc.checksum.ip_pseudo.IP_Pseudo + :members: + :show-inheritance: + +Auxiliary Functions +~~~~~~~~~~~~~~~~~~~ + +.. autofunction:: pcapkit.protocols.schema.misc.checksum.ip_pseudo.is_ipv6 +.. autofunction:: pcapkit.protocols.schema.misc.checksum.ip_pseudo.ip_pseudo_address_selector +.. autofunction:: pcapkit.protocols.schema.misc.checksum.ip_pseudo.ip_pseudo_zero_selector + +Data Models +----------- + +.. module:: pcapkit.protocols.data.misc.checksum.ip_pseudo + +.. autoclass:: pcapkit.protocols.data.misc.checksum.ip_pseudo.IP_Pseudo + :members: + :show-inheritance: diff --git a/docs/source/pcapkit/protocols/misc/index.rst b/docs/source/pcapkit/protocols/misc/index.rst index 1abf5ea54f..37a0d28219 100644 --- a/docs/source/pcapkit/protocols/misc/index.rst +++ b/docs/source/pcapkit/protocols/misc/index.rst @@ -19,3 +19,4 @@ empty payload. pcapng raw null + checksum diff --git a/pcapkit/__init__.py b/pcapkit/__init__.py index 30b0a2036b..8aedcdfbda 100644 --- a/pcapkit/__init__.py +++ b/pcapkit/__init__.py @@ -119,6 +119,7 @@ 'HIP', 'HOPOPT', 'IPv6_Frag', 'IPv6_Ext', 'IPv6_Opts', 'IPv6_Route', 'MH', # IPv6 Extension Header + 'IP_Pseudo', # IP Pseudo-Header 'TCP', 'UDP', 'SCTP', # Transport Layer diff --git a/pcapkit/all.py b/pcapkit/all.py index e56b24c991..7597eaf59f 100644 --- a/pcapkit/all.py +++ b/pcapkit/all.py @@ -128,6 +128,7 @@ 'HIP', 'HOPOPT', 'IPv6_Ext', 'IPv6_Frag', 'IPv6_Opts', 'IPv6_Route', 'MH', # IPv6 Extension Header + 'IP_Pseudo', # IP Pseudo-Header 'TCP', 'UDP', 'SCTP', # Transport Layer 'FTP', 'FTP_DATA', # Application Layer 'HTTP', 'HTTPv1', 'HTTPv2', 'NGAP', 'OSPF', 'RARP', 'DRARP', diff --git a/pcapkit/protocols/__init__.py b/pcapkit/protocols/__init__.py index 3121984b73..a9e2440646 100644 --- a/pcapkit/protocols/__init__.py +++ b/pcapkit/protocols/__init__.py @@ -57,6 +57,9 @@ 'AH', 'ESP', 'ICMP', 'ICMPv4', 'ICMPv6', 'IGMP', 'IGMPv1', 'IGMPv2', 'IGMPv3', 'IP', 'IPsec', 'IPv4', 'IPv6', 'IPX', + # IP Pseudo-Header + 'IP_Pseudo', + # IPv6 Extension Header 'HIP', 'HOPOPT', 'IPv6_Frag', 'IPv6_Ext', 'IPv6_Opts', 'IPv6_Route', 'MH', diff --git a/pcapkit/protocols/data/__init__.py b/pcapkit/protocols/data/__init__.py index fe22251fbd..844f2c2db8 100644 --- a/pcapkit/protocols/data/__init__.py +++ b/pcapkit/protocols/data/__init__.py @@ -115,6 +115,9 @@ 'HOPOPT_LineIdentificationOption', 'HOPOPT_JumboPayloadOption', 'HOPOPT_HomeAddressOption', 'HOPOPT_IPDFFOption', + # IP Pseudo-Header + 'IP_Pseudo', + # Internet Control Message Protocol 'ICMPv4', 'ICMPv4_Unknown', 'ICMPv4_Echo', 'ICMPv4_DestinationUnreachable', 'ICMPv4_TimeExceeded', diff --git a/pcapkit/protocols/data/misc/__init__.py b/pcapkit/protocols/data/misc/__init__.py index d85291df34..89d3ca843e 100644 --- a/pcapkit/protocols/data/misc/__init__.py +++ b/pcapkit/protocols/data/misc/__init__.py @@ -76,6 +76,7 @@ # misc protocols from pcapkit.protocols.data.misc.raw import Raw from pcapkit.protocols.data.misc.null import NoPayload +from pcapkit.protocols.data.misc.checksum import IP_Pseudo __all__ = [ # PCAP file headers @@ -105,6 +106,7 @@ 'PCAPNG_CustomBlock', 'PCAPNG_PacketBlock', # No Payload + 'IP_Pseudo', 'NoPayload', # Raw Packet diff --git a/pcapkit/protocols/data/misc/checksum/__init__.py b/pcapkit/protocols/data/misc/checksum/__init__.py new file mode 100644 index 0000000000..f9ef903c3d --- /dev/null +++ b/pcapkit/protocols/data/misc/checksum/__init__.py @@ -0,0 +1,6 @@ +# -*- coding: utf-8 -*- +"""data models for checksum pseudo-headers""" + +from pcapkit.protocols.data.misc.checksum.ip_pseudo import IP_Pseudo + +__all__ = ['IP_Pseudo'] diff --git a/pcapkit/protocols/data/misc/checksum/ip_pseudo.py b/pcapkit/protocols/data/misc/checksum/ip_pseudo.py new file mode 100644 index 0000000000..1c05e23cc7 --- /dev/null +++ b/pcapkit/protocols/data/misc/checksum/ip_pseudo.py @@ -0,0 +1,41 @@ +# -*- coding: utf-8 -*- +"""data model for the IP pseudo-header""" + +from typing import TYPE_CHECKING + +from pcapkit.corekit.infoclass import info_final +from pcapkit.protocols.data.protocol import Protocol + +if TYPE_CHECKING: + from ipaddress import IPv4Address, IPv6Address + + from typing_extensions import Literal + + from pcapkit.const.reg.transtype import TransType + +__all__ = ['IP_Pseudo'] + + +@info_final +class IP_Pseudo(Protocol): + """Data model for the IP pseudo-header. + + .. versionadded:: 1.6.0 + + """ + + #: IP version of the enclosing layer. + version: 'Literal[4, 6]' + #: Source address. + src: 'IPv4Address | IPv6Address' + #: Destination address. + dst: 'IPv4Address | IPv6Address' + #: Upper-layer protocol number. + protocol: 'TransType' + #: Upper-layer packet length, header included. + len: 'int' + #: Reserved octets, as parsed. + zero: 'bytes' + + if TYPE_CHECKING: + def __init__(self, version: 'Literal[4, 6]', src: 'IPv4Address | IPv6Address', dst: 'IPv4Address | IPv6Address', protocol: 'TransType', len: 'int', zero: 'bytes') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,redefined-builtin,line-too-long diff --git a/pcapkit/protocols/misc/__init__.py b/pcapkit/protocols/misc/__init__.py index ae44a05fb4..162cd894bf 100644 --- a/pcapkit/protocols/misc/__init__.py +++ b/pcapkit/protocols/misc/__init__.py @@ -8,7 +8,8 @@ :mod:`pcapkit.protocols.misc` contains the auxiliary protocol implementations. These include the :class:`~pcapkit.protocols.misc.raw.Raw` class for not-supported protocols, the :class:`~pcapkit.protocols.misc.null.NoPayload` class for -indication of empty payload, and the PCAP and PCAP-NG file format classes. +indication of empty payload, the PCAP and PCAP-NG file format classes, and the +checksum pseudo-headers in :mod:`pcapkit.protocols.misc.checksum`. """ @@ -22,6 +23,9 @@ from pcapkit.protocols.misc.raw import Raw from pcapkit.protocols.misc.null import NoPayload +# Checksum Pseudo-Headers +from pcapkit.protocols.misc.checksum import IP_Pseudo + __all__ = [ # PCAP Headers 'Header', 'Frame', @@ -34,4 +38,7 @@ # Raw Packet 'Raw', + + # Checksum Pseudo-Headers + 'IP_Pseudo', ] diff --git a/pcapkit/protocols/misc/checksum/__init__.py b/pcapkit/protocols/misc/checksum/__init__.py new file mode 100644 index 0000000000..47b0fce9d2 --- /dev/null +++ b/pcapkit/protocols/misc/checksum/__init__.py @@ -0,0 +1,14 @@ +# -*- coding: utf-8 -*- +"""Checksum Pseudo-Headers +============================= + +.. module:: pcapkit.protocols.misc.checksum + +:mod:`pcapkit.protocols.misc.checksum` contains the pseudo-header protocols +that upper-layer checksums cover, currently +:class:`~pcapkit.protocols.misc.checksum.ip_pseudo.IP_Pseudo` only. + +""" +from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + +__all__ = ['IP_Pseudo'] diff --git a/pcapkit/protocols/misc/checksum/ip_pseudo.py b/pcapkit/protocols/misc/checksum/ip_pseudo.py new file mode 100644 index 0000000000..9a7defa403 --- /dev/null +++ b/pcapkit/protocols/misc/checksum/ip_pseudo.py @@ -0,0 +1,631 @@ +# -*- coding: utf-8 -*- +"""IP Pseudo-Header +====================== + +.. module:: pcapkit.protocols.misc.checksum.ip_pseudo + +:mod:`pcapkit.protocols.misc.checksum.ip_pseudo` contains +:class:`~pcapkit.protocols.misc.checksum.ip_pseudo.IP_Pseudo` only, +which implements the pseudo-header of the IP protocol family +that the upper-layer checksums cover, whose structure is +described as below: + +IPv4 (:rfc:`793#section-3.1`, :rfc:`768`): + +======= ========= ========================= =============================== +Octets Bits Name Description +======= ========= ========================= =============================== + 0 0 ``ip_pseudo.src`` Source Address + 4 32 ``ip_pseudo.dst`` Destination Address + 8 64 ``ip_pseudo.zero`` Reserved + 9 72 ``ip_pseudo.protocol`` Protocol + 10 80 ``ip_pseudo.len`` Upper-Layer Packet Length +======= ========= ========================= =============================== + +IPv6 (:rfc:`8200#section-8.1`): + +======= ========= ========================= =============================== +Octets Bits Name Description +======= ========= ========================= =============================== + 0 0 ``ip_pseudo.src`` Source Address + 16 128 ``ip_pseudo.dst`` Destination Address + 32 256 ``ip_pseudo.len`` Upper-Layer Packet Length + 36 288 ``ip_pseudo.zero`` Reserved + 39 312 ``ip_pseudo.protocol`` Next Header +======= ========= ========================= =============================== + +""" +import ipaddress +from typing import TYPE_CHECKING, overload + +from pcapkit.const.ipv4.option_number import OptionNumber as Enum_IPv4Option +from pcapkit.const.ipv6.extension_header import ExtensionHeader as Enum_ExtensionHeader +from pcapkit.const.ipv6.option import Option as Enum_Option +from pcapkit.const.reg.transtype import TransType as Enum_TransType +from pcapkit.corekit.fields.ipaddress import parse_ip_address +from pcapkit.corekit.protochain import ProtoChain +from pcapkit.protocols.data.misc.checksum.ip_pseudo import IP_Pseudo as Data_IP_Pseudo +from pcapkit.protocols.internet.ipv4 import IPv4 +from pcapkit.protocols.internet.ipv6 import IPV6_HEADER_LEN, IPv6 +from pcapkit.protocols.misc.null import NoPayload +from pcapkit.protocols.protocol import ProtocolBase +from pcapkit.protocols.schema.internet.ipv6 import IP_MAX_LENGTH, jumbo_payload_length +from pcapkit.protocols.schema.misc.checksum.ip_pseudo import IPV4_PSEUDO_LEN, IPV6_PSEUDO_LEN +from pcapkit.protocols.schema.misc.checksum.ip_pseudo import IP_Pseudo as Schema_IP_Pseudo +from pcapkit.utilities.exceptions import FieldValueError, ProtocolError, UnsupportedCall + +if TYPE_CHECKING: + from enum import IntEnum as StdlibEnum + from ipaddress import IPv4Address, IPv6Address + from typing import IO, Any, NoReturn, Optional, Type + + from aenum import IntEnum as AenumEnum + from typing_extensions import Literal + +__all__ = ['IP_Pseudo'] + +#: Extension headers that end the chain, as the upper layer: the Mobility Header +#: (:rfc:`6275#section-6.1.1`) and HIP (:rfc:`7401#section-5.1.1`). +_TERMINAL_HEADERS = (Enum_ExtensionHeader.Mobility_Header, Enum_ExtensionHeader.HIP) + + +class IP_Pseudo(ProtocolBase[Data_IP_Pseudo, Schema_IP_Pseudo], + schema=Schema_IP_Pseudo, data=Data_IP_Pseudo): + """This class implements the IP pseudo-header. + + It is not a layer on the wire: it is built from the enclosing + :class:`~pcapkit.protocols.internet.ipv4.IPv4` or + :class:`~pcapkit.protocols.internet.ipv6.IPv6` layer, or from explicit + values, and its :attr:`data` is the octets an upper-layer checksum covers + ahead of that layer's own. The layout follows the IP version of the source + address, or the length of the octets parsed. + + .. versionadded:: 1.6.0 + + """ + + ########################################################################## + # Properties. + ########################################################################## + + @property + def name(self) -> 'Literal["IP Pseudo-Header"]': + """Name of current protocol.""" + return 'IP Pseudo-Header' + + @property + def length(self) -> 'Literal[12, 40]': + """Header length of current protocol.""" + return IPV6_PSEUDO_LEN if self._is_ipv6(self._info.src) else IPV4_PSEUDO_LEN # type: ignore[return-value] + + @property + def protocol(self) -> 'Enum_TransType': + """Upper-layer protocol.""" + return self._info.protocol + + @property + def version(self) -> 'Literal[4, 6]': + """IP version of the enclosing layer.""" + return self._info.version + + @property + def src(self) -> 'IPv4Address | IPv6Address': + """Source address.""" + return self._info.src + + @property + def dst(self) -> 'IPv4Address | IPv6Address': + """Destination address.""" + return self._info.dst + + ########################################################################## + # Methods. + ########################################################################## + + def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_IP_Pseudo': # pylint: disable=unused-argument + """Read IP pseudo-header. + + Structure of the IPv4 pseudo-header [:rfc:`793`]: + + .. code-block:: text + + +--------+--------+--------+--------+ + | Source Address | + +--------+--------+--------+--------+ + | Destination Address | + +--------+--------+--------+--------+ + | zero | PTCL | TCP Length | + +--------+--------+--------+--------+ + + Structure of the IPv6 pseudo-header [:rfc:`8200`]: + + .. code-block:: text + + +--------+--------+--------+--------+ + | | + + + + | | + + Source Address + + | | + + + + | | + +--------+--------+--------+--------+ + | | + + + + | | + + Destination Address + + | | + + + + | | + +--------+--------+--------+--------+ + | Upper-Layer Packet Length | + +--------+--------+--------+--------+ + | zero | Next | + +--------+--------+--------+--------+ + + Args: + length: Length of packet data. + **kwargs: Arbitrary keyword arguments. + + Returns: + Parsed packet data. + + """ + schema = self.__header__ + + version: 'Literal[4, 6]' = schema.src.version + return Data_IP_Pseudo( + version=version, + src=schema.src, + dst=schema.dst, + protocol=Enum_TransType.get(schema.protocol), + len=schema.ipv6_len if self._is_ipv6(schema.src) else schema.ipv4_len, + zero=schema.zero, + ) + + def make(self, + ip: 'Optional[IPv4 | IPv6]' = None, + src: 'Optional[IPv4Address | IPv6Address | str | int | bytes]' = None, + dst: 'Optional[IPv4Address | IPv6Address | str | int | bytes]' = None, + protocol: 'Optional[Enum_TransType | StdlibEnum | AenumEnum | int | str]' = None, + protocol_default: 'Optional[int]' = None, + protocol_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long + protocol_reversed: 'bool' = False, + upper_length: 'Optional[int]' = None, + zero: 'bytes' = b'', + **kwargs: 'Any') -> 'Schema_IP_Pseudo': + """Make (construct) packet data. + + Values not given are read from ``ip``. ``src`` is a Home Address option's + address (:rfc:`6275#section-11.3.1`); ``dst`` is the final destination of a + Routing header with segments left, or of a source route (:rfc:`791`) not + yet exhausted; ``protocol`` and ``upper_length`` lie past the extension + headers, a UDP upper layer giving its Length field (:rfc:`768`) and a + jumbogram its Jumbo Payload Length (:rfc:`2675`). Values ``ip`` lacks are + to be given: an unparsed Routing destination (:rfc:`8754`), a fragment's + length, or an upper layer behind an unparsed header such as ESP. + + Args: + ip: Parsed or constructed IP layer enclosing the upper layer. + src: Source address; its IP version selects the layout. + dst: Destination address, of the same IP version as ``src``. + protocol: Upper-layer protocol number. + protocol_default: Default value of ``protocol``. + protocol_namespace: Namespace of ``protocol``; + :class:`~pcapkit.const.reg.transtype.TransType` by default. + protocol_reversed: Whether to reverse the namespace of ``protocol``. + upper_length: Upper-layer packet length, header included. + zero: Reserved octets, truncated or zero-filled to the layout's width. + **kwargs: Arbitrary keyword arguments. + + Returns: + Constructed packet data. + + Raises: + ProtocolError: If ``ip`` is neither IPv4 nor IPv6, a value is given + by neither the arguments nor ``ip``, or a header of ``ip`` is + malformed or not parsed. + FieldValueError: If ``src`` or ``dst`` is not a valid address, + they differ in IP version, or ``upper_length`` does not fit + the layout's length field. + + """ + if ip is not None: + if not isinstance(ip, (IPv4, IPv6)): + raise ProtocolError(f'{self.alias}: unsupported enclosing layer ' + f'{type(ip).__name__}') + if src is None: + src = self._source(ip) + if dst is None: + dst = self._destination(ip) + from_layer = protocol is None + if from_layer: + protocol = self._upper_protocol(ip) + if upper_length is None: + upper_length = self._upper_length(ip, from_layer=from_layer) + + if src is None or dst is None or protocol is None or upper_length is None: + raise ProtocolError(f'{self.alias}: src, dst, protocol and upper_length ' + 'are required without ip') + + src_val = parse_ip_address(src, f'{self.alias}: source address') + dst_val = parse_ip_address(dst, f'{self.alias}: destination address', src_val.version) + is_v6 = self._is_ipv6(src_val) + limit = 0xFFFFFFFF if is_v6 else IP_MAX_LENGTH + if not 0 <= upper_length <= limit: + raise FieldValueError(f'{self.alias}: upper-layer length {upper_length} ' + f'out of range 0 to {limit:#x}') + if protocol_namespace is None: + protocol_namespace = Enum_TransType + proto = self._make_index(protocol, protocol_default, namespace=protocol_namespace, + reversed=protocol_reversed, pack=False) + + # NOTE: Kept at the layout's width, so that a constructed data model + # matches the parsed one. + width = 3 if is_v6 else 1 + zero = zero[:width].ljust(width, b'\x00') + + if is_v6: + return Schema_IP_Pseudo( + src=src_val, + dst=dst_val, + ipv6_len=upper_length, + zero=zero, + protocol=proto, + ) + return Schema_IP_Pseudo( + src=src_val, + dst=dst_val, + zero=zero, + protocol=proto, + ipv4_len=upper_length, + ) + + ########################################################################## + # Data models. + ########################################################################## + + @overload + def __post_init__(self, file: 'IO[bytes] | bytes', length: 'Optional[int]' = ..., **kwargs: 'Any') -> 'None': ... + @overload + def __post_init__(self, **kwargs: 'Any') -> 'None': ... # pylint: disable=arguments-differ + + def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None, + length: 'Optional[int]' = None, **kwargs: 'Any') -> 'None': + """Post initialisation hook. + + Args: + file: Source packet stream. + length: Length of packet data. + **kwargs: Arbitrary keyword arguments. + + Raises: + ProtocolError: If the octets parsed are not ``length`` long, or + not 12 (IPv4) or 40 (IPv6) long. + + See Also: + For construction arguments, please refer to :meth:`make`. + + """ + if file is not None: + file = file if isinstance(file, bytes) else file.read(length) # type: ignore[arg-type] + if length is None: + length = len(file) + if len(file) != length: + raise ProtocolError(f'{self.alias}: length {length} disagrees with ' + f'{len(file)} octets') + if length not in (IPV4_PSEUDO_LEN, IPV6_PSEUDO_LEN): + raise ProtocolError(f'{self.alias}: invalid length {length}; ' + f'expected {IPV4_PSEUDO_LEN} or {IPV6_PSEUDO_LEN}') + super().__post_init__(file, length, **kwargs) # type: ignore[arg-type] + + #: pcapkit.protocols.misc.null.NoPayload: Next layer (no payload). + self._next = NoPayload() + #: pcapkit.corekit.protochain.ProtoChain: Protocol chain from current layer. + self._protos = ProtoChain(self.__class__, self.alias) + + def __length_hint__(self) -> 'Literal[12]': + """Return an estimated length for the object.""" + return IPV4_PSEUDO_LEN # type: ignore[return-value] + + @classmethod + def __index__(cls) -> 'NoReturn': + """Numeral registry index of the protocol. + + Raises: + UnsupportedCall: This protocol has no registry entry. + + """ + raise UnsupportedCall(f'{cls.__name__!r} object cannot be interpreted as an integer') + + ########################################################################## + # Utilities. + ########################################################################## + + @classmethod + def _make_data(cls, data: 'Data_IP_Pseudo') -> 'dict[str, Any]': # type: ignore[override] + """Create key-value pairs from ``data`` for protocol construction. + + Args: + data: protocol data + + Returns: + Key-value pairs for protocol construction. + + """ + return { + 'src': data.src, + 'dst': data.dst, + 'protocol': data.protocol, + 'upper_length': data.len, + 'zero': data.zero, + } + + def _source(self, ip: 'IPv4 | IPv6') -> 'IPv4Address | IPv6Address': + """Read the pseudo-header source address from an IP layer. + + Args: + ip: IPv4 or IPv6 layer. + + Returns: + The address of a Home Address option (:rfc:`6275#section-11.3.1`) + if any, the source address otherwise. + + Raises: + ProtocolError: If a Destination Options header is not parsed. + + """ + src = ip.info.src + if isinstance(ip, IPv6): + for header in ip.extension_headers.getlist(Enum_ExtensionHeader.IPv6_Opts): # type: ignore[arg-type] + options = header.info.get('options') + if options is None: + raise ProtocolError(f'{self.alias}: Destination Options of {ip.alias} not ' + 'parsed; give src') + option = options.get(Enum_Option.Home_Address) + if option is not None: + src = option.address + + # If a malformed Destination Options header stopped the walk, it becomes the upper layer as Raw (#2480). + # Only raise an error if the Destination Options header is the next protocol but not in extension_headers. + if (ip.info.next == Enum_ExtensionHeader.IPv6_Opts and + not ip.extension_headers.getlist(Enum_ExtensionHeader.IPv6_Opts) and + hasattr(ip, 'payload') and ip.payload is not None): + from pcapkit.protocols.misc.raw import Raw + if type(ip.payload) is Raw: + raise ProtocolError(f'{self.alias}: Destination Options of {ip.alias} not ' + 'parsed; give src') + + return src + + def _destination(self, ip: 'IPv4 | IPv6') -> 'IPv4Address | IPv6Address': + """Read the pseudo-header destination address from an IP layer. + + Args: + ip: IPv4 or IPv6 layer. + + Returns: + The last address still to be visited of a source route + (:rfc:`791`) not yet exhausted, or of a Routing header with + segments left; the destination address otherwise. + + Raises: + ProtocolError: If a source route or Routing header is not parsed, + or a Routing header with segments left has no parsed last + address. + + """ + dst = ip.info.dst + if isinstance(ip, IPv4): + # NOTE: The header holds the next hop while the route has addresses + # left; once exhausted (RFC 791 section 3.1) it is the destination. + options = ip.info.get('options') + if options is None: + return dst + for code in (Enum_IPv4Option.LSR, Enum_IPv4Option.SSR): + for option in options.getlist(code): + if option.remaining: + dst = option.remaining[-1] + return dst + for header in ip.extension_headers.getlist(Enum_ExtensionHeader.IPv6_Route): # type: ignore[arg-type] + route = header.info + seg_left = route.get('seg_left') + if seg_left is None or route.get('__short_read__') is not None: + raise ProtocolError(f'{self.alias}: Routing header of {ip.alias} not parsed; give dst') + if not seg_left: + continue + # NOTE: Type 2 carries one address, the source routes a tuple. A + # truncated RPL address stays bytes; unparsed types carry none. + final = route.get('ip') + if isinstance(final, tuple) and final: + final = final[-1] + if not isinstance(final, ipaddress.IPv6Address): + raise ProtocolError(f'{self.alias}: final destination of Routing type ' + f'{route.type} not parsed; give dst') + dst = final + + # If a malformed Routing header stopped the walk, it becomes the upper layer as Raw (#2480). + # Only raise an error if the Routing header is the next protocol but not in extension_headers. + if (ip.info.next == Enum_ExtensionHeader.IPv6_Route and + not ip.extension_headers.getlist(Enum_ExtensionHeader.IPv6_Route) and + hasattr(ip, 'payload') and ip.payload is not None): + from pcapkit.protocols.misc.raw import Raw + if type(ip.payload) is Raw: + raise ProtocolError(f'{self.alias}: Routing header of {ip.alias} not parsed; give dst') + return dst + + def _upper_protocol(self, ip: 'IPv4 | IPv6') -> 'Enum_TransType': + """Read the upper-layer protocol from an IP layer. + + Args: + ip: IPv4 or IPv6 layer. + + Returns: + The protocol past the extension headers. + + Raises: + ProtocolError: If the walk ended at an unparsed header: none, or + an IPv6 extension header that was not walked, as past the + header limit, or a truncated Routing or Fragment header. + + """ + return self._upper(ip)[0] + + def _upper(self, ip: 'IPv4 | IPv6') -> 'tuple[Enum_TransType, int]': + """Read the upper-layer protocol and the octets of the headers ahead of it. + + A Mobility Header or HIP (:rfc:`6275#section-6.1.1`, :rfc:`7401#section-5.1.1`) + is the upper layer itself. An Authentication Header (:rfc:`4302#section-2.2`) + is walked past, its Next Header naming the upper layer, as in IPv6. + + Args: + ip: IPv4 or IPv6 layer. + + Returns: + The protocol past the extension headers, and the octets of those + headers ahead of it. + + Raises: + ProtocolError: If the walk ended at an unparsed header: none, an + IPv6 extension header that was not walked, an Authentication + Header cut short or in a non-first IPv4 fragment. An Encapsulating Security Payload + (:rfc:`4303`) is never walked, its upper layer being encrypted. + + """ + if isinstance(ip, IPv4): + code, before, offset = ip.info.protocol, 0, ip.info.hdr_len + # NOTE: A non-first fragment's payload does not start with the AH its Protocol names. + while code == Enum_TransType.AH and not ip.info.offset: + # NOTE: Payload Len counts 32-bit words, less two (RFC 4302 section 2.2). + octets = ip.data[offset:offset + 2] + size = (octets[1] + 2) * 4 if len(octets) == 2 else 0 + if not size or offset + size > len(ip.data): + raise ProtocolError(f'{self.alias}: Authentication Header of {ip.alias} ' + 'not parsed; give protocol and upper_length') + code = Enum_TransType.get(ip.data[offset]) + before, offset = before + size, offset + size + # NOTE: The upper layer of an ESP (RFC 4303) is encrypted, so it is never walked. + unparsed = code in (Enum_TransType.AH, Enum_TransType.ESP) + else: + code, before = ip.info.next, 0 + for _, header in ip.extension_headers.items(multi=True): + # NOTE: An ESP header is never walked, its upper layer being encrypted (RFC 4303). + if code in _TERMINAL_HEADERS or code == Enum_TransType.ESP: + break + before += header.length + code = header.info.next + unparsed = code not in _TERMINAL_HEADERS and self._is_extension_header(code) + + # If a malformed extension header stopped the walk, it becomes the upper layer as Raw (#2480). + # Provide a more specific error message identifying which extension header failed. + if unparsed: + from pcapkit.protocols.misc.raw import Raw + + # Check if the upper layer is Raw, indicating an unparsed extension header. + # Only raise a specific error if this extension header is NOT in the parsed headers. + if hasattr(ip, 'payload') and type(ip.payload) is Raw: + # Try to identify which extension header failed. ``ProtocolError`` + # subclasses ``ValueError``, so only the enum lookup itself -- not + # the raises below -- may sit inside this ``except ValueError``. + try: + ext_header = Enum_ExtensionHeader(code) # type: 'Optional[Enum_ExtensionHeader]' + except ValueError: + ext_header = None + if ext_header is not None: + # Check if this header type was successfully parsed by looking in extension_headers + ext_list = ip.extension_headers.getlist(ext_header) + # If the header is not in the parsed list, it failed to parse + if not ext_list: + if ext_header == Enum_ExtensionHeader.IPv6_Opts: + raise ProtocolError(f'{self.alias}: Destination Options of {ip.alias} not ' + 'parsed; give src') + elif ext_header == Enum_ExtensionHeader.IPv6_Route: + raise ProtocolError(f'{self.alias}: Routing header of {ip.alias} not parsed; give dst') + # Fall through to generic error message if not a recognized case + + if code is None or unparsed: + raise ProtocolError(f'{self.alias}: upper layer of {ip.alias} not parsed; ' + 'give protocol and upper_length') + return code, before + + @staticmethod + def _is_ipv6(address: 'IPv4Address | IPv6Address') -> 'bool': + """Tell whether ``address`` selects the IPv6 layout.""" + return isinstance(address, ipaddress.IPv6Address) + + @staticmethod + def _is_extension_header(code: 'int') -> 'bool': + """Tell whether ``code`` is an IPv6 extension header number.""" + try: + Enum_ExtensionHeader(code) + except ValueError: + return False + return True + + def _upper_length(self, ip: 'IPv4 | IPv6', *, from_layer: 'bool' = True) -> 'int': + """Read the upper-layer packet length from an IP layer. + + Args: + ip: IPv4 or IPv6 layer. + from_layer: Whether the upper layer is the one ``ip`` carries, + rather than one given by the caller. + + Returns: + The Length field of a UDP header that is present, else the length + past the IP header and extension headers. + + Raises: + ProtocolError: If ``ip`` is a fragment, ends at an unparsed header, + or has a zero Payload Length and no Jumbo Payload option. + + """ + if isinstance(ip, IPv4): + info4 = ip.info + if info4.offset or info4.flags.mf: + raise ProtocolError(f'{self.alias}: {ip.alias} fragment; give upper_length') + protocol, before = self._upper(ip) + return self._udp_length(ip, protocol, info4.hdr_len + before, + info4.len - info4.hdr_len - before, from_layer) + + protocol, before = self._upper(ip) + info = ip.info + for frag in ip.extension_headers.getlist(Enum_ExtensionHeader.IPv6_Frag): # type: ignore[arg-type] + if frag.info.offset or frag.info.mf: + raise ProtocolError(f'{self.alias}: {ip.alias} fragment; give upper_length') + + extent = info.payload + if not extent: + jumbo = jumbo_payload_length(info.next, ip.data[IPV6_HEADER_LEN:]) + if jumbo is None: + raise ProtocolError(f'{self.alias}: Payload Length 0 and no Jumbo Payload ' + 'option; give upper_length') + extent = jumbo + # NOTE: ``info.hdr_len`` ends at the first Fragment header, so headers + # after it are not counted; the fixed header plus ``before`` is the offset. + return self._udp_length(ip, protocol, IPV6_HEADER_LEN + before, extent - before, from_layer) + + def _udp_length(self, ip: 'IPv4 | IPv6', protocol: 'int', hdr_len: 'int', length: 'int', + from_layer: 'bool') -> 'int': + """Prefer the Length field of a UDP header to the length from the IP layer. + + Args: + ip: IPv4 or IPv6 layer. + protocol: Upper-layer protocol. + hdr_len: Length of the IP header and extension headers. + length: Length from the IP layer. + from_layer: Whether the upper layer is the one ``ip`` carries. + + Returns: + The UDP Length field (:rfc:`768`) if the upper layer is UDP and its + header is present with a non-zero field, ``length`` otherwise. + + """ + if not from_layer or protocol != Enum_TransType.UDP: + return length + # NOTE: Octets 4 and 5 of the UDP header are its Length field. A zero + # field is a jumbogram's, so the IP layer's length stands. + start = hdr_len + 4 + octets = ip.data[start:start + 2] + if len(octets) < 2: + return length + return int.from_bytes(octets, 'big') or length diff --git a/pcapkit/protocols/schema/__init__.py b/pcapkit/protocols/schema/__init__.py index a645628c6a..ae6ca0fe7b 100644 --- a/pcapkit/protocols/schema/__init__.py +++ b/pcapkit/protocols/schema/__init__.py @@ -64,6 +64,8 @@ 'HOPOPT_QuickStartReportOption', 'HOPOPT_RPLOption', 'HOPOPT_MPLOption', 'HOPOPT_ILNPOption', 'HOPOPT_LineIdentificationOption', 'HOPOPT_JumboPayloadOption', 'HOPOPT_HomeAddressOption', 'HOPOPT_IPDFFOption', + 'IP_Pseudo', + 'ICMPv4', 'ICMPv4_ICMPv4Type', 'ICMPv4_Unknown', 'ICMPv4_Echo', 'ICMPv4_DestinationUnreachable', 'ICMPv4_TimeExceeded', diff --git a/pcapkit/protocols/schema/misc/__init__.py b/pcapkit/protocols/schema/misc/__init__.py index 37f3542b8e..fb24c2edee 100644 --- a/pcapkit/protocols/schema/misc/__init__.py +++ b/pcapkit/protocols/schema/misc/__init__.py @@ -77,6 +77,7 @@ # misc protocols from pcapkit.protocols.schema.misc.raw import Raw from pcapkit.protocols.schema.misc.null import NoPayload +from pcapkit.protocols.schema.misc.checksum import IP_Pseudo __all__ = [ # PCAP file format @@ -107,6 +108,7 @@ 'PCAPNG_CustomBlock', 'PCAPNG_PacketBlock', # misc protocols + 'IP_Pseudo', 'NoPayload', 'Raw', ] diff --git a/pcapkit/protocols/schema/misc/checksum/__init__.py b/pcapkit/protocols/schema/misc/checksum/__init__.py new file mode 100644 index 0000000000..11bb658816 --- /dev/null +++ b/pcapkit/protocols/schema/misc/checksum/__init__.py @@ -0,0 +1,6 @@ +# -*- coding: utf-8 -*- +"""header schema for checksum pseudo-headers""" + +from pcapkit.protocols.schema.misc.checksum.ip_pseudo import IP_Pseudo + +__all__ = ['IP_Pseudo'] diff --git a/pcapkit/protocols/schema/misc/checksum/ip_pseudo.py b/pcapkit/protocols/schema/misc/checksum/ip_pseudo.py new file mode 100644 index 0000000000..72a0a1e099 --- /dev/null +++ b/pcapkit/protocols/schema/misc/checksum/ip_pseudo.py @@ -0,0 +1,118 @@ +# -*- coding: utf-8 -*- +# mypy: disable-error-code=assignment +"""header schema for the IP pseudo-header""" + +import ipaddress +from typing import TYPE_CHECKING + +from pcapkit.const.reg.transtype import TransType as Enum_TransType +from pcapkit.corekit.fields.ipaddress import IPv4AddressField, IPv6AddressField +from pcapkit.corekit.fields.misc import ConditionalField, SwitchField +from pcapkit.corekit.fields.numbers import EnumField, UInt16Field, UInt32Field +from pcapkit.corekit.fields.strings import PaddingField +from pcapkit.protocols.schema.schema import Schema, schema_final + +__all__ = ['IP_Pseudo'] + +if TYPE_CHECKING: + from ipaddress import IPv4Address, IPv6Address + from typing import Any + + from pcapkit.corekit.fields.field import FieldBase as Field + +#: Length of the IPv4 pseudo-header, :rfc:`793#section-3.1`: source, destination, +#: zero, protocol and length. +#: +#: .. versionadded:: 1.6.0 +IPV4_PSEUDO_LEN = 2 * (ipaddress.IPV4LENGTH // 8) + 1 + 1 + 2 + +#: Length of the IPv6 pseudo-header, :rfc:`8200#section-8.1`: source, destination, +#: length, zero and next header. +#: +#: .. versionadded:: 1.6.0 +IPV6_PSEUDO_LEN = 2 * (ipaddress.IPV6LENGTH // 8) + 4 + 3 + 1 + + +def is_ipv6(packet: 'dict[str, Any]') -> 'bool': + """Tell whether ``packet`` holds an IPv6 pseudo-header. + + Args: + packet: Packet data. + + Returns: + Whether the pseudo-header has the IPv6 layout: by the version of the + source address once it is known, by the octets to unpack before that. + + .. versionadded:: 1.6.0 + + """ + src = packet.get('src') + if isinstance(src, (ipaddress.IPv4Address, ipaddress.IPv6Address)): + return isinstance(src, ipaddress.IPv6Address) + return packet['__length__'] >= IPV6_PSEUDO_LEN + + +def ip_pseudo_address_selector(packet: 'dict[str, Any]') -> 'Field': + """Selector function for :attr:`IP_Pseudo.src` and :attr:`IP_Pseudo.dst`. + + Args: + packet: Packet data. + + Returns: + An :class:`~pcapkit.corekit.fields.ipaddress.IPv6AddressField` for the + IPv6 layout, an :class:`~pcapkit.corekit.fields.ipaddress.IPv4AddressField` + otherwise. + + .. versionadded:: 1.6.0 + + """ + if is_ipv6(packet): + return IPv6AddressField() + return IPv4AddressField() + + +def ip_pseudo_zero_selector(packet: 'dict[str, Any]') -> 'Field': + """Selector function for :attr:`IP_Pseudo.zero`. + + Args: + packet: Packet data. + + Returns: + A :class:`~pcapkit.corekit.fields.strings.PaddingField` of three octets + for the IPv6 layout, of one octet otherwise. + + .. versionadded:: 1.6.0 + + """ + return PaddingField(length=3 if is_ipv6(packet) else 1) + + +@schema_final +class IP_Pseudo(Schema): + """Header schema for the IP pseudo-header. + + The layout follows the source address: :rfc:`793#section-3.1` for IPv4, + whose length is :attr:`ipv4_len`, and :rfc:`8200#section-8.1` for IPv6, + whose length is :attr:`ipv6_len`. + + .. versionadded:: 1.6.0 + + """ + + #: Source address. + src: 'IPv4Address | IPv6Address' = SwitchField(selector=ip_pseudo_address_selector) + #: Destination address. + dst: 'IPv4Address | IPv6Address' = SwitchField(selector=ip_pseudo_address_selector) + #: Upper-layer packet length (IPv6 only). + ipv6_len: 'int' = ConditionalField(UInt32Field(), is_ipv6) + #: Zero octets. + zero: 'bytes' = SwitchField(selector=ip_pseudo_zero_selector) + #: Upper-layer protocol number. + protocol: 'Enum_TransType' = EnumField(length=1, namespace=Enum_TransType) + #: Upper-layer packet length (IPv4 only). + ipv4_len: 'int' = ConditionalField(UInt16Field(), lambda pkt: not is_ipv6(pkt)) + + if TYPE_CHECKING: + def __init__(self, src: 'IPv4Address | IPv6Address', dst: 'IPv4Address | IPv6Address', + protocol: 'Enum_TransType | int', zero: 'bytes' = b'', + ipv4_len: 'int' = ..., ipv6_len: 'int' = ...) -> 'None': ... diff --git a/tests/corekit/test_fields_misc.py b/tests/corekit/test_fields_misc.py index 95f88afa36..d89ff7afab 100644 --- a/tests/corekit/test_fields_misc.py +++ b/tests/corekit/test_fields_misc.py @@ -62,6 +62,17 @@ 'tcp mptcp add_addr IPv4', {'length': 8}), ('pcapkit.protocols.schema.transport.tcp', 'MPTCPAddAddress', 'address', 'tcp mptcp add_addr IPv6', {'length': 20}), + + # The layout follows ``__length__`` when no address is given yet, 12 for IPv4 + # and 40 for IPv6. + ('pcapkit.protocols.schema.misc.checksum.ip_pseudo', 'IP_Pseudo', 'src', + 'ip_pseudo src IPv4', {'__length__': 12}), + ('pcapkit.protocols.schema.misc.checksum.ip_pseudo', 'IP_Pseudo', 'src', + 'ip_pseudo src IPv6', {'__length__': 40}), + ('pcapkit.protocols.schema.misc.checksum.ip_pseudo', 'IP_Pseudo', 'dst', + 'ip_pseudo dst IPv4', {'__length__': 12}), + ('pcapkit.protocols.schema.misc.checksum.ip_pseudo', 'IP_Pseudo', 'dst', + 'ip_pseudo dst IPv6', {'__length__': 40}), ] diff --git a/tests/protocols/misc/checksum/__init__.py b/tests/protocols/misc/checksum/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/protocols/misc/checksum/test_ip_pseudo_2471_unit.py b/tests/protocols/misc/checksum/test_ip_pseudo_2471_unit.py new file mode 100644 index 0000000000..589f46abfa --- /dev/null +++ b/tests/protocols/misc/checksum/test_ip_pseudo_2471_unit.py @@ -0,0 +1,666 @@ +# -*- coding: utf-8 -*- +"""The IP pseudo-header packs to the octets the upper-layer checksum covers. + +GitHub issue #2471: :class:`~pcapkit.protocols.misc.checksum.ip_pseudo.IP_Pseudo` is +built from a parsed IPv4 or IPv6 layer, or from explicit values, and packs to +the :rfc:`793#section-3.1` or :rfc:`8200#section-8.1` layout. + +The captured cases come from the tracked :file:`examples/captures/in.pcap` and +:file:`examples/captures/dhcp.pcapng`. Each expected pseudo-header is sliced +from the captured IP header by hand, and the one's-complement sum over it and +the upper-layer octets is checked against the sender's own checksum. + +""" +from __future__ import annotations + +import importlib.util +import io +import ipaddress +import unittest +import warnings + +from tests._support import reimport_once_per_class, sample_path + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + +#: Ethernet header length. +ETH = 14 + + +def ones_complement_sum(octets: bytes) -> int: + """Return the 16-bit one's-complement sum of ``octets``, :rfc:`1071`.""" + if len(octets) % 2: + octets += b'\x00' + total = sum(int.from_bytes(octets[i:i + 2], 'big') for i in range(0, len(octets), 2)) + while total >> 16: + total = (total & 0xFFFF) + (total >> 16) + return total + + +def expected_ipv4(ip: bytes) -> bytes: + """Slice the IPv4 pseudo-header out of a captured IPv4 header without options.""" + upper = int.from_bytes(ip[2:4], 'big') - (ip[0] & 0x0F) * 4 + return ip[12:16] + ip[16:20] + b'\x00' + ip[9:10] + upper.to_bytes(2, 'big') + + +def expected_ipv6(ip: bytes) -> bytes: + """Slice the IPv6 pseudo-header out of a captured IPv6 header with no extension header.""" + return ip[8:24] + ip[24:40] + bytes(2) + ip[4:6] + bytes(3) + ip[6:7] + + +def ethernet_frames(name: str) -> 'list[bytes]': + """Return the Ethernet frames of a tracked capture.""" + import pcapkit + + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + extraction = pcapkit.extract(fin=sample_path(name), nofile=True, store=True, + engine='default', verbose=False) + return [frame.info.packet for frame in extraction.frame] + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class IPPseudoCapturedTests(unittest.TestCase): + """Pseudo-headers built from captured IPv4 and IPv6 layers.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def check(self, frame: bytes, version: int, expected: bytes) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + from pcapkit.protocols.link.ethernet import Ethernet + + ip = Ethernet(frame)[f'IPv{version}'] + pseudo = IP_Pseudo(ip=ip) + + self.assertEqual(pseudo.data, expected) + self.assertEqual(pseudo.version, version) + self.assertEqual(ones_complement_sum(pseudo.data + bytes(ip.payload)), 0xFFFF) + + def test_ipv4_tcp(self) -> None: + frame = ethernet_frames('in.pcap')[2] + expected = expected_ipv4(frame[ETH:]) + self.assertEqual(expected, bytes.fromhex('7b81d287' 'c0a80164' '00' '06' '0014')) + self.check(frame, 4, expected) + + def test_ipv4_udp(self) -> None: + frame = ethernet_frames('dhcp.pcapng')[0] + expected = expected_ipv4(frame[ETH:]) + self.assertEqual(expected, bytes.fromhex('00000000' 'ffffffff' '00' '11' '0118')) + self.check(frame, 4, expected) + + def test_ipv6_icmpv6(self) -> None: + for number in (0, 1): + frame = ethernet_frames('in.pcap')[number] + with self.subTest(frame=number): + self.assertEqual(frame[ETH + 6], 58) + self.check(frame, 6, expected_ipv6(frame[ETH:])) + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class IPPseudoUnitTests(unittest.TestCase): + """Pseudo-headers built from explicit values and constructed layers.""" + + #: IPv6 addresses of the IPv6 cases. + SRC6 = '2001:db8::1' + DST6 = '2001:db8::2' + #: Their packed octets. + ADDRS6 = ipaddress.IPv6Address(SRC6).packed + ipaddress.IPv6Address(DST6).packed + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_explicit_ipv4(self) -> None: + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + + pseudo = IP_Pseudo(src='192.0.2.1', dst='198.51.100.2', protocol=TransType.UDP, upper_length=28) + self.assertEqual(pseudo.data, bytes.fromhex('c0000201' 'c6336402' '00' '11' '001c')) + self.assertEqual(len(pseudo), 12) + self.assertEqual(pseudo.length, 12) + self.assertEqual(pseudo.protocol, TransType.UDP) + + def test_explicit_ipv6(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + + pseudo = IP_Pseudo(src=self.SRC6, dst=self.DST6, protocol='TCP', upper_length=0x10203) + self.assertEqual(pseudo.data, self.ADDRS6 + bytes.fromhex('00010203' '000000' '06')) + self.assertEqual(pseudo.length, 40) + self.assertEqual(pseudo.src, ipaddress.IPv6Address(self.SRC6)) + self.assertEqual(pseudo.dst, ipaddress.IPv6Address(self.DST6)) + + def test_protocol_namespace(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + + pseudo = IP_Pseudo(src='10.0.0.1', dst='10.0.0.2', protocol='spam', + protocol_namespace={99: 'spam'}, upper_length=0) + self.assertEqual(pseudo.data[9], 99) + self.assertEqual(pseudo.read().protocol, 99) + + def test_make_read_round_trip(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + + for made in (IP_Pseudo(src='10.0.0.1', dst='10.0.0.2', protocol=6, upper_length=40), + IP_Pseudo(src=self.SRC6, dst=self.DST6, protocol=17, upper_length=8)): + with self.subTest(version=made.version): + parsed = IP_Pseudo(made.data) + self.assertEqual(parsed.info, made.info) + self.assertEqual(bytes(IP_Pseudo.from_data(parsed.info)), made.data) + + def test_nonzero_reserved_octets_round_trip(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + + for octets in (bytes.fromhex('0a000001' '0a000002' 'ff' '06' '0014'), + self.ADDRS6 + bytes.fromhex('00000008' 'abcdef' '11')): + with self.subTest(length=len(octets)): + parsed = IP_Pseudo(octets) + self.assertEqual(bytes(IP_Pseudo.from_data(parsed.info)), octets) + + def test_explicit_values_override_the_layer(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + from pcapkit.protocols.internet.ipv4 import IPv4 + + ip = IPv4(src='10.0.0.1', dst='10.0.0.2', protocol='TCP', payload=bytes(20)) + self.assertEqual(IP_Pseudo(ip=ip).data, bytes.fromhex('0a000001' '0a000002' '00' '06' '0014')) + self.assertEqual(IP_Pseudo(ip=ip, dst='10.0.0.3', upper_length=7).data, + bytes.fromhex('0a000001' '0a000003' '00' '06' '0007')) + + def test_ipv6_extension_headers_are_skipped(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + from pcapkit.protocols.internet.ipv6 import IPv6 + + # Hop-by-Hop Options (next UDP, one PadN) ahead of an 8-octet UDP header. + hopopt = bytes.fromhex('1100' '0104' '00000000') + udp = bytes.fromhex('0035' '0035' '0008' '0000') + ip = IPv6(bytes.fromhex('60000000' '0010' '00' '40') + self.ADDRS6 + hopopt + udp) + + self.assertEqual(IP_Pseudo(ip=ip).data, self.ADDRS6 + bytes.fromhex('00000008' '000000' '11')) + + def test_ipv6_routing_header_final_destination(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + from pcapkit.protocols.internet.ipv6 import IPv6 + + final = ipaddress.IPv6Address('2001:db8::99').packed + # Type 0 Routing header, one segment left, next UDP. + route = bytes.fromhex('1102' '0001' '00000000') + final + udp = bytes.fromhex('0035' '0035' '0008' '0000') + ip = IPv6(bytes.fromhex('60000000' '0020' '2b' '40') + self.ADDRS6 + route + udp) + + expected = ipaddress.IPv6Address(self.SRC6).packed + final + bytes.fromhex('00000008' '000000' '11') + self.assertEqual(IP_Pseudo(ip=ip).data, expected) + + # Type 2 Routing header (RFC 6275), which carries one address. + type2 = bytes.fromhex('1102' '0201' '00000000') + final + ip2 = IPv6(bytes.fromhex('60000000' '0020' '2b' '40') + self.ADDRS6 + type2 + udp) + self.assertEqual(IP_Pseudo(ip=ip2).data, expected) + + # NOTE: No segment left: the header's own destination is the final one. + done = bytearray(ip.data) + done[40 + 3] = 0 + self.assertEqual(IP_Pseudo(ip=IPv6(bytes(done))).data, self.ADDRS6 + bytes.fromhex('00000008' '000000' '11')) + + def test_rejections(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + from pcapkit.protocols.misc.raw import Raw + from pcapkit.utilities.exceptions import FieldValueError, ProtocolError, UnsupportedCall + + with self.assertRaisesRegex(ProtocolError, 'invalid length 13'): + IP_Pseudo(bytes(13)) + with self.assertRaisesRegex(ProtocolError, 'invalid length 13'): + IP_Pseudo(io.BytesIO(bytes(40)), length=13) + self.assertEqual(IP_Pseudo(io.BytesIO(bytes(41)), length=40).version, 6) + with self.assertRaisesRegex(ProtocolError, 'required without ip'): + IP_Pseudo(src='10.0.0.1', dst='10.0.0.2', protocol=6) + with self.assertRaisesRegex(ProtocolError, 'unsupported enclosing layer Raw'): + IP_Pseudo(ip=Raw(b'')) + with self.assertRaises(FieldValueError): + IP_Pseudo(src='10.0.0.1', dst='::2', protocol=6, upper_length=0) + with self.assertRaises(UnsupportedCall): + IP_Pseudo.__index__() + + def test_name_and_chain(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + from pcapkit.protocols.misc.null import NoPayload + + pseudo = IP_Pseudo(src='10.0.0.1', dst='10.0.0.2', protocol=6, upper_length=20) + self.assertEqual(pseudo.name, 'IP Pseudo-Header') + self.assertIsInstance(pseudo.payload, NoPayload) + self.assertEqual(pseudo.protochain.protocols, (IP_Pseudo,)) + self.assertEqual(pseudo.__length_hint__(), 12) + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class IPPseudoBoundaryTests(unittest.TestCase): + """Values an IP layer does not determine, and values out of range. + + The packets are built by scapy, whose checksums the pseudo-headers verify. + + """ + + SRC6 = ipaddress.IPv6Address('2001:db8::1').packed + DST6 = ipaddress.IPv6Address('2001:db8::2').packed + #: An 8-octet UDP header, ports 53, with no checksum. + UDP = bytes.fromhex('0035' '0035' '0008' '0000') + + def setUp(self) -> None: + reimport_once_per_class(self) + + def ipv6(self, next_: int, rest: bytes, payload: 'int | None' = None) -> object: + from pcapkit.protocols.internet.ipv6 import IPv6 + + length = len(rest) if payload is None else payload + return IPv6(bytes.fromhex('60000000') + length.to_bytes(2, 'big') + bytes([next_, 64]) + + self.SRC6 + self.DST6 + rest) + + def verify(self, ip: object, expected: bytes) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + + pseudo = IP_Pseudo(ip=ip) + self.assertEqual(pseudo.data, expected) + self.assertEqual(ones_complement_sum(pseudo.data + bytes(ip.payload)), 0xFFFF) # type: ignore[attr-defined] + + def test_ipv4_options_excluded_from_length(self) -> None: + from pcapkit.protocols.internet.ipv4 import IPv4 + + ip = IPv4(bytes.fromhex('460000230001000040118b91c0000201c6336402' '01010100' + '00350035000b4ed4616263')) + self.verify(ip, bytes.fromhex('c0000201' 'c6336402' '00' '11' '000b')) + + def test_routing_header_last_address(self) -> None: + # Type 0 Routing header, two addresses, two segments left. + route = bytes.fromhex('1104' '0002' '00000000' + '20010db8000000000000000000000098' '20010db8000000000000000000000099') + ip = self.ipv6(0x2B, route + bytes.fromhex('003500350008a368')) + self.verify(ip, self.SRC6 + bytes.fromhex('20010db8000000000000000000000099' '00000008' '000000' '11')) + + def test_home_address_option_source(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + + opts = bytes.fromhex('1102' '01020000' 'c910' '20010db8000000000000000000000077') + ip = self.ipv6(0x3C, opts + bytes.fromhex('003500350008a389')) + self.verify(ip, bytes.fromhex('20010db8000000000000000000000077') + self.DST6 + + bytes.fromhex('00000008' '000000' '11')) + self.assertEqual(IP_Pseudo(ip=ip, src='2001:db8::1').data[:16], self.SRC6) + padded = self.ipv6(0x3C, bytes.fromhex('1100' '0104' '00000000') + self.UDP) + self.assertEqual(IP_Pseudo(ip=padded).data[:16], self.SRC6) + + def test_segment_routing_header(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + from pcapkit.utilities.exceptions import ProtocolError + + # Segment Routing Header, Segment List[0] 2001:db8::99, one segment left. + srh = bytes.fromhex('1104' '0401' '01000000' + '20010db8000000000000000000000099' '20010db8000000000000000000000002') + ip = self.ipv6(0x2B, srh + self.UDP) + with self.assertRaisesRegex(ProtocolError, 'Routing type 4 not parsed; give dst'): + IP_Pseudo(ip=ip) + self.assertEqual(IP_Pseudo(ip=ip, dst='2001:db8::99').data[16:32], srh[8:24]) + + done = bytearray(srh) + done[3] = 0 + self.assertEqual(IP_Pseudo(ip=self.ipv6(0x2B, bytes(done) + self.UDP)).data[16:32], self.DST6) + + def test_fragments_need_the_reassembled_length(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.utilities.exceptions import ProtocolError + + first = self.ipv6(0x2C, bytes.fromhex('1100' '0001' '00000005') + self.UDP) + later = IPv4(bytes.fromhex('4500001c0001000140008ea9c0000201c6336402') + bytes(8)) + more = IPv4(bytes.fromhex('4500001c000120004011' '0000' 'c0000201c6336402') + self.UDP) + for ip in (first, later, more): + with self.subTest(ip=ip): + with self.assertRaisesRegex(ProtocolError, 'fragment; give upper_length'): + IP_Pseudo(ip=ip) + self.assertEqual(IP_Pseudo(ip=ip, upper_length=9).info.len, 9) + + # An atomic fragment is a whole packet. + atomic = self.ipv6(0x2C, bytes.fromhex('1100' '0000' '00000005') + bytes.fromhex('003500350008a3ff')) + self.verify(atomic, self.SRC6 + self.DST6 + bytes.fromhex('00000008' '000000' '11')) + + def test_unparsed_next_header(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + from pcapkit.utilities.exceptions import ProtocolError + + # ESP, SPI 1, sequence 1, encrypted. + ip = self.ipv6(0x32, bytes.fromhex('00000001' '00000001') + bytes(16)) + for kwargs in ({}, {'protocol': 50}, {'upper_length': 24}): + with self.subTest(**kwargs): + with self.assertRaisesRegex(ProtocolError, 'upper layer of IPv6 not parsed'): + IP_Pseudo(ip=ip, **kwargs) + self.assertEqual(IP_Pseudo(ip=ip, protocol=50, upper_length=24).data[-8:], + bytes.fromhex('00000018' '000000' '32')) + + def test_jumbogram(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + from pcapkit.utilities.exceptions import ProtocolError + + # Hop-by-Hop Options with a Jumbo Payload option, then UDP of 65,544 octets. + jumbo = 8 + 8 + 65536 + hopopt = bytes.fromhex('1100' 'c204') + jumbo.to_bytes(4, 'big') + ip = self.ipv6(0x00, hopopt + bytes.fromhex('00350035' '00000000') + bytes(65536), payload=0) + self.assertEqual(IP_Pseudo(ip=ip).data[32:], (jumbo - 8).to_bytes(4, 'big') + bytes.fromhex('000000' '11')) + + with self.assertRaisesRegex(ProtocolError, 'Payload Length 0 and no Jumbo Payload option'): + IP_Pseudo(ip=self.ipv6(0x11, self.UDP, payload=0)) + + def test_upper_length_range(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.utilities.exceptions import FieldValueError + + for src, limit in (('10.0.0.1', 0xFFFF), ('::1', 0xFFFFFFFF)): + dst = '10.0.0.2' if src == '10.0.0.1' else '::2' + with self.subTest(src=src): + self.assertEqual(IP_Pseudo(src=src, dst=dst, protocol=6, upper_length=limit).info.len, limit) + for value in (limit + 1, -1): + with self.assertRaisesRegex(FieldValueError, f'length {value} out of range'): + IP_Pseudo(src=src, dst=dst, protocol=6, upper_length=value) + + # Total Length 0, as a TSO capture has it: the UDP Length field gives the length. + ip = IPv4(bytes.fromhex('450000000001000040118e99c0000201c6336402') + self.UDP) + self.assertEqual(IP_Pseudo(ip=ip).info.len, 8) + + # Without one, the IP layer's length is negative. + tcp = IPv4(bytes.fromhex('450000000001000040068e99c0000201c6336402') + bytes(20)) + with self.assertRaisesRegex(FieldValueError, 'length -20 out of range'): + IP_Pseudo(ip=tcp) + + def test_zero_is_kept_at_the_layout_width(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + + ipv4 = IP_Pseudo(src='10.0.0.1', dst='10.0.0.2', protocol=6, upper_length=20, zero=b'\xaa\xbb') + self.assertEqual(ipv4.data, bytes.fromhex('0a000001' '0a000002' 'aa' '06' '0014')) + self.assertEqual(ipv4.info.zero, b'\xaa') + for zero, kept in ((b'\xaa\xbb\xcc\xdd', 'aabbcc'), (b'\xaa', 'aa0000')): + with self.subTest(zero=zero): + ipv6 = IP_Pseudo(src='::1', dst='::2', protocol=6, upper_length=20, zero=zero) + self.assertEqual(ipv6.data[32:], bytes.fromhex('00000014') + bytes.fromhex(kept) + b'\x06') + self.assertEqual(ipv6.info.zero, bytes.fromhex(kept)) + + def test_each_value_is_required_without_ip(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + from pcapkit.utilities.exceptions import ProtocolError + + values = {'src': '10.0.0.1', 'dst': '10.0.0.2', 'protocol': 6, 'upper_length': 20} + for missing in values: + with self.subTest(missing=missing): + with self.assertRaisesRegex(ProtocolError, 'required without ip'): + IP_Pseudo(**{key: value for key, value in values.items() if key != missing}) + + def test_octets_must_match_the_length(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + from pcapkit.utilities.exceptions import ProtocolError + + with self.assertRaisesRegex(ProtocolError, 'invalid length 0; expected 12 or 40'): + IP_Pseudo(b'') + with self.assertRaisesRegex(ProtocolError, 'invalid length 0; expected 12 or 40'): + IP_Pseudo(io.BytesIO(b'')) + with self.assertRaisesRegex(ProtocolError, 'length 12 disagrees with 40 octets'): + IP_Pseudo(bytes(40), length=12) + with self.assertRaisesRegex(ProtocolError, 'length 12 disagrees with 5 octets'): + IP_Pseudo(io.BytesIO(bytes(5)), length=12) + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class IPPseudoWalkTests(unittest.TestCase): + """Upper layers the IP walk gives wrong values for, and malformed headers. + + Each expected pseudo-header is spelled out from its RFC, not from the code. + + """ + + SRC6 = ipaddress.IPv6Address('2001:db8::1').packed + DST6 = ipaddress.IPv6Address('2001:db8::2').packed + #: An 8-octet UDP header, ports 53, with no checksum. + UDP = bytes.fromhex('0035' '0035' '0008' '0000') + #: An 8-octet ICMPv6 echo request, whose checksum is not filled in. + ICMPV6 = bytes.fromhex('8000' '0000' '00000000') + + def setUp(self) -> None: + reimport_once_per_class(self) + + def ipv6(self, next_: int, rest: bytes, payload: 'int | None' = None) -> object: + from pcapkit.protocols.internet.ipv6 import IPv6 + + length = len(rest) if payload is None else payload + return IPv6(bytes.fromhex('60000000') + length.to_bytes(2, 'big') + bytes([next_, 64]) + + self.SRC6 + self.DST6 + rest) + + @staticmethod + def ipv4_route(kind: int, pointer: int, route: 'list[str]') -> object: + """An IPv4 packet whose first hop is 192.0.2.10, with a source route option.""" + from pcapkit.protocols.internet.ipv4 import IPv4 + + option = bytes([kind, 3 + 4 * len(route), pointer]) + option += b''.join(ipaddress.IPv4Address(address).packed for address in route) + option += bytes(-len(option) % 4) + header = (bytes([0x40 | (5 + len(option) // 4), 0]) + (20 + len(option) + 8).to_bytes(2, 'big') + + bytes(4) + bytes([64, 17, 0, 0]) + ipaddress.IPv4Address('10.0.0.1').packed + + ipaddress.IPv4Address('192.0.2.10').packed) + return IPv4(header + option + bytes.fromhex('0035' '0035' '0008' '0000')) + + @staticmethod + def ipv4(proto: int, rest: bytes) -> object: + """An IPv4 packet from 192.0.2.1 to 198.51.100.2 carrying ``rest``.""" + from pcapkit.protocols.internet.ipv4 import IPv4 + + header = (bytes([0x45, 0]) + (20 + len(rest)).to_bytes(2, 'big') + bytes(4) + + bytes([64, proto, 0, 0]) + ipaddress.IPv4Address('192.0.2.1').packed + + ipaddress.IPv4Address('198.51.100.2').packed) + return IPv4(header + rest) + + def test_udp_length_field(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + from pcapkit.protocols.internet.ipv4 import IPv4 + + # The UDP Length field is 8, though the IP layer carries 12 octets (RFC 768). + ip = self.ipv6(17, self.UDP + bytes(4)) + self.assertEqual(IP_Pseudo(ip=ip).data[32:], bytes.fromhex('00000008' '000000' '11')) + + ipv4 = IPv4(bytes.fromhex('45000020' '00010000' '4011' '0000' 'c0000201' 'c6336402') + + self.UDP + bytes(4)) + self.assertEqual(IP_Pseudo(ip=ipv4).info.len, 8) + + def test_source_route_final_destination(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + + # LSR (131) and SSR (137). With the pointer on or before the last address, + # that address is the destination; once the route is exhausted (RFC 791 + # section 3.1), the destination field is. + for kind in (131, 137): + for pointer, final in ((4, '192.0.2.30'), (8, '192.0.2.30'), (12, '192.0.2.10')): + with self.subTest(kind=kind, pointer=pointer): + ip = self.ipv4_route(kind, pointer, ['192.0.2.20', '192.0.2.30']) + self.assertEqual(ip.info.dst, ipaddress.IPv4Address('192.0.2.10')) # type: ignore[attr-defined] + self.assertEqual(IP_Pseudo(ip=ip).data[4:8], # type: ignore[arg-type] + ipaddress.IPv4Address(final).packed) + + def test_source_route_without_addresses_keeps_the_header_destination(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + + ip = self.ipv4_route(131, 4, []) + self.assertEqual(IP_Pseudo(ip=ip).data[4:8], ipaddress.IPv4Address('192.0.2.10').packed) + + def test_udp_length_after_an_atomic_fragment(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + + # An atomic fragment (Offset 0, More Fragments clear) whose Next Header is + # Destination Options, then UDP. The UDP header sits 16 octets past the + # fixed header, so its Length field is read from there (RFC 8200 section 4.5). + fragment = bytes([60, 0]) + bytes.fromhex('0000' '01020304') + # Tunnel Encapsulation Limit (type 4), then Pad1s; non-zero, so a wrong offset shows. + dest_opts = bytes([17, 0, 4, 1, 5, 0, 0, 0]) + ip = self.ipv6(44, fragment + dest_opts + self.UDP) + self.assertEqual(IP_Pseudo(ip=ip).data[32:], bytes.fromhex('00000008' '000000' '11')) + + def test_authentication_header_is_walked(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + from pcapkit.utilities.exceptions import ProtocolError + + # AH (RFC 4302 section 2.2): Next Header, then Payload Len in 32-bit words + # less two, so 24 octets with a 12-octet ICV. Two nested, then UDP: the + # upper layer is UDP, and the length is the UDP header's alone. + first = bytes([51, 4]) + bytes(22) + second = bytes([17, 4]) + bytes(22) + pseudo = IP_Pseudo(ip=self.ipv4(51, first + second + self.UDP)) + self.assertEqual(pseudo.data[8:], bytes.fromhex('00' '11' '0008')) + # A lone one cut short after its first octets. + with self.assertRaisesRegex(ProtocolError, 'Authentication Header of IPv4 not parsed'): + IP_Pseudo(ip=self.ipv4(51, bytes([17, 4]) + bytes(4))) + + def test_esp_is_never_walked(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + from pcapkit.utilities.exceptions import ProtocolError + + # ESP (RFC 4303): the upper layer is encrypted, so an IPv4 packet with + # one needs its protocol and length given, as an IPv6 packet does. + with self.assertRaisesRegex(ProtocolError, 'upper layer of IPv4 not parsed'): + IP_Pseudo(ip=self.ipv4(50, bytes(16))) + self.assertEqual(IP_Pseudo(ip=self.ipv4(50, bytes(16)), protocol=50, upper_length=24) + .data[8:], bytes.fromhex('00' '32' '0018')) + + def test_authentication_header_over_tcp(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + + # AH (RFC 4302) of 24 octets naming TCP, then a 20-octet TCP header: the length + # is the TCP octets alone, not the 44 that a walk ignoring the AH would count. + pseudo = IP_Pseudo(ip=self.ipv4(51, bytes([6, 4]) + bytes(22) + bytes(20))) + self.assertEqual(pseudo.info.len, 20) + self.assertEqual(pseudo.protocol, 6) + + def test_authentication_header_offsets_the_udp_length_field(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + + # The UDP header follows the 24-octet AH; a walk that drops the AH reads the + # Length field from the ICV, 0xAAAA, not 8. The 4 padding octets keep the two apart. + ip = self.ipv4(51, bytes([17, 4]) + bytes([0xAA] * 22) + self.UDP + bytes(4)) + self.assertEqual(IP_Pseudo(ip=ip).data[8:], bytes.fromhex('00' '11' '0008')) + + def test_authentication_header_in_a_non_first_fragment_is_not_walked(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.utilities.exceptions import ProtocolError + + # Fragment Offset 1 (8 octets), More Fragments clear: the payload is a continuation, + # not an AH header, so the 6 it starts with must not be read as the Next Header. + payload = bytes([6, 4]) + bytes(22) + self.UDP + header = (bytes([0x45, 0]) + (20 + len(payload)).to_bytes(2, 'big') + bytes(2) + + (1).to_bytes(2, 'big') + bytes([64, 51, 0, 0]) + + ipaddress.IPv4Address('192.0.2.1').packed + + ipaddress.IPv4Address('198.51.100.2').packed) + fragment = IPv4(header + payload) + with self.assertRaisesRegex(ProtocolError, 'upper layer of IPv4 not parsed'): + IP_Pseudo(ip=fragment, upper_length=8) + self.assertEqual(IP_Pseudo(ip=fragment, protocol=51, upper_length=8).data[8:], + bytes.fromhex('00' '33' '0008')) + + def test_esp_under_ipv6_with_an_sa_is_never_walked(self) -> None: + from pcapkit.corekit.context import ContextRegistry + from pcapkit.protocols.internet.esp import ESPContext, SecurityAssociation + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.utilities.exceptions import ProtocolError + + # ESP (SPI 0x4321, seq 1) under an ENCR_NULL SA, whose plaintext is UDP then a + # trailer naming UDP. The SA gives the ESP header a Next Header of 17, which + # the walk must not follow: the upper layer is encrypted (RFC 4303). + packet = bytes.fromhex('00004321' '00000001') + self.UDP + bytes([0, 17]) + registry = ContextRegistry.make(ESPContext(SecurityAssociation(spi=0x4321))) + ip = IPv6(bytes.fromhex('60000000') + len(packet).to_bytes(2, 'big') + bytes([50, 64]) + + self.SRC6 + self.DST6 + packet, __context__=registry) + with self.assertRaisesRegex(ProtocolError, 'upper layer of IPv6 not parsed'): + IP_Pseudo(ip=ip) + + def test_truncated_udp_header_keeps_the_layer_length(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + + # Payload Length 8, but the octets end before the UDP header does. + ip = self.ipv6(17, b'', payload=8) + self.assertEqual(IP_Pseudo(ip=ip).data[32:], bytes.fromhex('00000008' '000000' '11')) + + def test_mobility_header_and_hip_are_the_upper_layer(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + + # Mobility Header, Next Header TCP, 8 octets: the chain ends at it, so it is + # the upper layer, and the TCP named inside it is not (RFC 6275 section 6.1.1). + ip = self.ipv6(135, bytes([6, 0, 5, 0]) + bytes(4)) + pseudo = IP_Pseudo(ip=ip) + self.assertEqual(pseudo.data[32:], bytes.fromhex('00000008' '000000' '87')) + self.assertEqual(pseudo.protocol, 135) + + # Headers after it are not walked past it: 8 octets of HIP ahead of 8 of UDP. + ip = self.ipv6(139, bytes([17, 1, 0, 0, 0, 0, 0, 0]) + self.UDP) + pseudo = IP_Pseudo(ip=ip) + self.assertEqual(pseudo.data[32:], bytes.fromhex('00000010' '000000' '8b')) + self.assertEqual(pseudo.protocol, 139) + + def test_extension_header_limit(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + from pcapkit.utilities.exceptions import ProtocolError + + # The walk takes at most 32 headers; a 33rd Destination Options is unparsed. + def chain(count: int) -> bytes: + return b''.join(bytes([60 if index < count - 1 else 17, 0]) + bytes(6) + for index in range(count)) + + self.assertEqual(IP_Pseudo(ip=self.ipv6(60, chain(32) + self.UDP)).info.len, 8) + with self.assertRaisesRegex(ProtocolError, 'upper layer of IPv6 not parsed'): + IP_Pseudo(ip=self.ipv6(60, chain(33) + self.UDP)) + + def test_truncated_headers_are_unparsed(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + from pcapkit.utilities.exceptions import ProtocolError + + # A Type 0 Routing header claiming two segments, cut by snaplen. + with self.assertRaisesRegex(ProtocolError, 'Routing header of IPv6 not parsed; give dst'): + IP_Pseudo(ip=self.ipv6(43, bytes([17, 2, 0, 1, 0, 0, 0, 0]))) + # A Fragment header cut by snaplen. + with self.assertRaisesRegex(ProtocolError, 'upper layer of IPv6 not parsed'): + IP_Pseudo(ip=self.ipv6(44, bytes([17, 0, 0, 0]))) + + def test_malformed_extension_headers_raise_protocol_error(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + from pcapkit.utilities.exceptions import ProtocolError + + # Destination Options whose length runs past the payload. + with self.assertRaisesRegex(ProtocolError, 'Destination Options of IPv6 not parsed; give src'): + IP_Pseudo(ip=self.ipv6(60, bytes([17, 2]) + bytes(22), payload=8)) + # A Type 0 Routing header with an odd Hdr Ext Len. + with self.assertRaisesRegex(ProtocolError, 'Routing header of IPv6 not parsed; give dst'): + IP_Pseudo(ip=self.ipv6(43, bytes([17, 1, 0, 1, 0, 0, 0, 0]) + bytes(8))) + + def test_last_fragment_needs_the_reassembled_length(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + from pcapkit.utilities.exceptions import ProtocolError + + # Fragment Offset 1 (8 octets), More Fragments clear: the last fragment of a packet. + last = self.ipv6(44, bytes([17, 0]) + bytes.fromhex('0008' '00000005') + self.UDP) + with self.assertRaisesRegex(ProtocolError, 'fragment; give upper_length'): + IP_Pseudo(ip=last) + self.assertEqual(IP_Pseudo(ip=last, upper_length=8).info.len, 8) + + def test_extension_header_lengths_are_summed(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + + # Hop-by-Hop Options, then Destination Options, then ICMPv6: 16 octets of + # headers ahead of the 8-octet ICMPv6 message in a 24-octet payload. ICMPv6 + # has no length field, so the walk alone gives the 8. + hop_by_hop = bytes([60, 0]) + bytes(6) + dest_opts = bytes([58, 0]) + bytes(6) + ip = self.ipv6(0, hop_by_hop + dest_opts + self.ICMPV6) + self.assertEqual(IP_Pseudo(ip=ip).data[32:], bytes.fromhex('00000008' '000000' '3a')) + + def test_duplicate_header_types_are_each_summed(self) -> None: + from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + + # Two Destination Options headers, the second one carrying ICMPv6. + first = bytes([60, 0]) + bytes(6) + second = bytes([58, 0]) + bytes(6) + ip = self.ipv6(60, first + second + self.ICMPV6) + self.assertEqual(IP_Pseudo(ip=ip).data[32:], bytes.fromhex('00000008' '000000' '3a')) + + +if __name__ == '__main__': + unittest.main()