# -*- coding: utf-8 -*-
"""PCAP Tools
================
.. module:: pcapkit.toolkit.pcap
:mod:`pcapkit.toolkit.pcap` contains the adapters for the PCAP format.
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. The TCP adapters here and in
:mod:`pcapkit.toolkit.pcapng` read the segment with
:func:`~pcapkit.toolkit.pcap.tcp_segment`, and the DTLS ones the record with
:func:`~pcapkit.toolkit.pcap._dtls_fragments`.
"""
from typing import TYPE_CHECKING, NamedTuple, cast
from pcapkit.const.ipv6.extension_header import ExtensionHeader as Enum_ExtensionHeader
from pcapkit.const.reg.transtype import TransType as Enum_TransType
from pcapkit.const.tls.content_type import ContentType as Enum_ContentType
from pcapkit.foundation.reassembly.data.dtls import Packet as DTLS_Packet
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.application.dtls.data import DTLS as Data_DTLS
from pcapkit.protocols.application.dtls.data import DTLSHandshake as Data_DTLSHandshake
from pcapkit.protocols.application.dtls.protocol import DTLS
from pcapkit.protocols.internet.ip.protocol import IP as Protocol_IP
from pcapkit.protocols.internet.ipv4.protocol import IPv4
from pcapkit.protocols.internet.ipv6.protocol import IPv6
from pcapkit.protocols.misc.no_payload.protocol import NoPayload
from pcapkit.protocols.misc.raw.protocol import Raw
from pcapkit.protocols.protocol import ProtocolBase
from pcapkit.protocols.transport.tcp.protocol import TCP
from pcapkit.protocols.transport.tcp.schema import TCP_DATA_OFFSET_OCTET, TCP_MIN_HEADER_LEN
from pcapkit.protocols.transport.udp.protocol import UDP
if TYPE_CHECKING:
from ipaddress import IPv4Address, IPv6Address
from typing import Optional
from pcapkit.const.reg.linktype import LinkType
from pcapkit.protocols.internet.ipv4.data import IPv4 as Data_IPv4
from pcapkit.protocols.internet.ipv6.data import IPv6 as Data_IPv6
from pcapkit.protocols.internet.ipv6_frag.protocol import IPv6_Frag
from pcapkit.protocols.misc.pcap import Frame
__all__ = ['ipv4_reassembly', 'ipv6_reassembly', 'tcp_reassembly', 'dtls_reassembly',
'tcp_traceflow', 'tcp_segment', 'TCPSegment']
[docs]
class TCPSegment(NamedTuple):
"""The fields of a TCP segment that the reassembly and flow tracing records take.
:func:`tcp_segment` returns one.
"""
#: Info of the IP layer carrying the segment, the innermost one in a tunnel.
ip: 'Data_IPv4 | Data_IPv6'
#: TCP source port.
srcport: 'int'
#: TCP destination port.
dstport: 'int'
#: TCP sequence number.
seq: 'int'
#: TCP acknowledgement number.
ack: 'int'
#: TCP synchronise (SYN) flag.
syn: 'bool'
#: TCP finish (FIN) flag.
fin: 'bool'
#: TCP reset (RST) flag.
rst: 'bool'
#: Raw TCP header, as captured.
header: 'bytes'
#: Raw TCP payload, as captured.
payload: 'bytes'
def _carrier(frame: 'ProtocolBase') -> 'tuple[Optional[IPv4 | IPv6], Optional[TCP]]':
"""Fetch the TCP layer of a frame and the IP layer that carries it.
Args:
frame: PCAP frame or PCAP-NG block.
Returns:
The IPv4 or IPv6 layer nearest above the frame's TCP layer, and that TCP
layer; with no TCP layer, the last IPv4 or IPv6 layer of the frame and
:data:`None`. With no IP layer, neither.
Note:
The nearest IP layer is the innermost one of an IP-in-IP tunnel
(:rfc:`2003`, :rfc:`2473`, :rfc:`4213`), whose addresses are the TCP
endpoints', as in Wireshark's TCP conversations; the outermost, which
``frame['IP']`` finds, is the tunnel's (:issue:`1581`). It is found by
walking down from the outermost, so the headers between them do not
intervene: IPv6 dissects its own extension headers, and an AH or other
header inside IPv4 is a layer of its own but not an IP layer.
"""
if 'IP' not in frame:
return None, None
ip = cast('IPv4 | IPv6', frame['IP'])
layer = ip.payload
while isinstance(layer, ProtocolBase) and not isinstance(layer, (NoPayload, TCP)):
if isinstance(layer, Protocol_IP):
ip = cast('IPv4 | IPv6', layer)
layer = layer.payload
return ip, (cast('TCP', frame['TCP']) if 'TCP' in frame else None)
[docs]
def tcp_segment(frame: 'ProtocolBase') -> 'Optional[TCPSegment]':
"""The TCP segment ``frame`` carries, or :data:`None` if it carries none.
The TCP reassembly and flow tracing adapters of this module and of
:mod:`pcapkit.toolkit.pcapng` all read the segment with this.
Args:
frame: PCAP frame or PCAP-NG block.
Returns:
The fields of the TCP layer if :mod:`pcapkit` parsed one, with those of
the IP layer that carries it -- in a tunnel, the innermost
(:issue:`1581`). Otherwise, the fields of the fixed 20-octet TCP header
(:rfc:`9293#section-3.1`) if the payload of the innermost IP layer is a
:class:`~pcapkit.protocols.misc.raw.protocol.Raw` the TCP parser rejected, of
protocol TCP, from a datagram that is not a later fragment, with at
least those 20 octets captured; else :data:`None`. A segment past
:data:`~pcapkit.utilities.config.FRAME_LAYER_LIMIT` is kept raw
without being parsed, not rejected, so it is not read (:issue:`1610`).
Note:
A capture snapped inside the TCP header leaves a Data Offset that runs
past the captured octets, and the TCP parser rejects such a header by
design (:issue:`1404`), so the segment arrives as
:class:`~pcapkit.protocols.misc.raw.protocol.Raw`. The ports, numbers and flags
are all in the fixed header, so the segment still belongs to its flow,
as it does for |dpkt|_, :mod:`scapy` and Wireshark (:issue:`1518`).
The header and payload are then the captured octets either side of the
Data Offset. Inside a tunnel, this holds of the IP layer carrying the
segment, as it does without one.
"""
ip, tcp = _carrier(frame)
if ip is None:
return None
if tcp is not None:
tcp_info = tcp.info
return TCPSegment(
ip=ip.info,
srcport=tcp_info.srcport.port,
dstport=tcp_info.dstport.port,
seq=tcp_info.seq,
ack=tcp_info.ack,
syn=tcp_info.flags.syn,
fin=tcp_info.flags.fin,
rst=tcp_info.flags.rst,
header=tcp.packet.header,
payload=tcp.packet.payload,
)
ip_info = ip.info
raw = ip.payload
# NOTE: neither a Raw with no error (the caller stopped at a ``layer`` or
# ``protocol``) nor one past config.FRAME_LAYER_LIMIT is a rejected segment (#1610).
if (not isinstance(raw, Raw) or raw.info.error is None
or raw._past_layer_limit # pylint: disable=protected-access
or ip_info.protocol != Enum_TransType.TCP):
return None
# A later fragment's payload starts mid-datagram, with no TCP header in it.
# NOTE: the layer's class says which header it is (#1666).
if isinstance(ip, IPv4):
offset = ip.info.offset
else:
frag = ip.extension_headers.get(Enum_ExtensionHeader.IPv6_Frag) # type: ignore[call-overload]
offset = 0 if frag is None else cast('IPv6_Frag', frag).info.offset
if offset:
return None
octets = bytes(raw.packet.payload)
hdr_len = (octets[TCP_DATA_OFFSET_OCTET] >> 4) * 4 if len(octets) >= TCP_MIN_HEADER_LEN else 0
if hdr_len < TCP_MIN_HEADER_LEN:
return None
flags = octets[13]
return TCPSegment(
ip=ip_info,
srcport=int.from_bytes(octets[0:2], 'big'),
dstport=int.from_bytes(octets[2:4], 'big'),
seq=int.from_bytes(octets[4:8], 'big'),
ack=int.from_bytes(octets[8:12], 'big'),
syn=bool(flags & 0x02),
fin=bool(flags & 0x01),
rst=bool(flags & 0x04),
header=octets[:hdr_len],
payload=octets[hdr_len:],
)
[docs]
def _fragment(frame: 'ProtocolBase', version: 'type[IPv4 | IPv6]') -> 'Optional[IPv4 | IPv6]':
"""Fetch the IPv4 or IPv6 layer of a frame that IP reassembly reads.
Args:
frame: PCAP frame or PCAP-NG block.
version: :class:`~pcapkit.protocols.internet.ipv4.protocol.IPv4` or
:class:`~pcapkit.protocols.internet.ipv6.protocol.IPv6`, the layer to fetch.
Returns:
The first ``version`` layer, from the outside in, that is a fragment --
for IPv4, one with **MF** set or a non-zero Fragment Offset, for IPv6,
one with an IPv6 Fragment header; with none, the outermost ``version``
layer; with no ``version`` layer, :data:`None`.
Note:
In an IP-in-IP tunnel (:rfc:`2003`, :rfc:`2473`, :rfc:`4213`) either
datagram may be the fragmented one: the inner, each fragment of it
carried whole by the tunnel, or the tunnel's own, whose first fragment
alone carries the inner header. Each is reassembled on its own header
(:issue:`2307`). A frame of no tunnel has one layer to choose from.
"""
if version.__name__ not in frame:
return None
outermost = cast('IPv4 | IPv6', frame[version.__name__])
layer = outermost # type: ProtocolBase
while isinstance(layer, ProtocolBase) and not isinstance(layer, NoPayload):
if isinstance(layer, IPv4) and version is IPv4:
if layer.info.flags.mf or layer.info.offset:
return layer
elif isinstance(layer, IPv6) and version is IPv6:
if Enum_ExtensionHeader.IPv6_Frag in layer.extension_headers:
return layer
layer = layer.payload
return outermost
[docs]
def ipv4_reassembly(frame: 'Frame') -> 'IP_Packet[IPv4Address] | None':
"""Make data for IPv4 reassembly.
A frame can be reassembled if it contains an IPv4 layer
(:class:`~pcapkit.protocols.internet.ipv4.protocol.IPv4`) and the **DF**
(:attr:`IPv4.flags.df <pcapkit.protocols.internet.ipv4.data.Flags.df>`) flag is
:data:`False`. In an IP-in-IP tunnel, the layer is the first that is a fragment
(see :func:`_fragment`).
Args:
frame: PCAP frame.
Returns:
Data for IPv4 reassembly (:term:`reasm.ipv4.packet`) if the ``frame`` can be
reassembled; otherwise :data:`None`.
See Also:
:class:`pcapkit.foundation.reassembly.ipv4.IPv4`
"""
if (ipv4 := cast('Optional[IPv4]', _fragment(frame, IPv4))) is not None:
ipv4_info = ipv4.info
if ipv4_info.flags.df: # dismiss not fragmented frame
return None
data = IP_Packet(
bufid=(
ipv4_info.src, # source IP address
ipv4_info.dst, # destination IP address
ipv4_info.id, # identification
ipv4_info.protocol, # payload protocol type
),
num=frame.info.number, # original packet range number
fo=ipv4_info.offset, # fragment offset
ihl=ipv4_info.hdr_len, # internet header length
mf=ipv4_info.flags.mf, # more fragment flag
tl=ipv4_info.len, # total length, header includes
header=ipv4.packet.header, # raw bytes type header
payload=bytearray(ipv4.packet.payload), # raw bytearray type payload
timestamp=float(frame.info.time_epoch), # capture timestamp
)
return data
return None
[docs]
def ipv6_reassembly(frame: 'Frame') -> 'IP_Packet[IPv6Address] | None':
"""Make data for IPv6 reassembly.
A frame can be reassembled if it contains an IPv6 layer
(:class:`~pcapkit.protocols.internet.ipv6.protocol.IPv6`) and an IPv6 Fragment header
(:rfc:`2460#section-4.5`, i.e.,
:class:`~pcapkit.protocols.internet.ipv6_frag.protocol.IPv6_Frag`). In an IP-in-IP tunnel,
the layer is the first that has one (see :func:`_fragment`).
Args:
frame: PCAP frame.
Returns:
Data for IPv6 reassembly (:term:`reasm.ipv6.packet`) if the ``frame`` can be
reassembled; otherwise :data:`None`.
See Also:
:class:`pcapkit.foundation.reassembly.ipv6.IPv6`
"""
if (ipv6 := cast('Optional[IPv6]', _fragment(frame, IPv6))) is not None:
ipv6_info = ipv6.info
if (ipv6_frag := ipv6.extension_headers.get( # type: ignore[call-overload]
Enum_ExtensionHeader.IPv6_Frag
)) is None: # dismiss not fragmented frame
return None
ipv6_frag_info = cast('IPv6_Frag', ipv6_frag).info
# NOTE: ``Data_IPv6.hdr_len`` counts the Fragment header, which the
# reassembled packet does not carry (RFC 8200, section 4.5), so subtract it.
hdr_len = ipv6_info.hdr_len - cast('IPv6_Frag', ipv6_frag).length
# NOTE: ``tl`` comes from the payload handed over, not ``raw_len``, as
# reassembly writes the payload over ``tl - ihl``.
payload = bytearray(ipv6_info.fragment.payload)
data = IP_Packet(
bufid=(
ipv6_info.src, # source IP address
ipv6_info.dst, # destination IP address
ipv6_frag_info.id, # identification
ipv6_frag_info.next, # next header field in IPv6 Fragment Header
),
num=frame.info.number, # original packet range number
fo=ipv6_frag_info.offset, # fragment offset
ihl=hdr_len, # header length, only headers before IPv6-Frag
mf=ipv6_frag_info.mf, # more fragment flag
tl=hdr_len + len(payload), # total length, header includes
header=ipv6_info.fragment.header[:hdr_len], # raw bytes type header before IPv6-Frag
payload=payload, # raw bytearray type payload after IPv6-Frag
timestamp=float(frame.info.time_epoch), # capture timestamp
)
return data
return None
[docs]
def tcp_reassembly(frame: 'Frame') -> 'TCP_Packet | None':
"""Make data for TCP reassembly.
A frame can be reassembled if it contains a TCP layer
(:class:`~pcapkit.protocols.transport.tcp.protocol.TCP`), or a TCP segment whose header the
TCP parser rejected, as it does one the snapshot length cut short (:issue:`1518`).
Args:
frame: PCAP frame.
Returns:
Data for TCP reassembly (:term:`reasm.tcp.packet`) if the ``frame`` can be
reassembled; otherwise :data:`None`.
See Also:
:class:`pcapkit.foundation.reassembly.tcp.TCP`
"""
if (segment := tcp_segment(frame)) is not None:
raw_len = len(segment.payload)
data = TCP_Packet(
bufid=(
segment.ip.src, # source IP address
segment.srcport, # source port
segment.ip.dst, # destination IP address
segment.dstport, # destination port
),
num=frame.info.number, # original packet range number
ack=segment.ack, # acknowledgement
dsn=segment.seq, # data sequence number
syn=segment.syn, # synchronise flag
fin=segment.fin, # finish flag
rst=segment.rst, # reset connection flag
header=segment.header, # raw bytes type header
payload=bytearray(segment.payload), # raw bytearray type payload
first=segment.seq, # first sequence number of payload
last=segment.seq + raw_len - 1, # last sequence number of payload
len=raw_len, # payload length, header excludes
timestamp=float(frame.info.time_epoch), # capture timestamp
)
return data
return None
[docs]
def _dtls_fragments(frame: 'ProtocolBase', *, num: 'int',
timestamp: 'float') -> 'Optional[tuple[DTLS_Packet, ...]]':
"""Make data for DTLS handshake-message reassembly from a frame's DTLS record.
The DTLS adapters of this module and of :mod:`pcapkit.toolkit.pcapng` read
the record with this.
Args:
frame: PCAP frame or PCAP-NG block.
num: Original packet range number.
timestamp: Capture timestamp, in seconds since the Unix epoch.
Returns:
One record per fragmented handshake message in the frame's first
:class:`~pcapkit.protocols.application.dtls.protocol.DTLS` layer, if UDP carries
it and it is a cleartext ``handshake`` record holding one; otherwise
:data:`None`. The buffer ID is the IP and UDP endpoints nearest above
the record, with its epoch (:rfc:`6347#section-4.1`).
.. versionadded:: 1.6.0
"""
ip = prev = None # type: Optional[ProtocolBase]
layer = frame # type: ProtocolBase
while isinstance(layer, ProtocolBase) and not isinstance(layer, (NoPayload, DTLS)):
if isinstance(layer, Protocol_IP):
ip = layer
prev, layer = layer, layer.payload
if not isinstance(layer, DTLS) or not isinstance(prev, UDP) or ip is None:
return None
dtls_info = layer.info
# NOTE: a record past epoch 0 is ciphertext, so its content is bytes, not messages.
if (not isinstance(dtls_info, Data_DTLS) or dtls_info.content_type != Enum_ContentType.handshake
or not isinstance(dtls_info.content, tuple)):
return None
ip_info = cast('Data_IPv4 | Data_IPv6', ip.info)
udp_info = prev.info
bufid = (
ip_info.src, # source IP address
udp_info.srcport.port, # source port
ip_info.dst, # destination IP address
udp_info.dstport.port, # destination port
dtls_info.epoch, # DTLS epoch
)
packets = tuple(
DTLS_Packet( # type: ignore[type-var]
bufid=bufid, # buffer identifier
num=num, # original packet range number
msg_type=message.msg_type, # handshake message type
length=message.length, # total message length
message_seq=message.message_seq, # message sequence number
fragment_offset=message.fragment_offset, # fragment offset
fragment_length=message.fragment_length, # fragment length
payload=cast('bytes', message.body), # raw bytes type fragment body
timestamp=timestamp, # capture timestamp
)
for message in dtls_info.content
if isinstance(message, Data_DTLSHandshake) and message.is_fragmented
)
return packets or None
[docs]
def dtls_reassembly(frame: 'Frame') -> 'Optional[tuple[DTLS_Packet, ...]]':
"""Make data for DTLS handshake-message reassembly.
A frame can be reassembled if UDP carries a DTLS layer
(:class:`~pcapkit.protocols.application.dtls.protocol.DTLS`) whose cleartext
``handshake`` record holds a fragmented message
(:attr:`DTLSHandshake.is_fragmented
<pcapkit.protocols.application.dtls.data.DTLSHandshake.is_fragmented>`,
:rfc:`6347#section-4.2.3`). Only the frame's first record is read.
Args:
frame: PCAP frame.
Returns:
Data for DTLS reassembly (:term:`reasm.dtls.packet`), one per fragmented
message, if the ``frame`` can be reassembled; otherwise :data:`None`.
See Also:
:class:`pcapkit.foundation.reassembly.dtls.DTLS`
.. versionadded:: 1.6.0
"""
return _dtls_fragments(frame, num=frame.info.number, timestamp=float(frame.info.time_epoch))
[docs]
def tcp_traceflow(frame: 'Frame', *, data_link: 'LinkType') -> 'TF_TCP_Packet | None':
"""Trace packet flow for TCP.
A frame can be traced if it contains a TCP layer
(:class:`~pcapkit.protocols.transport.tcp.protocol.TCP`), or a TCP segment whose header the
TCP parser rejected, as it does one the snapshot length cut short (:issue:`1518`).
Args:
frame: PCAP frame.
data_link: Data link layer protocol (from global header).
Returns:
Data for TCP flow tracing (:term:`trace.tcp.packet`) if the ``frame`` can be
traced; otherwise :data:`None`.
See Also:
:class:`pcapkit.foundation.traceflow.tcp.TCP`
"""
if (segment := tcp_segment(frame)) is not None:
data = TF_TCP_Packet( # type: ignore[type-var]
protocol=data_link, # data link type from global header
index=frame.info.number, # frame number
frame=frame.info, # extracted frame info
syn=segment.syn, # TCP synchronise (SYN) flag
fin=segment.fin, # TCP finish (FIN) flag
rst=segment.rst, # TCP reset (RST) flag
src=segment.ip.src, # source IP
dst=segment.ip.dst, # destination IP
srcport=segment.srcport, # TCP source port
dstport=segment.dstport, # TCP destination port
timestamp=float(frame.info.time_epoch), # frame timestamp
seq=segment.seq, # TCP sequence number
ack=segment.ack, # TCP acknowledgement number
header=segment.header, # raw bytes type header
payload=bytearray(segment.payload), # raw bytearray type payload
)
return data
return None