# -*- 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.data.misc.pcap.frame import Frame as Data_Frame
from pcapkit.protocols.data.misc.pcap.frame import FrameInfo as Data_FrameInfo
from pcapkit.protocols.schema.misc.pcap.header import RESOLUTION
from pcapkit.toolkit.pcap import tcp_segment
if TYPE_CHECKING:
from ipaddress import IPv4Address, IPv6Address
from pcapkit.protocols.internet.ipv4 import IPv4
from pcapkit.protocols.internet.ipv6 import IPv6
from pcapkit.protocols.internet.ipv6_frag import IPv6_Frag
from pcapkit.protocols.misc.pcapng import PCAPNG, Packet
__all__ = ['ipv4_reassembly', 'ipv6_reassembly', 'tcp_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.IPv4`) and the **DF**
(:attr:`IPv4.flags.df <pcapkit.protocols.data.internet.ipv4.Flags.df>`) flag is
:data:`False`.
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' in frame:
ipv4 = cast('IPv4', frame['IPv4'])
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.IPv6`) and an IPv6 Fragment header
(:rfc:`2460#section-4.5`, i.e.,
:class:`~pcapkit.protocols.internet.ipv6_frag.IPv6_Frag`).
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' in frame:
ipv6 = cast('IPv6', frame['IPv6'])
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, since
# :meth:`IPv6._decode_next_layer <pcapkit.protocols.internet.ipv6.IPv6._decode_next_layer>`
# adds each extension header's length before the Fragment-header check
# breaks its loop. That is correct for a header length, but the
# reassembled packet carries no Fragment header at all
# (:rfc:`8200#section-4.5`), so the boundary wanted here is the one
# *before* it -- which is that same length less the Fragment header's,
# undoing the addition that included it.
hdr_len = ipv6_info.hdr_len - cast('IPv6_Frag', ipv6_frag).length
# NOTE: The reassembly machinery writes the payload into its datagram
# buffer over the span ``tl - ihl``, so ``tl`` is derived from the
# payload actually handed over rather than from ``raw_len``: the two
# agree, but only the former cannot drift out of step with it.
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.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 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.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 from global header
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`` is the on-wire length and ``cap_len`` the captured one,
# which is the way round Wireshark's ``packet-frame.c`` registers the two
# field names this pair borrows. Do not swap them to match some other
# reader: this is the reference, and the PCAP reader in
# :mod:`pcapkit.protocols.misc.pcap.frame` follows it (#618), on the
# strength of those registered names.
len=block.original_len,
cap_len=block.captured_len,
)
frame.__update__(packet=block.packet)
return frame