refactor(protocols): one home per duplicated RFC-fixed constant (#1739)

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:

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:

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:

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:

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:

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:

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

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',

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.

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:

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

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':

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)

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,15 +209,15 @@ 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
 
     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'