# -*- coding: utf-8 -*-
"""PCAP-NG Tools
===================
.. module:: pcapkit.toolkit.pcapng
:mod:`pcapkit.toolkit.pcapng` contains the adapters for the PCAP-NG 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.
"""
from typing import TYPE_CHECKING, cast
from pcapkit.const.ipv6.extension_header import ExtensionHeader as Enum_ExtensionHeader
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.ipv4.protocol import IPv4
from pcapkit.protocols.internet.ipv6.protocol import IPv6
from pcapkit.protocols.misc.pcap.frame.data import Frame as Data_Frame
from pcapkit.protocols.misc.pcap.frame.data import FrameInfo as Data_FrameInfo
from pcapkit.protocols.misc.pcap.header.schema import RESOLUTION
from pcapkit.toolkit.pcap import _dtls_fragments, _fragment, tcp_segment
if TYPE_CHECKING:
from ipaddress import IPv4Address, IPv6Address
from typing import Optional
from pcapkit.foundation.reassembly.data.dtls import Packet as DTLS_Packet
from pcapkit.protocols.internet.ipv6_frag.protocol import IPv6_Frag
from pcapkit.protocols.misc.pcapng.protocol import PCAPNG, Packet
__all__ = ['ipv4_reassembly', 'ipv6_reassembly', 'tcp_reassembly', 'dtls_reassembly',
'tcp_traceflow', 'block2frame']
[docs]
def ipv4_reassembly(frame: 'PCAPNG') -> '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:`pcapkit.toolkit.pcap._fragment`).
Args:
frame: PCAP-NG 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
frame_info = cast('Packet', frame.info)
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.timestamp_epoch), # capture timestamp
)
return data
return None
[docs]
def ipv6_reassembly(frame: 'PCAPNG') -> '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:`pcapkit.toolkit.pcap._fragment`).
Args:
frame: PCAP-NG 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)
frame_info = cast('Packet', frame.info)
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.timestamp_epoch), # capture timestamp
)
return data
return None
[docs]
def tcp_reassembly(frame: 'PCAPNG') -> '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-NG 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)
frame_info = cast('Packet', frame.info)
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.timestamp_epoch), # capture timestamp
)
return data
return None
[docs]
def dtls_reassembly(frame: 'PCAPNG') -> '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 (see
:func:`pcapkit.toolkit.pcap._dtls_fragments`).
Args:
frame: PCAP-NG 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
"""
frame_info = cast('Packet', frame.info)
return _dtls_fragments(frame, num=frame_info.number, timestamp=float(frame_info.timestamp_epoch))
[docs]
def tcp_traceflow(frame: 'PCAPNG', *, nanosecond: 'bool' = False) -> '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-NG frame.
nanosecond: Whether to use nanosecond precision.
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:
frame_info = cast('Packet', frame.info)
data = TF_TCP_Packet( # type: ignore[type-var]
protocol=frame.linktype, # data link type of the block's interface
index=frame_info.number, # frame number
frame=block2frame(frame_info, nanosecond=nanosecond),
# 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.timestamp_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
[docs]
def block2frame(block: 'Packet', *, nanosecond: 'bool' = False) -> 'Data_Frame':
"""Convert PCAP-NG block to PCAP frame.
Args:
block: PCAP-NG block.
nanosecond: Whether to use nanosecond precision.
Returns:
PCAP frame.
"""
ts_sec = int(block.timestamp_epoch)
ts_usec = int((block.timestamp_epoch - ts_sec) * RESOLUTION[bool(nanosecond)])
frame = Data_Frame(
frame_info=Data_FrameInfo(
ts_sec=ts_sec,
ts_usec=ts_usec,
incl_len=block.captured_len,
orig_len=block.original_len,
),
time=block.timestamp,
number=block.number,
time_epoch=block.timestamp_epoch,
# NOTE: ``len`` on-wire and ``cap_len`` captured, as Wireshark's
# ``packet-frame.c`` registers them; the PCAP reader follows suit (#618).
len=block.original_len,
cap_len=block.captured_len,
)
frame.__update__(packet=block.packet)
return frame