# -*- coding: utf-8 -*-
"""Internet Checksum
=======================
.. module:: pcapkit.corekit.checksum.internet
:mod:`pcapkit.corekit.checksum.internet` contains the one's-complement
Internet checksum (:rfc:`1071`) of the IPv4 header, TCP, UDP, ICMP, ICMPv6,
IGMP, GRE and OSPF: :class:`ChecksumStatus`, the outcome a parse records, and
the functions that compute and verify the checksum. It imports nothing from
:mod:`pcapkit.protocols`, which builds on it (:issue:`2580`).
Verification is opt-in and computation opt-out, by
:data:`~pcapkit.utilities.config.CHECKSUM_VERIFY` and
:data:`~pcapkit.utilities.config.CHECKSUM_COMPUTE` (:issue:`2471`). A
mismatch warns with a :exc:`~pcapkit.utilities.warnings.ChecksumWarning`,
which a filter can silence alone (:issue:`2470`).
A checksum over the IP pseudo-header is left to the enclosing IP layer, see
:mod:`pcapkit.protocols.misc.checksum.ip_pseudo.protocol`. A checksum over the message
alone, as ICMP's, IGMP's and GRE's, is verified as it is parsed, unless the IP
layer it is parsed from is a fragment, see :func:`verify_message`.
.. versionadded:: 1.6.0
"""
import contextlib
import contextvars
import enum
import struct
from typing import TYPE_CHECKING
from pcapkit.corekit.enum import EnumLookup
from pcapkit.utilities.contextvars import frame_length, rebuilding
from pcapkit.utilities.exceptions import stacklevel
from pcapkit.utilities.warnings import ChecksumWarning, warn
if TYPE_CHECKING:
from typing import Collection, Iterator, Optional
__all__ = [
'ChecksumStatus',
'internet_checksum', 'compute_checksum', 'partial_checksum', 'verify_checksum',
'verify_message', 'fragment_payload',
]
#: contextvars.ContextVar[bool]: Whether the layer being parsed is the payload
#: of an IP fragment, as :func:`fragment_payload` sets it and
#: :func:`verify_message` reads it. Only this module uses it, so it stays here
#: (:issue:`2616`).
_in_fragment = contextvars.ContextVar('pcapkit.corekit.checksum.internet._in_fragment', default=False)
[docs]
class ChecksumStatus(EnumLookup, enum.IntEnum):
"""Outcome of verifying a checksum, as recorded in ``checksum_status``.
``get`` / ``get_all`` come from :class:`~pcapkit.corekit.enum.EnumLookup`;
the class defines no ``_missing_``, so an unknown value raises
:exc:`ValueError`.
"""
#: The checksum matches.
VALID = 0
#: The checksum does not match, which warns with a
#: :exc:`~pcapkit.utilities.warnings.ChecksumWarning`.
INVALID = 1
#: The field holds no checksum: one the protocol lets the sender leave
#: out, as UDP over IPv4 does with zero (:rfc:`768`), or what checksum
#: offload leaves in a capture taken on the sending host. This is the
#: expected state for such a capture, and is **not** an error.
NOT_COMPUTED = 2
#: Fewer octets were captured than the checksum covers.
TRUNCATED = 3
#: The enclosing IP layer does not give what the checksum covers: it is
#: a fragment, holding only part of the datagram, or the pseudo-header
#: cannot be derived from it.
NO_PSEUDO_HEADER = 4
def _sum(data: 'bytes') -> 'int':
"""One's-complement sum of ``data`` in 16-bit words (:rfc:`1071`).
Args:
data: Octets to sum, padded with a zero octet if of odd length.
Returns:
The sum, folded to 16 bits.
"""
if len(data) % 2:
data += b'\x00'
total = sum(struct.unpack(f'!{len(data) // 2}H', data))
while total > 0xFFFF:
total = (total & 0xFFFF) + (total >> 16)
return total
[docs]
def internet_checksum(data: 'bytes') -> 'int':
"""Compute the Internet checksum of ``data`` (:rfc:`1071`).
Args:
data: Octets the checksum covers, with the checksum field zeroed.
Returns:
The one's complement of the one's-complement sum of ``data``.
"""
return ~_sum(data) & 0xFFFF
[docs]
def compute_checksum(data: 'bytes', *, prefix: 'bytes' = b'', nonzero: 'bool' = False) -> 'bytes':
"""Compute the checksum field of ``data``.
Args:
data: Octets the checksum covers, with the checksum field zeroed.
prefix: Pseudo-header octets the checksum covers ahead of ``data``.
nonzero: Whether a computed zero is written as ``0xFFFF``, as UDP does
(:rfc:`768`), since its zero means no checksum.
Returns:
The two octets of the checksum field.
"""
value = internet_checksum(prefix + data)
if nonzero and not value:
value = 0xFFFF
return value.to_bytes(2, 'big')
[docs]
def partial_checksum(prefix: 'bytes') -> 'bytes':
"""The field checksum offload leaves for a pseudo-header checksum.
A sender that offloads the checksum to its NIC writes the one's-complement
sum of the pseudo-header alone, uncomplemented, for the NIC to finish, as
Linux does for TCP and UDP; a capture taken on that host records it so.
Args:
prefix: Pseudo-header octets.
Returns:
The two octets of that partial checksum.
"""
return _sum(prefix).to_bytes(2, 'big')
[docs]
def verify_checksum(alias: 'str', data: 'bytes', field: 'bytes', *, length: 'Optional[int]' = None,
prefix: 'bytes' = b'', absent: 'Optional[bytes]' = None,
offload: 'Collection[bytes]' = ()) -> 'ChecksumStatus':
"""Verify the checksum ``field`` of ``data``.
Args:
alias: Protocol alias, to name in the warning.
data: Octets the checksum covers, the checksum field included.
field: The checksum field, as on the wire.
length: Octets the checksum covers, as the headers declare it; the
length of ``data`` by default.
prefix: Pseudo-header octets the checksum covers ahead of ``data``.
absent: Field value that means the sender left the checksum out.
offload: Field values checksum offload leaves on the sending host.
Returns:
:attr:`~ChecksumStatus.VALID` if the sum of ``prefix`` and ``data`` is
all ones (:rfc:`1071`), or the reason it is not.
Warns:
ChecksumWarning: If the checksum is :attr:`~ChecksumStatus.INVALID`,
unless a payload is being rebuilt: the layer it is rebuilt for
parses the same octets again, so a rebuilt frame warns once.
"""
if length is not None:
if len(data) < length:
return ChecksumStatus.TRUNCATED
data = data[:length]
if absent is not None and field == absent:
return ChecksumStatus.NOT_COMPUTED
total = _sum(prefix + data)
if total == 0xFFFF:
return ChecksumStatus.VALID
if field in offload:
return ChecksumStatus.NOT_COMPUTED
# NOTE: Adding the field's complement takes it back out of the sum, so the
# value expected is found without knowing where the field sits.
value = int.from_bytes(field, 'big')
rest = total + (~value & 0xFFFF)
expected = ~((rest & 0xFFFF) + (rest >> 16)) & 0xFFFF
if not rebuilding.get():
warn(f'{alias}: checksum {value:#06x} is wrong; {expected:#06x} expected',
ChecksumWarning, stacklevel=stacklevel())
return ChecksumStatus.INVALID
[docs]
def verify_message(alias: 'str', data: 'bytes', field: 'bytes', *,
length: 'Optional[int]' = None) -> 'ChecksumStatus':
"""Verify the checksum ``field`` of a message that covers itself alone.
ICMP's (:rfc:`792`), IGMP's (:rfc:`2236#section-2.3`,
:rfc:`3376#section-4.1.2`) and GRE's (:rfc:`2784#section-2.5`) checksums
cover the octets of their own message, with no pseudo-header, so they are
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.
field: The checksum field, as on the wire.
length: Octets the checksum covers, as the headers declare it; the
length of ``data`` by default.
Returns:
:attr:`~ChecksumStatus.NO_PSEUDO_HEADER` for the payload of an IP
fragment, otherwise as :func:`verify_checksum`.
Warns:
ChecksumWarning: As :func:`verify_checksum` does.
"""
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)
[docs]
@contextlib.contextmanager
def fragment_payload(fragment: 'bool') -> 'Iterator[None]':
"""Mark the payload an IP layer parses as a fragment's, or not.
A checksum over the message alone, as ICMP's (:rfc:`792`), covers all
of the datagram's data, which a fragment carries part of, so
:func:`verify_message` records :attr:`~ChecksumStatus.NO_PSEUDO_HEADER` for
the upper layer of one rather than verifying it (:issue:`2470`). The
nearest enclosing IP layer decides.
Args:
fragment: Whether the IP layer is a fragment; an atomic one is not.
"""
token = _in_fragment.set(fragment)
try:
yield
finally:
_in_fragment.reset(token)