diff --git a/docs/source/contributing/pep.rst b/docs/source/contributing/pep.rst index f03eba26d0..d21d6c573e 100644 --- a/docs/source/contributing/pep.rst +++ b/docs/source/contributing/pep.rst @@ -959,13 +959,17 @@ offload heuristic on :issue:`2470`. :rfc:`3376#section-4.1.2`, :rfc:`2784#section-2.5`). The IGMPv1 rule follows :rfc:`2236`, which updates :rfc:`1112`'s 8-octet one. A GRE header with ``C`` clear is :attr:`~pcapkit.corekit.checksum.internet.ChecksumStatus.NOT_COMPUTED`, - and ``make()`` computes one when ``checksum_present`` asks without giving it. - Neither GRE nor the Linux cooked-capture header carries a length, so a GRE frame - the capture cut short there is - :attr:`~pcapkit.corekit.checksum.internet.ChecksumStatus.INVALID`, settled at - implementation (:issue:`2618`); behind IPv4 or IPv6, which state the payload - length, it is :attr:`~pcapkit.corekit.checksum.internet.ChecksumStatus.TRUNCATED` - (:issue:`2649`). + and ``make()`` computes one when ``checksum_present`` asks without giving it + (:issue:`2618`). Neither GRE nor the Linux cooked-capture header carries a + length, so a GRE message the capture cut short there is + :attr:`~pcapkit.corekit.checksum.internet.ChecksumStatus.TRUNCATED` by the + original length its PCAP record or PCAP-NG packet block states (:issue:`2653`); + behind IPv4 or IPv6, by the payload length they state (:issue:`2649`). An IPv4 + Total Length of 0 states none, so behind one, and through an AH after it, GRE, + ICMP and IGMP go by the original length as well (:issue:`2689`). A GRE or IGMP + header the data ends inside is + :attr:`~pcapkit.corekit.checksum.internet.ChecksumStatus.TRUNCATED` too + (:issue:`2689`). * **DCCP covers what its Checksum Coverage gives** (:rfc:`4340#section-9.2`), over a pseudo-header that carries the whole packet's length, as TCP's does, so a fragment is diff --git a/pcapkit/corekit/checksum/internet.py b/pcapkit/corekit/checksum/internet.py index 5e360c717f..b6ab3a2aa2 100644 --- a/pcapkit/corekit/checksum/internet.py +++ b/pcapkit/corekit/checksum/internet.py @@ -31,7 +31,7 @@ from typing import TYPE_CHECKING from pcapkit.corekit.enum import EnumLookup -from pcapkit.utilities.contextvars import rebuilding +from pcapkit.utilities.contextvars import frame_length, rebuilding from pcapkit.utilities.exceptions import stacklevel from pcapkit.utilities.warnings import ChecksumWarning, warn @@ -204,6 +204,11 @@ def verify_message(alias: 'str', data: 'bytes', field: 'bytes', *, verified as the message is parsed, unless an IP fragment carries it; see :func:`fragment_payload`. + Where no layer before the message declares its length, the message runs to + the end of its frame, and the octets the capture cut off the frame, as + :data:`~pcapkit.utilities.contextvars.frame_length` gives them, are + covered too (:issue:`2653`, :issue:`2689`). + Args: alias: Protocol alias, to name in the warning. data: Octets the checksum covers, the checksum field included. @@ -221,6 +226,9 @@ def verify_message(alias: 'str', data: 'bytes', field: 'bytes', *, """ if _in_fragment.get(): return ChecksumStatus.NO_PSEUDO_HEADER + lengths = frame_length.get() + if lengths is not None: + length = (len(data) if length is None else length) + max(lengths[1] - lengths[0], 0) return verify_checksum(alias, data, field, length=length) diff --git a/pcapkit/protocols/internet/ah.py b/pcapkit/protocols/internet/ah.py index 8f3cbcb3f7..f13039b209 100644 --- a/pcapkit/protocols/internet/ah.py +++ b/pcapkit/protocols/internet/ah.py @@ -195,7 +195,9 @@ def read(self, length: 'Optional[int]' = None, *, version: 'Literal[4, 6]' = 4, # IPv6 header, so hold NoPayload here so that membership does not raise (#2484). self._next = NoPayload() # pylint: disable=attribute-defined-outside-init return ah - return self._decode_next_layer(ah, schema.next, length - ah.length) + # NOTE: The payload is the rest of the parent's, so a frame's cut the + # parent handed on is handed on in turn (#2689). + return self._decode_next_layer(ah, schema.next, length - ah.length, declared=False) def make(self, next: 'Enum_TransType | StdlibEnum | AenumEnum | str | int' = Enum_TransType.UDP, diff --git a/pcapkit/protocols/internet/igmp.py b/pcapkit/protocols/internet/igmp.py index 5784b507b1..52a958e8f2 100644 --- a/pcapkit/protocols/internet/igmp.py +++ b/pcapkit/protocols/internet/igmp.py @@ -412,7 +412,10 @@ def _verify_checksum(self, igmp: '_PT', checksum: 'bytes', length: 'int') -> 'No :rfc:`3376#section-4.1.2`, :rfc:`3376#section-4.2.2`). That holds for IGMPv1 as well: :rfc:`1112` defines an 8-octet message, and :rfc:`2236`, which updates it, sums a longer one whole (:rfc:`2236#section-2.5`). See - :func:`~pcapkit.corekit.checksum.internet.verify_message`. + :func:`~pcapkit.corekit.checksum.internet.verify_message`. A message the + data ends inside, as its header's ``__short_read__`` records, is + :attr:`~pcapkit.corekit.checksum.internet.ChecksumStatus.TRUNCATED` + (:issue:`2689`). Args: igmp: Parsed message data. @@ -421,9 +424,16 @@ def _verify_checksum(self, igmp: '_PT', checksum: 'bytes', length: 'int') -> 'No """ if config.CHECKSUM_VERIFY: - from pcapkit.corekit.checksum.internet import verify_message # isort: skip # pylint: disable=import-outside-toplevel + from pcapkit.corekit.checksum.internet import ( # isort: skip # pylint: disable=import-outside-toplevel + ChecksumStatus, verify_message) + # NOTE: The checksum covers the whole message, so a message the data + # ends inside leaves part of what it covers uncaptured (#2689). + if '__short_read__' in self.__header__.__dict__: + status = ChecksumStatus.TRUNCATED + else: + status = verify_message(self.alias, self._data, checksum, length=length) igmp.__update__([ - ('checksum_status', verify_message(self.alias, self._data, checksum, length=length)), + ('checksum_status', status), ]) @classmethod diff --git a/pcapkit/protocols/internet/internet.py b/pcapkit/protocols/internet/internet.py index 7bd348c673..21d7281845 100644 --- a/pcapkit/protocols/internet/internet.py +++ b/pcapkit/protocols/internet/internet.py @@ -449,7 +449,7 @@ class registered for ``proto``. Paired with it, the keywords that def _decode_next_layer(self, dict_: '_PT', proto: 'Optional[int]' = None, # pylint: disable=arguments-differ length: 'Optional[int]' = None, *, packet: 'Optional[dict[str, Any]]' = None, version: 'Literal[4, 6]' = 4, ipv6_exthdr: 'Optional[ProtoChain]' = None, - payload: 'Optional[bytes]' = None) -> '_PT': + payload: 'Optional[bytes]' = None, declared: 'bool' = True) -> '_PT': r"""Decode next layer extractor. Arguments: @@ -461,6 +461,7 @@ def _decode_next_layer(self, dict_: '_PT', proto: 'Optional[int]' = None, # pyl ipv6_exthdr: protocol chain of IPv6 extension headers payload: payload from packet. If not provided, will extract from :meth:`self.__header__.get_payload ` + declared: whether this layer declares the length of its payload Returns: Current protocol with next layer extracted. @@ -474,18 +475,22 @@ def _decode_next_layer(self, dict_: '_PT', proto: 'Optional[int]' = None, # pyl The next layer is parsed with :data:`~pcapkit.utilities.contextvars.frame_length` cleared, since this layer, not the frame, bounds its payload (:issue:`2653`). + With ``declared`` false it is left as it is, for a payload that + runs to the end of the frame, as behind an IPv4 Total Length of 0, + or to the end of the parent's (:issue:`2689`). """ # NOTE: The length this layer hands on bounds the payload, so octets the # capture cut off the frame past it are none of the next layer's (#2653). - # An IPv4 Total Length of 0, read as the rest of the frame, loses the cut. - token = frame_length.set(None) + # A layer declaring none hands the frame's cut on instead (#2689). + token = frame_length.set(None) if declared else None try: next_ = cast('ProtocolBase', # type: ignore[redundant-cast] self._import_next_layer(proto, length, packet=packet, version=version, payload=payload)) # type: ignore[misc,call-arg] finally: - frame_length.reset(token) + if token is not None: + frame_length.reset(token) info, chain = next_.info, next_.protochain # make next layer protocol name diff --git a/pcapkit/protocols/internet/ipv4.py b/pcapkit/protocols/internet/ipv4.py index 632e044f40..cd65390fd6 100644 --- a/pcapkit/protocols/internet/ipv4.py +++ b/pcapkit/protocols/internet/ipv4.py @@ -442,13 +442,16 @@ def read(self, length: 'Optional[int]' = None, *, # pylint: disable=unused-argu # (#1553). Reassembly takes ``packet.payload``, the octets as # captured, which this leaves as they were. proto = None if ipv4.offset else ipv4.protocol + # NOTE: A Total Length of 0 declares none, so the frame's lengths are the + # only record of a cut and the payload keeps them (#2689). + declared = bool(ipv4.len) if config.CHECKSUM_VERIFY: # NOTE: An upper layer checksummed alone, as ICMP is, is not verified # in a fragment; see pcapkit.corekit.checksum.internet (#2470). from pcapkit.corekit.checksum.internet import fragment_payload # isort: skip # pylint: disable=import-outside-toplevel with fragment_payload(bool(ipv4.offset or ipv4.flags.mf)): - return self._decode_next_layer(ipv4, proto, _plen, packet=__packet__) - return self._decode_next_layer(ipv4, proto, _plen, packet=__packet__) + return self._decode_next_layer(ipv4, proto, _plen, packet=__packet__, declared=declared) + return self._decode_next_layer(ipv4, proto, _plen, packet=__packet__, declared=declared) def make(self, tos_pre: 'Enum_ToSPrecedence | StdlibEnum | AenumEnum | int | str' = Enum_ToSPrecedence.Routine, diff --git a/pcapkit/protocols/link/gre.py b/pcapkit/protocols/link/gre.py index 2489237132..b470ebd0ff 100644 --- a/pcapkit/protocols/link/gre.py +++ b/pcapkit/protocols/link/gre.py @@ -78,7 +78,6 @@ from pcapkit.protocols.link.link import Link from pcapkit.protocols.schema.link.gre import GRE as Schema_GRE from pcapkit.utilities import config -from pcapkit.utilities.contextvars import frame_length from pcapkit.utilities.exceptions import ProtocolError if TYPE_CHECKING: @@ -113,7 +112,8 @@ class GRE(Link[Data_GRE, Schema_GRE], capture cut short, holding fewer octets than its original length, reads as :attr:`~pcapkit.corekit.checksum.internet.ChecksumStatus.TRUNCATED`, as :data:`~pcapkit.utilities.contextvars.frame_length` tells it - (:issue:`2653`). + (:issue:`2653`), and so does a header the data ends inside + (:issue:`2689`). """ @@ -223,14 +223,15 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_GRE': ChecksumStatus, verify_message) if _checksum is None: status = ChecksumStatus.NOT_COMPUTED + elif '__short_read__' in schema.__dict__: + # NOTE: The checksum covers the whole message, so a header the data + # ends inside leaves part of what it covers uncaptured (#2689). + status = ChecksumStatus.TRUNCATED else: # NOTE: Behind LinuxSLL the message runs to the end of its frame, so - # what the capture cut off is covered too. Behind IP the value is None: - # Internet clears it, as the length IP declares bounds GRE (#2653). - lengths = frame_length.get() - missing = 0 if lengths is None else max(lengths[1] - lengths[0], 0) + # verify_message covers what the capture cut off too (#2653). status = verify_message(self.alias, self._data, _checksum.to_bytes(2, 'big'), - length=length + missing) + length=length) gre.__update__([ ('checksum_status', status), ]) diff --git a/pcapkit/utilities/contextvars.py b/pcapkit/utilities/contextvars.py index 3802fc92ff..92ea2b9679 100644 --- a/pcapkit/utilities/contextvars.py +++ b/pcapkit/utilities/contextvars.py @@ -38,11 +38,12 @@ #: contextvars.ContextVar[tuple[int, int] | None]: The captured and the original #: length of the frame being dissected, in octets, as its PCAP record or PCAP-NG #: packet block states them; :data:`None` outside a frame, and inside the payload -#: of an :class:`~pcapkit.protocols.internet.internet.Internet` layer, which -#: bounds it. A frame the capture cut short holds fewer octets than were on the -#: wire, which a layer stating no length of its own, as -#: :class:`~pcapkit.protocols.link.gre.GRE`, cannot tell from the octets it is -#: handed (:issue:`2653`). +#: of an :class:`~pcapkit.protocols.internet.internet.Internet` layer that +#: bounds it, which neither an IPv4 Total Length of 0 nor +#: :class:`~pcapkit.protocols.internet.ah.AH` does (:issue:`2689`). A frame +#: the capture cut short holds fewer octets than were on the wire, which a layer +#: stating no length of its own, as :class:`~pcapkit.protocols.link.gre.GRE`, +#: cannot tell from the octets it is handed (:issue:`2653`). frame_length = contextvars.ContextVar( 'pcapkit.utilities.contextvars.frame_length', default=None, ) # type: contextvars.ContextVar[Optional[tuple[int, int]]] diff --git a/tests/protocols/internet/test_igmp_header_cut_2689_unit.py b/tests/protocols/internet/test_igmp_header_cut_2689_unit.py new file mode 100644 index 0000000000..488010c62c --- /dev/null +++ b/tests/protocols/internet/test_igmp_header_cut_2689_unit.py @@ -0,0 +1,121 @@ +# -*- coding: utf-8 -*- +"""An IGMP message the data ends inside reads ``TRUNCATED``, not ``INVALID``. + +GitHub issue #2689: built directly on octets that end inside it, with no enclosing +length, an IGMPv1, IGMPv2 or IGMPv3 message reads the fields the data never held as +zeros, and verifying that zero-filled checksum would read ``INVALID`` with a +``ChecksumWarning``. Every version's checksum covers the whole message +(:rfc:`2236#section-2.3`, :rfc:`3376#section-4.1.2`), so +:meth:`IGMP._verify_checksum ` +reads the header's own ``__short_read__`` record as ``TRUNCATED``, as GRE does. A +whole message is verified. + +The messages are built here; this module is unit-tier and reads no capture file. + +""" +from __future__ import annotations + +import unittest +import warnings +from typing import TYPE_CHECKING +from unittest import mock + +from tests._support import reimport_once_per_class +from tests.foundation import _roundtrip as wire + +if TYPE_CHECKING: + from typing import Any + +#: Test tier (#1958): hand-picked parse -> rebuild round trips. +__tier__ = 'ALL' + + +def message(head: bytes, rest: bytes) -> bytes: + """``head`` (type and Max Resp Code), the checksum over the whole message, then ``rest``.""" + return head + wire.checksum(head + b'\x00\x00' + rest).to_bytes(2, 'big') + rest + + +#: A group record of type MODE_IS_INCLUDE for 239.1.1.1 with one source, 10.0.0.1. +RECORD = bytes.fromhex('01000001ef0101010a000001') + +#: Each message: the class it is read as, its octets, and the lengths of the cuts +#: that leave a short read on its header. Shorter than eight octets, an IGMPv2 or +#: IGMPv3 Query is refused by its class, which classifies a Query by its length. +#: Eight octets of the Report are a whole header whose record count no octet +#: backs, which records no short read. +MESSAGES = { + 'v1-report': ('IGMPv1', message(b'\x12\x00', bytes([224, 0, 0, 1])), range(1, 8)), + 'v2-report': ('IGMPv2', message(b'\x16\x00', bytes([239, 1, 2, 3])), range(1, 8)), + 'v2-leave': ('IGMPv2', message(b'\x17\x00', bytes([239, 1, 2, 3])), range(1, 8)), + 'v3-report': ('IGMPv3', message(b'\x22\x00', bytes.fromhex('00000001') + RECORD), + [*range(1, 8), *range(9, 20)]), + # A General Query with QRV 2, QQIC 125 and one source, 10.0.0.1. + 'v3-query': ('IGMPv3', message(b'\x11\x64', bytes(4) + bytes.fromhex('027d00010a000001')), + range(12, 16)), +} + + +def wrong(msg: bytes) -> bytes: + """``msg`` with its checksum flipped in both octets.""" + return msg[:2] + (int.from_bytes(msg[2:4], 'big') ^ 0x0101).to_bytes(2, 'big') + msg[4:] + + +class IGMPHeaderCutTests(unittest.TestCase): + """An IGMP message the data ends inside holds no checksum to verify.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + from pcapkit.corekit.checksum.internet import ChecksumStatus + from pcapkit.protocols.internet.igmpv1 import IGMPv1 + from pcapkit.protocols.internet.igmpv2 import IGMPv2 + from pcapkit.protocols.internet.igmpv3 import IGMPv3 + from pcapkit.utilities import config + from pcapkit.utilities.warnings import ChecksumWarning + + self.classes = {'IGMPv1': IGMPv1, 'IGMPv2': IGMPv2, 'IGMPv3': IGMPv3} + self.Status = ChecksumStatus + self.config = config + self.ChecksumWarning = ChecksumWarning + + def read(self, name: str, data: bytes) -> 'tuple[Any, list[str]]': + """``data`` read as class ``name`` with verification on, and the checksum warnings.""" + with mock.patch.object(self.config, 'CHECKSUM_VERIFY', True), \ + warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + layer = self.classes[name](data, len(data)) + return layer, [str(item.message) for item in caught + if issubclass(item.category, self.ChecksumWarning)] + + def test_every_cut_reads_truncated(self) -> None: + for label, (name, msg, cuts) in MESSAGES.items(): + for bad in (False, True): + data = wrong(msg) if bad else msg + for count in cuts: + with self.subTest(message=label, bad=bad, captured=count): + layer, caught = self.read(name, data[:count]) + self.assertIsNotNone(layer.info.get('__short_read__')) + self.assertEqual((layer.info.checksum_status, caught), (self.Status.TRUNCATED, [])) + + def test_a_whole_message_is_verified(self) -> None: + for label, (name, msg, _) in MESSAGES.items(): + with self.subTest(message=label): + layer, caught = self.read(name, msg) + self.assertIsNone(layer.info.get('__short_read__')) + self.assertEqual((layer.info.checksum_status, caught), (self.Status.VALID, [])) + layer, caught = self.read(name, wrong(msg)) + self.assertIs(layer.info.checksum_status, self.Status.INVALID) + self.assertEqual(len(caught), 1) + self.assertRegex(caught[0], r'^IGMP(v[23])?: checksum 0x[0-9a-f]{4} is wrong;') + + def test_parse_rebuild_is_exact(self) -> None: + for label, (name, msg, cuts) in MESSAGES.items(): + for count in cuts: + with self.subTest(message=label, captured=count): + layer, _ = self.read(name, msg[:count]) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + self.assertEqual(self.classes[name].from_data(layer.info).data, msg[:count]) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/internet/test_ipv4_tso_cut_checksum_2689_unit.py b/tests/protocols/internet/test_ipv4_tso_cut_checksum_2689_unit.py new file mode 100644 index 0000000000..2287c78a6e --- /dev/null +++ b/tests/protocols/internet/test_ipv4_tso_cut_checksum_2689_unit.py @@ -0,0 +1,252 @@ +# -*- coding: utf-8 -*- +"""A message behind an IPv4 Total Length of 0 the capture cut short reads ``TRUNCATED``. + +GitHub issue #2689: :meth:`IPv4.read ` reads +a Total Length of 0, as TCP segmentation offload leaves it, as the rest of the frame +(#1547), so the IPv4 layer declares no length and the frame's original length is the +only record of a snaplen cut. Cleared there by :meth:`Internet._decode_next_layer +`, the cut is lost, and +GRE, ICMP and IGMP sum the octets captured and read ``INVALID`` with a false +``ChecksumWarning``. So the IPv4 layer leaves +:data:`~pcapkit.utilities.contextvars.frame_length` as it is, an AH between it and the +message hands it on, and :func:`~pcapkit.corekit.checksum.internet.verify_message`, which +all three share, counts the cut. A layer declaring its payload's length clears it, as +#2653 has it. + +Each capture is built here and read through :func:`pcapkit.extract`. This module is +unit-tier and reads no capture file. + +""" +from __future__ import annotations + +import io +import struct +import unittest +import warnings +from typing import TYPE_CHECKING +from unittest import mock + +from tests._support import reimport_once_per_class +from tests.foundation import _roundtrip as wire + +if TYPE_CHECKING: + from typing import Any, Callable + + #: A writer of a capture: link type, the octets captured and the original length. + Writer = Callable[[int, bytes, int], bytes] + +#: Test tier (#1958): hand-picked parse -> rebuild round trips. +__tier__ = 'ALL' + +#: ``LINKTYPE_ETHERNET`` and ``LINKTYPE_IPV4``. +ETHERNET, RAW_IPV4 = 1, 228 +#: Octets of the fixed header of each message below, which a cut must leave whole +#: for the message to parse as its own layer. +HEADER = 8 +#: The octets one cut takes off the end of a frame, where a test needs only one. +CUT = 4 + + +def message(head: bytes, rest: bytes) -> bytes: + """``head``, then the checksum over the whole message, then ``rest``.""" + return head + wire.checksum(head + b'\x00\x00' + rest).to_bytes(2, 'big') + rest + + +def wrong(msg: bytes, at: int) -> bytes: + """``msg`` with the checksum field at ``at`` flipped in both octets.""" + field = int.from_bytes(msg[at:at + 2], 'big') ^ 0x0101 + return msg[:at] + field.to_bytes(2, 'big') + msg[at + 2:] + + +#: An IGMPv3 group record of type MODE_IS_INCLUDE for 239.1.2.3, with one source. +RECORD = bytes([1, 0, 0, 1, 239, 1, 2, 3, 192, 0, 2, 7]) + +#: Each message: its IP protocol number, its octets, where its checksum sits, and +#: the layer it parses as. GRE carries EtherType ``0x88B5``, local experimental, +#: so its payload is Raw; the IGMPv2 report carries trailing octets, so the cut +#: can end inside them as well as inside the message. +MESSAGES = { + 'GRE': (47, message(b'\x80\x00\x88\xb5', bytes(2) + b'payload-octets!!' * 2), 4, 'GRE'), + 'ICMP': (1, message(b'\x08\x00', b'\x00\x01\x00\x01' + b'echo-data-octets' * 2), 2, 'ICMP'), + 'IGMPv2': (2, message(b'\x16\x00', bytes([224, 0, 0, 9]) + b'trailing-octets!' * 2), 2, 'IGMPv2'), + 'IGMPv3': (2, message(b'\x22\x00', b'\x00\x00\x00\x02' + RECORD * 2), 2, 'IGMPv3'), +} + + +def ipv4(payload: bytes, proto: int, *, tso: bool) -> bytes: + """An IPv4 packet carrying ``payload``, with a Total Length of 0 if ``tso``.""" + packet = bytearray(wire.ipv4(payload, proto=proto)) + if tso: + packet[2:4] = packet[10:12] = b'\x00\x00' + packet[10:12] = wire.checksum(bytes(packet[:20])).to_bytes(2, 'big') + return bytes(packet) + + +def ah(nxt: int, payload: bytes) -> bytes: + """An AH header of a 4-octet ICV, Payload Len 2 (RFC 4302 section 2.2), before ``payload``.""" + return bytes([nxt, 2, 0, 0]) + struct.pack('>II', 0x100, 1) + bytes(4) + payload + + +def frame(name: str, *, tso: bool = True, via_ah: bool = False, bad: bool = False, + linktype: int = ETHERNET) -> bytes: + """The frame of link type ``linktype`` carrying message ``name``.""" + proto, msg, at, _ = MESSAGES[name] + if bad: + msg = wrong(msg, at) + packet = ipv4(ah(proto, msg), 51, tso=tso) if via_ah else ipv4(msg, proto, tso=tso) + return wire.ethernet(packet, 0x0800) if linktype == ETHERNET else packet + + +def pcap(linktype: int, packet: bytes, original: int) -> bytes: + """A PCAP savefile holding ``packet``, whose record says ``original`` octets were on the wire.""" + return (struct.pack(' bytes: + """A little-endian PCAP-NG block of ``kind`` around ``body``, padded to 32 bits.""" + body += bytes(-len(body) % 4) + size = 12 + len(body) + return struct.pack(' bytes: + """A PCAP-NG file of one interface and one Enhanced Packet Block holding ``packet``.""" + shb = block(0x0A0D0D0A, struct.pack(' None: + reimport_once_per_class(self) + import pcapkit + from pcapkit.corekit.checksum.internet import ChecksumStatus + from pcapkit.utilities import config + from pcapkit.utilities.warnings import ChecksumWarning + + self.pcapkit = pcapkit + self.Status = ChecksumStatus + self.config = config + self.ChecksumWarning = ChecksumWarning + + def extract(self, capture: bytes) -> 'tuple[Any, list[str]]': + """The first frame of ``capture``, and the checksum warnings reading it gave.""" + with mock.patch.object(self.config, 'CHECKSUM_VERIFY', True), \ + warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + first = self.pcapkit.extract(fin=io.BytesIO(capture), nofile=True).frame[0] + return first, [str(item.message) for item in caught + if issubclass(item.category, self.ChecksumWarning)] + + def status(self, packet: bytes, original: int, layer: str, *, + write: 'Writer' = pcap, linktype: int = ETHERNET) -> 'tuple[Any, list[str]]': + """The checksum status ``layer`` reads, :data:`None` if it did not parse, and the warnings.""" + first, caught = self.extract(write(linktype, packet, original)) + if layer not in first: + return None, caught + return first[layer].info.checksum_status, caught + + def test_every_cut_reads_truncated(self) -> None: + for name, (_, msg, _, layer) in MESSAGES.items(): + for via_ah in (False, True): + for bad in (False, True): + whole = frame(name, via_ah=via_ah, bad=bad) + for cut in range(1, len(msg)): + with self.subTest(message=name, ah=via_ah, bad=bad, cut=cut): + status, caught = self.status(whole[:-cut], len(whole), layer) + self.assertEqual(caught, []) + # A cut into the fixed header leaves the octets as Raw. + if cut <= len(msg) - HEADER: + self.assertIs(status, self.Status.TRUNCATED) + else: + self.assertIn(status, (None, self.Status.TRUNCATED)) + + def test_pcapng_and_raw_ipv4_hand_the_cut_down_too(self) -> None: + writers = {'pcap': pcap, 'pcapng': pcapng} + for name, (_, _, _, layer) in MESSAGES.items(): + for (label, write), linktype in ((item, link) for item in writers.items() + for link in (ETHERNET, RAW_IPV4)): + for via_ah in (False, True): + whole = frame(name, via_ah=via_ah, bad=True, linktype=linktype) + with self.subTest(message=name, capture=label, linktype=linktype, ah=via_ah): + self.assertEqual(self.status(whole[:-CUT], len(whole), layer, + write=write, linktype=linktype), + (self.Status.TRUNCATED, [])) + + def test_complete_messages_are_verified(self) -> None: + for name, (_, _, _, layer) in MESSAGES.items(): + for tso in (True, False): + for via_ah in (False, True): + with self.subTest(message=name, tso=tso, ah=via_ah): + good = frame(name, tso=tso, via_ah=via_ah) + self.assertEqual(self.status(good, len(good), layer), (self.Status.VALID, [])) + bad = frame(name, tso=tso, via_ah=via_ah, bad=True) + status, caught = self.status(bad, len(bad), layer) + self.assertIs(status, self.Status.INVALID) + self.assertEqual(len(caught), 1) + self.assertRegex(caught[0], rf'^{layer}: checksum 0x[0-9a-f]{{4}} is wrong;') + + def test_a_declared_total_length_reads_its_own_cut(self) -> None: + # The control: the length IPv4 declares gives TRUNCATED on its own. + for name, (_, _, _, layer) in MESSAGES.items(): + for via_ah in (False, True): + whole = frame(name, tso=False, via_ah=via_ah, bad=True) + with self.subTest(message=name, ah=via_ah): + self.assertEqual(self.status(whole[:-CUT], len(whole), layer), + (self.Status.TRUNCATED, [])) + + def test_only_a_layer_declaring_no_length_hands_the_cut_on(self) -> None: + from pcapkit.corekit.checksum import internet as checksum + from pcapkit.utilities.contextvars import frame_length + + seen = [] # type: list[object] + verify = checksum.verify_message + + def spy(*args: 'Any', **kwargs: 'Any') -> 'Any': + seen.append(frame_length.get()) + return verify(*args, **kwargs) + + _, msg, _, _ = MESSAGES['ICMP'] + cases = { + 'Total Length 0': (frame('ICMP'), True), + 'Total Length 0, then AH': (frame('ICMP', via_ah=True), True), + 'Total Length declared': (frame('ICMP', tso=False), False), + 'Total Length declared, then AH': (frame('ICMP', tso=False, via_ah=True), False), + # The inner IPv4 declares the length, so it clears what the outer handed on. + 'Total Length 0, then IPv4 declaring one': ( + wire.ethernet(ipv4(ipv4(msg, 1, tso=False), 4, tso=True), 0x0800), False), + } + for name, (whole, kept) in cases.items(): + with self.subTest(case=name), mock.patch.object(checksum, 'verify_message', spy): + seen.clear() + self.extract(pcap(ETHERNET, whole[:-CUT], len(whole))) + self.assertEqual(seen, [(len(whole) - CUT, len(whole)) if kept else None]) + self.assertIsNone(frame_length.get()) + + def test_parse_rebuild_is_exact(self) -> None: + from pcapkit.protocols.misc.pcap.frame import Frame + + for name, (_, msg, _, layer) in MESSAGES.items(): + for via_ah in (False, True): + whole = frame(name, via_ah=via_ah, bad=True) + for cut in (1, CUT, len(msg) - HEADER): + capture = pcap(ETHERNET, whole[:-cut], len(whole)) + with self.subTest(message=name, ah=via_ah, cut=cut): + parsed, _ = self.extract(capture) + with mock.patch.object(self.config, 'CHECKSUM_VERIFY', True), \ + warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + rebuilt = Frame.from_data(parsed.info, num=parsed.info.number, + header=parsed.header) + self.assertEqual(rebuilt.data, capture[24:]) + # The rebuilt frame hands its lengths down as the parsed one did. + self.assertEqual(rebuilt[layer].info.checksum_status, self.Status.TRUNCATED) + self.assertEqual([item.category for item in caught + if issubclass(item.category, self.ChecksumWarning)], []) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/link/test_gre_header_cut_2689_unit.py b/tests/protocols/link/test_gre_header_cut_2689_unit.py new file mode 100644 index 0000000000..3a3ba93f3e --- /dev/null +++ b/tests/protocols/link/test_gre_header_cut_2689_unit.py @@ -0,0 +1,129 @@ +# -*- coding: utf-8 -*- +"""A GRE header the data ends inside reads ``TRUNCATED``, not ``INVALID``. + +GitHub issue #2689: built directly on octets that end inside its header, a +:class:`~pcapkit.protocols.link.gre.GRE` layer reads the fields the data never held as +zeros, and verifying that zero-filled checksum would read ``INVALID`` with a +``ChecksumWarning``. The checksum covers the whole message (:rfc:`2784#section-2.5`), +so a header cut anywhere, in the Checksum or after it, leaves part of what it covers +uncaptured, and :meth:`GRE.read ` reads the +schema's own ``__short_read__`` record as ``TRUNCATED``. A whole header is verified. + +The messages are built here; this module is unit-tier and reads no capture file. + +""" +from __future__ import annotations + +import struct +import unittest +import warnings +from typing import TYPE_CHECKING +from unittest import mock + +from tests._support import reimport_once_per_class +from tests.foundation import _roundtrip as wire + +if TYPE_CHECKING: + from typing import Any + +#: Test tier (#1958): hand-picked parse -> rebuild round trips. +__tier__ = 'ALL' + +#: The payload every message carries, of EtherType ``0x88B5``, local experimental, +#: which no layer is registered for. +PAYLOAD = b'twenty-octet-payload' + + +def gre(flags: int, *, key: 'int | None' = None, sequence: 'int | None' = None, + bad: bool = False) -> bytes: + """A GRE message with ``flags``, carrying :data:`PAYLOAD`. + + The checksum is right, or wrong in both octets when ``bad``; there is none + unless ``flags`` sets ``C``. + + """ + head = struct.pack('>HH', flags, 0x88B5) + extra = (b'' if key is None else struct.pack('>I', key)) \ + + (b'' if sequence is None else struct.pack('>I', sequence)) + if not flags & 0x8000: + return head + extra + PAYLOAD + value = wire.checksum(head + bytes(4) + extra + PAYLOAD) ^ (0x0101 if bad else 0) + return head + struct.pack('>HH', value, 0) + extra + PAYLOAD + + +#: The fields of each message with ``C`` set, for :func:`gre`, and its header's length. +CHECKSUMMED = { + 'C': ({'flags': 0x8000}, 8), + 'C, K': ({'flags': 0xA000, 'key': 0x11223344}, 12), + 'C, K, S': ({'flags': 0xB000, 'key': 0x11223344, 'sequence': 7}, 16), +} # type: dict[str, tuple[dict[str, int], int]] + + +class GREHeaderCutTests(unittest.TestCase): + """A GRE header the data ends inside holds no checksum to verify.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + from pcapkit.corekit.checksum.internet import ChecksumStatus + from pcapkit.protocols.link.gre import GRE + from pcapkit.utilities import config + from pcapkit.utilities.warnings import ChecksumWarning + + self.GRE = GRE + self.Status = ChecksumStatus + self.config = config + self.ChecksumWarning = ChecksumWarning + + def read(self, data: bytes) -> 'tuple[Any, list[str]]': + """``data`` read as GRE with verification on, and the checksum warnings that gave.""" + with mock.patch.object(self.config, 'CHECKSUM_VERIFY', True), \ + warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + layer = self.GRE(data, len(data)) + return layer, [str(item.message) for item in caught + if issubclass(item.category, self.ChecksumWarning)] + + def test_every_cut_inside_the_header_reads_truncated(self) -> None: + for name, (fields, header) in CHECKSUMMED.items(): + for bad in (False, True): + data = gre(**fields, bad=bad) + # From one octet on, the octets held set ``C``. + for count in range(1, header): + with self.subTest(flags=name, bad=bad, captured=count): + layer, caught = self.read(data[:count]) + self.assertIsNotNone(layer.info.get('__short_read__')) + self.assertEqual((layer.info.checksum_status, caught), (self.Status.TRUNCATED, [])) + + def test_a_whole_header_is_verified(self) -> None: + for name, (fields, _) in CHECKSUMMED.items(): + with self.subTest(flags=name): + msg, bad = gre(**fields), gre(**fields, bad=True) + layer, caught = self.read(msg) + self.assertIsNone(layer.info.get('__short_read__')) + self.assertEqual((layer.info.checksum_status, caught), (self.Status.VALID, [])) + layer, caught = self.read(bad) + self.assertIs(layer.info.checksum_status, self.Status.INVALID) + self.assertEqual(caught, [f'GRE: checksum {int.from_bytes(bad[4:6], "big"):#06x} is wrong; ' + f'{int.from_bytes(msg[4:6], "big"):#06x} expected']) + + def test_a_cut_header_without_a_checksum_holds_none(self) -> None: + # With ``C`` clear there is nothing to verify, cut or not. + msg = gre(0x2000, key=0x11223344) + for count in range(1, 8): + with self.subTest(captured=count): + layer, caught = self.read(msg[:count]) + self.assertEqual((layer.info.checksum_status, caught), (self.Status.NOT_COMPUTED, [])) + + def test_parse_rebuild_is_exact(self) -> None: + for name, (fields, header) in CHECKSUMMED.items(): + msg = gre(**fields) + for count in range(1, header + 1): + with self.subTest(flags=name, captured=count): + layer, _ = self.read(msg[:count]) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + self.assertEqual(self.GRE.from_data(layer.info).data, msg[:count]) + + +if __name__ == '__main__': + unittest.main()