# -*- coding: utf-8 -*-
"""IPv6 Datagram Reassembly
==============================
.. module:: pcapkit.foundation.reassembly.ipv6
:mod:`pcapkit.foundation.reassembly.ipv6` contains
:class:`~pcapkit.foundation.reassembly.ipv6.IPv6`
only, which reconstructs fragmented IPv6 packets back to
origin. Please refer to :doc:`ip` for more information.
"""
from typing import TYPE_CHECKING
from pcapkit.const.reg.transtype import TransType as Enum_TransType
from pcapkit.foundation.reassembly.ip import IP
from pcapkit.protocols.internet.ipv6 import IPV6_HEADER_LEN, IPV6_NEXT_HEADER_OFFSET
from pcapkit.protocols.internet.ipv6 import IPv6 as IPv6_Protocol
if TYPE_CHECKING:
from pcapkit.const.reg.transtype import TransType
__all__ = ['IPv6']
def _next_header_offset(header: 'bytes') -> 'int':
"""Locate the Next Header field of a datagram's last header.
:rfc:`8200#section-4.5` gives the Next Header field of the *last* header of
the unfragmentable part -- not necessarily the fixed IPv6 header's, since
Hop-by-Hop Options, Routing and Destination Options headers may precede the
Fragment header. Each of those starts with its own Next Header field, so the
answer is found by walking the chain to its end.
Args:
header: Unfragmentable part of a datagram, i.e. every octet of the
fragment before its Fragment header.
Returns:
Offset, within ``header``, of the Next Header field to rewrite.
"""
offset = IPV6_NEXT_HEADER_OFFSET
position = IPV6_HEADER_LEN
proto = header[offset]
# NOTE: ``header[position]`` is the Next Header field of the extension
# header starting at ``position`` and ``header[position + 1]`` its length,
# but which of the two length encodings applies is decided by the *previous*
# header's Next Header value -- hence ``proto`` trailing one step behind.
while position + 1 < len(header):
offset = position
# NOTE: AH is the one extension header that does not measure its
# length in 8-octet units (:rfc:`4302#section-2.2`), which is why it
# is singled out rather than falling to the common case below.
if proto == Enum_TransType.AH:
position += (header[position + 1] + 2) * 4
else:
position += (header[position + 1] + 1) * 8
proto = header[offset]
return offset
# BUG: It is supposed to be ``IP[IPv6Address]``. But somehow Python
# thinks that ``IP`` should take 4 arguments as in its parent class
# ``Reassembly``. So we have to drop the type hint here.
[docs]
class IPv6(IP):
"""Reassembly for IPv6 payload.
Args:
strict: if :data:`True`, report a datagram that is not completely
reassembled as the tuple of its received runs; otherwise as one
contiguous payload, its holes zero-filled -- or, while the total
length is unknown, only the prefix up to the first hole
store: if store reassembled datagram in memory, i.e.,
:attr:`self._dtgram <pcapkit.foundation.reassembly.reassembly.Reassembly._dtgram>`
(if not, datagram will be discarded after callback)
timeout: reassembly timeout in seconds, on the capture's own clock;
:data:`None` selects :attr:`__timeout__`
Example:
>>> from pcapkit.foundation.reassembly import IPv6
# Initialise instance:
>>> ipv6_reassembly = IPv6()
# Call reassembly:
>>> ipv6_reassembly(packet_dict)
# Fetch result:
>>> result = ipv6_reassembly.datagram
.. versionchanged:: 1.5.0
The reassembled ``header`` leaves out the Fragment header and carries
its Next Header, per :rfc:`8200#section-4.5`, whichever engine fed it;
1.4.x kept the Fragment header on the ``default`` and ``pcapng``
engines (:issue:`415`).
"""
##########################################################################
# Defaults.
##########################################################################
#: Protocol name of current reassembly object.
__protocol_name__ = 'IPv6'
#: Protocol of current reassembly object.
__protocol_type__ = IPv6_Protocol
#: float: Default reassembly timeout, in seconds, on the capture's clock.
#:
#: :rfc:`8200#section-4.5` is unambiguous and mandatory: "If insufficient
#: fragments are received to complete reassembly of a packet within 60
#: seconds of the reception of the first-arriving fragment of that packet,
#: reassembly of that packet must be abandoned and all the fragments that
#: have been received for that packet must be discarded."
#:
#: Note "first-arriving", not "first" -- the deadline is counted from
#: whichever fragment opened the buffer, not from the one at fragment offset
#: zero, which is why
#: :attr:`Buffer.timestamp <pcapkit.foundation.reassembly.data.ip.Buffer.timestamp>`
#: is set once and never revised. The offset-zero fragment matters only to
#: the second half of the rule, the ICMP Time Exceeded that a live stack
#: "should" send and that an offline parser cannot -- see
#: :meth:`~pcapkit.foundation.reassembly.reassembly.Reassembly.expire`,
#: which reports the condition in its log record instead.
__timeout__ = 60.0
##########################################################################
# Methods.
##########################################################################
def _rectify_header(self, header: 'bytes', proto: 'TransType') -> 'bytes':
"""Remove the Fragment header from a datagram's header chain.
:rfc:`8200#section-4.5` states that the Fragment header is not present in
the reassembled packet, and that the Next Header field of the last header
of the unfragmentable part comes from the Fragment header's. Left alone,
the reassembled datagram would advertise a Fragment header on a datagram
that is by definition no longer a fragment. The toolkit adapters
that build IPv6 reassembly data, :mod:`~pcapkit.toolkit.pcap`,
:mod:`~pcapkit.toolkit.pcapng`, :mod:`~pcapkit.toolkit.dpkt` and
:mod:`~pcapkit.toolkit.scapy`, leave the Fragment header's *octets* out
of ``header``, but none rewrites the field that points at it, so only
that field is left to fix.
Args:
header: Raw header octets of the fragment at fragment offset zero.
proto: Payload protocol type, i.e. the Fragment header's Next Header
field, which is what the rewritten field must carry.
Returns:
Header octets to keep for the reassembled datagram.
"""
# a header too short to hold the fixed IPv6 header cannot be walked, and
# a chain not ending in the Fragment header has nothing to rewrite --
# which also makes this idempotent
if len(header) < IPV6_HEADER_LEN:
return header
offset = _next_header_offset(header)
if header[offset] != Enum_TransType.IPv6_Frag:
return header
return header[:offset] + bytes((int(proto),)) + header[offset + 1:]
def _header_overhead(self, length: 'int') -> 'int':
"""Count the header octets that the Payload Length includes.
The Payload Length counts every octet after the fixed IPv6 header
(:rfc:`8200#section-3`), so a reassembled packet's is its unfragmentable
part's extension headers plus its data (:rfc:`8200#section-4.5`). The
header here is the unfragmentable part, its Fragment header excluded.
Args:
length: Length of the datagram's header.
Returns:
How many of those octets the Payload Length counts.
"""
return max(length - IPV6_HEADER_LEN, 0)