# -*- coding: utf-8 -*-
"""numerical field class
.. versionremoved:: 1.6.0
``pcapkit.corekit.fields.numbers._reduce_unregistered_member``, with no
alias; :func:`pcapkit.utilities.pickle.reduce_to` installs the reduction it
built (:issue:`2714`).
"""
import enum
import math
from typing import TYPE_CHECKING, Generic, TypeVar, Union, cast
import aenum
from pcapkit.corekit.fields.field import Field
from pcapkit.corekit.sentinels import NO_VALUE
from pcapkit.utilities.exceptions import BaseError, FieldValueError, IntError, ProtocolError
from pcapkit.utilities.pickle import reduce_to
__all__ = [
'NumberField',
'Int32Field', 'UInt32Field',
'Int16Field', 'UInt16Field',
'Int64Field', 'UInt64Field',
'Int8Field', 'UInt8Field',
'EnumField', 'PortEnumField',
'BitmapField',
]
if TYPE_CHECKING:
from enum import IntEnum as StdlibEnum
from typing import Any, Callable, Mapping, Optional, Type
from aenum import IntEnum as AenumEnum
from typing_extensions import Literal, Self
from pcapkit.const.reg.apptype import AppType, TransportProtocol
from pcapkit.corekit.sentinels import NoValueType
_T = TypeVar('_T', bound='int')
#: Instance attribute a parsed value records its bits outside
#: :attr:`NumberField.bit_length` under, as ``(outside, length, bit_length,
#: signed, byteorder)``. See GitHub issue :issue:`1488`.
_OUTSIDE_BITS = '_outside_bits_'
[docs]
class _CapturedInt(int):
"""An :obj:`int` that also records the bits outside its field's
:attr:`~NumberField.bit_length` as they were captured.
It equals, hashes and renders as the plain :obj:`int` it holds, and any
arithmetic on it yields a plain :obj:`int`, which is what lets a value a
caller derives from a parsed one keep a plain :obj:`int`'s truncating
semantics. See
GitHub issue :issue:`1488`.
"""
def _capture_key(field: 'NumberField') -> 'tuple[int, int, bool, str]':
"""Return the shape of ``field`` a captured value's bits belong to.
Args:
field: The field packing or unpacking the value.
Returns:
``(length, bit_length, signed, byteorder)``.
"""
return (field._length, field._bit_length, # pylint: disable=protected-access
field._signed, field._byteorder) # pylint: disable=protected-access
def _capture_record(value: 'Any') -> 'Optional[tuple[int, int, int, bool, str]]':
"""Return the capture ``value`` records, or :data:`None` if it records none.
Args:
value: Field value.
Returns:
``(outside, length, bit_length, signed, byteorder)``, or :data:`None`.
The instance dictionary is read rather than :func:`getattr`, so that no
class-level attribute -- an enumeration's member lookup among them -- can
answer for it.
"""
return getattr(value, '__dict__', {}).get(_OUTSIDE_BITS)
[docs]
def _outside_bits(value: 'Any', field: 'NumberField') -> 'Optional[int]':
"""Return the captured bits ``value`` records outside ``field``'s
:attr:`~NumberField.bit_length`, or :data:`None`.
Args:
value: Field value.
field: The field packing it.
Returns:
The bits, in place (not shifted down), if ``value`` was captured by a
field of the same length, bit length, sign and byte order as
``field``; :data:`None` otherwise, so that any other field packs it
exactly as it would the plain value.
"""
record = _capture_record(value)
if record is None or tuple(record[1:]) != _capture_key(field):
return None
return record[0]
[docs]
def _with_outside_bits(value: 'int', record: 'tuple[int, int, int, bool, str]') -> 'int':
"""Return ``value`` recording ``record`` as its captured bits outside
:attr:`~NumberField.bit_length`.
Args:
value: Field value, the low :attr:`~NumberField.bit_length` bits.
record: ``(outside, length, bit_length, signed, byteorder)``.
Returns:
A :class:`_CapturedInt` equal to ``value``.
"""
captured = _CapturedInt(value)
setattr(captured, _OUTSIDE_BITS, record)
return captured
[docs]
class NumberField(Field[int], Generic[_T]):
"""Numerical value 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.
default: Field default value, if any.
signed: Whether the field is signed; :data:`None` defers to the
class-level ``__signed__``, which this class leaves unset and so
means unsigned.
byteorder: Field byte order.
bit_length: Field bit length.
callback: Callback function to be called upon
:meth:`self.__call__ <pcapkit.corekit.fields.field.FieldBase.__call__>`.
Raises:
IntError: If no ``length`` is given and ``__length__`` fixes none either.
FieldValueError: If ``signed`` contradicts a sign already fixed by
``__signed__`` -- never from this class, which fixes none.
ProtocolError: If ``bit_length`` is given negative. Left alone,
``(1 << bit_length) - 1`` raises a bare, uncatchable
:exc:`ValueError` (``negative shift count``) here, before
:meth:`__call__`'s own negative-``length`` guard (:issue:`828`) or
:attr:`~pcapkit.corekit.fields.field.FieldBase.length`'s
(:issue:`805`) ever see anything -- this one fires at construction
time, on the argument itself rather than on a resolved wire length.
See GitHub issue :issue:`831`.
Notes:
A subclass such as :class:`UInt32Field` fixes the sign through
``__signed__``, so ``signed`` there is at best redundant. A
contradicting value is rejected rather than discarded, since
discarding it would make ``UInt32Field(signed=True)`` hand back an
unsigned field whose values only look wrong once the high bit is set
-- see GitHub issue :issue:`545`. Omitting it, or passing the sign the
class already fixes, stays legal.
Packing does not range-check a value: :meth:`pre_process` masks it to
:attr:`bit_length` bits, so e.g. ``70000`` in a two-octet field packs
as ``0x1170`` and ``-1`` as ``0xffff``, with no error or warning. This
is the documented contract a parsed value's round trip relies on
(:issue:`1488`). A range check belongs in the protocol's ``make`` or in
a field whose values have a range of their own, such as
:class:`PortEnumField` (:issue:`2314`).
"""
__length__ = None # type: Optional[int]
__template__ = None # type: Optional[str]
__signed__ = None # type: Optional[bool]
@property
def bit_length(self) -> 'int':
"""Field bit length."""
return self._bit_length
def __init__(self, length: 'Optional[int | Callable[[dict[str, Any]], int]]' = None,
default: 'int | NoValueType' = NO_VALUE, signed: 'Optional[bool]' = None,
byteorder: 'Literal["little", "big"]' = 'big',
bit_length: 'Optional[int]' = None,
callback: 'Callable[[Self, dict[str, Any]], None]' = lambda *_: None) -> 'None':
if length is None:
if self.__length__ is None:
raise IntError(f'Field has no length.')
length = self.__length__
super().__init__(length, default, callback)
if bit_length is not None:
if bit_length < 0:
raise ProtocolError(
f'Field {self.name} resolved to a negative length; '
f'bit_length={bit_length!r}'
)
self._bit_length = bit_length
self._bit_mask = (1 << bit_length) - 1
else:
self._bit_length, self._bit_mask = -1, -1
# NOTE: ``__signed__`` fixes a subclass's sign, so a contradicting ``signed`` is rejected rather
# than discarded; ``None`` means not given, which tells it apart from the default (#545).
if self.__signed__ is None:
self._signed = False if signed is None else bool(signed)
elif signed is None or bool(signed) == self.__signed__:
self._signed = self.__signed__
else:
raise FieldValueError(
f'{type(self).__name__}: field is fixed as '
f'{"signed" if self.__signed__ else "unsigned"}, '
f'but signed={signed!r} was given'
)
self._byteorder = byteorder
self._need_process = False
endian = '>' if byteorder == 'big' else '<'
if self.__template__ is not None:
struct_fmt = self.__template__
else:
# NOTE: ``self._signed``, not the argument, so a subclass fixing ``__signed__`` but not
# ``__template__`` parses with the sign it declared.
struct_fmt = self.build_template(self._length, self._signed)
self._template = f'{endian}{struct_fmt}'
[docs]
def __call__(self, packet: 'dict[str, Any]') -> 'Self':
"""Update field attributes.
Args:
packet: Packet data.
Returns:
New instance of :class:`NumberField`.
This method will return a new instance of :class:`NumberField` instead of
updating the current instance.
Raises:
ProtocolError: If the resolved ``length`` is negative -- e.g. a
``length`` callback such as ``lambda pkt: pkt['len'] - 4``
resolving below zero once the wire value it reads is smaller
than the subtrahend. Left alone, ``1 << (length * 8)`` raises
a bare, uncatchable :exc:`ValueError` (``negative shift
count``) when ``bit_length`` was not supplied, before
:attr:`~pcapkit.corekit.fields.field.FieldBase.length` (see
its own :exc:`~pcapkit.utilities.exceptions.ProtocolError`
guard, :issue:`805`/:issue:`825`)
or :meth:`build_template` ever sees the value: this method sets
``self._bit_length`` from the resolved length eagerly, as a
cache, and shifts by it immediately, so the crash happens on
*this* line rather than on the later, already-guarded ones. See
GitHub issue :issue:`828`. This guard runs regardless of
whether ``bit_length`` was supplied, so a field constructed
with a fixed ``bit_length`` *and* a callable ``length`` that
resolves negative raises the identical message as one with no
``bit_length`` at all, rather than falling through to a
``template='...-1s'``
:exc:`~pcapkit.utilities.exceptions.ProtocolError` from
:attr:`~pcapkit.corekit.fields.field.FieldBase.length` later --
see GitHub issue :issue:`831`. A resolved length of exactly
``0`` is a legitimate empty field (e.g. ``len=4`` above
resolving to ``0``) and is left alone.
Notes:
Rebuilding the template here is what applies a callable ``length``,
and :meth:`build_template` recomputes ``self._need_process`` as it
goes, so the flag and the template always describe the same width.
See GitHub issue :issue:`591`.
"""
new_self = super().__call__(packet)
if new_self._length < 0:
raise ProtocolError(
f'Field {new_self.name} resolved to a negative length; '
f'length={new_self._length!r}'
)
if new_self._bit_length < 0:
new_self._bit_length = new_self._length * 8
new_self._bit_mask = (1 << new_self._bit_length) - 1
endian = '>' if new_self._byteorder == 'big' else '<'
struct_fmt = new_self.build_template(new_self._length, new_self._signed)
new_self._template = f'{endian}{struct_fmt}'
return new_self
[docs]
def build_template(self, length: 'int', signed: 'bool') -> 'str':
"""Build template for field.
Arguments:
length: Field size (in bytes)
signed: Whether the field is signed
Returns:
Template for field.
Notes:
``self._need_process`` is **assigned** here rather than only ever
raised, so that it always describes the ``length`` this template
was built for. Only ever setting it :data:`True` in the
fall-through branch would make it a latch: a callable ``length`` is
a placeholder of ``-1`` at construction, ``-1`` takes the
fall-through branch, and the flag would then survive the rebuild in
:meth:`__call__` that resolves the real width. :meth:`pre_process`
would consequently hand :obj:`bytes` to a template that had become
``>Q`` -- or ``>I``, ``>H``, ``>B`` -- and :func:`struct.pack` would
refuse it. See GitHub issue :issue:`591`.
Assigning it is what tells a placeholder apart from a width that
genuinely needs byte packing, without having to remember that a
placeholder was ever in play: the answer for ``-1`` is
:data:`True`, the answer for ``8`` is :data:`False`, and whichever
width is in force now is the one that decides. A callable
resolving to, say, ``3`` still takes the fall-through branch and
still gets :data:`True`, because for ``3`` that is the correct
answer rather than a leftover one.
"""
if length == 8: # unpack to 8-byte integer (long long)
struct_fmt, need_process = 'q' if signed else 'Q', False
elif length == 4: # unpack to 4-byte integer (int / long)
struct_fmt, need_process = 'i' if signed else 'I', False
elif length == 2: # unpack to 2-byte integer (short)
struct_fmt, need_process = 'h' if signed else 'H', False
elif length == 1: # unpack to 1-byte integer (char)
struct_fmt, need_process = 'b' if signed else 'B', False
else: # do not unpack
struct_fmt, need_process = f'{length}s', True
self._need_process = need_process
return struct_fmt
[docs]
def pre_process(self, value: 'int', packet: 'dict[str, Any]') -> 'int | bytes': # pylint: disable=unused-argument
"""Process field value before construction (packing).
Arguments:
value: Field value.
packet: Packet data.
Returns:
Processed field value.
Notes:
Masking against :attr:`self._bit_mask <NumberField.bit_length>`
truncates the value to the field's bit length, but it also turns a
negative value into its unsigned two's-complement pattern, which
neither :func:`struct.pack` nor :meth:`int.to_bytes` accepts for a
signed field. A signed field therefore maps the pattern back into
its signed range afterwards, so that e.g. a PCAP-NG section length
of ``-1`` (section length not specified) can be written out.
A field packed without having been resolved -- so with ``_length``
still negative -- has its width derived from the value instead, and
that rebuild can land on a width :mod:`struct` has a native
integer code for. The flag is therefore consulted **after** the
rebuild rather than before it, since deciding first and rebuilding
second would let the template and the value being returned
disagree. Cf. :issue:`591`.
That width is a **ceiling** of the bit length over eight, and it is
written as one. ``math.ceil(value.bit_length() // 8)`` is not a
ceiling at all: :func:`math.ceil` of an :obj:`int` is that
:obj:`int`, so the ``//`` has already floored the quotient and the
outer call does nothing. Every value whose bit length is not an
exact multiple of eight would be sized one octet short -- ``256`` at
one octet, ``65536`` at two, and ``1`` itself at *zero* -- which
:meth:`int.to_bytes` and :func:`struct.pack` both refuse. See GitHub
issue :issue:`599`.
A value :meth:`post_process` returned for octets whose bits outside
:attr:`bit_length` were not what packing writes there carries those
bits (see :func:`_outside_bits`), and they are written back as
captured, so ``pack(unpack(b)) == b`` holds for every ``b``. Only a
field of the same length, bit length, sign and byte order as the
one that captured them honours them; any other field, and any other
value -- one a caller built, or the result of arithmetic on a
parsed one, which is a plain :obj:`int` -- is truncated to
:attr:`bit_length` bits. See GitHub issue
:issue:`1488`.
"""
outside = _outside_bits(value, self)
value = value & self._bit_mask
if outside is not None and self._length >= 0:
value |= outside
width = self._length * 8
if self._signed and value >> (width - 1):
value -= 1 << width
elif self._signed and value > self._bit_mask >> 1:
value -= self._bit_mask + 1
if self._need_process and self._length < 0:
self._length = math.ceil(value.bit_length() / 8)
endian = '>' if self._byteorder == 'big' else '<'
struct_fmt = self.build_template(self._length, self._signed)
self._template = f'{endian}{struct_fmt}'
if not self._need_process:
return value
return value.to_bytes(
self._length, self._byteorder, signed=self._signed
)
[docs]
def post_process(self, value: 'int | bytes', packet: 'dict[str, Any]') -> 'int': # pylint: disable=unused-argument
"""Process field value after parsing (unpacked).
Args:
value: Field value.
packet: Packet data.
Returns:
Processed field value.
Notes:
Masking against :attr:`self._bit_mask <NumberField.bit_length>`
yields the unsigned pattern, so a signed field maps it back into
its signed range afterwards, mirroring :meth:`pre_process`. A
resolved signed field therefore decodes ``ffffffff`` as ``-1``
rather than ``4294967295``. See GitHub issue :issue:`1274`.
The value is the low :attr:`bit_length` bits, but the octets hold
more. When the bits outside it are not what :meth:`pre_process`
would write there for this value -- zeros, or for a negative signed
value the sign extension -- they are captured but uninterpreted,
and the value records them (see :func:`_with_outside_bits`) so that
packing it writes them back. Otherwise the value is a plain
:obj:`int`. See GitHub issue :issue:`1488`.
"""
if not self._need_process:
raw = cast('int', value)
else:
raw = int.from_bytes(
cast('bytes', value), self._byteorder, signed=self._signed
)
value = raw & self._bit_mask
if self._signed and value > self._bit_mask >> 1:
value -= self._bit_mask + 1
if 0 <= self._bit_length < self._length * 8:
spare = ((1 << (self._length * 8)) - 1) & ~self._bit_mask
outside = raw & spare
if outside != (spare if value < 0 else 0):
return _with_outside_bits(value, (outside, *_capture_key(self)))
return value
[docs]
class Int32Field(NumberField):
"""Integer value for protocol fields.
Args:
length: Field size (in bytes).
default: Field default value, if any.
signed: Whether the field is signed; fixed as :data:`True` here, so a
contradicting :data:`False` is rejected rather than ignored.
byteorder: Field byte order.
bit_length: Field bit length.
callback: Callback function to be called upon
:meth:`self.__call__ <pcapkit.corekit.fields.field.FieldBase.__call__>`.
Raises:
FieldValueError: If ``signed`` is given as :data:`False`, contradicting
the sign this class fixes.
"""
__length__ = 4
__template__ = 'i'
__signed__ = True
[docs]
class UInt32Field(NumberField):
"""Unsigned integer value for protocol fields.
Args:
length: Field size (in bytes).
default: Field default value, if any.
signed: Whether the field is signed; fixed as :data:`False` here, so a
contradicting :data:`True` is rejected rather than ignored.
byteorder: Field byte order.
bit_length: Field bit length.
callback: Callback function to be called upon
:meth:`self.__call__ <pcapkit.corekit.fields.field.FieldBase.__call__>`.
Raises:
FieldValueError: If ``signed`` is given as :data:`True`, contradicting
the sign this class fixes.
"""
__length__ = 4
__template__ = 'I'
__signed__ = False
[docs]
class Int16Field(NumberField):
"""Short integer value for protocol fields.
Args:
length: Field size (in bytes).
default: Field default value, if any.
signed: Whether the field is signed; fixed as :data:`True` here, so a
contradicting :data:`False` is rejected rather than ignored.
byteorder: Field byte order.
bit_length: Field bit length.
callback: Callback function to be called upon
:meth:`self.__call__ <pcapkit.corekit.fields.field.FieldBase.__call__>`.
Raises:
FieldValueError: If ``signed`` is given as :data:`False`, contradicting
the sign this class fixes.
"""
__length__ = 2
__template__ = 'h'
__signed__ = True
[docs]
class UInt16Field(NumberField):
"""Unsigned short integer value for protocol fields.
Args:
length: Field size (in bytes).
default: Field default value, if any.
signed: Whether the field is signed; fixed as :data:`False` here, so a
contradicting :data:`True` is rejected rather than ignored.
byteorder: Field byte order.
bit_length: Field bit length.
callback: Callback function to be called upon
:meth:`self.__call__ <pcapkit.corekit.fields.field.FieldBase.__call__>`.
Raises:
FieldValueError: If ``signed`` is given as :data:`True`, contradicting
the sign this class fixes.
"""
__length__ = 2
__template__ = 'H'
__signed__ = False
[docs]
class Int64Field(NumberField):
"""Long integer value for protocol fields.
Args:
length: Field size (in bytes).
default: Field default value, if any.
signed: Whether the field is signed; fixed as :data:`True` here, so a
contradicting :data:`False` is rejected rather than ignored.
byteorder: Field byte order.
bit_length: Field bit length.
callback: Callback function to be called upon
:meth:`self.__call__ <pcapkit.corekit.fields.field.FieldBase.__call__>`.
Raises:
FieldValueError: If ``signed`` is given as :data:`False`, contradicting
the sign this class fixes.
"""
__length__ = 8
__template__ = 'q'
__signed__ = True
[docs]
class UInt64Field(NumberField):
"""Unsigned long integer value for protocol fields.
Args:
length: Field size (in bytes).
default: Field default value, if any.
signed: Whether the field is signed; fixed as :data:`False` here, so a
contradicting :data:`True` is rejected rather than ignored.
byteorder: Field byte order.
bit_length: Field bit length.
callback: Callback function to be called upon
:meth:`self.__call__ <pcapkit.corekit.fields.field.FieldBase.__call__>`.
Raises:
FieldValueError: If ``signed`` is given as :data:`True`, contradicting
the sign this class fixes.
"""
__length__ = 8
__template__ = 'Q'
__signed__ = False
[docs]
class Int8Field(NumberField):
"""Byte value for protocol fields.
Args:
length: Field size (in bytes).
default: Field default value, if any.
signed: Whether the field is signed; fixed as :data:`True` here, so a
contradicting :data:`False` is rejected rather than ignored.
byteorder: Field byte order.
bit_length: Field bit length.
callback: Callback function to be called upon
:meth:`self.__call__ <pcapkit.corekit.fields.field.FieldBase.__call__>`.
Raises:
FieldValueError: If ``signed`` is given as :data:`False`, contradicting
the sign this class fixes.
"""
__length__ = 1
__template__ = 'b'
__signed__ = True
[docs]
class UInt8Field(NumberField):
"""Unsigned byte value for protocol fields.
Args:
length: Field size (in bytes).
default: Field default value, if any.
signed: Whether the field is signed; fixed as :data:`False` here, so a
contradicting :data:`True` is rejected rather than ignored.
byteorder: Field byte order.
bit_length: Field bit length.
callback: Callback function to be called upon
:meth:`self.__call__ <pcapkit.corekit.fields.field.FieldBase.__call__>`.
Raises:
FieldValueError: If ``signed`` is given as :data:`True`, contradicting
the sign this class fixes.
"""
__length__ = 1
__template__ = 'B'
__signed__ = False
[docs]
class EnumField(NumberField[Union[enum.IntEnum, aenum.IntEnum]]):
"""Enumerated value 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.
default: Field default value, if any.
signed: Whether the field is signed; :data:`None` defers to the
class-level ``__signed__``, which this class leaves unset and so
means unsigned.
byteorder: Field byte order.
bit_length: Field bit length.
namespace: Field namespace (a :class:`enum.IntEnum` class).
callback: Callback function to be called upon
:meth:`self.__call__ <pcapkit.corekit.fields.field.FieldBase.__call__>`.
Notes:
A wire value the ``namespace`` registry has no member for resolves to a
nameless pseudo-member rather than failing the parse -- see
:meth:`post_process`.
"""
def __init__(self, length: 'int | Callable[[dict[str, Any]], int]',
default: 'StdlibEnum | AenumEnum | NoValueType' = NO_VALUE,
signed: 'Optional[bool]' = None,
byteorder: 'Literal["little", "big"]' = 'big',
bit_length: 'Optional[int]' = None,
namespace: 'Optional[Type[StdlibEnum] | Type[AenumEnum]]' = None,
callback: 'Callable[[Self, dict[str, Any]], None]' = lambda *_: None) -> 'None':
super().__init__(length, default, signed, byteorder, bit_length, callback)
self._namespace = namespace
[docs]
def post_process(self, value: 'int | bytes', packet: 'dict[str, Any]') -> 'StdlibEnum | AenumEnum':
"""Process field value after parsing (unpacked).
Args:
value: Field value.
packet: Packet data.
Returns:
Processed field value -- the registry member declared for the value,
or a nameless pseudo-member carrying the value itself when the
registry declares none.
Raises:
BaseError: Whatever in-library error the registry raised for the
value, re-raised untouched.
Notes:
The registry is consulted through its constructor, which raises for
a value no member and no ``_missing_`` rule accounts for. That raise
is |aenum|_'s own bare :exc:`ValueError`: not one of
:mod:`pcapkit.utilities.exceptions`, so a caller cannot tell it from
a bug of its own, and not an :exc:`EOFError`, so
:meth:`Extractor.record_frames
<pcapkit.foundation.extraction.Extractor.record_frames>` does not
catch it. Left to propagate, one unassigned code would cost the
whole extraction.
It would also make the "unknown" reader the formats require
unreachable for any genuinely unassigned code -- PCAP-NG's
:class:`~pcapkit.protocols.misc.pcapng.schema.UnknownBlock`, and the
``unassigned`` option readers of IPv4, TCP, HOPOPT, MH and HIP --
because the lookup would fail before the dispatch that selects it.
PCAP-NG repeats a block's total length at both ends precisely so
that a reader can skip a block type it does not recognise; that skip
is what this fallback keeps working. See GitHub issue :issue:`701`.
The fallback is the same nameless pseudo-member this method already
builds for a field carrying no registry at all, so it is a value
shape the package already produces and the dump layer already
renders -- as ``<unknown>::<unassigned> [28]``, through
:func:`~pcapkit.dumpkit.common.render_enum`, not through its
``name is None`` branch (:issue:`648`), which a member named
``<unassigned>`` never takes -- and one an :class:`int`-keyed
dispatch registry looks up by value like any declared member. It is
built per value rather than grafted onto
the registry with ``aenum.extend_enum``, for two reasons: a
capture carrying many distinct unassigned codes would otherwise grow
a process-global registry without bound, which is the growth
:meth:`Protocol._lookup_registry
<pcapkit.protocols.protocol.Protocol._lookup_registry>` exists
to avoid; and a stdlib :class:`enum.IntEnum` registry is then
handled exactly like an ``aenum.IntEnum`` one.
Only a *foreign* rejection is absorbed. A registry rejecting a value
with one of :mod:`pcapkit.utilities.exceptions` has made a
deliberate decision that this layer -- which sees only that the value
arrived in a field of some width -- is in no position to overrule, so
an in-library error propagates unchanged and it is only |aenum|_'s
and :mod:`enum`'s "no member has this value" that becomes a
pseudo-member. That is what keeps the fallback from being an
unconditional ``except ValueError: pass``.
No registry under :mod:`pcapkit.const` raises an in-library error
from its guard, and deliberately so: a generated guard raises a
bare, unlogged :exc:`ValueError` precisely because the ``except
ValueError`` fallback of :meth:`~pcapkit.corekit.enum.EnumLookup.get`
has to keep catching it (GitHub issues :issue:`584` and
:issue:`647`). That bare :exc:`ValueError` is also how a generated
guard bounds its registry to a width, rejecting a value outside
``0 <= value <= N``. Bit-flag registries such as
:class:`pcapkit.const.tcp.flags.Flags` do so, and so do plain ones
such as :class:`pcapkit.const.ipv4.option_number.OptionNumber`, which
is the namespace of a plain :class:`EnumField`. That is harmless:
the out-of-width rejection is the bare :exc:`ValueError` this method
falls back on, so no in-library guard loses its force through this
method. The distinction is therefore for a registry registered from
outside :mod:`pcapkit.const`, which has no such obligation to stay
quiet.
Octets carrying bits outside :attr:`~NumberField.bit_length` that
:meth:`NumberField.post_process` recorded resolve to a member that
records them too, so that packing it writes them back. A registry
member is shared, so that one is an unregistered copy of it, built
by :meth:`_unregistered_member`: ``isinstance`` against the
registry and equality with the member still hold, identity does
not. See GitHub issue :issue:`1488`.
.. versionchanged:: 1.5.0
A value the registry has no member for resolves to a nameless
pseudo-member, where |aenum|_'s bare :exc:`ValueError` escaped
(:issue:`701`).
"""
value = super().post_process(value, packet)
record = _capture_record(value)
if self._namespace is not None:
try:
member = self._namespace(value)
except ValueError as error:
# NOTE: an in-library rejection is pcapkit's own verdict on the
# value, not a missing member, so it is not this layer's to absorb.
if isinstance(error, BaseError):
raise
else:
if record is None:
return member
# NOTE: a registry member is shared, so the bits outside
# ``bit_length`` go on an unregistered copy of it (#1488).
attrs = {key: val for key, val in vars(member).items()
if key not in ('_name_', '_value_', '__reduce_ex__')}
attrs[_OUTSIDE_BITS] = record
return self._unregistered_member(
type(member), member._value_, member._name_, **attrs) # pylint: disable=protected-access
pseudo = self._pseudo_member(int(value))
if record is not None:
setattr(pseudo, _OUTSIDE_BITS, record)
return pseudo
[docs]
def _pseudo_member(self, value: 'int') -> 'StdlibEnum | AenumEnum':
"""Build the bounded nameless pseudo-member this method falls back to
when ``value`` is a foreign miss rather than an in-library rejection.
Returns:
A single-member, throwaway :class:`enum.IntEnum` instance, built
fresh per call rather than ``aenum.extend_enum``-ed onto
``self._namespace``, per this method's own docstring above.
"""
unknown = enum.IntEnum('<unknown>', {
'<unassigned>': value,
}, module='pcapkit.const', qualname='pcapkit.const.<unknown>')
return getattr(unknown, '<unassigned>')
[docs]
@staticmethod
def _unregistered_member(namespace: 'Type[StdlibEnum] | Type[AenumEnum]',
value: 'Any', name: 'str' = '<unassigned>',
**attrs: 'Any') -> 'Any':
"""Build a member of ``namespace``, absent from every one of its own
lookup tables, for a value a *parse* -- rather than a direct call to
the registry's own ``get()`` -- resolved without anyone asking for a
name.
GitHub issue :issue:`575`: the owner's ruling is that an unassigned
wire value should resolve to a real member of the registry the field
names -- ``isinstance`` against it and every ancestor holds, and it
renders and dispatches exactly like a declared one -- provided
building it never grows the registry.
This is what gets there: it calls ``namespace``'s own storage base's
``__new__`` directly -- :class:`str` or :class:`int`, whichever
``namespace`` derives from -- which skips ``namespace``'s *own*
``__new__`` entirely, and with it the ``cls.__registry__.add(...)`` /
``cls.__members_ns__[...] = ...`` line every registry in this package
uses to record a member it mints. No entry is added to
``_member_map_`` or ``_value2member_map_`` either, since those are
only ever touched by the metaclass machinery ``aenum.extend_enum``
drives, which this bypasses completely.
Note:
Building a member this way, rather than as some other type
altogether, is why :meth:`PortEnumField.post_process` and
:meth:`OptionEnumField.post_process <pcapkit.protocols.misc.pcapng.schema.OptionEnumField.post_process>`
need this rather than :meth:`_pseudo_member`:
the result answers ``isinstance(result, AppType)`` truthfully,
which :meth:`PortEnumField.pre_process` tests before reading
``.port`` off it, and
:meth:`Transport._make_port <pcapkit.protocols.transport.transport.Transport._make_port>`
in :mod:`pcapkit.protocols.transport.transport` before passing it through unchanged (see
``test_a_member_is_still_an_apptype`` in
``tests/const/test_const_apptype_split_unit.py``); a value that fails them would be a
defect of its own.
The member this returns is absent from ``_value2member_map_``, so
a *value*-keyed lookup on it -- ``self._namespace(value)`` --
still raises. Nothing on the
parse or reconstruction path does that to a value it just resolved
this way, which is what keeps this safe to return from
``post_process``. A direct call to
:meth:`AppType.get() <pcapkit.const.reg.apptype.apptype.AppType.get>` for the same port
does not come through here, and it does not mint either: it builds an unregistered
member of its own, through the registry's ``_unregistered_member`` -- measured on this
tree, ``AppType.get(54321, proto=tcp)`` returns ``<TCP.unknown: 54321 [tcp]>``,
``TCP.__members__`` does not grow, and a second call with the same port returns an equal
member rather than the same one. Two unregistered members for the same value also
compare equal without being identical, since
:class:`~pcapkit.const.reg.apptype.apptype.AppType` and
:class:`~pcapkit.const.pcapng.option_type.OptionType` both define
``__eq__``/``__hash__`` off an attribute (``.port`` / ``.opt_value``) rather than object
identity -- harmless for every reader in this package, since none compares one with
``is`` or keys a mapping on it expecting identity, but worth knowing before reusing this
elsewhere.
:mod:`pickle` and :func:`copy.copy`/:func:`copy.deepcopy` all
reduce an :class:`~enum.Enum` member through
``Enum.__reduce_ex__``, which returns ``(cls, (value,))`` -- the
one lookup this member is deliberately absent from. A minted
member is registered and so round-trips; left alone, this one
would not: without the override, an unknown TCP port's member
made ``pickle.loads(pickle.dumps(member))`` raise
``ValueError: 'unknown [53406 - tcp]' is not a valid TCP`` -- while
``pickle.dumps`` still succeeded, so the failure surfaced only on
read-back rather than where it was caused.
So ``__reduce_ex__`` is set on the member itself, reducing it to
:func:`_rebuild_unregistered_member` instead of to a value lookup.
Both :mod:`pickle` and :mod:`copy` fetch that attribute with
:func:`getattr` on the object rather than on its type, so a
per-instance override is honoured: verified against the C
:mod:`pickle` accelerator on every protocol from 0 to 5, against
the pure-Python ``pickle._Pickler``, and against
:func:`copy.copy`/:func:`copy.deepcopy` both as they are on 3.11+
and with CPython's ``Enum.__copy__``/``__deepcopy__`` deleted to
emulate 3.10, where those two do not exist. Rebuilding re-enters
this method rather than ``namespace.__new__``, so an unpickled
member is unregistered exactly as the original was and the
registry does not grow -- ``TCP.__members__`` measured unchanged
before and after. On 3.11+ ``copy``/``deepcopy`` still return the
member itself, since ``Enum.__copy__`` short-circuits ahead of
any reduction; on 3.10 they return an equal rebuilt one, which is
the same answer for an immutable value.
Args:
namespace: The concrete registry class to build the member as an
instance of. ``isinstance`` holds against it and every
ancestor; it never gains an entry in any of its own tables.
value: The value ``namespace``'s own constructor would have
wrapped -- e.g. the crafted string
:meth:`~pcapkit.const.reg.apptype.apptype.AppType.__new__` builds from
a name, a port and a transport, or
:meth:`~pcapkit.const.pcapng.option_type.OptionType.__new__`'s
equivalent -- kept the same shape here so a rendered or
re-keyed member reads the same either way.
name: The member's own ``.name``; ``'<unassigned>'`` matches every
other nameless value this package produces.
**attrs: Extra attributes to set on the returned member, matching
the shape the caller's registry gives its real members --
:class:`~pcapkit.const.reg.apptype.apptype.AppType`'s ``.port``,
``.svc`` and ``.proto``, or
:class:`~pcapkit.const.pcapng.option_type.OptionType`'s
``.opt_name`` and ``.opt_value``.
Returns:
The unregistered member.
Raises:
TypeError: If ``namespace`` derives from neither :class:`str` nor
:class:`int` -- every registry this package builds does one or
the other, and guessing wrong for some future one would ship
a member silently missing whatever its storage base provides,
rather than saying plainly that this needs extending first.
"""
obj: 'Any'
if issubclass(namespace, str):
obj = str.__new__(namespace, value)
elif issubclass(namespace, int):
obj = int.__new__(namespace, value)
else:
raise TypeError(
f'{namespace!r} derives from neither str nor int; '
'_unregistered_member does not know how to build one of its members')
# NOTE: setting name/value directly rather than through the namespace's
# __new__ is what skips the registration __new__ would do.
obj._name_ = name # pylint: disable=protected-access
obj._value_ = value # pylint: disable=protected-access
for attr_name, attr_value in attrs.items():
setattr(obj, attr_name, attr_value)
# NOTE: ``Enum.__reduce_ex__`` would reduce to the value lookup this member is absent from, so a
# per-instance override lets pickle and copy rebuild an equivalent unregistered member.
return reduce_to(obj, _rebuild_unregistered_member, namespace, value, name, attrs)
[docs]
def _rebuild_unregistered_member(namespace: 'Type[StdlibEnum] | Type[AenumEnum]',
value: 'Any', name: 'str',
attrs: 'dict[str, Any]') -> 'Any':
"""Rebuild the member :meth:`EnumField._unregistered_member` returned.
This is what :mod:`pickle` and :mod:`copy` reconstruct through, in place of
the value lookup ``Enum.__reduce_ex__`` would otherwise have reduced the
member to. It is a module-level function rather than a method so that every
:mod:`pickle` protocol can name it: protocols below 4 cannot reference a
callable nested inside a class.
Args:
namespace: The registry class to rebuild the member as an instance of.
value: The member's ``_value_``.
name: The member's ``_name_``.
attrs: The extra attributes the member carried.
Returns:
A member equal to the original and, like it, absent from every one of
``namespace``'s lookup tables -- rebuilding goes back through
:meth:`EnumField._unregistered_member` and never through
``namespace.__new__``, so it cannot register anything either.
"""
return EnumField._unregistered_member( # pylint: disable=protected-access
namespace, value, name, **attrs)
[docs]
class PortEnumField(EnumField):
"""Enumerated value for transport-layer port 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.
default: Field default value, if any.
signed: Whether the field is signed; :data:`None` defers to the
class-level ``__signed__``, which this class leaves unset and so
means unsigned.
byteorder: Field byte order.
bit_length: Field bit length.
namespace: The port registry, i.e.
:class:`~pcapkit.const.reg.apptype.apptype.AppType`.
callback: Callback function to be called upon
:meth:`self.__call__ <pcapkit.corekit.fields.field.FieldBase.__call__>`.
proto: The transport protocol whose ports this field carries -- a single
:class:`~pcapkit.const.reg.apptype.apptype.TransportProtocol` member, which
is what tells e.g. TCP/80 from UDP/80. The TCP, UDP and SCTP
schemas each pass their own.
Important:
This class is specifically designed for :class:`~pcapkit.const.reg.apptype.apptype.AppType`
as it is actually an ``aenum.StrEnum`` class.
.. versionadded:: 1.5.0
"""
if TYPE_CHECKING:
_namespace: 'Type[AppType]'
def __init__(self, length: 'int | Callable[[dict[str, Any]], int]',
default: 'StdlibEnum | AenumEnum | NoValueType' = NO_VALUE,
signed: 'Optional[bool]' = None,
byteorder: 'Literal["little", "big"]' = 'big',
bit_length: 'Optional[int]' = None,
namespace: 'Optional[Type[AppType]]' = None,
callback: 'Callable[[Self, dict[str, Any]], None]' = lambda *_: None, *,
proto: 'TransportProtocol') -> 'None':
super().__init__(length, default, signed, byteorder, bit_length, namespace, callback)
self._proto = proto
[docs]
def pre_process(self, value: 'int | AppType', packet: 'dict[str, Any]') -> 'int | bytes':
"""Process field value before construction (packing).
Arguments:
value: Field value.
packet: Packet data.
Returns:
Processed field value.
Raises:
FieldValueError: If ``value`` is a port outside this field's width,
i.e. ``0``-``65535`` for the two-octet TCP, UDP and SCTP
ports.
Notes:
:meth:`NumberField.pre_process` masks a value to the field's width
rather than range-checking it, so without this check a schema built
directly would write port ``70000`` as ``0x1170`` and ``-1`` as
``0xffff``. A port is a whole field with no bits to carry outside
its width, so the check costs no wire value its round trip: every
port :meth:`post_process` can return is in range. The bound is the
one :meth:`post_process` tests. See GitHub issue :issue:`2314`.
"""
# NOTE: imported here since :mod:`pcapkit.const` imports
# :mod:`pcapkit.corekit`; tested against AppType, not ``self._namespace`` (#1516).
from pcapkit.const.reg.apptype import AppType # pylint: disable=import-outside-toplevel
if isinstance(value, AppType):
value = value.port
if isinstance(value, int) and not 0 <= value < (1 << (8 * self.length)):
raise FieldValueError(
f'Field {self.name} has invalid value: port {value!r} is outside '
f'0-{(1 << (8 * self.length)) - 1}.'
)
return super().pre_process(value, packet)
[docs]
def post_process(self, value: 'int | bytes', packet: 'dict[str, Any]') -> 'AppType':
"""Process field value after parsing (unpacked).
Args:
value: Field value.
packet: Packet data.
Returns:
Processed field value -- the registry member declared for the
port, or an unregistered member of the same registry, carrying
the port itself, when the registry declares none.
Notes:
This peeks at what :meth:`AppType.get() <pcapkit.const.reg.apptype.apptype.AppType.get>` itself would
consult -- the per-port rows via ``__registry__.getlist``, then the
documented IANA spans via
:meth:`~pcapkit.const.reg.apptype.apptype.AppType._missing_` -- and
only calls ``get`` once one of those is known to hold. It is not a
guard against growth: ``get`` answers a port neither accounts for
with an unregistered member rather than minting one (GitHub issue
:issue:`860`), and the registry does not grow either way
(:issue:`1774`). It exists because that member is not the one a
parse needs. ``get`` builds it through
:meth:`AppType._unregistered_member <pcapkit.const.reg.apptype.apptype.AppType._unregistered_member>`,
which installs no ``__reduce_ex__``, so ``pickle.loads(pickle.dumps(...))`` raises
``ValueError: 'unknown [50000 - tcp]' is not a valid TCP``; and its ``.name`` is
``'unknown'``, so it dumps as ``TCP::unknown [...]`` rather than as the
``TCP::<unassigned> [...]`` of every other unassigned wire value. A genuine miss
therefore builds its member through :meth:`EnumField._unregistered_member` instead.
That member carries the registry's own ``__transport__``, as
``get``'s does, rather than ``self._proto``. The two are the same
member for each of the TCP, UDP and SCTP schemas, but a composite built by hand
such as ``TransportProtocol.tcp | TransportProtocol.udp`` is a bare
:class:`int` (``3``, which dispatches to SCTP), and reading ``.name``
off it would raise :exc:`AttributeError` on a miss ``get`` resolves.
A port outside this field's own width is rejected *before* any of
that, rather than being let through to
:meth:`~pcapkit.const.reg.apptype.apptype.AppType._missing_`, which
raises a bare :exc:`ValueError` for it. That call sits inside an
``except ValueError`` that absorbs a foreign miss, and
:meth:`AppType.get() <pcapkit.const.reg.apptype.apptype.AppType.get>` rejects the
out-of-range case deliberately, so that it is never minted over -- a catch keyed on
exception type alone cannot see the difference between that and a "no member has this
value", so it would quietly turn the rejection into a miss. Within the width,
``_missing_`` raises for no port of any per-transport registry: it returns a member for
a documented span and :data:`None` otherwise. Checking the width first needs no
exception-based distinction at all. ``self.length`` is the field's own declared byte
width (``2`` for every caller of this class, hence ``0``-``65535``) rather than a
hard-coded ``65535`` borrowed from
:meth:`~pcapkit.const.reg.apptype.apptype.AppType._missing_`'s own guard, so the two
stay in lockstep by construction.
"""
value = super(EnumField, self).post_process(value, packet)
proto = self._proto
if not (isinstance(value, int) and 0 <= value < (1 << (8 * self.length))):
# NOTE: lets AppType.get() raise its own rejection for a port this
# field's width cannot hold, rather than absorbing it below as a miss.
return self._namespace.get(value, proto=proto)
owner = self._namespace._dispatch(value, proto) # pylint: disable=protected-access
if not owner.__registry__.getlist(value): # type: ignore[union-attr]
try:
declared = owner._missing_(value) # pylint: disable=protected-access
except ValueError as error:
# NOTE: ``value`` is in width here, so a ValueError is a foreign "no such member" miss; a
# ``pcapkit.utilities.exceptions`` error is a deliberate registry decision and propagates.
if isinstance(error, BaseError):
raise
declared = None
if declared is None:
# NOTE: An unregistered ``owner`` member, attributes set in ``AppType.__new__``'s order, which
# dumps follow; ``owner.__transport__`` is used as ``proto`` may be a bare int (#1774).
transport = owner.__transport__
return self._unregistered_member(
owner, f'unknown [{value:d} - {transport.name}]',
svc='unknown', port=value, proto=transport)
return self._namespace.get(value, proto=proto)
[docs]
class BitmapField(NumberField):
"""Integer word for protocol fields, split into subfields numbered from its
least-significant bit.
The word is read as an integer in its byte order first, so
a subfield may cross an octet boundary of a little-endian word, which
:class:`~pcapkit.corekit.fields.strings.BitField`, numbering the bits of the
raw octets from the most-significant one, cannot express.
Args:
length: Field size (in bytes).
namespace: Field namespace (a dict mapping each subfield's name to a tuple
of its start bit, counted from the least-significant bit, and its
width in bits).
remainder: Name the bits no subfield names are kept under, in place; with
:obj:`None`, they are kept under ``__unnamed__`` where any is set, as
:class:`~pcapkit.corekit.fields.strings.BitField` keeps them.
flags: Flag enumeration :attr:`remainder` holds a member of; with
:obj:`None`, it holds an :class:`int`.
byteorder: Field byte order.
callback: Callback function to be called upon
:meth:`self.__call__ <pcapkit.corekit.fields.field.FieldBase.__call__>`.
Raises:
FieldValueError: If a subfield reaches past the word, two subfields share
a bit, or ``flags`` is given without ``remainder``.
.. versionadded:: 1.6.0
"""
#: Key of the unpacked value that carries the unnamed bits, as
#: :attr:`BitField.UNNAMED <pcapkit.corekit.fields.strings.BitField.UNNAMED>`.
UNNAMED = '__unnamed__'
@property
def subfields(self) -> 'dict[str, tuple[int, int]]':
"""Each subfield's start bit and width."""
return dict(self._subfields)
@property
def remainder(self) -> 'Optional[str]':
"""Name the bits no subfield names are kept under."""
return self._remainder
@property
def remainder_mask(self) -> 'int':
"""Mask of the bits no subfield names."""
return self._remainder_mask
@property
def flags(self) -> 'Optional[type]':
"""Flag enumeration :attr:`remainder` holds a member of."""
return self._registry
def __init__(self, length: 'int', namespace: 'dict[str, tuple[int, int]]', *,
remainder: 'Optional[str]' = None, flags: 'Optional[Type[enum.Flag] | Type[aenum.Flag]]' = None,
byteorder: 'Literal["little", "big"]' = 'big',
callback: 'Callable[[Self, dict[str, Any]], None]' = lambda *_: None) -> 'None':
super().__init__(length, byteorder=byteorder, callback=callback)
if flags is not None and remainder is None:
raise FieldValueError(f'{type(self).__name__}: flags {flags.__name__} given without a remainder')
width = 8 * length
named = 0
for name, (start, size) in namespace.items():
bits = ((1 << size) - 1) << start
if start < 0 or size < 1 or start + size > width or named & bits:
raise FieldValueError(f'{type(self).__name__}: subfield {name!r} spans bits '
f'{start}-{start + size - 1} of a {width}-bit word, or another\'s')
named |= bits
self._subfields = dict(namespace)
self._remainder = remainder
self._registry = flags
self._remainder_mask = ((1 << width) - 1) & ~named
[docs]
def pre_process(self, value: 'Mapping[str, Any]', # type: ignore[override]
packet: 'dict[str, Any]') -> 'int | bytes':
"""Process field value before construction (packing).
Arguments:
value: Field value.
packet: Packet data.
Returns:
Processed field value.
Raises:
FieldValueError: If a subfield is missing or its value does not fit
its bits, or the remainder is not an :class:`int` or sets a
subfield's bit.
"""
word = 0
for name, (start, size) in self._subfields.items():
if name not in value:
raise FieldValueError(f'{type(self).__name__}: subfield {name!r} is missing')
part = value[name]
if isinstance(part, bool) or not isinstance(part, int) or not 0 <= part < 1 << size:
raise FieldValueError(f'{type(self).__name__}: subfield {name!r} value {part!r} '
f'does not fit in {size} bit(s)')
word |= part << start
rest = value.get(self.UNNAMED if self._remainder is None else self._remainder, 0)
if isinstance(rest, bool) or not isinstance(rest, int) or int(rest) & ~self._remainder_mask:
raise FieldValueError(f'{type(self).__name__}: remainder {rest!r} is not an int of the bits '
'no subfield names')
return super().pre_process(word | int(rest), packet)
[docs]
def post_process(self, value: 'int | bytes', # type: ignore[override]
packet: 'dict[str, Any]') -> 'dict[str, Any]':
"""Process field value after parsing (unpacked).
Args:
value: Field value.
packet: Packet data.
Returns:
Each subfield's value, by name, then the remainder: a member of the
field's ``flags`` where it has them, else an :class:`int`.
"""
word = int(super().post_process(value, packet))
out = {name: word >> start & ((1 << size) - 1)
for name, (start, size) in self._subfields.items()} # type: dict[str, Any]
rest = word & self._remainder_mask
if self._remainder is not None:
out[self._remainder] = rest if self._registry is None else self._registry(rest)
elif rest:
out[self.UNNAMED] = rest
return out