# -*- coding: utf-8 -*-
"""DPKT Tools
================
.. module:: pcapkit.toolkit.dpkt
:mod:`pcapkit.toolkit.dpkt` contains the adapters for the `DPKT`_ 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.
"""
import copy
import decimal
import ipaddress
from typing import TYPE_CHECKING, cast
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.data import FrameRecord
from pcapkit.foundation.traceflow.data.tcp import Packet as TF_TCP_Packet
from pcapkit.protocols.data.misc.pcap.frame import FrameInfo as Data_FrameInfo
from pcapkit.protocols.schema.misc.pcap.header import RESOLUTION, RESOLUTION_PLACES
from pcapkit.utilities.exceptions import UnsupportedCall
if TYPE_CHECKING:
from decimal import Decimal
from ipaddress import IPv4Address, IPv6Address
from typing import Any, Optional
from dpkt.dpkt import Packet
from dpkt.ip import IP
from dpkt.ip6 import IP6, IP6FragmentHeader
from dpkt.tcp import TCP
from pcapkit.const.reg.linktype import LinkType as Enum_LinkType
__all__ = [
'RecordTimestamp', 'Timestamp', 'DecimalTimestamp', 'ipv6_hdr_len', 'attach_timestamp', 'packet2timestamp', 'attach_buffer', 'packet2bytes',
'packet2frame', 'packet2chain', 'packet2dict',
'ipv4_reassembly', 'ipv6_reassembly', 'tcp_reassembly', 'tcp_traceflow'
]
#: Attribute a frame's capture timestamp is stashed under.
#:
#: `DPKT`_ keeps the two halves of a record apart -- its reader yields
#: ``(timestamp, bytes)`` and only the bytes become a packet -- so a frame on its
#: own does not know when it was captured. Anything that reads a frame *after* the
#: extraction loop has moved on therefore has no way back to the timestamp unless
#: the engine puts it somewhere, and this is where
#: :class:`~pcapkit.foundation.engines.dpkt.DPKT` puts it.
#:
#: A `DPKT`_ packet carries a :attr:`~object.__dict__`, so this is an ordinary
#: attribute rather than anything exotic; the name is spelled out here so that
#: nothing has to know it by hand.
#:
#: .. _DPKT: https://dpkt.readthedocs.io
TIMESTAMP_ATTR = '__pcapkit_timestamp__'
#: Attribute a frame's record octets are stashed under.
#:
#: A `DPKT`_ packet keeps no copy of the octets it was parsed from, and
#: serialising it again is not the same thing: ``dpkt.ip.IP.__bytes__``
#: recomputes a zeroed checksum and length and writes them back into the
#: packet, as do the transport layers beneath it. So the adapters below slice
#: the wire octets out of this buffer instead, which
#: :class:`~pcapkit.foundation.engines.dpkt.DPKT` attaches alongside
#: :data:`TIMESTAMP_ATTR`.
#:
#: .. _DPKT: https://dpkt.readthedocs.io
BUFFER_ATTR = '__pcapkit_buffer__'
[docs]
class RecordTimestamp:
"""A record's capture timestamp, with what a PCAP frame record needs of it.
The readers :class:`~pcapkit.foundation.engines.dpkt.DPKT` uses yield
``(timestamp, octets)`` and nothing else, so a record's exact timestamp, its
resolution and its original length would be lost between the reader and
:func:`packet2frame`. The engine's readers yield a :class:`Timestamp` or a
:class:`DecimalTimestamp` instead, which *is* the value `DPKT`_ yields --
so a stored frame's timestamp is exactly what it was before -- and also
carries what would otherwise be lost. Reassembly and flow tracing are handed a plain
:obj:`float` of it (:issue:`1501`): of its exact value, since `DPKT`_'s
own for a microsecond PCAP rounds twice (:issue:`1668`; see
:func:`_capture_time`).
A copy, a deep copy and a pickle at any protocol all rebuild a timestamp
from what its :meth:`~object.__reduce__` returns, so each keeps the exact
timestamp, the resolution and the original length.
This class answers each of those protocols itself, ahead of the numeric base
class: the pure-Python :class:`~decimal.Decimal` copies a subclass as
``type(self)(str(self))``, with the constructor's defaults for the rest
(:issue:`1647`).
"""
__slots__ = ()
if TYPE_CHECKING:
#: Exact timestamp, in seconds since the UNIX epoch.
exact: 'Decimal'
#: Timestamp resolution of the record, in units per second.
resolution: 'int'
#: Original length of the packet, or :data:`None` if unknown.
orig_len: 'Optional[int]'
def __reduce__(self) -> 'tuple[Any, ...]':
raise NotImplementedError(f'{type(self).__name__} does not say how it is rebuilt')
def __reduce_ex__(self, protocol: 'Any') -> 'tuple[Any, ...]':
return self.__reduce__()
def __copy__(self) -> 'Any':
factory, args = self.__reduce__()
return factory(*args)
def __deepcopy__(self, memo: 'Any') -> 'Any':
factory, args = self.__reduce__()
return factory(*copy.deepcopy(args, memo))
[docs]
class Timestamp(RecordTimestamp, float):
"""A :obj:`float` capture timestamp; see :class:`RecordTimestamp`.
Args:
value: The timestamp, as `DPKT`_ yields it.
exact: Exact timestamp; ``value`` itself if not given.
resolution: Timestamp resolution of the record, in units per second.
orig_len: Original length of the packet, or :data:`None` if unknown.
"""
__slots__ = ('exact', 'resolution', 'orig_len')
def __new__(cls, value: 'Any' = 0.0, exact: 'Optional[Decimal]' = None,
resolution: 'int' = RESOLUTION[False], orig_len: 'Optional[int]' = None) -> 'Timestamp':
self = super().__new__(cls, value)
self.exact = decimal.Decimal(float(self)) if exact is None else exact
self.resolution = resolution
self.orig_len = orig_len
return self
def __reduce__(self) -> 'tuple[Any, ...]':
return type(self), (float(self), self.exact, self.resolution, self.orig_len)
[docs]
class DecimalTimestamp(RecordTimestamp, decimal.Decimal):
"""A :class:`~decimal.Decimal` capture timestamp; see :class:`RecordTimestamp`.
`DPKT`_ yields a :class:`~decimal.Decimal` for a nanosecond PCAP.
Args:
value: The timestamp, as `DPKT`_ yields it, which is exact.
resolution: Timestamp resolution of the record, in units per second.
orig_len: Original length of the packet, or :data:`None` if unknown.
"""
__slots__ = ('exact', 'resolution', 'orig_len')
def __new__(cls, value: 'Any' = '0', resolution: 'int' = RESOLUTION[True],
orig_len: 'Optional[int]' = None) -> 'DecimalTimestamp':
self = super().__new__(cls, value)
self.exact = decimal.Decimal(self)
self.resolution = resolution
self.orig_len = orig_len
return self
def __reduce__(self) -> 'tuple[Any, ...]':
return type(self), (str(self), self.resolution, self.orig_len)
[docs]
def attach_timestamp(packet: 'Packet', timestamp: 'float') -> 'None':
"""Stash a frame's capture timestamp on the frame.
Args:
packet: DPKT packet.
timestamp: Capture timestamp of the packet, as `DPKT`_'s reader yielded it
beside the record's octets.
"""
setattr(packet, TIMESTAMP_ATTR, timestamp)
[docs]
def packet2timestamp(packet: 'Packet') -> 'float':
"""Read back the capture timestamp of a DPKT packet.
Args:
packet: DPKT packet, as stored by
:class:`~pcapkit.foundation.engines.dpkt.DPKT`.
Returns:
Capture timestamp of the packet, in seconds since the epoch. As
:class:`~pcapkit.foundation.engines.dpkt.DPKT` attaches it, that is a
:class:`Timestamp`, or for a nanosecond PCAP file a
:class:`DecimalTimestamp`, which is a :class:`~decimal.Decimal`
subclass rather than a :obj:`float`.
Raises:
UnsupportedCall: If the packet carries no timestamp, i.e. it did not come
through :class:`~pcapkit.foundation.engines.dpkt.DPKT`. Raised rather
than defaulted, because a plausible-looking zero would silently
misdate whatever was going to use it.
"""
timestamp = getattr(packet, TIMESTAMP_ATTR, None)
if timestamp is None:
raise UnsupportedCall(
f'{type(packet).__name__} carries no capture timestamp; only a frame read by '
"'Extractor(engine=dpkt)' has one attached"
)
return cast('float', timestamp)
[docs]
def _capture_time(timestamp: 'float | Decimal') -> 'float':
"""The capture timestamp reassembly and flow tracing are handed.
Args:
timestamp: Capture timestamp of the packet.
Returns:
The :obj:`float` nearest the exact timestamp. `DPKT`_ reads a
microsecond PCAP's as ``tv_sec + tv_usec / 1e6``, which rounds twice,
so can be an ulp off at a small ``tv_sec``:
``1.0036909999999999`` for ``1.003691`` (:issue:`1668`). A
:class:`RecordTimestamp` is therefore read by its exact value. Any other
timestamp -- one passed by hand, or yielded by a |dpkt|_ whose reader
the engine could not extend -- carries none, and is handed on as it is.
"""
if isinstance(timestamp, RecordTimestamp):
return float(timestamp.exact)
return float(timestamp)
[docs]
def attach_buffer(packet: 'Packet', buffer: 'bytes') -> 'None':
"""Stash a frame's record octets on the frame.
Args:
packet: DPKT packet.
buffer: Octets of the record the packet was parsed from, as `DPKT`_'s
reader yielded them.
"""
setattr(packet, BUFFER_ATTR, bytes(buffer))
[docs]
def packet2bytes(packet: 'Packet') -> 'bytes':
"""Fetch the octets of a DPKT packet without modifying it.
Args:
packet: DPKT packet.
Returns:
The record octets attached by :func:`attach_buffer`, i.e. the frame as
it was captured. A packet that carries none -- one built by hand rather
than read by :class:`~pcapkit.foundation.engines.dpkt.DPKT` -- is
serialised from a copy instead, so that `DPKT`_'s checksum and length
recomputation never reaches the packet itself.
"""
buffer = getattr(packet, BUFFER_ATTR, None)
if buffer is not None:
return cast('bytes', buffer)
return copy.deepcopy(packet).pack()
[docs]
def packet2frame(packet: 'Packet', timestamp: 'float | Decimal', *,
data_link: 'Enum_LinkType') -> 'FrameRecord':
"""Report a DPKT packet to the flow tracer, with its PCAP frame record.
Args:
packet: DPKT packet.
timestamp: Capture timestamp of the packet. The one attached by
:func:`attach_timestamp`, if any, is preferred, since it is the value
`DPKT`_'s reader yielded.
data_link: Data link type.
Returns:
The :func:`packet2dict` mapping of the packet, which every trace format
but PCAP writes unchanged, carrying the captured octets, from
:func:`packet2bytes`, the PCAP record header the PCAP trace dumper
writes, and the exact timestamp it reads the header's resolution from.
Note:
The fraction in ``frame_info.ts_usec`` is in the capture's own resolution,
and both it and ``orig_len`` come from the reader:
:class:`~pcapkit.foundation.engines.dpkt.DPKT` reads PCAP and PCAP-NG
alike into a :class:`RecordTimestamp`. A timestamp that is not one
-- a packet built by hand, or read by a |dpkt|_ whose reader the engine
could not extend -- is read as `DPKT`_ yields it: a
:class:`~decimal.Decimal` counts nanoseconds and a :obj:`float`
microseconds, and the octets are taken to be whole.
"""
stamp = getattr(packet, TIMESTAMP_ATTR, timestamp)
octets = packet2bytes(packet)
if isinstance(stamp, RecordTimestamp):
epoch, resolution, orig_len = stamp.exact, stamp.resolution, stamp.orig_len
elif isinstance(stamp, decimal.Decimal):
epoch, resolution, orig_len = stamp, RESOLUTION[True], None
else:
epoch, resolution, orig_len = (round(decimal.Decimal(stamp), RESOLUTION_PLACES[False]),
RESOLUTION[False], None)
ts_sec = int(epoch)
ts_usec = int((epoch - ts_sec) * RESOLUTION[resolution > RESOLUTION[False]])
# the mapping is exactly the one reported before #1507, plain value types
# included, so that the dict-capable trace formats write the same bytes
if isinstance(timestamp, float):
timestamp = float(timestamp)
elif isinstance(timestamp, decimal.Decimal):
timestamp = decimal.Decimal(timestamp)
return FrameRecord(
packet2dict(packet, timestamp, data_link=data_link),
packet=octets,
frame_info=Data_FrameInfo(
ts_sec=ts_sec,
ts_usec=ts_usec,
incl_len=len(octets),
orig_len=len(octets) if orig_len is None else orig_len,
),
time_epoch=decimal.Decimal(epoch),
)
def _ipv6_ext_hdrs(ipv6: 'IP6') -> 'list[Any]':
"""Fetch an IPv6 packet's extension headers, in the order they were parsed.
Args:
ipv6: DPKT IPv6 packet.
Returns:
``dpkt.ip6.IP6.all_extension_headers`` where `DPKT`_ recorded it,
which keeps the on-wire order and any header type that occurs twice;
otherwise the values of ``dpkt.ip6.IP6.extension_hdrs``, which
``dpkt.ip6.IP6.__len__`` falls back to in the same way.
"""
ext_hdrs = getattr(ipv6, 'all_extension_headers', None)
if ext_hdrs:
return list(ext_hdrs)
return list(getattr(ipv6, 'extension_hdrs', {}).values())
def _wire_hdr_len(packet: 'Packet', payload: 'Packet') -> 'int':
"""Calculate the octets a layer occupies before its payload begins.
Args:
packet: DPKT packet, i.e. one layer of a frame.
payload: That layer's payload, i.e. ``packet.data``.
Returns:
The advance :meth:`~dpkt.dpkt.Packet.unpack` itself made to reach the
payload.
For IPv6 that is the fixed header plus each extension header's
``dpkt.ip6.IP6ExtensionHeader.length``, which is what
:meth:`dpkt.ip6.IP6.unpack` walks the buffer by. It cannot be taken from
``len()``: ``dpkt.ip6.IP6.__len__`` sums ``len()`` of the extension
headers instead, and that is 12 octets plus the *rest of the frame* for
an Authentication header -- :class:`dpkt.ip6.IP6AHHeader` trims only into
``dpkt.ip6.IP6AHHeader.auth_data`` and leaves
``dpkt.dpkt.Packet.data`` running to the end of the buffer -- and 8
octets short of the wire for a Routing header whose Hdr Ext Len is odd
(:class:`dpkt.ip6.IP6RoutingHeader` keeps whole addresses only).
Every other layer's ``len()`` does bound its payload, so the difference
of the two is used there. For IPv4 that difference is the Internet
Header Length, options included. An Ethernet frame's falls short of the
wire where `DPKT`_ dropped some of its VLAN tags, which
:func:`_layer2bytes` then finds (:issue:`1646`).
"""
if hasattr(packet, 'extension_hdrs'):
return packet.__hdr_len__ + sum(
ext_hdr.length for ext_hdr in _ipv6_ext_hdrs(cast('IP6', packet))
)
return len(packet) - len(payload)
def _layer2bytes(packet: 'Packet', layer: 'Packet') -> 'tuple[bytes, bool]':
"""Fetch the octets of a layer of a DPKT packet, from its first octet on.
Args:
packet: DPKT packet, i.e. the outermost layer.
layer: A layer nested within ``packet``.
Returns:
The octets, and whether they are the captured ones. With a buffer
attached by :func:`attach_buffer` they are the record octets from the
layer's offset to the end of the frame, so any link-layer trailer is
still there for the caller to trim; otherwise they are the layer alone,
serialised from a copy (see :func:`packet2bytes`).
They are that serialisation too if the buffer does not carry the
layer's own header where the walk above placed it. `DPKT`_ unpacks a
header field by field and :meth:`~dpkt.dpkt.Packet.pack_hdr` packs those
fields back without recomputing any of them, so the two agree octet for
octet wherever the offset is right -- which makes this an exact check
rather than a plausibility one. A misplaced slice would reach the
reassembler as a believable datagram, invisible in its output, so it is
never returned. `DPKT`_'s own lengths misplace the layers behind a stack
of VLAN tags it reads in two goes, such as 802.1Q then 802.1ad, as it
keeps only the last go's tags (:issue:`1646`), and behind a Cisco ISL
header followed by an 802.1Q tag, which it keeps in place of the ISL
one.
"""
buffer = getattr(packet, BUFFER_ATTR, None)
if buffer is not None:
offset = 0
current = packet
while current is not layer:
payload = getattr(current, 'data', None)
if payload is None or isinstance(payload, (bytes, bytearray)):
break
offset += _wire_hdr_len(current, payload)
current = payload
else:
wire = cast('bytes', buffer)[offset:]
hdr_len = getattr(layer, '__hdr_len__', 0)
if not hdr_len or wire[:hdr_len] == layer.pack_hdr():
return wire, True
return packet2bytes(layer), False
def _network(packet: 'Packet') -> 'Optional[IP | IP6]':
"""Fetch the network layer of a DPKT packet.
Args:
packet: DPKT packet, i.e. the outermost layer.
Returns:
The packet itself if it is an IPv4 or IPv6 packet, as `DPKT`_ reads every
frame of a raw-IP link type (:issue:`1548`); otherwise the IPv4 or IPv6
payload of its link layer, which `DPKT`_ names ``ip`` or ``ip6`` -- a
bare IP packet names neither -- or :data:`None` if it has none.
"""
import dpkt # isort:skip
if isinstance(packet, (dpkt.ip.IP, dpkt.ip6.IP6)):
return packet
if hasattr(packet, 'ip'):
return cast('IP', packet.ip)
return getattr(packet, 'ip6', None)
[docs]
def ipv6_hdr_len(ipv6: 'IP6') -> 'int':
"""Calculate length of headers before IPv6 Fragment header.
Args:
ipv6: DPKT IPv6 packet.
Returns:
Length of headers before IPv6 Fragment header
:class:`dpkt.ip6.IP6FragmentHeader` (:rfc:`2460#section-4.5`).
As specified in :rfc:`2460#section-4.1`, such a header may be a Hop-by-Hop Options
header :class:`dpkt.ip6.IP6HopOptsHeader` (:rfc:`2460#section-4.3`), a Destination
Options header :class:`dpkt.ip6.IP6DstOptsHeader` (:rfc:`2460#section-4.6`) or a
Routing header :class:`dpkt.ip6.IP6RoutingHeader` (:rfc:`2460#section-4.4`). An
Authentication header (:rfc:`4302`) reaches here as well, though, so the headers are
walked in the order they were parsed rather than looked up by a fixed list of types.
Walking them is also what counts a repeated header type once per occurrence, and what leaves
out one that sits *after* the Fragment header rather than before it.
"""
frag = ipv6.extension_hdrs.get(44)
hdr_len = ipv6.__hdr_len__
for ext_hdr in _ipv6_ext_hdrs(ipv6):
if ext_hdr is frag:
break
hdr_len += ext_hdr.length
return hdr_len
[docs]
def packet2chain(packet: 'Packet') -> 'str':
"""Fetch DPKT packet protocol chain.
Args:
packet: DPKT packet.
Returns:
Colon (``:``) separated list of protocol chain. It ends at the first
payload that is no layer, as in :func:`packet2dict`, e.g. VRRP's ``''``.
"""
chain = [type(packet).__name__]
payload = packet.data
while hasattr(payload, '__hdr_fields__'):
chain.append(type(payload).__name__)
payload = payload.data
return ':'.join(chain)
[docs]
def packet2dict(packet: 'Packet', timestamp: 'float', *,
data_link: 'Enum_LinkType') -> 'dict[str, Any]':
"""Convert DPKT packet into :obj:`dict`.
Args:
packet: DPKT packet.
timestamp: Timestamp of packet.
data_link: Data link type.
Returns:
A :obj:`dict` mapping of packet data. A layer's payload is mapped as a
layer of its own only where it is one, i.e. has the header fields a
:class:`dpkt.dpkt.Packet` has; any other is leaf data.
`DPKT`_ leaves some undecoded payloads as a :obj:`str` rather than as
:obj:`bytes`, e.g. ``''`` for VRRP's. A frame the engine keeps as raw
data, as it keeps one `DPKT`_ cannot decode, has no header fields, so
maps to an empty :obj:`dict`.
"""
def wrapper(packet: 'Packet') -> 'dict[str, Any]':
dict_ = {} # type: dict[str, Any]
for field in getattr(packet, '__hdr_fields__', ()):
dict_[field] = getattr(packet, field, None)
payload = packet.data
if hasattr(payload, '__hdr_fields__'):
dict_[type(payload).__name__] = wrapper(payload)
return dict_
return {
'timestamp': timestamp,
'packet': packet2bytes(packet),
data_link.name: wrapper(packet),
}
[docs]
def ipv4_reassembly(packet: 'Packet', timestamp: 'float', *,
count: 'int' = -1) -> 'IP_Packet[IPv4Address] | None':
"""Make data for IPv4 reassembly.
A packet can be reassembled if its network layer is IPv4 (:class:`dpkt.ip.IP`) --
one carried in IPv6 is not read -- whose header was captured whole, and the **DF**
(:attr:`dpkt.ip.IP.df`) flag is :data:`False`.
Args:
packet: DPKT packet.
timestamp: Capture timestamp of the packet, which drives the reassembly
timeout. `DPKT`_'s reader yields it beside the record's octets rather
than on the packet, so it is passed in -- as :func:`tcp_traceflow`
does. A caller holding only a frame can read it back with
:func:`packet2timestamp`, which is where
:class:`~pcapkit.foundation.engines.dpkt.DPKT` leaves it. Handed on
as a :obj:`float`, though `DPKT`_ reads
a nanosecond PCAP's as a :class:`~decimal.Decimal` (:issue:`1501`),
and as the one nearest the exact value it carries, if any, where
`DPKT`_'s own may be an ulp off (:issue:`1668`; see
:func:`_capture_time`).
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`.
Note:
The header and payload are sliced out of the record octets
:class:`~pcapkit.foundation.engines.dpkt.DPKT` attached to the frame, so
they are the captured ones. A packet built by hand carries no such
buffer, and its octets are then |dpkt|_'s re-serialisation of it, in
which a zeroed checksum or length comes back recomputed --
:func:`attach_buffer` is what makes them the wire's again. They are that
re-serialisation too where |dpkt|_'s own lengths misplace the layer
in the record, as behind some stacks of VLAN tags (:issue:`1646`). The
packet itself is left untouched either way.
The header is the IHL's octets, and the payload ends at the Total
Length, or at the end of the frame for a zero one. A Total Length
shorter than the IHL leaves the header whole and the payload empty
(:issue:`1616`).
See Also:
:class:`pcapkit.foundation.reassembly.ipv4.IPv4`
.. versionchanged:: 1.5.0
``timestamp`` is a new second positional parameter, so a 1.4.x call
with ``packet`` alone fails.
"""
network = _network(packet)
# an IPv6 layer is the one with extension headers, as in ``_wire_hdr_len``
ipv4 = None if hasattr(network, 'extension_hdrs') else network
if ipv4 is not None:
if ipv4.df: # dismiss not fragmented packet
return None
# internet header length, in octets -- ``IP.hl`` counts 32-bit words and
# covers any IP options, whereas ``IP.__hdr_len__`` is the fixed 20-octet
# struct size and would leave option octets at the head of the payload
ihl = ipv4.hl * 4
wire, captured = _layer2bytes(packet, ipv4)
# a zero Total Length is what TCP segmentation offload leaves behind, in
# which case the datagram runs to the end of the frame, as with DPKT; nor
# does one shorter than the IHL cut into the header (#1616)
if captured and ipv4.len:
wire = wire[:max(ipv4.len, ihl)]
# DPKT decodes a header whose IHL runs past the captured octets, but
# there is no whole header to hand over
if len(wire) < ihl:
return None
data = IP_Packet(
bufid=(
cast('IPv4Address',
ipaddress.ip_address(ipv4.src)), # source IP address
cast('IPv4Address',
ipaddress.ip_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.offset * 8, # fragment offset
ihl=ihl, # internet header length
mf=bool(ipv4.mf), # more fragment flag
tl=ipv4.len, # total length, header includes
header=wire[:ihl], # raw bytes type header
payload=bytearray(wire[ihl:]), # raw bytearray type payload
timestamp=_capture_time(timestamp), # capture timestamp
)
return data
return None
[docs]
def ipv6_reassembly(packet: 'Packet', timestamp: 'float', *,
count: 'int' = -1) -> 'IP_Packet[IPv6Address] | None':
"""Make data for IPv6 reassembly.
A packet can be reassembled if its network layer is IPv6 (:class:`dpkt.ip6.IP6`) --
one carried in IPv4 is not read -- and has an IPv6 Fragment header
(:rfc:`2460#section-4.5`, i.e., :class:`dpkt.ip6.IP6FragmentHeader`).
Args:
packet: DPKT packet.
timestamp: Capture timestamp of the packet, which drives the reassembly
timeout; :func:`packet2timestamp` reads it back off a stored frame.
Handed on as a :obj:`float`, as for :func:`ipv4_reassembly`.
count: Packet index. If not provided, defaults to ``-1``.
Returns:
Data for IPv6 reassembly (:term:`reasm.ipv6.packet`) if the ``packet`` can be
reassembled; otherwise :data:`None`.
Note:
The header and payload are sliced out of the record octets
:class:`~pcapkit.foundation.engines.dpkt.DPKT` attached to the frame, so
they are the captured ones. A packet built by hand carries no such
buffer, and its octets are then |dpkt|_'s re-serialisation of it, in
which a zeroed checksum or length comes back recomputed --
:func:`attach_buffer` is what makes them the wire's again. They are that
re-serialisation too where |dpkt|_'s own lengths misplace the layer
in the record, as behind some stacks of VLAN tags (:issue:`1646`). The
packet itself is left untouched either way.
See Also:
:class:`pcapkit.foundation.reassembly.ipv6.IPv6`
.. versionchanged:: 1.5.0
``timestamp`` is a new second positional parameter, so a 1.4.x call
with ``packet`` alone fails.
"""
network = _network(packet)
ipv6 = cast('IP6', network) if hasattr(network, 'extension_hdrs') else None
if ipv6 is not None:
# dpkt keeps each extension header under its protocol number, so 44 is a
# Fragment header; the dict's value type cannot say so
ipv6_frag = cast('Optional[IP6FragmentHeader]', ipv6.extension_hdrs.get(44))
if ipv6_frag is None: # dismiss not fragmented packet
return None
hdr_len = ipv6_hdr_len(ipv6)
wire, captured = _layer2bytes(packet, ipv6)
# a zero Payload Length is a jumbogram or TCP segmentation offload, in
# which case the datagram runs to the end of the frame, as with DPKT
if captured and ipv6.plen:
wire = wire[:ipv6.__hdr_len__ + ipv6.plen]
# payload following the IPv6 Fragment header
payload = wire[hdr_len + ipv6_frag.__hdr_len__:]
data = IP_Packet(
bufid=(
cast('IPv6Address',
ipaddress.ip_address(ipv6.src)), # source IP address
cast('IPv6Address',
ipaddress.ip_address(ipv6.dst)), # destination IP address
# NOTE: The reassembly key is the Fragment header's Identification
# (:rfc:`8200#section-4.5`), not the IPv6 header's Flow Label. The
# label is optional and routinely zero, so keying on it collapses
# every datagram between one address pair into a single buffer and
# interleaves their fragments.
ipv6_frag.id, # identification
Enum_TransType.get(ipv6_frag.nxt), # next header field in IPv6 Fragment Header
),
num=count, # original packet range number
# NOTE: ``IP6FragmentHeader.frag_off`` is a ``__bit_fields__`` property
# over the 13-bit on-wire Fragment Offset, i.e. already shifted out of
# the flags word, so it counts 8-octet units (:rfc:`8200#section-4.5`).
# The reassembly machinery indexes the datagram buffer with ``fo``, so
# the units have to become octets here, exactly as
# :func:`pcapkit.toolkit.scapy.ipv6_reassembly` does.
fo=ipv6_frag.frag_off * 8, # fragment offset
ihl=hdr_len, # header length, only headers before IPv6-Frag
mf=bool(ipv6_frag.m_flag), # more fragment flag
tl=hdr_len + len(payload), # total length, header includes
header=wire[:hdr_len], # raw bytes type header before IPv6-Frag
payload=bytearray(payload), # raw bytearray type payload after IPv6-Frag
timestamp=_capture_time(timestamp), # capture timestamp
)
return data
return None
def _tcp_flags(tcp: 'TCP') -> 'tuple[bool, bool, bool]':
"""Read the RST, SYN and FIN flags of a DPKT TCP segment.
Args:
tcp: DPKT TCP packet.
Returns:
Whether RST, SYN and FIN are set, in that order.
Note:
Each flag is masked out of :attr:`dpkt.tcp.TCP.flags` by its ``TH_*``
value, so it is read from the flags octet alone. `DPKT`_ 1.9.8 makes
that field nine bits wide, its top bit the NS bit (:rfc:`3540`): read by
its position among the field's binary digits, every flag was one bit low
with NS set, a SYN reading as a FIN (:issue:`1618`).
"""
from dpkt.tcp import TH_FIN, TH_RST, TH_SYN # isort:skip
flags = tcp.flags
return bool(flags & TH_RST), bool(flags & TH_SYN), bool(flags & TH_FIN)
[docs]
def tcp_reassembly(packet: 'Packet', timestamp: 'float', *,
count: 'int' = -1) -> 'TCP_Packet | None':
"""Make data for TCP reassembly.
A packet can be reassembled if its network layer carries a TCP layer
(:class:`dpkt.tcp.TCP`), behind any IPv6 extension headers |dpkt|_ decodes. TCP
in an IP-in-IP tunnel, or behind an AH or IPv6 extension header carried in IPv4, is
not read.
Args:
packet: DPKT packet.
timestamp: Capture timestamp of the packet, which drives the reassembly
timeout; :func:`packet2timestamp` reads it back off a stored frame.
Handed on as a :obj:`float`, as for :func:`ipv4_reassembly`.
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`.
Note:
The header and payload are sliced out of the record octets
:class:`~pcapkit.foundation.engines.dpkt.DPKT` attached to the frame, so
they are the captured ones. A packet built by hand carries no such
buffer, and its octets are then |dpkt|_'s re-serialisation of it, in
which a zeroed checksum or length comes back recomputed --
:func:`attach_buffer` is what makes them the wire's again. They are that
re-serialisation too where |dpkt|_'s own lengths misplace the layer
in the record, as behind some stacks of VLAN tags (:issue:`1646`). The
packet itself is left untouched either way.
See Also:
:class:`pcapkit.foundation.reassembly.tcp.TCP`
.. versionchanged:: 1.5.0
``timestamp`` is a new second positional parameter, so a 1.4.x call
with ``packet`` alone fails.
"""
ip = _network(packet)
if ip is None:
return None
tcp = getattr(ip, 'tcp', None) # type: Optional[TCP]
if tcp is None and type(getattr(ip, 'data', None)).__name__ == 'TCP':
tcp = cast('TCP', ip.data)
if tcp is not None:
rst, syn, fin = _tcp_flags(tcp)
wire, _ = _layer2bytes(packet, tcp)
raw_len = len(tcp.data) # payload length, header excludes
data = TCP_Packet(
bufid=(
ipaddress.ip_address(ip.src), # source IP address
tcp.sport, # source port
ipaddress.ip_address(ip.dst), # destination IP address
tcp.dport, # destination port
),
num=count, # original packet range number
ack=tcp.ack, # acknowledgement
dsn=tcp.seq, # data sequence number
rst=rst, # reset connection flag
syn=syn, # synchronise flag
fin=fin, # finish flag
header=wire[:tcp.off * 4], # raw bytes type header
payload=bytearray(bytes(tcp.data)), # raw bytearray type payload
first=tcp.seq, # first sequence number of payload
last=tcp.seq + raw_len - 1, # last sequence number of payload
len=raw_len, # payload length, header excludes
timestamp=_capture_time(timestamp), # capture timestamp
)
return data
return None
[docs]
def tcp_traceflow(packet: 'Packet', timestamp: 'float', *,
data_link: 'Enum_LinkType', count: 'int' = -1) -> 'TF_TCP_Packet | None':
"""Trace packet flow for TCP.
A packet can be traced if its network layer carries a TCP layer
(:class:`dpkt.tcp.TCP`), behind any IPv6 extension headers |dpkt|_ decodes. TCP
in an IP-in-IP tunnel, or behind an AH or IPv6 extension header carried in IPv4, is
not read.
Args:
packet: DPKT packet.
timestamp: Timestamp of the packet, handed on as a :obj:`float`, as for
:func:`ipv4_reassembly`. The PCAP frame record is built from the
exact timestamp attached by :func:`attach_timestamp`, if any; see
:func:`packet2frame`.
data_link: Data link layer protocol (from global 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`.
Note:
The header and payload are sliced out of the record octets
:class:`~pcapkit.foundation.engines.dpkt.DPKT` attached to the frame, so
they are the captured ones. A packet built by hand carries no such
buffer, and its octets are then |dpkt|_'s re-serialisation of it, in
which a zeroed checksum or length comes back recomputed --
:func:`attach_buffer` is what makes them the wire's again. They are that
re-serialisation too where |dpkt|_'s own lengths misplace the layer
in the record, as behind some stacks of VLAN tags (:issue:`1646`). The
packet itself is left untouched either way.
See Also:
:class:`pcapkit.foundation.traceflow.tcp.TCP`
"""
ip = _network(packet)
if ip is None:
return None
tcp = getattr(ip, 'tcp', None) # type: Optional[TCP]
if tcp is None and type(getattr(ip, 'data', None)).__name__ == 'TCP':
tcp = cast('TCP', ip.data)
if tcp is not None:
rst, syn, fin = _tcp_flags(tcp)
wire, _ = _layer2bytes(packet, tcp)
data = TF_TCP_Packet( # type: ignore[type-var]
protocol=data_link, # data link type from global header
index=count, # frame number
frame=packet2frame(packet, timestamp, data_link=data_link), # extracted packet
syn=syn, # TCP synchronise (SYN) flag
fin=fin, # TCP finish (FIN) flag
rst=rst, # TCP reset (RST) flag
src=ipaddress.ip_address(ip.src), # source IP
dst=ipaddress.ip_address(ip.dst), # destination IP
srcport=tcp.sport, # TCP source port
dstport=tcp.dport, # TCP destination port
timestamp=_capture_time(timestamp), # timestamp
seq=tcp.seq, # TCP sequence number
ack=tcp.ack, # TCP acknowledgement number
header=wire[:tcp.off * 4], # raw bytes type header
payload=bytearray(bytes(tcp.data)), # raw bytearray type payload
)
return data
return None