diff --git a/docs/source/pcapkit/protocols/internet/ah.rst b/docs/source/pcapkit/protocols/internet/ah.rst index 12be9b7e3a..b734034a49 100644 --- a/docs/source/pcapkit/protocols/internet/ah.rst +++ b/docs/source/pcapkit/protocols/internet/ah.rst @@ -45,6 +45,8 @@ Header Schemas .. module:: pcapkit.protocols.schema.internet.ah +.. autodata:: pcapkit.protocols.schema.internet.ah.AH_HEADER_LEN + .. autoclass:: pcapkit.protocols.schema.internet.ah.AH :members: :show-inheritance: diff --git a/docs/source/pcapkit/protocols/internet/ipv4.rst b/docs/source/pcapkit/protocols/internet/ipv4.rst index b27245d710..87ecfe4650 100644 --- a/docs/source/pcapkit/protocols/internet/ipv4.rst +++ b/docs/source/pcapkit/protocols/internet/ipv4.rst @@ -95,6 +95,8 @@ Header Schemas .. module:: pcapkit.protocols.schema.internet.ipv4 +.. autodata:: pcapkit.protocols.schema.internet.ipv4.IPV4_HEADER_LEN + .. autoclass:: pcapkit.protocols.schema.internet.ipv4.IPv4 :members: :show-inheritance: diff --git a/docs/source/pcapkit/protocols/internet/ipv6.rst b/docs/source/pcapkit/protocols/internet/ipv6.rst index ce7ee42316..e2aedb2c10 100644 --- a/docs/source/pcapkit/protocols/internet/ipv6.rst +++ b/docs/source/pcapkit/protocols/internet/ipv6.rst @@ -44,11 +44,20 @@ Octets Bits Name Description .. automethod:: __index__ +Protocol Constants +------------------ + +.. autodata:: pcapkit.protocols.internet.ipv6.IPV6_HEADER_LEN + +.. autodata:: pcapkit.protocols.internet.ipv6.IPV6_NEXT_HEADER_OFFSET + Header Schemas -------------- .. module:: pcapkit.protocols.schema.internet.ipv6 +.. autodata:: pcapkit.protocols.schema.internet.ipv6.IP_MAX_LENGTH + .. autoclass:: pcapkit.protocols.schema.internet.ipv6.IPv6 :members: :show-inheritance: diff --git a/docs/source/pcapkit/protocols/misc/pcap.rst b/docs/source/pcapkit/protocols/misc/pcap.rst index 5c964daab4..d8f4a4ede0 100644 --- a/docs/source/pcapkit/protocols/misc/pcap.rst +++ b/docs/source/pcapkit/protocols/misc/pcap.rst @@ -61,6 +61,17 @@ Header Schemas .. module:: pcapkit.protocols.schema.misc.pcap.header +.. autodata:: pcapkit.protocols.schema.misc.pcap.header.GLOBAL_HEADER_LEN + +.. autodata:: pcapkit.protocols.schema.misc.pcap.header.MAGIC_NUMBER + :no-value: + +.. autodata:: pcapkit.protocols.schema.misc.pcap.header.RESOLUTION + :no-value: + +.. autodata:: pcapkit.protocols.schema.misc.pcap.header.RESOLUTION_PLACES + :no-value: + .. autoclass:: pcapkit.protocols.schema.misc.pcap.header.Header :members: :show-inheritance: @@ -127,6 +138,8 @@ Header Schemas .. module:: pcapkit.protocols.schema.misc.pcap.frame +.. autodata:: pcapkit.protocols.schema.misc.pcap.frame.RECORD_HEADER_LEN + .. autoclass:: pcapkit.protocols.schema.misc.pcap.frame.Frame :members: :show-inheritance: diff --git a/docs/source/pcapkit/protocols/misc/pcapng.rst b/docs/source/pcapkit/protocols/misc/pcapng.rst index bc76982bc0..af992f6608 100644 --- a/docs/source/pcapkit/protocols/misc/pcapng.rst +++ b/docs/source/pcapkit/protocols/misc/pcapng.rst @@ -213,6 +213,15 @@ Header Schemas .. module:: pcapkit.protocols.schema.misc.pcapng +.. autodata:: pcapkit.protocols.schema.misc.pcapng.BYTE_ORDER_MAGIC + +.. autodata:: pcapkit.protocols.schema.misc.pcapng.GENERIC_BLOCK_MINIMUM + +.. autodata:: pcapkit.protocols.schema.misc.pcapng.BLOCK_MINIMUM + :no-value: + +.. autodata:: pcapkit.protocols.schema.misc.pcapng.DEFAULT_RESOLUTION + .. autoclass:: pcapkit.protocols.schema.misc.pcapng.PCAPNG :members: :show-inheritance: diff --git a/docs/source/pcapkit/protocols/transport/tcp.rst b/docs/source/pcapkit/protocols/transport/tcp.rst index 3106e6b358..2f8645a315 100644 --- a/docs/source/pcapkit/protocols/transport/tcp.rst +++ b/docs/source/pcapkit/protocols/transport/tcp.rst @@ -143,6 +143,10 @@ Header Schemas .. module:: pcapkit.protocols.schema.transport.tcp +.. autodata:: pcapkit.protocols.schema.transport.tcp.TCP_MIN_HEADER_LEN + +.. autodata:: pcapkit.protocols.schema.transport.tcp.TCP_DATA_OFFSET_OCTET + .. autoclass:: pcapkit.protocols.schema.transport.tcp.TCP :members: :show-inheritance: diff --git a/docs/source/pcapkit/toolkit/3rdparty.rst b/docs/source/pcapkit/toolkit/3rdparty.rst index 2057e93252..a8ef78fbad 100644 --- a/docs/source/pcapkit/toolkit/3rdparty.rst +++ b/docs/source/pcapkit/toolkit/3rdparty.rst @@ -227,8 +227,6 @@ Auxiliary Functions Internal Definitions -------------------- -.. autodata:: pcapkit.toolkit.pypcapfile.TCP_MIN_HEADER_LEN - .. autodata:: pcapkit.toolkit.pypcapfile.IPV4_FLAG_DF .. autodata:: pcapkit.toolkit.pypcapfile.IPV4_FLAG_MF diff --git a/examples/generators/endian.py b/examples/generators/endian.py index 4074dac116..1f91be66cb 100644 --- a/examples/generators/endian.py +++ b/examples/generators/endian.py @@ -77,10 +77,11 @@ SAMPLE = ROOT / 'examples' / 'captures' #: PCAP file magic numbers, keyed by endianness and nanosecond-resolution flag. -#: The same table as ``_MAGIC_NUM`` in -#: :file:`pcapkit/protocols/misc/pcap/header.py`, which is the reader these -#: fixtures are written for; it is repeated rather than imported so that a -#: fixture stays generatable without :mod:`pcapkit` importing cleanly. +#: The same values as +#: :data:`pcapkit.protocols.schema.misc.pcap.header.MAGIC_NUMBER`, which the +#: reader these fixtures are written for dispatches on; they are repeated +#: rather than imported so that a fixture stays generatable without +#: :mod:`pcapkit` importing cleanly. MAGIC_NUMBER = { ('big', False): b'\xa1\xb2\xc3\xd4', ('big', True): b'\xa1\xb2\x3c\x4d', diff --git a/pcapkit/dumpkit/pcap.py b/pcapkit/dumpkit/pcap.py index 7645079016..e0e80786e7 100644 --- a/pcapkit/dumpkit/pcap.py +++ b/pcapkit/dumpkit/pcap.py @@ -17,6 +17,7 @@ from pcapkit.dumpkit.common import DumperBase from pcapkit.protocols.data.misc.pcap.header import Header as Data_Header from pcapkit.protocols.misc.pcap.header import Header +from pcapkit.protocols.schema.misc.pcap.header import RESOLUTION if TYPE_CHECKING: from enum import IntEnum as StdlibIntEnum @@ -51,12 +52,6 @@ #: the two spellings writing the same octets for every input. _UINT32_MASK = 0xFFFF_FFFF -#: Units per second of a record's ``ts_usec`` field, keyed by the global header's -#: nanosecond flag. -_RESOLUTION = { - False: 1_000_000, - True: 1_000_000_000, -} def _truncates_to(scaled: 'int', den: 'int', fraction: 'int') -> 'bool': @@ -199,8 +194,8 @@ def _dump_header(self, *, protocol: 'Enum_LinkType | StdlibIntEnum | AenumIntEnu #: int: Units per second of the fraction this file's records carry, and #: of the one the other resolution would, taken from :attr:`self._ghdr #: <_ghdr>` for the same reason as :attr:`self._rechdr <_rechdr>`. - self._tsres = _RESOLUTION[self._ghdr.magic_number.nanosecond] - self._tsalt = _RESOLUTION[not self._ghdr.magic_number.nanosecond] + self._tsres = RESOLUTION[self._ghdr.magic_number.nanosecond] + self._tsalt = RESOLUTION[not self._ghdr.magic_number.nanosecond] def _append_value(self, value: 'Data_Frame', file: 'IO[bytes]', name: 'str') -> 'None': # pylint: disable=unused-argument """Call this function to write contents. diff --git a/pcapkit/foundation/engines/dpkt.py b/pcapkit/foundation/engines/dpkt.py index a13fa1008c..fa6f554d96 100644 --- a/pcapkit/foundation/engines/dpkt.py +++ b/pcapkit/foundation/engines/dpkt.py @@ -18,6 +18,9 @@ from pcapkit.const.pcapng.option_type import OptionType as Enum_OptionType from pcapkit.const.reg.linktype import LinkType as Enum_LinkType from pcapkit.foundation.engines.engine import EngineBase +from pcapkit.protocols.schema.misc.pcap.header import RESOLUTION +from pcapkit.protocols.schema.misc.pcapng import (BYTE_ORDER_MAGIC, DEFAULT_RESOLUTION, + GENERIC_BLOCK_MINIMUM) from pcapkit.utilities.compat import localcontext from pcapkit.utilities.exceptions import FormatError, stacklevel from pcapkit.utilities.logging import get_logger @@ -45,6 +48,14 @@ #: :data:`pcapkit.utilities.logging.logger`. logger = get_logger(__name__) +#: The Block Type of a Section Header Block as the octets read, the same in +#: either byte order, spelled once rather than per block read. +_SHB_TYPE_OCTETS = Enum_BlockType.Section_Header_Block.to_bytes(4, 'big') + +#: :data:`~pcapkit.protocols.schema.misc.pcapng.BYTE_ORDER_MAGIC` as a +#: little-endian section writes it, spelled once rather than per block read. +_LITTLE_BYTE_ORDER_OCTETS = BYTE_ORDER_MAGIC.to_bytes(4, 'little') + class PCAPNGReader: """PCAP-NG reader for the DPKT engine, aware of every interface. @@ -106,14 +117,14 @@ def _read(self) -> 'Iterator[tuple[float, bytes]]': # the SHB block type is a palindrome, so it reads the same in # either byte order; any other block uses the section's order byteorder: 'Literal["big", "little"]' - if head[:4] == b'\x0a\x0d\x0d\x0a': + if head[:4] == _SHB_TYPE_OCTETS: head += self._file.read(4) - byteorder = 'little' if head[8:] == b'\x4d\x3c\x2b\x1a' else 'big' + byteorder = 'little' if head[8:] == _LITTLE_BYTE_ORDER_OCTETS else 'big' else: byteorder = self._ctx.section.byteorder block_type = int.from_bytes(head[:4], byteorder) length = int.from_bytes(head[4:8], byteorder) - if length < 12: + if length < GENERIC_BLOCK_MINIMUM: raise FormatError(f'PCAP-NG: [Block {block_type}] invalid block length: {length}') buf = head + self._file.read(length - len(head)) @@ -172,7 +183,7 @@ def _read_packet(self, buf: 'bytes', block_type: 'int', tsresol = interface.options.get(Enum_OptionType.if_tsresol) tsoffset = interface.options.get(Enum_OptionType.if_tsoffset) - resolution = 1_000_000 if tsresol is None else tsresol.resolution + resolution = DEFAULT_RESOLUTION if tsresol is None else tsresol.resolution offset = 0 if tsoffset is None else tsoffset.offset # same arithmetic as :meth:`PCAPNG._read_timestamp @@ -217,7 +228,7 @@ def _read_simple(self, buf: 'bytes') -> 'tuple[float, bytes]': tsresol = cast('Optional[Data_IF_TSResolOption]', interface.options.get(Enum_OptionType.if_tsresol)) - resolution = 1_000_000 if tsresol is None else tsresol.resolution + resolution = DEFAULT_RESOLUTION if tsresol is None else tsresol.resolution self._linktype = interface.linktype return (Timestamp(0.0, decimal.Decimal(0), resolution, spb.original_len), @@ -280,7 +291,7 @@ def __iter__(self) -> 'Iterator[tuple[float, bytes]]': file, header = self._Reader__f, self._Reader__ph # pylint: disable=no-member nanosecond = isinstance(self._divisor, decimal.Decimal) - resolution = 1_000_000_000 if nanosecond else 1_000_000 + resolution = RESOLUTION[nanosecond] while True: buf = file.read(header.__hdr_len__) if not buf: diff --git a/pcapkit/foundation/engines/pcap.py b/pcapkit/foundation/engines/pcap.py index 67e8399731..4f286079cc 100644 --- a/pcapkit/foundation/engines/pcap.py +++ b/pcapkit/foundation/engines/pcap.py @@ -15,6 +15,8 @@ from pcapkit.foundation.engines.engine import EngineBase from pcapkit.protocols.misc.pcap.frame import Frame from pcapkit.protocols.misc.pcap.header import Header +from pcapkit.protocols.schema.misc.pcap.frame import RECORD_HEADER_LEN +from pcapkit.protocols.schema.misc.pcap.header import MAGIC_NUMBER as Schema_MAGIC_NUMBER from pcapkit.utilities.logging import get_logger __all__ = ['PCAP'] @@ -46,12 +48,10 @@ class PCAP(EngineBase[Frame]): #: Nanosecond flag. _nnsec: 'bool' - MAGIC_NUMBER = ( - b'\xa1\xb2\x3c\x4d', - b'\xa1\xb2\xc3\xd4', - b'\x4d\x3c\xb2\xa1', - b'\xd4\xc3\xb2\xa1', - ) + #: The PCAP magic numbers this engine claims, read from their one home, + #: :data:`pcapkit.protocols.schema.misc.pcap.header.MAGIC_NUMBER` + #: (:issue:`1739`). + MAGIC_NUMBER = tuple(Schema_MAGIC_NUMBER) ########################################################################## # Defaults. @@ -234,10 +234,10 @@ def _bound_read_ahead(self) -> 'None': return position = ifile.tell() - head = ifile.read(16) + head = ifile.read(RECORD_HEADER_LEN) ifile.seek(position, io.SEEK_SET) - size = 16 - if len(head) == 16: + size = RECORD_HEADER_LEN + if len(head) == RECORD_HEADER_LEN: size += int.from_bytes(head[8:12], self._gbhdr.byteorder) ifile._read_ahead_size = size diff --git a/pcapkit/foundation/engines/pcapng.py b/pcapkit/foundation/engines/pcapng.py index dc79a38ca6..8c4aab2ac9 100644 --- a/pcapkit/foundation/engines/pcapng.py +++ b/pcapkit/foundation/engines/pcapng.py @@ -18,6 +18,8 @@ from pcapkit.dumpkit.pcap import PCAPIO from pcapkit.foundation.engines.engine import EngineBase from pcapkit.protocols.misc.pcapng import PCAPNG as P_PCAPNG +from pcapkit.protocols.schema.misc.pcapng import (BLOCK_MINIMUM, BYTE_ORDER_MAGIC, + GENERIC_BLOCK_MINIMUM) from pcapkit.utilities.exceptions import FormatError, stacklevel from pcapkit.utilities.logging import get_logger from pcapkit.utilities.warnings import DeprecatedFormatWarning, warn @@ -104,31 +106,23 @@ class PCAPNG(EngineBase[P_PCAPNG]): #: across the whole file rather than per section. _block_count: 'collections.Counter[str]' + #: The PCAP-NG magic number this engine claims: the Block Type of a Section + #: Header Block, read from its one home, + #: :class:`~pcapkit.const.pcapng.block_type.BlockType` (:issue:`1739`). It + #: reads the same in either byte order. MAGIC_NUMBER = ( - b'\x0a\x0d\x0d\x0a', + Enum_BlockType.Section_Header_Block.to_bytes(4, 'big'), ) - #: A section header block's Byte-Order Magic, mapped to the byte order it - #: declares for its section. - BYTE_ORDER_MAGIC = { - b'\x1a\x2b\x3c\x4d': 'big', - b'\x4d\x3c\x2b\x1a': 'little', + #: A Section Header Block's Byte-Order Magic, as the octets read, mapped to + #: the byte order it declares for its section; built from the one home of + #: the value, :data:`pcapkit.protocols.schema.misc.pcapng.BYTE_ORDER_MAGIC` + #: (:issue:`1739`). + _BYTE_ORDER_OCTETS = { + BYTE_ORDER_MAGIC.to_bytes(4, 'big'): 'big', + BYTE_ORDER_MAGIC.to_bytes(4, 'little'): 'little', } # type: dict[bytes, Literal['big', 'little']] - #: Block types mapped to the octets their fixed fields span, Block Type and - #: both copies of Block Total Length included; any other block spans 12. - BLOCK_MINIMUM = { - Enum_BlockType.Section_Header_Block: 28, - Enum_BlockType.Interface_Description_Block: 20, - Enum_BlockType.Packet_Block: 32, - Enum_BlockType.Simple_Packet_Block: 16, - Enum_BlockType.Interface_Statistics_Block: 24, - Enum_BlockType.Enhanced_Packet_Block: 32, - Enum_BlockType.Decryption_Secrets_Block: 20, - Enum_BlockType.Custom_Block_that_rewriters_can_copy_into_new_files: 16, - Enum_BlockType.Custom_Block_that_rewriters_should_not_copy_into_new_files: 16, - } # type: dict[int, int] - #: Block types that carry a captured packet, mapped to the tag used when #: reporting them. Parsing any of them resolves the interface the packet was #: captured on, so the enclosing section must describe at least one @@ -447,18 +441,18 @@ def _bound_read_ahead(self) -> 'None': head = self._peek(12) if head[:4] == self.MAGIC_NUMBER[0]: - byteorder = self.BYTE_ORDER_MAGIC.get(head[8:12]) + byteorder = self._BYTE_ORDER_OCTETS.get(head[8:12]) else: byteorder = self._ctx.section.byteorder if self._ctx is not None else None if byteorder is not None and len(head) >= 8: block_type = int.from_bytes(head[:4], byteorder) length = int.from_bytes(head[4:8], byteorder) - if length >= self.BLOCK_MINIMUM.get(block_type, 12) and length % 4 == 0: + if length >= BLOCK_MINIMUM.get(block_type, GENERIC_BLOCK_MINIMUM) and length % 4 == 0: size = length if block_type == Enum_BlockType.Decryption_Secrets_Block: secrets = int.from_bytes(self._peek(16)[12:16], byteorder) - size = max(size, 20 + secrets + -secrets % 4) + size = max(size, BLOCK_MINIMUM[block_type] + secrets + -secrets % 4) ifile._read_ahead_size = size def _check_packet_block_context(self) -> 'None': diff --git a/pcapkit/foundation/engines/pypcapfile.py b/pcapkit/foundation/engines/pypcapfile.py index bcbc588413..c3cf18317c 100644 --- a/pcapkit/foundation/engines/pypcapfile.py +++ b/pcapkit/foundation/engines/pypcapfile.py @@ -20,6 +20,7 @@ from pcapkit.corekit.io import NamedStream from pcapkit.foundation.engines.engine import EngineBase from pcapkit.foundation.reassembly import ReassemblyManager +from pcapkit.protocols.schema.misc.pcap.frame import RECORD_HEADER_LEN from pcapkit.utilities.exceptions import FormatError, stacklevel from pcapkit.utilities.warnings import AttributeWarning, ProtocolWarning, warn @@ -363,8 +364,8 @@ def _read_records(self, file: 'NamedStream', number = 0 while True: number += 1 - record = file.read(16) - if len(record) < 16: + record = file.read(RECORD_HEADER_LEN) + if len(record) < RECORD_HEADER_LEN: return timestamp, timestamp_us, capture_len, packet_len = struct.unpack(f'{order}IIII', record) data = file.read(capture_len) diff --git a/pcapkit/foundation/engines/scapy.py b/pcapkit/foundation/engines/scapy.py index ad3776674f..c4647b04fd 100644 --- a/pcapkit/foundation/engines/scapy.py +++ b/pcapkit/foundation/engines/scapy.py @@ -41,6 +41,9 @@ from typing import TYPE_CHECKING, cast from pcapkit.foundation.engines.engine import EngineBase +from pcapkit.protocols.schema.misc.pcap.frame import RECORD_HEADER_LEN +from pcapkit.protocols.schema.misc.pcap.header import RESOLUTION +from pcapkit.protocols.schema.misc.pcapng import DEFAULT_RESOLUTION from pcapkit.utilities.exceptions import VersionError, stacklevel from pcapkit.utilities.logging import get_logger from pcapkit.utilities.warnings import AttributeWarning, warn @@ -186,8 +189,8 @@ class _PcapRecords(RawPcapReader): # self)._read_packet``, which looks in the classes after ``PcapReader`` # in the reader's MRO, so this one is a base listed after it. def _read_packet(self, size: 'int' = _CAPLEN_MAX) -> 'tuple[bytes, Any]': - header = self.f.read(16) - if len(header) < 16: + header = self.f.read(RECORD_HEADER_LEN) + if len(header) < RECORD_HEADER_LEN: raise EOFError sec, usec, caplen, wirelen = struct.unpack(f'{self.endian}IIII', header) chunks = [] # type: list[bytes] @@ -206,7 +209,7 @@ def _read_packet(self, size: 'int' = _CAPLEN_MAX) -> 'tuple[bytes, Any]': class _PcapReader(_Closing, _Uncapped, PcapReader, _PcapRecords): def read_packet(self, *args: 'Any', **kwargs: 'Any') -> 'ScapyPacket': packet = super().read_packet(*args, **kwargs) - attach_resolution(packet, 1_000_000_000 if self.nano else 1_000_000) + attach_resolution(packet, RESOLUTION[bool(self.nano)]) attach_linktype(packet, self.linktype) return packet @@ -214,7 +217,7 @@ class _PcapNgReader(_Closing, _Uncapped, PcapNgReader): # the metaclass makes this ``_PcapReader``'s alternative in turn alternative = _PcapReader - _resolution = 1_000_000 + _resolution = DEFAULT_RESOLUTION # each set from the packet block last read, before ``read_packet`` reads it _linktype: 'int' _timed: 'bool' diff --git a/pcapkit/foundation/reassembly/ip.py b/pcapkit/foundation/reassembly/ip.py index 112c05a676..fb8a0874df 100644 --- a/pcapkit/foundation/reassembly/ip.py +++ b/pcapkit/foundation/reassembly/ip.py @@ -20,6 +20,7 @@ from pcapkit.foundation.reassembly.data.ip import (_AT, Buffer, BufferID, Datagram, DatagramID, Deferred, Packet) from pcapkit.foundation.reassembly.reassembly import ReassemblyBase +from pcapkit.protocols.schema.internet.ipv6 import IP_MAX_LENGTH if TYPE_CHECKING: from typing import Type @@ -29,11 +30,6 @@ __all__ = ['IP'] -#: The most octets a 16-bit length field declares -- IPv4's Total Length, the -#: header included, and IPv6's Payload Length, the fixed header excluded -- and -#: so the length the data buffer is preallocated to. -_MAX_LENGTH = 65535 - class IP(ReassemblyBase[Packet[_AT], Datagram[_AT], BufferID, Buffer[_AT]], Generic[_AT]): # pylint: disable=abstract-method """Reassembly for IP payload. @@ -160,7 +156,7 @@ def reassembly(self, info: 'Packet[_AT]') -> 'None': RCVBT=bytearray(8192), # Fragment Received Bit Table index=[], # index record header=header, # header buffer - datagram=bytearray(_MAX_LENGTH), # data buffer + datagram=bytearray(IP_MAX_LENGTH), # data buffer timestamp=TS, # first-arriving fragment's clock reading conflict=[], # conflicting octet ranges ) @@ -203,8 +199,8 @@ def reassembly(self, info: 'Packet[_AT]') -> 'None': if TL == 0: length = len(info.payload) extent = FO + length - if self._header_overhead(IHL) + extent > _MAX_LENGTH: - size = max(extent, _MAX_LENGTH + 1, len(buf.datagram)) + if self._header_overhead(IHL) + extent > IP_MAX_LENGTH: + size = max(extent, IP_MAX_LENGTH + 1, len(buf.datagram)) buf.datagram.extend(bytes(size - len(buf.datagram))) buf.RCVBT.extend(bytes((size + 7) // 8 - len(buf.RCVBT))) else: @@ -279,7 +275,8 @@ def _is_complete(self, buf: 'Buffer[_AT]') -> 'bool': than it may be, and every 8-octet block up to it is marked received. - A datagram is no longer than ``_MAX_LENGTH`` with its header, as + A datagram is no longer than + :data:`~pcapkit.protocols.schema.internet.ipv6.IP_MAX_LENGTH` with its header, as :meth:`_header_overhead` counts it: IPv4's Total Length counts the whole header, and IPv6's Payload Length every header but the fixed one. A longer datagram cannot be declared, so it is never complete, and @@ -292,10 +289,10 @@ def _is_complete(self, buf: 'Buffer[_AT]') -> 'bool': """ TDL = buf.TDL - if len(buf.datagram) > _MAX_LENGTH: + if len(buf.datagram) > IP_MAX_LENGTH: limit = len(buf.datagram) else: - limit = _MAX_LENGTH - self._header_overhead(len(buf.header)) + limit = IP_MAX_LENGTH - self._header_overhead(len(buf.header)) return 0 <= TDL <= limit and all(buf.RCVBT[:(TDL + 7) // 8]) @staticmethod diff --git a/pcapkit/foundation/reassembly/ipv6.py b/pcapkit/foundation/reassembly/ipv6.py index 774a562c33..bfa02d15e7 100644 --- a/pcapkit/foundation/reassembly/ipv6.py +++ b/pcapkit/foundation/reassembly/ipv6.py @@ -14,6 +14,7 @@ from pcapkit.const.reg.transtype import TransType as Enum_TransType from pcapkit.foundation.reassembly.ip import IP +from pcapkit.protocols.internet.ipv6 import IPV6_HEADER_LEN, IPV6_NEXT_HEADER_OFFSET from pcapkit.protocols.internet.ipv6 import IPv6 as IPv6_Protocol if TYPE_CHECKING: @@ -21,14 +22,6 @@ __all__ = ['IPv6'] -#: Length of the fixed IPv6 header, i.e. the offset of the first extension -#: header (:rfc:`8200#section-3`). A literal because it is a *byte offset* into -#: the header rather than a protocol number, so no enumeration carries it. -_IPV6_HDR_LEN = 40 - -#: Offset of the Next Header field within the fixed IPv6 header. A literal for -#: the same reason as :data:`_IPV6_HDR_LEN`. -_IPV6_NEXT_HEADER = 6 def _next_header_offset(header: 'bytes') -> 'int': """Locate the Next Header field of a datagram's last header. @@ -47,8 +40,8 @@ def _next_header_offset(header: 'bytes') -> 'int': Offset, within ``header``, of the Next Header field to rewrite. """ - offset = _IPV6_NEXT_HEADER - position = _IPV6_HDR_LEN + offset = IPV6_NEXT_HEADER_OFFSET + position = IPV6_HEADER_LEN proto = header[offset] # NOTE: ``header[position]`` is the Next Header field of the extension @@ -152,7 +145,7 @@ def _rectify_header(self, header: 'bytes', proto: 'TransType') -> 'bytes': # a header too short to hold the fixed IPv6 header cannot be walked, and # a chain not ending in the Fragment header has nothing to rewrite -- # which also makes this idempotent - if len(header) < _IPV6_HDR_LEN: + if len(header) < IPV6_HEADER_LEN: return header offset = _next_header_offset(header) if header[offset] != Enum_TransType.IPv6_Frag: @@ -174,4 +167,4 @@ def _header_overhead(self, length: 'int') -> 'int': How many of those octets the Payload Length counts. """ - return max(length - _IPV6_HDR_LEN, 0) + return max(length - IPV6_HEADER_LEN, 0) diff --git a/pcapkit/protocols/internet/ah.py b/pcapkit/protocols/internet/ah.py index 908a9fbd67..0325c32abc 100644 --- a/pcapkit/protocols/internet/ah.py +++ b/pcapkit/protocols/internet/ah.py @@ -31,6 +31,7 @@ from pcapkit.protocols.internet.ipsec import IPsec from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext from pcapkit.protocols.schema.internet.ah import AH as Schema_AH +from pcapkit.protocols.schema.internet.ah import AH_HEADER_LEN from pcapkit.utilities.exceptions import UnsupportedCall if TYPE_CHECKING: @@ -219,7 +220,7 @@ def make(self, """ next_value = self._make_index(next, next_default, namespace=next_namespace, reversed=next_reversed, pack=False) - length = (len(icv) + 12) // 4 - 2 + length = (len(icv) + AH_HEADER_LEN) // 4 - 2 return Schema_AH( next=next_value, # type: ignore[arg-type] diff --git a/pcapkit/protocols/internet/ipv4.py b/pcapkit/protocols/internet/ipv4.py index 23f419d50a..782f0e096a 100644 --- a/pcapkit/protocols/internet/ipv4.py +++ b/pcapkit/protocols/internet/ipv4.py @@ -82,6 +82,7 @@ from pcapkit.protocols.data.internet.ipv4 import TSOption as Data_TSOption from pcapkit.protocols.data.internet.ipv4 import UnassignedOption as Data_UnassignedOption from pcapkit.protocols.internet.ip import IP +from pcapkit.protocols.schema.internet.ipv4 import IPV4_HEADER_LEN from pcapkit.protocols.schema.internet.ipv4 import EOOLOption as Schema_EOOLOption from pcapkit.protocols.schema.internet.ipv4 import ESECOption as Schema_ESECOption from pcapkit.protocols.schema.internet.ipv4 import IPv4 as Schema_IPv4 @@ -328,14 +329,14 @@ def read(self, length: 'Optional[int]' = None, *, # pylint: disable=unused-argu raise ProtocolError(f'{self.alias}: header length {ipv4.hdr_len} runs past the end of ' f'the data ({len(self)} octets)') - _optl = ipv4.hdr_len - 20 + _optl = ipv4.hdr_len - IPV4_HEADER_LEN if _optl: options = self._read_ipv4_options(_optl) _used = sum(opt.length for _, opt in options.items(multi=True)) ipv4.__update__([ ('options', options), # octets after the End of Option List, as captured - ('padding', self._data[20 + _used:ipv4.hdr_len]), + ('padding', self._data[IPV4_HEADER_LEN + _used:ipv4.hdr_len]), ]) # NOTE: Octets captured past the Total Length belong to this layer, @@ -578,7 +579,7 @@ def register_option(cls, code: 'Enum_OptionNumber', meth: 'str | tuple[OptionPar def __length_hint__(self) -> 'Literal[20]': """Return an estimated length for the object.""" - return 20 + return IPV4_HEADER_LEN # type: ignore[return-value] @classmethod def __index__(cls) -> 'Enum_TransType': # pylint: disable=invalid-index-returned diff --git a/pcapkit/protocols/internet/ipv6.py b/pcapkit/protocols/internet/ipv6.py index c3a8093848..1397f6b820 100644 --- a/pcapkit/protocols/internet/ipv6.py +++ b/pcapkit/protocols/internet/ipv6.py @@ -38,6 +38,7 @@ from pcapkit.protocols.data.internet.ipv6 import IPv6 as Data_IPv6 from pcapkit.protocols.internet.ip import IP from pcapkit.protocols.protocol import _frame_depth +from pcapkit.protocols.schema.internet.ipv6 import IP_MAX_LENGTH from pcapkit.protocols.schema.internet.ipv6 import IPv6 as Schema_IPv6 from pcapkit.protocols.schema.internet.ipv6 import jumbo_payload_length, jumbogram_error from pcapkit.utilities import config @@ -56,6 +57,19 @@ __all__ = ['IPv6'] +#: Length of the fixed IPv6 header, i.e. the offset of the first extension +#: header, :rfc:`8200#section-3`. The one home of this value (:issue:`1739`); it +#: is fixed by the RFC, so it is never configurable. +#: +#: .. versionadded:: 1.5.0 +IPV6_HEADER_LEN = 40 + +#: Offset of the Next Header field within the fixed IPv6 header, +#: :rfc:`8200#section-3`. The one home of this value (:issue:`1739`). +#: +#: .. versionadded:: 1.5.0 +IPV6_NEXT_HEADER_OFFSET = 6 + class IPv6(IP[Data_IPv6, Schema_IPv6], schema=Schema_IPv6, data=Data_IPv6): @@ -128,7 +142,7 @@ def name(self) -> 'Literal["Internet Protocol version 6"]': @property def length(self) -> 'Literal[40]': """Header length of corresponding protocol.""" - return 40 + return IPV6_HEADER_LEN # type: ignore[return-value] @property def protocol(self) -> 'Enum_TransType': @@ -311,7 +325,7 @@ def make(self, # (:issue:`1687`). if payload_length is None: payload_length = len(payload) - if payload_length > 0xFFFF or jumbo_payload_length(next_val, bytes(payload)) is not None: + if payload_length > IP_MAX_LENGTH or jumbo_payload_length(next_val, bytes(payload)) is not None: payload_length = 0 return Schema_IPv6( @@ -345,7 +359,7 @@ def id(cls) -> 'tuple[Literal["IPv6"]]': # type: ignore[override] def __length_hint__(self) -> 'Literal[40]': """Return an estimated length for the object.""" - return 40 + return IPV6_HEADER_LEN # type: ignore[return-value] @classmethod def __index__(cls) -> 'Enum_TransType': # pylint: disable=invalid-index-returned diff --git a/pcapkit/protocols/misc/pcap/frame.py b/pcapkit/protocols/misc/pcap/frame.py index c3a6de617e..d5c8ed7002 100644 --- a/pcapkit/protocols/misc/pcap/frame.py +++ b/pcapkit/protocols/misc/pcap/frame.py @@ -35,7 +35,9 @@ from pcapkit.protocols.data.misc.pcap.frame import Frame as Data_Frame from pcapkit.protocols.data.misc.pcap.frame import FrameInfo as Data_FrameInfo from pcapkit.protocols.protocol import ProtocolBase +from pcapkit.protocols.schema.misc.pcap.frame import RECORD_HEADER_LEN from pcapkit.protocols.schema.misc.pcap.frame import Frame as Schema_Frame +from pcapkit.protocols.schema.misc.pcap.header import RESOLUTION from pcapkit.utilities.compat import localcontext from pcapkit.utilities.exceptions import StreamEOFError, UnsupportedCall, stacklevel from pcapkit.utilities.warnings import ProtocolWarning, RegistryWarning, warn @@ -118,7 +120,7 @@ def name(self) -> 'str': @property def length(self) -> 'Literal[16]': """Header length of corresponding protocol.""" - return 16 + return RECORD_HEADER_LEN # type: ignore[return-value] @property def header(self) -> 'Data_Header': @@ -261,10 +263,7 @@ def read(self, length: 'Optional[int]' = None, *, _read: 'bool' = True, _olen = schema.orig_len with localcontext(prec=64): - if self._nsec: - _epch = _tsss + decimal.Decimal(_tsus) / 1_000_000_000 - else: - _epch = _tsss + decimal.Decimal(_tsus) / 1_000_000 + _epch = _tsss + decimal.Decimal(_tsus) / RESOLUTION[bool(self._nsec)] _irat = _epch.as_integer_ratio() try: @@ -470,7 +469,7 @@ def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None, length: 'Opt def __length_hint__(self) -> 'Literal[16]': """Return an estimated length for the object.""" - return 16 + return RECORD_HEADER_LEN # type: ignore[return-value] def _check_record_floor(self, length: 'Optional[int]') -> 'None': """Refuse a tail too short to hold a record header. @@ -497,9 +496,10 @@ def _check_record_floor(self, length: 'Optional[int]') -> 'None': length = self._file.seek(0, io.SEEK_END) - current self._file.seek(current, io.SEEK_SET) - if length < 16: + if length < RECORD_HEADER_LEN: raise StreamEOFError(f'PCAP: record truncated: {length} octet(s) left, ' - 'fewer than the 16 a record header needs', quiet=True) + f'fewer than the {RECORD_HEADER_LEN} a record header needs', + quiet=True) # NOTE: This is a hack to make the ``__index__`` method work both as a # class method and an instance method. @@ -579,7 +579,7 @@ def _make_timestamp(self, timestamp: 'Optional[float | Decimal | dt_type | int]' ts_sec = int(timestamp) if ts_usec is None: - ts_usec = int((timestamp - ts_sec) * (1_000_000_000 if nanosecond else 1_000_000)) + ts_usec = int((timestamp - ts_sec) * RESOLUTION[bool(nanosecond)]) return ts_sec, ts_usec diff --git a/pcapkit/protocols/misc/pcap/header.py b/pcapkit/protocols/misc/pcap/header.py index 80bd185eec..1ec2c64bf7 100644 --- a/pcapkit/protocols/misc/pcap/header.py +++ b/pcapkit/protocols/misc/pcap/header.py @@ -34,6 +34,7 @@ from pcapkit.protocols.data.misc.pcap.header import Header as Data_Header from pcapkit.protocols.data.misc.pcap.header import MagicNumber as Data_MagicNumber from pcapkit.protocols.protocol import ProtocolBase +from pcapkit.protocols.schema.misc.pcap.header import GLOBAL_HEADER_LEN, MAGIC_NUMBER from pcapkit.protocols.schema.misc.pcap.header import Header as Schema_Header from pcapkit.protocols.schema.schema import keep_short_read, replay_short_read from pcapkit.utilities.exceptions import EndianError, FileError, UnsupportedCall @@ -47,17 +48,11 @@ __all__ = ['Header'] -#: dict[tuple[str, bool], bytes]: Mapping of PCAP file magic numbers. The key -#: is a :obj:`tuple` of endianness (``big`` or ``little``) and nanosecond- -#: timestamp resolution flag (:data:`True` refers to nanosecond-resolution -#: timestamp, :data:`False` refers to microsecond-resolution timestamp); the -#: value is the corresponding PCAP file magic number sequence. -_MAGIC_NUM = { - ('big', True): b'\xa1\xb2\x3c\x4d', - ('big', False): b'\xa1\xb2\xc3\xd4', - ('little', True): b'\x4d\x3c\xb2\xa1', - ('little', False): b'\xd4\xc3\xb2\xa1', -} +#: The PCAP magic number for each byte order and nanosecond flag, the reverse of +#: :data:`~pcapkit.protocols.schema.misc.pcap.header.MAGIC_NUMBER`, so that +#: :meth:`Header._make_magic` looks one up rather than searching for it -- an +#: undeclared pair raises :exc:`KeyError`, as it did before :issue:`1739`. +_DECLARED_MAGIC = {declared: magic for magic, declared in MAGIC_NUMBER.items()} class Header(ProtocolBase[Data_Header, Schema_Header], @@ -76,7 +71,7 @@ def name(self) -> 'Literal["Global Header"]': @property def length(self) -> 'Literal[24]': """Header length of corresponding protocol.""" - return 24 + return GLOBAL_HEADER_LEN # type: ignore[return-value] @property def version(self) -> 'VersionInfo': @@ -149,20 +144,14 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_Header' schema = self.__header__ _magn = schema.magic_number - if _magn == b'\xd4\xc3\xb2\xa1': - self._nsec = False - self._byte = 'little' - elif _magn == b'\xa1\xb2\xc3\xd4': - self._nsec = False - self._byte = 'big' - elif _magn == b'\x4d\x3c\xb2\xa1': - self._nsec = True - self._byte = 'little' - elif _magn == b'\xa1\xb2\x3c\x4d': - self._nsec = True - self._byte = 'big' - else: + # NOTE: looked up as :obj:`bytes`, so that a :obj:`bytearray` magic + # number is matched as it was compared before :issue:`1739` rather + # than rejected as unhashable. + _decl = (MAGIC_NUMBER.get(bytes(_magn)) + if isinstance(_magn, (bytes, bytearray, memoryview)) else None) + if _decl is None: raise FileError(5, 'Unknown file format', self._file.name) # pylint: disable=no-member + self._byte, self._nsec = _decl header = Data_Header( magic_number=Data_MagicNumber( @@ -280,11 +269,11 @@ def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None, def __len__(self) -> 'Literal[24]': """Total length of corresponding protocol.""" - return 24 + return GLOBAL_HEADER_LEN # type: ignore[return-value] def __length_hint__(self) -> 'Literal[24]': """Return an estimated length for the object.""" - return 24 + return GLOBAL_HEADER_LEN # type: ignore[return-value] @classmethod def from_data(cls, data: 'Data_Header | dict[str, Any]', **kwargs: 'Any') -> 'Self': @@ -377,5 +366,5 @@ def _make_magic(self, byteorder: 'Literal["big", "little"]' = sys.byteorder, if byteorder not in ('little', 'big'): raise EndianError(f"unknown byte order: {byteorder!r}") - magic_number = _MAGIC_NUM[(byteorder, nanosecond)] + magic_number = _DECLARED_MAGIC[(byteorder, nanosecond)] return magic_number, (byteorder == 'little') diff --git a/pcapkit/protocols/misc/pcapng.py b/pcapkit/protocols/misc/pcapng.py index 9b68b742cf..3696e78163 100644 --- a/pcapkit/protocols/misc/pcapng.py +++ b/pcapkit/protocols/misc/pcapng.py @@ -111,6 +111,9 @@ from pcapkit.protocols.data.misc.pcapng import ZigBeeNWKKey as Data_ZigBeeNWKKey from pcapkit.protocols.data.protocol import Packet as Data_Packet from pcapkit.protocols.protocol import ProtocolBase +from pcapkit.protocols.schema.misc.pcap.header import RESOLUTION +from pcapkit.protocols.schema.misc.pcapng import (BLOCK_MINIMUM, BYTE_ORDER_MAGIC, + DEFAULT_RESOLUTION, GENERIC_BLOCK_MINIMUM) from pcapkit.protocols.schema.misc.pcapng import PCAPNG as Schema_PCAPNG from pcapkit.protocols.schema.misc.pcapng import BlockType as Schema_BlockType from pcapkit.protocols.schema.misc.pcapng import CommentOption as Schema_CommentOption @@ -678,10 +681,12 @@ class PCAPNG(ProtocolBase[Data_PCAPNG, Schema_PCAPNG], #: Header blocks mapped to the octets their fixed fields span, so that a #: shorter Block Total Length is kept as captured (see - #: :meth:`_read_block_short_header`). + #: :meth:`_read_block_short_header`). The lengths are read from their one + #: home, :data:`pcapkit.protocols.schema.misc.pcapng.BLOCK_MINIMUM` + #: (:issue:`1739`). SHORT_HEADER_TYPES = { - Enum_BlockType.Section_Header_Block: 28, - Enum_BlockType.Interface_Description_Block: 20, + block: BLOCK_MINIMUM[block] for block in (Enum_BlockType.Section_Header_Block, + Enum_BlockType.Interface_Description_Block) } #: Blocks that resolve an interface of their section, mapped to the tag used @@ -932,7 +937,7 @@ def byteorder(self) -> 'Literal["big", "little"]': @property def nanosecond(self) -> 'bool': """Whether the timestamp is in nanosecond.""" - return self.ts_resolution > 1_000_000 + return self.ts_resolution > RESOLUTION[False] @property def ts_resolution(self) -> 'int': @@ -941,7 +946,7 @@ def ts_resolution(self) -> 'int': #raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute 'ts_resolution'") warn(f"'{self.__class__.__name__}' object has no attribute 'ts_resolution'", AttributeWarning, stacklevel=stacklevel()) - return 1_000_000 + return DEFAULT_RESOLUTION info = cast('Packet', self._info) return self._get_resolution(info.interface_id) @@ -1258,7 +1263,7 @@ def read(self, length: 'Optional[int]' = None, *, _read: 'bool' = True, """ schema = self.__header__ - if schema.block.length < 12 or schema.block.length % 4 != 0: + if schema.block.length < GENERIC_BLOCK_MINIMUM or schema.block.length % 4 != 0: raise ProtocolError(f'PCAP-NG: [Block {schema.type}] invalid length: {schema.block.length}') self._type = schema.type @@ -1267,7 +1272,7 @@ def read(self, length: 'Optional[int]' = None, *, _read: 'bool' = True, self._ctx = None name = self._lookup_registry(self.__block__, schema.type) - if schema.type in self.SHORT_PACKET_TYPES and schema.block.length < 32: + if schema.type in self.SHORT_PACKET_TYPES and schema.block.length < BLOCK_MINIMUM[schema.type]: meth = cast('BlockParser', self._read_block_short) elif schema.block.length < self.SHORT_HEADER_TYPES.get(schema.type, 0): meth = cast('BlockParser', self._read_block_short_header) @@ -1472,7 +1477,7 @@ def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None, length: 'Opt def __length_hint__(self) -> 'Literal[12]': """Return an estimated length for the object.""" - return 12 + return GENERIC_BLOCK_MINIMUM # type: ignore[return-value] # NOTE: This is a hack to make the ``__index__`` method work both as a # class method and an instance method. @@ -1553,9 +1558,9 @@ def _check_block_floor(self, length: 'Optional[int]') -> 'None': length = self._file.seek(0, io.SEEK_END) - current self._file.seek(current, io.SEEK_SET) - if length < 12: + if length < GENERIC_BLOCK_MINIMUM: raise StreamEOFError(f'PCAP-NG: block truncated: {length} octet(s) left, ' - 'fewer than the 12 a block needs', quiet=True) + f'fewer than the {GENERIC_BLOCK_MINIMUM} a block needs', quiet=True) def _captured(self, offset: 'int') -> 'int': """Count the octets left in the stream from ``offset`` on. @@ -1594,7 +1599,7 @@ def _read_block_cut(self, header: 'Schema_PCAPNG', offset: 'int', current = self._file.tell() self._file.seek(offset + 8, io.SEEK_SET) - body = self._read_fileng(header.block.length - 12) + body = self._read_fileng(header.block.length - GENERIC_BLOCK_MINIMUM) self._file.seek(current, io.SEEK_SET) return Data_UnknownBlock( @@ -1706,13 +1711,13 @@ def _get_resolution(self, interface_id: 'int' = 0) -> 'int': # raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute '_get_resolution'") warn(f"'{self.__class__.__name__}' object has no attribute '_get_resolution'", AttributeWarning, stacklevel=stacklevel()) - return 1_000_000 + return DEFAULT_RESOLUTION options = self._get_interface(interface_id).options tsresol = cast('Optional[Data_IF_TSResolOption]', options.get(Enum_OptionType.if_tsresol)) if tsresol is None: - return 1_000_000 + return DEFAULT_RESOLUTION return tsresol.resolution def _get_offset(self, interface_id: 'int' = 0) -> 'int': @@ -4308,7 +4313,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) - length = len(data) + 12 + length = len(data) + GENERIC_BLOCK_MINIMUM return Schema_UnknownBlock( length=length, @@ -4364,13 +4369,13 @@ def _make_block_shb(self, block: 'Optional[Data_SectionHeaderBlock]' = None, *, total_length += self._make_options_trailer(block, options_value) return Schema_SectionHeaderBlock( - length=total_length + 28, - magic=0x1A2B3C4D, + length=total_length + BLOCK_MINIMUM[Enum_BlockType.Section_Header_Block], + magic=BYTE_ORDER_MAGIC, major=major_version, minor=minor_version, section_length=section_length, options=options_value, - length2=total_length + 28, + length2=total_length + BLOCK_MINIMUM[Enum_BlockType.Section_Header_Block], ) def _make_block_idb(self, block: 'Optional[Data_InterfaceDescriptionBlock]' = None, *, @@ -4414,11 +4419,11 @@ def _make_block_idb(self, block: 'Optional[Data_InterfaceDescriptionBlock]' = No total_length += self._make_options_trailer(block, options_value) return self._make_padding(Schema_InterfaceDescriptionBlock( - length=total_length + 20, + length=total_length + BLOCK_MINIMUM[Enum_BlockType.Interface_Description_Block], linktype=linktype_val, snaplen=snaplen, options=options_value, - length2=total_length + 20, + length2=total_length + BLOCK_MINIMUM[Enum_BlockType.Interface_Description_Block], ), block, 'reserved') def _make_block_epb(self, block: 'Optional[Data_EnhancedPacketBlock]' = None, *, @@ -4479,7 +4484,7 @@ def _make_block_epb(self, block: 'Optional[Data_EnhancedPacketBlock]' = None, *, total_length += self._make_options_trailer(block, options_value) return self._make_padding(Schema_EnhancedPacketBlock( - length=total_length + 32 + packet_len, + length=total_length + BLOCK_MINIMUM[Enum_BlockType.Enhanced_Packet_Block] + packet_len, interface_id=interface_id, timestamp_high=timestamp_high, timestamp_low=timestamp_low, @@ -4487,7 +4492,7 @@ def _make_block_epb(self, block: 'Optional[Data_EnhancedPacketBlock]' = None, *, original_len=original_len, packet_data=packet_data, options=options_value, - length2=total_length + 32 + packet_len, + length2=total_length + BLOCK_MINIMUM[Enum_BlockType.Enhanced_Packet_Block] + packet_len, ), block, 'padding_data') def _make_block_spb(self, block: 'Optional[Data_SimplePacketBlock]' = None, *, @@ -4528,10 +4533,10 @@ def _make_block_spb(self, block: 'Optional[Data_SimplePacketBlock]' = None, *, packet_len = math.ceil(len(packet_data) / 4) * 4 return self._make_padding(Schema_SimplePacketBlock( - length=packet_len + 16, + length=packet_len + BLOCK_MINIMUM[Enum_BlockType.Simple_Packet_Block], original_len=original_len, packet_data=packet_data, - length2=packet_len + 16, + length2=packet_len + BLOCK_MINIMUM[Enum_BlockType.Simple_Packet_Block], ), block, 'padding', 'padding_data') def _make_block_short( @@ -4582,10 +4587,10 @@ def _make_block_nrb(self, block: 'Optional[Data_NameResolutionBlock]' = None, *, options_length += self._make_options_trailer(block, options_value) return Schema_NameResolutionBlock( - length=options_length + records_length + 12, + length=options_length + records_length + GENERIC_BLOCK_MINIMUM, records=records_value, options=options_value, - length2=options_length + records_length + 12, + length2=options_length + records_length + GENERIC_BLOCK_MINIMUM, ) def _make_block_isb(self, block: 'Optional[Data_InterfaceStatisticsBlock]' = None, *, @@ -4623,12 +4628,12 @@ def _make_block_isb(self, block: 'Optional[Data_InterfaceStatisticsBlock]' = Non total_length += self._make_options_trailer(block, options_value) return Schema_InterfaceStatisticsBlock( - length=total_length + 24, + length=total_length + BLOCK_MINIMUM[Enum_BlockType.Interface_Statistics_Block], interface_id=interface_id, timestamp_high=timestamp_high, timestamp_low=timestamp_low, options=options_value, - length2=total_length + 24, + length2=total_length + BLOCK_MINIMUM[Enum_BlockType.Interface_Statistics_Block], ) def _make_block_systemd(self, block: 'Optional[Data_SystemdJournalExportBlock]' = None, *, @@ -4662,9 +4667,9 @@ def _make_block_systemd(self, block: 'Optional[Data_SystemdJournalExportBlock]' data += bytes(-len(data) % 4) return Schema_SystemdJournalExportBlock( - length=len(data) + 12, + length=len(data) + GENERIC_BLOCK_MINIMUM, entry=data, - length2=len(data) + 12, + length2=len(data) + GENERIC_BLOCK_MINIMUM, ) def _make_block_dsb(self, block: 'Optional[Data_DecryptionSecretsBlock]' = None, *, @@ -4730,12 +4735,12 @@ def _make_block_dsb(self, block: 'Optional[Data_DecryptionSecretsBlock]' = None, total_length += self._make_options_trailer(block, options_value) return self._make_padding(Schema_DecryptionSecretsBlock( - length=total_length + secrets_padded + 20, + length=total_length + secrets_padded + BLOCK_MINIMUM[Enum_BlockType.Decryption_Secrets_Block], secrets_type=secrets_type_val, secrets_length=secrets_length, secrets_data=secrets_data_val, options=options_value, - length2=total_length + secrets_padded + 20, + length2=total_length + secrets_padded + BLOCK_MINIMUM[Enum_BlockType.Decryption_Secrets_Block], ), block, 'padding_data') def _make_block_cb(self, block: 'Optional[Data_CustomBlock]' = None, *, @@ -4782,11 +4787,14 @@ def _make_block_cb(self, block: 'Optional[Data_CustomBlock]' = None, *, cb_data += option.pack() if isinstance(option, Schema) else option cb_data += bytes(-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. + minimum = BLOCK_MINIMUM[Enum_BlockType.Custom_Block_that_rewriters_can_copy_into_new_files] return Schema_CustomBlock( - length=len(cb_data) + 16, + length=len(cb_data) + minimum, pen=pen, data=cb_data, - length2=len(cb_data) + 16, + length2=len(cb_data) + minimum, ) def _make_block_packet(self, block: 'Optional[Data_PacketBlock]' = None, *, @@ -4853,7 +4861,7 @@ def _make_block_packet(self, block: 'Optional[Data_PacketBlock]' = None, *, total_length += self._make_options_trailer(block, options_value) return self._make_padding(Schema_PacketBlock( - length=total_length + 32 + packet_len, + length=total_length + BLOCK_MINIMUM[Enum_BlockType.Packet_Block] + packet_len, interface_id=interface_id, drop_count=drop_count, timestamp_high=timestamp_high, @@ -4862,7 +4870,7 @@ def _make_block_packet(self, block: 'Optional[Data_PacketBlock]' = None, *, original_length=original_len, packet_data=packet_data, options=options_value, - length2=total_length + 32 + packet_len, + length2=total_length + BLOCK_MINIMUM[Enum_BlockType.Packet_Block] + packet_len, ), block, 'padding_data') def _make_pcapng_options(self, options: 'Option | list[Schema_Option | tuple[Enum_OptionType, dict[str, Any]] | bytes] | bytes', @@ -5269,7 +5277,7 @@ def _make_option_if_speed(self, type: 'Enum_OptionType', option: 'Optional[Data_ ) def _make_option_if_tsresol(self, type: 'Enum_OptionType', option: 'Optional[Data_IF_TSResolOption]' = None, *, - resolution: 'int' = 1000000, + resolution: 'int' = DEFAULT_RESOLUTION, **kwargs: 'Any') -> 'Schema_IF_TSResolOption': """Make PCAP-NG ``if_tsresol`` option. diff --git a/pcapkit/protocols/schema/internet/ah.py b/pcapkit/protocols/schema/internet/ah.py index 012e063cb8..d80570933d 100644 --- a/pcapkit/protocols/schema/internet/ah.py +++ b/pcapkit/protocols/schema/internet/ah.py @@ -15,6 +15,15 @@ if TYPE_CHECKING: from pcapkit.protocols.protocol import ProtocolBase +#: Length of the fixed part of an IP Authentication Header (AH), before its +#: Integrity Check Value, :rfc:`4302#section-2`. The one home of this value +#: (:issue:`1739`), here so that the schema's ICV length and the protocol +#: modules read the same one; it is fixed by the RFC, so it is never +#: configurable. +#: +#: .. versionadded:: 1.5.0 +AH_HEADER_LEN = 12 + @schema_final class AH(Schema): @@ -31,7 +40,7 @@ class AH(Schema): #: Sequence number field. seq: 'int' = UInt32Field() #: Integrity check value. - icv: 'bytes' = BytesField(length=lambda pkt: (pkt['len'] + 2) * 4 - 12) + icv: 'bytes' = BytesField(length=lambda pkt: (pkt['len'] + 2) * 4 - AH_HEADER_LEN) #: Payload. payload: 'bytes' = PayloadField() diff --git a/pcapkit/protocols/schema/internet/ipv4.py b/pcapkit/protocols/schema/internet/ipv4.py index 36aee6b79d..e6502ffbb9 100644 --- a/pcapkit/protocols/schema/internet/ipv4.py +++ b/pcapkit/protocols/schema/internet/ipv4.py @@ -44,6 +44,14 @@ from pcapkit.corekit.fields.field import FieldBase as Field from pcapkit.protocols.protocol import ProtocolBase +#: Length of the fixed IPv4 header, i.e. with no options, :rfc:`791#section-3.1`. +#: The one home of this value (:issue:`1739`), here so that the schema's option +#: length and the protocol modules read the same one; it is fixed by the RFC, so +#: it is never configurable. +#: +#: .. versionadded:: 1.5.0 +IPV4_HEADER_LEN = 20 + if SPHINX_TYPE_CHECKING: # pragma: no cover from typing_extensions import TypedDict @@ -724,7 +732,7 @@ class IPv4(Schema): dst: 'IPv4Address' = IPv4AddressField() #: Options. options: 'list[Option]' = OptionField( - length=lambda pkt: pkt['vihl']['ihl'] * 4 - 20, + length=lambda pkt: pkt['vihl']['ihl'] * 4 - IPV4_HEADER_LEN, base_schema=Option, type_name='type', registry=Option.registry, diff --git a/pcapkit/protocols/schema/internet/ipv6.py b/pcapkit/protocols/schema/internet/ipv6.py index b2c11ffca4..524d7fd58d 100644 --- a/pcapkit/protocols/schema/internet/ipv6.py +++ b/pcapkit/protocols/schema/internet/ipv6.py @@ -21,6 +21,17 @@ from pcapkit.protocols.protocol import ProtocolBase +#: The most octets a 16-bit IP length field declares: IPv4's Total Length, the +#: header included (:rfc:`791#section-3.1`), and IPv6's Payload Length, the +#: fixed header excluded (:rfc:`8200#section-3`) -- so a longer IPv6 payload is +#: a jumbogram, whose Jumbo Payload Length must exceed it (:rfc:`2675#section-2`). +#: The one home of this value (:issue:`1739`), here so that the schema's +#: jumbogram check and the protocol modules read the same one; it is fixed by +#: the RFCs, so it is never configurable. +#: +#: .. versionadded:: 1.5.0 +IP_MAX_LENGTH = 65535 + if SPHINX_TYPE_CHECKING: # pragma: no cover from typing_extensions import TypedDict @@ -179,8 +190,8 @@ def jumbogram_error(length: 'int', next: 'int', data: 'bytes', # pylint: disabl return None if length: return f'Jumbo Payload option with a non-zero Payload Length {length}' - if jumbo <= 0xFFFF: - return f'Jumbo Payload Length {jumbo}, not greater than 65,535' + if jumbo <= IP_MAX_LENGTH: + return f'Jumbo Payload Length {jumbo}, not greater than {IP_MAX_LENGTH:,}' if fragment is not None and fragment(): return 'Jumbo Payload option with a Fragment header' return None diff --git a/pcapkit/protocols/schema/misc/pcap/frame.py b/pcapkit/protocols/schema/misc/pcap/frame.py index 64e8a7ac1e..20508286a8 100644 --- a/pcapkit/protocols/schema/misc/pcap/frame.py +++ b/pcapkit/protocols/schema/misc/pcap/frame.py @@ -14,6 +14,13 @@ if TYPE_CHECKING: from pcapkit.protocols.protocol import ProtocolBase +#: Length of a PCAP record header, which precedes each record's packet data. +#: The one home of this value (:issue:`1739`); the format fixes it, so it is +#: never configurable. +#: +#: .. versionadded:: 1.5.0 +RECORD_HEADER_LEN = 16 + @schema_final class Frame(Schema): diff --git a/pcapkit/protocols/schema/misc/pcap/header.py b/pcapkit/protocols/schema/misc/pcap/header.py index 8c18a70257..42c66a886e 100644 --- a/pcapkit/protocols/schema/misc/pcap/header.py +++ b/pcapkit/protocols/schema/misc/pcap/header.py @@ -13,10 +13,53 @@ __all__ = ['Header'] if TYPE_CHECKING: - from typing import Any + from typing import Any, Dict + + from typing_extensions import Literal from pcapkit.corekit.fields.numbers import NumberField +#: Length of the PCAP global header, which opens the file. The one home of +#: this value (:issue:`1739`); the format fixes it, so it is never configurable. +#: +#: .. versionadded:: 1.5.0 +GLOBAL_HEADER_LEN = 24 + +#: PCAP magic numbers, as the octets open the file, mapped to the byte order +#: (``'big'`` or ``'little'``) and the timestamp resolution (:data:`True` for +#: nanoseconds, :data:`False` for microseconds) each one declares. The one home +#: of these values (:issue:`1739`); the format fixes them, so they are never +#: configurable. +#: +#: .. versionadded:: 1.5.0 +MAGIC_NUMBER = { + b'\xa1\xb2\x3c\x4d': ('big', True), + b'\xa1\xb2\xc3\xd4': ('big', False), + b'\x4d\x3c\xb2\xa1': ('little', True), + b'\xd4\xc3\xb2\xa1': ('little', False), +} # type: Dict[bytes, tuple[Literal['big', 'little'], bool]] + +#: Units per second of a record's ``ts_usec`` field, keyed by the nanosecond +#: flag its file's magic number declares (:data:`MAGIC_NUMBER`): microseconds, +#: or nanoseconds. The one home of these values (:issue:`1739`); the format +#: fixes them, so they are never configurable. +#: +#: .. versionadded:: 1.5.0 +RESOLUTION = { + False: 1_000_000, + True: 1_000_000_000, +} # type: Dict[bool, int] + +#: Decimal places of a ``ts_usec`` fraction at each resolution of +#: :data:`RESOLUTION`, keyed the same way -- six for microseconds, nine for +#: nanoseconds -- and derived from it, so that an epoch rounded to its file's +#: resolution follows the one home (:issue:`1739`). +#: +#: .. versionadded:: 1.5.0 +RESOLUTION_PLACES = { + flag: len(str(units)) - 1 for flag, units in RESOLUTION.items() +} # type: Dict[bool, int] + def magic_number_callback(field: 'NumberField', packet: 'dict[str, Any]') -> 'None': """Calculate byte order of PCAP file. @@ -27,16 +70,14 @@ def magic_number_callback(field: 'NumberField', packet: 'dict[str, Any]') -> 'No """ magic_number = packet['magic_number'] - if magic_number == b'\xd4\xc3\xb2\xa1': - field._byteorder = 'little' - elif magic_number == b'\xa1\xb2\xc3\xd4': - field._byteorder = 'big' - elif magic_number == b'\x4d\x3c\xb2\xa1': - field._byteorder = 'little' - elif magic_number == b'\xa1\xb2\x3c\x4d': - field._byteorder = 'big' - else: + # NOTE: looked up as :obj:`bytes`, so that a :obj:`bytearray` magic number + # is matched as it was compared before :issue:`1739` rather than rejected + # as unhashable. + declared = (MAGIC_NUMBER.get(bytes(magic_number)) + if isinstance(magic_number, (bytes, bytearray, memoryview)) else None) + if declared is None: raise ProtocolError('invalid magic number') + field._byteorder = declared[0] @schema_final diff --git a/pcapkit/protocols/schema/misc/pcapng.py b/pcapkit/protocols/schema/misc/pcapng.py index 3c4f48dd6f..437f6e9423 100644 --- a/pcapkit/protocols/schema/misc/pcapng.py +++ b/pcapkit/protocols/schema/misc/pcapng.py @@ -62,7 +62,7 @@ if TYPE_CHECKING: from ipaddress import IPv4Address, IPv4Interface, IPv6Address, IPv6Interface - from typing import IO, Any, Callable, DefaultDict, Iterable, Optional, Type + from typing import IO, Any, Callable, DefaultDict, Dict, Iterable, Optional, Type from typing_extensions import Literal, Self @@ -167,6 +167,51 @@ class PACKFlags(TypedDict): #: Link-layer-dependent error - symbol error (bit 31). symbol_error: int +#: Byte-Order Magic of a Section Header Block, read in the byte order of the +#: section that wrote it. The one home of this value (:issue:`1739`); the format +#: fixes it, so it is never configurable. +#: +#: .. versionadded:: 1.5.0 +BYTE_ORDER_MAGIC = 0x1A2B3C4D + +#: :data:`BYTE_ORDER_MAGIC` read in the other byte order, i.e. as a big-endian +#: reader sees a little-endian section's. +_SWAPPED_BYTE_ORDER_MAGIC = int.from_bytes(BYTE_ORDER_MAGIC.to_bytes(4, 'little'), 'big') + +#: The octets every block's fixed fields span: Block Type and both copies of +#: Block Total Length, and so the shortest Block Total Length of any block, and +#: the whole of it for a block not in :data:`BLOCK_MINIMUM`. The one home of +#: this value (:issue:`1739`); the format fixes it, so it is never configurable. +#: +#: .. versionadded:: 1.5.0 +GENERIC_BLOCK_MINIMUM = 12 + +#: Block types mapped to the octets their fixed fields span, Block Type and +#: both copies of Block Total Length included; any other block spans +#: :data:`GENERIC_BLOCK_MINIMUM`. The one home of these values (:issue:`1739`); +#: the format fixes them, so they are never configurable. +#: +#: .. versionadded:: 1.5.0 +BLOCK_MINIMUM = { + Enum_BlockType.Section_Header_Block: 28, + Enum_BlockType.Interface_Description_Block: 20, + Enum_BlockType.Packet_Block: 32, + Enum_BlockType.Simple_Packet_Block: 16, + Enum_BlockType.Interface_Statistics_Block: 24, + Enum_BlockType.Enhanced_Packet_Block: 32, + Enum_BlockType.Decryption_Secrets_Block: 20, + Enum_BlockType.Custom_Block_that_rewriters_can_copy_into_new_files: 16, + Enum_BlockType.Custom_Block_that_rewriters_should_not_copy_into_new_files: 16, +} # type: Dict[int, int] + +#: Units per second of a timestamp on an interface that carries no +#: ``if_tsresol`` option: microseconds, the default the format gives that +#: option. The one home of this value (:issue:`1739`); the format fixes it, so +#: it is never configurable. +#: +#: .. versionadded:: 1.5.0 +DEFAULT_RESOLUTION = 1_000_000 + def shb_byteorder_callback(field: 'NumberField', packet: 'dict[str, Any]') -> 'None': """Update byte order of PCAP-NG file for SHB. @@ -192,9 +237,9 @@ def shb_byteorder_callback(field: 'NumberField', packet: 'dict[str, Any]') -> 'N """ magic = packet['match']['byteorder'] # type: int - if magic in (0x1A2B3C4D, 0x0000000C): + if magic in (BYTE_ORDER_MAGIC, 0x0000000C): field._byteorder = 'big' - elif magic in (0x4D3C2B1A, 0x0C000000): + elif magic in (_SWAPPED_BYTE_ORDER_MAGIC, 0x0C000000): field._byteorder = 'little' else: raise ProtocolError(f'unknown byteorder magic: {magic:#x}') @@ -449,8 +494,12 @@ def captured_area(name: 'str') -> 'Callable[[dict[str, Any]], int]': (:issue:`1405`). """ + # NOTE: an EPB and the obsolete Packet Block span the same fixed fields, so + # the one entry stands for both. + framing = BLOCK_MINIMUM[Enum_BlockType.Enhanced_Packet_Block] + def callback(pkt: 'dict[str, Any]') -> 'int': - return max(0, min(pkt[name], pkt['length'] - 32)) + return max(0, min(pkt[name], pkt['length'] - framing)) return callback @@ -785,7 +834,7 @@ class UnknownBlock(BlockType): #: Block total length. length: 'int' = UInt32Field(callback=byteorder_callback) #: Block body (including padding). - body: 'bytes' = BytesField(length=nonnegative(lambda pkt: pkt['length'] - 12)) + body: 'bytes' = BytesField(length=nonnegative(lambda pkt: pkt['length'] - GENERIC_BLOCK_MINIMUM)) #: Block total length. length2: 'int' = UInt32Field(callback=byteorder_callback) @@ -1023,7 +1072,7 @@ class SectionHeaderBlock(BlockType, code=Enum_BlockType.Section_Header_Block): section_length: 'int' = Int64Field(callback=shb_byteorder_callback, default=0xFFFF_FFFF_FFFF_FFFF) #: Options. options: 'list[Option]' = OptionAreaField( - length=nonnegative(lambda pkt: pkt['length'] - 28), + length=nonnegative(lambda pkt: pkt['length'] - BLOCK_MINIMUM[Enum_BlockType.Section_Header_Block]), base_schema=_OPT_Option, type_name='type', registry=Option.registry['opt'], @@ -1055,7 +1104,8 @@ def pre_pack(self, packet: 'dict[str, Any]') -> 'None': return packet['match'] = { - 'byteorder': 0x1A2B3C4D if packet_byteorder(packet) == 'big' else 0x4D3C2B1A, + 'byteorder': (BYTE_ORDER_MAGIC if packet_byteorder(packet) == 'big' + else _SWAPPED_BYTE_ORDER_MAGIC), } def post_process(self, packet: 'dict[str, Any]') -> 'SectionHeaderBlock': @@ -1079,16 +1129,16 @@ def post_process(self, packet: 'dict[str, Any]') -> 'SectionHeaderBlock': Block Total Length (see :func:`shb_byteorder_callback`). """ - if self.length >= 28: + if self.length >= BLOCK_MINIMUM[Enum_BlockType.Section_Header_Block]: self = cast('Self', super().post_process(packet)) if self.section_length == 0xFFFF_FFFF_FFFF_FFFF: self.section_length = -1 magic = packet['match']['byteorder'] # type: int - if magic == 0x1A2B3C4D or (magic == 0x0000000C and self.length == 12): + if magic == BYTE_ORDER_MAGIC or (magic == 0x0000000C and self.length == GENERIC_BLOCK_MINIMUM): self.byteorder = 'big' - elif magic == 0x4D3C2B1A or (magic == 0x0C000000 and self.length == 12): + elif magic == _SWAPPED_BYTE_ORDER_MAGIC or (magic == 0x0C000000 and self.length == GENERIC_BLOCK_MINIMUM): self.byteorder = 'little' else: raise ProtocolError(f'unknown byteorder magic: {magic:#x}') @@ -1356,7 +1406,7 @@ class InterfaceDescriptionBlock(BlockType, code=Enum_BlockType.Interface_Descrip snaplen: 'int' = UInt32Field(default=0, callback=byteorder_callback) #: Options. options: 'list[Option]' = OptionAreaField( - length=nonnegative(lambda pkt: pkt['length'] - 20), + length=nonnegative(lambda pkt: pkt['length'] - BLOCK_MINIMUM[Enum_BlockType.Interface_Description_Block]), base_schema=_IF_Option, type_name='type', registry=Option.registry['if'], @@ -1384,7 +1434,7 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': the block as captured and compares them itself (:issue:`1422`). """ - if self.length < 20: + if self.length < BLOCK_MINIMUM[Enum_BlockType.Interface_Description_Block]: return self return super().post_process(packet) @@ -1512,7 +1562,7 @@ class EnhancedPacketBlock(BlockType, code=Enum_BlockType.Enhanced_Packet_Block): # ``padding_data``: a PaddingField is written straight into the schema # 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'] - 32 + length=bounded_area(lambda pkt: pkt['length'] - BLOCK_MINIMUM[Enum_BlockType.Enhanced_Packet_Block] - (captured_area('captured_len')(pkt) + 3) // 4 * 4), base_schema=_EPB_Option, type_name='type', @@ -1541,7 +1591,7 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': as captured and compares them itself (:issue:`1414`). """ - if self.length < 32: + if self.length < BLOCK_MINIMUM[Enum_BlockType.Enhanced_Packet_Block]: return self return super().post_process(packet) @@ -1568,7 +1618,7 @@ def _spb_data_length(packet: 'dict[str, Any]') -> 'int': read where it sits. """ - area = max(0, packet['length'] - 16) + area = max(0, packet['length'] - BLOCK_MINIMUM[Enum_BlockType.Simple_Packet_Block]) size = min(packet['original_len'], area) snaplen = packet.get('snaplen') if snaplen is not None and snaplen < size: @@ -1780,7 +1830,7 @@ class NameResolutionBlock(BlockType, code=Enum_BlockType.Name_Resolution_Block): length: 'int' = UInt32Field(callback=byteorder_callback) #: Name resolution records. records: 'list[NameResolutionRecord]' = OptionAreaField( - length=nonnegative(lambda pkt: pkt['length'] - 12), + length=nonnegative(lambda pkt: pkt['length'] - GENERIC_BLOCK_MINIMUM), base_schema=NameResolutionRecord, type_name='type', registry=NameResolutionRecord.registry, @@ -1940,7 +1990,7 @@ class InterfaceStatisticsBlock(BlockType, code=Enum_BlockType.Interface_Statisti timestamp_low: 'int' = UInt32Field(callback=byteorder_callback) #: Options. options: 'list[Option]' = OptionAreaField( - length=nonnegative(lambda pkt: pkt['length'] - 24), + length=nonnegative(lambda pkt: pkt['length'] - BLOCK_MINIMUM[Enum_BlockType.Interface_Statistics_Block]), base_schema=_ISB_Option, type_name='type', registry=Option.registry['isb'], @@ -1965,7 +2015,7 @@ class SystemdJournalExportBlock(BlockType, code=Enum_BlockType.systemd_Journal_E #: Block total length. length: 'int' = UInt32Field(callback=byteorder_callback) #: Journal entry. - entry: 'bytes' = BytesField(length=nonnegative(lambda pkt: pkt['length'] - 12)) + entry: 'bytes' = BytesField(length=nonnegative(lambda pkt: pkt['length'] - GENERIC_BLOCK_MINIMUM)) #: Block total length. length2: 'int' = UInt32Field(callback=byteorder_callback) @@ -2322,8 +2372,8 @@ class DecryptionSecretsBlock(BlockType, code=Enum_BlockType.Decryption_Secrets_B 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'] - 20 - pkt['secrets_length'] - - (4 - pkt['secrets_length'] % 4) % 4), + length=nonnegative(lambda pkt: pkt['length'] - BLOCK_MINIMUM[Enum_BlockType.Decryption_Secrets_Block] + - pkt['secrets_length'] - (4 - pkt['secrets_length'] % 4) % 4), base_schema=_DSB_Option, type_name='type', registry=Option.registry['dsb'], @@ -2360,7 +2410,10 @@ class CustomBlock(BlockType, code=[Enum_BlockType.Custom_Block_that_rewriters_ca #: Private enterprise number. pen: 'int' = UInt32Field(callback=byteorder_callback) #: Custom data (incl. padding and options). - data: 'bytes' = BytesField(length=nonnegative(lambda pkt: pkt['length'] - 16)) + # NOTE: both custom block types span the same fixed fields, so the one + # entry stands for both. + data: 'bytes' = BytesField(length=nonnegative( + lambda pkt: pkt['length'] - BLOCK_MINIMUM[Enum_BlockType.Custom_Block_that_rewriters_can_copy_into_new_files])) #: Block total length. length2: 'int' = UInt32Field(callback=byteorder_callback) @@ -2433,7 +2486,7 @@ class PacketBlock(BlockType, code=Enum_BlockType.Packet_Block): 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'] - 32 + length=bounded_area(lambda pkt: pkt['length'] - BLOCK_MINIMUM[Enum_BlockType.Packet_Block] - (captured_area('captured_length')(pkt) + 3) // 4 * 4), base_schema=_PACK_Option, type_name='type', @@ -2462,7 +2515,7 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': as captured and compares them itself (:issue:`1414`). """ - if self.length < 32: + if self.length < BLOCK_MINIMUM[Enum_BlockType.Packet_Block]: return self return super().post_process(packet) diff --git a/pcapkit/protocols/schema/transport/tcp.py b/pcapkit/protocols/schema/transport/tcp.py index 2a89a227fa..e440ffdf4e 100644 --- a/pcapkit/protocols/schema/transport/tcp.py +++ b/pcapkit/protocols/schema/transport/tcp.py @@ -46,6 +46,22 @@ from pcapkit.corekit.fields.field import FieldBase as Field from pcapkit.protocols.protocol import ProtocolBase +#: Minimum length of a TCP header, i.e. the fixed part with no options, +#: :rfc:`9293#section-3.1`. The one home of this value (:issue:`1739`), here so +#: that the schema's option length and the protocol modules read the same one; +#: it is fixed by the RFC, so it is never configurable. +#: +#: .. versionadded:: 1.5.0 +TCP_MIN_HEADER_LEN = 20 + +#: Offset of the octet whose high four bits are the Data Offset field -- the +#: header length in 32-bit words -- within a TCP header, :rfc:`9293#section-3.1`. +#: The one home of this value (:issue:`1739`), beside :data:`TCP_MIN_HEADER_LEN`; +#: it is fixed by the RFC, so it is never configurable. +#: +#: .. versionadded:: 1.5.0 +TCP_DATA_OFFSET_OCTET = 12 + if SPHINX_TYPE_CHECKING: # pragma: no cover from typing_extensions import TypedDict @@ -1093,7 +1109,7 @@ class TCP(Schema): urgent: 'int' = UInt16Field() #: Options. options: 'list[Option]' = OptionField( - length=lambda pkt: pkt['offset']['offset'] * 4 - 20, + length=lambda pkt: pkt['offset']['offset'] * 4 - TCP_MIN_HEADER_LEN, base_schema=Option, type_name='kind', registry=Option.registry, diff --git a/pcapkit/protocols/transport/tcp.py b/pcapkit/protocols/transport/tcp.py index 6e0345a56c..795db663eb 100644 --- a/pcapkit/protocols/transport/tcp.py +++ b/pcapkit/protocols/transport/tcp.py @@ -97,6 +97,7 @@ from pcapkit.protocols.schema.transport.tcp import MPTCPDSS as Schema_MPTCPDSS from pcapkit.protocols.schema.transport.tcp import SACK as Schema_SACK from pcapkit.protocols.schema.transport.tcp import TCP as Schema_TCP +from pcapkit.protocols.schema.transport.tcp import TCP_MIN_HEADER_LEN from pcapkit.protocols.schema.transport.tcp import \ AlternateChecksumData as Schema_AlternateChecksumData from pcapkit.protocols.schema.transport.tcp import \ @@ -519,7 +520,7 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_TCP': raise ProtocolError(f'{self.alias}: header length {tcp.hdr_len} runs past the end of ' f'the data ({len(self)} octets)') - _optl = tcp.hdr_len - 20 + _optl = tcp.hdr_len - TCP_MIN_HEADER_LEN if _optl: # NOTE: the octets after an End of Option List option are kept as # read, so that :meth:`make` rebuilds the option area -- and the @@ -629,7 +630,7 @@ def make(self, else: options_value, total_length = [], 0 - offset = math.ceil((20 + total_length) / 4) + offset = math.ceil((TCP_MIN_HEADER_LEN + total_length) / 4) return Schema_TCP( srcport=self._make_port(srcport, Enum_TransportProtocol.tcp), @@ -701,7 +702,7 @@ def register_mp_option(cls, code: 'Enum_MPTCPOption', meth: 'str | tuple[MPOptio def __length_hint__(self) -> 'Literal[20]': """Return an estimated length for the object.""" - return 20 + return TCP_MIN_HEADER_LEN # type: ignore[return-value] @classmethod def __index__(cls) -> 'TransType': # pylint: disable=invalid-index-returned diff --git a/pcapkit/toolkit/dpkt.py b/pcapkit/toolkit/dpkt.py index e1e6aafd9e..cf77ad4155 100644 --- a/pcapkit/toolkit/dpkt.py +++ b/pcapkit/toolkit/dpkt.py @@ -23,6 +23,7 @@ from pcapkit.foundation.traceflow.data.data import FrameRecord from pcapkit.foundation.traceflow.data.tcp import Packet as TF_TCP_Packet from pcapkit.protocols.data.misc.pcap.frame import FrameInfo as Data_FrameInfo +from pcapkit.protocols.schema.misc.pcap.header import RESOLUTION, RESOLUTION_PLACES from pcapkit.utilities.exceptions import UnsupportedCall if TYPE_CHECKING: @@ -139,7 +140,7 @@ class Timestamp(RecordTimestamp, float): __slots__ = ('exact', 'resolution', 'orig_len') def __new__(cls, value: 'Any' = 0.0, exact: 'Optional[Decimal]' = None, - resolution: 'int' = 1_000_000, orig_len: 'Optional[int]' = None) -> 'Timestamp': + resolution: 'int' = RESOLUTION[False], orig_len: 'Optional[int]' = None) -> 'Timestamp': self = super().__new__(cls, value) self.exact = decimal.Decimal(float(self)) if exact is None else exact self.resolution = resolution @@ -166,7 +167,7 @@ class DecimalTimestamp(RecordTimestamp, decimal.Decimal): __slots__ = ('exact', 'resolution', 'orig_len') - def __new__(cls, value: 'Any' = '0', resolution: 'int' = 1_000_000_000, + def __new__(cls, value: 'Any' = '0', resolution: 'int' = RESOLUTION[True], orig_len: 'Optional[int]' = None) -> 'DecimalTimestamp': self = super().__new__(cls, value) self.exact = decimal.Decimal(self) @@ -312,11 +313,12 @@ def packet2frame(packet: 'Packet', timestamp: 'float | Decimal', *, if isinstance(stamp, RecordTimestamp): epoch, resolution, orig_len = stamp.exact, stamp.resolution, stamp.orig_len elif isinstance(stamp, decimal.Decimal): - epoch, resolution, orig_len = stamp, 1_000_000_000, None + epoch, resolution, orig_len = stamp, RESOLUTION[True], None else: - epoch, resolution, orig_len = round(decimal.Decimal(stamp), 6), 1_000_000, None + epoch, resolution, orig_len = (round(decimal.Decimal(stamp), RESOLUTION_PLACES[False]), + RESOLUTION[False], None) ts_sec = int(epoch) - ts_usec = int((epoch - ts_sec) * (1_000_000_000 if resolution > 1_000_000 else 1_000_000)) + ts_usec = int((epoch - ts_sec) * RESOLUTION[resolution > RESOLUTION[False]]) # the mapping is exactly the one reported before #1507, plain value types # included, so that the dict-capable trace formats write the same bytes diff --git a/pcapkit/toolkit/pcap.py b/pcapkit/toolkit/pcap.py index 400ecf6d60..77ca8dc91a 100644 --- a/pcapkit/toolkit/pcap.py +++ b/pcapkit/toolkit/pcap.py @@ -24,6 +24,7 @@ from pcapkit.protocols.misc.null import NoPayload from pcapkit.protocols.misc.raw import Raw from pcapkit.protocols.protocol import ProtocolBase +from pcapkit.protocols.schema.transport.tcp import TCP_DATA_OFFSET_OCTET, TCP_MIN_HEADER_LEN from pcapkit.protocols.transport.tcp import TCP if TYPE_CHECKING: @@ -174,8 +175,8 @@ def tcp_segment(frame: 'ProtocolBase') -> 'Optional[TCPSegment]': return None octets = bytes(raw.packet.payload) - hdr_len = (octets[12] >> 4) * 4 if len(octets) >= 20 else 0 - if hdr_len < 20: + hdr_len = (octets[TCP_DATA_OFFSET_OCTET] >> 4) * 4 if len(octets) >= TCP_MIN_HEADER_LEN else 0 + if hdr_len < TCP_MIN_HEADER_LEN: return None flags = octets[13] return TCPSegment( diff --git a/pcapkit/toolkit/pcapng.py b/pcapkit/toolkit/pcapng.py index 6aedcfeb22..25f4411848 100644 --- a/pcapkit/toolkit/pcapng.py +++ b/pcapkit/toolkit/pcapng.py @@ -18,6 +18,7 @@ from pcapkit.foundation.traceflow.data.tcp import Packet as TF_TCP_Packet from pcapkit.protocols.data.misc.pcap.frame import Frame as Data_Frame from pcapkit.protocols.data.misc.pcap.frame import FrameInfo as Data_FrameInfo +from pcapkit.protocols.schema.misc.pcap.header import RESOLUTION from pcapkit.toolkit.pcap import tcp_segment if TYPE_CHECKING: @@ -249,7 +250,7 @@ def block2frame(block: 'Packet', *, nanosecond: 'bool' = False) -> 'Data_Frame': """ ts_sec = int(block.timestamp_epoch) - ts_usec = int((block.timestamp_epoch - ts_sec) * (1_000_000_000 if nanosecond else 1_000_000)) + ts_usec = int((block.timestamp_epoch - ts_sec) * RESOLUTION[bool(nanosecond)]) frame = Data_Frame( frame_info=Data_FrameInfo( diff --git a/pcapkit/toolkit/pypcapfile.py b/pcapkit/toolkit/pypcapfile.py index 3c95b58a51..1356f32636 100644 --- a/pcapkit/toolkit/pypcapfile.py +++ b/pcapkit/toolkit/pypcapfile.py @@ -61,7 +61,12 @@ from pcapkit.foundation.reassembly.data.ip import Packet as IP_Packet from pcapkit.foundation.reassembly.data.tcp import Packet as TCP_Packet from pcapkit.foundation.traceflow.data.tcp import Packet as TF_TCP_Packet +from pcapkit.protocols.internet import ipv6 as Module_IPv6 from pcapkit.protocols.internet.ipv4 import IPv4 as Protocol_IPv4 +from pcapkit.protocols.schema.internet import ah as Schema_AH_Module +from pcapkit.protocols.schema.internet import ipv4 as Schema_IPv4_Module +from pcapkit.protocols.schema.misc.pcap.header import RESOLUTION +from pcapkit.protocols.schema.transport import tcp as Schema_TCP_Module from pcapkit.protocols.transport.tcp import TCP as Protocol_TCP from pcapkit.utilities import config from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall, stacklevel @@ -87,21 +92,12 @@ #: :meth:`pcapkit.protocols.link.ethernet.Ethernet._read_mac_addr`. py38 = ((version_info := sys.version_info).major >= 3 and version_info.minor >= 8) -#: Minimum length of a TCP header, i.e. the fixed part with no options. -TCP_MIN_HEADER_LEN = 20 - #: IPv4 **DF** (don't fragment) bit, within the three-bit flags field. IPV4_FLAG_DF = 0b010 #: IPv4 **MF** (more fragments) bit, within the three-bit flags field. IPV4_FLAG_MF = 0b001 -#: Length of the fixed IPv4 header, i.e. with no options, :rfc:`791#section-3.1`. -IPV4_HEADER_LEN = 20 - -#: Length of the fixed IPv6 header, :rfc:`8200#section-3`. -IPV6_HEADER_LEN = 40 - #: The captures :func:`_decline_ipv6_tcp` has warned about, oldest first. Each #: key :func:`_capture_of` gives maps to the anchor that keeps that key unique. _ipv6_tcp_warned = collections.OrderedDict() # type: OrderedDict[int, Any] @@ -109,14 +105,10 @@ # NOTE: the definitions from here to ``packet2timestamp``, and # ``_default_accepts``, ``_default_extension_header`` and # ``_walk_extension_headers`` below, are kept for :mod:`pcapkit.toolkit.scapy`: -# it imports the rest, and ``_walk_extension_headers`` uses ``AH_HEADER_LEN`` -# and ``_default_extension_header``. This module's adapters use none of them +# it imports the rest, and ``_walk_extension_headers`` uses +# ``_default_extension_header``. This module's adapters use none of them # (#1708). -#: Length of the fixed part of an IP Authentication Header (AH), before its -#: Integrity Check Value, :rfc:`4302#section-2`. -AH_HEADER_LEN = 12 - #: IP protocol numbers of the headers :func:`_walk_extension_headers` steps #: over inside IPv4: AH and the IPv6 Hop-by-Hop Options, Routing, Fragment, #: Destination Options, Mobility and HIP headers. @@ -156,7 +148,7 @@ def packet2timestamp(packet: 'Packet') -> 'float': .. _PyPCAPFile: https://github.com/kisom/pypcapfile """ - divisor = 1_000_000_000 if packet.header[0].ns_resolution else 1_000_000 + divisor = RESOLUTION[bool(packet.header[0].ns_resolution)] return (packet.timestamp * divisor + packet.timestamp_us) / divisor @@ -355,8 +347,8 @@ def _network(packet: 'Packet') -> 'Optional[IP]': payload = getattr(link, 'payload', None) # type: Any if type(payload).__name__ != 'IP': return None - if payload.hl * 4 > IPV4_HEADER_LEN and ( - IPV4_HEADER_LEN + len(_maybe_unhex(bytes(payload.opt))) < payload.hl * 4): + if payload.hl * 4 > Schema_IPv4_Module.IPV4_HEADER_LEN and ( + Schema_IPv4_Module.IPV4_HEADER_LEN + len(_maybe_unhex(bytes(payload.opt))) < payload.hl * 4): return None return payload @@ -388,7 +380,8 @@ def _ipv6_carries_tcp(packet: 'Packet') -> 'bool': return False data = _maybe_unhex(bytes(payload)) - return len(data) >= IPV6_HEADER_LEN and data[6] == Enum_TransType.TCP + return (len(data) >= Module_IPv6.IPV6_HEADER_LEN + and data[Module_IPv6.IPV6_NEXT_HEADER_OFFSET] == Enum_TransType.TCP) def _capture_of(packet: 'Packet') -> 'tuple[int, Any]': @@ -486,7 +479,8 @@ def _transport(ipv4: 'IP') -> 'Optional[bytes]': segment = _maybe_unhex(bytes(segment)) if ipv4.len: segment = segment[:max(ipv4.len - ipv4.hl * 4, 0)] - if len(segment) < TCP_MIN_HEADER_LEN or segment[12] >> 4 < 5: # Data Offset, in words + if (len(segment) < Schema_TCP_Module.TCP_MIN_HEADER_LEN + or (segment[Schema_TCP_Module.TCP_DATA_OFFSET_OCTET] >> 4) * 4 < Schema_TCP_Module.TCP_MIN_HEADER_LEN): return None return segment @@ -558,7 +552,8 @@ def _walk_extension_headers(protocol: 'int', data: 'bytes') -> 'Optional[tuple[i is cut short or rejected, or more than :data:`~pcapkit.utilities.config.EXTENSION_HEADER_LIMIT` follow in a row. An AH is ``(Payload Length + 2) * 4`` octets long - (:rfc:`4302#section-2.2`), and no shorter than :data:`AH_HEADER_LEN`; + (:rfc:`4302#section-2.2`), and no shorter than + :data:`~pcapkit.protocols.schema.internet.ah.AH_HEADER_LEN`; any other is read by :func:`_default_extension_header`. """ @@ -568,8 +563,8 @@ def _walk_extension_headers(protocol: 'int', data: 'bytes') -> 'Optional[tuple[i return None depth += 1 if protocol == Enum_TransType.AH: - length = (data[1] + 2) * 4 if len(data) >= AH_HEADER_LEN else 0 - if length < AH_HEADER_LEN or len(data) < length: + length = (data[1] + 2) * 4 if len(data) >= Schema_AH_Module.AH_HEADER_LEN else 0 + if length < Schema_AH_Module.AH_HEADER_LEN or len(data) < length: return None protocol, data = data[0], data[length:] continue @@ -846,7 +841,7 @@ def tcp_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'TCP_Packet | None from pcapfile.protocols.transport.tcp import TCP # isort:skip tcp = TCP(segment) - hdr_len = tcp.data_offset # at least TCP_MIN_HEADER_LEN, see ``_transport`` + hdr_len = tcp.data_offset # at least ``TCP_MIN_HEADER_LEN``, see ``_transport`` payload = segment[hdr_len:] return TCP_Packet( @@ -920,7 +915,7 @@ def tcp_traceflow(packet: 'Packet', *, data_link: 'Enum_LinkType', from pcapfile.protocols.transport.tcp import TCP # isort:skip tcp = TCP(segment) - hdr_len = tcp.data_offset # at least TCP_MIN_HEADER_LEN, see ``_transport`` + hdr_len = tcp.data_offset # at least ``TCP_MIN_HEADER_LEN``, see ``_transport`` return TF_TCP_Packet( # type: ignore[type-var] protocol=data_link, # data link type from savefile header index=count, # frame number diff --git a/pcapkit/toolkit/scapy.py b/pcapkit/toolkit/scapy.py index a70e9e13ee..6b6e279398 100644 --- a/pcapkit/toolkit/scapy.py +++ b/pcapkit/toolkit/scapy.py @@ -46,9 +46,12 @@ from pcapkit.foundation.traceflow.data.data import FrameRecord from pcapkit.foundation.traceflow.data.tcp import Packet as TF_TCP_Packet from pcapkit.protocols.data.misc.pcap.frame import FrameInfo as Data_FrameInfo +from pcapkit.protocols.internet.ipv6 import IPV6_HEADER_LEN +from pcapkit.protocols.schema.internet.ipv4 import IPV4_HEADER_LEN from pcapkit.protocols.schema.internet.ipv6 import jumbo_payload_length -from pcapkit.toolkit.pypcapfile import (IP_TUNNELS, IPV4_EXTENSION_HEADERS, IPV4_HEADER_LEN, - IPV6_HEADER_LEN, TCP_MIN_HEADER_LEN, _default_accepts, +from pcapkit.protocols.schema.misc.pcap.header import RESOLUTION, RESOLUTION_PLACES +from pcapkit.protocols.schema.transport.tcp import TCP_DATA_OFFSET_OCTET, TCP_MIN_HEADER_LEN +from pcapkit.toolkit.pypcapfile import (IP_TUNNELS, IPV4_EXTENSION_HEADERS, _default_accepts, _walk_extension_headers) from pcapkit.utilities.compat import ModuleNotFoundError # pylint: disable=redefined-builtin from pcapkit.utilities.exceptions import MissingKeyError, ModuleNotFound, stacklevel @@ -218,12 +221,12 @@ def packet2frame(packet: 'Packet') -> 'FrameRecord': wirelen = getattr(packet, 'wirelen', None) orig_len = len(octets) if wirelen is None else int(wirelen) - nanosecond = getattr(packet, RESOLUTION_ATTR, 1_000_000) > 1_000_000 + nanosecond = getattr(packet, RESOLUTION_ATTR, RESOLUTION[False]) > RESOLUTION[False] stamp = packet.time epoch = (decimal.Decimal(stamp) if isinstance(stamp, decimal.Decimal) - else round(decimal.Decimal(stamp), 9 if nanosecond else 6)) + else round(decimal.Decimal(stamp), RESOLUTION_PLACES[nanosecond])) ts_sec = int(epoch) - ts_usec = int((epoch - ts_sec) * (1_000_000_000 if nanosecond else 1_000_000)) + ts_usec = int((epoch - ts_sec) * RESOLUTION[nanosecond]) return FrameRecord(mapping, packet=octets, frame_info=Data_FrameInfo( ts_sec=ts_sec, @@ -336,7 +339,7 @@ def _tcp_segment(tcp: 'TCP') -> 'tuple[bytes, bytes] | None': """ segment = _within_datagrams(tcp) - hdr_len = (segment[12] >> 4) * 4 if len(segment) >= TCP_MIN_HEADER_LEN else 0 + hdr_len = (segment[TCP_DATA_OFFSET_OCTET] >> 4) * 4 if len(segment) >= TCP_MIN_HEADER_LEN else 0 if hdr_len < TCP_MIN_HEADER_LEN: return None return segment[:hdr_len], segment[hdr_len:] @@ -352,7 +355,7 @@ def _tcp_behind_extension_headers(segment: 'bytes') -> 'tuple[bytes, bytes] | No Returns: The header, as many octets as the Data Offset counts, and the payload after it; or :data:`None` if there are fewer than - :data:`TCP_MIN_HEADER_LEN ` + :data:`TCP_MIN_HEADER_LEN ` octets, the Data Offset counts fewer or more than were captured, or pcapkit's own TCP parser rejects the options (see :func:`~pcapkit.toolkit.pypcapfile._default_accepts`). @@ -363,7 +366,7 @@ def _tcp_behind_extension_headers(segment: 'bytes') -> 'tuple[bytes, bytes] | No pcapkit's parser to check. """ - hdr_len = (segment[12] >> 4) * 4 if len(segment) >= TCP_MIN_HEADER_LEN else 0 + hdr_len = (segment[TCP_DATA_OFFSET_OCTET] >> 4) * 4 if len(segment) >= TCP_MIN_HEADER_LEN else 0 if hdr_len < TCP_MIN_HEADER_LEN or hdr_len > len(segment): return None if hdr_len > TCP_MIN_HEADER_LEN and not _default_accepts('TCP', segment[:hdr_len]): diff --git a/tests/foundation/engines/test_pcapng_read_ahead_byte_order_1739_unit.py b/tests/foundation/engines/test_pcapng_read_ahead_byte_order_1739_unit.py new file mode 100644 index 0000000000..898a7741af --- /dev/null +++ b/tests/foundation/engines/test_pcapng_read_ahead_byte_order_1739_unit.py @@ -0,0 +1,70 @@ +# -*- coding: utf-8 -*- +"""The PCAP-NG engine sizes a Section Header Block's read-ahead in the order it declares. C.f. #1739. + +:meth:`PCAPNG._bound_read_ahead ` +reads a Section Header Block's Block Total Length in the byte order its +Byte-Order Magic declares, mapped by ``_BYTE_ORDER_OCTETS`` from the magic's one +home, :data:`pcapkit.protocols.schema.misc.pcapng.BYTE_ORDER_MAGIC`. Read in +the wrong order, the length of a 28-octet section header is ``0x1C000000``, so +the reader would read ahead hundreds of megabytes instead of to the block's end. + +Both byte orders are checked, as no capture in the test suite is a big-endian +PCAP-NG file read through this probe. 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 + + +def _shb(byteorder: 'str', length: 'int' = 28) -> 'bytes': + """A Section Header Block with no options, in ``byteorder``.""" + end = '>' if byteorder == 'big' else '<' + return (struct.pack(f'{end}IIIHHq', 0x0A0D0D0A, length, 0x1A2B3C4D, 1, 0, -1) + + struct.pack(f'{end}I', length)) + + +class TestReadAheadByteOrder(unittest.TestCase): + """A Section Header Block is measured in its own byte order.""" + + def _read_ahead(self, data: 'bytes') -> '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=16, 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 = None # 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_big_endian(self) -> 'None': + data = _shb('big') + self.assertEqual(data[8:12], b'\x1a\x2b\x3c\x4d') + self.assertEqual(self._read_ahead(data), 28) + + def test_little_endian(self) -> 'None': + data = _shb('little') + self.assertEqual(data[8:12], b'\x4d\x3c\x2b\x1a') + self.assertEqual(self._read_ahead(data), 28) + + def test_longer_block_both_orders(self) -> 'None': + for byteorder in ('big', 'little'): + with self.subTest(byteorder=byteorder): + data = _shb(byteorder, 40) + self.assertEqual(self._read_ahead(data), 40) + + def test_unknown_magic_keeps_the_buffer(self) -> 'None': + data = bytearray(_shb('big')) + data[8:12] = b'\x00\x11\x22\x33' + self.assertEqual(self._read_ahead(bytes(data)), 16) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/foundation/reassembly/test_ip_max_length_edge_1739_unit.py b/tests/foundation/reassembly/test_ip_max_length_edge_1739_unit.py new file mode 100644 index 0000000000..6d7fd6b129 --- /dev/null +++ b/tests/foundation/reassembly/test_ip_max_length_edge_1739_unit.py @@ -0,0 +1,81 @@ +# -*- coding: utf-8 -*- +"""IP reassembly reads :data:`IP_MAX_LENGTH` as a strict bound at its edge. C.f. #1739. + +:class:`pcapkit.foundation.reassembly.ip.IP` reads +:data:`~pcapkit.protocols.schema.internet.ipv6.IP_MAX_LENGTH` from its home. A +fragment with Total Length 0 declares no length (#1555), and grows the buffer +past the bound only when its datagram, header included, is *longer* than +65,535 octets, since growing the buffer is what lifts the bound (#1585). One of +exactly 65,535 octets fits, so it leaves the bound in place, and a declared +final fragment that then runs past it still never completes. + +Every case builds its fragments in memory and reads no capture. Classes are +imported inside each test, after :func:`~tests._support.reimport_once_per_class`. + +""" + +from ipaddress import ip_address +import unittest + +from tests._support import reimport_once_per_class + +#: Datagram octets, with no zero octet in them, so a hole shows. +DATA = (bytes(range(1, 256)) * 258)[:65536] + +#: Octets of the IPv4 header in front of every fragment. +IHL = 20 + +#: Fragment Offset of the final fragment, the start of the block a cut +#: fragment ending at octet 65,515 leaves partial. +FINAL = 65512 + + +class TestIPMaxLengthEdge(unittest.TestCase): + """A Total Length 0 datagram of exactly 65,535 octets does not lift the bound.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _packet(self, *, num: int, fo: int, mf: bool, payload: bytes, tl: 'int | None' = None): + from pcapkit.const.reg.transtype import TransType + from pcapkit.foundation.reassembly.data.ip import Packet + + bufid = (ip_address('192.0.2.1'), ip_address('198.51.100.2'), 7, TransType.UDP) + return Packet(bufid, num, fo, IHL, mf, IHL + len(payload) if tl is None else tl, + b'H' * IHL, bytearray(payload), 1000.0 + num) + + def _completed(self, *, first: int, strict: bool): + """Reassemble a Total Length 0 fragment of ``first`` octets, then a declared final one. + + The final fragment declares its data to end at octet 65,536, one past + what a 20-octet header leaves room for. + + """ + from pcapkit.foundation.reassembly.ipv4 import IPv4 + + reasm = IPv4(strict=strict) + reasm(self._packet(num=1, fo=0, mf=True, payload=DATA[:first], tl=0)) + reasm(self._packet(num=2, fo=FINAL, mf=False, payload=DATA[FINAL:])) + datagram, = reasm.datagram + self.assertEqual(datagram.index, (1, 2)) + return datagram.completed + + def test_exactly_the_bound_keeps_it(self) -> None: + from pcapkit.foundation.reassembly.data.data import Completion + + for strict in (True, False): + with self.subTest(strict=strict): + # 20 + 65,515 is 65,535 octets: the buffer is not grown + self.assertIs(self._completed(first=65535 - IHL, strict=strict), Completion.PARTIAL) + + def test_one_past_the_bound_lifts_it(self) -> None: + from pcapkit.foundation.reassembly.data.data import Completion + + for strict in (True, False): + with self.subTest(strict=strict): + # 20 + 65,516 is 65,536 octets: the buffer grows past the bound + self.assertIs(self._completed(first=65536 - IHL, strict=strict), Completion.COMPLETE) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/project/test_rfc_constants_one_home_1739_unit.py b/tests/project/test_rfc_constants_one_home_1739_unit.py new file mode 100644 index 0000000000..044478737c --- /dev/null +++ b/tests/project/test_rfc_constants_one_home_1739_unit.py @@ -0,0 +1,725 @@ +# -*- coding: utf-8 -*- +"""Each duplicated RFC-fixed constant has one home, and its readers import it. + +Ruling 7 on GitHub issue #1726, tracked as #1739: a protocol constant that the +RFC (or the file format) fixes, and that more than one module spelled out, is +defined once, in the module that owns the protocol -- or that protocol's schema +or :mod:`pcapkit.const` module -- and every reader imports it from there. Such a +constant is never configurable, and the old homes do not re-export it. + +Four checks pin that: + +* :class:`TestHomes` -- each constant is defined in its home, with the value the + RFC or the format gives it; +* :class:`TestOneDefinition` -- no other module under :mod:`pcapkit` assigns the + name, at module or class level, and none spells the distinctive literals (the + PCAP and PCAP-NG magic numbers and the PCAP timestamp resolutions) itself; +* :class:`TestReaders` -- each reader imports the constant from its home and + reads it, and :mod:`pcapkit.toolkit.pypcapfile`, which defined four of them, + no longer binds those names at all; +* :class:`TestNotReinlined` -- no module in :data:`SWEPT`, the homes included, + spells a homed value anywhere but in the home's own definition: not as an + operand, a call argument, a subscript, a default, a returned value, a tuple + or table element, a branch of a conditional, nor an assignment at any level. + Each match that means something else is listed in :data:`NOT_A_COPY` or + :data:`UNRELATED` with the reason, and the values that are equal but not the + same constant in :data:`DISTINCT`. + +The source checks read each module with :mod:`ast` rather than importing it. +:mod:`pcapkit.vendor` and :mod:`pcapkit.const` are generated and are not read. + +This module is unit-tier and reads no capture. + +""" +from __future__ import annotations + +import ast +import importlib +import pathlib +import sys +import unittest + +#: Repository root. +ROOT = pathlib.Path(__file__).resolve().parents[2] + +#: The package directory. +PACKAGE = ROOT / 'pcapkit' + +#: Each consolidated constant: its home module, its name and its value. +HOMES = ( + ('pcapkit.protocols.schema.internet.ipv4', 'IPV4_HEADER_LEN', 20), + ('pcapkit.protocols.internet.ipv6', 'IPV6_HEADER_LEN', 40), + ('pcapkit.protocols.internet.ipv6', 'IPV6_NEXT_HEADER_OFFSET', 6), + ('pcapkit.protocols.schema.internet.ipv6', 'IP_MAX_LENGTH', 65535), + ('pcapkit.protocols.schema.internet.ah', 'AH_HEADER_LEN', 12), + ('pcapkit.protocols.schema.transport.tcp', 'TCP_MIN_HEADER_LEN', 20), + ('pcapkit.protocols.schema.transport.tcp', 'TCP_DATA_OFFSET_OCTET', 12), + ('pcapkit.protocols.schema.misc.pcap.header', 'GLOBAL_HEADER_LEN', 24), + ('pcapkit.protocols.schema.misc.pcap.frame', 'RECORD_HEADER_LEN', 16), + ('pcapkit.protocols.schema.misc.pcap.header', 'MAGIC_NUMBER', { + b'\xa1\xb2\x3c\x4d': ('big', True), + b'\xa1\xb2\xc3\xd4': ('big', False), + b'\x4d\x3c\xb2\xa1': ('little', True), + b'\xd4\xc3\xb2\xa1': ('little', False), + }), + ('pcapkit.protocols.schema.misc.pcap.header', 'RESOLUTION', {False: 10 ** 6, True: 10 ** 9}), + ('pcapkit.protocols.schema.misc.pcap.header', 'RESOLUTION_PLACES', {False: 6, True: 9}), + ('pcapkit.protocols.schema.misc.pcapng', 'BYTE_ORDER_MAGIC', 0x1A2B3C4D), + ('pcapkit.protocols.schema.misc.pcapng', 'DEFAULT_RESOLUTION', 10 ** 6), + ('pcapkit.protocols.schema.misc.pcapng', 'GENERIC_BLOCK_MINIMUM', 12), + ('pcapkit.protocols.schema.misc.pcapng', 'BLOCK_MINIMUM', { + 0x0A0D0D0A: 28, 0x00000001: 20, 0x00000002: 32, 0x00000003: 16, 0x00000005: 24, + 0x00000006: 32, 0x0000000A: 20, 0x00000BAD: 16, 0x40000BAD: 16, + }), +) + +#: Names that used to be defined elsewhere and now must not be, mapped to the +#: module that held them. ``MAGIC_NUMBER`` stays a class attribute of the PCAP +#: and PCAP-NG engines -- the extractor dispatches on it -- but is built from +#: the home rather than spelled out, which :class:`TestOneDefinition` checks +#: through the literals. +OLD_NAMES = ( + ('pcapkit/toolkit/pypcapfile.py', 'TCP_MIN_HEADER_LEN'), + ('pcapkit/toolkit/pypcapfile.py', 'IPV4_HEADER_LEN'), + ('pcapkit/toolkit/pypcapfile.py', 'IPV6_HEADER_LEN'), + ('pcapkit/toolkit/pypcapfile.py', 'AH_HEADER_LEN'), + ('pcapkit/foundation/reassembly/ip.py', '_MAX_LENGTH'), + ('pcapkit/foundation/reassembly/ipv6.py', '_IPV6_HDR_LEN'), + ('pcapkit/foundation/reassembly/ipv6.py', '_IPV6_NEXT_HEADER'), + ('pcapkit/dumpkit/pcap.py', '_RESOLUTION'), + ('pcapkit/protocols/misc/pcap/header.py', '_MAGIC_NUM'), + ('pcapkit/foundation/engines/pcapng.py', 'BLOCK_MINIMUM'), + ('pcapkit/foundation/engines/pcapng.py', 'BYTE_ORDER_MAGIC'), + ('pcapkit/protocols/internet/ip.py', 'IP_MAX_LENGTH'), + ('pcapkit/protocols/internet/ipv4.py', 'IPV4_HEADER_LEN'), + ('pcapkit/protocols/internet/ah.py', 'AH_HEADER_LEN'), + ('pcapkit/protocols/transport/tcp.py', 'TCP_MIN_HEADER_LEN'), + ('pcapkit/protocols/transport/tcp.py', 'TCP_DATA_OFFSET_OCTET'), +) + +#: Literals distinctive enough to find by value, mapped to the one file allowed +#: to spell them. ``1_000_000_000`` is also Python's nanoseconds per second, so a +#: division of :func:`time.time_ns` by it is exempt (see :func:`_time_ns_divisor`). +LITERALS = { + b'\xa1\xb2\x3c\x4d': 'pcapkit/protocols/schema/misc/pcap/header.py', + b'\xa1\xb2\xc3\xd4': 'pcapkit/protocols/schema/misc/pcap/header.py', + b'\x4d\x3c\xb2\xa1': 'pcapkit/protocols/schema/misc/pcap/header.py', + b'\xd4\xc3\xb2\xa1': 'pcapkit/protocols/schema/misc/pcap/header.py', + b'\x0a\x0d\x0d\x0a': None, + b'\x1a\x2b\x3c\x4d': None, + b'\x4d\x3c\x2b\x1a': None, + 0x1A2B3C4D: 'pcapkit/protocols/schema/misc/pcapng.py', + 0x4D3C2B1A: None, + 10 ** 9: 'pcapkit/protocols/schema/misc/pcap/header.py', + 10 ** 6: ('pcapkit/protocols/schema/misc/pcap/header.py', 'pcapkit/protocols/schema/misc/pcapng.py'), +} + +#: Each reader module, the home it imports from, and the names it imports. +READERS = ( + ('pcapkit/toolkit/scapy.py', 'pcapkit.protocols.schema.internet.ipv4', 'IPV4_HEADER_LEN'), + ('pcapkit/toolkit/scapy.py', 'pcapkit.protocols.internet.ipv6', 'IPV6_HEADER_LEN'), + ('pcapkit/toolkit/scapy.py', 'pcapkit.protocols.schema.transport.tcp', 'TCP_MIN_HEADER_LEN'), + ('pcapkit/toolkit/scapy.py', 'pcapkit.protocols.schema.transport.tcp', 'TCP_DATA_OFFSET_OCTET'), + ('pcapkit/toolkit/scapy.py', 'pcapkit.protocols.schema.misc.pcap.header', 'RESOLUTION'), + ('pcapkit/toolkit/scapy.py', 'pcapkit.protocols.schema.misc.pcap.header', 'RESOLUTION_PLACES'), + ('pcapkit/toolkit/pcap.py', 'pcapkit.protocols.schema.transport.tcp', 'TCP_MIN_HEADER_LEN'), + ('pcapkit/toolkit/pcap.py', 'pcapkit.protocols.schema.transport.tcp', 'TCP_DATA_OFFSET_OCTET'), + ('pcapkit/toolkit/pcapng.py', 'pcapkit.protocols.schema.misc.pcap.header', 'RESOLUTION'), + ('pcapkit/toolkit/dpkt.py', 'pcapkit.protocols.schema.misc.pcap.header', 'RESOLUTION'), + ('pcapkit/toolkit/dpkt.py', 'pcapkit.protocols.schema.misc.pcap.header', 'RESOLUTION_PLACES'), + ('pcapkit/toolkit/pypcapfile.py', 'pcapkit.protocols.schema.misc.pcap.header', 'RESOLUTION'), + ('pcapkit/foundation/reassembly/ip.py', 'pcapkit.protocols.schema.internet.ipv6', 'IP_MAX_LENGTH'), + ('pcapkit/foundation/reassembly/ipv6.py', 'pcapkit.protocols.internet.ipv6', 'IPV6_HEADER_LEN'), + ('pcapkit/foundation/reassembly/ipv6.py', 'pcapkit.protocols.internet.ipv6', + 'IPV6_NEXT_HEADER_OFFSET'), + ('pcapkit/protocols/internet/ipv6.py', 'pcapkit.protocols.schema.internet.ipv6', 'IP_MAX_LENGTH'), + ('pcapkit/dumpkit/pcap.py', 'pcapkit.protocols.schema.misc.pcap.header', 'RESOLUTION'), + ('pcapkit/protocols/misc/pcap/header.py', 'pcapkit.protocols.schema.misc.pcap.header', + 'MAGIC_NUMBER'), + ('pcapkit/protocols/misc/pcap/frame.py', 'pcapkit.protocols.schema.misc.pcap.header', + 'RESOLUTION'), + ('pcapkit/protocols/misc/pcapng.py', 'pcapkit.protocols.schema.misc.pcap.header', 'RESOLUTION'), + ('pcapkit/protocols/misc/pcapng.py', 'pcapkit.protocols.schema.misc.pcapng', 'BLOCK_MINIMUM'), + ('pcapkit/protocols/misc/pcapng.py', 'pcapkit.protocols.schema.misc.pcapng', 'BYTE_ORDER_MAGIC'), + ('pcapkit/protocols/misc/pcapng.py', 'pcapkit.protocols.schema.misc.pcapng', + 'DEFAULT_RESOLUTION'), + ('pcapkit/protocols/misc/pcapng.py', 'pcapkit.protocols.schema.misc.pcapng', + 'GENERIC_BLOCK_MINIMUM'), + ('pcapkit/foundation/engines/pcap.py', 'pcapkit.protocols.schema.misc.pcap.header', + 'MAGIC_NUMBER'), + ('pcapkit/foundation/engines/pcapng.py', 'pcapkit.protocols.schema.misc.pcapng', + 'BLOCK_MINIMUM'), + ('pcapkit/foundation/engines/pcapng.py', 'pcapkit.protocols.schema.misc.pcapng', + 'BYTE_ORDER_MAGIC'), + ('pcapkit/foundation/engines/pcapng.py', 'pcapkit.protocols.schema.misc.pcapng', + 'GENERIC_BLOCK_MINIMUM'), + ('pcapkit/foundation/engines/scapy.py', 'pcapkit.protocols.schema.misc.pcap.header', + 'RESOLUTION'), + ('pcapkit/foundation/engines/scapy.py', 'pcapkit.protocols.schema.misc.pcapng', + 'DEFAULT_RESOLUTION'), + ('pcapkit/foundation/engines/dpkt.py', 'pcapkit.protocols.schema.misc.pcap.header', + 'RESOLUTION'), + ('pcapkit/foundation/engines/dpkt.py', 'pcapkit.protocols.schema.misc.pcapng', + 'BYTE_ORDER_MAGIC'), + ('pcapkit/foundation/engines/dpkt.py', 'pcapkit.protocols.schema.misc.pcapng', + 'DEFAULT_RESOLUTION'), + ('pcapkit/foundation/engines/dpkt.py', 'pcapkit.protocols.schema.misc.pcapng', + 'GENERIC_BLOCK_MINIMUM'), + ('pcapkit/foundation/engines/pcap.py', 'pcapkit.protocols.schema.misc.pcap.frame', + 'RECORD_HEADER_LEN'), + ('pcapkit/foundation/engines/pypcapfile.py', 'pcapkit.protocols.schema.misc.pcap.frame', + 'RECORD_HEADER_LEN'), + ('pcapkit/foundation/engines/scapy.py', 'pcapkit.protocols.schema.misc.pcap.frame', + 'RECORD_HEADER_LEN'), + ('pcapkit/protocols/misc/pcap/frame.py', 'pcapkit.protocols.schema.misc.pcap.frame', + 'RECORD_HEADER_LEN'), + ('pcapkit/protocols/misc/pcap/header.py', 'pcapkit.protocols.schema.misc.pcap.header', + 'GLOBAL_HEADER_LEN'), + ('pcapkit/protocols/internet/ipv4.py', 'pcapkit.protocols.schema.internet.ipv4', 'IPV4_HEADER_LEN'), + ('pcapkit/protocols/internet/ah.py', 'pcapkit.protocols.schema.internet.ah', 'AH_HEADER_LEN'), + ('pcapkit/protocols/transport/tcp.py', 'pcapkit.protocols.schema.transport.tcp', + 'TCP_MIN_HEADER_LEN'), +) + +#: Readers that reach a home as a module attribute, ``Module.NAME``, rather than +#: importing the name: the module, the home's module path and the name. +#: :mod:`pcapkit.toolkit.pypcapfile` does so because binding the four names it +#: used to define would re-export them (see :meth:`TestReaders.test_pypcapfile_binds_none`). +ATTRIBUTE_READERS = ( + ('pcapkit/toolkit/pypcapfile.py', 'pcapkit.protocols.schema.internet.ipv4', 'IPV4_HEADER_LEN'), + ('pcapkit/toolkit/pypcapfile.py', 'pcapkit.protocols.internet.ipv6', 'IPV6_HEADER_LEN'), + ('pcapkit/toolkit/pypcapfile.py', 'pcapkit.protocols.internet.ipv6', 'IPV6_NEXT_HEADER_OFFSET'), + ('pcapkit/toolkit/pypcapfile.py', 'pcapkit.protocols.schema.internet.ah', 'AH_HEADER_LEN'), + ('pcapkit/toolkit/pypcapfile.py', 'pcapkit.protocols.schema.transport.tcp', 'TCP_MIN_HEADER_LEN'), + ('pcapkit/toolkit/pypcapfile.py', 'pcapkit.protocols.schema.transport.tcp', + 'TCP_DATA_OFFSET_OCTET'), +) + +#: The trees :class:`TestNotReinlined` sweeps. +SWEPT = ('pcapkit/protocols', 'pcapkit/foundation', 'pcapkit/toolkit', 'pcapkit/dumpkit') + +#: Values the sweep leaves to :meth:`TestOneDefinition.test_literals_at_home`, +#: which finds them anywhere because they are distinctive. +DISTINCTIVE = (10 ** 6, 10 ** 9, 0x1A2B3C4D) + +#: Sweep hits of a homed value that are not a copy of it, one expression each: +#: the module, the expression (see :func:`_context`) and why. A hit here is in a +#: module whose own protocol owns a constant of that value, or that reads one. +NOT_A_COPY = { + ('pcapkit/protocols/internet/ipv6.py', "schema.hextet['version'] != 6"): + 'the IP version number, not the Next Header offset', + ('pcapkit/protocols/internet/ipv6.py', 'self._import_next_layer(version=6)'): + 'the IP version number, not the Next Header offset', + ('pcapkit/protocols/internet/ipv6.py', "{'version': 6}"): + 'the IP version number, not the Next Header offset', + ('pcapkit/protocols/schema/internet/ipv6.py', 'ptr + 6'): + 'the end of a Jumbo Payload option (2 octets of header, 4 of value)', + ('pcapkit/protocols/misc/pcap/frame.py', 'version == 6'): + 'the IP version number, not the decimal places of a resolution', + ('pcapkit/protocols/misc/pcapng.py', 'version == 6'): + 'the IP version number, not the decimal places of a resolution', + ('pcapkit/protocols/misc/pcapng.py', 'schema.length != 6'): + 'a MAC address option is 6 octets', + ('pcapkit/protocols/misc/pcapng.py', 'Schema_IF_MACAddrOption(length=6)'): + 'a MAC address option is 6 octets', + ('pcapkit/protocols/misc/pcapng.py', 'schema.length != 16'): + 'an IPv6 address option is 16 octets, not a Simple Packet Block', + ('pcapkit/protocols/misc/pcapng.py', 'Schema_NS_DNSIP6AddrOption(length=16)'): + 'an IPv6 address option is 16 octets, not a Simple Packet Block', + ('pcapkit/protocols/misc/pcapng.py', '(body + bytes(20))[:20]'): + 'an epb_hash option body padded to its 20-octet SHA-1 digest', + ('pcapkit/protocols/misc/pcapng.py', 'bytes(20)'): + 'an epb_hash option body padded to its 20-octet SHA-1 digest', + ('pcapkit/protocols/misc/pcapng.py', '(body + bytes(16))[4:16]'): + 'an epb_hash option body padded to its 16-octet MD5 digest', + ('pcapkit/protocols/misc/pcapng.py', 'bytes(16)'): + 'an epb_hash option body padded to its 16-octet MD5 digest', + ('pcapkit/protocols/misc/pcapng.py', '16 + _utf8_length(records)'): + 'an IPv6 name record is 16 octets of address, not a block', + ('pcapkit/protocols/misc/pcapng.py', 'timestamp_high << 32'): + 'a bit width: the high half of a 64-bit timestamp', + ('pcapkit/protocols/misc/pcapng.py', 'ts_info >> 32'): + 'a bit width: the high half of a 64-bit timestamp', + ('pcapkit/protocols/misc/pcapng.py', 'short_address >> 16'): + 'a bit width: the high half of a 32-bit address', + ('pcapkit/protocols/misc/pcapng.py', 'schema.addr_high << 16'): + 'a bit width: the high half of a 32-bit address', + ('pcapkit/protocols/schema/misc/pcapng.py', 'BytesField(length=6)'): + 'a MAC address is 6 octets', + ('pcapkit/protocols/schema/misc/pcapng.py', 'BytesField(length=16)'): + 'an IPv6 address is 16 octets', + ('pcapkit/protocols/schema/misc/pcapng.py', "pkt['length'] - 16"): + 'an IPv6 name record is 16 octets of address, not a block', + ('pcapkit/protocols/schema/misc/pcapng.py', 'magic == 12'): + 'the Block Total Length of a 12-octet Section Header Block read where the ' + 'Byte-Order Magic belongs, spelled as the octets read to pair with the ' + 'byte-swapped 0x0C000000 beside it', + ('pcapkit/protocols/schema/misc/pcapng.py', 'magic in (BYTE_ORDER_MAGIC, 12)'): + 'the same octets read where the Byte-Order Magic belongs, paired with the ' + 'byte-swapped 0x0C000000 beside it, as in the check above', + ('pcapkit/protocols/schema/misc/pcapng.py', "{'reserved': (12, 12)}"): + 'an epb_flags bit-field: 12 reserved bits from bit 12', + ('pcapkit/protocols/schema/misc/pcapng.py', "{'crc_error': (24, 1)}"): + 'an epb_flags bit offset, not an Interface Statistics Block', + ('pcapkit/protocols/schema/misc/pcapng.py', "{'unaligned_error': (28, 1)}"): + 'an epb_flags bit offset, not a Section Header Block', + ('pcapkit/protocols/schema/misc/pcapng.py', "{'byteorder': (32, 32)}"): + "a bit-field: a Section Header Block's Byte-Order Magic, the 32 bits after " + 'its Block Total Length, not a Packet Block', + ('pcapkit/protocols/schema/transport/tcp.py', "pkt['length'] > 12"): + 'an MP_CAPABLE option length, not the Data Offset octet', + ('pcapkit/protocols/schema/transport/tcp.py', "pkt['length'] > 20"): + 'an MP_CAPABLE option length, not the minimum header length', + ('pcapkit/protocols/schema/transport/tcp.py', 'BytesField(length=20)'): + 'an MPTCP MP_JOIN SYN/ACK truncated HMAC is 20 octets', + ('pcapkit/protocols/schema/transport/tcp.py', "pkt['length'] in (20, 22, 28, 30)"): + 'the MP_ADD_ADDR option lengths with an IPv6 address, not the minimum header length', + ('pcapkit/protocols/schema/transport/tcp.py', "{'reserved': (4, 12)}"): + 'an MPTCP bit-field: 12 reserved bits after the 4-bit subtype, not the Data ' + 'Offset octet', + ('pcapkit/protocols/transport/tcp.py', 'schema.length != 12'): + 'an MPTCP option length, not the Data Offset octet', + ('pcapkit/protocols/transport/tcp.py', 'schema.length > 12'): + 'an MP_CAPABLE option length, not the Data Offset octet', + ('pcapkit/protocols/transport/tcp.py', 'schema.length > 20'): + 'a field offset within the MP_CAPABLE option, not the minimum header length', + ('pcapkit/protocols/transport/tcp.py', 'Schema_MPTCPJoinSYN(length=12)'): + 'an MP_JOIN SYN option is 12 octets', + ('pcapkit/protocols/transport/tcp.py', 'Schema_MPTCPFallback(length=12)'): + 'an MP_FAIL option is 12 octets', + ('pcapkit/protocols/transport/tcp.py', 'Schema_MPTCPFastclose(length=12)'): + 'an MP_FASTCLOSE option is 12 octets', + ('pcapkit/protocols/transport/tcp.py', 'bytes(20)'): + 'an MP_JOIN SYN/ACK HMAC default of 20 octets', + ('pcapkit/protocols/transport/tcp.py', + "{20: ModuleDescriptor('pcapkit.protocols.application.ftp', 'FTP_DATA')}"): + 'the FTP data port, not the minimum header length', + ('pcapkit/protocols/transport/tcp.py', 'schema.length not in (4, 12, 20, 22, 24)'): + 'the MP_CAPABLE option lengths, :rfc:`8684#section-3.1`', + ('pcapkit/protocols/transport/tcp.py', '(4, 12, 20, 22, 24)[present]'): + 'the MP_CAPABLE option lengths, :rfc:`8684#section-3.1`', + ('pcapkit/protocols/transport/tcp.py', 'schema.length not in (8, 10, 16, 18, 20, 22, 28, 30)'): + 'the MP_ADD_ADDR option lengths, :rfc:`8684#section-3.4.1`', + ('pcapkit/foundation/engines/dpkt.py', 'pkt.ts_high << 32'): + 'a bit width: the high half of a 64-bit timestamp', + ('pcapkit/foundation/engines/dpkt.py', 'version == 6'): + 'the IP version number', + ('pcapkit/foundation/engines/dpkt.py', '{24: ip6}'): + 'the BSD loopback address family of IPv6 on NetBSD and OpenBSD', + ('pcapkit/foundation/engines/dpkt.py', '{28: ip6}'): + 'the BSD loopback address family of IPv6 on FreeBSD', + ('pcapkit/foundation/engines/pcap.py', 'head[8:12]'): + "a record header's incl_len field, octets 8 to 12", + ('pcapkit/foundation/engines/pcapng.py', 'self._peek(12)'): + "the first 12 octets of a block -- type, length and a Section Header Block's " + 'Byte-Order Magic -- not the generic block minimum', + ('pcapkit/foundation/engines/pcapng.py', 'head[8:12]'): + "the Byte-Order Magic field of a Section Header Block, octets 8 to 12", + ('pcapkit/foundation/engines/pcapng.py', 'self._peek(16)[12:16]'): + "a Decryption Secrets Block's Secrets Length field, octets 12 to 16", + ('pcapkit/foundation/engines/pcapng.py', 'self._peek(16)'): + "a Decryption Secrets Block up to its Secrets Length, not a block minimum", + ('pcapkit/foundation/engines/scapy.py', 'len(options) >= 12'): + 'an if_tsoffset option: 4 octets of header and 8 of value', + ('pcapkit/foundation/engines/scapy.py', 'options[4:12]'): + 'the 8-octet value of an if_tsoffset option', + ('pcapkit/foundation/reassembly/tcp.py', '1 << 24'): + 'a bit width, not a PCAP global header', + ('pcapkit/toolkit/pcap.py', 'octets[8:12]'): + "a TCP header's Acknowledgment Number, octets 8 to 12", + ('pcapkit/toolkit/pypcapfile.py', 'len(raw) == 6'): + 'a MAC address is 6 octets', + ('pcapkit/toolkit/pypcapfile.py', '{Enum_TransType.IPv6: 6}'): + 'the IP version an IPv6-in-IP tunnel carries, not the Next Header offset', + ('pcapkit/dumpkit/common.py', 'ord(char) > 65535'): + 'the last code point of the Basic Multilingual Plane, not an IP length', +} + +#: Sweep hits of a homed value in a module whose own protocol owns no constant +#: of that value, nor reads one: the module mapped to the values left there and +#: why. A value not listed still fails, so a copy added later is still caught. +UNRELATED = { + 'pcapkit/protocols/application/http.py': ((6, 9, 12), 'the length of an HTTP/1 version ' + 'token, and HTTP/2 frame types and flag masks'), + 'pcapkit/protocols/application/httpv2.py': ((6, 9), 'HTTP/2 frame header and setting sizes'), + 'pcapkit/protocols/internet/ah.py': ((20,), "AH's length estimate, its 12-octet fixed part " + 'plus a minimal ICV, not the fixed part itself'), + 'pcapkit/protocols/schema/internet/ipv6_option.py': ((16,), 'an MPL Seed-ID of 16 octets'), + 'pcapkit/protocols/schema/application/httpv2.py': ((6, 32), + 'an HTTP/2 setting size and a flag bit'), + 'pcapkit/protocols/application/ospf.py': ((24, 65535), 'OSPF header length and field bounds'), + 'pcapkit/protocols/schema/application/ospf.py': ((24,), 'the OSPF header length'), + 'pcapkit/protocols/internet/esp.py': ((12, 16, 20, 24, 32), 'ESP ICV and key sizes'), + 'pcapkit/protocols/internet/hip.py': ((6, 12, 16, 20, 32, 40), 'HIP parameter fields and sizes'), + 'pcapkit/protocols/schema/internet/hip.py': ((12, 16, 20, 65535), + 'HIP parameter fields, sizes and bit-fields'), + 'pcapkit/protocols/internet/hopopt.py': ((6, 16, 32), 'HOPOPT option fields, widths and sizes'), + 'pcapkit/protocols/schema/internet/hopopt.py': ((6, 16), 'HOPOPT option field sizes ' + 'and bit-fields'), + 'pcapkit/protocols/internet/ipv6_opts.py': ((6, 16, 32), + 'IPv6-Opts option fields, widths and sizes'), + 'pcapkit/protocols/schema/internet/ipv6_opts.py': ((6, 16), 'IPv6-Opts option field sizes ' + 'and bit-fields'), + 'pcapkit/protocols/internet/ipv6_ext.py': ((6,), 'the IP version number'), + 'pcapkit/protocols/internet/ipv6_route.py': ((6, 16), 'Routing header address sizes'), + 'pcapkit/protocols/schema/internet/ipv6_route.py': ((16, 20), 'an IPv6 address is 16 ' + 'octets; a 20-bit reserved bit-field'), + 'pcapkit/protocols/schema/internet/ipv4.py': ((6, 16), 'the ECN and option-function ' + 'bit-field offsets'), + 'pcapkit/protocols/schema/internet/ipv6.py': ((12, 20), 'the 20-bit Flow Label from bit 12 ' + 'of the first 32 bits'), + 'pcapkit/protocols/internet/ipv6.py': ((16,), 'a hex base and an IPv6 address size'), + 'pcapkit/protocols/internet/ipv4.py': ((9, 12, 16, 40), 'IPv4 option counts and sizes; ' + '40 is the option area maximum, not an IPv6 header'), + 'pcapkit/protocols/internet/ipx.py': ((12,), 'an IPX address is 12 octets'), + 'pcapkit/protocols/schema/internet/ipx.py': ((12,), 'an IPX address is 12 octets'), + 'pcapkit/protocols/internet/mh.py': ((6, 16, 20, 24, 32), + 'Mobility Header option fields and sizes'), + 'pcapkit/protocols/schema/internet/mh.py': ((6, 9, 12, 16, 20, 24, 32), + 'Mobility Header option fields, sizes and bit-fields'), + 'pcapkit/protocols/link/arp.py': ((6, 9, 16, 28), 'address sizes, InARP operation codes, ' + 'and the 28-octet IPv4 ARP length'), + 'pcapkit/protocols/link/loopback.py': ((24, 28), 'the BSD loopback address families of IPv6 ' + '(NetBSD and OpenBSD 24, FreeBSD 28)'), + 'pcapkit/protocols/link/l2tpv2.py': ((6, 12, 16), 'L2TPv2 header flags, widths and length'), + 'pcapkit/protocols/schema/link/ethernet.py': ((6,), 'a MAC address is 6 octets'), + 'pcapkit/protocols/schema/link/l2tp.py': ((6, 12), 'an L2TP header field offset, and the ' + 'version bit-field'), + 'pcapkit/protocols/schema/link/vlan.py': ((12,), 'the 12-bit VLAN identifier'), + 'pcapkit/protocols/misc/pcapng.py': ((65535,), 'a 16-bit mask'), + 'pcapkit/protocols/schema/misc/pcapng.py': ((9, 65535), 'an epb_flags bit offset, and a ' + '16-bit field default'), + 'pcapkit/protocols/transport/sctp.py': ((12, 16, 20), 'SCTP chunk and parameter sizes'), + 'pcapkit/protocols/schema/transport/sctp.py': ((6, 16, 20), 'SCTP chunk and parameter sizes, ' + 'and a flag bit'), + 'pcapkit/protocols/transport/tcp.py': ((6, 16, 24, 32, 65535), 'TCP and MPTCP option sizes, ' + 'and the 16-bit window and MSS defaults'), + 'pcapkit/protocols/schema/transport/tcp.py': ((6, 16, 28), 'TCP and MPTCP option sizes, and ' + 'the MP_ADD_ADDR lengths'), + 'pcapkit/toolkit/pyshark.py': ((6, 9, 20, 24, 28), 'Wireshark encapsulation numbers mapped ' + 'to link types'), + 'pcapkit/foundation/engines/pypcapfile.py': ((12,), 'the Python 3.12 version ceiling of ' + 'PyPCAPFile, not a length'), +} + +#: Values equal to one of the homes' but meaning something else, so left where +#: they are rather than read from a home: the module, the name and the value. +#: ``0xFFFF_FFFF`` is the largest 32-bit captured length in the Scapy engine, +#: and the mask a PCAP record field is truncated to in the dumper. +DISTINCT = ( + ('pcapkit.foundation.engines.scapy', '_CAPLEN_MAX', 0xFFFF_FFFF), + ('pcapkit.dumpkit.pcap', '_UINT32_MASK', 0xFFFF_FFFF), +) + + +def _sources() -> 'list[pathlib.Path]': + """Every module the rule governs: all of :mod:`pcapkit` but the generated trees.""" + skip = (PACKAGE / 'vendor', PACKAGE / 'const') + return sorted(path for path in PACKAGE.rglob('*.py') + if not any(base in path.parents for base in skip)) + + +def _rel(path: 'pathlib.Path') -> 'str': + """The path relative to the repository root, with forward slashes.""" + return path.relative_to(ROOT).as_posix() + + +def _assigned(tree: 'ast.Module') -> 'set[str]': + """Names assigned at module level, or at the top level of a class body.""" + names = set() # type: set[str] + bodies = [tree.body] + [node.body for node in tree.body if isinstance(node, ast.ClassDef)] + for body in bodies: + for node in body: + targets = [] # type: list[ast.expr] + if isinstance(node, ast.Assign): + targets = node.targets + elif isinstance(node, (ast.AnnAssign, ast.AugAssign)): + targets = [node.target] + names.update(target.id for target in targets if isinstance(target, ast.Name)) + return names + + +def _swept_values() -> 'set[int]': + """Every homed scalar value, a table's values each, less the distinctive ones.""" + values = set() # type: set[int] + for _, _, value in HOMES: + for scalar in (value.values() if isinstance(value, dict) else (value,)): + if type(scalar) is int and scalar not in DISTINCTIVE: + values.add(scalar) + return values + + +def _segment(text: 'str', node: 'ast.AST') -> 'str': + """The source of ``node``, on one line.""" + return ' '.join((ast.get_source_segment(text, node) or '').split()) + + +#: Nodes a constant may sit inside without changing what it means: the +#: expression that holds the container names it (see :func:`_context`). +_CONTAINERS = (ast.Tuple, ast.List, ast.Set, ast.Starred, ast.UnaryOp, ast.keyword) + +#: Statements with a body, too long to name a constant by. +_COMPOUND = (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef, ast.For, ast.AsyncFor, + ast.While, ast.If, ast.With, ast.AsyncWith, ast.Try) + + +def _parents(tree: 'ast.AST') -> 'dict[int, ast.AST]': + """``id`` of every node mapped to its parent.""" + return {id(child): node for node in ast.walk(tree) for child in ast.iter_child_nodes(node)} + + +def _home_sites(rel: 'str', tree: 'ast.Module') -> 'set[int]': + """``id`` of every constant in a home's own definition in ``rel``: its value and table.""" + names = {name for module, name, _ in HOMES if module.replace('.', '/') + '.py' == rel} + sites = set() # type: set[int] + for node in tree.body: + if isinstance(node, ast.Assign): + targets = node.targets + elif isinstance(node, ast.AnnAssign) and node.value is not None: + targets = [node.target] + else: + continue + if any(isinstance(target, ast.Name) and target.id in names for target in targets): + sites.update(id(sub) for sub in ast.walk(node.value)) + return sites + + +def _default_of(arguments: 'ast.arguments', operand: 'ast.expr') -> 'str': + """The parameter whose default is ``operand``.""" + positional = [*getattr(arguments, 'posonlyargs', []), *arguments.args] + pairs = list(zip(positional[len(positional) - len(arguments.defaults):], arguments.defaults)) + pairs += list(zip(arguments.kwonlyargs, arguments.kw_defaults)) + return next(arg.arg for arg, default in pairs if default is operand) + + +def _context(text: 'str', operand: 'ast.Constant', parents: 'dict[int, ast.AST]') -> 'str': + """How :data:`NOT_A_COPY` names the expression ``operand`` sits in. + + The constant is named by the nearest node holding it that is not a mere + container (a tuple, list, set, starred or unary operand, or a keyword) -- + unless that is a statement with a body, when the container names it. A call + is named by its callee and the one argument, ``f(name=value)``, so that a + multi-line call keeps a short key; a default by its function and parameter, + ``def f(name=value)``; a dictionary entry by its pair, ``{key: value}``; a + subscript or slice bound by the whole subscript; anything else by its source. + + """ + child, node = operand, parents[id(operand)] # type: ast.AST, ast.AST + while isinstance(node, _CONTAINERS) and not isinstance(parents.get(id(node)), _COMPOUND): + child, node = node, parents[id(node)] + if sys.version_info < (3, 9) and isinstance(node, ast.Index): # an ast.Index wraps the subscript + child, node = node, parents[id(node)] # pragma: no cover + if isinstance(node, ast.Slice): + child, node = node, parents[id(node)] + if isinstance(node, ast.Call): + if isinstance(child, ast.keyword): + return f'{_segment(text, node.func)}({child.arg}={operand.value})' + return f'{_segment(text, node.func)}({operand.value})' + if isinstance(node, ast.arguments): + function = parents[id(node)] + name = getattr(function, 'name', 'lambda') + return f'def {name}({_default_of(node, child)}={operand.value})' + if isinstance(node, ast.Dict): + for key, value in zip(node.keys, node.values): + if child is key or child is value: + return f'{{{_segment(text, key) if key else "**"}: {_segment(text, value)}}}' + if isinstance(node, (ast.Compare, ast.BinOp)): + # hex spellings are named by value, as the scan compares them + return _segment(text, node).replace('0x0000000C', '12').replace('0xFFFF', '65535') + if isinstance(node, _COMPOUND): + return _segment(text, child) + return _segment(text, node) + + +def _sweep() -> 'list[tuple[str, int, int, str]]': + """Every homed value in :data:`SWEPT`: module, value, line, expression. + + A value is swept wherever it is spelled -- an operand, a call argument, a + subscript, a default, a returned value, an element of a tuple or a table, + a branch of a conditional, or an assignment at any level, which is how a + renamed copy (``_TCP_HDR = 20``) would read. Only the homes' own + definitions, their tables included, are not hits. + + """ + values = _swept_values() + hits = [] # type: list[tuple[str, int, int, str]] + for tree_root in SWEPT: + for path in sorted((ROOT / tree_root).rglob('*.py')): + rel = _rel(path) + text = path.read_text(encoding='utf-8') + tree = ast.parse(text) + parents = _parents(tree) + sites = _home_sites(rel, tree) + for node in ast.walk(tree): + if (isinstance(node, ast.Constant) and type(node.value) is int + and node.value in values and id(node) not in sites): + hits.append((rel, node.value, node.lineno, _context(text, node, parents))) + return hits + + +def _time_ns_divisor(tree: 'ast.Module') -> 'set[int]': + """``id`` of every constant that divides a :func:`time.time_ns` reading.""" + exempt = set() # type: set[int] + for node in ast.walk(tree): + if (isinstance(node, ast.BinOp) and isinstance(node.op, ast.Div) + and isinstance(node.right, ast.Constant) + and any(isinstance(call, ast.Call) and isinstance(call.func, ast.Attribute) + and call.func.attr == 'time_ns' for call in ast.walk(node.left))): + exempt.add(id(node.right)) + return exempt + + +class TestHomes(unittest.TestCase): + """Each constant is defined in its home, with its RFC or format value.""" + + def test_value(self) -> 'None': + for module, name, value in HOMES: + with self.subTest(module=module, name=name): + home = importlib.import_module(module) + self.assertTrue(hasattr(home, name), f'{module} defines no {name}') + self.assertEqual(getattr(home, name), value) + + def test_defined_not_imported(self) -> 'None': + for module, name, _ in HOMES: + with self.subTest(module=module, name=name): + path = ROOT / (module.replace('.', '/') + '.py') + self.assertIn(name, _assigned(ast.parse(path.read_text(encoding='utf-8')))) + + +class TestOneDefinition(unittest.TestCase): + """No module but the home defines the name or spells the literal.""" + + def test_no_second_definition(self) -> 'None': + homes = {name: module.replace('.', '/') + '.py' for module, name, _ in HOMES} + offenders = [] # type: list[str] + for path in _sources(): + assigned = _assigned(ast.parse(path.read_text(encoding='utf-8'))) + for name, home in homes.items(): + # NOTE: the engines' ``MAGIC_NUMBER`` is their dispatch contract + # and is built from the home; the literal check covers it. + if name == 'MAGIC_NUMBER' and _rel(path).startswith('pcapkit/foundation/engines/'): + continue + if name in assigned and _rel(path) != home: + offenders.append(f'{_rel(path)}: {name}') + self.assertEqual(offenders, []) + + def test_old_names_gone(self) -> 'None': + for rel, name in OLD_NAMES: + with self.subTest(module=rel, name=name): + tree = ast.parse((ROOT / rel).read_text(encoding='utf-8')) + self.assertNotIn(name, _assigned(tree)) + + def test_literals_at_home(self) -> 'None': + offenders = [] # type: list[str] + for path in _sources(): + tree = ast.parse(path.read_text(encoding='utf-8')) + exempt = _time_ns_divisor(tree) + for node in ast.walk(tree): + if not isinstance(node, ast.Constant) or isinstance(node.value, bool): + continue + if not isinstance(node.value, (bytes, int)) or node.value not in LITERALS: + continue + allowed = LITERALS[node.value] + allowed = (allowed,) if isinstance(allowed, str) or allowed is None else allowed + if _rel(path) not in allowed and id(node) not in exempt: + offenders.append(f'{_rel(path)}:{node.lineno}: {node.value!r}') + self.assertEqual(offenders, []) + + +class TestReaders(unittest.TestCase): + """Each reader imports the constant from its home.""" + + def test_imports_home(self) -> 'None': + for rel, module, name in READERS: + with self.subTest(reader=rel, name=name): + tree = ast.parse((ROOT / rel).read_text(encoding='utf-8')) + imported = {alias.name for node in ast.walk(tree) + if isinstance(node, ast.ImportFrom) and node.module == module + for alias in node.names} + self.assertIn(name, imported, f'{rel} does not import {name} from {module}') + + def test_imported_name_is_read(self) -> 'None': + # NOTE: an import left behind by a reader that spelled the value out + # again would still pass :meth:`test_imports_home`. + for rel, module, name in READERS: + with self.subTest(reader=rel, name=name): + tree = ast.parse((ROOT / rel).read_text(encoding='utf-8')) + bound = {alias.asname or alias.name for node in ast.walk(tree) + if isinstance(node, ast.ImportFrom) and node.module == module + for alias in node.names if alias.name == name} + loads = [node for node in ast.walk(tree) if isinstance(node, ast.Name) + and node.id in bound and isinstance(node.ctx, ast.Load)] + self.assertTrue(loads, f'{rel} never reads {name}') + + def test_attribute_readers(self) -> 'None': + for rel, module, name in ATTRIBUTE_READERS: + with self.subTest(reader=rel, name=name): + tree = ast.parse((ROOT / rel).read_text(encoding='utf-8')) + package, _, leaf = module.rpartition('.') + aliases = {alias.asname or alias.name for node in ast.walk(tree) + if isinstance(node, ast.ImportFrom) and node.module == package + for alias in node.names if alias.name == leaf} + self.assertTrue(aliases, f'{rel} does not import {module}') + loads = [node for node in ast.walk(tree) if isinstance(node, ast.Attribute) + and node.attr == name and isinstance(node.value, ast.Name) + and node.value.id in aliases and isinstance(node.ctx, ast.Load)] + self.assertTrue(loads, f'{rel} never reads {module}.{name}') + + def test_pypcapfile_binds_none(self) -> 'None': + # The four names defined there until #1739: reading them through + # ``pcapkit.toolkit.pypcapfile`` would be a re-export, which ruling 1 on + # #1726 rules out, so the module reaches them as module attributes. + pypcapfile = importlib.import_module('pcapkit.toolkit.pypcapfile') + for name in ('TCP_MIN_HEADER_LEN', 'IPV4_HEADER_LEN', 'IPV6_HEADER_LEN', 'AH_HEADER_LEN'): + with self.subTest(name=name): + self.assertFalse(hasattr(pypcapfile, name)) + + def test_engine_tables_gone(self) -> 'None': + engine = importlib.import_module('pcapkit.foundation.engines.pcapng').PCAPNG + for name in ('BLOCK_MINIMUM', 'BYTE_ORDER_MAGIC'): + with self.subTest(name=name): + self.assertNotIn(name, vars(engine)) + + def test_engine_magic_from_home(self) -> 'None': + pcap = importlib.import_module('pcapkit.foundation.engines.pcap').PCAP + pcapng = importlib.import_module('pcapkit.foundation.engines.pcapng').PCAPNG + magic = importlib.import_module('pcapkit.protocols.schema.misc.pcap.header').MAGIC_NUMBER + self.assertEqual(pcap.MAGIC_NUMBER, tuple(magic)) + self.assertEqual(pcapng.MAGIC_NUMBER, (b'\x0a\x0d\x0d\x0a',)) + + +class TestNotReinlined(unittest.TestCase): + """No module spells a homed value again where the name belongs.""" + + def test_sweep(self) -> 'None': + offenders = [] # type: list[str] + seen_copy = set() # type: set[tuple[str, str]] + seen_unrelated = set() # type: set[tuple[str, int]] + for rel, value, lineno, context in _sweep(): + if (rel, context) in NOT_A_COPY: + seen_copy.add((rel, context)) + elif value in UNRELATED.get(rel, ((), ''))[0]: + seen_unrelated.add((rel, value)) + else: + offenders.append(f'{rel}:{lineno}: {context} (a copy of a homed constant?)') + self.assertEqual(offenders, []) + # an entry that no longer matches anything is stale, and would hide a + # copy that later took its place + self.assertEqual(sorted(set(NOT_A_COPY) - seen_copy), []) + listed = {(rel, value) for rel, (values, _) in UNRELATED.items() for value in values} + self.assertEqual(sorted(listed - seen_unrelated), []) + + def test_every_entry_has_a_reason(self) -> 'None': + reasons = list(NOT_A_COPY.values()) + [reason for _, reason in UNRELATED.values()] + self.assertTrue(all(reason.strip() for reason in reasons)) + + def test_distinct_values_stay_apart(self) -> 'None': + homes = {name for _, name, _ in HOMES} + for module, name, value in DISTINCT: + with self.subTest(module=module, name=name): + self.assertEqual(getattr(importlib.import_module(module), name), value) + self.assertNotIn(name, homes) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/misc/test_pcap_magic_lookup_1739_unit.py b/tests/protocols/misc/test_pcap_magic_lookup_1739_unit.py new file mode 100644 index 0000000000..9f7912f466 --- /dev/null +++ b/tests/protocols/misc/test_pcap_magic_lookup_1739_unit.py @@ -0,0 +1,100 @@ +# -*- coding: utf-8 -*- +"""The PCAP magic number is looked up in its one home as it was compared before. C.f. #1739. + +Moving the four magic numbers to +:data:`pcapkit.protocols.schema.misc.pcap.header.MAGIC_NUMBER` turned three +comparisons into lookups, which pins two behaviours the literals had: + +* :meth:`Header._make_magic ` + raises :exc:`KeyError` for a nanosecond flag that is neither, not a + :exc:`StopIteration` that a generator would turn into :exc:`RuntimeError`; +* :func:`~pcapkit.protocols.schema.misc.pcap.header.magic_number_callback` and + :meth:`Header.read ` match a + :obj:`bytearray` magic number, which compared equal to the literals, rather + than reject it as unhashable. + +Every value is built here, so the module is in the unit tier. + +""" +from __future__ import annotations + +import types +import unittest + +#: Each magic number as the octets read, and the byte order and nanosecond flag +#: it declares. +MAGICS = ( + (b'\xa1\xb2\x3c\x4d', 'big', True), + (b'\xa1\xb2\xc3\xd4', 'big', False), + (b'\x4d\x3c\xb2\xa1', 'little', True), + (b'\xd4\xc3\xb2\xa1', 'little', False), +) + + +def _header() -> 'object': + """A :class:`~pcapkit.protocols.misc.pcap.header.Header` with no file behind it.""" + from pcapkit.protocols.misc.pcap.header import Header # pylint: disable=import-outside-toplevel + return Header.__new__(Header) + + +class TestMakeMagic(unittest.TestCase): + """:meth:`Header._make_magic` looks the pair up.""" + + def test_each_pair(self) -> 'None': + for magic, byteorder, nanosecond in MAGICS: + with self.subTest(byteorder=byteorder, nanosecond=nanosecond): + self.assertEqual(_header()._make_magic(byteorder, nanosecond=nanosecond), # type: ignore[attr-defined] + (magic, byteorder == 'little')) + + def test_undeclared_flag_is_a_key_error(self) -> 'None': + for nanosecond in (None, 2, 'x'): + with self.subTest(nanosecond=nanosecond): + with self.assertRaises(KeyError): + _header()._make_magic('big', nanosecond=nanosecond) # type: ignore[attr-defined] + + def test_undeclared_flag_inside_a_generator(self) -> 'None': + # NOTE: a ``StopIteration`` escaping a generator is re-raised as + # ``RuntimeError`` (PEP 479); a ``KeyError`` stays itself. + def magics() -> 'object': + yield _header()._make_magic('little', nanosecond=None) # type: ignore[attr-defined] + with self.assertRaises(KeyError): + list(magics()) # type: ignore[call-overload] + + +class TestBytearrayMagic(unittest.TestCase): + """A :obj:`bytearray` magic number is matched, as it was compared.""" + + def test_callback(self) -> 'None': + from pcapkit.protocols.schema.misc.pcap.header import \ + magic_number_callback # pylint: disable=import-outside-toplevel + for magic, byteorder, _ in MAGICS: + with self.subTest(magic=magic): + field = types.SimpleNamespace() + magic_number_callback(field, {'magic_number': bytearray(magic)}) # type: ignore[arg-type] + self.assertEqual(field._byteorder, byteorder) # pylint: disable=protected-access + + def test_callback_rejects_others(self) -> 'None': + from pcapkit.protocols.schema.misc.pcap.header import \ + magic_number_callback # pylint: disable=import-outside-toplevel + from pcapkit.utilities.exceptions import ProtocolError # pylint: disable=import-outside-toplevel + for magic in (bytearray(b'\x00\x00\x00\x00'), 0xA1B2C3D4, None): + with self.subTest(magic=magic): + with self.assertRaises(ProtocolError): + magic_number_callback(types.SimpleNamespace(), {'magic_number': magic}) # type: ignore[arg-type] + + def test_read(self) -> 'None': + for magic, byteorder, nanosecond in MAGICS: + with self.subTest(magic=magic): + header = _header() + header.__header__ = types.SimpleNamespace( # type: ignore[attr-defined] + magic_number=bytearray(magic), version_major=2, version_minor=4, + thiszone=0, sigfigs=0, snaplen=0x40000, network=1, + ) + data = header.read() # type: ignore[attr-defined] + self.assertEqual(data.magic_number.byteorder, byteorder) + self.assertEqual(data.magic_number.nanosecond, nanosecond) + self.assertEqual(data.magic_number.data, magic) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/toolkit/test_rfc_constants_readers_1739_unit.py b/tests/toolkit/test_rfc_constants_readers_1739_unit.py new file mode 100644 index 0000000000..9063890761 --- /dev/null +++ b/tests/toolkit/test_rfc_constants_readers_1739_unit.py @@ -0,0 +1,75 @@ +# -*- coding: utf-8 -*- +"""Toolkit readers of the RFC-fixed constants behave at the constants' edges. C.f. #1739. + +The constants moved to their homes in :mod:`pcapkit.protocols`; these pin the +behaviour of the toolkit code that reads them, at exactly the value each one +gives: + +* :func:`pcapkit.toolkit.pypcapfile._walk_extension_headers` steps over an AH of + exactly :data:`~pcapkit.protocols.schema.internet.ah.AH_HEADER_LEN` octets -- one + carrying no Integrity Check Value -- and rejects a shorter one; +* :class:`pcapkit.toolkit.dpkt.Timestamp` defaults to microseconds and + :class:`~pcapkit.toolkit.dpkt.DecimalTimestamp` to nanoseconds, the two + resolutions of + :data:`~pcapkit.protocols.schema.misc.pcap.header.RESOLUTION`. + +Every value is built here and neither PyPCAPFile nor DPKT is imported, so the +module is in the unit tier. + +""" +from __future__ import annotations + +import decimal +import unittest + +#: IP protocol numbers of AH and TCP. +AH, TCP = 51, 6 + + +def _ah(payload_length: 'int', next_header: 'int' = TCP) -> 'bytes': + """An AH whose Payload Length is ``payload_length``, zero-filled.""" + return bytes([next_header, payload_length]) + bytes((payload_length + 2) * 4 - 2) + + +class TestWalkAH(unittest.TestCase): + """An AH is stepped over from :data:`AH_HEADER_LEN` octets up.""" + + def test_shortest_ah_is_stepped_over(self) -> 'None': + from pcapkit.toolkit.pypcapfile import \ + _walk_extension_headers # pylint: disable=import-outside-toplevel + segment = b'TCP header octets...' + ah = _ah(1) + self.assertEqual(len(ah), 12) + self.assertEqual(_walk_extension_headers(AH, ah + segment), (TCP, segment, 1)) + + def test_longer_ah_is_stepped_over(self) -> 'None': + from pcapkit.toolkit.pypcapfile import \ + _walk_extension_headers # pylint: disable=import-outside-toplevel + segment = b'TCP header octets...' + self.assertEqual(_walk_extension_headers(AH, _ah(4) + segment), (TCP, segment, 1)) + + def test_shorter_ah_is_rejected(self) -> 'None': + from pcapkit.toolkit.pypcapfile import \ + _walk_extension_headers # pylint: disable=import-outside-toplevel + # a Payload Length of 0 declares 8 octets, fewer than the fixed part + self.assertIsNone(_walk_extension_headers(AH, bytes([TCP, 0]) + bytes(30))) + # fewer octets than the fixed part at all + self.assertIsNone(_walk_extension_headers(AH, bytes([TCP, 1]) + bytes(9))) + + +class TestTimestampDefaults(unittest.TestCase): + """DPKT timestamps default to the resolution of their type.""" + + def test_float_timestamp_is_microseconds(self) -> 'None': + from pcapkit.toolkit.dpkt import Timestamp # pylint: disable=import-outside-toplevel + self.assertEqual(Timestamp(1.5).resolution, 1_000_000) + self.assertEqual(Timestamp().resolution, 1_000_000) + + def test_decimal_timestamp_is_nanoseconds(self) -> 'None': + from pcapkit.toolkit.dpkt import DecimalTimestamp # pylint: disable=import-outside-toplevel + self.assertEqual(DecimalTimestamp(decimal.Decimal('1.5')).resolution, 1_000_000_000) + self.assertEqual(DecimalTimestamp().resolution, 1_000_000_000) + + +if __name__ == '__main__': + unittest.main()