diff --git a/docs/source/pcapkit/corekit/fields/strings.rst b/docs/source/pcapkit/corekit/fields/strings.rst index 2d96259cdc..a03eadb226 100644 --- a/docs/source/pcapkit/corekit/fields/strings.rst +++ b/docs/source/pcapkit/corekit/fields/strings.rst @@ -14,6 +14,8 @@ Bytes-based Fields :members: :show-inheritance: +.. autofunction:: pcapkit.corekit.fields.strings.pad_length + String-based Fields ------------------- diff --git a/pcapkit/corekit/fields/strings.py b/pcapkit/corekit/fields/strings.py index cd22e7267a..985c07bbf6 100644 --- a/pcapkit/corekit/fields/strings.py +++ b/pcapkit/corekit/fields/strings.py @@ -1,6 +1,7 @@ # -*- coding: utf-8 -*- """text field class""" +import operator import urllib.parse as urllib_parse from typing import TYPE_CHECKING, Any, Generic, TypeVar @@ -8,7 +9,7 @@ from pcapkit.corekit.sentinels import NO_VALUE from pcapkit.utilities.chardet import detect from pcapkit.utilities.compat import Dict, unquote -from pcapkit.utilities.exceptions import FieldValueError +from pcapkit.utilities.exceptions import FieldValueError, IntError __all__ = [ '_TextField', @@ -17,6 +18,7 @@ 'DecodedString', 'BitField', 'PaddingField', + 'pad_length', ] if TYPE_CHECKING: @@ -378,9 +380,23 @@ class PaddingField(BytesField): Args: length: Field size (in bytes); if a callable is given, it should return an integer value and accept the current packet as its only argument. + Left out when ``align`` sizes the field. default: Field default value, if any. callback: Callback function to be called upon :meth:`self.__call__ `. + align: Boundary, in octets, that the padding reaches, given with + ``after`` in place of ``length``. The field is then + :func:`pad_length` octets long. + after: What the padding follows: the name of an earlier field whose + value is its length in octets, or a callable that accepts the + current packet and returns that length. + + Raises: + FieldValueError: If ``length`` and ``align`` are both given or both + left out, if only one of ``align`` and ``after`` is given, if + ``after`` is neither a :obj:`str` nor a callable, or if ``align`` + is not positive. + IntError: If ``align`` is not an :obj:`int`. Padding packs back as the octets it was given, so a schema unpacked from non-zero padding rebuilds byte for byte. An absent or empty value packs as @@ -388,8 +404,43 @@ class PaddingField(BytesField): width is truncated or zero-filled to the field's, as :mod:`struct`'s ``s`` format does. + .. versionchanged:: 1.6.0 + + Takes ``align`` and ``after`` as an alternative to ``length`` + (:issue:`2614`). + """ + def __init__(self, length: 'Optional[int | Callable[[dict[str, Any]], int]]' = None, + default: 'bytes | NoValueType' = NO_VALUE, + callback: 'Callable[[Self, dict[str, Any]], None]' = lambda *_: None, *, + align: 'Optional[int]' = None, + after: 'Optional[str | Callable[[dict[str, Any]], int]]' = None) -> 'None': + name = type(self).__name__ + if align is not None: + if length is not None: + raise FieldValueError(f'{name}: takes length or align, not both') + # NOTE: A bad ``align`` is a mistake in the schema, so it is + # rejected here rather than once per packet. + pad_length(0, align) + if isinstance(after, str): + span = operator.itemgetter(after) # type: Callable[[dict[str, Any]], int] + elif callable(after): + span = after + else: + raise FieldValueError(f'{name}: align needs after, a field name or a callable, ' + f'not {after!r}') + boundary = align + + def aligned(packet: 'dict[str, Any]') -> 'int': + return pad_length(span(packet), boundary) + length = aligned + elif after is not None: + raise FieldValueError(f'{name}: after needs align') + elif length is None: + raise FieldValueError(f'{name}: needs a length or align') + super().__init__(length, default, callback) + def pack(self, value: 'Optional[bytes]', packet: 'dict[str, Any]') -> 'bytes': # pylint: disable=unused-argument """Pack field value into :obj:`bytes`. @@ -407,3 +458,32 @@ def pack(self, value: 'Optional[bytes]', packet: 'dict[str, Any]') -> 'bytes': if isinstance(value, (bytes, bytearray)) and value: return bytes(value[:length]).ljust(length, b'\x00') return bytes(length) + + +def pad_length(length: 'int', align: 'int') -> 'int': + """Octets of padding that take ``length`` octets to an ``align``-octet boundary. + + Args: + length: Octets to be padded. + align: Boundary, in octets. + + Returns: + ``-length % align``: from ``0``, where ``length`` is already a multiple + of ``align``, to ``align - 1``. + + Raises: + IntError: If ``align`` is not an :obj:`int`, or is a :obj:`bool`. + FieldValueError: If ``align`` is not positive. + + A schema's ``length`` callbacks are handed the packet data, not a + protocol, so this is a plain function and not a protocol method. + :class:`PaddingField` sizes itself by it when given ``align``. + + .. versionadded:: 1.6.0 + + """ + if isinstance(align, bool) or not isinstance(align, int): + raise IntError(f'align must be an int, not {align!r}') + if align < 1: + raise FieldValueError(f'align must be positive, not {align!r}') + return -length % align diff --git a/pcapkit/protocols/schema/internet/hip.py b/pcapkit/protocols/schema/internet/hip.py index d90abc54d6..26996947df 100644 --- a/pcapkit/protocols/schema/internet/hip.py +++ b/pcapkit/protocols/schema/internet/hip.py @@ -27,7 +27,7 @@ from pcapkit.corekit.fields.misc import ConditionalField, PayloadField, SchemaField, SwitchField from pcapkit.corekit.fields.numbers import (EnumField, NumberField, UInt8Field, UInt16Field, UInt32Field, UInt64Field) -from pcapkit.corekit.fields.strings import BitField, BytesField, PaddingField +from pcapkit.corekit.fields.strings import BitField, BytesField, PaddingField, pad_length from pcapkit.protocols.schema.schema import EnumSchema, Schema, schema_final from pcapkit.utilities.exceptions import FieldValueError from pcapkit.utilities.logging import SPHINX_TYPE_CHECKING @@ -376,7 +376,10 @@ def parameter_total_len(length: 'int') -> 'int': Total Length = 11 + Length - (Length + 3) % 8; - -- and this function is that formula, spelled the way the RFC spells it. + -- which is ``4 + Length``, the header and the contents, padded to the next + multiple of eight. This function spells it that way, through + :func:`~pcapkit.corekit.fields.strings.pad_length`; the two agree at every + ``Length``. The distinction that matters is *which* quantity gets aligned. ``Length`` is "Length of the Contents, in bytes, excluding Type, Length, and @@ -412,7 +415,8 @@ def parameter_total_len(length: 'int') -> 'int': """ if length < 0: raise FieldValueError(f'HIP: invalid parameter length: {length}') - return 11 + length - (length + 3) % 8 + total = 4 + length + return total + pad_length(total, 8) def parameter_padding_len(pkt: 'dict[str, Any]') -> 'int': diff --git a/pcapkit/protocols/schema/misc/pcapng.py b/pcapkit/protocols/schema/misc/pcapng.py index 71d717f397..3ffef2a34b 100644 --- a/pcapkit/protocols/schema/misc/pcapng.py +++ b/pcapkit/protocols/schema/misc/pcapng.py @@ -26,7 +26,7 @@ from pcapkit.corekit.fields.numbers import (EnumField, Int32Field, Int64Field, NumberField, UInt8Field, UInt16Field, UInt32Field, UInt64Field) from pcapkit.corekit.fields.strings import (BitField, BytesField, DecodedString, PaddingField, - StringField) + StringField, pad_length) from pcapkit.corekit.multidict import MultiDict, OrderedMultiDict from pcapkit.protocols.schema.schema import EnumSchema, Schema, schema_final from pcapkit.utilities.exceptions import FieldValueError, ProtocolError, stacklevel @@ -1233,7 +1233,7 @@ class UnknownOption(_OPT_Option): #: Option value. data: 'bytes' = BytesField(length=bounded_option(lambda pkt: pkt['length'])) #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', data: 'bytes') -> 'None': ... @@ -1254,7 +1254,7 @@ class CommentOption(_OPT_Option, code=Enum_OptionType.opt_comment): #: Comment text. comment: 'str' = StringField(length=bounded_option(lambda pkt: pkt['length']), encoding='utf-8') #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', comment: 'str') -> 'None': ... @@ -1272,7 +1272,7 @@ class CustomOption(_OPT_Option, code=[Enum_OptionType.opt_custom_2988, #: Custom data. data: 'bytes' = BytesField(length=bounded_option(lambda pkt: pkt['length'] - 4)) #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'int', length: 'int', pen: 'int', data: 'bytes') -> 'None': ... @@ -1398,7 +1398,7 @@ class IF_NameOption(_IF_Option, code=Enum_OptionType.if_name): #: Interface name. name: 'str' = StringField(length=bounded_option(lambda pkt: pkt['length']), encoding='utf-8') #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', name: 'str') -> 'None': ... @@ -1411,7 +1411,7 @@ class IF_DescriptionOption(_IF_Option, code=Enum_OptionType.if_description): #: Interface description. description: 'str' = StringField(length=bounded_option(lambda pkt: pkt['length']), encoding='utf-8') #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', description: 'str') -> 'None': ... @@ -1424,7 +1424,7 @@ class IF_IPv4AddrOption(_IF_Option, code=Enum_OptionType.if_IPv4addr): #: IPv4 interface. interface: 'IPv4Interface' = IPv4InterfaceField() #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', interface: 'IPv4Interface | str') -> 'None': ... @@ -1437,7 +1437,7 @@ class IF_IPv6AddrOption(_IF_Option, code=Enum_OptionType.if_IPv6addr): #: IPv6 interface. interface: 'IPv6Interface' = IPv6InterfaceField() #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', interface: 'IPv6Interface | str') -> 'None': ... @@ -1450,7 +1450,7 @@ class IF_MACAddrOption(_IF_Option, code=Enum_OptionType.if_MACaddr): #: MAC interface. interface: 'bytes' = BytesField(length=6) #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', interface: 'bytes') -> 'None': ... @@ -1463,7 +1463,7 @@ class IF_EUIAddrOption(_IF_Option, code=Enum_OptionType.if_EUIaddr): #: EUI interface. interface: 'bytes' = BytesField(length=8) #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', interface: 'bytes') -> 'None': ... @@ -1476,7 +1476,7 @@ class IF_SpeedOption(_IF_Option, code=Enum_OptionType.if_speed): #: Interface speed, in bits per second. speed: 'int' = UInt64Field(callback=byteorder_callback) #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', speed: 'int') -> 'None': ... @@ -1493,7 +1493,7 @@ class IF_TSResolOption(_IF_Option, code=Enum_OptionType.if_tsresol): 'resolution': (1, 7), }) #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') def post_process(self, packet: 'dict[str, Any]') -> 'IF_TSResolOption': """Revise ``schema`` data after unpacking process. @@ -1523,7 +1523,7 @@ class IF_TZoneOption(_IF_Option, code=Enum_OptionType.if_tzone): #: Interface time zone (as in seconds difference from GMT). tzone: 'int' = Int32Field(callback=byteorder_callback) #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', tzone: 'int') -> 'None': ... @@ -1538,7 +1538,7 @@ class IF_FilterOption(_IF_Option, code=Enum_OptionType.if_filter): #: Capture filter. filter: 'bytes' = BytesField(length=bounded_option(lambda pkt: pkt['length'] - 1)) #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', code: 'Enum_FilterType', filter: 'bytes') -> 'None': ... @@ -1551,7 +1551,7 @@ class IF_OSOption(_IF_Option, code=Enum_OptionType.if_os): #: OS information. os: 'str' = StringField(length=bounded_option(lambda pkt: pkt['length']), encoding='utf-8') #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', os: 'str') -> 'None': ... @@ -1564,7 +1564,7 @@ class IF_FCSLenOption(_IF_Option, code=Enum_OptionType.if_fcslen): #: FCS length. fcslen: 'int' = UInt8Field(callback=byteorder_callback) #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', fcslen: 'int') -> 'None': ... @@ -1577,7 +1577,7 @@ class IF_TSOffsetOption(_IF_Option, code=Enum_OptionType.if_tsoffset): #: Timestamp offset (in seconds). tsoffset: 'int' = Int64Field(callback=byteorder_callback) #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', tsoffset: 'int') -> 'None': ... @@ -1590,7 +1590,7 @@ class IF_HardwareOption(_IF_Option, code=Enum_OptionType.if_hardware): #: Hardware information. hardware: 'str' = StringField(length=bounded_option(lambda pkt: pkt['length']), encoding='utf-8') #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', hardware: 'str') -> 'None': ... @@ -1603,7 +1603,7 @@ class IF_TxSpeedOption(_IF_Option, code=Enum_OptionType.if_txspeed): #: Interface transmit speed, in bits per second. tx_speed: 'int' = UInt64Field(callback=byteorder_callback) #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', tx_speed: 'int') -> 'None': ... @@ -1616,7 +1616,7 @@ class IF_RxSpeedOption(_IF_Option, code=Enum_OptionType.if_rxspeed): #: Interface receive speed, in bits per second. rx_speed: 'int' = UInt64Field(callback=byteorder_callback) #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', rx_speed: 'int') -> 'None': ... @@ -1689,7 +1689,7 @@ class EPB_FlagsOption(_EPB_Option, code=Enum_OptionType.epb_flags): #: Flags. flags: 'EPBFlags' = FlagsField(namespace=PACKET_FLAGS, callback=byteorder_callback) #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', flags: 'EPBFlags') -> 'None': ... @@ -1704,7 +1704,7 @@ class EPB_HashOption(_EPB_Option, code=Enum_OptionType.epb_hash): #: Hash value. data: 'bytes' = BytesField(length=bounded_option(lambda pkt: pkt['length'] - 1)) #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', func: 'Enum_HashAlgorithm', data: 'bytes') -> 'None': ... @@ -1717,7 +1717,7 @@ class EPB_DropCountOption(_EPB_Option, code=Enum_OptionType.epb_dropcount): #: Number of packets dropped by the interface. drop_count: 'int' = UInt64Field(callback=byteorder_callback) #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', drop_count: 'int') -> 'None': ... @@ -1730,7 +1730,7 @@ class EPB_PacketIDOption(_EPB_Option, code=Enum_OptionType.epb_packetid): #: Packet ID. packet_id: 'int' = UInt64Field(callback=byteorder_callback) #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', packet_id: 'int') -> 'None': ... @@ -1743,7 +1743,7 @@ class EPB_QueueOption(_EPB_Option, code=Enum_OptionType.epb_queue): #: Queue ID. queue_id: 'int' = UInt32Field(callback=byteorder_callback) #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', queue_id: 'int') -> 'None': ... @@ -1758,7 +1758,7 @@ class EPB_VerdictOption(_EPB_Option, code=Enum_OptionType.epb_verdict): #: Verdict value. value: 'bytes' = BytesField(length=bounded_option(lambda pkt: pkt['length'] - 1)) #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', verdict: 'Enum_VerdictType', value: 'bytes') -> 'None': ... @@ -1785,7 +1785,7 @@ class EnhancedPacketBlock(BlockType, code=Enum_BlockType.Enhanced_Packet_Block): #: Packet data, bounded by the block (see :func:`captured_area`). packet_data: 'bytes' = PayloadField(length=captured_area('captured_len')) #: Padding. - padding_data: 'bytes' = PaddingField(length=lambda pkt: (4 - captured_area('captured_len')(pkt) % 4) % 4) + padding_data: 'bytes' = PaddingField(align=4, after=captured_area('captured_len')) #: Options. options: 'list[Option]' = OptionAreaField( # NOTE: The padding is recomputed here rather than read back from @@ -1793,7 +1793,7 @@ class EnhancedPacketBlock(BlockType, code=Enum_BlockType.Enhanced_Packet_Block): # buffer while packing and never lands in the packet data, so its name # is not a key here on the packing path. length=bounded_area(lambda pkt: pkt['length'] - BLOCK_MINIMUM[Enum_BlockType.Enhanced_Packet_Block] - - (captured_area('captured_len')(pkt) + 3) // 4 * 4), + - (size := captured_area('captured_len')(pkt)) - pad_length(size, 4)), base_schema=_EPB_Option, type_name='type', registry=Option.registry['epb'], @@ -1853,7 +1853,7 @@ def _spb_data_length(packet: 'dict[str, Any]') -> 'int': snaplen = packet.get('snaplen') if snaplen is not None and snaplen < size: size = snaplen - if area > (size + 3) // 4 * 4: + if area > size + pad_length(size, 4): size = area return size @@ -1872,7 +1872,7 @@ class SimplePacketBlock(BlockType, code=Enum_BlockType.Simple_Packet_Block): #: block holds; see :func:`_spb_data_length`. packet_data: 'bytes' = PayloadField(length=_spb_data_length) #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - len(pkt['packet_data']) % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after=lambda pkt: len(pkt['packet_data'])) #: Block total length. length2: 'int' = UInt32Field(callback=byteorder_callback) @@ -1932,7 +1932,7 @@ class UnknownRecord(NameResolutionRecord): #: Unknown record data. data: 'bytes' = BytesField(length=bounded_option(lambda pkt: pkt['length'])) #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'Enum_RecordType', length: 'int', data: 'bytes') -> 'None': ... @@ -1955,7 +1955,7 @@ class IPv4Record(NameResolutionRecord, code=Enum_RecordType.nrb_record_ipv4): #: Name resolution data. resol: 'str' = StringField(length=bounded_option(lambda pkt: pkt['length'] - 4), encoding='utf-8') #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') def post_process(self, packet: 'dict[str, Any]') -> 'Schema': """Revise ``schema`` data after unpacking process. @@ -1986,7 +1986,7 @@ class IPv6Record(NameResolutionRecord, code=Enum_RecordType.nrb_record_ipv6): #: Name resolution data. resol: 'str' = StringField(length=bounded_option(lambda pkt: pkt['length'] - 16), encoding='utf-8') #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') def post_process(self, packet: 'dict[str, Any]') -> 'Schema': """Revise ``schema`` data after unpacking process. @@ -2024,7 +2024,7 @@ class NS_DNSNameOption(_NS_Option, code=Enum_OptionType.ns_dnsname): #: DNS name. name: 'str' = StringField(length=bounded_option(lambda pkt: pkt['length']), encoding='utf-8') #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', name: 'str') -> 'None': ... @@ -2302,7 +2302,7 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Self': try: self.data = self._read_entries() rebuilt = self.dump_entries(cast('list[OrderedMultiDict[str, str | bytes]]', self.data)) - if self.entry not in (rebuilt, rebuilt + bytes(-len(rebuilt) % 4)): + if self.entry not in (rebuilt, rebuilt + bytes(pad_length(len(rebuilt), 4))): raise ProtocolError('PCAP-NG: [systemd Journal Export] parsed entry does not ' 'rebuild to the octets captured') except ProtocolError as error: @@ -2630,13 +2630,13 @@ class DecryptionSecretsBlock(BlockType, code=Enum_BlockType.Decryption_Secrets_B selector=dsb_secrets_selector, ) #: Padding. - padding_data: 'bytes' = PaddingField(length=lambda pkt: (4 - secrets_area(pkt) % 4) % 4) + padding_data: 'bytes' = PaddingField(align=4, after=secrets_area) #: Options. options: 'list[Option]' = OptionAreaField( # NOTE: see EnhancedPacketBlock.options on why the padding is recomputed # here instead of being read back from ``padding_data``. length=nonnegative(lambda pkt: pkt['length'] - BLOCK_MINIMUM[Enum_BlockType.Decryption_Secrets_Block] - - (secrets_area(pkt) + 3) // 4 * 4), + - (size := secrets_area(pkt)) - pad_length(size, 4)), base_schema=_DSB_Option, type_name='type', registry=Option.registry['dsb'], @@ -2731,7 +2731,7 @@ class PACK_FlagsOption(_PACK_Option, code=Enum_OptionType.pack_flags): #: Flags. flags: 'PACKFlags' = FlagsField(namespace=PACKET_FLAGS, callback=byteorder_callback) #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', flags: 'PACKFlags') -> 'None': ... @@ -2746,7 +2746,7 @@ class PACK_HashOption(_PACK_Option, code=Enum_OptionType.pack_hash): #: Hash value. data: 'bytes' = BytesField(length=bounded_option(lambda pkt: pkt['length'] - 1)) #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) + padding: 'bytes' = PaddingField(align=4, after='length') if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', func: 'Enum_HashAlgorithm', data: 'bytes') -> 'None': ... @@ -2775,13 +2775,13 @@ class PacketBlock(BlockType, code=Enum_BlockType.Packet_Block): #: Packet data, bounded by the block (see :func:`captured_area`). packet_data: 'bytes' = PayloadField(length=captured_area('captured_length')) #: Padding. - padding_data: 'bytes' = PaddingField(length=lambda pkt: (4 - captured_area('captured_length')(pkt) % 4) % 4) + padding_data: 'bytes' = PaddingField(align=4, after=captured_area('captured_length')) #: Options. options: 'list[Option]' = OptionAreaField( # NOTE: see EnhancedPacketBlock.options on why the padding is recomputed # here instead of being read back from ``padding_data``. length=bounded_area(lambda pkt: pkt['length'] - BLOCK_MINIMUM[Enum_BlockType.Packet_Block] - - (captured_area('captured_length')(pkt) + 3) // 4 * 4), + - (size := captured_area('captured_length')(pkt)) - pad_length(size, 4)), base_schema=_PACK_Option, type_name='type', registry=Option.registry['pack'], diff --git a/pcapkit/protocols/schema/transport/sctp.py b/pcapkit/protocols/schema/transport/sctp.py index b501173591..f40ae8017f 100644 --- a/pcapkit/protocols/schema/transport/sctp.py +++ b/pcapkit/protocols/schema/transport/sctp.py @@ -16,7 +16,7 @@ from pcapkit.corekit.fields.ipaddress import IPv4AddressField, IPv6AddressField from pcapkit.corekit.fields.misc import PayloadField, SchemaField from pcapkit.corekit.fields.numbers import EnumField, PortEnumField, UInt16Field, UInt32Field -from pcapkit.corekit.fields.strings import BitField, BytesField, PaddingField +from pcapkit.corekit.fields.strings import BitField, BytesField, PaddingField, pad_length from pcapkit.protocols.schema.schema import EnumSchema, Schema, schema_final from pcapkit.utilities.logging import SPHINX_TYPE_CHECKING @@ -103,7 +103,7 @@ def padding_length(pkt: 'dict[str, Any]') -> 'int': """ length = pkt.get('length') or 0 - padding = -length % 4 + padding = pad_length(length, 4) width = pkt.get('__padding__') if isinstance(width, int) and width >= 0: @@ -206,7 +206,7 @@ def nested_length(base: 'int') -> 'Callable[[dict[str, Any]], int]': """ def callback(pkt: 'dict[str, Any]') -> 'int': length = pkt.get('length') or 0 - return max(length - base + (-length % 4), 0) + return max(length - base + pad_length(length, 4), 0) return callback @@ -883,7 +883,7 @@ def whole_chunks_length(data: 'bytes') -> 'int': stop = end + length if length < 4 or stop > size: break - end = min(stop + (-length % 4), size) + end = min(stop + pad_length(length, 4), size) return end diff --git a/tests/corekit/test_fields_all_exports_2174_unit.py b/tests/corekit/test_fields_all_exports_2174_unit.py index 61076a8ed6..4a3008b39a 100644 --- a/tests/corekit/test_fields_all_exports_2174_unit.py +++ b/tests/corekit/test_fields_all_exports_2174_unit.py @@ -46,11 +46,12 @@ #: * ``Field`` and ``_TextField`` are abstract bases, kept out of ``import *`` as #: every package's base class is (see #: :data:`tests.project.test_public_api.DELIBERATE_NON_EXPORTS`). -#: * ``packet_byteorder``, ``byteorder_callback``, ``parse_ip_address`` and -#: ``nested_packet_context`` are helpers for writing a field or its callbacks, -#: reached by their module path. ``nested_packet_context`` in particular builds -#: the packet context a nested schema's callbacks receive; star-exporting it -#: would drop a generic name into the caller's namespace (#2174). +#: * ``packet_byteorder``, ``byteorder_callback``, ``parse_ip_address``, +#: ``nested_packet_context`` and ``pad_length`` are helpers for writing a field +#: or its callbacks, reached by their module path. ``nested_packet_context`` +#: in particular builds the packet context a nested schema's callbacks +#: receive; star-exporting it would drop a generic name into the caller's +#: namespace (#2174). #: * ``DecodedString`` is the value type :class:`~pcapkit.corekit.fields.strings.StringField` #: returns, not a field. FIELDS_DELIBERATE_NON_EXPORTS = { @@ -58,7 +59,7 @@ 'pcapkit.corekit.fields.field': ('Field',), 'pcapkit.corekit.fields.ipaddress': ('parse_ip_address',), 'pcapkit.corekit.fields.misc': ('nested_packet_context',), - 'pcapkit.corekit.fields.strings': ('DecodedString', '_TextField'), + 'pcapkit.corekit.fields.strings': ('DecodedString', '_TextField', 'pad_length'), } diff --git a/tests/corekit/test_fields_pad_length_2614_unit.py b/tests/corekit/test_fields_pad_length_2614_unit.py new file mode 100644 index 0000000000..2da2f3c35a --- /dev/null +++ b/tests/corekit/test_fields_pad_length_2614_unit.py @@ -0,0 +1,306 @@ +# -*- coding: utf-8 -*- +"""``pad_length`` and ``PaddingField(align=...)`` size padding to a boundary. + +GitHub issue #2614 found the octets that reach the next ``N``-octet boundary +spelled several ways, ``-n % 4``, ``(4 - n % 4) % 4`` and ``(n + 3) // 4 * 4`` +among them. :func:`~pcapkit.corekit.fields.strings.pad_length` states it once, and +:class:`~pcapkit.corekit.fields.strings.PaddingField` takes ``align`` and +``after`` so a schema says what it aligns rather than how. + +The expected values here never call ``pad_length``: :func:`spec_pad` counts the +octets up to the boundary one at a time, as the specifications word it. The +PCAP-NG cases check every schema in +:mod:`pcapkit.protocols.schema.misc.pcapng` that sizes its padding with +``align``, so a site sized against the wrong field or boundary fails by name. + +:mod:`pcapkit` is imported inside each test, after +:func:`~tests._support.reimport_once_per_class`, not at module load. + +""" + +import struct +import unittest +import warnings +from typing import TYPE_CHECKING + +from tests._support import reimport_once_per_class + +if TYPE_CHECKING: + from typing import Any + +#: Test tier (#1958): exercises integer edge cases -- operands of ``%`` far past +#: any wire field's width -- as ``test_fields_strings`` does. +__tier__ = 'ALL' + +#: Boundaries checked, the ones the schemas use and odd ones besides. +ALIGNS = (1, 2, 3, 4, 7, 8, 16) + +#: Lengths far past any wire field, where a width-limited formula would wrap. +LARGE = (2**16 - 1, 2**32 + 1, 2**64, 2**64 + 3, 10**30 + 7) + + +def spec_pad(length: int, align: int) -> int: + """Octets added to ``length`` until it is a multiple of ``align``, counted one at a time.""" + pad = 0 + while (length + pad) % align: + pad += 1 + return pad + + +class PadLengthTests(unittest.TestCase): + """:func:`~pcapkit.corekit.fields.strings.pad_length` itself.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + from pcapkit.corekit.fields.strings import pad_length + + self.pad_length = pad_length + + def test_boundaries(self) -> None: + for align in ALIGNS: + for length, expected in ((0, 0), (1, (align - 1) % align), (align - 1, 1 % align), + (align, 0), (align + 1, (align - 1) % align), (2 * align, 0)): + with self.subTest(align=align, length=length): + self.assertEqual(self.pad_length(length, align), expected) + + def test_matches_the_count_up_to_the_boundary(self) -> None: + for align in ALIGNS: + for length in range(0, 4 * align + 2): + with self.subTest(align=align, length=length): + self.assertEqual(self.pad_length(length, align), spec_pad(length, align)) + + def test_large_lengths(self) -> None: + for align in ALIGNS: + for length in LARGE: + with self.subTest(align=align, length=length): + pad = self.pad_length(length, align) + self.assertEqual((length + pad) % align, 0) + self.assertTrue(0 <= pad < align) + + def test_result_is_an_int(self) -> None: + self.assertIs(type(self.pad_length(5, 4)), int) + + def test_non_positive_align_is_refused(self) -> None: + from pcapkit.utilities.exceptions import FieldValueError + + for align in (0, -1, -4): + with self.subTest(align=align): + with self.assertRaisesRegex(FieldValueError, 'align must be positive'): + self.pad_length(4, align) + self.assertTrue(issubclass(FieldValueError, ValueError)) + + def test_non_int_align_is_refused(self) -> None: + from pcapkit.utilities.exceptions import IntError + + for align in (4.0, True, False, None, '4', b'\x04'): + with self.subTest(align=align): + with self.assertRaisesRegex(IntError, 'align must be an int'): + self.pad_length(4, align) # type: ignore[arg-type] + self.assertTrue(issubclass(IntError, TypeError)) + + +class PaddingFieldAlignTests(unittest.TestCase): + """:class:`~pcapkit.corekit.fields.strings.PaddingField` with ``align`` and ``after``.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + from pcapkit.corekit.fields.strings import PaddingField + + self.PaddingField = PaddingField + + def test_after_a_field_name(self) -> None: + field = self.PaddingField(align=4, after='length') + for length in (0, 1, 3, 4, 5, 7, 8) + LARGE: + with self.subTest(length=length): + self.assertEqual(field({'length': length}).length, spec_pad(length, 4)) + + def test_after_a_callable(self) -> None: + field = self.PaddingField(align=8, after=lambda pkt: len(pkt['data'])) + for size in range(0, 18): + with self.subTest(size=size): + self.assertEqual(field({'data': bytes(size)}).length, spec_pad(size, 8)) + + def test_each_packet_resolves_its_own_width(self) -> None: + field = self.PaddingField(align=4, after='length') + self.assertEqual(field({'length': 5}).length, 3) + self.assertEqual(field({'length': 8}).length, 0) + + def test_length_callers_are_unchanged(self) -> None: + self.assertEqual(self.PaddingField(length=3).length, 3) + self.assertEqual(self.PaddingField(3).length, 3) + self.assertEqual(self.PaddingField(length=lambda pkt: pkt['n'])({'n': 2}).length, 2) + + def test_pack_and_unpack_in_a_schema(self) -> None: + from pcapkit.corekit.fields.numbers import UInt8Field, UInt16Field + from pcapkit.corekit.fields.strings import BytesField + from pcapkit.protocols.schema.schema import Schema, schema_final + + @schema_final + class TLV(Schema): + """A value padded to four octets, then a trailing octet.""" + + length: 'int' = UInt16Field() + value: 'bytes' = BytesField(length=lambda pkt: pkt['length']) + padding: 'bytes' = self.PaddingField(align=4, after='length') + tail: 'int' = UInt8Field() + + for value in (b'', b'a', b'abc', b'abcd', b'abcde'): + with self.subTest(value=value): + built = TLV(length=len(value), value=value, tail=9).pack() + expected = len(value).to_bytes(2, 'big') + value + bytes(spec_pad(len(value), 4)) + b'\x09' + self.assertEqual(built, expected) + + # captured non-zero padding parses at the right width and rebuilds as captured + raw = b'\x00\x03abc\xee\x09' + parsed = TLV.unpack(raw, len(raw), None) + self.assertEqual(parsed.padding, b'\xee') + self.assertEqual(parsed.tail, 9) + self.assertEqual(parsed.pack(), raw) + + def test_bad_declarations_are_refused_at_declaration(self) -> None: + from pcapkit.utilities.exceptions import FieldValueError, IntError + + for kwargs, kind, message in ( + ({}, FieldValueError, 'needs a length or align'), + ({'length': 2, 'align': 4, 'after': 'length'}, FieldValueError, 'not both'), + ({'after': 'length'}, FieldValueError, 'after needs align'), + ({'align': 4}, FieldValueError, 'align needs after'), + ({'align': 4, 'after': 3}, FieldValueError, 'align needs after'), + ({'align': 0, 'after': 'length'}, FieldValueError, 'align must be positive'), + ({'align': -4, 'after': 'length'}, FieldValueError, 'align must be positive'), + ({'align': 4.0, 'after': 'length'}, IntError, 'align must be an int'), + ({'align': True, 'after': 'length'}, IntError, 'align must be an int'), + ): + with self.subTest(kwargs=kwargs): + with self.assertRaisesRegex(kind, message): + self.PaddingField(**kwargs) + + +class PCAPNGAlignedPaddingTests(unittest.TestCase): + """Every PCAP-NG ``align`` site pads its value to a 32-bit boundary. + + The PCAP-NG specification pads an option's value, a name resolution + record's value, and a block's packet or secrets data to 32 bits, and counts + none of that padding in the length that precedes it. + + """ + + def setUp(self) -> None: + reimport_once_per_class(self) + + from pcapkit.protocols.schema.misc import pcapng + + self.pcapng = pcapng + + def _length_sized(self) -> 'dict[str, type]': + """Options and records, whose padding follows their ``length`` field.""" + found = {} + for name, obj in vars(self.pcapng).items(): + if (isinstance(obj, type) and obj.__module__ == self.pcapng.__name__ + and issubclass(obj, (self.pcapng.Option, self.pcapng.NameResolutionRecord)) + and 'padding' in obj.__fields__): + found[name] = obj + return found + + def test_options_and_records_pad_their_value(self) -> None: + found = self._length_sized() + for name in ('CommentOption', 'CustomOption', 'UnknownOption', 'IF_NameOption', + 'EPB_HashOption', 'IPv4Record', 'UnknownRecord'): + self.assertIn(name, found) + for name, cls in found.items(): + field = cls.__fields__['padding'] + for length in range(0, 9): + with self.subTest(schema=name, length=length): + self.assertEqual(field({'length': length}).length, spec_pad(length, 4)) + + def test_packet_blocks_pad_their_packet_data(self) -> None: + for name, key in (('EnhancedPacketBlock', 'captured_len'), ('PacketBlock', 'captured_length')): + field = getattr(self.pcapng, name).__fields__['padding_data'] + for size in range(0, 9): + with self.subTest(schema=name, captured=size): + packet = {key: size, 'length': 32 + 64} + self.assertEqual(field(packet).length, spec_pad(size, 4)) + + def _parse(self, octets: 'bytes') -> 'tuple[Any, bytes, list[str]]': + """Parse ``octets`` as one little-endian block, rebuild it, and keep the warnings.""" + from pcapkit.const.pcapng.block_type import BlockType + from pcapkit.const.reg.linktype import LinkType + from pcapkit.foundation.engines.pcapng import Context + from pcapkit.protocols.misc.pcapng import PCAPNG + + section = PCAPNG(num=0, sct=1, ctx=None, type=BlockType.Section_Header_Block, block={}) + context = Context(section.info) # type: Any + interface = PCAPNG(num=1, sct=1, ctx=context, type=BlockType.Interface_Description_Block, + block={'linktype': LinkType.ETHERNET, 'snaplen': 0x40000}) + context.interfaces.append(interface.info) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + parsed = PCAPNG(octets, len(octets), num=2, sct=1, ctx=context) + rebuilt = PCAPNG.from_data(parsed.info, num=2, sct=1, ctx=context).data + return parsed, rebuilt, [str(item.message) for item in caught] + + def test_simple_packet_block_pads_its_packet_data(self) -> None: + field = self.pcapng.SimplePacketBlock.__fields__['padding'] + for size in range(0, 9): + with self.subTest(size=size): + self.assertEqual(field({'packet_data': bytes(size)}).length, spec_pad(size, 4)) + + def test_simple_packet_block_padding_is_not_packet_data(self) -> None: + """Five octets of an ``original_len`` of five, then three of padding. + + The block's area is the octets past its 16 of framing. From the data's + own five octets up to the data and its padding, the data stays + ``original_len`` octets and the rest is padding, even where the area + stops short of the padding; only an area past both keeps the excess as + data (#1384, #1414). + + """ + for area, expected in ((5, 5), (6, 5), (7, 5), (8, 5), (9, 9), (12, 12)): + with self.subTest(area=area): + packet = {'length': 16 + area, 'original_len': 5} + self.assertEqual(self.pcapng._spb_data_length(packet), expected) # pylint: disable=protected-access + + octets = struct.pack(' None: + field = self.pcapng.DecryptionSecretsBlock.__fields__['padding_data'] + for size in range(0, 9): + with self.subTest(size=size): + packet = {'secrets_length': size, 'length': 20 + 64} + self.assertEqual(field(packet).length, spec_pad(size, 4)) + + def test_decryption_secrets_block_pads_only_the_secrets_it_holds(self) -> None: + """An over-declared Secrets Length pads the secrets the block holds (#1980).""" + field = self.pcapng.DecryptionSecretsBlock.__fields__['padding_data'] + for held in (4, 5, 6, 7): + for declared in (held + 1, held + 2, held + 3): + with self.subTest(held=held, declared=declared): + self.assertNotEqual(spec_pad(held, 4), spec_pad(declared, 4)) + packet = {'secrets_length': declared, 'length': 20 + held} + self.assertEqual(field(packet).length, spec_pad(held, 4)) + + for declared in (5, 6, 7): + with self.subTest(block='abcd', declared=declared): + octets = (struct.pack(' None: + """``opt_comment`` of five octets: four of header, five of value, three of padding.""" + from pcapkit.const.pcapng.option_type import OptionType + + option = self.pcapng.CommentOption(type=OptionType.opt_comment, length=5, comment='hello') + raw = option.pack({'byteorder': 'big'}) + self.assertEqual(raw, b'\x00\x01\x00\x05hello\x00\x00\x00')