# -*- coding: utf-8 -*-
"""PyPCAPFile Tools
=====================
.. module:: pcapkit.toolkit.pypcapfile
:mod:`pcapkit.toolkit.pypcapfile` contains the adapters for the `PyPCAPFile`_ engine.
The reassembly and flow tracing adapters return the data their
:mod:`pcapkit.foundation` counterpart consumes, or :data:`None` if the frame
cannot be used for it.
.. important::
`PyPCAPFile`_ decodes Ethernet, IPv4, TCP and UDP and nothing else -- there
is no IPv6 decoder at all. :func:`ipv6_reassembly` therefore cannot be
implemented and raises :exc:`~pcapkit.utilities.exceptions.UnsupportedCall`
instead of quietly returning :data:`None`, which would be indistinguishable
from "this frame carries no IPv6 fragment".
For the same reason :func:`tcp_reassembly` and :func:`tcp_traceflow` see TCP
over IPv4 only, and decline a frame carrying TCP over IPv6. When the IPv6
packet is straight over Ethernet (below), that loss is reported with one
:exc:`~pcapkit.utilities.warnings.AttributeWarning` per capture, on its first
such frame, rather than one per frame.
That warning covers an IPv6 packet straight over Ethernet whose Next Header
is TCP. `PyPCAPFile`_ decodes no VLAN tag (802.1Q), no QinQ (802.1ad) and no
tunnel (e.g. IPv6 in IPv4, or GRE), so TCP inside any of them, or behind an
extension header (an IPv6 one, or AH in IPv4), is left out without a
warning.
Note also that `PyPCAPFile`_ decoders *replace* the payload bytes of the layer
they decode. :class:`~pcapkit.foundation.engines.pypcapfile.PyPCAPFile` therefore
stops at the network layer, which keeps the transport segment verbatim so that
the header/payload split this module reports is exact. The transport header is
decoded on demand below, using `PyPCAPFile`_'s own
``pcapfile.protocols.transport.tcp.TCP`` class rather than a
reimplementation.
"""
import binascii
import collections
import ctypes
import functools
import ipaddress
import struct
import sys
import textwrap
from typing import TYPE_CHECKING
from pcapkit.const.reg.ethertype import EtherType as Enum_EtherType
from pcapkit.const.reg.transtype import TransType as Enum_TransType
from pcapkit.foundation.reassembly.data.ip import Packet as IP_Packet
from pcapkit.foundation.reassembly.data.tcp import Packet as TCP_Packet
from pcapkit.foundation.traceflow.data.tcp import Packet as TF_TCP_Packet
from pcapkit.protocols.internet import ipv6 as Module_IPv6
from pcapkit.protocols.internet.ipv4 import IPv4 as Protocol_IPv4
from pcapkit.protocols.schema.internet import ah as Schema_AH_Module
from pcapkit.protocols.schema.internet import ipv4 as Schema_IPv4_Module
from pcapkit.protocols.schema.misc.pcap.header import RESOLUTION
from pcapkit.protocols.schema.transport import tcp as Schema_TCP_Module
from pcapkit.protocols.transport.tcp import TCP as Protocol_TCP
from pcapkit.utilities import config
from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall, stacklevel
from pcapkit.utilities.warnings import AttributeWarning, warn
if TYPE_CHECKING:
from collections import OrderedDict
from ipaddress import IPv4Address, IPv6Address
from typing import Any, Optional
from pcapfile.protocols.network.ip import IP
from pcapfile.structs import pcap_packet as Packet
from pcapkit.const.reg.linktype import LinkType as Enum_LinkType
__all__ = [
'packet2timestamp', 'ipv4_header', 'packet2chain', 'packet2dict',
'ipv4_reassembly', 'ipv6_reassembly', 'tcp_reassembly', 'tcp_traceflow',
]
#: Whether :meth:`bytes.hex` supports the ``sep`` argument (Python 3.8+), as used
#: below by :func:`_parse_mac_address` -- mirrors the same guard in
#: :meth:`pcapkit.protocols.link.ethernet.Ethernet._read_mac_addr`.
py38 = ((version_info := sys.version_info).major >= 3 and version_info.minor >= 8)
#: IPv4 **DF** (don't fragment) bit, within the three-bit flags field.
IPV4_FLAG_DF = 0b010
#: IPv4 **MF** (more fragments) bit, within the three-bit flags field.
IPV4_FLAG_MF = 0b001
#: The captures :func:`_decline_ipv6_tcp` has warned about, oldest first. Each
#: key :func:`_capture_of` gives maps to the anchor that keeps that key unique.
_ipv6_tcp_warned = collections.OrderedDict() # type: OrderedDict[int, Any]
# NOTE: the definitions from here to ``packet2timestamp``, and
# ``_default_accepts``, ``_default_extension_header`` and
# ``_walk_extension_headers`` below, are kept for :mod:`pcapkit.toolkit.scapy`:
# it imports the rest, and ``_walk_extension_headers`` uses
# ``_default_extension_header``. This module's adapters use none of them
# (#1708).
#: IP protocol numbers of the headers :func:`_walk_extension_headers` steps
#: over inside IPv4: AH and the IPv6 Hop-by-Hop Options, Routing, Fragment,
#: Destination Options, Mobility and HIP headers.
IPV4_EXTENSION_HEADERS = frozenset((
Enum_TransType.HOPOPT, Enum_TransType.IPv6_Route, Enum_TransType.IPv6_Frag,
Enum_TransType.AH, Enum_TransType.IPv6_Opts, Enum_TransType.Mobility_Header,
Enum_TransType.HIP,
))
#: IP protocol numbers of the IP-in-IP tunnels, each mapped to the IP version
#: it carries: IPv4 (protocol 4, :rfc:`2003`) and IPv6 (protocol 41,
#: :rfc:`2473` and :rfc:`4213`).
IP_TUNNELS = {Enum_TransType.IPv4: 4, Enum_TransType.IPv6: 6}
[docs]
def packet2timestamp(packet: 'Packet') -> 'float':
"""Calculate the timestamp of a PyPCAPFile packet.
Args:
packet: PyPCAPFile packet.
Returns:
Timestamp of the packet, in seconds since the epoch, as the :obj:`float`
nearest the exact value.
Note:
`PyPCAPFile`_ keeps the two halves of the per-packet timestamp apart, and
the sub-second half is nanoseconds rather than microseconds when the
savefile carries the nanosecond magic number -- which is recorded on the
savefile header as ``ns_resolution``.
The two are joined in one integer division, which rounds once.
``timestamp + timestamp_us / divisor`` rounds twice, so can be an ulp
off at a small ``timestamp``: ``1.0036909999999999`` for ``1.003691``
(:issue:`1668`).
"""
divisor = RESOLUTION[bool(packet.header[0].ns_resolution)]
return (packet.timestamp * divisor + packet.timestamp_us) / divisor
def _parse_ipv4_address(value: 'Any') -> 'IPv4Address':
"""Parse a raw PyPCAPFile IPv4 ``src``/``dst`` field.
Args:
value: Raw value of an ``IP.src``
or ``IP.dst`` field.
Returns:
The parsed address.
Note:
The released `PyPCAPFile`_ (0.12.0) stores ``src``/``dst`` as
dotted-decimal ASCII text in a :class:`ctypes.c_char_p` -- e.g.
``b'10.1.1.2'`` -- rather than as a packed 4-byte value, which is all
:class:`ipaddress.IPv4Address` accepts from a :obj:`bytes` argument
(it raises :exc:`~ipaddress.AddressValueError` otherwise). This helper
also accepts a packed 4-byte :obj:`bytes` value directly, so a
differently-represented `PyPCAPFile`_ fork or release still works.
Anything that is not :obj:`bytes`, a plain :obj:`int` included, is
rejected outright. The unit-test stand-ins model the field as
dotted-decimal :obj:`bytes`, as the released library does, rather than
this helper being widened to fit a fixture.
Raises:
ProtocolError: If ``value`` cannot be parsed as an IPv4 address.
"""
if not isinstance(value, bytes):
raise ProtocolError(f'invalid PyPCAPFile IPv4 address: {value!r}')
try:
if len(value) != 4:
# Dotted-decimal ASCII text, the actual PyPCAPFile 0.12.0 form --
# a packed 4-byte address is never a valid dotted-quad string,
# since the shortest one (``0.0.0.0``) is already 7 bytes long.
return ipaddress.IPv4Address(value.decode('ascii'))
return ipaddress.IPv4Address(value)
except (ValueError, UnicodeDecodeError) as error:
raise ProtocolError(f'invalid PyPCAPFile IPv4 address: {value!r}') from error
def _parse_mac_address(value: 'Any') -> 'str':
"""Parse a raw PyPCAPFile Ethernet ``src``/``dst`` field.
Args:
value: Raw value of an ``Ethernet.src`` or
``Ethernet.dst``
field.
Returns:
The MAC address as text. A raw 6-octet ``value`` is formatted as
lowercase hex, two digits per octet, colon-separated, e.g.
``'01:00:5e:01:03:03'``; any other is decoded as ASCII and returned
as is.
Note:
The released `PyPCAPFile`_ (0.12.0) already stores ``src``/``dst`` pre-formatted
exactly this way, as ASCII text in a :class:`ctypes.c_char_p` -- not as a raw
6-byte value, which :func:`bytes` alone passes straight through unexamined. This
helper also accepts a raw 6-byte value directly, so a differently-represented
`PyPCAPFile`_ fork or release, and the stand-ins used in unit tests, keep working.
Raises:
ProtocolError: If ``value`` cannot be parsed as a MAC address.
"""
if not isinstance(value, (bytes, bytearray)):
raise ProtocolError(f'invalid PyPCAPFile MAC address: {value!r}')
raw = bytes(value)
if len(raw) == 6:
return raw.hex(':') if py38 else ':'.join(textwrap.wrap(raw.hex(), 2))
try:
return raw.decode('ascii')
except UnicodeDecodeError as error:
raise ProtocolError(f'invalid PyPCAPFile MAC address: {value!r}') from error
[docs]
def _maybe_unhex(value: 'bytes') -> 'bytes':
"""Undo :func:`binascii.hexlify`, if ``value`` looks hex-encoded.
Args:
value: Raw bytes, possibly hex-encoded.
Returns:
``value`` hex-decoded, if it is a valid even-length hex string;
otherwise ``value`` itself, unchanged.
Note:
The released `PyPCAPFile`_ (0.12.0) stores an IPv4 packet's options
(``IP.opt``) and payload
(``IP.payload``) as
:func:`binascii.hexlify`'d ASCII text once decoding stops short of the
transport layer -- which is exactly how
:class:`~pcapkit.foundation.engines.pypcapfile.PyPCAPFile` calls it --
rather than as the raw bytes this module otherwise assumes. Both forms
are :obj:`bytes`, hex-encoded or not, which is why :func:`_is_raw`
cannot tell them apart, and falling back to the original value when
it does not decode as hex is what keeps this a no-op for the raw,
non-hex stand-ins used in unit tests.
This is **not** a guarantee against every false positive, only a
cheap and, in practice, reliable heuristic: a genuinely raw payload
whose every byte happens to fall in the ASCII hex-digit range (e.g. a
24-byte TCP header built entirely from digits and ``A``-``F``) is
indistinguishable from real hex text and gets silently halved. There
is no way to tell the two forms apart from the bytes alone; this is
acceptable here because that byte range is a small fraction of the
possible values a real header or payload can take.
"""
try:
return binascii.unhexlify(value)
except binascii.Error:
return value
[docs]
def _is_raw(layer: 'Any') -> 'bool':
"""Test if a decoded layer is in fact undecoded raw bytes.
Args:
layer: Decoded layer, or raw bytes.
"""
return isinstance(layer, (bytes, bytearray, memoryview))
[docs]
def _network(packet: 'Packet') -> 'Optional[IP]':
"""Fetch the decoded IPv4 layer of a PyPCAPFile packet, if any.
Args:
packet: PyPCAPFile packet.
Returns:
The decoded ``pcapfile.protocols.network.ip.IP`` layer, or
:data:`None` when the frame was not decoded that far -- which is the
case for every non-IPv4 frame, `PyPCAPFile`_ having no other network
layer decoder -- or its header length (IHL) runs past the captured
octets (:issue:`1591`).
Note:
`PyPCAPFile`_ slices the header at its IHL however few octets there
are, so a header cut short has fewer options than its IHL declares.
"""
link = packet.packet
if _is_raw(link):
return None
payload = getattr(link, 'payload', None) # type: Any
if type(payload).__name__ != 'IP':
return None
if payload.hl * 4 > Schema_IPv4_Module.IPV4_HEADER_LEN and (
Schema_IPv4_Module.IPV4_HEADER_LEN + len(_maybe_unhex(bytes(payload.opt))) < payload.hl * 4):
return None
return payload
[docs]
def _ipv6_carries_tcp(packet: 'Packet') -> 'bool':
"""Test if a PyPCAPFile frame is an IPv6 packet carrying TCP.
Args:
packet: PyPCAPFile packet.
Returns:
:data:`True` if the frame is IPv6 straight over Ethernet and its Next
Header is TCP; otherwise :data:`False`.
Note:
`PyPCAPFile`_ leaves an IPv6 frame's payload undecoded and hex-encoded
(see :func:`_maybe_unhex`), so this reads the Next Header field of the
raw IPv6 header and decodes nothing.
"""
link = packet.packet
if _is_raw(link) or getattr(link, 'type', None) != Enum_EtherType.Internet_Protocol_version_6:
return False
payload = getattr(link, 'payload', None) # type: Any
if not _is_raw(payload):
return False
data = _maybe_unhex(bytes(payload))
return (len(data) >= Module_IPv6.IPV6_HEADER_LEN
and data[Module_IPv6.IPV6_NEXT_HEADER_OFFSET] == Enum_TransType.TCP)
def _capture_of(packet: 'Packet') -> 'tuple[int, Any]':
"""Identify the capture a PyPCAPFile packet was read from.
Args:
packet: PyPCAPFile packet.
Returns:
A key, equal for every packet of one capture, and an anchor: while the
anchor is alive, no packet of another capture has the same key.
Note:
``pcap_packet.header`` is a
:mod:`ctypes` pointer to the savefile header. Each read of the field
builds a new pointer object, but every one of them points at the same
header, so the header's address is the key. The pointer object read
here is the anchor: it keeps its packet alive and, through that packet,
the header itself, so no other capture's header can be allocated at
that address while it is held. A header that is not a :mod:`ctypes`
pointer is keyed by its identity instead, and anchors itself.
"""
anchor = packet.header
try:
return ctypes.addressof(anchor.contents), anchor
except (AttributeError, TypeError, ValueError):
return id(anchor), anchor
def _decline_ipv6_tcp(packet: 'Packet', count: 'int') -> 'None':
"""Warn that a frame carrying TCP over IPv6 is left out, once per capture.
Args:
packet: PyPCAPFile packet, which :func:`_network` found no IPv4 layer in.
count: Packet index.
Warns:
AttributeWarning: If ``packet`` carries TCP over IPv6 (see
:func:`_ipv6_carries_tcp`) and is the first such frame of its
capture to reach :func:`tcp_reassembly` or :func:`tcp_traceflow`.
Later frames of the same capture are left out without another
warning, and every other capture warns once of its own, however
their frames interleave. Only the :data:`~pcapkit.utilities.config.IPV6_TCP_WARNED_LIMIT`
captures warned about most recently are remembered, so with more
than that read at once a forgotten one warns again: the warning can
repeat, but is never omitted.
"""
key, anchor = _capture_of(packet)
if key in _ipv6_tcp_warned:
return
if not _ipv6_carries_tcp(packet):
return
_ipv6_tcp_warned[key] = anchor
while config.IPV6_TCP_WARNED_LIMIT != -1 and len(_ipv6_tcp_warned) > config.IPV6_TCP_WARNED_LIMIT:
_ipv6_tcp_warned.popitem(last=False)
warn(f"Frame {count}: TCP over IPv6 is left out of TCP reassembly and flow tracing, as "
"'pypcapfile' has no IPv6 decoder; later such frames of this capture are left out "
'without another warning', AttributeWarning, stacklevel=stacklevel())
[docs]
def _transport(ipv4: 'IP') -> 'Optional[bytes]':
"""Fetch the verbatim TCP segment carried by a PyPCAPFile IPv4 packet.
Args:
ipv4: PyPCAPFile IPv4 packet.
Returns:
The raw TCP segment, up to the packet's Total Length, or :data:`None`
if the packet does not carry a TCP header. A later fragment carries
none: from its offset on, its data is a slice of the datagram's payload
(:rfc:`791#section-3.2`, :issue:`1576`). Nor does a segment too short
for a header, or whose Data Offset is below 5 words (:issue:`1590`).
Note:
`PyPCAPFile`_ keeps every octet after the header as the payload, so a
frame padded to the Ethernet minimum carries its padding there. The
segment is cut at the Total Length to leave it out (:issue:`1577`). A
Total Length of ``0``, which TCP segmentation offload leaves in a
capture taken on the sending host, keeps the rest of the frame
(:issue:`1547`).
"""
if ipv4.p != Enum_TransType.TCP or ipv4.off:
return None
segment = ipv4.payload
if not _is_raw(segment):
return None
segment = _maybe_unhex(bytes(segment))
if ipv4.len:
segment = segment[:max(ipv4.len - ipv4.hl * 4, 0)]
if (len(segment) < Schema_TCP_Module.TCP_MIN_HEADER_LEN
or (segment[Schema_TCP_Module.TCP_DATA_OFFSET_OCTET] >> 4) * 4 < Schema_TCP_Module.TCP_MIN_HEADER_LEN):
return None
return segment
[docs]
@functools.lru_cache(maxsize=2)
def _default_accepts(name: 'str', header: 'bytes') -> 'bool':
"""Test if pcapkit's own IPv4 or TCP parser accepts a header.
Args:
name: ``'IPv4'`` or ``'TCP'``, the parser to ask.
header: The whole header, options included, and nothing after it.
Returns:
:data:`False` if the parser raises anything at all; else :data:`True`.
Note:
The parser stops at its own layer, so nothing past the header is
dissected. The answer depends on the octets alone, so the most recent
answers are remembered, as many as the cache's ``maxsize``.
"""
protocol = Protocol_IPv4 if name == 'IPv4' else Protocol_TCP
try:
protocol(header, len(header), protocol=name)
except Exception: # as beholder does
return False
return True
def _ethernet2dict(link: 'Any') -> 'dict[str, Any]':
"""Convert a PyPCAPFile Ethernet frame into :obj:`dict`.
Args:
link: PyPCAPFile Ethernet frame.
Raises:
ProtocolError: If ``link.src`` or ``link.dst`` cannot be parsed as a
MAC address.
"""
return {
'dst': _parse_mac_address(link.dst),
'src': _parse_mac_address(link.src),
'type': link.type,
}
def _ipv4_2dict(ipv4: 'IP') -> 'dict[str, Any]':
"""Convert a PyPCAPFile IPv4 packet into :obj:`dict`.
Args:
ipv4: PyPCAPFile IPv4 packet.
"""
return {
'v': ipv4.v,
'hl': ipv4.hl,
'tos': ipv4.tos,
'len': ipv4.len,
'id': ipv4.id,
'flags': ipv4.flags,
'off': ipv4.off,
'ttl': ipv4.ttl,
'p': ipv4.p,
'sum': ipv4.sum,
'src': str(_parse_ipv4_address(ipv4.src)),
'dst': str(_parse_ipv4_address(ipv4.dst)),
'opt': _maybe_unhex(bytes(ipv4.opt)),
}
def _layer2dict(layer: 'Any') -> 'dict[str, Any]':
"""Convert a decoded PyPCAPFile layer into :obj:`dict`, recursively.
Args:
layer: Decoded PyPCAPFile layer, or raw bytes.
Note:
`PyPCAPFile`_'s own ``pcapfile.protocols.linklayer.ethernet.Ethernet``
and ``pcapfile.protocols.network.ip.IP`` decoders hex-encode the
payload they retain rather than decoding it further (see
:func:`_maybe_unhex`), so a raw payload reached by recursing *from* one of
those classes is un-hexed before being reported here. Without that,
an ``IP`` layer with options would report its own ``opt`` correctly
(:func:`_ipv4_2dict` already un-hexes it) while its nested ``'Raw'``
payload stayed hex-encoded and twice the true length -- an inconsistency
within the very same :obj:`dict`. Any other, unrecognised layer type's
raw payload is left untouched, since nothing here establishes that it is
`PyPCAPFile`_-hexlified rather than genuinely raw.
"""
if _is_raw(layer):
raw = bytes(layer)
return {'raw_len': len(raw), 'raw': raw}
name = type(layer).__name__
if name == 'Ethernet':
dict_ = _ethernet2dict(layer)
elif name == 'IP':
dict_ = _ipv4_2dict(layer)
else:
dict_ = {field[0]: getattr(layer, field[0], None)
for field in getattr(type(layer), '_fields_', ())}
payload = getattr(layer, 'payload', None)
if payload is not None:
if name in ('Ethernet', 'IP') and _is_raw(payload):
payload = _maybe_unhex(bytes(payload))
dict_['Raw' if _is_raw(payload) else type(payload).__name__] = _layer2dict(payload)
return dict_
[docs]
def packet2chain(packet: 'Packet', *, data_link: 'Enum_LinkType') -> 'str':
"""Fetch PyPCAPFile packet protocol chain.
Args:
packet: PyPCAPFile packet.
data_link: Data link type, from the savefile header.
Returns:
Colon (``:``) separated list of protocol chain.
Note:
The chain reports what `PyPCAPFile`_ actually decoded and no more, so it
ends in ``Raw`` -- the transport segment is left undecoded on purpose,
see the module notes.
"""
layer = packet.packet
if _is_raw(layer):
return f'{data_link.name}:Raw'
chain = [] # type: list[str]
while not _is_raw(layer):
chain.append(type(layer).__name__)
layer = getattr(layer, 'payload', b'')
chain.append('Raw')
return ':'.join(chain)
[docs]
def packet2dict(packet: 'Packet', *, data_link: 'Enum_LinkType') -> 'dict[str, Any]':
"""Convert PyPCAPFile packet into :obj:`dict`.
Args:
packet: PyPCAPFile packet.
data_link: Data link type, from the savefile header.
Returns:
A :obj:`dict` mapping of packet data.
Raises:
ProtocolError: If a decoded IPv4 layer's ``src``/``dst`` cannot be
parsed as an IPv4 address, or a decoded Ethernet layer's
``src``/``dst`` cannot be parsed as a MAC address.
Note:
A frame `PyPCAPFile`_ left undecoded is hex-encoded, as the engine reads
it (see :meth:`PyPCAPFile._read_records
<pcapkit.foundation.engines.pypcapfile.PyPCAPFile._read_records>`), so
it is un-hexed (see :func:`_maybe_unhex`) and reported as its octets,
not as twice as many characters of hex text (:issue:`1667`).
"""
frame = packet.packet
if _is_raw(frame):
frame = _maybe_unhex(bytes(frame))
return {
'timestamp': packet2timestamp(packet),
'capture_len': packet.capture_len,
'packet_len': packet.packet_len,
data_link.name: _layer2dict(frame),
}
[docs]
def ipv4_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv4Address] | None':
"""Make data for IPv4 reassembly.
A packet can be reassembled if it contains an IPv4 layer
(``pcapfile.protocols.network.ip.IP``) whose header was captured whole (see
:func:`_network`), and the **DF** flag is :data:`False`.
Args:
packet: PyPCAPFile packet.
count: Packet index. If not provided, defaults to ``-1``.
Returns:
Data for IPv4 reassembly (:term:`reasm.ipv4.packet`) if the ``packet`` can be
reassembled; otherwise :data:`None`.
Raises:
ProtocolError: If ``ipv4.src`` or ``ipv4.dst`` cannot be parsed as an
IPv4 address.
See Also:
:class:`pcapkit.foundation.reassembly.ipv4.IPv4`
"""
ipv4 = _network(packet)
if ipv4 is None:
return None
if ipv4.flags & IPV4_FLAG_DF: # dismiss not fragmented packet
return None
header = ipv4_header(ipv4)
return IP_Packet(
bufid=(
_parse_ipv4_address(ipv4.src), # source IP address
_parse_ipv4_address(ipv4.dst), # destination IP address
ipv4.id, # identification
Enum_TransType.get(ipv4.p), # payload protocol type
),
num=count, # original packet range number
fo=ipv4.off * 8, # fragment offset, in octets
ihl=len(header), # internet header length
mf=bool(ipv4.flags & IPV4_FLAG_MF), # more fragment flag
tl=ipv4.len, # total length, header includes
header=header, # raw bytes type header
payload=bytearray(_maybe_unhex(bytes(ipv4.payload))), # raw bytearray type payload
timestamp=packet2timestamp(packet), # capture timestamp
)
[docs]
def ipv6_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv6Address] | None':
"""Make data for IPv6 reassembly.
Args:
packet: PyPCAPFile packet.
count: Packet index. If not provided, defaults to ``-1``.
Raises:
UnsupportedCall: Always, as `PyPCAPFile`_ has no IPv6 decoder.
"""
raise UnsupportedCall('IPv6 reassembly is not supported by the PyPCAPFile engine: '
"'pypcapfile' has no IPv6 decoder, so no IPv6 fragment header "
'can be read')
[docs]
def tcp_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'TCP_Packet | None':
"""Make data for TCP reassembly.
A packet can be reassembled if it contains an IPv4 layer carrying a TCP segment (see
:func:`_transport`); TCP over IPv6 cannot be, as `PyPCAPFile`_ has no IPv6 decoder.
Args:
packet: PyPCAPFile packet.
count: Packet index. If not provided, defaults to ``-1``.
Returns:
Data for TCP reassembly (:term:`reasm.tcp.packet`) if the ``packet`` can be
reassembled; otherwise :data:`None`.
Raises:
ProtocolError: If ``ipv4.src`` or ``ipv4.dst`` cannot be parsed as an
IPv4 address.
Warns:
AttributeWarning: If ``packet`` is an IPv6 packet straight over Ethernet
whose Next Header is TCP (see :func:`_ipv6_carries_tcp`), which is
left out, and is the first such frame of its capture to reach either
:func:`tcp_reassembly` or :func:`tcp_traceflow`. The rest of that
capture's such frames are left out without another warning.
See Also:
:class:`pcapkit.foundation.reassembly.tcp.TCP`
"""
ipv4 = _network(packet)
if ipv4 is None:
_decline_ipv6_tcp(packet, count)
return None
segment = _transport(ipv4)
if segment is None:
return None
# NOTE: imported only once there is something to decode, so that declining a
# frame does not require ``pcapfile`` to be installed.
from pcapfile.protocols.transport.tcp import TCP # isort:skip
tcp = TCP(segment)
hdr_len = tcp.data_offset # at least ``TCP_MIN_HEADER_LEN``, see ``_transport``
payload = segment[hdr_len:]
return TCP_Packet(
bufid=(
_parse_ipv4_address(ipv4.src), # source IP address
tcp.src_port, # source port
_parse_ipv4_address(ipv4.dst), # destination IP address
tcp.dst_port, # destination port
),
num=count, # original packet range number
ack=tcp.acknum, # acknowledgement
dsn=tcp.seqnum, # data sequence number
rst=bool(tcp.rst), # reset connection flag
syn=bool(tcp.syn), # synchronise flag
fin=bool(tcp.fin), # finish flag
header=segment[:hdr_len], # raw bytes type header
payload=bytearray(payload), # raw bytearray type payload
first=tcp.seqnum, # first sequence number of payload
last=tcp.seqnum + len(payload) - 1, # last sequence number of payload
len=len(payload), # payload length, header excludes
timestamp=packet2timestamp(packet), # capture timestamp
)
[docs]
def tcp_traceflow(packet: 'Packet', *, data_link: 'Enum_LinkType',
count: 'int' = -1) -> 'TF_TCP_Packet | None':
"""Trace packet flow for TCP.
A packet can be traced if it contains an IPv4 layer carrying a TCP segment (see
:func:`_transport`); TCP over IPv6 cannot be, as `PyPCAPFile`_ has no IPv6 decoder.
Args:
packet: PyPCAPFile packet.
data_link: Data link layer protocol (from the savefile header).
count: Packet index. If not provided, defaults to ``-1``.
Returns:
Data for TCP flow tracing (:term:`trace.tcp.packet`) if the ``packet`` can be
traced; otherwise :data:`None`.
Raises:
ProtocolError: If ``ipv4.src`` or ``ipv4.dst`` cannot be parsed as an
IPv4 address.
Warns:
AttributeWarning: If ``packet`` is an IPv6 packet straight over Ethernet
whose Next Header is TCP (see :func:`_ipv6_carries_tcp`), which is
left out, and is the first such frame of its capture to reach either
:func:`tcp_reassembly` or :func:`tcp_traceflow`. The rest of that
capture's such frames are left out without another warning.
See Also:
:class:`pcapkit.foundation.traceflow.tcp.TCP`
"""
ipv4 = _network(packet)
if ipv4 is None:
_decline_ipv6_tcp(packet, count)
return None
segment = _transport(ipv4)
if segment is None:
return None
# NOTE: imported only once there is something to decode, so that declining a
# frame does not require ``pcapfile`` to be installed.
from pcapfile.protocols.transport.tcp import TCP # isort:skip
tcp = TCP(segment)
hdr_len = tcp.data_offset # at least ``TCP_MIN_HEADER_LEN``, see ``_transport``
return TF_TCP_Packet( # type: ignore[type-var]
protocol=data_link, # data link type from savefile header
index=count, # frame number
frame=packet2dict(packet, data_link=data_link), # extracted packet
syn=bool(tcp.syn), # TCP synchronise (SYN) flag
fin=bool(tcp.fin), # TCP finish (FIN) flag
rst=bool(tcp.rst), # TCP reset (RST) flag
src=_parse_ipv4_address(ipv4.src), # source IP
dst=_parse_ipv4_address(ipv4.dst), # destination IP
srcport=tcp.src_port, # TCP source port
dstport=tcp.dst_port, # TCP destination port
timestamp=packet2timestamp(packet), # timestamp
seq=tcp.seqnum, # TCP sequence number
ack=tcp.acknum, # TCP acknowledgement number
header=segment[:hdr_len], # raw bytes type header
payload=bytearray(segment[hdr_len:]), # raw bytearray type payload
)