# -*- coding: utf-8 -*-
"""numerical field class"""
import enum
import functools
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
__all__ = [
'NumberField',
'Int32Field', 'UInt32Field',
'Int16Field', 'UInt16Field',
'Int64Field', 'UInt64Field',
'Int8Field', 'UInt8Field',
'EnumField', 'PortEnumField',
]
if TYPE_CHECKING:
from enum import IntEnum as StdlibEnum
from typing import Any, Callable, 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_'
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)
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]
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.
"""
__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 the sign for a subclass such as
# :class:`UInt32Field`; discarding a contradicting ``signed`` would make
# ``UInt32Field(signed=True)`` return an unsigned field and
# ``Int8Field(signed=False)`` a signed one, both without a word.
# ``None`` is what "not given" looks like, which is what
# lets a contradicting value be told apart from the default and rejected
# while leaving an agreeing one alone. See #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 ``signed`` argument. A subclass
# that fixes ``__signed__`` without also fixing ``__template__``
# would otherwise build its template from the argument and parse
# with the opposite sign to the one 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. C.f. :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 :mod:`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.schema.misc.pcapng.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 :func:`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:`ProtocolBase._lookup_registry
<pcapkit.protocols.protocol.ProtocolBase._lookup_registry>` exists
to avoid; and a stdlib :class:`enum.IntEnum` registry is then
handled exactly like an :class:`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 :mod:`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`). The registries that
*do* bound themselves to a width and reject outside it are the
bit-flag ones -- :class:`pcapkit.const.tcp.flags.Flags` among them --
and none of those is named as the namespace of a plain
:class:`EnumField` anywhere in the package, 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 :mod:`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 decision about
# the value, rather than the enumeration library reporting that
# no member carries it, 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 captured bits
# outside ``bit_length`` go on an unregistered copy of it
# rather than on the member itself (:issue:`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 :func:`~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 :func:`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.schema.misc.pcapng.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 ``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:`~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. Measured on CPython 3.14.7, resolving port 53406
through the TCP schema's :class:`PortEnumField`
with nothing in place of the override,
``pickle.loads(pickle.dumps(member))`` raised
``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 the enum protocol's own name/value attributes
# directly, rather than through namespace's own __new__, is what
# skips the registration that __new__ would otherwise have done.
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__`` reduces a member to ``(cls, (value,))``,
# i.e. to the one lookup this member is deliberately absent from, so
# pickle and copy would both raise on it without this. Overriding it
# per instance -- which pickle and copy both honour, since both fetch
# it with getattr on the object rather than on its type -- rebuilds an
# equivalent unregistered member instead. functools.partial rather
# than a closure keeps this off obj itself, so the member does not
# become part of a reference cycle merely by being reducible.
obj.__reduce_ex__ = functools.partial(
_reduce_unregistered_member, namespace, value, name, attrs)
return obj
[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]
def _reduce_unregistered_member( # pylint: disable=unused-argument
namespace: 'Type[StdlibEnum] | Type[AenumEnum]',
value: 'Any', name: 'str', attrs: 'dict[str, Any]',
protocol: 'int') -> 'tuple[Callable[..., Any], tuple[Any, ...]]':
"""The ``__reduce_ex__`` :meth:`EnumField._unregistered_member` installs.
Bound to its first four arguments with :func:`functools.partial`, so that
the reducer holds the ingredients of the member rather than the member
itself.
Args:
namespace: The registry class the member is an instance of.
value: The member's ``_value_``.
name: The member's ``_name_``.
attrs: The extra attributes the member carries.
protocol: The :mod:`pickle` protocol version, ignored -- the reduction
is the same for all of them, and :mod:`copy` passes 4 here.
Returns:
A two-tuple of :func:`_rebuild_unregistered_member` and its arguments.
"""
return (_rebuild_unregistered_member, (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 a :class:`~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.
"""
# NOTE: imported here rather than at module level, as
# :mod:`pcapkit.const` itself imports :mod:`pcapkit.corekit`; and tested
# against AppType rather than ``self._namespace``, as the
# per-transport copies this class replaced did (:issue:`1516`).
from pcapkit.const.reg.apptype import AppType # pylint: disable=import-outside-toplevel
if isinstance(value, AppType):
value = value.port
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:`~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
``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:`~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() -- unmodified -- raise its rejection
# for a port this field's own width cannot represent, rather than
# risking it being absorbed below as a foreign 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 already known to be in-width here, so this
# ValueError is never the out-of-range rejection, which never
# reaches this branch; it could only be a foreign "no member has
# this value", absorbed as a miss. AppType._missing_ raises
# neither for an in-width port: it answers a miss with None. A
# pcapkit.utilities.exceptions error is still a deliberate
# registry decision and propagates unchanged.
if isinstance(error, BaseError):
raise
declared = None
if declared is None:
# NOTE: an unregistered member of ``owner`` itself, per the
# ruling in EnumField._unregistered_member, rather than a foreign
# pseudo-enum. ``.port``, ``.svc`` and ``.proto`` are what a real
# AppType member carries -- read unconditionally by e.g.
# Transport._decode_next_layer's ``srcport.port`` -- and
# ``svc='unknown'`` marks a port with no declared service. They
# are passed in the order AppType.__new__ sets them, because
# DictDumper.object_hook renders a member's addon keys straight
# out of its ``__dict__`` in insertion order -- so any other
# order here would make an unassigned port dump ``port`` before
# ``svc`` while every declared one dumps ``svc`` first. The
# transport is ``owner.__transport__``, as AppType.get uses,
# since ``proto`` may be a bare int with no ``.name`` (#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)