diff --git a/docs/source/pcapkit/protocols/internet/hopopt.rst b/docs/source/pcapkit/protocols/internet/hopopt.rst index 21f946a228..177d0bdba5 100644 --- a/docs/source/pcapkit/protocols/internet/hopopt.rst +++ b/docs/source/pcapkit/protocols/internet/hopopt.rst @@ -220,8 +220,15 @@ Type Stubs Auxiliary Functions ~~~~~~~~~~~~~~~~~~~ +The option helpers are shared with +:mod:`~pcapkit.protocols.schema.internet.ipv6_opts` and documented in +:mod:`pcapkit.protocols.schema.internet.ipv6_option`. The functions below bind +them to this module's error prefix and option schemas; +:func:`~pcapkit.protocols.schema.internet.ipv6_option.rpl_opt_sub_tlv_len` and +:func:`~pcapkit.protocols.schema.internet.ipv6_option.pad_opt_data_len` are +used unchanged. + .. autofunction:: pcapkit.protocols.schema.internet.hopopt.mpl_opt_seed_id_len -.. autofunction:: pcapkit.protocols.schema.internet.hopopt.pad_opt_data_len .. autofunction:: pcapkit.protocols.schema.internet.hopopt.calipso_pad_len .. autofunction:: pcapkit.protocols.schema.internet.hopopt.mpl_opt_pad_len diff --git a/docs/source/pcapkit/protocols/internet/index.rst b/docs/source/pcapkit/protocols/internet/index.rst index 9c40ef532d..017f87e814 100644 --- a/docs/source/pcapkit/protocols/internet/index.rst +++ b/docs/source/pcapkit/protocols/internet/index.rst @@ -20,6 +20,7 @@ internet layer, with detailed implementation and methods. ipv6_opts ipv6_route hopopt + ipv6_option ipsec ah esp diff --git a/docs/source/pcapkit/protocols/internet/ipv6_option.rst b/docs/source/pcapkit/protocols/internet/ipv6_option.rst new file mode 100644 index 0000000000..93520ae6e8 --- /dev/null +++ b/docs/source/pcapkit/protocols/internet/ipv6_option.rst @@ -0,0 +1,31 @@ +IPv6 Options - Shared Option Helpers +==================================== + +.. module:: pcapkit.protocols.schema.internet.ipv6_option + +:mod:`pcapkit.protocols.schema.internet.ipv6_option` contains the field helpers +shared by the option schemas of +:mod:`~pcapkit.protocols.schema.internet.hopopt` (HOPOPT) and +:mod:`~pcapkit.protocols.schema.internet.ipv6_opts` (IPv6-Opts), since both +headers carry the same IPv6 options [*]_. A helper that reports an error takes +the header's name as ``prefix``, and a selector takes the header's option +schemas. Each header module binds them in a function of the same name, which +is what its fields use. + +Auxiliary Functions +------------------- + +.. autofunction:: pcapkit.protocols.schema.internet.ipv6_option.rpl_opt_sub_tlv_len +.. autofunction:: pcapkit.protocols.schema.internet.ipv6_option.mpl_opt_seed_id_len +.. autofunction:: pcapkit.protocols.schema.internet.ipv6_option.pad_opt_data_len +.. autofunction:: pcapkit.protocols.schema.internet.ipv6_option.calipso_pad_len +.. autofunction:: pcapkit.protocols.schema.internet.ipv6_option.mpl_opt_pad_len + +.. autofunction:: pcapkit.protocols.schema.internet.ipv6_option.smf_dpd_data_selector +.. autofunction:: pcapkit.protocols.schema.internet.ipv6_option.smf_i_dpd_tid_selector +.. autofunction:: pcapkit.protocols.schema.internet.ipv6_option.smf_i_dpd_id_len +.. autofunction:: pcapkit.protocols.schema.internet.ipv6_option.quick_start_data_selector + +.. rubric:: Footnotes + +.. [*] https://www.rfc-editor.org/rfc/rfc8200#section-4.2 diff --git a/docs/source/pcapkit/protocols/internet/ipv6_opts.rst b/docs/source/pcapkit/protocols/internet/ipv6_opts.rst index 10351c4df5..bb8a2ce144 100644 --- a/docs/source/pcapkit/protocols/internet/ipv6_opts.rst +++ b/docs/source/pcapkit/protocols/internet/ipv6_opts.rst @@ -220,8 +220,15 @@ Type Stubs Auxiliary Functions ~~~~~~~~~~~~~~~~~~~ +The option helpers are shared with +:mod:`~pcapkit.protocols.schema.internet.hopopt` and documented in +:mod:`pcapkit.protocols.schema.internet.ipv6_option`. The functions below bind +them to this module's error prefix and option schemas; +:func:`~pcapkit.protocols.schema.internet.ipv6_option.rpl_opt_sub_tlv_len` and +:func:`~pcapkit.protocols.schema.internet.ipv6_option.pad_opt_data_len` are +used unchanged. + .. autofunction:: pcapkit.protocols.schema.internet.ipv6_opts.mpl_opt_seed_id_len -.. autofunction:: pcapkit.protocols.schema.internet.ipv6_opts.pad_opt_data_len .. autofunction:: pcapkit.protocols.schema.internet.ipv6_opts.calipso_pad_len .. autofunction:: pcapkit.protocols.schema.internet.ipv6_opts.mpl_opt_pad_len diff --git a/docs/source/pcapkit/protocols/misc/byteorder.rst b/docs/source/pcapkit/protocols/misc/byteorder.rst new file mode 100644 index 0000000000..1e6a1eb5b6 --- /dev/null +++ b/docs/source/pcapkit/protocols/misc/byteorder.rst @@ -0,0 +1,15 @@ +Byte Order - Shared Byte-Order Helpers +====================================== + +.. module:: pcapkit.protocols.schema.misc.byteorder + +:mod:`pcapkit.protocols.schema.misc.byteorder` contains the field callback that +reads a capture file's byte order out of the packet data. It is shared by the +PCAP frame header (:mod:`~pcapkit.protocols.schema.misc.pcap.frame`) and the +PCAP-NG blocks (:mod:`~pcapkit.protocols.schema.misc.pcapng`). + +Auxiliary Functions +------------------- + +.. autofunction:: pcapkit.protocols.schema.misc.byteorder.byteorder_callback +.. autofunction:: pcapkit.protocols.schema.misc.byteorder.packet_byteorder diff --git a/docs/source/pcapkit/protocols/misc/index.rst b/docs/source/pcapkit/protocols/misc/index.rst index 2ae424fe85..b9a6310a74 100644 --- a/docs/source/pcapkit/protocols/misc/index.rst +++ b/docs/source/pcapkit/protocols/misc/index.rst @@ -16,5 +16,6 @@ indication of empty payload, the PCAP header classes, and the PCAP-NG pcap pcapng + byteorder raw null diff --git a/docs/source/pcapkit/protocols/misc/pcapng.rst b/docs/source/pcapkit/protocols/misc/pcapng.rst index d147d744f8..b03f833903 100644 --- a/docs/source/pcapkit/protocols/misc/pcapng.rst +++ b/docs/source/pcapkit/protocols/misc/pcapng.rst @@ -510,7 +510,11 @@ Type Stubs Auxiliary Functions ------------------- -.. autofunction:: pcapkit.protocols.schema.misc.pcapng.byteorder_callback +The byte order of every block but the Section Header Block is read by +:func:`~pcapkit.protocols.schema.misc.byteorder.byteorder_callback`, which is +shared with the PCAP frame header and documented in +:mod:`pcapkit.protocols.schema.misc.byteorder`. + .. autofunction:: pcapkit.protocols.schema.misc.pcapng.shb_byteorder_callback .. autofunction:: pcapkit.protocols.schema.misc.pcapng.pcapng_block_selector .. autofunction:: pcapkit.protocols.schema.misc.pcapng.dsb_secrets_selector diff --git a/examples/generators/endian.py b/examples/generators/endian.py index b5c0fb05a4..622321e2e9 100644 --- a/examples/generators/endian.py +++ b/examples/generators/endian.py @@ -11,8 +11,9 @@ instead of the file's produced the right answer anyway. That is how GitHub issue #605 survived: ``Frame.unpack`` seeded the byte order -under the key ``bytesorder`` where ``byteorder_callback`` in -:file:`pcapkit/protocols/schema/misc/pcap/frame.py` reads ``byteorder``, so the +under the key ``bytesorder`` where ``byteorder_callback`` -- then in +:file:`pcapkit/protocols/schema/misc/pcap/frame.py`, now in +:file:`pcapkit/protocols/schema/misc/byteorder.py` -- reads ``byteorder``, so the lookup always missed and always fell back to :data:`sys.byteorder`. Reading a big-endian capture then byte-swapped ``ts_sec``, ``ts_usec``, ``incl_len`` and ``orig_len`` -- and because ``incl_len`` is the payload length, the first record diff --git a/pcapkit/corekit/fields/misc.py b/pcapkit/corekit/fields/misc.py index 44abe038c1..a60f386687 100644 --- a/pcapkit/corekit/fields/misc.py +++ b/pcapkit/corekit/fields/misc.py @@ -567,7 +567,7 @@ def nested_packet_context(packet: 'dict[str, Any]') -> 'dict[str, Any]': reserved ``__packet__`` key, for a callback that needs to name the outer schema specifically rather than whichever schema happens to declare a given field -- see - :func:`pcapkit.protocols.schema.misc.pcapng.packet_byteorder` and + :func:`pcapkit.protocols.schema.misc.byteorder.packet_byteorder` and :meth:`~pcapkit.protocols.schema.misc.pcapng.BlockType.post_process` for why that distinction matters, and note that both already hand-roll this exact fallback and so are unaffected by (and do not diff --git a/pcapkit/protocols/schema/internet/hopopt.py b/pcapkit/protocols/schema/internet/hopopt.py index 735ef0662e..4ad3dcd94a 100644 --- a/pcapkit/protocols/schema/internet/hopopt.py +++ b/pcapkit/protocols/schema/internet/hopopt.py @@ -10,18 +10,18 @@ from pcapkit.const.ipv6.router_alert import RouterAlert as Enum_RouterAlert from pcapkit.const.ipv6.seed_id import SeedID as Enum_SeedID from pcapkit.const.ipv6.smf_dpd_mode import SMFDPDMode as Enum_SMFDPDMode -from pcapkit.const.ipv6.tagger_id import TaggerID as Enum_TaggerID from pcapkit.const.reg.transtype import TransType as Enum_TransType from pcapkit.corekit.fields.collections import OptionField from pcapkit.corekit.fields.field import NO_VALUE -from pcapkit.corekit.fields.ipaddress import IPv4AddressField, IPv6AddressField -from pcapkit.corekit.fields.misc import (ConditionalField, ForwardMatchField, NoValueField, - PayloadField, SchemaField, SwitchField) +from pcapkit.corekit.fields.ipaddress import IPv6AddressField +from pcapkit.corekit.fields.misc import (ConditionalField, ForwardMatchField, PayloadField, + SwitchField) from pcapkit.corekit.fields.numbers import (EnumField, NumberField, UInt8Field, UInt16Field, UInt32Field) from pcapkit.corekit.fields.strings import BitField, BytesField, PaddingField +from pcapkit.protocols.schema.internet import ipv6_option +from pcapkit.protocols.schema.internet.ipv6_option import pad_opt_data_len, rpl_opt_sub_tlv_len from pcapkit.protocols.schema.schema import EnumSchema, Schema, schema_final -from pcapkit.utilities.exceptions import FieldValueError from pcapkit.utilities.logging import SPHINX_TYPE_CHECKING __all__ = [ @@ -129,244 +129,83 @@ class QSNonce(TypedDict): reserved: int -def rpl_opt_sub_tlv_len(pkt: 'dict[str, Any]') -> 'int': - """Return RPL option sub-TLV length. - - Args: - pkt: RPL option unpacked schema. - - Returns: - RPL option sub-TLV length, i.e. ``Opt Data Len`` less the four octets - of the fixed fields. An ``Opt Data Len`` below ``4`` gives ``0``, and - the reader then rejects the option. - - """ - return max(pkt['len'] - 4, 0) +# NOTE: The option helpers are shared with IPv6-Opts and live in +# :mod:`pcapkit.protocols.schema.internet.ipv6_option`. The functions below bind +# what differs between the two headers: the ``HOPOPT`` error prefix, and the +# option schemas of this module. def mpl_opt_seed_id_len(pkt: 'dict[str, Any]') -> 'int': """Return MPL Seed-ID length. - Args: - pkt: MPL option unpacked schema. - - Returns: - MPL Seed-ID length. + See :func:`pcapkit.protocols.schema.internet.ipv6_option.mpl_opt_seed_id_len`; + errors are reported as ``HOPOPT``. """ - s_type = pkt['flags']['type'] - if s_type == 0: - return 0 - if s_type == 1: - return 2 - if s_type == 2: - return 8 - if s_type == 3: - return 16 - raise FieldValueError(f'HOPOPT: invalid MPL Seed-ID type: {s_type}') + return ipv6_option.mpl_opt_seed_id_len(pkt, prefix='HOPOPT') def smf_i_dpd_id_len(pkt: 'dict[str, Any]') -> 'int': """Return SMF I-DPD identifier length. - Args: - pkt: SMF identification-based DPD option unpacked schema. - - Returns: - SMF I-DPD identifier length. - - Raises: - FieldValueError: If ``Opt Data Len`` on the wire is too short to hold - the TaggerID it declares, which would otherwise underflow the - identifier length below zero. + See :func:`pcapkit.protocols.schema.internet.ipv6_option.smf_i_dpd_id_len`; + errors are reported as ``HOPOPT``. """ - length = pkt['len'] - (1 if pkt['info']['type'] == 0 else (pkt['info']['len'] + 2)) - if length < 0: - raise FieldValueError(f'HOPOPT: invalid SMF I-DPD option length: {pkt["len"]}') - return length + return ipv6_option.smf_i_dpd_id_len(pkt, prefix='HOPOPT') def smf_dpd_data_selector(pkt: 'dict[str, Any]') -> 'Field': """Selector function for :attr:`_SMFDPDOption.data` field. - Args: - pkt: Packet data. - - Returns: - * If ``mode`` is ``0``, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField` - wrapped :class:`~pcapkit.protocols.schema.internet.hopopt.SMFIdentificationBasedDPDOption` - instance. - * If ``mode`` is ``1``, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField` - wrapped :class:`~pcapkit.protocols.schema.internet.hopopt.SMFHashBasedDPDOption` - instance. - - Note: - The field is sized ``Opt Data Len + 2`` rather than ``Opt Data Len``. - ``Opt Data Len`` counts only what follows the option header - [:rfc:`8200#section-4.2`], while both schemas this may return inherit - :attr:`Option.type` and :attr:`Option.len` and so parse those two octets - themselves. Sizing the field at ``Opt Data Len`` would hand them an area two - octets short of the option they read, which - :class:`~pcapkit.corekit.fields.collections.OptionField` would then - mis-count against the option area. + See :func:`pcapkit.protocols.schema.internet.ipv6_option.smf_dpd_data_selector`, + choosing among the schemas registered with :class:`SMFDPDOption`; errors + are reported as ``HOPOPT``. """ - # NOTE: With ``Opt Data Len`` 0 the option has no data octet, so the mode - # bit forward-matched above belongs to whatever follows the option. - if pkt['test']['len'] == 0: - raise FieldValueError('HOPOPT: invalid SMF DPD option length: 0') - - mode = Enum_SMFDPDMode.get(pkt['test']['mode']) - schema = SMFDPDOption.registry[mode] - if schema is None: - raise FieldValueError(f'HOPOPT: invalid SMF DPD mode: {mode}') - return SchemaField(length=pkt['test']['len'] + 2, schema=schema) + return ipv6_option.smf_dpd_data_selector(pkt, prefix='HOPOPT', base=SMFDPDOption) def smf_i_dpd_tid_selector(pkt: 'dict[str, Any]') -> 'Field': """Selector function for :attr:`SMFIdentificationBasedDPDOption.tid` field. - Args: - pkt: Packet data. - - Returns: - * If ``tid_type`` is ``0``, returns a :class:`~pcapkit.corekit.fields.misc.NoValueField` - instance. - * If ``tid_type`` is ``1``, returns a :class:`~pcapkit.corekit.fields.ipaddress.IPv4AddressField` - instance. - * If ``tid_type`` is ``2``, returns a :class:`~pcapkit.corekit.fields.ipaddress.IPv6AddressField` - instance. - * Otherwise, returns a :class:`~pcapkit.corekit.fields.strings.BytesField` instance. + See :func:`pcapkit.protocols.schema.internet.ipv6_option.smf_i_dpd_tid_selector`; + errors are reported as ``HOPOPT``. """ - tid_type = Enum_TaggerID.get(pkt['info']['type']) - tid_len = pkt['info']['len'] - - # update type - pkt['info']['type'] = tid_type - - if tid_type == Enum_TaggerID.NULL: - if tid_len != 0: - raise FieldValueError(f'HOPOPT: invalid TaggerID length: {tid_len}') - return NoValueField() - if tid_type == Enum_TaggerID.IPv4: - if tid_len != 3: - raise FieldValueError(f'HOPOPT: invalid TaggerID length: {tid_len}') - return IPv4AddressField() - if tid_type == Enum_TaggerID.IPv6: - if tid_len != 15: - raise FieldValueError(f'HOPOPT: invalid TaggerID length: {tid_len}') - return IPv6AddressField() - return BytesField(length=tid_len + 1) + return ipv6_option.smf_i_dpd_tid_selector(pkt, prefix='HOPOPT') def quick_start_data_selector(pkt: 'dict[str, Any]') -> 'Field': """Selector function for :attr:`_QuickStartOption.data` field. - Args: - pkt: Packet data. - - Returns: - * If ``func`` is ``0``, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField` - wrapped :class:`~pcapkit.protocols.schema.internet.hopopt.QuickStartRequestOption` - instance. - * If ``func`` is ``8``, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField` - wrapped :class:`~pcapkit.protocols.schema.internet.hopopt.QuickStartReportOption` - instance. - * Otherwise, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField` - wrapped :class:`~pcapkit.protocols.schema.internet.hopopt.UnassignedOption` - instance, sized by the option's own ``Opt Data Len``. - - """ - func = Enum_QSFunction.get(pkt['flags']['func']) - pkt['flags']['func'] = func - - schema = QuickStartOption.registry[func] - if schema is None: - # An unassigned function has no defined layout, so the option is kept - # as raw data, sized by its own ``Opt Data Len``. - return SchemaField(length=pkt['flags']['len'] + 2, schema=UnassignedOption) - - # NOTE: The suboption schemas carry the option's own ``type`` and ``len`` - # octets, so this is the width of the whole option: the two header octets - # plus the six data octets :rfc:`4782#section-3.2` gives both functions. - return SchemaField(length=8, schema=schema) - - -def pad_opt_data_len(pkt: 'dict[str, Any]') -> 'int': - """Return the length of the padding data of a padding option. - - Args: - pkt: Padding option unpacked schema. - - Returns: - Number of padding octets carried after the option header, i.e. the - value of the ``Opt Data Len`` field of a ``PadN`` option, and zero for - a ``Pad1`` option, which carries no such field. - - Note: - A ``Pad1`` option is a single octet with neither an ``Opt Data Len`` - field nor any option data (c.f. :rfc:`8200#section-4.2`), which is why - :attr:`Option.len` is declared as a - :class:`~pcapkit.corekit.fields.misc.ConditionalField` and is skipped - for it. A skipped conditional field is *recorded* in the packet data as - :data:`~pcapkit.corekit.fields.field.NO_VALUE`, rather than being left - out of it, so the test below has to be on the **value** and not on the - presence of the key: ``pkt.get('len', 0)`` on its own hands that - :obj:`~pcapkit.corekit.fields.field.NoValueType` straight to - :class:`~pcapkit.corekit.fields.strings.PaddingField`, where it becomes - an unusable :mod:`struct` template and surfaces much later as an opaque - :exc:`struct.error`, which is how a ``Pad1`` option would fail to parse. + See :func:`pcapkit.protocols.schema.internet.ipv6_option.quick_start_data_selector`, + choosing among the schemas registered with :class:`QuickStartOption`, and + :class:`UnassignedOption` for an unassigned QS function. """ - length = pkt.get('len', 0) - if not isinstance(length, int): # ``NO_VALUE`` (skipped) or :obj:`None` (unset) - return 0 - return length + return ipv6_option.quick_start_data_selector(pkt, base=QuickStartOption, + unassigned=UnassignedOption) def calipso_pad_len(pkt: 'dict[str, Any]') -> 'int': """Return CALIPSO option padding length. - Args: - pkt: CALIPSO option unpacked schema. - - Returns: - CALIPSO option padding length. - - Raises: - FieldValueError: If ``Opt Data Len`` on the wire is too short to hold - the fixed header and the compartment bitmap declared by - ``cmpt_len``, which would otherwise underflow the padding length - below zero. + See :func:`pcapkit.protocols.schema.internet.ipv6_option.calipso_pad_len`; + errors are reported as ``HOPOPT``. """ - length = pkt['len'] - 8 - pkt['cmpt_len'] * 4 - if length < 0: - raise FieldValueError(f'HOPOPT: invalid CALIPSO option length: {pkt["len"]}') - return length + return ipv6_option.calipso_pad_len(pkt, prefix='HOPOPT') def mpl_opt_pad_len(pkt: 'dict[str, Any]') -> 'int': """Return MPL option padding length. - Args: - pkt: MPL option unpacked schema. - - Returns: - MPL option padding length. - - Raises: - FieldValueError: If ``Opt Data Len`` on the wire is too short to hold - the fixed header and the Seed-ID declared by ``flags.type``, which - would otherwise underflow the padding length below zero. + See :func:`pcapkit.protocols.schema.internet.ipv6_option.mpl_opt_pad_len`; + errors are reported as ``HOPOPT``. """ - length = pkt['len'] - 2 - (0 if pkt['flags']['type'] == 0 else mpl_opt_seed_id_len(pkt)) - if length < 0: - raise FieldValueError(f'HOPOPT: invalid MPL option length: {pkt["len"]}') - return length + return ipv6_option.mpl_opt_pad_len(pkt, prefix='HOPOPT') class Option(EnumSchema[Enum_Option]): diff --git a/pcapkit/protocols/schema/internet/ipv6_option.py b/pcapkit/protocols/schema/internet/ipv6_option.py new file mode 100644 index 0000000000..287b2c3dbd --- /dev/null +++ b/pcapkit/protocols/schema/internet/ipv6_option.py @@ -0,0 +1,299 @@ +# -*- coding: utf-8 -*- +"""helpers shared by the IPv6 option schemas + +:mod:`pcapkit.protocols.schema.internet.ipv6_option` holds the field callbacks +that size and select the IPv6 options [:rfc:`8200#section-4.2`] carried by +both the Hop-by-Hop Options header +(:mod:`~pcapkit.protocols.schema.internet.hopopt`) and the Destination Options +header (:mod:`~pcapkit.protocols.schema.internet.ipv6_opts`). + +Each of those two modules declares its own option schemas, so the helpers take +what differs between them as parameters: ``prefix`` is the name of the header, +which leads each error message (``HOPOPT`` or ``IPv6-Opts``), and the selectors +take the option schemas they choose from. Each module binds the parameters in +a function of the same name, which is what its fields use. + +""" + +from typing import TYPE_CHECKING + +from pcapkit.const.ipv6.qs_function import QSFunction as Enum_QSFunction +from pcapkit.const.ipv6.smf_dpd_mode import SMFDPDMode as Enum_SMFDPDMode +from pcapkit.const.ipv6.tagger_id import TaggerID as Enum_TaggerID +from pcapkit.corekit.fields.ipaddress import IPv4AddressField, IPv6AddressField +from pcapkit.corekit.fields.misc import NoValueField, SchemaField +from pcapkit.corekit.fields.strings import BytesField +from pcapkit.utilities.exceptions import FieldValueError + +__all__ = [ + 'rpl_opt_sub_tlv_len', 'mpl_opt_seed_id_len', 'smf_i_dpd_id_len', + 'smf_dpd_data_selector', 'smf_i_dpd_tid_selector', 'quick_start_data_selector', + 'pad_opt_data_len', 'calipso_pad_len', 'mpl_opt_pad_len', +] + +if TYPE_CHECKING: + from typing import Any, Optional, Type + + from pcapkit.corekit.fields.field import FieldBase as Field + from pcapkit.protocols.schema.schema import EnumMeta, Schema + + +def rpl_opt_sub_tlv_len(pkt: 'dict[str, Any]') -> 'int': + """Return RPL option sub-TLV length. + + Args: + pkt: RPL option unpacked schema. + + Returns: + RPL option sub-TLV length, i.e. ``Opt Data Len`` less the four octets + of the fixed fields. An ``Opt Data Len`` below ``4`` gives ``0``, and + the reader then rejects the option. + + """ + return max(pkt['len'] - 4, 0) + + +def mpl_opt_seed_id_len(pkt: 'dict[str, Any]', *, prefix: 'str') -> 'int': + """Return MPL Seed-ID length. + + Args: + pkt: MPL option unpacked schema. + prefix: Name of the extension header, leading the error message. + + Returns: + MPL Seed-ID length. + + Raises: + FieldValueError: If ``flags.type`` is not a defined Seed-ID type. + + """ + s_type = pkt['flags']['type'] + if s_type == 0: + return 0 + if s_type == 1: + return 2 + if s_type == 2: + return 8 + if s_type == 3: + return 16 + raise FieldValueError(f'{prefix}: invalid MPL Seed-ID type: {s_type}') + + +def smf_i_dpd_id_len(pkt: 'dict[str, Any]', *, prefix: 'str') -> 'int': + """Return SMF I-DPD identifier length. + + Args: + pkt: SMF identification-based DPD option unpacked schema. + prefix: Name of the extension header, leading the error message. + + Returns: + SMF I-DPD identifier length. + + Raises: + FieldValueError: If ``Opt Data Len`` on the wire is too short to hold + the TaggerID it declares, which would otherwise underflow the + identifier length below zero. + + """ + length = pkt['len'] - (1 if pkt['info']['type'] == 0 else (pkt['info']['len'] + 2)) + if length < 0: + raise FieldValueError(f'{prefix}: invalid SMF I-DPD option length: {pkt["len"]}') + return length + + +def smf_dpd_data_selector(pkt: 'dict[str, Any]', *, prefix: 'str', + base: 'EnumMeta') -> 'Field': + """Selector function for the ``data`` field of an ``SMF_DPD`` option. + + Args: + pkt: Packet data. + prefix: Name of the extension header, leading the error message. + base: The header's ``SMFDPDOption`` schema, whose registry maps the + DPD mode to the schema that reads the option. + + Returns: + * If ``mode`` is ``0``, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField` + wrapped ``SMFIdentificationBasedDPDOption`` instance. + * If ``mode`` is ``1``, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField` + wrapped ``SMFHashBasedDPDOption`` instance. + + Raises: + FieldValueError: If ``Opt Data Len`` is ``0``, or if no schema is + registered for the DPD mode. + + Note: + The field is sized ``Opt Data Len + 2`` rather than ``Opt Data Len``. + ``Opt Data Len`` counts only what follows the option header + [:rfc:`8200#section-4.2`], while both schemas this may return inherit + the option's ``type`` and ``len`` fields and so parse those two octets + themselves. Sizing the field at ``Opt Data Len`` would hand them an area two + octets short of the option they read, which + :class:`~pcapkit.corekit.fields.collections.OptionField` would then + mis-count against the option area. + + """ + # NOTE: With ``Opt Data Len`` 0 the option has no data octet, so the mode + # bit forward-matched above belongs to whatever follows the option. + if pkt['test']['len'] == 0: + raise FieldValueError(f'{prefix}: invalid SMF DPD option length: 0') + + mode = Enum_SMFDPDMode.get(pkt['test']['mode']) + schema: 'Optional[Type[Schema]]' = base.registry[mode] + if schema is None: + raise FieldValueError(f'{prefix}: invalid SMF DPD mode: {mode}') + return SchemaField(length=pkt['test']['len'] + 2, schema=schema) + + +def smf_i_dpd_tid_selector(pkt: 'dict[str, Any]', *, prefix: 'str') -> 'Field': + """Selector function for the ``tid`` field of an SMF I-DPD option. + + Args: + pkt: Packet data. + prefix: Name of the extension header, leading the error message. + + Returns: + * If ``tid_type`` is ``0`` (``NULL``), returns a + :class:`~pcapkit.corekit.fields.misc.NoValueField` instance. + * If ``tid_type`` is ``2`` (``IPv4``), returns a + :class:`~pcapkit.corekit.fields.ipaddress.IPv4AddressField` instance. + * If ``tid_type`` is ``3`` (``IPv6``), returns a + :class:`~pcapkit.corekit.fields.ipaddress.IPv6AddressField` instance. + * Otherwise, returns a :class:`~pcapkit.corekit.fields.strings.BytesField` instance. + + Raises: + FieldValueError: If ``tid_len`` does not match the length of the + ``NULL``, ``IPv4`` or ``IPv6`` TaggerID type. + + """ + tid_type = Enum_TaggerID.get(pkt['info']['type']) + tid_len = pkt['info']['len'] + + # update type + pkt['info']['type'] = tid_type + + if tid_type == Enum_TaggerID.NULL: + if tid_len != 0: + raise FieldValueError(f'{prefix}: invalid TaggerID length: {tid_len}') + return NoValueField() + if tid_type == Enum_TaggerID.IPv4: + if tid_len != 3: + raise FieldValueError(f'{prefix}: invalid TaggerID length: {tid_len}') + return IPv4AddressField() + if tid_type == Enum_TaggerID.IPv6: + if tid_len != 15: + raise FieldValueError(f'{prefix}: invalid TaggerID length: {tid_len}') + return IPv6AddressField() + return BytesField(length=tid_len + 1) + + +def quick_start_data_selector(pkt: 'dict[str, Any]', *, base: 'EnumMeta', + unassigned: 'Type[Schema]') -> 'Field': + """Selector function for the ``data`` field of a ``Quick_Start`` option. + + Args: + pkt: Packet data. + base: The header's ``QuickStartOption`` schema, whose registry maps + the QS function to the schema that reads the option. + unassigned: The header's ``UnassignedOption`` schema. + + Returns: + * If ``func`` is ``0``, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField` + wrapped ``QuickStartRequestOption`` instance. + * If ``func`` is ``8``, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField` + wrapped ``QuickStartReportOption`` instance. + * Otherwise, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField` + wrapped ``unassigned`` instance, sized by the option's own + ``Opt Data Len``. + + """ + func = Enum_QSFunction.get(pkt['flags']['func']) + pkt['flags']['func'] = func + + schema: 'Optional[Type[Schema]]' = base.registry[func] + if schema is None: + # An unassigned function has no defined layout, so the option is kept + # as raw data, sized by its own ``Opt Data Len``. + return SchemaField(length=pkt['flags']['len'] + 2, schema=unassigned) + + # NOTE: The suboption schemas carry the option's own ``type`` and ``len`` + # octets, so this is the width of the whole option: the two header octets + # plus the six data octets :rfc:`4782#section-3.2` gives both functions. + return SchemaField(length=8, schema=schema) + + +def pad_opt_data_len(pkt: 'dict[str, Any]') -> 'int': + """Return the length of the padding data of a padding option. + + Args: + pkt: Padding option unpacked schema. + + Returns: + Number of padding octets carried after the option header, i.e. the + value of the ``Opt Data Len`` field of a ``PadN`` option, and zero for + a ``Pad1`` option, which carries no such field. + + Note: + A ``Pad1`` option is a single octet with neither an ``Opt Data Len`` + field nor any option data (c.f. :rfc:`8200#section-4.2`), which is why + the option's ``len`` field is declared as a + :class:`~pcapkit.corekit.fields.misc.ConditionalField` and is skipped + for it. A skipped conditional field is *recorded* in the packet data as + :data:`~pcapkit.corekit.fields.field.NO_VALUE`, rather than being left + out of it, so the test below has to be on the **value** and not on the + presence of the key: ``pkt.get('len', 0)`` on its own hands that + :obj:`~pcapkit.corekit.fields.field.NoValueType` straight to + :class:`~pcapkit.corekit.fields.strings.PaddingField`, where it becomes + an unusable :mod:`struct` template and surfaces much later as an opaque + :exc:`struct.error`, which is how a ``Pad1`` option would fail to parse. + + """ + length = pkt.get('len', 0) + if not isinstance(length, int): # ``NO_VALUE`` (skipped) or :obj:`None` (unset) + return 0 + return length + + +def calipso_pad_len(pkt: 'dict[str, Any]', *, prefix: 'str') -> 'int': + """Return CALIPSO option padding length. + + Args: + pkt: CALIPSO option unpacked schema. + prefix: Name of the extension header, leading the error message. + + Returns: + CALIPSO option padding length. + + Raises: + FieldValueError: If ``Opt Data Len`` on the wire is too short to hold + the fixed header and the compartment bitmap declared by + ``cmpt_len``, which would otherwise underflow the padding length + below zero. + + """ + length = pkt['len'] - 8 - pkt['cmpt_len'] * 4 + if length < 0: + raise FieldValueError(f'{prefix}: invalid CALIPSO option length: {pkt["len"]}') + return length + + +def mpl_opt_pad_len(pkt: 'dict[str, Any]', *, prefix: 'str') -> 'int': + """Return MPL option padding length. + + Args: + pkt: MPL option unpacked schema. + prefix: Name of the extension header, leading the error message. + + Returns: + MPL option padding length. + + Raises: + FieldValueError: If ``Opt Data Len`` on the wire is too short to hold + the fixed header and the Seed-ID declared by ``flags.type``, which + would otherwise underflow the padding length below zero. + + """ + length = pkt['len'] - 2 - (0 if pkt['flags']['type'] == 0 + else mpl_opt_seed_id_len(pkt, prefix=prefix)) + if length < 0: + raise FieldValueError(f'{prefix}: invalid MPL option length: {pkt["len"]}') + return length diff --git a/pcapkit/protocols/schema/internet/ipv6_opts.py b/pcapkit/protocols/schema/internet/ipv6_opts.py index a21d2c0588..6b53c9560c 100644 --- a/pcapkit/protocols/schema/internet/ipv6_opts.py +++ b/pcapkit/protocols/schema/internet/ipv6_opts.py @@ -10,18 +10,18 @@ from pcapkit.const.ipv6.router_alert import RouterAlert as Enum_RouterAlert from pcapkit.const.ipv6.seed_id import SeedID as Enum_SeedID from pcapkit.const.ipv6.smf_dpd_mode import SMFDPDMode as Enum_SMFDPDMode -from pcapkit.const.ipv6.tagger_id import TaggerID as Enum_TaggerID from pcapkit.const.reg.transtype import TransType as Enum_TransType from pcapkit.corekit.fields.collections import OptionField from pcapkit.corekit.fields.field import NO_VALUE -from pcapkit.corekit.fields.ipaddress import IPv4AddressField, IPv6AddressField -from pcapkit.corekit.fields.misc import (ConditionalField, ForwardMatchField, NoValueField, - PayloadField, SchemaField, SwitchField) +from pcapkit.corekit.fields.ipaddress import IPv6AddressField +from pcapkit.corekit.fields.misc import (ConditionalField, ForwardMatchField, PayloadField, + SwitchField) from pcapkit.corekit.fields.numbers import (EnumField, NumberField, UInt8Field, UInt16Field, UInt32Field) from pcapkit.corekit.fields.strings import BitField, BytesField, PaddingField +from pcapkit.protocols.schema.internet import ipv6_option +from pcapkit.protocols.schema.internet.ipv6_option import pad_opt_data_len, rpl_opt_sub_tlv_len from pcapkit.protocols.schema.schema import EnumSchema, Schema, schema_final -from pcapkit.utilities.exceptions import FieldValueError from pcapkit.utilities.logging import SPHINX_TYPE_CHECKING __all__ = [ @@ -129,244 +129,83 @@ class QSNonce(TypedDict): reserved: int -def rpl_opt_sub_tlv_len(pkt: 'dict[str, Any]') -> 'int': - """Return RPL option sub-TLV length. - - Args: - pkt: RPL option unpacked schema. - - Returns: - RPL option sub-TLV length, i.e. ``Opt Data Len`` less the four octets - of the fixed fields. An ``Opt Data Len`` below ``4`` gives ``0``, and - the reader then rejects the option. - - """ - return max(pkt['len'] - 4, 0) +# NOTE: The option helpers are shared with HOPOPT and live in +# :mod:`pcapkit.protocols.schema.internet.ipv6_option`. The functions below bind +# what differs between the two headers: the ``IPv6-Opts`` error prefix, and the +# option schemas of this module. def mpl_opt_seed_id_len(pkt: 'dict[str, Any]') -> 'int': """Return MPL Seed-ID length. - Args: - pkt: MPL option unpacked schema. - - Returns: - MPL Seed-ID length. + See :func:`pcapkit.protocols.schema.internet.ipv6_option.mpl_opt_seed_id_len`; + errors are reported as ``IPv6-Opts``. """ - s_type = pkt['flags']['type'] - if s_type == 0: - return 0 - if s_type == 1: - return 2 - if s_type == 2: - return 8 - if s_type == 3: - return 16 - raise FieldValueError(f'IPv6-Opts: invalid MPL Seed-ID type: {s_type}') + return ipv6_option.mpl_opt_seed_id_len(pkt, prefix='IPv6-Opts') def smf_i_dpd_id_len(pkt: 'dict[str, Any]') -> 'int': """Return SMF I-DPD identifier length. - Args: - pkt: SMF identification-based DPD option unpacked schema. - - Returns: - SMF I-DPD identifier length. - - Raises: - FieldValueError: If ``Opt Data Len`` on the wire is too short to hold - the TaggerID it declares, which would otherwise underflow the - identifier length below zero. + See :func:`pcapkit.protocols.schema.internet.ipv6_option.smf_i_dpd_id_len`; + errors are reported as ``IPv6-Opts``. """ - length = pkt['len'] - (1 if pkt['info']['type'] == 0 else (pkt['info']['len'] + 2)) - if length < 0: - raise FieldValueError(f'IPv6-Opts: invalid SMF I-DPD option length: {pkt["len"]}') - return length + return ipv6_option.smf_i_dpd_id_len(pkt, prefix='IPv6-Opts') def smf_dpd_data_selector(pkt: 'dict[str, Any]') -> 'Field': """Selector function for :attr:`_SMFDPDOption.data` field. - Args: - pkt: Packet data. - - Returns: - * If ``mode`` is ``0``, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField` - wrapped :class:`~pcapkit.protocols.schema.internet.ipv6_opts.SMFIdentificationBasedDPDOption` - instance. - * If ``mode`` is ``1``, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField` - wrapped :class:`~pcapkit.protocols.schema.internet.ipv6_opts.SMFHashBasedDPDOption` - instance. - - Note: - The field is sized ``Opt Data Len + 2`` rather than ``Opt Data Len``. - ``Opt Data Len`` counts only what follows the option header - [:rfc:`8200#section-4.2`], while both schemas this may return inherit - :attr:`Option.type` and :attr:`Option.len` and so parse those two octets - themselves. Sizing the field at ``Opt Data Len`` would hand them an area two - octets short of the option they read, which - :class:`~pcapkit.corekit.fields.collections.OptionField` would then - mis-count against the option area. + See :func:`pcapkit.protocols.schema.internet.ipv6_option.smf_dpd_data_selector`, + choosing among the schemas registered with :class:`SMFDPDOption`; errors + are reported as ``IPv6-Opts``. """ - # NOTE: With ``Opt Data Len`` 0 the option has no data octet, so the mode - # bit forward-matched above belongs to whatever follows the option. - if pkt['test']['len'] == 0: - raise FieldValueError('IPv6-Opts: invalid SMF DPD option length: 0') - - mode = Enum_SMFDPDMode.get(pkt['test']['mode']) - schema = SMFDPDOption.registry[mode] - if schema is None: - raise FieldValueError(f'IPv6-Opts: invalid SMF DPD mode: {mode}') - return SchemaField(length=pkt['test']['len'] + 2, schema=schema) + return ipv6_option.smf_dpd_data_selector(pkt, prefix='IPv6-Opts', base=SMFDPDOption) def smf_i_dpd_tid_selector(pkt: 'dict[str, Any]') -> 'Field': """Selector function for :attr:`SMFIdentificationBasedDPDOption.tid` field. - Args: - pkt: Packet data. - - Returns: - * If ``tid_type`` is ``0``, returns a :class:`~pcapkit.corekit.fields.misc.NoValueField` - instance. - * If ``tid_type`` is ``1``, returns a :class:`~pcapkit.corekit.fields.ipaddress.IPv4AddressField` - instance. - * If ``tid_type`` is ``2``, returns a :class:`~pcapkit.corekit.fields.ipaddress.IPv6AddressField` - instance. - * Otherwise, returns a :class:`~pcapkit.corekit.fields.strings.BytesField` instance. + See :func:`pcapkit.protocols.schema.internet.ipv6_option.smf_i_dpd_tid_selector`; + errors are reported as ``IPv6-Opts``. """ - tid_type = Enum_TaggerID.get(pkt['info']['type']) - tid_len = pkt['info']['len'] - - # update type - pkt['info']['type'] = tid_type - - if tid_type == Enum_TaggerID.NULL: - if tid_len != 0: - raise FieldValueError(f'IPv6-Opts: invalid TaggerID length: {tid_len}') - return NoValueField() - if tid_type == Enum_TaggerID.IPv4: - if tid_len != 3: - raise FieldValueError(f'IPv6-Opts: invalid TaggerID length: {tid_len}') - return IPv4AddressField() - if tid_type == Enum_TaggerID.IPv6: - if tid_len != 15: - raise FieldValueError(f'IPv6-Opts: invalid TaggerID length: {tid_len}') - return IPv6AddressField() - return BytesField(length=tid_len + 1) + return ipv6_option.smf_i_dpd_tid_selector(pkt, prefix='IPv6-Opts') def quick_start_data_selector(pkt: 'dict[str, Any]') -> 'Field': """Selector function for :attr:`_QuickStartOption.data` field. - Args: - pkt: Packet data. - - Returns: - * If ``func`` is ``0``, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField` - wrapped :class:`~pcapkit.protocols.schema.internet.ipv6_opts.QuickStartRequestOption` - instance. - * If ``func`` is ``8``, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField` - wrapped :class:`~pcapkit.protocols.schema.internet.ipv6_opts.QuickStartReportOption` - instance. - * Otherwise, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField` - wrapped :class:`~pcapkit.protocols.schema.internet.ipv6_opts.UnassignedOption` - instance, sized by the option's own ``Opt Data Len``. - - """ - func = Enum_QSFunction.get(pkt['flags']['func']) - pkt['flags']['func'] = func - - schema = QuickStartOption.registry[func] - if schema is None: - # An unassigned function has no defined layout, so the option is kept - # as raw data, sized by its own ``Opt Data Len``. - return SchemaField(length=pkt['flags']['len'] + 2, schema=UnassignedOption) - - # NOTE: The suboption schemas carry the option's own ``type`` and ``len`` - # octets, so this is the width of the whole option: the two header octets - # plus the six data octets :rfc:`4782#section-3.2` gives both functions. - return SchemaField(length=8, schema=schema) - - -def pad_opt_data_len(pkt: 'dict[str, Any]') -> 'int': - """Return the length of the padding data of a padding option. - - Args: - pkt: Padding option unpacked schema. - - Returns: - Number of padding octets carried after the option header, i.e. the - value of the ``Opt Data Len`` field of a ``PadN`` option, and zero for - a ``Pad1`` option, which carries no such field. - - Note: - A ``Pad1`` option is a single octet with neither an ``Opt Data Len`` - field nor any option data (c.f. :rfc:`8200#section-4.2`), which is why - :attr:`Option.len` is declared as a - :class:`~pcapkit.corekit.fields.misc.ConditionalField` and is skipped - for it. A skipped conditional field is *recorded* in the packet data as - :data:`~pcapkit.corekit.fields.field.NO_VALUE`, rather than being left - out of it, so the test below has to be on the **value** and not on the - presence of the key: ``pkt.get('len', 0)`` on its own hands that - :obj:`~pcapkit.corekit.fields.field.NoValueType` straight to - :class:`~pcapkit.corekit.fields.strings.PaddingField`, where it becomes - an unusable :mod:`struct` template and surfaces much later as an opaque - :exc:`struct.error`, which is how a ``Pad1`` option would fail to parse. + See :func:`pcapkit.protocols.schema.internet.ipv6_option.quick_start_data_selector`, + choosing among the schemas registered with :class:`QuickStartOption`, and + :class:`UnassignedOption` for an unassigned QS function. """ - length = pkt.get('len', 0) - if not isinstance(length, int): # ``NO_VALUE`` (skipped) or :obj:`None` (unset) - return 0 - return length + return ipv6_option.quick_start_data_selector(pkt, base=QuickStartOption, + unassigned=UnassignedOption) def calipso_pad_len(pkt: 'dict[str, Any]') -> 'int': """Return CALIPSO option padding length. - Args: - pkt: CALIPSO option unpacked schema. - - Returns: - CALIPSO option padding length. - - Raises: - FieldValueError: If ``Opt Data Len`` on the wire is too short to hold - the fixed header and the compartment bitmap declared by - ``cmpt_len``, which would otherwise underflow the padding length - below zero. + See :func:`pcapkit.protocols.schema.internet.ipv6_option.calipso_pad_len`; + errors are reported as ``IPv6-Opts``. """ - length = pkt['len'] - 8 - pkt['cmpt_len'] * 4 - if length < 0: - raise FieldValueError(f'IPv6-Opts: invalid CALIPSO option length: {pkt["len"]}') - return length + return ipv6_option.calipso_pad_len(pkt, prefix='IPv6-Opts') def mpl_opt_pad_len(pkt: 'dict[str, Any]') -> 'int': """Return MPL option padding length. - Args: - pkt: MPL option unpacked schema. - - Returns: - MPL option padding length. - - Raises: - FieldValueError: If ``Opt Data Len`` on the wire is too short to hold - the fixed header and the Seed-ID declared by ``flags.type``, which - would otherwise underflow the padding length below zero. + See :func:`pcapkit.protocols.schema.internet.ipv6_option.mpl_opt_pad_len`; + errors are reported as ``IPv6-Opts``. """ - length = pkt['len'] - 2 - (0 if pkt['flags']['type'] == 0 else mpl_opt_seed_id_len(pkt)) - if length < 0: - raise FieldValueError(f'IPv6-Opts: invalid MPL option length: {pkt["len"]}') - return length + return ipv6_option.mpl_opt_pad_len(pkt, prefix='IPv6-Opts') class Option(EnumSchema[Enum_Option]): diff --git a/pcapkit/protocols/schema/misc/byteorder.py b/pcapkit/protocols/schema/misc/byteorder.py new file mode 100644 index 0000000000..07196ba097 --- /dev/null +++ b/pcapkit/protocols/schema/misc/byteorder.py @@ -0,0 +1,72 @@ +# -*- coding: utf-8 -*- +"""byte-order helpers shared by the PCAP and PCAP-NG schemas + +:mod:`pcapkit.protocols.schema.misc.byteorder` holds the field callback that +reads a capture file's byte order out of the packet data. The record header of +a classic PCAP file (:mod:`~pcapkit.protocols.schema.misc.pcap.frame`) and the +blocks of a PCAP-NG file (:mod:`~pcapkit.protocols.schema.misc.pcapng`) both use +it. + +""" + +import sys +from typing import TYPE_CHECKING + +__all__ = ['packet_byteorder', 'byteorder_callback'] + +if TYPE_CHECKING: + from typing import Any + + from typing_extensions import Literal + + from pcapkit.corekit.fields.numbers import NumberField + + +def packet_byteorder(packet: 'dict[str, Any]') -> 'Literal["big", "little"]': + """Byte order declared for the file or section that ``packet`` belongs to. + + A nested schema is handed its parent's packet data under a ``__packet__`` + key (see :meth:`SchemaField.pack + `), so the byte order may + live one level up. + + Args: + packet: Packet data. + + Returns: + Byte order of the enclosing file or section, falling back to the host + byte order when the packet data declares none. + + """ + if 'byteorder' not in packet and '__packet__' in packet: + return packet['__packet__'].get('byteorder', sys.byteorder) + return packet.get('byteorder', sys.byteorder) + + +def byteorder_callback(field: 'NumberField', packet: 'dict[str, Any]') -> 'None': + """Update byte order of a PCAP or PCAP-NG file. + + Args: + field: Field instance. + packet: Packet data. + + Notes: + ``byteorder`` is the key a caller has to seed, and + :func:`packet_byteorder` is what reads it. + :meth:`Frame.pack ` and + :meth:`Frame.unpack ` + both write it from the global header's magic number. A PCAP-NG Section + Header Block writes it from its own Byte-Order Magic (see + :func:`~pcapkit.protocols.schema.misc.pcapng.shb_byteorder_callback`), + and the blocks, options and records of that section read it, nested or + not. The fallback to :data:`sys.byteorder` is for a schema packed or + unpacked on its own, with no header to ask. That also means a + *misspelled* key looks exactly like an absent one and reports nothing: + every field would be read in the host's order rather than the file's, + which is right by coincidence on a little-endian capture and + byte-swapped on a big-endian one. See + :file:`tests/protocols/misc/pcap/test_frame_endian_runtime.py` for the + fixtures that take the other side of the branch. + + """ + field._byteorder = packet_byteorder(packet) diff --git a/pcapkit/protocols/schema/misc/pcap/frame.py b/pcapkit/protocols/schema/misc/pcap/frame.py index f3d6edc616..1c07adfdc4 100644 --- a/pcapkit/protocols/schema/misc/pcap/frame.py +++ b/pcapkit/protocols/schema/misc/pcap/frame.py @@ -2,46 +2,19 @@ # mypy: disable-error-code=assignment """header schema for frame header of PCAP file format""" -import sys from typing import TYPE_CHECKING from pcapkit.corekit.fields.misc import PayloadField from pcapkit.corekit.fields.numbers import UInt32Field +from pcapkit.protocols.schema.misc.byteorder import byteorder_callback from pcapkit.protocols.schema.schema import Schema, schema_final __all__ = ['Frame'] if TYPE_CHECKING: - from typing import Any - - from pcapkit.corekit.fields.numbers import NumberField as Field from pcapkit.protocols.protocol import ProtocolBase -def byteorder_callback(field: 'Field', packet: 'dict[str, Any]') -> 'None': - """Update byte order of PCAP file. - - Args: - field: Field instance. - packet: Packet data. - - Notes: - ``byteorder`` is the key a caller has to seed, and this function is what - defines it: :meth:`Frame.pack ` - and :meth:`Frame.unpack ` - both write it from the global header's magic number. The fallback to - :data:`sys.byteorder` is for a schema packed or unpacked on its own, with - no global header to ask -- which also means a *misspelled* key looks - exactly like an absent one and reports nothing: every field here would be - read in the host's order rather than the file's, which is right by - coincidence on a little-endian capture and byte-swapped on a big-endian - one. See :file:`tests/protocols/misc/pcap/test_frame_endian_runtime.py` - for the fixtures that take the other side of the branch. - - """ - field._byteorder = packet.get('byteorder', sys.byteorder) - - @schema_final class Frame(Schema): """Frame header of PCAP file format.""" diff --git a/pcapkit/protocols/schema/misc/pcapng.py b/pcapkit/protocols/schema/misc/pcapng.py index dcb4acdb48..8765cf0ca9 100644 --- a/pcapkit/protocols/schema/misc/pcapng.py +++ b/pcapkit/protocols/schema/misc/pcapng.py @@ -7,7 +7,6 @@ import collections.abc import io import struct -import sys from typing import TYPE_CHECKING, Any, cast from pcapkit.const.pcapng.block_type import BlockType as Enum_BlockType @@ -27,6 +26,7 @@ from pcapkit.corekit.fields.strings import (BitField, BytesField, DecodedString, PaddingField, StringField) from pcapkit.corekit.multidict import MultiDict, OrderedMultiDict +from pcapkit.protocols.schema.misc.byteorder import byteorder_callback, packet_byteorder from pcapkit.protocols.schema.schema import EnumSchema, Schema, schema_final from pcapkit.utilities.exceptions import FieldValueError, ProtocolError, stacklevel from pcapkit.utilities.logging import SPHINX_TYPE_CHECKING @@ -168,38 +168,6 @@ class PACKFlags(TypedDict): symbol_error: int -def packet_byteorder(packet: 'dict[str, Any]') -> 'Literal["big", "little"]': - """Byte order declared for the section that ``packet`` belongs to. - - A nested schema is handed its parent's packet data under a ``__packet__`` - key (see :meth:`SchemaField.pack - `), so the section byte order - may live one level up. - - Args: - packet: Packet data. - - Returns: - Byte order of the enclosing section, falling back to the host byte - order when the packet data declares none. - - """ - if 'byteorder' not in packet and '__packet__' in packet: - return packet['__packet__'].get('byteorder', sys.byteorder) - return packet.get('byteorder', sys.byteorder) - - -def byteorder_callback(field: 'NumberField', packet: 'dict[str, Any]') -> 'None': - """Update byte order of PCAP-NG file. - - Args: - field: Field instance. - packet: Packet data. - - """ - field._byteorder = packet_byteorder(packet) - - def shb_byteorder_callback(field: 'NumberField', packet: 'dict[str, Any]') -> 'None': """Update byte order of PCAP-NG file for SHB. @@ -208,7 +176,8 @@ def shb_byteorder_callback(field: 'NumberField', packet: 'dict[str, Any]') -> 'N the first SHB of a file has no section context by construction, and a later one would otherwise inherit the *previous* section's byte order. The magic is therefore also written back as ``packet['byteorder']``, which is what the - SHB's own options -- read by :func:`byteorder_callback`, after this field -- + SHB's own options -- read by + :func:`~pcapkit.protocols.schema.misc.byteorder.byteorder_callback`, after this field -- resolve their byte order from. A Section Header Block of 12 octets has no room for the magic, and its diff --git a/tests/corekit/test_fields_misc.py b/tests/corekit/test_fields_misc.py index af0312447b..f74f34e997 100644 --- a/tests/corekit/test_fields_misc.py +++ b/tests/corekit/test_fields_misc.py @@ -113,6 +113,14 @@ def discover_address_typed_switches() -> 'set[tuple[str, str, str]]': source = inspect.getsource(field._selector) except (OSError, TypeError): # pragma: no cover continue + # a selector may only bind a shared helper's parameters, as the + # HOPOPT and IPv6-Opts ones do (#1519), so read what it calls too + code, scope = field._selector.__code__, field._selector.__globals__ + for owner in (scope.get(name) for name in code.co_names): + if inspect.ismodule(owner): + source += ''.join(inspect.getsource(getattr(owner, name)) + for name in code.co_names + if inspect.isfunction(getattr(owner, name, None))) if any(name in source for name in ('IPv4AddressField', 'IPv6AddressField', 'IPv4InterfaceField', 'IPv6InterfaceField')): diff --git a/tests/corekit/test_fields_misc_packet_context.py b/tests/corekit/test_fields_misc_packet_context.py index efd950834f..38f2b220cd 100644 --- a/tests/corekit/test_fields_misc_packet_context.py +++ b/tests/corekit/test_fields_misc_packet_context.py @@ -179,7 +179,7 @@ def test_nested_schema_reads_enclosing_field_by_name_and_does_not_leak_writes(se def test_pcapng_byteorder_consumer_still_works_with_both_shapes(self) -> None: """The one existing, hand-rolled ``__packet__`` fallback is unaffected. - ``packet_byteorder`` (``pcapkit/protocols/schema/misc/pcapng.py:168``) + ``packet_byteorder`` (``pcapkit/protocols/schema/misc/byteorder.py:25``) is called both with a plain dict built by hand -- as several tests and :func:`~pcapkit.foundation.engines.pcapng` construct -- and with what ``SchemaField`` now actually builds. Both must keep working, and diff --git a/tests/corekit/test_schema_roundtrip_unit.py b/tests/corekit/test_schema_roundtrip_unit.py index 00bb7e74ef..28669632d9 100644 --- a/tests/corekit/test_schema_roundtrip_unit.py +++ b/tests/corekit/test_schema_roundtrip_unit.py @@ -141,15 +141,16 @@ def _strings(fn: 'Any', depth: 'int' = 0, out: 'Optional[set[str]]' = None) -> ' code = getattr(fn, '__code__', None) if depth > 4 or code is None: return out - stack = [code] + stack, names = [code], set() # type: list[Any], set[str] while stack: current = stack.pop() - out.update(current.co_names) + names.update(current.co_names) for const in current.co_consts: if isinstance(const, str): out.add(const) elif hasattr(const, 'co_consts'): stack.append(const) + out.update(names) for cell in getattr(fn, '__closure__', None) or (): try: value = cell.cell_contents @@ -157,6 +158,14 @@ def _strings(fn: 'Any', depth: 'int' = 0, out: 'Optional[set[str]]' = None) -> ' continue if callable(value): _strings(value, depth + 1, out) + # a callback may only bind a shared helper's parameters, as the HOPOPT and + # IPv6-Opts ones do (#1519), so read what it calls through a module too + scope = getattr(fn, '__globals__', {}) + for owner in {id(module): module for module in (scope.get(name) for name in names) + if inspect.ismodule(module)}.values(): + for name in names: + if inspect.isfunction(getattr(owner, name, None)): + _strings(getattr(owner, name), depth + 1, out) return out diff --git a/tests/protocols/internet/test_ipv6_option_helpers_1519_unit.py b/tests/protocols/internet/test_ipv6_option_helpers_1519_unit.py new file mode 100644 index 0000000000..1dd22be25f --- /dev/null +++ b/tests/protocols/internet/test_ipv6_option_helpers_1519_unit.py @@ -0,0 +1,213 @@ +# -*- coding: utf-8 -*- +"""The IPv6 option helpers HOPOPT and IPv6-Opts share. + +GitHub issue #1519. The nine field helpers of the HOPOPT and IPv6-Opts option +schemas were two copies differing only in the error prefix and in which +module's option schemas the selectors return. They now live once, in +:mod:`pcapkit.protocols.schema.internet.ipv6_option`, and each header module +binds its own prefix and schemas. This module pins the shared helpers under +both prefixes, that each header delegates to them with its own, and that every +error message keeps its exact text. + +The modules are imported inside each test, after +:func:`~tests._support.reimport_once_per_class`, so that they belong to the +:mod:`pcapkit` import that is live when the test runs. + +""" + +import collections +import copy +import importlib +import unittest +from unittest import mock + +from tests._support import reimport_once_per_class + +SHARED = 'pcapkit.protocols.schema.internet.ipv6_option' + +#: Each header's schema module, and the prefix its errors carry. +HEADERS = ( + ('pcapkit.protocols.schema.internet.hopopt', 'HOPOPT'), + ('pcapkit.protocols.schema.internet.ipv6_opts', 'IPv6-Opts'), +) + +#: ``(helper, packet, message after the prefix)`` for every ``raise`` in the +#: helpers whose packet alone triggers it. +ERRORS = ( + ('mpl_opt_seed_id_len', {'flags': {'type': 4}}, 'invalid MPL Seed-ID type: 4'), + ('smf_i_dpd_id_len', {'len': 0, 'info': {'type': 0, 'len': 0}}, + 'invalid SMF I-DPD option length: 0'), + ('smf_i_dpd_id_len', {'len': 3, 'info': {'type': 2, 'len': 3}}, + 'invalid SMF I-DPD option length: 3'), + ('smf_dpd_data_selector', {'test': {'len': 0, 'mode': 0}}, + 'invalid SMF DPD option length: 0'), + ('smf_i_dpd_tid_selector', {'info': {'type': 0, 'len': 1}}, 'invalid TaggerID length: 1'), + ('smf_i_dpd_tid_selector', {'info': {'type': 2, 'len': 4}}, 'invalid TaggerID length: 4'), + ('smf_i_dpd_tid_selector', {'info': {'type': 3, 'len': 16}}, 'invalid TaggerID length: 16'), + ('calipso_pad_len', {'len': 11, 'cmpt_len': 1}, 'invalid CALIPSO option length: 11'), + ('mpl_opt_pad_len', {'len': 3, 'flags': {'type': 1}}, 'invalid MPL option length: 3'), + ('mpl_opt_pad_len', {'len': 1, 'flags': {'type': 0}}, 'invalid MPL option length: 1'), +) + +#: The helpers each header binds, and the keywords it binds them with beyond +#: ``prefix`` -- by attribute name of the header's schema module. +BOUND = { + 'mpl_opt_seed_id_len': {}, + 'smf_i_dpd_id_len': {}, + 'smf_dpd_data_selector': {'base': 'SMFDPDOption'}, + 'smf_i_dpd_tid_selector': {}, + 'calipso_pad_len': {}, + 'mpl_opt_pad_len': {}, +} + +#: The helpers both headers use unchanged. +UNBOUND = ('rpl_opt_sub_tlv_len', 'pad_opt_data_len') + + +class TestIPv6OptionHelpers(unittest.TestCase): + """Pin the shared helpers, and the header modules' use of them.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _shared(self) -> 'object': + return importlib.import_module(SHARED) + + def _headers(self) -> 'list[tuple[object, str]]': + return [(importlib.import_module(name), prefix) for name, prefix in HEADERS] + + def test_shared_module_holds_all_nine_helpers(self) -> None: + shared = self._shared() + self.assertEqual(sorted(shared.__all__), sorted([*BOUND, 'quick_start_data_selector', *UNBOUND])) + + def test_shared_helpers_report_the_prefix_they_are_given(self) -> None: + from pcapkit.utilities.exceptions import FieldValueError + + shared = self._shared() + hopopt = importlib.import_module(HEADERS[0][0]) + for prefix in ('HOPOPT', 'IPv6-Opts', 'any-header'): + for name, packet, message in ERRORS: + extra = {'base': hopopt.SMFDPDOption} if name == 'smf_dpd_data_selector' else {} + with self.subTest(prefix=prefix, helper=name, packet=packet): + with self.assertRaises(FieldValueError) as caught: + getattr(shared, name)(copy.deepcopy(packet), prefix=prefix, **extra) + self.assertEqual(str(caught.exception), f'{prefix}: {message}') + + def test_shared_dpd_selector_reports_an_unregistered_mode(self) -> None: + from pcapkit.const.ipv6.smf_dpd_mode import SMFDPDMode + from pcapkit.utilities.exceptions import FieldValueError + + shared = self._shared() + empty = mock.Mock(registry=collections.defaultdict(lambda: None)) + for prefix in ('HOPOPT', 'IPv6-Opts'): + with self.subTest(prefix=prefix): + with self.assertRaises(FieldValueError) as caught: + shared.smf_dpd_data_selector({'test': {'len': 4, 'mode': 1}}, + prefix=prefix, base=empty) + self.assertEqual(str(caught.exception), + f'{prefix}: invalid SMF DPD mode: {SMFDPDMode.H_DPD}') + + def test_shared_length_helpers(self) -> None: + from pcapkit.corekit.fields.field import NO_VALUE + + shared = self._shared() + self.assertEqual(shared.rpl_opt_sub_tlv_len({'len': 9}), 5) + self.assertEqual(shared.rpl_opt_sub_tlv_len({'len': 2}), 0) + self.assertEqual(shared.pad_opt_data_len({'len': 3}), 3) + self.assertEqual(shared.pad_opt_data_len({}), 0) + self.assertEqual(shared.pad_opt_data_len({'len': NO_VALUE}), 0) + self.assertEqual(shared.pad_opt_data_len({'len': None}), 0) + for s_type, length in ((0, 0), (1, 2), (2, 8), (3, 16)): + self.assertEqual(shared.mpl_opt_seed_id_len({'flags': {'type': s_type}}, prefix='X'), length) + self.assertEqual(shared.smf_i_dpd_id_len({'len': 5, 'info': {'type': 0, 'len': 0}}, prefix='X'), 4) + self.assertEqual(shared.smf_i_dpd_id_len({'len': 7, 'info': {'type': 2, 'len': 3}}, prefix='X'), 2) + self.assertEqual(shared.calipso_pad_len({'len': 14, 'cmpt_len': 1}, prefix='X'), 2) + self.assertEqual(shared.mpl_opt_pad_len({'len': 6, 'flags': {'type': 0}}, prefix='X'), 4) + self.assertEqual(shared.mpl_opt_pad_len({'len': 6, 'flags': {'type': 1}}, prefix='X'), 2) + + def test_shared_tid_selector_fields_and_type_update(self) -> None: + from pcapkit.const.ipv6.tagger_id import TaggerID + from pcapkit.corekit.fields.ipaddress import IPv4AddressField, IPv6AddressField + from pcapkit.corekit.fields.misc import NoValueField + from pcapkit.corekit.fields.strings import BytesField + + shared = self._shared() + for tid_type, tid_len, kind in ((0, 0, NoValueField), (2, 3, IPv4AddressField), + (3, 15, IPv6AddressField), (1, 4, BytesField)): + packet = {'info': {'type': tid_type, 'len': tid_len}} + with self.subTest(tid_type=tid_type): + self.assertIs(type(shared.smf_i_dpd_tid_selector(packet, prefix='X')), kind) + self.assertIs(packet['info']['type'], TaggerID.get(tid_type)) + self.assertEqual(shared.smf_i_dpd_tid_selector({'info': {'type': 1, 'len': 4}}, prefix='X').length, 5) + + def test_shared_selectors_return_the_schemas_they_are_given(self) -> None: + shared = self._shared() + for module, _ in self._headers(): + with self.subTest(module=module.__name__): + for mode, schema in ((0, module.SMFIdentificationBasedDPDOption), + (1, module.SMFHashBasedDPDOption)): + field = shared.smf_dpd_data_selector({'test': {'len': 4, 'mode': mode}}, + prefix='X', base=module.SMFDPDOption) + self.assertIs(field.schema, schema) + self.assertEqual(field.length, 6) + for func, schema, length in ((0, module.QuickStartRequestOption, 8), + (8, module.QuickStartReportOption, 8), + (1, module.UnassignedOption, 5)): + field = shared.quick_start_data_selector( + {'flags': {'func': func, 'len': 3}}, + base=module.QuickStartOption, unassigned=module.UnassignedOption, + ) + self.assertIs(field.schema, schema) + self.assertEqual(field.length, length) + + def test_headers_keep_their_exact_error_text(self) -> None: + from pcapkit.utilities.exceptions import FieldValueError + + for module, prefix in self._headers(): + for name, packet, message in ERRORS: + with self.subTest(module=module.__name__, helper=name, packet=packet): + with self.assertRaises(FieldValueError) as caught: + getattr(module, name)(copy.deepcopy(packet)) + self.assertEqual(str(caught.exception), f'{prefix}: {message}') + + def test_headers_delegate_with_their_own_prefix_and_schemas(self) -> None: + shared = self._shared() + for module, prefix in self._headers(): + for name, extra in BOUND.items(): + packet = {'probe': name} + with self.subTest(module=module.__name__, helper=name): + with mock.patch.object(shared, name, return_value=mock.sentinel.out) as stub: + self.assertIs(getattr(module, name)(packet), mock.sentinel.out) + stub.assert_called_once_with(packet, prefix=prefix, **{ + key: getattr(module, value) for key, value in extra.items() + }) + with self.subTest(module=module.__name__, helper='quick_start_data_selector'): + packet = {'probe': 'quick_start_data_selector'} + with mock.patch.object(shared, 'quick_start_data_selector', + return_value=mock.sentinel.out) as stub: + self.assertIs(module.quick_start_data_selector(packet), mock.sentinel.out) + stub.assert_called_once_with(packet, base=module.QuickStartOption, + unassigned=module.UnassignedOption) + + def test_headers_use_the_identical_helpers_unchanged(self) -> None: + shared = self._shared() + for module, _ in self._headers(): + for name in UNBOUND: + with self.subTest(module=module.__name__, helper=name): + self.assertIs(getattr(module, name), getattr(shared, name)) + + def test_header_selectors_return_their_own_modules_schemas(self) -> None: + hopopt, ipv6_opts = (module for module, _ in self._headers()) + self.assertIsNot(hopopt.SMFHashBasedDPDOption, ipv6_opts.SMFHashBasedDPDOption) + for module in (hopopt, ipv6_opts): + with self.subTest(module=module.__name__): + self.assertIs(module.smf_dpd_data_selector({'test': {'len': 4, 'mode': 1}}).schema, + module.SMFHashBasedDPDOption) + self.assertIs(module.quick_start_data_selector({'flags': {'func': 8, 'len': 6}}).schema, + module.QuickStartReportOption) + self.assertIs(module.quick_start_data_selector({'flags': {'func': 1, 'len': 6}}).schema, + module.UnassignedOption) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/misc/pcap/test_frame_endian_runtime.py b/tests/protocols/misc/pcap/test_frame_endian_runtime.py index 73a4d969e7..a9d32af581 100644 --- a/tests/protocols/misc/pcap/test_frame_endian_runtime.py +++ b/tests/protocols/misc/pcap/test_frame_endian_runtime.py @@ -11,8 +11,9 @@ That is what GitHub issue #605 was. :meth:`Frame.unpack ` seeded the -order under the key ``bytesorder``, while ``byteorder_callback`` in -:file:`pcapkit/protocols/schema/misc/pcap/frame.py` reads ``byteorder``, so the +order under the key ``bytesorder``, while ``byteorder_callback`` -- then in +:file:`pcapkit/protocols/schema/misc/pcap/frame.py`, now in +:file:`pcapkit/protocols/schema/misc/byteorder.py` -- reads ``byteorder``, so the ``.get()`` never found the key and always fell back to the host's order -- the sibling :meth:`Frame.pack ` eleven lines earlier spelled it correctly, which is what marks it as a slip diff --git a/tests/protocols/misc/pcap/test_header_frame_unit.py b/tests/protocols/misc/pcap/test_header_frame_unit.py index 3e82807029..e1b0916b1d 100644 --- a/tests/protocols/misc/pcap/test_header_frame_unit.py +++ b/tests/protocols/misc/pcap/test_header_frame_unit.py @@ -446,8 +446,9 @@ def test_frame_header_is_read_in_the_files_byte_order(self) -> None: :meth:`Frame.unpack ` seeded the file's byte order under the key ``bytesorder``, where - ``byteorder_callback`` in - :file:`pcapkit/protocols/schema/misc/pcap/frame.py` reads ``byteorder``, + ``byteorder_callback`` -- then in + :file:`pcapkit/protocols/schema/misc/pcap/frame.py`, now in + :file:`pcapkit/protocols/schema/misc/byteorder.py` -- reads ``byteorder``, so the lookup always missed and always fell back to :data:`sys.byteorder`. On a little-endian host reading a little-endian capture that is the right answer by coincidence, which is why every diff --git a/tests/protocols/misc/test_schema_byteorder_1519_unit.py b/tests/protocols/misc/test_schema_byteorder_1519_unit.py new file mode 100644 index 0000000000..5f545dc96d --- /dev/null +++ b/tests/protocols/misc/test_schema_byteorder_1519_unit.py @@ -0,0 +1,90 @@ +# -*- coding: utf-8 -*- +"""The byte-order callback the PCAP and PCAP-NG schemas share. + +GitHub issue #1519. The PCAP frame header and the PCAP-NG blocks each had a +``byteorder_callback``, and the PCAP-NG one additionally fell back to an +enclosing ``__packet__``. One helper now serves both, in +:mod:`pcapkit.protocols.schema.misc.byteorder`, with the PCAP-NG body: a +superset that gives the frame header the same answer on every packet the +library hands it, since :meth:`Frame.pack +` and :meth:`Frame.unpack +` always seed ``byteorder``. +This pins the helper, its ``__packet__`` fallback, and that both schemas use it. + +""" + +import importlib +import struct +import sys +import unittest + +from tests._support import reimport_once_per_class + +SHARED = 'pcapkit.protocols.schema.misc.byteorder' + +#: The byte order that is not the host's, so a fallback to the host is visible. +OTHER = 'big' if sys.byteorder == 'little' else 'little' + +#: ``(packet, expected byte order)``: its own key first, then the enclosing +#: packet's, then the host's. +CASES = ( + ({'byteorder': OTHER}, OTHER), + ({'byteorder': sys.byteorder}, sys.byteorder), + ({'__packet__': {'byteorder': OTHER}}, OTHER), + ({'byteorder': OTHER, '__packet__': {'byteorder': sys.byteorder}}, OTHER), + ({'byteorder': sys.byteorder, '__packet__': {'byteorder': OTHER}}, sys.byteorder), + ({'__packet__': {}}, sys.byteorder), + ({}, sys.byteorder), +) + + +class TestSchemaByteorder(unittest.TestCase): + """Pin the shared byte-order helper and its use by both schemas.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _shared(self) -> 'object': + return importlib.import_module(SHARED) + + def test_packet_byteorder_resolution(self) -> None: + shared = self._shared() + for packet, expected in CASES: + with self.subTest(packet=packet): + self.assertEqual(shared.packet_byteorder(packet), expected) + + def test_callback_sets_the_field_byte_order(self) -> None: + from pcapkit.corekit.fields.numbers import UInt32Field + + shared = self._shared() + for packet, expected in CASES: + with self.subTest(packet=packet): + field = UInt32Field() + shared.byteorder_callback(field, packet) + self.assertEqual(field._byteorder, expected) # pylint: disable=protected-access + + def test_frame_and_pcapng_use_the_shared_helper(self) -> None: + frame = importlib.import_module('pcapkit.protocols.schema.misc.pcap.frame') + pcapng = importlib.import_module('pcapkit.protocols.schema.misc.pcapng') + shared = self._shared() + + self.assertIs(frame.byteorder_callback, shared.byteorder_callback) + self.assertIs(pcapng.byteorder_callback, shared.byteorder_callback) + self.assertIs(pcapng.packet_byteorder, shared.packet_byteorder) + for name in ('ts_sec', 'ts_usec', 'incl_len', 'orig_len'): + with self.subTest(field=name): + self.assertIs(frame.Frame.__fields__[name]._callback, # pylint: disable=protected-access + shared.byteorder_callback) + + def test_frame_header_reads_the_byte_order_it_is_given(self) -> None: + frame = importlib.import_module('pcapkit.protocols.schema.misc.pcap.frame') + raw = struct.pack('>IIII' if OTHER == 'big' else '