diff --git a/pcapkit/foundation/engines/pcapng.py b/pcapkit/foundation/engines/pcapng.py index 81462e1b81..7d38f5a0e1 100644 --- a/pcapkit/foundation/engines/pcapng.py +++ b/pcapkit/foundation/engines/pcapng.py @@ -14,6 +14,7 @@ from typing import TYPE_CHECKING, cast from pcapkit.const.pcapng.block_type import BlockType as Enum_BlockType +from pcapkit.corekit.fields.strings import pad_length from pcapkit.corekit.infoclass import Info, info_final from pcapkit.corekit.io import SeekableReader from pcapkit.dumpkit.pcap import PCAPIO @@ -458,7 +459,7 @@ def _bound_read_ahead(self) -> 'None': size = length if block_type == Enum_BlockType.Decryption_Secrets_Block: secrets = int.from_bytes(self._peek(16)[12:16], byteorder) - size = max(size, BLOCK_MINIMUM[block_type] + secrets + -secrets % 4) + size = max(size, BLOCK_MINIMUM[block_type] + secrets + pad_length(secrets, 4)) # NOTE: No buffer holds more than :data:`sys.maxsize` octets, so a block # longer than that is read ahead only so far; the reader moves past the # rest itself (#1975). diff --git a/pcapkit/foundation/engines/scapy.py b/pcapkit/foundation/engines/scapy.py index 05153d8824..ab401b0684 100644 --- a/pcapkit/foundation/engines/scapy.py +++ b/pcapkit/foundation/engines/scapy.py @@ -38,6 +38,7 @@ import struct from typing import TYPE_CHECKING, cast +from pcapkit.corekit.fields.strings import pad_length from pcapkit.foundation.engines.engine import EngineBase from pcapkit.foundation.reassembly import ReassemblyManager from pcapkit.protocols.schema.misc.pcap.frame import RECORD_HEADER_LEN @@ -106,7 +107,7 @@ def _read_tsoffset(options: 'bytes', endian: 'str') -> 'int': break if code == 14 and length == 8 and len(options) >= 12: return cast('int', struct.unpack(f'{endian}q', options[4:12])[0]) - options = options[4 + length + (-length % 4):] + options = options[4 + length + pad_length(length, 4):] return 0 diff --git a/pcapkit/protocols/link/arp.py b/pcapkit/protocols/link/arp.py index 13735aa06e..6d62d09994 100644 --- a/pcapkit/protocols/link/arp.py +++ b/pcapkit/protocols/link/arp.py @@ -44,7 +44,6 @@ """ import ipaddress -import re from typing import TYPE_CHECKING, cast from pcapkit.const.arp.hardware import Hardware as Enum_Hardware @@ -72,9 +71,6 @@ __all__ = ['ARP'] -# Ethernet address pattern, of any number of octets (``hlen`` is not fixed at 6) -PAT_MAC_ADDR = re.compile(rb'(?i)(?:[0-9a-f]{2}(?:[:-][0-9a-f]{2})*)?') - class ARP(Link[Data_ARP, Schema_ARP], schema=Schema_ARP, data=Data_ARP): @@ -374,10 +370,8 @@ def _read_addr_resolve(self, addr: 'bytes', htype: 'int') -> 'str': """ if htype == Enum_Hardware.Ethernet: # Ethernet - _addr = addr.hex(':') - else: - _addr = addr.hex() - return _addr + return self._read_mac_addr(addr) + return addr.hex() def _make_addr_resolve(self, addr: 'str | bytes | bytearray', htype: 'int') -> 'bytes': """Resolve hardware address according to protocol. @@ -389,8 +383,9 @@ def _make_addr_resolve(self, addr: 'str | bytes | bytearray', htype: 'int') -> ' Returns: Hardware address packed as raw octets. If ``htype`` is ``1``, i.e. MAC address, the ``:``- or ``-``-separated *hex* encoded MAC address - is validated and packed, one octet per group (an empty string packs - as no octets); otherwise, a :class:`str` is + is packed by :meth:`~pcapkit.protocols.protocol.Protocol._make_mac_addr`, + one octet per group, of any number since ``hlen`` is not fixed at 6 + (an empty string packs as no octets); otherwise, a :class:`str` is taken as the *hex* encoded address that :meth:`_read_addr_resolve` returns, and :class:`bytes`/:class:`bytearray` are returned as :class:`bytes`. @@ -409,20 +404,16 @@ def _make_addr_resolve(self, addr: 'str | bytes | bytearray', htype: 'int') -> ' :meth:`_make_proto_resolve` gives. """ - name = 'MAC address' if htype == Enum_Hardware.Ethernet else 'hardware address' + if htype == Enum_Hardware.Ethernet: + return self._make_mac_addr(addr, size=None, alias=self.__class__.__name__) if not isinstance(addr, (str, bytes, bytearray)): - raise ProtocolError(f'{self.__class__.__name__}: invalid {name}: must be a str, ' + raise ProtocolError(f'{self.__class__.__name__}: invalid hardware address: must be a str, ' f'bytes or bytearray, not {addr!r}') - if htype == Enum_Hardware.Ethernet: - _addr = addr.encode() if isinstance(addr, str) else addr - if PAT_MAC_ADDR.fullmatch(_addr) is not None: - return bytes.fromhex(_addr.replace(b':', b'').replace(b'-', b'').decode()) - raise ProtocolError(f'{self.__class__.__name__}: invalid {name}: {addr!r}') if isinstance(addr, str): try: return bytes.fromhex(addr.replace(':', '').replace('-', '')) except ValueError: - raise ProtocolError(f'{self.__class__.__name__}: invalid {name}: ' + raise ProtocolError(f'{self.__class__.__name__}: invalid hardware address: ' f'{addr!r}') from None return bytes(addr) diff --git a/pcapkit/protocols/misc/pcapng.py b/pcapkit/protocols/misc/pcapng.py index 9ec4629ae1..d4d0f5e3a0 100644 --- a/pcapkit/protocols/misc/pcapng.py +++ b/pcapkit/protocols/misc/pcapng.py @@ -48,7 +48,6 @@ import io import math import platform -import re import struct import sys import time @@ -65,7 +64,7 @@ from pcapkit.const.reg.linktype import LinkType as Enum_LinkType from pcapkit.corekit.enum import EnumLookup from pcapkit.corekit.fields.byteorder import packet_byteorder -from pcapkit.corekit.fields.strings import DecodedString +from pcapkit.corekit.fields.strings import DecodedString, pad_length from pcapkit.corekit.infoclass import OrderedMultiInfo from pcapkit.corekit.io import advance from pcapkit.corekit.module import ModuleDescriptor @@ -260,9 +259,6 @@ _FlagsEnum = TypeVar('_FlagsEnum', 'PacketDirection', 'PacketReception') _S = TypeVar('_S', bound='Schema | bytes') -# EUI address pattern -PAT_EUI_ADDR = re.compile(rb'(?i)(?:[0-9a-f]{2}[:-]){7}[0-9a-f]{2}') - def _option_key(code: 'Enum_OptionType') -> 'Union[Enum_OptionType, Tuple[str, int]]': """Return a collision-free key for PCAP-NG option registries. @@ -2336,65 +2332,30 @@ def _make_timestamp(self, timestamp: 'Optional[float | Decimal | dt_type | int]' ) return (ts_info >> 32) & 0xFFFF_FFFF, ts_info & 0xFFFF_FFFF - def _read_eui_addr(self, addr: 'bytes') -> 'str': - """Read EUI address. - - Args: - addr: EUI address. - - Returns: - Colon (``:``) separated *hex* encoded EUI address. - - """ - return addr.hex(':') - def _make_mac_addr(self, addr: 'str | bytes | bytearray', *, size: 'Optional[int]' = 6, name: 'str' = 'MAC address', alias: 'Optional[str]' = None) -> 'bytes': - """Make MAC address. + """Make a MAC address, or an EUI-64 one with ``size=8``. Args: - addr: MAC address. + addr: MAC or EUI address. size: Octets in the address, as for :meth:`Protocol._make_mac_addr `. name: Name of the address, for the error message. alias: Prefix of the error message, ``'PCAP-NG'`` by default. Returns: - MAC address packed as ``size`` octets, 6 by default. + The address packed as ``size`` octets, 6 by default. Raises: ProtocolError: If ``addr`` is not a :class:`str`, :class:`bytes` or :class:`bytearray` (a :obj:`bool` included), or is not a - ``:``- or ``-``-separated *hex* encoded MAC address. + ``:``- or ``-``-separated *hex* encoded address of ``size`` + octets. """ prefix = 'PCAP-NG' if alias is None else alias return super()._make_mac_addr(addr, size=size, name=name, alias=prefix) - def _make_eui_addr(self, addr: 'str | bytes | bytearray') -> 'bytes': - """Make EUI address. - - Args: - addr: EUI address. - - Returns: - EUI address packed as 8 octets. - - Raises: - ProtocolError: If ``addr`` is not a :class:`str`, :class:`bytes` - or :class:`bytearray` (a :obj:`bool` included), or is not a - ``:``- or ``-``-separated *hex* encoded EUI-64 address. - - """ - if not isinstance(addr, (str, bytes, bytearray)): - raise ProtocolError(f'PCAP-NG: invalid EUI address: must be a str, bytes or ' - f'bytearray, not {addr!r}') - _addr = addr.encode() if isinstance(addr, str) else addr - - if PAT_EUI_ADDR.fullmatch(_addr) is not None: - return bytes.fromhex(_addr.replace(b':', b'').replace(b'-', b'').decode()) - raise ProtocolError(f'PCAP-NG: invalid EUI address: {addr!r}') - def _decode_next_layer(self, dict_: 'Data_PCAPNG', proto: 'Optional[int]' = None, length: 'Optional[int]' = None, *, packet: 'Optional[dict[str, Any]]' = None) -> 'Data_PCAPNG': # pylint: disable=arguments-differ r"""Decode next layer protocol. @@ -2724,7 +2685,7 @@ def _read_block_spb(self, schema: 'Schema_SimplePacketBlock', *, ProtocolWarning, stacklevel=stacklevel()) # NOTE: Likewise for more packet data than the original length, padding # aside (#1414). - if data.captured_len > (data.original_len + 3) // 4 * 4: + if data.captured_len > data.original_len + pad_length(data.original_len, 4): warn(f'PCAP-NG: [Block {header.type}] packet data of {data.captured_len} octet(s) ' f'exceeds the original length of {data.original_len}; kept as captured', ProtocolWarning, stacklevel=stacklevel()) @@ -3385,7 +3346,7 @@ def _read_option_if_eui(self, schema: 'Schema_IF_EUIAddrOption', *, option = Data_IF_EUIAddrOption( type=schema.type, length=schema.length, - interface=self._read_eui_addr(schema.interface), + interface=self._read_mac_addr(schema.interface), ) return option @@ -4506,7 +4467,7 @@ def _make_block_unknown(self, block: 'Optional[Data_UnknownBlock]' = None, *, # NOTE: The body is padded to 32 bits, and Block Total Length counts it # together with the 12 octets of block type and both length fields. - data += bytes(-len(data) % 4) + data += bytes(pad_length(len(data), 4)) length = len(data) + GENERIC_BLOCK_MINIMUM return Schema_UnknownBlock( @@ -4530,7 +4491,8 @@ def _make_block_shb(self, block: 'Optional[Data_SectionHeaderBlock]' = None, *, version: Version information. major_version: Major version number. minor_version: Minor version number. - section_length: Section length. + section_length: Section length, ``-1`` (or ``0xFFFFFFFFFFFFFFFF``) + for "not specified". options: Block options. byteorder: Byte order of the section; a parsed block keeps its own. **kwargs: Arbitrary keyword arguments. @@ -4538,6 +4500,22 @@ def _make_block_shb(self, block: 'Optional[Data_SectionHeaderBlock]' = None, *, Returns: Constructed block schema. + Raises: + ProtocolError: If ``section_length`` is not an :obj:`int` (a + :obj:`bool` included), or is outside ``-2 ** 63`` to + ``2 ** 64 - 1``. + + .. versionchanged:: 1.6.0 + + ``section_length`` is checked by + :meth:`~pcapkit.protocols.protocol.Protocol._make_sint`, so a value + outside the 64-bit field in either spelling, a :obj:`bool` or a + non-integer raises + :exc:`~pcapkit.utilities.exceptions.ProtocolError` rather than being + written masked or failing to pack; ``-1`` and + ``0xFFFFFFFFFFFFFFFF`` still write the same octets + (:issue:`2614`). + """ if block is not None: byteorder = block.byteorder @@ -4555,6 +4533,10 @@ def _make_block_shb(self, block: 'Optional[Data_SectionHeaderBlock]' = None, *, major_version = version[0] if minor_version is None: minor_version = version[1] + # NOTE: the field is signed and the specification spells "not specified" + # both -1 and 0xFFFFFFFFFFFFFFFF, so both spellings are accepted and the + # field packs them to the same octets; any other value is refused (#2617). + self._make_sint(section_length, bits=64, name='[SHB] section_length', alias='PCAP-NG', allow_unsigned=True) if options is not None: options_value, total_length = self._make_pcapng_options(options, namespace='opt') @@ -4597,8 +4579,9 @@ def _make_block_idb(self, block: 'Optional[Data_InterfaceDescriptionBlock]' = No Constructed block schema. Raises: - ProtocolError: If ``snaplen`` is outside ``0`` to ``0xFFFFFFFF``, - which the 32-bit SnapLen field would truncate (:issue:`2278`). + ProtocolError: If ``snaplen`` is not an :obj:`int` (a :obj:`bool` + included), or is outside ``0`` to ``0xFFFFFFFF``, which the + 32-bit SnapLen field would truncate (:issue:`2278`). .. versionchanged:: 1.5.0 @@ -4607,6 +4590,14 @@ def _make_block_idb(self, block: 'Optional[Data_InterfaceDescriptionBlock]' = No and a value outside the field's range now raises instead of being truncated (:issue:`2278`). + .. versionchanged:: 1.6.0 + + ``snaplen`` is checked by + :meth:`~pcapkit.protocols.protocol.Protocol._make_uint`, so a + :obj:`bool` or a non-integer raises + :exc:`~pcapkit.utilities.exceptions.ProtocolError` too + (:issue:`2614`). + """ if block is not None and hasattr(block, 'block_raw'): return self._make_block_short(block) # type: ignore[return-value] @@ -4617,8 +4608,7 @@ def _make_block_idb(self, block: 'Optional[Data_InterfaceDescriptionBlock]' = No else: linktype_val = self._make_index(linktype, linktype_default, namespace=linktype_namespace, # type: ignore[assignment] reversed=linktype_reversed, pack=False) - if not 0 <= snaplen <= 0xFFFF_FFFF: - raise ProtocolError(f'PCAP-NG: [IDB] snaplen must be in range 0 to 0xFFFFFFFF: {snaplen!r}') + snaplen = self._make_uint(snaplen, bits=32, name='[IDB] snaplen', alias='PCAP-NG') if options is not None: options_value, total_length = self._make_pcapng_options(options, namespace='if') @@ -4683,7 +4673,7 @@ def _make_block_epb(self, block: 'Optional[Data_EnhancedPacketBlock]' = None, *, # to it rather than written out whole behind a clipped length. if len(packet_data) > captured_len: packet_data = bytes(packet_data)[:captured_len] - packet_len = math.ceil(len(packet_data) / 4) * 4 + packet_len = len(packet_data) + pad_length(len(packet_data), 4) if options is not None: options_value, total_length = self._make_pcapng_options(options, namespace='epb') @@ -4738,7 +4728,7 @@ def _make_block_spb(self, block: 'Optional[Data_SimplePacketBlock]' = None, *, captured_len = min(original_len, snaplen) if len(packet_data) > captured_len: packet_data = bytes(packet_data)[:captured_len] - packet_len = math.ceil(len(packet_data) / 4) * 4 + packet_len = len(packet_data) + pad_length(len(packet_data), 4) return self._make_padding(Schema_SimplePacketBlock( length=packet_len + BLOCK_MINIMUM[Enum_BlockType.Simple_Packet_Block], @@ -4856,7 +4846,7 @@ def _make_block_systemd(self, block: 'Optional[Data_SystemdJournalExportBlock]' data = Schema_SystemdJournalExportBlock.dump_entries(entries) # NOTE: The entry MUST be zero-padded to 32 bits, and the Block Total # Length counts that padding, as the ``entry`` field reads it back. - data += bytes(-len(data) % 4) + data += bytes(pad_length(len(data), 4)) return Schema_SystemdJournalExportBlock( length=len(data) + GENERIC_BLOCK_MINIMUM, @@ -4946,7 +4936,7 @@ def _make_block_dsb(self, block: 'Optional[Data_DecryptionSecretsBlock]' = None, }})) else: secrets_held = len(secrets_data_val) - secrets_padded = secrets_held + -secrets_held % 4 + secrets_padded = secrets_held + pad_length(secrets_held, 4) # NOTE: A Secrets Length running past the block is kept as declared, as # an EPB keeps its captured length, so the block rebuilds as captured; # the block length is still sized from the secrets written (#1980). @@ -5012,15 +5002,11 @@ def _make_block_cb(self, block: 'Optional[Data_CustomBlock]' = None, *, else: options_value, _ = [], 0 - # NOTE: ``-n % 4`` is the octet count needed to reach the next 32-bit - # boundary, and zero when already there. Integer arithmetic throughout, - # unlike the ``math.ceil(n / 4) * 4`` idiom used elsewhere in this - # file, which routes a length through a float. cb_data = data.pack() if isinstance(data, Schema) else data - cb_data += bytes(-len(cb_data) % 4) + cb_data += bytes(pad_length(len(cb_data), 4)) for option in options_value: cb_data += option.pack() if isinstance(option, Schema) else option - cb_data += bytes(-len(cb_data) % 4) + cb_data += bytes(pad_length(len(cb_data), 4)) # NOTE: both custom block types span the same fixed fields, and this # method makes either, so the one entry stands for both. @@ -5046,7 +5032,7 @@ def _make_block_packet(self, block: 'Optional[Data_PacketBlock]' = None, *, Args: block: Block data model. interface_id: Interface ID. - drop_count: Drops count. + drop_count: Drops count, ``0xFFFF`` for "not available". timestamp: Packet timestamp. captured_len: Captured length. original_len: Original length. @@ -5057,6 +5043,18 @@ def _make_block_packet(self, block: 'Optional[Data_PacketBlock]' = None, *, Returns: Constructed block schema. + Raises: + ProtocolError: If ``drop_count`` is not an :obj:`int` (a + :obj:`bool` included), or is outside ``0`` to ``0xFFFF``. + + .. versionchanged:: 1.6.0 + + ``drop_count`` is checked by + :meth:`~pcapkit.protocols.protocol.Protocol._make_uint`, so a value + the 16-bit field cannot hold, ``-1`` included, or a :obj:`bool` + raises :exc:`~pcapkit.utilities.exceptions.ProtocolError` rather + than being written masked (:issue:`2614`). + """ warn('PCAP-NG: Packet Block has been obsolete! Please use Enhanced Packet Block and/or ' 'Simple Packet Block instead.', DeprecatedFormatWarning, stacklevel=stacklevel()) @@ -5071,6 +5069,7 @@ def _make_block_packet(self, block: 'Optional[Data_PacketBlock]' = None, *, original_len = block.original_len packet_data = self._make_packet_data(block) options = self._block_area(block, 'options') + drop_count = self._make_uint(drop_count, bits=16, name='[PB] drop_count', alias='PCAP-NG') timestamp_high, timestamp_low = self._make_timestamp(timestamp, interface_id=interface_id) if captured_len is None: @@ -5087,7 +5086,7 @@ def _make_block_packet(self, block: 'Optional[Data_PacketBlock]' = None, *, # to it rather than written out whole behind a clipped length. if len(packet_data) > captured_len: packet_data = bytes(packet_data)[:captured_len] - packet_len = math.ceil(len(packet_data) / 4) * 4 + packet_len = len(packet_data) + pad_length(len(packet_data), 4) if options is not None: options_value, total_length = self._make_pcapng_options(options, namespace='pack') @@ -5477,7 +5476,7 @@ def _make_option_if_eui(self, type: 'Enum_OptionType', option: 'Optional[Data_IF return Schema_IF_EUIAddrOption( type=type, length=8, - interface=self._make_eui_addr(interface), + interface=self._make_mac_addr(interface, size=8, name='EUI address'), ) def _make_option_if_speed(self, type: 'Enum_OptionType', option: 'Optional[Data_IF_SpeedOption]' = None, *, @@ -5573,6 +5572,16 @@ def _make_option_if_tzone(self, type: 'Enum_OptionType', option: 'Optional[Data_ Returns: Constructed option schema. + Raises: + ProtocolError: If ``tzone`` is none of those types, or is a + :obj:`bool`, or its seconds do not fit the signed 32-bit field. + + .. versionchanged:: 1.6.0 + + An offset the signed 32-bit field cannot hold, or a :obj:`bool`, + raises :exc:`~pcapkit.utilities.exceptions.ProtocolError` rather + than being written truncated (:issue:`2614`). + """ if self._type != Enum_BlockType.Interface_Description_Block: raise ProtocolError(f'PCAP-NG: [if_tzone] option must be in Interface Description Block, ' @@ -5593,6 +5602,9 @@ def _make_option_if_tzone(self, type: 'Enum_OptionType', option: 'Optional[Data_ else: raise ProtocolError(f'PCAP-NG: [if_tzone] option timezone must be int, timedelta or timezone, ' f'not {builtins.type(tzone).__name__}') + # NOTE: checked rather than masked, see ``IF_TZoneOption.tzone``; the + # signed value is written, as the field reads it back. + self._make_sint(tzone_val, bits=32, name='[if_tzone] tzone', alias='PCAP-NG') return Schema_IF_TZoneOption( type=type, @@ -5721,6 +5733,17 @@ def _make_option_if_tsoffset(self, type: 'Enum_OptionType', option: 'Optional[Da Returns: Constructed option schema. + Raises: + ProtocolError: If ``offset`` is not an :obj:`int`, is a + :obj:`bool`, or does not fit the signed 64-bit field. + + .. versionchanged:: 1.6.0 + + An offset the signed 64-bit field cannot hold, a :obj:`bool` or a + non-integer raises :exc:`~pcapkit.utilities.exceptions.ProtocolError` + rather than being written truncated, written as zero (``None``) or + failing to pack (:issue:`2614`). + """ if self._type != Enum_BlockType.Interface_Description_Block: raise ProtocolError(f'PCAP-NG: [if_tsoffset] option must be in Interface Description Block, ' @@ -5731,6 +5754,8 @@ def _make_option_if_tsoffset(self, type: 'Enum_OptionType', option: 'Optional[Da if option is not None: offset = option.offset + # NOTE: checked rather than masked, see ``IF_TSOffsetOption.tsoffset``. + self._make_sint(offset, bits=64, name='[if_tsoffset] offset', alias='PCAP-NG') return Schema_IF_TSOffsetOption( type=type, diff --git a/pcapkit/protocols/protocol.py b/pcapkit/protocols/protocol.py index be41e3ace6..370ad5512e 100644 --- a/pcapkit/protocols/protocol.py +++ b/pcapkit/protocols/protocol.py @@ -2211,7 +2211,8 @@ def _make_uint(self, value: 'Any', *, bits: 'int', name: 'str', alias: 'Optional raise ProtocolError(f'{self.alias if alias is None else alias}: invalid {name}: {value!r}') return int(value) - def _make_sint(self, value: 'Any', *, bits: 'int', name: 'str') -> 'int': + def _make_sint(self, value: 'Any', *, bits: 'int', name: 'str', alias: 'Optional[str]' = None, + allow_unsigned: 'bool' = False) -> 'int': """Check that ``value`` is a signed integer that fits in ``bits`` bits. The inverse of :meth:`_read_sint`. A :obj:`bool` is refused whatever @@ -2221,6 +2222,13 @@ def _make_sint(self, value: 'Any', *, bits: 'int', name: 'str') -> 'int': value: the value given for the field bits: width of the field, in bits name: name of the field, for the error message + alias: prefix of the error message, for a caller whose + :attr:`alias` raises before ``read()`` sets it or is not the + prefix its messages carry; :attr:`alias` by default + allow_unsigned: also accept the unsigned spelling of the field's + ``bits`` bits, i.e. up to ``2 ** bits - 1``, for a field whose + specification spells one value both ways, as a sentinel + written ``-1`` and ``0xFF...F``; both denote the same bits Returns: The unsigned integer whose ``bits`` bits denote ``value`` in @@ -2229,14 +2237,16 @@ def _make_sint(self, value: 'Any', *, bits: 'int', name: 'str') -> 'int': Raises: ProtocolError: If ``value`` is not an :class:`int`, is a :obj:`bool`, or does not fit in ``bits`` bits as two's - complement. The message names :attr:`alias`. + complement (or, with ``allow_unsigned``, as an unsigned + integer either). The message names ``alias``. .. versionadded:: 1.6.0 """ bound = 1 << (bits - 1) - if not isinstance(value, int) or isinstance(value, bool) or not -bound <= value < bound: - raise ProtocolError(f'{self.alias}: invalid {name}: {value!r}') + limit = 2 * bound if allow_unsigned else bound + if not isinstance(value, int) or isinstance(value, bool) or not -bound <= value < limit: + raise ProtocolError(f'{self.alias if alias is None else alias}: invalid {name}: {value!r}') return value & ((1 << bits) - 1) @staticmethod diff --git a/pcapkit/protocols/schema/misc/pcapng.py b/pcapkit/protocols/schema/misc/pcapng.py index 3ffef2a34b..e5fe5139f1 100644 --- a/pcapkit/protocols/schema/misc/pcapng.py +++ b/pcapkit/protocols/schema/misc/pcapng.py @@ -1294,6 +1294,7 @@ class SectionHeaderBlock(BlockType, code=Enum_BlockType.Section_Header_Block): major: 'int' = UInt16Field(callback=shb_byteorder_callback, default=1) #: Minor version number. minor: 'int' = UInt16Field(callback=shb_byteorder_callback, default=0) + # NOTE: -1, "not specified", is spelled 0xFFFFFFFFFFFFFFFF too: ``_make_sint`` accepts both, refuses the rest (#2617). #: Section length. section_length: 'int' = Int64Field(callback=shb_byteorder_callback, default=0xFFFF_FFFF_FFFF_FFFF) #: Options. @@ -1520,6 +1521,8 @@ def __init__(self, type: 'Enum_OptionType', length: 'int', tsresol: 'ResolutionD class IF_TZoneOption(_IF_Option, code=Enum_OptionType.if_tzone): """Header schema for PCAP-NG file ``if_tzone`` options.""" + # NOTE: range-checked by ``_make_sint`` rather than masked, pcapkit's choice so it round-trips + # (#2617). The spec fixes only its length; it is signed as this field reads it (#1340). #: Interface time zone (as in seconds difference from GMT). tzone: 'int' = Int32Field(callback=byteorder_callback) #: Padding. @@ -1574,6 +1577,7 @@ def __init__(self, type: 'Enum_OptionType', length: 'int', fcslen: 'int') -> 'No class IF_TSOffsetOption(_IF_Option, code=Enum_OptionType.if_tsoffset): """Header schema for PCAP-NG file ``if_tsoffset`` options.""" + # NOTE: range-checked by ``_make_sint`` rather than masked: a signed 64-bit count, no sentinel. #: Timestamp offset (in seconds). tsoffset: 'int' = Int64Field(callback=byteorder_callback) #: Padding. @@ -1632,6 +1636,7 @@ class InterfaceDescriptionBlock(BlockType, code=Enum_BlockType.Interface_Descrip linktype: 'Enum_LinkType' = EnumField(length=2, namespace=Enum_LinkType, callback=byteorder_callback) #: Reserved. reserved: 'bytes' = PaddingField(length=2) + # NOTE: range-checked by ``_make_uint`` rather than masked: its sentinel, 0 for no limit, is in range. #: Snap length. snaplen: 'int' = UInt32Field(default=0, callback=byteorder_callback) #: Options. @@ -2762,6 +2767,7 @@ class PacketBlock(BlockType, code=Enum_BlockType.Packet_Block): length: 'int' = UInt32Field(callback=byteorder_callback) #: Interface ID. interface_id: 'int' = UInt16Field(callback=byteorder_callback) + # NOTE: range-checked by ``_make_uint`` rather than masked: its "not available" 0xFFFF is in range, not -1. #: Drops count. drop_count: 'int' = UInt16Field(callback=byteorder_callback, default=0xFFFF) #: Timestamp (high). diff --git a/tests/foundation/engines/test_pcapng_read_ahead_dsb_unit.py b/tests/foundation/engines/test_pcapng_read_ahead_dsb_unit.py new file mode 100644 index 0000000000..94c1f91f61 --- /dev/null +++ b/tests/foundation/engines/test_pcapng_read_ahead_dsb_unit.py @@ -0,0 +1,58 @@ +# -*- coding: utf-8 -*- +"""The PCAP-NG engine reads a Decryption Secrets Block ahead as far as its padded secrets run. + +:meth:`PCAPNG._bound_read_ahead ` +has a non-seekable input read ahead to a block's Total Length, and a Decryption +Secrets Block further when its Secrets Length runs past it: to the block's fixed +fields plus the secrets padded to 32 bits by +:func:`~pcapkit.corekit.fields.strings.pad_length`, the span the block is +parsed over (#1586, #2614). Every value is built here, so the module is in the +unit tier. + +""" +from __future__ import annotations + +import io +import struct +import types +import unittest + +#: Octets of a Decryption Secrets Block's fixed fields: type, both lengths, +#: Secrets Type and Secrets Length. +DSB_MINIMUM = 20 + + +def _dsb(length: 'int', secrets: 'int') -> 'bytes': + """The first 16 octets of a little-endian Decryption Secrets Block.""" + return struct.pack(' 'int': + from pcapkit.corekit.io import SeekableReader # pylint: disable=import-outside-toplevel + from pcapkit.foundation.engines.pcapng import PCAPNG # pylint: disable=import-outside-toplevel + + reader = SeekableReader(io.BytesIO(data), buffer_size=8, stream_closing=False) + self.addCleanup(reader.close) + engine = PCAPNG.__new__(PCAPNG) + engine._extractor = types.SimpleNamespace(_ifile=reader) # type: ignore[assignment] # pylint: disable=protected-access + engine._ctx = types.SimpleNamespace(section=types.SimpleNamespace(byteorder='little')) # type: ignore[assignment] # pylint: disable=protected-access + engine._peek = lambda size: data[:size] # type: ignore[method-assign] # pylint: disable=protected-access + engine._bound_read_ahead() # pylint: disable=protected-access + return reader._read_ahead_size # pylint: disable=protected-access + + def test_secrets_within_the_block_keep_its_length(self) -> 'None': + for secrets in (0, 1, 8, 12): + with self.subTest(secrets=secrets): + self.assertEqual(self._read_ahead(_dsb(32, secrets)), 32) + + def test_secrets_past_the_block_read_to_their_padding(self) -> 'None': + for secrets, padded in ((13, 16), (14, 16), (15, 16), (16, 16), (17, 20), (100, 100)): + with self.subTest(secrets=secrets): + self.assertEqual(self._read_ahead(_dsb(32, secrets)), DSB_MINIMUM + padded) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/link/test_arp_mac_helpers_2614_unit.py b/tests/protocols/link/test_arp_mac_helpers_2614_unit.py new file mode 100644 index 0000000000..7a32cf171b --- /dev/null +++ b/tests/protocols/link/test_arp_mac_helpers_2614_unit.py @@ -0,0 +1,105 @@ +# -*- coding: utf-8 -*- +"""GitHub issue #2614: ARP's Ethernet hardware address goes through the base MAC pair. + +:meth:`ARP._read_addr_resolve ` +and :meth:`ARP._make_addr_resolve ` +read and make an Ethernet address with +:meth:`~pcapkit.protocols.protocol.Protocol._read_mac_addr` and +:meth:`~pcapkit.protocols.protocol.Protocol._make_mac_addr`, of any number of +octets since ``hlen`` is not fixed at 6, rather than with a pattern of the +module's own. The messages keep the class name as their prefix, which +:attr:`~pcapkit.protocols.protocol.Protocol.alias` cannot give before +:meth:`~pcapkit.protocols.link.arp.ARP.read` has run. + +Every case builds its own octets in memory and reads no capture. + +The classes are imported inside each test, after +:func:`~tests._support.reimport_once_per_class`. + +""" + +import importlib +import unittest +from unittest import mock + +from tests._support import reimport_once_per_class + +#: ``(module, class)`` for every ARP-family class. +CLASSES = ( + ('pcapkit.protocols.link.arp', 'ARP'), + ('pcapkit.protocols.link.inarp', 'InARP'), + ('pcapkit.protocols.application.rarp', 'RARP'), + ('pcapkit.protocols.application.drarp', 'DRARP'), +) + + +class TestARPMacHelpers(unittest.TestCase): + """Pin the Ethernet branch of ARP's hardware address helpers.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_the_module_keeps_no_address_pattern(self) -> None: + import pcapkit.protocols.link.arp as module + from pcapkit.protocols.link.arp import ARP + from pcapkit.protocols.protocol import ProtocolBase + + self.assertFalse(hasattr(module, 'PAT_MAC_ADDR')) + self.assertIs(ARP._read_mac_addr, ProtocolBase._read_mac_addr) + self.assertIs(ARP._make_mac_addr, ProtocolBase._make_mac_addr) + + def test_the_ethernet_branch_calls_the_base_pair(self) -> None: + from pcapkit.const.arp.hardware import Hardware + from pcapkit.protocols.link.arp import ARP + from pcapkit.protocols.protocol import ProtocolBase + + arp = object.__new__(ARP) + with mock.patch.object(ProtocolBase, '_make_mac_addr', autospec=True, return_value=b'made') as make, \ + mock.patch.object(ProtocolBase, '_read_mac_addr', autospec=True, return_value='read') as read: + self.assertEqual(arp._make_addr_resolve('01:02', Hardware.Ethernet), b'made') + self.assertEqual(arp._read_addr_resolve(b'\x01\x02', Hardware.Ethernet), 'read') + # another hardware type keeps ARP's own hex form + self.assertEqual(arp._make_addr_resolve('0102', Hardware.Frame_Relay), b'\x01\x02') + self.assertEqual(arp._read_addr_resolve(b'\x01\x02', Hardware.Frame_Relay), '0102') + make.assert_called_once_with(arp, '01:02', size=None, alias='ARP') + read.assert_called_once_with(arp, b'\x01\x02') + + def test_any_number_of_octets_packs_and_reads_back(self) -> None: + from pcapkit.const.arp.hardware import Hardware + from pcapkit.protocols.link.arp import ARP + + arp = object.__new__(ARP) + for addr, octets in (('', b''), (b'', b''), ('ab', b'\xab'), ('01-02-03', b'\x01\x02\x03'), + (bytearray(b'AA:bb:CC:dd:EE:ff'), bytes.fromhex('aabbccddeeff')), + ('01:02:03:04:05:06:07:08', bytes.fromhex('0102030405060708'))): + with self.subTest(addr=addr): + self.assertEqual(arp._make_addr_resolve(addr, Hardware.Ethernet), octets) + self.assertEqual(arp._read_addr_resolve(octets, Hardware.Ethernet), octets.hex(':')) + + def test_the_messages_name_the_class(self) -> None: + from pcapkit.const.arp.hardware import Hardware + from pcapkit.utilities.exceptions import ProtocolError + + for module, name in CLASSES: + proto = object.__new__(getattr(importlib.import_module(module), name)) + for addr, message in (('01:2', "invalid MAC address: '01:2'"), + (':', "invalid MAC address: ':'"), + ('01:02:', "invalid MAC address: '01:02:'"), + (None, 'invalid MAC address: must be a str, bytes or bytearray, not None'), + (True, 'invalid MAC address: must be a str, bytes or bytearray, not True')): + with self.subTest(cls=name, addr=addr): + with self.assertRaises(ProtocolError) as context: + proto._make_addr_resolve(addr, Hardware.Ethernet) + self.assertEqual(str(context.exception), f'{name}: {message}') + # another hardware type names a hardware address instead + for addr, message in (('zz', "invalid hardware address: 'zz'"), + (None, 'invalid hardware address: must be a str, bytes or bytearray, ' + 'not None')): + with self.subTest(cls=name, addr=addr, htype='Frame_Relay'): + with self.assertRaises(ProtocolError) as context: + proto._make_addr_resolve(addr, Hardware.Frame_Relay) + self.assertEqual(str(context.exception), f'{name}: {message}') + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/misc/test_pcapng_addr_type_2222_unit.py b/tests/protocols/misc/test_pcapng_addr_type_2222_unit.py index f74f414b60..b7f01fcc4a 100644 --- a/tests/protocols/misc/test_pcapng_addr_type_2222_unit.py +++ b/tests/protocols/misc/test_pcapng_addr_type_2222_unit.py @@ -1,14 +1,13 @@ # -*- coding: utf-8 -*- """GitHub issue #2222: PCAP-NG rejects a non-string MAC or EUI address. -:meth:`PCAPNG._make_mac_addr -` and -:meth:`PCAPNG._make_eui_addr -` passed a :obj:`bool`, -:class:`int`, :obj:`None` or :class:`float` straight to the address pattern, -which raised a bare :class:`TypeError`. Anything not a :class:`str`, -:class:`bytes` or :class:`bytearray`, a :obj:`bool` included, raises -:class:`~pcapkit.utilities.exceptions.ProtocolError`. +PCAP-NG's MAC and EUI address makers passed a :obj:`bool`, :class:`int`, +:obj:`None` or :class:`float` straight to the address pattern, which raised a +bare :class:`TypeError`. Anything not a :class:`str`, :class:`bytes` or +:class:`bytearray`, a :obj:`bool` included, raises +:class:`~pcapkit.utilities.exceptions.ProtocolError`. Both addresses go through +:meth:`PCAPNG._make_mac_addr `, +the EUI one with ``size=8`` (#2614). Each case reaches the maker through the ``if_MACaddr`` or ``if_EUIaddr`` option of an Interface Description Block. diff --git a/tests/protocols/misc/test_pcapng_unit.py b/tests/protocols/misc/test_pcapng_unit.py index 5abe2b819f..c801612fa5 100644 --- a/tests/protocols/misc/test_pcapng_unit.py +++ b/tests/protocols/misc/test_pcapng_unit.py @@ -451,16 +451,16 @@ def test_pcapng_context_timestamp_and_address_helpers(self) -> None: self.assertEqual(pcapng._make_timestamp(ts_decimal), (0, 2000)) self.assertEqual(pcapng._read_mac_addr(b'\x00\x01\x02\x03\x04\x05'), '00:01:02:03:04:05') - self.assertEqual(pcapng._read_eui_addr(bytes.fromhex('023456fffe789abc')), + self.assertEqual(pcapng._read_mac_addr(bytes.fromhex('023456fffe789abc')), '02:34:56:ff:fe:78:9a:bc') self.assertEqual(pcapng._make_mac_addr('00:01:02:03:04:05'), b'\x00\x01\x02\x03\x04\x05') self.assertEqual(pcapng._make_mac_addr(b'00-01-02-03-04-05'), b'\x00\x01\x02\x03\x04\x05') - self.assertEqual(pcapng._make_eui_addr('02:34:56:ff:fe:78:9a:bc'), + self.assertEqual(pcapng._make_mac_addr('02:34:56:ff:fe:78:9a:bc', size=8, name='EUI address'), bytes.fromhex('023456fffe789abc')) with self.assertRaises(ProtocolError): pcapng._make_mac_addr('bad') with self.assertRaises(ProtocolError): - pcapng._make_eui_addr('bad') + pcapng._make_mac_addr('bad', size=8, name='EUI address') def test_pcapng_top_level_properties_registry_and_dispatch_paths(self) -> None: from pcapkit.const.pcapng.block_type import BlockType @@ -582,7 +582,7 @@ def now(cls, tz): self.assertEqual(pcapng._make_timestamp(instant, interface_id=0), (0, 4_000_000)) self.assertEqual(pcapng._read_mac_addr(b'\x00\x01\x02\x03\x04\x05'), '00:01:02:03:04:05') - self.assertEqual(pcapng._read_eui_addr(bytes.fromhex('023456fffe789abc')), + self.assertEqual(pcapng._read_mac_addr(bytes.fromhex('023456fffe789abc')), '02:34:56:ff:fe:78:9a:bc') proto_map = PCAPNG.__dict__['__proto__'] diff --git a/tests/protocols/misc/test_pcapng_value_helpers_2614_unit.py b/tests/protocols/misc/test_pcapng_value_helpers_2614_unit.py new file mode 100644 index 0000000000..4784ac79e0 --- /dev/null +++ b/tests/protocols/misc/test_pcapng_value_helpers_2614_unit.py @@ -0,0 +1,347 @@ +# -*- coding: utf-8 -*- +"""GitHub issue #2614: PCAP-NG's signed, sentinel and address fields on the base helpers. + +The signed and sentinel fields were reviewed one by one against the PCAP-NG +specification (draft-ietf-opsawg-pcapng): + +* ``if_tsoffset`` is a signed 64-bit integer with no sentinel, so a value its + 64 bits cannot hold raises through + :meth:`~pcapkit.protocols.protocol.Protocol._make_sint` rather than being + written truncated, which read back as another value; +* ``if_tzone`` gets the same check over 32 bits, by pcapkit's choice so that + it round-trips (#2617): the specification fixes only its length, and the + value is signed because the schema's ``Int32Field`` reads it so (#1340); +* the IDB's SnapLen is unsigned, and its sentinel -- 0, no limit -- is in range, + so it is checked by :meth:`~pcapkit.protocols.protocol.Protocol._make_uint`; +* the obsolete Packet Block's Drops Count is a 16-bit counter the schema's + ``UInt16Field`` reads unsigned, and its "not available" sentinel, ``xFFFF`` in + hexadecimal (Appendix A, Drops Count), is in range -- like SnapLen's -- so it is + checked by :meth:`~pcapkit.protocols.protocol.Protocol._make_uint` over 16 bits. + The specification never spells it -1 and the reader returns 65535, so there is + no spelling for masking to keep, and a ``-1``, a value past ``0xFFFF`` or a + :obj:`bool` is refused rather than written as another count; +* the SHB's Section Length is signed, and the specification spells its "not + specified" -1 as ``0xFFFFFFFFFFFFFFFF`` as well, the reader returning -1. So + :meth:`~pcapkit.protocols.protocol.Protocol._make_sint` takes ``-2 ** 63`` to + ``2 ** 64 - 1``, both spellings of the sentinel write the same octets, and the + field's own packing maps the unsigned spelling onto the signed one -- the only + masking left. A value past either end, a :obj:`bool` or a non-integer is + refused (#2617). That is what sets it apart from Drops Count above, whose + sentinel has no signed spelling. + +``if_tsoffset`` and ``if_tzone`` are checked over the signed range alone, which +already refuses a :obj:`bool` and a value past either end: neither has a +sentinel, so no second spelling is needed, and one would read back as another +value (#2617). + +The EUI-64 address of ``if_EUIaddr`` goes through the base MAC address pair with +``size=8``, so the module keeps no address pattern of its own. + +The padding sites on :func:`~pcapkit.corekit.fields.strings.pad_length` that no +other test reaches are pinned too: a custom block's data before its options, its +option area, and a Simple Packet Block whose data ends exactly at the padding of +its original length. + +Every case builds its own octets in memory and reads no capture. + +:mod:`pcapkit` is imported inside each test, after +:func:`~tests._support.reimport_once_per_class`. + +""" + +import datetime +import struct +import sys +import unittest +import warnings +from typing import TYPE_CHECKING + +from tests.protocols.misc.test_pcapng_roundtrip2_unit import PCAPNGTestCase + +if TYPE_CHECKING: + from typing import Any + +#: Test tier (#1958): exercises the integer bounds of signed and unsigned fields, +#: which are ``int`` and :mod:`struct` edge cases. +__tier__ = 'ALL' + + +class TestPCAPNGValueHelpers(PCAPNGTestCase): + """Pin how the reviewed PCAP-NG fields check, pack and parse back.""" + + def _idb(self, snaplen: 'Any' = 0x40000, **options: 'Any') -> 'Any': + """Make an Ethernet Interface Description Block with the given options.""" + from pcapkit.const.pcapng.block_type import BlockType + from pcapkit.const.pcapng.option_type import OptionType + from pcapkit.const.reg.linktype import LinkType + from pcapkit.protocols.misc.pcapng import PCAPNG + + return PCAPNG(num=2, sct=1, ctx=self._context(), type=BlockType.Interface_Description_Block, + block={'linktype': LinkType.ETHERNET, 'snaplen': snaplen, + 'options': [(OptionType[code], args) for code, args in options.items()]}) + + def _reparse(self, made: 'Any') -> 'Any': + """Parse the octets of a made block back into its data model.""" + return self._parse(made.data).info + + def assertRefused(self, message: 'str', **kwargs: 'Any') -> None: # pylint: disable=invalid-name + """Making an IDB from ``kwargs`` raises :exc:`ProtocolError` with exactly ``message``.""" + from pcapkit.utilities.exceptions import ProtocolError + + with self.assertRaises(ProtocolError) as context: + self._idb(**kwargs) + self.assertEqual(str(context.exception), message) + + def test_tzone_outside_the_signed_field_is_refused(self) -> None: + for tzone, shown in ((2 ** 31, 2 ** 31), (-2 ** 31 - 1, -2 ** 31 - 1), (2 ** 40, 2 ** 40), + (datetime.timedelta(seconds=2 ** 31), 2 ** 31), (True, True)): + with self.subTest(tzone=tzone): + self.assertRefused(f'PCAP-NG: invalid [if_tzone] tzone: {shown!r}', if_tzone={'tzone': tzone}) + + def test_tzone_bounds_read_back_unchanged(self) -> None: + from pcapkit.const.pcapng.option_type import OptionType + + for seconds in (-2 ** 31, -86400, -1, 0, 3600, 2 ** 31 - 1): + with self.subTest(tzone=seconds): + # an offset of a day or more warns as it is read, and is kept (#1340) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + info = self._reparse(self._idb(if_tzone={'tzone': seconds})) + zone = info.options[OptionType.if_tzone].timezone + offset = zone if isinstance(zone, datetime.timedelta) else zone.utcoffset(None) + self.assertEqual(offset, datetime.timedelta(seconds=seconds)) + + def test_tsoffset_outside_the_signed_field_is_refused(self) -> None: + # ``None`` is refused rather than written as zero, a non-integer rather than failing to pack + for offset in (2 ** 63, -2 ** 63 - 1, True, 1.5, None, '1'): + with self.subTest(offset=offset): + self.assertRefused(f'PCAP-NG: invalid [if_tsoffset] offset: {offset!r}', + if_tsoffset={'offset': offset}) + + def test_tsoffset_bounds_read_back_unchanged(self) -> None: + from pcapkit.const.pcapng.option_type import OptionType + + for offset in (-2 ** 63, -1, 0, 1, 2 ** 63 - 1): + with self.subTest(offset=offset): + info = self._reparse(self._idb(if_tsoffset={'offset': offset})) + self.assertEqual(info.options[OptionType.if_tsoffset].offset, offset) + + def test_snaplen_is_checked_by_make_uint(self) -> None: + for snaplen in (-1, 2 ** 32, True, 1.5, '1'): + with self.subTest(snaplen=snaplen): + self.assertRefused(f'PCAP-NG: invalid [IDB] snaplen: {snaplen!r}', snaplen=snaplen) + + def test_snaplen_bounds_read_back_unchanged(self) -> None: + for snaplen in (0, 1, 0xFFFF_FFFF): + with self.subTest(snaplen=snaplen): + self.assertEqual(self._reparse(self._idb(snaplen=snaplen)).snaplen, snaplen) + + def test_eui_address_is_the_base_mac_pair(self) -> None: + import pcapkit.protocols.misc.pcapng as module + from pcapkit.const.pcapng.option_type import OptionType + from pcapkit.protocols.misc.pcapng import PCAPNG + from pcapkit.protocols.protocol import ProtocolBase + + self.assertFalse(hasattr(module, 'PAT_EUI_ADDR')) + for name in ('_read_eui_addr', '_make_eui_addr'): + with self.subTest(name=name): + self.assertFalse(hasattr(PCAPNG, name)) + self.assertIs(PCAPNG._read_mac_addr, ProtocolBase._read_mac_addr) + + made = self._idb(if_EUIaddr={'interface': '02-34-56-FF-FE-78-9A-BC'}) + self.assertIn(bytes.fromhex('023456fffe789abc'), made.data) + self.assertEqual(self._reparse(made).options[OptionType.if_EUIaddr].interface, + '02:34:56:ff:fe:78:9a:bc') + for interface in ('02:34:56:ff:fe:78:9a', '02:34:56:ff:fe:78:9a:bc:de'): + with self.subTest(interface=interface): + self.assertRefused(f'PCAP-NG: invalid EUI address: {interface!r}', + if_EUIaddr={'interface': interface}) + + +class TestPCAPNGSectionLength(PCAPNGTestCase): + """Pin the SHB's Section Length: signed 64 bits, with the unsigned spelling of the same bits.""" + + #: Section Length is octets 16 to 24 of an SHB: type, length, magic, versions. + FIELD = slice(16, 24) + + def _shb(self, **block: 'Any') -> 'Any': + """Make a little-endian section header block from the given fields.""" + from pcapkit.const.pcapng.block_type import BlockType + from pcapkit.protocols.misc.pcapng import PCAPNG + + return PCAPNG(num=0, sct=1, ctx=None, type=BlockType.Section_Header_Block, + block={'byteorder': 'little', **block}) + + def test_section_length_outside_the_field_is_refused(self) -> None: + """Each was written masked or failed bare: ``2**64`` as 0, ``True`` as 1, ``None`` as the sentinel.""" + from pcapkit.utilities.exceptions import ProtocolError + + for section_length in (2 ** 64, -2 ** 63 - 1, 2 ** 70, True, False, 1.5, None, '1'): + with self.subTest(section_length=section_length): + with self.assertRaises(ProtocolError) as context: + self._shb(section_length=section_length) + self.assertEqual(str(context.exception), + f'PCAP-NG: invalid [SHB] section_length: {section_length!r}') + + def test_section_length_edges_write_the_octets_of_their_bits(self) -> None: + """Either spelling of a 64-bit pattern writes it; the reader returns the signed one.""" + made = {} + for section_length in (-2 ** 63, -1, 0, 1024, 2 ** 63 - 1, 2 ** 63, 2 ** 64 - 1): + with self.subTest(section_length=section_length): + made[section_length] = self._shb(section_length=section_length).data + self.assertEqual(made[section_length][self.FIELD], + (section_length % 2 ** 64).to_bytes(8, 'little')) + signed = section_length - 2 ** 64 if section_length >= 2 ** 63 else section_length + self.assertEqual(self._parse(made[section_length]).info.section_length, signed) + # the two spellings of each pattern write one block, whole + self.assertEqual(made[-1], made[2 ** 64 - 1]) + self.assertEqual(made[-2 ** 63], made[2 ** 63]) + self.assertEqual(made[-1][self.FIELD], b'\xff' * 8) + + def test_section_length_defaults_to_the_sentinel(self) -> None: + """A block made without one is "not specified", whichever spelling is given.""" + default = self._shb().data + self.assertEqual(default[self.FIELD], b'\xff' * 8) + self.assertEqual(default, self._shb(section_length=-1).data) + self.assertEqual(default, self._shb(section_length=0xFFFF_FFFF_FFFF_FFFF).data) + self.assertEqual(self._parse(default).info.section_length, -1) + + def test_section_length_of_a_data_model_is_checked_too(self) -> None: + """A parsed block rebuilds byte-exact; one given a value the field cannot hold is refused.""" + from pcapkit.protocols.misc.pcapng import PCAPNG + from pcapkit.utilities.exceptions import ProtocolError + + for section_length in (-1, 0, 1024): + with self.subTest(section_length=section_length): + self.assertRebuilds(self._shb(section_length=section_length).data) + + info = self._parse(self._shb(section_length=1024).data).info + for section_length in (2 ** 64, -2 ** 63 - 1, True): + with self.subTest(section_length=section_length): + info.__update__([('section_length', section_length)]) + with self.assertRaises(ProtocolError) as raised: + PCAPNG.from_data(info, num=0, sct=1, ctx=None) + self.assertEqual(str(raised.exception), + f'PCAP-NG: invalid [SHB] section_length: {section_length!r}') + + +class TestPCAPNGPacketBlockDropCount(PCAPNGTestCase): + """Pin the Packet Block's Drops Count: unsigned 16 bits, checked and not masked.""" + + def _packet_block(self, context: 'Any' = None, **block: 'Any') -> 'Any': + """Make an obsolete Packet Block from the given fields, whose warning is silenced.""" + from pcapkit.const.pcapng.block_type import BlockType + from pcapkit.protocols.misc.pcapng import PCAPNG + + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + return PCAPNG(num=2, sct=1, ctx=self._context() if context is None else context, + type=BlockType.Packet_Block, + block={'packet_data': b'abcd', 'timestamp': 0, **block}) + + def test_drop_count_outside_the_unsigned_field_is_refused(self) -> None: + """Each was written masked: -1 as 0xFFFF, 65536 as 0, ``True`` as 1, ``None`` raised bare.""" + from pcapkit.utilities.exceptions import ProtocolError + + for drop_count in (-1, 0x1_0000, -0xFFFF, 2 ** 32, True, False, 1.5, None, '1'): + with self.subTest(drop_count=drop_count): + with self.assertRaises(ProtocolError) as context: + self._packet_block(drop_count=drop_count) + self.assertEqual(str(context.exception), f'PCAP-NG: invalid [PB] drop_count: {drop_count!r}') + + def test_drop_count_bounds_read_back_unchanged(self) -> None: + """0 and 1 are counts, ``0xFFFF`` is "not available"; the reader returns 65535 for it, never -1.""" + for drop_count in (0, 1, 0x7FFF, 0x8000, 0xFFFF): + with self.subTest(drop_count=drop_count): + made = self._packet_block(drop_count=drop_count) + # Block Type, Block Total Length, Interface ID, then Drops Count + self.assertEqual(made.data[10:12], drop_count.to_bytes(2, sys.byteorder)) + self.assertEqual(made.info.drop_count, drop_count) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + self.assertEqual(self._parse(made.data).info.drop_count, drop_count) + + def test_drop_count_defaults_to_zero_and_the_sentinel_rebuilds(self) -> None: + """A block made without one counts no drops; one parsed with the sentinel rebuilds with it.""" + context = self._context() + self.assertEqual(self._packet_block(context).info.drop_count, 0) + sentinel = self._packet_block(context, drop_count=0xFFFF) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + self.assertRebuilds(sentinel.data, context) + self.assertEqual(self._parse(sentinel.data, context).info.drop_count, 0xFFFF) + + def test_drop_count_of_a_data_model_is_checked_too(self) -> None: + """The data model path of the maker checks the field as the keyword path does.""" + from pcapkit.protocols.misc.pcapng import PCAPNG + from pcapkit.utilities.exceptions import ProtocolError + + context = self._context() + info = self._packet_block(context, drop_count=3).info + for drop_count in (-1, 0x1_0000, True): + with self.subTest(drop_count=drop_count): + info.__update__([('drop_count', drop_count)]) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + with self.assertRaises(ProtocolError) as raised: + PCAPNG.from_data(info, num=2, sct=1, ctx=context) + self.assertEqual(str(raised.exception), f'PCAP-NG: invalid [PB] drop_count: {drop_count!r}') + + +class TestPCAPNGPadLength(PCAPNGTestCase): + """Pin the padding sites that only a custom block or a clipped SPB reaches.""" + + def _custom(self, type_: 'Any', data: 'bytes', options: 'Any') -> 'Any': + """Make a custom block of ``type_`` with private enterprise number 1.""" + from pcapkit.protocols.misc.pcapng import PCAPNG + + return PCAPNG(num=2, sct=1, ctx=None, type=type_, block={'pen': 1, 'data': data, 'options': options}) + + def _custom_types(self) -> 'Any': + """Both custom block types, which one maker makes.""" + from pcapkit.const.pcapng.block_type import BlockType + + return (BlockType.Custom_Block_that_rewriters_can_copy_into_new_files, + BlockType.Custom_Block_that_rewriters_should_not_copy_into_new_files) + + def test_custom_data_is_padded_before_its_options(self) -> None: + from pcapkit.const.pcapng.option_type import OptionType + + comment = bytes.fromhex('01000200 68690000') # opt_comment 'hi', padded + for type_ in self._custom_types(): + for data in (b'o', b'od', b'odd'): + with self.subTest(type=type_, data=data): + made = self._custom(type_, data, [(OptionType.opt_comment, {'comment': 'hi'})]) + self.assertEqual(made.info.data, data + bytes(4 - len(data)) + comment) + self.assertEqual(made.info.length, 28) + self.assertRebuilds(made.data) + + def test_custom_option_area_is_padded(self) -> None: + """An option area of seven octets is written whole, with one pad octet after it.""" + area = bytes.fromhex('01000300 616263') # opt_comment 'abc', unpadded + for type_ in self._custom_types(): + with self.subTest(type=type_): + made = self._custom(type_, b'', area) + self.assertEqual(made.info.data, area + b'\x00') + self.assertEqual(made.info.length, 24) + self.assertEqual(made.info.length, len(made.data)) + + def test_spb_data_ending_at_its_padding_is_not_past_original_len(self) -> None: + """Eight octets past a snaplen of 2 end at the padding of ``original_len=5``.""" + body = struct.pack('`, :meth:`PCAPNG._make_mac_addr -` and :meth:`PCAPNG._make_eui_addr -` stripped the separators -from a ``aa:bb:cc:dd:ee:ff`` string and returned the remaining *hex text*, so -``make()`` wrote the ASCII digits of the address onto the wire. ARP did the -same for the plain hex string it parses a non-Ethernet hardware address into. +` and PCAP-NG's MAC and EUI +address makers stripped the separators from a ``aa:bb:cc:dd:ee:ff`` string and +returned the remaining *hex text*, so ``make()`` wrote the ASCII digits of the +address onto the wire. ARP did the same for the plain hex string it parses a +non-Ethernet hardware address into. PCAP-NG makes both addresses with +:meth:`PCAPNG._make_mac_addr `, +the EUI one with ``size=8`` (#2614). Every case builds its own octets in memory and reads no capture. @@ -90,7 +91,7 @@ def test_if_euiaddr_packs_eight_octets(self) -> None: schema = self.pcapng._make_option_if_eui(OptionType.if_EUIaddr, interface=addr) self.assertEqual(bytes(schema), bytes.fromhex('0700' '0800' '023456fffe789abc')) unpacked = IF_EUIAddrOption.unpack(bytes(schema)) - self.assertEqual(self.pcapng._read_eui_addr(unpacked.interface), + self.assertEqual(self.pcapng._read_mac_addr(unpacked.interface), '02:34:56:ff:fe:78:9a:bc') diff --git a/tests/protocols/test_protocol_value_helpers_2514_unit.py b/tests/protocols/test_protocol_value_helpers_2514_unit.py index 019d5a5119..430bc0a2d2 100644 --- a/tests/protocols/test_protocol_value_helpers_2514_unit.py +++ b/tests/protocols/test_protocol_value_helpers_2514_unit.py @@ -23,7 +23,8 @@ integer of so many, out of a buffer, refusing a short read; ``_make_octets`` checks the octets a field is given; * ``_read_sint`` and ``_make_sint`` -- a two's-complement field, read from its - unsigned value and checked on the way back; + unsigned value and checked on the way back, the unsigned spelling of the same + bits accepted too when ``allow_unsigned`` asks for it; * ``_uint_size`` -- the octets an unsigned integer needs; * ``_read_mac_addr`` and ``_make_mac_addr`` -- a MAC address, an EUI-64 one, or one of any length; @@ -356,6 +357,41 @@ def test_a_bool_or_a_non_int_is_refused(self) -> None: with self.subTest(value=value): self.assertRefused(f'invalid field: {shown}$', self.proto._make_sint, value, bits=8, name='field') + def test_the_unsigned_spelling_is_accepted_only_when_asked(self) -> None: + """``allow_unsigned`` widens the range to ``-half`` to ``2 ** bits - 1``; both spellings give the same bits.""" + for bits in (2, 8, 24): + half, top = 1 << (bits - 1), 1 << bits + for value in (-half, -1, 0, half - 1, half, top - 1): + with self.subTest(bits=bits, value=value): + self.assertEqual(self.proto._make_sint(value, bits=bits, name='field', allow_unsigned=True), + value % top) + self.assertEqual(self.proto._make_sint(-1, bits=bits, name='field', allow_unsigned=True), + self.proto._make_sint(top - 1, bits=bits, name='field', allow_unsigned=True)) + # without it the unsigned half stays refused, as the signed-only tests above pin at ``half`` + with self.subTest(bits=bits, value=top - 1, allow_unsigned=False): + self.assertRefused(rf'invalid field: {top - 1}$', self.proto._make_sint, top - 1, bits=bits, + name='field') + + def test_one_past_either_end_is_refused_with_the_unsigned_spelling(self) -> None: + for bits in (2, 8, 24): + half, top = 1 << (bits - 1), 1 << bits + for value in (-half - 1, top): + with self.subTest(bits=bits, value=value): + self.assertRefused(rf'invalid field: {value}$', self.proto._make_sint, value, bits=bits, + name='field', allow_unsigned=True) + for value, shown in ((True, 'True'), (False, 'False'), ('1', "'1'"), (1.0, r'1\.0'), (None, 'None')): + with self.subTest(value=value): + self.assertRefused(f'invalid field: {shown}$', self.proto._make_sint, value, bits=8, name='field', + allow_unsigned=True) + + def test_an_alias_given_prefixes_the_message_before_read(self) -> None: + from pcapkit.utilities.exceptions import ProtocolError + + proto = self.unread() + self.assertEqual(proto._make_sint(-1, bits=16, name='offset', alias='PCAP-NG'), 0xFFFF) + with self.assertRaisesRegex(ProtocolError, '^PCAP-NG: invalid offset: 32768$'): + proto._make_sint(0x8000, bits=16, name='offset', alias='PCAP-NG') + class UintSizeTests(_HelperTestCase): """``_uint_size`` counts whole octets, rounding a partial one up."""