# -*- coding: utf-8 -*-
"""container field class"""
import io
from typing import TYPE_CHECKING, Generic, TypeVar, cast
from pcapkit.corekit.fields.field import FieldBase
from pcapkit.corekit.fields.misc import SchemaField
from pcapkit.corekit.fields.numbers import NumberField
from pcapkit.corekit.multidict import OrderedMultiDict
from pcapkit.utilities.compat import List
from pcapkit.utilities.exceptions import FieldValueError, ProtocolError
__all__ = [
'ListField', 'OptionField',
]
if TYPE_CHECKING:
from collections import defaultdict
from enum import IntEnum as StdlibEnum
from typing import IO, Any, Callable, Optional, Type
from aenum import IntEnum as AenumEnum
from typing_extensions import Self
from pcapkit.protocols.schema.schema import Schema
_TL = TypeVar('_TL', 'Schema', 'FieldBase', 'bytes')
_TS = TypeVar('_TS', bound='Schema')
class _ShortReadTracker:
"""Stream proxy recording whether a read came back short.
Args:
file: Stream to read from.
Only the ``read``, ``seek`` and ``tell`` calls that
:meth:`Schema.unpack <pcapkit.protocols.schema.schema.Schema.unpack>` and
:meth:`OptionField.unpack` make are delegated.
"""
__slots__ = ('_file', 'short')
def __init__(self, file: 'IO[bytes]') -> 'None':
self._file = file
#: bool: Whether a read since the last reset returned fewer octets than
#: it asked for.
self.short = False
def read(self, size: 'Optional[int]' = -1) -> 'bytes':
"""Read ``size`` octets, noting a short read."""
if size is None:
size = -1
data = self._file.read(size)
if 0 <= size and len(data) < size:
self.short = True
return data
def seek(self, offset: 'int', whence: 'int' = io.SEEK_SET) -> 'int':
"""Seek the underlying stream."""
return self._file.seek(offset, whence)
def tell(self) -> 'int':
"""Tell the underlying stream's position."""
return self._file.tell()
[docs]
class ListField(FieldBase[List[_TL]], Generic[_TL]):
"""Field list for protocol fields.
Args:
length: Field size (in bytes); if a callable is given, it should return
an integer value and accept the current packet as its only argument.
item_type: Field type of the contained items.
callback: Callback function to be called upon
:meth:`self.__call__ <pcapkit.corekit.fields.field.FieldBase.__call__>`.
This field is used to represent a list of fields, as in the case of lists of
constant-length-field items in a protocol.
"""
@property
def length(self) -> 'int':
"""Field size."""
return self._length
@property
def optional(self) -> 'bool':
"""Field is optional."""
return True
def __init__(self, length: 'int | Callable[[dict[str, Any]], int]' = lambda _: -1,
item_type: 'Optional[FieldBase]' = None,
callback: 'Callable[[Self, dict[str, Any]], None]' = lambda *_: None) -> 'None':
#self._name = '<list>'
if not hasattr(self, '_name'):
self._name = f'<{type(self).__name__[:-5].lower()}>'
self._callback = callback
self._item_type = item_type
self._length_callback = None
if not isinstance(length, int):
self._length_callback, length = length, -1
self._length = length
self._template = '0s'
[docs]
def __call__(self, packet: 'dict[str, Any]') -> 'Self':
"""Update field attributes.
Args:
packet: Packet data.
Returns:
Updated field instance.
This method will return a new instance of :class:`ListField`
instead of updating the current instance.
"""
new_self = self.__copy__()
new_self._callback(self, packet)
if new_self._length_callback is not None:
new_self._length = new_self._length_callback(packet)
new_self._template = f'{new_self._length}s'
return new_self
[docs]
def pack(self, value: 'Optional[list[_TL]]', packet: 'dict[str, Any]') -> 'bytes':
"""Pack field value into :obj:`bytes`.
Args:
value: Field value.
packet: Packet data.
Returns:
Packed field value.
Without an ``item_type``, :meth:`unpack` returns the field's octets as
a single :obj:`bytes`, so such a value is written as-is rather than
iterated item by item.
"""
if value is None:
return b''
if self._item_type is None and isinstance(value, bytes):
return value
from pcapkit.protocols.schema.schema import \
Schema # pylint: disable=import-outside-top-level
temp = [] # type: list[bytes]
for item in value:
if isinstance(item, bytes):
temp.append(item)
elif isinstance(item, Schema):
temp.append(item.pack(packet))
elif self._item_type is not None:
temp.append(self._item_type.pack(item, packet))
else:
raise FieldValueError(f'Field {self.name} has invalid value.')
return b''.join(temp)
[docs]
def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> 'bytes | list[_TL]':
"""Unpack field value from :obj:`bytes`.
Args:
buffer: Field buffer.
packet: Packet data.
Returns:
Unpacked field value.
Raises:
FieldValueError: If the items overrun the field, or if a schema item
consumes nothing from ``buffer`` -- see the note below.
"""
length = self._length
if isinstance(buffer, bytes):
file = io.BytesIO(buffer) # type: IO[bytes]
else:
file = buffer
if self._item_type is None:
return file.read(length)
# NOTE: ``SchemaField`` is imported at module level rather than inside
# this method body -- there is no cyclic import to dodge by doing so;
# :mod:`pcapkit.corekit.fields.misc` does not import this module. A local
# ``from ... import`` re-resolves against :data:`sys.modules` on *every*
# call, and normally that is a cheap dict lookup returning the same
# cached module -- except when something has popped
# :mod:`pcapkit.corekit.fields.misc` out of :data:`sys.modules` since
# this ``ListField`` was built (the ``#439`` regression tests do exactly
# that, to get a clean ABC-cache state between cases). The next call
# would then re-execute the module from scratch and mint a *second*,
# distinct ``SchemaField`` class, while ``self._item_type`` -- built long
# before, from the first one -- is still an instance of the original.
# ``isinstance`` against the new class would report :data:`False` for an
# item that plainly is a ``SchemaField``, ``is_schema`` would go the
# wrong way, and this method's other branch would bill each item by its
# *declared* length instead of by what it actually consumed -- so a
# malformed TCP SACK option's length check could raise
# :exc:`~pcapkit.utilities.exceptions.FieldValueError` from here instead
# of :exc:`~pcapkit.utilities.exceptions.ProtocolError` from
# :meth:`TCP._read_mode_sack
# <pcapkit.protocols.transport.tcp.TCP._read_mode_sack>`, depending on
# unrelated things that ran earlier in the same process. C.f. #525.
# Importing at module level, once, for the life of this module object,
# closes that off: ``SchemaField`` then stays whatever class this
# module resolved at its own import time, regardless of what happens to
# :mod:`pcapkit.corekit.fields.misc`'s entry in :data:`sys.modules`
# afterwards.
is_schema = isinstance(self._item_type, SchemaField)
# NOTE: The item-typed branch below sizes each item by ``field.length``,
# which is what it read, but the schema branch sizes it by ``len(data)``,
# which is only what the schema *recorded*. A schema reading a stream that
# has already run out records nothing, so ``length -= len(data)`` makes no
# progress and the loop spins forever. Reachable from a TCP segment: a
# ``SACK`` option declaring more octets than the option area holds leaves
# ``sack``'s ``ListField`` reading ``SACKBlock`` off an exhausted stream.
# Remembering where the previous item ended is what bounds the iteration
# count, since it does not depend on what the schema reports. C.f. #431,
# which is the same defect in the ``OptionField`` subclass.
#
# ``start`` is where the field itself begins. The comparison needs stream
# positions, but the diagnostic wants an offset into the field, and the two
# only coincide when the field happens to be reading from the front of its
# stream -- which it does when handed a :obj:`bytes` buffer and does not
# when handed a live file.
start = offset = file.tell()
temp = [] # type: list[_TL]
while length > 0:
field = self._item_type(packet)
if is_schema:
data = cast('SchemaField', field).unpack(file, packet)
end = file.tell()
if end <= offset:
# NOTE: ``len(temp)`` counts the items already parsed, so it
# names the failing one as a count rather than as an ordinal --
# "after 2 item(s)" rather than "item 2", which would read as
# the second item when it is the third. The ``OptionField``
# message below names the option code in this slot and so has
# no index to be read either way.
raise FieldValueError(
f'Field {self.name} has an item that consumed no data: '
f'after {len(temp)} item(s), at offset {offset - start} of '
f'{self._length}, with {length} octet(s) of the field '
f'left to parse'
)
offset = end
length -= len(data)
if length < 0:
raise FieldValueError(f'Field {self.name} has invalid length.')
else:
length -= field.length
if length < 0:
raise FieldValueError(f'Field {self.name} has invalid length.')
buffer = file.read(field.length)
data = field.unpack(buffer, packet)
temp.append(data)
return temp
[docs]
class OptionField(ListField, Generic[_TS]):
"""Field list for protocol options.
Args:
length: Field size (in bytes); if a callable is given, it should return
an integer value and accept the current packet as its only argument.
base_schema: Base schema for option fields.
type_name: Name of the option type field.
registry: Option registry, as in a mapping from option types (enumeration
values) to option schemas, with the default value being the unknown
option schema.
eool: Enumeration of the EOOL (end-of-option-list, or equivalent) option
callback: Callback function to be called upon
:meth:`self.__call__ <pcapkit.corekit.fields.field.FieldBase.__call__>`.
This field is used to represent a list of fields, as in the case of lists of
options and/or parameters in a protocol.
Note:
:meth:`self.unpack <unpack>` selects an option's schema by reading the
``type_name`` field of ``base_schema`` **alone** off the front of the
option, instead of unpacking the whole base schema and keeping only that
one value. That is only the same read when the following hold of the type
field:
1. it is the base schema's **first** field, so that it is what sits at the
front of the option;
2. it is a :class:`~pcapkit.corekit.fields.numbers.NumberField`, so that
:meth:`Schema.unpack <pcapkit.protocols.schema.schema.Schema.unpack>`
reads it through its ordinary per-field branch;
3. its length is a fixed integer rather than a callable, so that its width
does not depend on packet data the shortcut has not read.
All of that is true of every ``OptionField`` declared in this package. A
base schema registered from outside it -- c.f.
:mod:`pcapkit.foundation.registry` -- need not satisfy it, and is **not**
rejected: such a base schema is unpacked in full. It parses correctly and
pays the cost of the second parse. Only the shortcut is withheld, so a base schema whose type field
comes second cannot be silently misread.
"""
@property
def base_schema(self) -> 'Type[_TS]':
"""Base schema."""
return self._base_schema
@property
def type_name(self) -> 'str':
"""Type name."""
return self._type_name
@property
def registry(self) -> 'defaultdict[int | StdlibEnum | AenumEnum, Type[_TS]]':
"""Option registry."""
return self._registry
@property
def eool(self) -> 'int | StdlibEnum | AenumEnum':
"""EOOL option."""
return self._eool
@property
def option_padding(self) -> 'int':
"""Length option padding data."""
return self._option_padding
def __init__(self, length: 'int | Callable[[dict[str, Any]], int]' = lambda _: -1,
base_schema: 'Optional[Type[_TS]]' = None,
type_name: 'str' = 'type',
registry: 'Optional[defaultdict[int | StdlibEnum | AenumEnum, Type[_TS]]]' = None,
eool: 'Optional[int | StdlibEnum | AenumEnum]' = None,
callback: 'Callable[[Self, dict[str, Any]], None]' = lambda *_: None) -> 'None':
super().__init__(length, None, callback)
#self._name = '<option>'
self._eool = eool
self._option_padding = 0
if base_schema is None:
raise FieldValueError('Field <option> has no base schema.')
self._base_schema = base_schema
if not hasattr(self._base_schema, type_name):
raise FieldValueError(f'Field <option> has no type field "{type_name}".')
self._type_name = type_name
if registry is None:
raise FieldValueError('Field <option> has no registry.')
self._registry = registry
# NOTE: Decided once, here, rather than per option in ``unpack``. See the
# docstring above for what the fast path requires and why a base schema
# that does not meet it is accommodated rather than rejected.
#
# The conditions above are exactly what makes reading the type field alone
# the same read that ``Schema.unpack`` performs for it: it must sit at the
# front of the option, ``Schema.unpack`` must handle it through its generic
# branch rather than one of the special ones (a ``NumberField`` is never
# ``PayloadField``, ``PaddingField``, ``ConditionalField``,
# ``ForwardMatchField``, ``SwitchField`` or ``OptionField``), and its width
# must not depend on packet state that the full unpack would have
# established first.
#
# Every test is written so that an unexpected base schema selects the full
# unpack instead of raising, so this cannot turn a working registration
# into an import-time failure.
fields = getattr(self._base_schema, '__fields__', {})
type_field = fields.get(type_name)
#: Optional[FieldBase]: Base schema's type field, when reading it on its own
#: is equivalent to unpacking the whole base schema to obtain it; otherwise
#: :data:`None`, and :meth:`self.unpack <unpack>` unpacks the base schema.
self._type_field = type_field if (
isinstance(type_field, NumberField)
and list(fields)[:1] == [type_name]
and type_field._length_callback is None # pylint: disable=protected-access
) else None
[docs]
def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> 'list[_TS]':
"""Unpack field value from :obj:`bytes`.
Args:
buffer: Field buffer.
packet: Packet data.
Returns:
Unpacked field value.
Important:
If the option list ended before the specified size limit,
set :attr:`self.option_padding <OptionField.option_padding>`
as the remaining length to the ``packet`` argument such that
the next fields can be aware of such informations.
Raises:
FieldValueError: If an option consumes nothing from ``buffer``, since
the loop below has then no way to get past it.
ProtocolError: If an option that begins with octets present reads
past the end of the data.
.. versionchanged:: 1.5.0
An option that begins with octets present and reads past the end of
the data raises :exc:`~pcapkit.utilities.exceptions.ProtocolError`,
where it was parsed from zero-filled octets (:issue:`1325`).
"""
length = self._length
if isinstance(buffer, bytes):
file = io.BytesIO(buffer) # type: IO[bytes]
else:
file = buffer
# NOTE: The loop below sizes each option by ``len(data)`` -- the size of
# the schema the option reported -- and that is not always the number of
# octets the option took from ``file``. The two part company for a schema
# whose ``post_process`` returns a *nested* schema, since ``len(data)``
# then measures the nested schema rather than what the outer one read. So
# ``len(data)`` cannot be the loop's progress measure: an option that
# over-reads leaves ``length`` above zero with ``file`` already exhausted,
# every field of the next option reads ``b''``, and an option that read
# nothing reports ``len(data) == 0`` and leaves ``length`` untouched --
# which spins forever, with no exception and no diagnostic. C.f. #431.
#
# Remembering where the previous option ended gives the loop a measure of
# progress that does not depend on what a schema reports, and one octet of
# it per iteration is what bounds the iteration count. ``length`` is
# decremented by ``len(data)`` as well, so the guard changes nothing for
# an option area that parses.
#
# ``start`` is where the option area itself begins. The comparison needs
# stream positions, but the diagnostic wants an offset into the area, and
# the two only coincide when the field happens to be reading from the front
# of its stream -- which it does when handed a :obj:`bytes` buffer and does
# not when handed a live file.
start = offset = file.tell()
# make a copy of the ``packet`` dict so that we can include
# parsed option schema in the ``packet`` dict
new_packet = packet.copy()
new_packet[self.name] = OrderedMultiDict()
# NOTE: Where it can, this reads the base schema's type field alone rather
# than the whole base schema. The option schema below re-reads the same
# octets from the rewound stream, and the base schema's result is used for
# nothing but the type code, so unpacking it in full parses every option
# twice -- 2274 schema unpacks for 1137 options of
# ``examples/captures/profile.pcapng``.
#
# Reading the type field alone deliberately mirrors what
# :meth:`Schema.unpack <pcapkit.protocols.schema.schema.Schema.unpack>`
# does for one field -- ``field(packet)``, read ``field.length`` octets,
# then ``field.unpack(byte, packet.copy())`` -- rather than reaching for
# :func:`struct.unpack`. That keeps ``code``'s enumeration type, which both
# the ``self._eool`` comparison and the ``OrderedMultiDict`` key depend on,
# and it is what makes the two spellings equivalent rather than merely
# similar.
#
# ``self._type_field`` is :data:`None` for a base schema that the shortcut
# does not fit -- see the class docstring -- and the full unpack is used
# for it instead. Whichever way the code was read, ``consumed`` is the
# number of octets to rewind to get back to the front of the option.
type_field = self._type_field
# NOTE: An option whose declared length runs past the octets behind it
# reads short, and every field reading off the end pads the missing
# octets with zeros, so the option would rebuild octets that were never
# there: a HIP ``ACK`` declaring 256 octets with 4 present, or an MH
# ``PadN`` declaring 3 with none. Nothing downstream can tell, because
# the schema records the zeros as values. C.f. #1325.
#
# So the stream is read through a proxy that notes any short read, and
# an option that began with at least one octet present and then read
# short is rejected. An option that begins at the end of the data reads
# nothing at all and is left to the end-of-option-list and progress
# handling below, which is what an option area declared longer than its
# octets relies on.
tracker = file = _ShortReadTracker(file) # type: ignore[assignment]
temp = [] # type: list[_TS]
while length > 0:
tracker.short = False
if type_field is None:
# unpack option type using base schema
meta = self._base_schema.unpack(file, length, packet) # type: ignore[call-arg,misc,var-annotated]
code = cast('int', meta[self._type_name])
consumed = len(meta)
else:
# unpack option type using the base schema's type field. No cast
# is needed here, unlike the branch above: the type field is known
# to be a ``NumberField``, so ``unpack`` is already typed ``int``.
field = type_field(packet)
byte = file.read(field.length)
code = field.unpack(byte, packet.copy())
consumed = len(byte)
schema = self._registry[code]
# rewind to the beginning of the option
file.seek(-consumed, io.SEEK_CUR)
# unpack option using option schema
data = schema.unpack(file, length, packet) # type: ignore[call-arg,misc,var-annotated]
if tracker.short and consumed > 0:
raise ProtocolError(
f'Field {self.name} has an option that runs past the end of '
f'the data: {code!r} at offset {offset - start} of '
f'{self._length}, with only {file.tell() - offset} octet(s) '
f'of it present'
)
new_packet[self.name].add(code, data)
temp.append(data)
# update length
length -= len(data)
# check for EOOL
if code == self._eool:
break
# NOTE: The progress check comes *after* the end-of-option-list break,
# and that order is not incidental. An area declared longer than the
# octets behind it -- an over-long ``ihl``, or a capture cut short by
# the snapshot length -- exhausts ``file`` early, and the exhausted
# read then decodes the type field as 0. For the IPv4, TCP and PCAP-NG
# registries 0 *is* the end-of-option-list code, so the break above
# absorbs that case and reports the rest of the area as padding.
# Checking progress first would turn all of those into errors: measured
# on ``IPv4(bytes.fromhex('4a00001800010000400600000a0000010a000002'))``,
# 20 octets of options declared with none present, and on a TCP segment
# with a data offset of 10 and four option octets, both of which parse.
#
# The registries that spin are the ones where 0 is *not* the
# end-of-option-list code, so they never reach the break: HOPOPT,
# IPv6-Opts and MH read 0 as ``Pad1``, HIP as an unassigned parameter,
# SCTP as a DATA chunk. Those are what this guards, and one octet of
# measured progress per surviving iteration is what bounds the loop.
end = file.tell()
if end <= offset:
raise FieldValueError(
f'Field {self.name} has an option that consumed no data: '
f'{code!r} at offset {offset - start} of {self._length}, with '
f'{length} octet(s) of the option area left to parse'
)
offset = end
self._option_padding = length
return temp