# -*- coding: utf-8 -*-
"""schema for protocol headers"""
import abc
import collections
import collections.abc
import io
import itertools
from typing import TYPE_CHECKING, Any, Generic, TypeVar, cast
from pcapkit.corekit.decorators import prepare
from pcapkit.corekit.fields.collections import ListField, OptionField
from pcapkit.corekit.fields.field import FieldBase
from pcapkit.corekit.fields.misc import (ConditionalField, ForwardMatchField, PayloadField,
SchemaField)
from pcapkit.corekit.fields.strings import PaddingField
from pcapkit.corekit.infoclass import FinalisedState
from pcapkit.corekit.sentinels import NO_VALUE
from pcapkit.utilities.compat import Mapping, final
from pcapkit.utilities.exceptions import NoDefaultValue, ProtocolUnbound, SchemaError, stacklevel
from pcapkit.utilities.warnings import RegistryWarning, SchemaWarning, UnknownFieldWarning, warn
if TYPE_CHECKING:
from collections import OrderedDict
from enum import Enum
from typing import IO, Any, Callable, DefaultDict, Iterable, Iterator, Optional, Type
from typing_extensions import Self
from pcapkit.protocols.protocol import ProtocolBase
_PB = TypeVar('_PB', bound=ProtocolBase)
__all__ = ['Schema', 'EnumSchema', 'schema_final']
_VT = TypeVar('_VT')
_ET = TypeVar('_ET', bound='Enum')
_ST = TypeVar('_ST', bound='Type[Schema]')
[docs]
def schema_final(cls: '_ST', *, _finalised: 'bool' = True) -> '_ST':
"""Finalise schema class.
This decorator function is used to generate necessary
attributes and methods for the decorated :class:`Schema`
class. It can be useful to reduce runtime generation
time as well as caching already generated attributes.
Notes:
The decorator should only be used on the *final* class. Applying it
with ``_finalised=True`` seals the class against *subclassing*, which
:meth:`Schema.__init_subclass__` enforces, and marks it with
:func:`~pcapkit.utilities.compat.final` -- so ``@schema_final`` implies
``@final`` and there is never a reason to write both. Writing both is
harmless in either order, though; what is not is ``@final`` *without*
this decorator, which :meth:`Schema.__new__` refuses -- provided
``@final`` is :func:`~pcapkit.utilities.compat.final` (or, on Python
3.11 and up, :func:`typing.final` directly). Below 3.11, plain
:func:`typing.final` does not set ``__final__`` at all (gh-90500), so a
class marked with a user's own ``from typing import final`` on 3.10 is
not refused: there is nothing on the class for :meth:`Schema.__new__` to
read.
Applying this decorator to the same class a second time only warns: the
first application already did the work, so the duplicate is redundant
rather than wrong, and the class comes back finalised and usable.
Args:
cls: Schema class.
_finalised: Whether to make the schema class finalised.
Returns:
Finalised schema class.
Warns:
pcapkit.utilities.warnings.SchemaWarning: If ``cls`` has already been
finalised *by this function*, i.e. it carries
:attr:`~pcapkit.corekit.infoclass.FinalisedState.FINAL` in its own
``__dict__``. The class is returned untouched.
.. versionchanged:: 1.5.0
A class finalised with ``_finalised=True`` refuses to be subclassed,
and a schema marked ``@final`` without this decorator cannot be
instantiated (:issue:`778`).
:meta decorator:
"""
# NOTE: keyed on ``__finalised__`` rather than on ``final``'s ``__final__``,
# and read out of ``cls.__dict__`` rather than through ``getattr``, for the
# reasons given in :func:`pcapkit.corekit.infoclass.info_final`, which
# makes both choices: only ``__finalised__`` records *this function* having
# run, so a ``__final__`` test would make ``@schema_final`` over ``@final``
# skip the generation while the opposite order did the work; and
# ``__finalised__`` is inherited, so ``getattr`` would skip a subclass
# declared before its parent was finalised.
if cls.__dict__.get('__finalised__') == FinalisedState.FINAL:
warn(f'{cls.__name__}: schema has been finalised; now skipping',
SchemaWarning, stacklevel=stacklevel())
return cls
# NOTE: short-circuit for ``Schema`` itself, as in
# :func:`pcapkit.corekit.infoclass.info_final`: ``Schema`` never reaches
# ``FinalisedState.BASE``, so a bare ``Schema()`` would otherwise re-enter
# this function, and redo the ``dir()``-over-the-MRO scan below, on *every*
# call. ``__base_ready__`` is
# its own marker, own-``__dict__`` only, so it never touches ``__finalised__``.
if cls is Schema and cls.__dict__.get('__base_ready__'):
return cls
temp = ['__map__', '__map_reverse__', '__builtin__',
'__fields__', '__buffer__', '__updated__',
'__payload__', '__finalised__', '__remainder__', '__remainder_offset__',
'__short_read__']
temp.extend(cls.__additional__)
for obj in cls.mro():
temp.extend(el for el in dir(obj) if el not in cls.__fields__)
cls.__builtin__ = set(temp)
cls.__excluded__.extend(cls.__builtin__)
args_ = [f'{key}=NO_VALUE' for key in cls.__fields__]
dict_ = [f'{key}={key}' for key in cls.__fields__]
# NOTE: We shall only generate ``__init__`` if the class does not define one,
# which is a test on ``cls.__dict__``, not on ``hasattr``: every class
# inherits ``__init__`` from :obj:`object`, so ``hasattr(cls, '__init__')``
# is always true. Without the generated method, ``Schema(...)`` would run
# :meth:`Schema.__update__` alone and never reach
# :meth:`Schema.__post_init__`, leaving a schema built from a subset of its
# fields holding :class:`~pcapkit.corekit.fields.field.FieldBase` objects in
# place of the omitted values, so that it could not be packed at all and
# failed with an error naming a field class rather than a field.
#
# :class:`~pcapkit.protocols.schema.misc.null.NoPayload` is what the test
# protects: it declares an argument-less ``__init__`` of its own so that no
# generated one displaces it.
if '__init__' not in cls.__dict__:
# NOTE: We only generate typed ``__init__`` method if only the class
# has field definition from any of itself and its base classes.
if args_:
# NOTE: The following code is to make the ``__init__`` method work.
# It is inspired from the :func:`dataclasses._create_fn` function.
#
# ``**kwargs`` is forwarded rather than rejected, so that a keyword
# naming something other than a field reaches
# :meth:`Schema.__update__` and draws its
# :class:`~pcapkit.utilities.warnings.UnknownFieldWarning`. Several
# schemas are constructed that way on purpose -- the Multipath TCP
# options take a ``kind`` and a ``length`` that the enclosing option
# owns and that ``MPTCP`` declares only for the type checker -- so a
# strict signature would turn a warning into a :exc:`TypeError`.
init_ = (
f'def __create_fn__():\n'
f' def __init__(self, {", ".join(args_)}, *, __packet__=None, **kwargs):\n'
f' self.__update__({", ".join(dict_)}, **kwargs)\n'
f' self.__post_init__(__packet__)\n'
f' return __init__\n'
)
else:
init_ = (
'def __create_fn__():\n'
' def __init__(self, dict_=None, *, __packet__=None, **kwargs):\n'
' self.__update__(dict_, **kwargs)\n'
' self.__post_init__(__packet__)\n'
' return __init__\n'
)
ns = {} # type: dict[str, Any]
exec(init_, None, ns) # pylint: disable=exec-used # nosec
cls.__init__ = ns['__create_fn__']() # type: ignore[misc]
cls.__init__.__qualname__ = f'{cls.__name__}.__init__' # type: ignore[misc]
if not _finalised:
# NOTE: ``Schema`` itself must never receive this promotion, for the same
# reason ``Info`` must not -- see :func:`pcapkit.corekit.infoclass.info_final`.
# This branch binds ``cls`` to ``Schema``
# only when something bare-constructs ``Schema()`` directly, and writing
# ``BASE`` onto ``Schema.__dict__`` there would make every subclass declared
# afterwards inherit ``BASE`` and skip :meth:`Schema.__new__`'s own
# ``FinalisedState.NONE`` branch -- silently defeating the bare-``@final``
# guard nested inside it for the rest of the process.
#
# ``__base_ready__`` is set instead, for the short-circuit at the top of
# this function -- own-``__dict__`` only, so it is not inherited and
# does not touch ``__finalised__``.
if cls is not Schema:
cls.__finalised__ = FinalisedState.BASE
else:
cls.__base_ready__ = True
return cls
cls.__finalised__ = FinalisedState.FINAL
return final(cls)
[docs]
class Schema(Mapping[str, _VT], Generic[_VT], metaclass=SchemaMeta):
"""Schema for protocol headers."""
if TYPE_CHECKING:
#: Mapping of name conflicts with builtin methods (original names to
#: transformed names).
__map__: 'dict[str, str]'
#: Mapping of name conflicts with builtin methods (transformed names to
#: original names).
__map_reverse__: 'dict[str, str]'
#: List of builtin methods.
__builtin__: 'set[str]'
#: Mapping of fields.
__fields__: 'OrderedDict[str, FieldBase]'
#: Mapping of field names to packed values.
__buffer__: 'dict[str, bytes]'
#: Whether the schema has changed since it was last packed, so that
#: :meth:`__bytes__` packs it again.
__updated__: 'bool'
#: Octets of an enclosing :class:`~pcapkit.corekit.fields.misc.SchemaField`
#: span that this nested schema left unread; set only when non-empty.
__remainder__: 'bytes'
#: Octets this nested schema read before :attr:`__remainder__`.
__remainder_offset__: 'int'
#: Flag for finalised class initialisation.
__finalised__: 'FinalisedState' = FinalisedState.NONE
#: Field name of the payload.
__payload__: 'str' = 'payload'
#: List of additional built-in names.
__additional__: 'list[str]' = []
#: List of names to be excluded from :obj:`dict` conversion.
__excluded__: 'list[str]' = []
[docs]
def __init_subclass__(cls, /, *args: 'Any', **kwargs: 'Any') -> 'None':
"""Refuse to derive from a finalised schema.
:func:`~pcapkit.utilities.compat.final` is a promise to the type
checker and nothing more: it records ``__final__`` on the class and
leaves the interpreter free to subclass it anyway. Every schema
:func:`schema_final` finalises carries a generated ``__init__`` built
from the :attr:`__fields__` that were declared at that moment, so a
subclass adding a field afterwards inherits a constructor that cannot
set it. This turns that promise into a rule the interpreter keeps.
Args:
*args: Arbitrary positional arguments.
**kwargs: Arbitrary keyword arguments in class definition.
Raises:
SchemaError: If any class in ``cls``'s ancestry carries the
``__final__`` marker in its own ``__dict__``.
"""
# NOTE: the whole ancestry rather than ``cls.__bases__``, and
# ``base.__dict__`` rather than ``getattr`` -- see
# :meth:`pcapkit.corekit.infoclass.Info.__init_subclass__`, which makes
# the same choices for the same reasons.
for base in cls.__mro__[1:]:
if base.__dict__.get('__final__'):
raise SchemaError(f'{cls.__name__}: cannot subclass {base.__name__}, '
'which is final')
# NOTE: forwarded rather than swallowed, so that a class keyword nobody
# accepts still reaches ``object`` and fails there.
# :meth:`EnumSchema.__init_subclass__` is the one caller that reaches here
# with nothing: it consumes its own ``code`` keyword and discards the
# rest, which is why a stray class keyword is tolerated on an
# :class:`EnumSchema` subclass and rejected on a plain one.
super().__init_subclass__(*args, **kwargs)
[docs]
def __new__(cls, *args: '_VT', **kwargs: '_VT') -> 'Self': # pylint: disable=unused-argument
"""Create a new instance.
The class will try to automatically generate ``__init__`` method with
the same signature as specified in class variables' type annotations,
which is inspired by :pep:`557` (:mod:`dataclasses`).
Note:
Whether ``cls`` still needs finalising is read from its *own*
``__dict__``, never inherited (:issue:`1498`). Each
class is finalised on its first construction, whatever its
ancestors' state, so a subclass of a
:attr:`~pcapkit.corekit.infoclass.FinalisedState.BASE` ancestor
still gets its own ``__builtin__``, ``__excluded__`` and generated
``__init__``, and a ``@final`` class under a ``BASE`` ancestor is
refused like any other.
Args:
*args: Arbitrary positional arguments.
**kwargs: Arbitrary keyword arguments.
Raises:
SchemaError: If ``cls`` was marked with
:func:`~pcapkit.utilities.compat.final` but never finalised by
:func:`schema_final`, i.e. it carries ``__final__`` in its own
``__dict__`` without
:attr:`~pcapkit.corekit.infoclass.FinalisedState.FINAL`.
"""
# NOTE: first instantiation is the earliest point a bare ``@final`` can
# be caught -- ``final`` runs after the class object exists, so
# :meth:`__init_subclass__` has already returned. Nested inside the NONE
# branch, which is already one-shot per class, so a finalised schema pays
# nothing for it. See :meth:`pcapkit.corekit.infoclass.Info.__new__`,
# which carries the full reasoning; on this side the missing generated
# ``__init__`` is the one built from :attr:`__fields__`.
#
# Both lookups read ``cls.__dict__`` rather than ``getattr``, because
# both markers are ordinary, inherited class attributes. The outer one:
# :func:`schema_final` writes ``BASE`` onto the class it finalises, and a
# subclass reading that state through inheritance was never finalised
# itself -- it would keep the ``__excluded__`` it copied at declaration,
# without the names :func:`schema_final` adds, and leak ``__map__``,
# ``__map_reverse__``, ``__buffer__`` and ``__updated__`` from
# :meth:`to_dict` (GitHub issue #1498). The inner one: an ancestor
# hand-marked bare ``@final`` passes its marker to every subclass
# declared before it landed, and ``getattr`` would blame such a subclass
# for a mismarking that is its ancestor's.
if cls.__dict__.get('__finalised__', FinalisedState.NONE) == FinalisedState.NONE:
if cls.__dict__.get('__final__'):
raise SchemaError(f'{cls.__name__}: marked final but never finalised, so it has no generated '
'__init__; apply schema_final, which applies final itself, not final alone')
cls = schema_final(cls, _finalised=False)
self = super().__new__(cls)
# NOTE: We define the ``__map__`` and ``__map_reverse__`` attributes
# here under ``self`` to avoid them being considered as class variables
# and thus being shared by all instances.
super().__setattr__(self, '__map__', {})
super().__setattr__(self, '__map_reverse__', {})
# NOTE: We only create the attributes for the instance itself,
# to avoid creating shared attributes for the class.
super().__setattr__(self, '__buffer__', {name: b'' for name in self.__fields__.keys()})
super().__setattr__(self, '__updated__', True)
return self
def __post_init__(self, packet: 'Optional[dict[str, Any]]' = None) -> 'None':
"""Fill in the fields the caller left unset.
Args:
packet: Packet data, as forwarded from the ``__packet__`` keyword
argument of the generated ``__init__``. The schema is packed
here only when one is given; see the note below.
"""
for name, field in self.__fields__.items():
# NOTE: Read with a fallback rather than by subscript, since the
# generated ``__init__`` is not the only caller: :meth:`from_dict`
# seeds only the keys its argument carries, so a field the caller left
# out is missing from ``__dict__`` entirely rather than holding
# ``NO_VALUE``.
#
# What is tested is ``NO_VALUE`` alone, not ``NO_VALUE`` or ``None``.
# This method fills in what the caller did not say, and a ``None`` the
# caller passed *is* something said: on an optional field it is the
# chosen value, meaning this packet does not carry the field. It is
# also what :meth:`unpack` stores for a
# :class:`~pcapkit.corekit.fields.misc.ConditionalField` whose test
# fails -- including one that declares a default of its own -- so
# substituting the default here would leave a constructed schema
# disagreeing with a parsed one about the same packet, and
# ``from_dict(parsed.to_dict())`` no longer reproducing what it was
# given. Telling the two apart is what ``NO_VALUE`` is for.
value = self.__dict__.get(name, NO_VALUE)
if value is not NO_VALUE:
continue
default = field.default
if default is not NO_VALUE:
self.__dict__[name] = default
else:
# NOTE: Nothing to fill an unset field with, so the ``NO_VALUE``
# the generated ``__init__`` seeded it with is dropped rather than
# kept: it is a *field* sentinel, not a value a schema may hold.
# Dropping it rather than storing ``None`` also keeps the name out
# of the context :meth:`pack` builds from ``__dict__``, which a
# schema may be relying on to seed for itself -- the PCAP-NG
# section header block reads its Byte-Order Magic from a ``match``
# its own :meth:`pre_pack` supplies, and only when the context
# does not name one already. :meth:`pack` reads an absent field as
# ``None`` regardless.
self.__dict__.pop(name, None)
# NOTE: Packed here only when a packet context was actually handed over.
# A schema is not in general packable from its own fields alone: a field
# callback may read a key that the *enclosing* layer owns, as the
# Multipath TCP options do with the ``length`` of the TCP option that
# carries them, and packing without it raises rather than producing
# octets. ``__updated__`` is still set, so a schema left unpacked here is
# packed by :meth:`__bytes__` on first use -- by which time the enclosing
# layer has supplied the context.
if packet is not None:
self.pack(packet)
def __update__(self, dict_: 'Optional[Mapping[str, _VT] | Iterable[tuple[str, _VT]]]' = None,
**kwargs: '_VT') -> 'None':
# NOTE: Keys with the same names as the class's builtin methods will be
# renamed with the class name prefixed as mangled class variables
# implicitly and internally. Such mapping information will be stored
# within the ``__map__`` attribute.
__name__ = type(self).__name__ # pylint: disable=redefined-builtin
if dict_ is None:
data_iter = kwargs.items() # type: Iterable[tuple[str, Any]]
elif isinstance(dict_, collections.abc.Mapping):
data_iter = itertools.chain(dict_.items(), kwargs.items())
else:
data_iter = itertools.chain(dict_, kwargs.items())
for (key, value) in data_iter:
# NOTE: Not a field, but what :meth:`to_dict` emits for the octets an
# enclosing ``SchemaField`` span held beyond this schema, and where,
# so that a rebuilt schema still re-emits them (:issue:`1380`).
# NOTE: Nor is the short read :meth:`unpack` records (:issue:`1458`).
if key in ('__remainder__', '__remainder_offset__', '__short_read__'):
self.__dict__[key] = value
continue
if key not in self.__buffer__:
warn(f'{key!r} is not a valid field name', UnknownFieldWarning)
continue
if key in self.__builtin__:
new_key = f'_{__name__}{key}'
# NOTE: We keep record of the mapping bidirectionally.
self.__map__[key] = new_key
self.__map_reverse__[new_key] = key
key = new_key
# if key in self.__dict__:
# raise KeyExists(f'{key!r} already exists')
# NOTE: We don't rewrite the key names here, just keep the
# original ones, even though they might break on the ``.``
# (:obj:`getattr`) operator.
# if isinstance(key, str):
# key = re.sub(r'\W', '_', key)
self.__dict__[key] = value
self.__updated__ = True
__init__ = __update__
def __str__(self) -> 'str':
temp = [] # type: list[str]
for (key, value) in self.__dict__.items():
if key in self.__excluded__:
continue
out_key = self.__map_reverse__.get(key, key)
temp.append(f'{out_key}={value}')
args = ', '.join(temp)
return f'{type(self).__name__}({args})'
def __repr__(self) -> 'str':
temp = [] # type: list[str]
for (key, value) in self.__dict__.items():
if key in self.__excluded__:
continue
out_key = self.__map_reverse__.get(key, key)
if isinstance(value, Schema):
temp.append(f'{out_key}={type(value).__name__}(...)')
else:
temp.append(f'{out_key}={value!r}')
args = ', '.join(temp)
return f'{type(self).__name__}({args})'
def __bytes__(self) -> 'bytes':
if self.__updated__:
self.pack()
buffer = [] # type: list[bytes]
for name in self.__fields__.keys():
value = self.__buffer__[name]
buffer.append(value)
return b''.join(buffer)
def __len__(self) -> 'int':
return len(self.__bytes__())
def __iter__(self) -> 'Iterator[str]':
for key in self.__dict__:
if key in self.__builtin__:
continue
yield self.__map_reverse__.get(key, key)
def __getitem__(self, name: 'str') -> '_VT':
# NOTE: Serves every key :meth:`__iter__` yields, not only the declared
# fields: a ``post_process`` may store derived values on the instance,
# as :class:`~pcapkit.protocols.schema.internet.ipv4.TSOption` does, and
# iteration reports them, so ``dict(schema)`` must be able to read them
# back (:issue:`1291`). A mangled key is reachable only under the name
# it was mangled from, as in iteration.
if name in self.__fields__ or (name not in self.__builtin__ and name not in self.__map_reverse__):
key = self.__map__.get(name, name)
return self.__dict__[key]
raise KeyError(name)
def __setattr__(self, name: 'str', value: '_VT') -> 'None':
if name in self.__fields__:
key = self.__map__.get(name, name)
self.__dict__[key] = value
# NOTE: ``self.__updated__ = True`` would re-enter this method once
# per field assigned, only to miss the ``__fields__`` test and fall
# through to ``object.__setattr__``. ``__updated__`` is an instance
# attribute established in ``__new__``, so the direct store is the same
# write with none of the round trip.
self.__dict__['__updated__'] = True
return
return super().__setattr__(name, value)
def __delattr__(self, name: 'str') -> 'None':
if name in self.__fields__:
key = self.__map__.get(name, name)
del self.__dict__[key]
self.__updated__ = True
return
return super().__delattr__(name)
[docs]
@classmethod
def from_dict(cls, dict_: 'Optional[Mapping[str, _VT] | Iterable[tuple[str, _VT]]]' = None,
**kwargs: '_VT') -> 'Self':
r"""Create a new instance.
* If ``dict_`` is present and has a ``.keys()`` method, then does:
``for k in dict_: self[k] = dict_[k]``.
* If ``dict_`` is present and has no ``.keys()`` method, then does:
``for k, v in dict_: self[k] = v``.
* If ``dict_`` is not present, then does:
``for k, v in kwargs.items(): self[k] = v``.
Args:
dict\_: Source data.
**kwargs: Arbitrary keyword arguments.
"""
self = cls.__new__(cls)
self.__update__(dict_, **kwargs)
self.__post_init__()
return self
[docs]
def to_dict(self) -> 'dict[str, _VT]':
"""Convert :class:`Schema` into :obj:`dict`.
Important:
We only convert nested :class:`Schema` objects into :obj:`dict` if
they are the direct value of the :class:`Schema` object's attribute.
Should such :class:`Schema` objects be nested within other data,
types, such as :obj:`list`, :obj:`tuple`, :obj:`set`, etc., we
leave them intact rather than converting them into :obj:`dict`.
Note:
A schema unpacked as the value of a
:class:`~pcapkit.corekit.fields.misc.SchemaField` that left part of
the field's span unread carries those octets under a
``__remainder__`` key, and the number of octets it did read under
``__remainder_offset__``. :meth:`from_dict` accepts both back, so
that the rebuilt field still packs them (:issue:`1380`). The keys
are absent otherwise.
A schema whose data ended inside one of its fields carries the
``__short_read__`` that :meth:`unpack` recorded, and
:meth:`from_dict` accepts it back, likewise.
"""
dict_ = {} # type: dict[str, Any]
for (key, value) in self.__dict__.items():
if key in self.__excluded__:
continue
out_key = self.__map_reverse__.get(key, key)
if isinstance(value, Schema):
dict_[out_key] = value.to_dict()
else:
dict_[out_key] = value
for key in ('__remainder__', '__remainder_offset__', '__short_read__'):
if key in self.__dict__:
dict_[key] = self.__dict__[key]
return dict_
[docs]
def to_bytes(self) -> 'bytes':
"""Convert :class:`Schema` into :obj:`bytes`."""
return self.__bytes__()
[docs]
def get_payload(self, name: 'Optional[str]' = None) -> 'bytes':
"""Get payload of :class:`Schema`.
Args:
name: Name of the payload field.
Returns:
Payload of :class:`Schema` as :obj:`bytes`.
"""
if name is None:
name = self.__payload__
field = self.__fields__.get(name)
if field is None:
raise ProtocolUnbound(f'unknown field: {name!r}')
if not isinstance(field, PayloadField):
raise ProtocolUnbound(f'not a payload field: {name!r}')
return self.__buffer__[name]
[docs]
def pack(self, packet: 'Optional[dict[str, Any]]' = None) -> 'bytes':
"""Pack :class:`Schema` into :obj:`bytes`.
Args:
packet: Packet data.
Returns:
Packed :class:`Schema` as :obj:`bytes`.
Notes:
Since we do not know the length of the packet, we use a
reasonable default value ``-1`` for the ``__length__``
field, as the :class:`~pcapkit.corekit.fields.field.Field`
class will consider negative value as a placeholder.
If you want to pack the packet with the correct length,
please provide the ``__length__`` value before packing.
"""
if packet is None:
packet = {}
packet.update(self.__dict__)
# NOTE: ``pre_unpack`` is called here as well as from the unpacking path
# since a schema may seed ``packet`` keys that both paths need, as
# ``pcapkit.protocols.schema.internet.hip.EncryptedParameter`` does.
# ``pre_pack`` is what carries the packing-only preparation, such as the
# PCAP-NG section header block's byte order magic, which no field of the
# schema holds and which therefore cannot be recovered from
# ``self.__dict__`` above.
self.pre_unpack(packet)
self.pre_pack(packet)
if '__length__' not in packet:
packet['__length__'] = -1 # reasonable default value
for field in self.__fields__.values():
field = field(packet)
# NOTE: Read from the instance rather than with :func:`getattr`, which
# finds the *class* attribute when the instance has none -- and a
# schema's class attribute for a field is the
# :class:`~pcapkit.corekit.fields.field.FieldBase` object itself. A
# field the caller never set would arrive below as the field rather
# than as a value: ``getattr(self, name, None)`` could not return its
# ``None``, so the absent-value branches would not fire, and packing
# would fail from inside a field object -- naming a field *class*, and
# so saying nothing about which field had been left out.
data = self.__dict__.get(self.__map__.get(field.name, field.name))
if isinstance(field, PayloadField):
# NOTE: ``ProtocolBase``, not ``Protocol``: ``Protocol`` is a thin
# subclass that only adds auto-registration for externally defined
# protocols, and no protocol in the library subclasses it. Testing
# against it would send any protocol instance -- which is exactly
# what :meth:`ProtocolBase._make_payload
# <pcapkit.protocols.protocol.ProtocolBase._make_payload>` returns,
# and what ``make``'s own ``bytes | Protocol | Schema`` signature
# advertises -- to the ``ProtocolUnbound`` below instead of packing
# it, and ``from_data`` could not reconstruct a parsed packet. An
# external ``Protocol`` subclass is still a ``ProtocolBase``. The
# import is inside the method rather than at module level.
from pcapkit.protocols.protocol import \
ProtocolBase # pylint: disable=import-outside-toplevel
if data is None:
self.__buffer__[field.name] = b''
elif isinstance(data, ProtocolBase):
self.__buffer__[field.name] = bytes(data)
elif isinstance(data, bytes):
self.__buffer__[field.name] = data
elif isinstance(data, Schema):
self.__buffer__[field.name] = data.pack(packet)
else:
raise ProtocolUnbound(f'unsupported type {type(data)}')
continue
if isinstance(field, ListField):
if data is None:
self.__buffer__[field.name] = b''
elif isinstance(data, bytes):
self.__buffer__[field.name] = data
elif isinstance(data, (list, tuple)):
# NOTE: a data model may declare a field ``tuple[...]``
# rather than ``list[...]`` -- e.g. HIP's ``group_id:
# 'tuple[Group, ...]'`` in pcapkit/protocols/data/internet/
# hip.py -- and ``_read_*`` then hands one straight back
# here on reconstruction. ``ListField.pack`` only ever
# iterates its argument, so it accepts either, and
# rejecting the tuple would break parse-then-reconstruct for
# such a field. ``list(data)`` is a no-op for an actual list
# and keeps ``ListField.pack``'s own ``Optional[list[_TL]]``
# signature honest rather than widening it too.
self.__buffer__[field.name] = field.pack(list(data), packet)
else:
raise ProtocolUnbound(f'unsupported type {type(data)}')
if isinstance(field, OptionField):
# NOTE: The packing mirror of the ``__option_padding__`` that
# :meth:`unpack` records: whatever of the declared option area
# the packed options do not fill. Without it every padding
# field sized from that key packs nothing, and the header
# comes out shorter than its own length field says
# (:issue:`1289`). An undeclared area (a negative length)
# leaves no remainder.
packet['__option_padding__'] = max(field.length - len(self.__buffer__[field.name]), 0)
continue
if isinstance(field, PaddingField):
# NOTE: the octets as captured, or zeros when there are none
# (:issue:`1223`) -- see :meth:`PaddingField.pack
# <pcapkit.corekit.fields.strings.PaddingField.pack>`.
self.__buffer__[field.name] = field.pack(data, packet)
continue
if isinstance(field, ConditionalField):
if not field.test(packet):
self.__buffer__[field.name] = b''
continue
field = field.field(packet)
if isinstance(field, ForwardMatchField):
# NOTE: a forward match consumes nothing, so it contributes no
# octets to ``bytes(self)``/``len(self)`` either. :meth:`unpack`
# mirrors this for the same reason -- see the ``ForwardMatchField``
# branch there.
self.__buffer__[field.name] = b''
continue
try:
temp = field.pack(data, packet)
except NoDefaultValue:
temp = bytes(field.length)
if isinstance(field, SchemaField) and len(temp) < field.length:
# NOTE: :meth:`unpack` hands a nested schema its whole declared
# span, and the schema may read less of it than that. The octets
# it left unread travel with the nested value as its
# ``__remainder__`` -- on the schema itself, or as a key of the
# mapping :meth:`to_dict` made of it -- so they are re-emitted
# from there, as long as the nested schema still packs to the
# ``__remainder_offset__`` octets it was parsed at. Re-emitting
# them after a nested schema that has shrunk or grown would put
# stale octets at a shifted offset, so then, or for a schema that
# never had any, the rest of the span is zero-filled.
# Either way the field keeps the width its length field declares
# (:issue:`1292`), and a schema rebuilt through :meth:`from_dict`
# packs what the parsed one did (:issue:`1380`).
#
# A field reached through a
# :class:`~pcapkit.corekit.fields.misc.SwitchField` is not handled
# here: the PCAP-NG block selector declares the rest of the
# *stream* as its span, which is no part of the field.
if isinstance(data, Schema):
record = data.__dict__ # type: Mapping[str, Any]
elif isinstance(data, collections.abc.Mapping):
record = data
else:
record = {}
tail = record.get('__remainder__', b'')
if record.get('__remainder_offset__') != len(temp) or len(temp) + len(tail) > field.length:
tail = b''
temp += tail + bytes(field.length - len(temp) - len(tail))
self.__buffer__[field.name] = temp
self.post_process(packet)
self._cut_short_read()
self.__updated__ = False
return self.__bytes__()
def _note_short_read(self, name: 'str', length: 'int', byte: 'bytes') -> 'None':
"""Record the first field the data ends inside, for :meth:`_cut_short_read`.
Args:
name: Field name.
length: Octets the field declares.
byte: Octets actually read for it.
"""
if len(byte) < length and '__short_read__' not in self.__dict__:
self.__dict__['__short_read__'] = (name, len(byte))
def _cut_short_read(self) -> 'None':
"""Cut the packed buffers back to the octets a short read consumed.
A schema whose data ended inside one of its fields records, as
``__short_read__``, that field's name and how many of its octets were
there (see :meth:`unpack`). Its value is read as though the missing
octets were zeros, so packing it writes the field at full width and
every field after it besides, none of which was captured. Only the
octets read are kept here, so a truncated header packs back to exactly
what was captured (:issue:`1458`).
"""
record = self.__dict__.get('__short_read__')
if record is None:
return
name, count = record
cut = False
for key in self.__fields__:
if cut:
self.__buffer__[key] = b''
elif key == name:
self.__buffer__[key] = self.__buffer__[key][:count]
cut = True
[docs]
def pre_pack(self, packet: 'dict[str, Any]') -> 'None':
"""Prepare ``packet`` data for packing process.
Args:
packet: packet data
Note:
This method is expected to directly modify any data stored
in the ``packet`` and thus no return is required.
"""
[docs]
@classmethod
@prepare
def unpack(cls, data: 'bytes | IO[bytes]',
length: 'Optional[int]' = None,
packet: 'Optional[dict[str, Any]]' = None) -> 'Self':
"""Unpack :obj:`bytes` into :class:`Schema`.
Args:
data: Packed data.
length: Length of data.
packet: Unpacked data.
Returns:
Unpacked data as :class:`Schema`.
Notes:
We used a ``__length__`` key in ``packet`` to record the length
of the remaining data, which is used to determine the length of
the payload field.
When this schema is nested -- unpacked through a
:class:`~pcapkit.corekit.fields.misc.SchemaField` rather than
directly -- ``packet`` is not the enclosing schema's own data, but
a context built by
:func:`~pcapkit.corekit.fields.misc.nested_packet_context`: a name
this schema does not itself
declare falls through to the enclosing schema, and the enclosing
schema is also reachable unconditionally under a ``__packet__``
key. See that function for the exact lookup, write and iteration
semantics.
We also use an ``__option_padding__`` key in ``packet`` to record how
much of an
:class:`~pcapkit.corekit.fields.collections.OptionField`'s declared
area it did not consume. What follows that area decides what the
remainder means: usually padding, to be skipped, but a schema may
equally read it as further options, as the PCAP-NG name resolution
block does with its records and options.
An :class:`~pcapkit.corekit.fields.collections.OptionField`
declares the size of the whole area it may read, but stops at the
end-of-option-list marker and reports the unconsumed remainder
through ``__option_padding__``. Since that remainder has not been
parsed, we rewind ``data`` by it, so that the fields which size
themselves from ``__option_padding__`` read the remainder itself
rather than the same number of octets from beyond it.
"""
# force cast arg type since decorator changed their signatures
if TYPE_CHECKING:
data = cast('IO[bytes]', data)
length = cast('int', length)
packet = cast('dict[str, Any]', packet)
self = cls.__new__(cls)
for field in self.__fields__.values():
field = field(packet)
# NOTE: ``Field.length`` recomputes struct.calcsize() on every read,
# so it is read once per field here rather than at each use.
if isinstance(field, PayloadField):
# NOTE: A negative length is the unset default and takes what
# the declared length leaves, never less than nothing; a
# computed zero is an empty payload, not an unset one, or it
# would swallow the fields after it (:issue:`1275`).
length = field.length
if length < 0:
length = max(cast('int', packet['__length__']), 0)
payload = data.read(length)
self.__buffer__[field.name] = payload
packet['__length__'] -= length
packet[field.name] = payload
setattr(self, field.name, payload)
continue
if isinstance(field, PaddingField):
length = field.length
byte = data.read(length)
self.__buffer__[field.name] = byte
self._note_short_read(field.name, length, byte)
packet[field.name] = byte
packet['__length__'] -= length
setattr(self, field.name, byte)
continue
if isinstance(field, ConditionalField):
if not field.test(packet):
self.__buffer__[field.name] = b''
setattr(self, field.name, None)
packet[field.name] = NO_VALUE
continue
field = field.field(packet)
length = field.length
byte = data.read(length)
self.__buffer__[field.name] = byte
if not isinstance(field, ForwardMatchField):
self._note_short_read(field.name, length, byte)
value = field.unpack(byte, packet.copy())
if isinstance(field, SchemaField) and isinstance(value, Schema):
# NOTE: Kept on the nested schema rather than only in
# ``__buffer__``, so that the octets it left unread go wherever
# the value goes, :meth:`to_dict` included; see :meth:`pack`.
# The width read is summed from the nested buffers rather than
# taken with :func:`len`, which may pack the schema again.
consumed = sum(map(len, value.__buffer__.values()))
if len(byte) > consumed:
value.__dict__['__remainder__'] = byte[consumed:]
value.__dict__['__remainder_offset__'] = consumed
setattr(self, field.name, value)
packet[field.name] = value
if isinstance(field, OptionField):
packet['__option_padding__'] = field.option_padding
if isinstance(field, ForwardMatchField):
# NOTE: a forward match reads ``length`` octets so a later field
# can size itself from them, but consumes neither the stream
# (the rewind below) nor ``__length__`` (no decrement in this
# branch). ``self.__buffer__[field.name]`` above still holds the
# octets just read, so it is zeroed here, as :meth:`pack` does for
# the same field type: ``__bytes__``/``__len__`` concatenate every
# slot in ``__buffer__``, and the octets are read again, for real,
# by whichever field actually needs them, so keeping them would
# over-report the schema's length by the forward match's width.
# Zeroing is also what lets a declared area checked against
# ``len(self)`` -- :class:`~pcapkit.corekit.fields.collections.OptionField`
# and :class:`~pcapkit.corekit.fields.collections.ListField` both
# do this -- see the octets actually consumed.
#
# The rewind is by the octets actually read, not by ``length``:
# near the end of the stream the read comes back short, and
# rewinding the full width would move the stream *before* the
# forward match, so the fields after it would re-read the
# preceding octet as their first.
data.seek(-len(byte), io.SEEK_CUR)
self.__buffer__[field.name] = b''
elif isinstance(field, OptionField) and field.option_padding > 0:
# the option list ended before the declared field length was
# exhausted; give the unconsumed remainder back to ``data``
# so that the following fields can read it
data.seek(-field.option_padding, io.SEEK_CUR)
consumed = length - field.option_padding
self.__buffer__[field.name] = byte[:consumed]
packet['__length__'] -= consumed
else:
packet['__length__'] -= length
if packet['__length__'] < 0:
warn(f'packet length < 0: {packet["__length__"]}',
SchemaWarning, stacklevel=stacklevel())
self.__updated__ = False
return self
[docs]
@classmethod
def pre_unpack(cls, packet: 'dict[str, Any]') -> 'None':
"""Prepare ``packet`` data for unpacking process.
Args:
packet: packet data
Note:
This method is expected to directly modify any data stored
in the ``packet`` and thus no return is required.
"""
[docs]
def post_process(self, packet: 'dict[str, Any]') -> 'Schema':
"""Revise ``schema`` data after packing and/or unpacking process.
Args:
packet: Unpacked data.
Returns:
Revised schema.
"""
return self
[docs]
class _EnumRegistry(collections.defaultdict):
"""A registry :class:`collections.defaultdict` that never inserts a miss.
:attr:`EnumSchema.registry` (and its class-level twin,
:attr:`EnumMeta.registry`) is read with a bare ``registry[code]`` at many
call sites across the schema layer, e.g. ``Option.registry[type]``. A
plain :class:`collections.defaultdict` inserts whatever
:attr:`~collections.defaultdict.default_factory` returns the *first time*
an unregistered ``code`` is looked up -- and since the registry lives on
the *class*, that insertion is permanent and shared by every instance of
every subclass in the process. Parsing one packet carrying an unrecognised
code would therefore grow the registry for the remainder of the process,
and make a later, entirely legitimate :meth:`EnumSchema.register` call
report an overwrite that never happened.
The fallback itself is deliberate -- it is how an unknown option, chunk or
block falls back to its ``Unknown*``/``Unassigned*`` schema -- so this
subclass keeps returning it and just stops recording it.
:meth:`ProtocolBase._lookup_registry
<pcapkit.protocols.protocol.Protocol._lookup_registry>`
does the same for the protocol-layer ``__proto__`` family.
"""
[docs]
def __missing__(self, key: 'Any') -> 'Any':
if self.default_factory is None:
raise KeyError(key)
return self.default_factory()
[docs]
class EnumSchema(Schema, Generic[_ET], metaclass=EnumMeta):
""":class:`Schema` with enumeration mapping support.
Examples:
To create an enumeration mapping supported schema, simply
.. code-block:: python
class MySchema(EnumSchema[MyEnum]):
# optional, set the default schema for enumeration mapping
# if the enumeration number is not found in the mapping
__default__ = lambda: UnknownSchema # by default, None
then, you can use inheritance to create a list of schemas
for this given enumeration mapping:
.. code-block:: python
class OneSchema(MySchema, code=MyEnum.ONE):
...
class MultipleSchema(MySchema, code=[MyEnum.TWO, MyEnum.THREE]):
...
or optionally, using the :meth:`register` method to register a
schema to the enumeration mapping:
.. code-block:: python
MySchema.register(MyEnum.ZERO, ZeroSchema)
And now you can access the enumeration mapping via the :attr:`registry`
property (more specifically, class attribute):
.. code-block:: python
>>> MySchema.registry[MyEnum.ONE] # OneSchema
"""
__additional__ = ['__enum__', '__default__']
__excluded__ = ['__enum__', '__default__']
#: Callback to return the default schema for enumeration mapping,
#: by default is a ``lambda: None`` statement.
__default__: 'Callable[[], Type[Self]]' = lambda: None # type: ignore[assignment,return-value]
if TYPE_CHECKING:
#: Mapping of enumeration numbers to schemas.
__enum__: 'DefaultDict[_ET, Type[Self]]'
@property
def registry(self) -> 'DefaultDict[_ET, Type[Self]]':
"""Mapping of enumeration numbers to schemas.
Note:
This property answers on an instance; on the class, the same
mapping is served by ``EnumMeta.registry``.
Important:
See :attr:`EnumMeta.registry`: the returned mapping is a
:class:`_EnumRegistry`, so looking up an unregistered ``code``
returns the default schema without recording ``code`` as if it
had been registered.
"""
return self.__enum__
[docs]
def __init_subclass__(cls, /, code: 'Optional[_ET | Iterable[_ET]]' = None, *args: 'Any', **kwargs: 'Any') -> 'None':
"""Register enumeration to :attr:`registry` mapping.
Args:
code: Enumeration code. It can be either a single enumeration
or a list of enumerations.
*args: Arbitrary positional arguments.
**kwargs: Arbitrary keyword arguments.
If ``code`` is provided, the subclass will be registered to the
:attr:`registry` mapping with the given ``code``. If ``code`` is
not given, the subclass will not be registered.
Warns:
pcapkit.utilities.warnings.RegistryWarning: If any of ``code`` is
already registered, naming the displaced schema and its
replacement. This is the guard :meth:`register` applies, repeated
here because a class declaration is the *other* way into
:attr:`__enum__`: ``class MyOption(Option, code=...)`` writes the
registry without calling :meth:`register`.
Notes:
If :attr:`__enum__` is not yet defined at function call,
it will automatically be defined as a :class:`_EnumRegistry`
object, with the default value set to :attr:`__default__`.
If intended to customise the :attr:`__enum__` mapping,
it is possible to override the :meth:`__init_subclass__` method and
define :attr:`__enum__` manually. Such a manual definition may use
a plain :class:`collections.defaultdict` -- e.g. to seed a
namespaced or nested mapping such as
:class:`pcapkit.protocols.schema.misc.pcapng.Option`'s -- so it is
swapped for the retention-safe :class:`_EnumRegistry` below, before
anything else can hold a reference to the original object.
"""
# NOTE: the base hook goes first, before any of the registry work below.
# :meth:`Schema.__init_subclass__` is what refuses to derive from a
# finalised schema, and a refusal has to land before this method writes
# ``cls`` into ``__enum__``: raising afterwards would discard the class
# object while leaving the registry pointing at it, so a rejected
# declaration would still have displaced a built-in schema. It takes no
# arguments: ``code`` is this method's own and the rest are deliberately
# dropped.
super().__init_subclass__()
if not hasattr(cls, '__enum__'):
cls.__enum__ = _EnumRegistry(cls.__default__)
elif '__enum__' in cls.__dict__ and not isinstance(cls.__dict__['__enum__'], _EnumRegistry):
# ``cls`` set its own ``__enum__`` in the class body, as a plain
# ``collections.defaultdict`` -- swap it for the retention-safe
# variant now, while ``cls`` is still being constructed and no
# external code has had a chance to capture a reference to the
# original dict. Every later access, through :attr:`registry` or
# otherwise, then sees the same safe object, and identity across
# repeated ``.registry`` reads is preserved.
manual = cls.__dict__['__enum__']
cls.__enum__ = _EnumRegistry(getattr(manual, 'default_factory', None), manual)
if code is not None:
# One loop over both shapes, so the overwrite guard below is written
# once rather than once per branch. ``register`` cannot be delegated
# to here: :class:`pcapkit.protocols.schema.misc.pcapng.Option`
# overrides it with an incompatible signature, so ``cls.register``
# does not mean the same thing for every subclass.
codes = code if isinstance(code, collections.abc.Iterable) else (code,)
for _code in codes:
# A repeated member of an iterable ``code`` -- or two members that
# are the same object, as an :class:`enum.Enum` alias makes
# possible -- reaches this twice for the same key with ``cls`` on
# both sides: incumbent on the first pass, replacement on the
# second. So ordinary subclassing syntax reaches this guard, e.g.
# ``class X(Base, code=[A, A])`` or ``code=[A, B]`` with
# ``A is B``.
incumbent = cls.__enum__.get(_code) # type: ignore[arg-type]
if incumbent is not None and incumbent is not cls:
warn(f'schema {_code} already registered, overwriting '
f'{incumbent!r} with {cls!r}', RegistryWarning)
cls.__enum__[_code] = (cls) # type: ignore[index]
[docs]
@classmethod
def register(cls, code: '_ET', schema: 'Type[Self]') -> 'None':
"""Register enumeration to :attr:`__enum__` mapping.
Args:
code: Enumeration code.
schema: Enumeration schema.
Warns:
pcapkit.utilities.warnings.RegistryWarning: If ``code`` is already
registered, naming the displaced schema and its replacement.
Note:
Every public registrar in
:mod:`pcapkit.foundation.registry.protocols` that accepts a
``schema`` registers two halves of one binding -- a parser class
through e.g. :meth:`IPv4.register_option
<pcapkit.protocols.internet.ipv4.IPv4.register_option>`, and a
schema class through this method. Both warn on an overwrite, so one
``register_ipv4_option`` call replacing a built-in reports the
schema it displaced as well as the parser.
The warning fires only when the incumbent differs from the
replacement: a repeat call that names the exact same schema object
is a caller replaying a registration, not a mistake, so it is a
silent no-op.
Presence is a faithful "was this really registered" test only because
:class:`_EnumRegistry` returns a miss without recording it. A plain
:class:`collections.defaultdict` would insert :attr:`__default__` the
first time any unregistered ``code`` was looked up, so parsing a
single packet carrying an unknown code would make the next legitimate
registration for that code warn about an entry no caller ever asked
for.
:class:`pcapkit.protocols.schema.misc.pcapng.Option` overrides this
method with a namespaced registry of its own and does not delegate
here, so it is guarded separately.
"""
incumbent = cls.__enum__.get(code) # type: ignore[arg-type]
if incumbent is not None and incumbent is not schema:
warn(f'schema {code} already registered, overwriting '
f'{incumbent!r} with {schema!r}', RegistryWarning)
cls.__enum__[code] = schema # type: ignore[index]
def keep_short_read(protocol: 'ProtocolBase') -> 'None':
"""Carry a parsed header's short read into the protocol's info.
Args:
protocol: Protocol instance, just parsed.
Called by the layer base classes once a parse is done. When the data ended
inside one of the header's fields, ``info`` gains the ``__short_read__``
record :meth:`Schema.unpack` made, so that :func:`replay_short_read` can
cut the rebuild back to what was captured (:issue:`1458`). The key is an
ordinary one, so :meth:`Info.to_dict <pcapkit.corekit.infoclass.Info.to_dict>`
carries it and ``from_data(info.to_dict())`` replays it too; it is absent
for a header that was captured whole.
"""
header = cast('Optional[Schema]', getattr(protocol, '__header__', None))
if header is None:
return
record = header.__dict__.get('__short_read__')
if record is not None:
protocol._info.__update__({'__short_read__': record}) # pylint: disable=protected-access
def replay_short_read(protocol: '_PB', data: 'Any') -> '_PB':
"""Cut a protocol rebuilt by ``from_data`` back to the octets its header had.
Args:
protocol: Protocol instance, just built by ``from_data``.
data: The info it was built from.
Returns:
``protocol`` itself.
The rebuild packs every field at full width, the missing octets as the
zeros they were read as. The ``__short_read__`` record that
:func:`keep_short_read` left in ``data`` names the field the capture ended
inside and how much of it there was, so everything from that point on is
dropped (:issue:`1458`). A header cut short has no payload past it.
"""
record = data.get('__short_read__') if isinstance(data, collections.abc.Mapping) else None
header = cast('Optional[Schema]', getattr(protocol, '__header__', None))
if record is None or header is None:
return protocol
name, count = record
offset = 0
for key in header.__fields__:
if key == name:
offset += min(count, len(header.__buffer__[key]))
break
offset += len(header.__buffer__[key])
else:
return protocol
header.__dict__['__short_read__'] = record
header._cut_short_read() # pylint: disable=protected-access
# pylint: disable=protected-access
protocol._data = protocol._data[:offset]
protocol._file = io.BytesIO(protocol._data)
protocol.__cached__.pop('__len__', None)
protocol._info.__update__({'__short_read__': record})
return protocol