diff --git a/README.rst b/README.rst index 152922b299..b58f0eeb37 100644 --- a/README.rst +++ b/README.rst @@ -386,6 +386,17 @@ and ``make test-all`` regenerates them before running the whole suite: The same fixtures back the demonstration scripts in ``examples/legacy_smoke/``, which read them as ``../captures/…``. +One of the generators works differently from the rest and is worth knowing about. +``examples/generators/options.py`` does not describe packets it wants; it asks the +library which option, chunk, parameter, message, frame and block codes it +registers, and then constructs one of each through the public construction API, +parses it back, and constructs it again from what was parsed. The +``options-*.pcap`` captures are the cases that survive that round trip, so they +record what this version of ``pcapkit`` builds rather than what a third-party +tool builds. ``tests/protocols/test_option_roundtrip_unit.py`` runs the same +cases without needing a fixture at all, and carries a table of the ones that do +not yet close the cycle, each named against the defect that stops it. + Continuous integration runs the ``make test`` selection, since the fixtures are not in the repository. ``tshark`` is only required to exercise the PyShark engine, and is not needed by the test suite. diff --git a/examples/generators/make_samples.py b/examples/generators/make_samples.py index d05c715d75..c07c24a452 100644 --- a/examples/generators/make_samples.py +++ b/examples/generators/make_samples.py @@ -20,6 +20,7 @@ :file:`pcap.py` the ``.pcap`` captures the unit and runtime tests read :file:`pcapng.py` the ``.pcapng`` captures the regression tests read :file:`legacy.py` the extra captures ``examples/legacy_smoke/`` reads +:file:`options.py` the ``options-*.pcap`` option-coverage captures =================== ========================================================== They are loaded by path rather than imported by name, since this directory is @@ -44,8 +45,12 @@ HERE = pathlib.Path(__file__).resolve().parent #: Destination directory for every generated capture. DEST = ROOT / 'examples' / 'captures' -#: Generator modules, in the order they are run. -GENERATORS = ('pcap', 'pcapng', 'legacy') +#: Generator modules, in the order they are run. :file:`options.py` runs last +#: because it is the only one that builds its captures out of :mod:`pcapkit`'s +#: own construction output, so a failure in it is a statement about the library +#: rather than about the fixture -- and reading it after the others have already +#: printed keeps that distinction visible in the log. +GENERATORS = ('pcap', 'pcapng', 'legacy', 'options') def load(name: 'str') -> 'ModuleType': diff --git a/examples/generators/options.py b/examples/generators/options.py new file mode 100644 index 0000000000..b7b09b57da --- /dev/null +++ b/examples/generators/options.py @@ -0,0 +1,1767 @@ +# -*- coding: utf-8 -*- +"""Generate the option-coverage sample captures, and the case table behind them. + +The other generators in this directory cover the *parse* path for common +protocols. What they do not cover is the option, chunk, parameter and block +space -- and in particular they do not cover the **construction** side of it at +all, the ``_make_*`` methods and the schema ``pack`` path. This module closes +that gap, and it is deliberately built the other way round from its siblings: +they know which packets they want and spell them out, whereas this one asks the +library which codes it claims to support and then exercises every one of them. + +=============================== ============================================= +Capture Codes +=============================== ============================================= +:file:`options-tcp.pcap` TCP options and Multipath TCP subtypes +:file:`options-ipv4.pcap` IPv4 options +:file:`options-ipv6.pcap` HOPOPT, IPv6-Opts and IPv6-Route +:file:`options-transport.pcap` SCTP chunks, parameters and error causes, + and HTTP/2 frames +:file:`options-internet.pcap` Mobility Header messages, options and + extensions, and HIP parameters +=============================== ============================================= + +Each frame carries exactly one item of exactly one code, so a parse failure +names the option that caused it instead of a soup of several. Every frame is an +Ethernet envelope around octets that :mod:`pcapkit` itself constructed, rather +than a packet assembled by :mod:`scapy` from its own protocol model. That is the +point: the octets under test are the library's own output, so a capture is a +record of what this version of ``pcapkit`` builds, and re-parsing it exercises +``_read_*`` against ``_make_*``. + +Only the codes that construct *successfully* can appear in a capture -- there +are no octets for the ones that do not. Roughly a fifth of the space does not, +and that shortfall is not swept up: it is recorded case by case in +:file:`tests/protocols/test_option_roundtrip_unit.py`, against the defect that +causes it. + +Why the case table is not a list of options +------------------------------------------- + +:func:`cases` walks the dispatch registries -- ``TCP.__option__``, +``SCTP.__chunk__``, ``MH.__message__`` and the rest -- so a code registered +tomorrow appears here tomorrow, and the unit test fails until somebody gives it +a case. What the tables below hold is only the *arguments*, which cannot be +derived: each option means something different, so each needs its own value. +Most need nothing at all, since nearly every ``_make_*`` in the tree is fully +defaulted; an entry exists where the default is degenerate (a zero-length +variable field), machine-dependent, or outright rejected. + +Every one of these registries is a :class:`collections.defaultdict` held on a +*class* attribute, so ``registry[code]`` inserts on a miss and permanently grows +a registry shared by every instance in the process. Nothing here subscripts one: +:func:`cases` only ever *iterates*, which touches nothing, and :func:`handler` -- +which resolves one particular code, and so cannot iterate -- goes through +:meth:`ProtocolBase._lookup_registry +` instead. + +IPv4 and HIP have no such registry yet -- they dispatch on the attribute name +``_make_opt_${name}`` / ``_make_param_${name}`` -- so for those two the source +of truth is the enumeration crossed with the presence of the handler, which is +what :func:`_named_registry` builds. + +Determinism +----------- + +The captures have to regenerate byte-identically; a fixture that churns is +worse than none, because every unrelated change then shows up as a diff. Four +things would otherwise vary between runs and are pinned: + +1. Frame timestamps, from :data:`EPOCH` rather than the clock. +2. The order cases are emitted in, from :func:`sort_key` rather than from + registry iteration order -- which is insertion order, and therefore an + implementation detail of whichever module last registered a code. +3. ``MH``'s ``MESG_ID_OPTION_TYPE``, whose ``_make_opt_mesg_id`` falls back to + :meth:`datetime.datetime.now` when given neither ``timestamp`` nor + ``interval``. It is given one below. +4. Anything reading the host. No case here uses PCAP-NG's + ``_make_option_if_os``, which defaults to the live :mod:`platform` string. + +Two deliberate deviations from the sibling generators +---------------------------------------------------- + +Almost every import of :mod:`pcapkit` and of :mod:`scapy` in this module is +inside the function that needs it, which is why :mod:`pylint`'s +``import-outside-toplevel`` is switched off for the file rather than argued with +forty times. Both are load-bearing. :mod:`scapy` is needed only by +:func:`generate`, while the case table and :func:`roundtrip` are useful without +it -- the unit-tier test uses both and has to run on a checkout that installed no +:mod:`scapy`, so a top-level import would make that test skip for a dependency it +never touches. The :mod:`pcapkit` imports are deferred so that importing this +module does not drag in every protocol in the tree, which is what lets the unit +test purge :mod:`pcapkit` and *then* load the case table without the two +orderings fighting. + +:func:`roundtrip` puts a :func:`signal.alarm` deadline around each case. That +is not defensive dressing: GitHub issue #431 is a parser defect that +degenerates into a loop making no progress, ``HOPOPT``'s ``SMF_DPD`` reaches it, +and without a deadline ``make samples`` does not fail -- it *hangs*, taking the +whole build with it. A signal is what interrupts it, because the loop is pure +Python and holds the GIL for the whole of an iteration, so no watchdog thread +would ever get to run. + +""" + +# Deferring these is the design, not an oversight -- see the module docstring. +# pylint: disable=import-outside-toplevel + +from __future__ import annotations + +import collections +import datetime +import pathlib +import signal +import warnings +from typing import TYPE_CHECKING, NamedTuple + +from pcapkit.protocols.protocol import ProtocolBase + +if TYPE_CHECKING: + from typing import Any, Callable, Iterator, Optional + +__all__ = ['generate', 'cases', 'roundtrip', 'outcomes', 'Case', 'Outcome', 'FAMILIES'] + +#: Repository root, i.e. the grandparent of the directory holding this file. +ROOT = pathlib.Path(__file__).resolve().parents[2] +#: Default destination directory for the generated captures. +SAMPLE = ROOT / 'examples' / 'captures' + +#: Capture start time, fixed so that regenerating gives identical files. +EPOCH = 1500000000.0 + +#: Source MAC for every generated frame. +SRC_MAC = '02:00:00:00:00:01' +#: Destination MAC for every generated frame. +DST_MAC = '02:00:00:00:00:02' +#: Source address for every generated IPv4 envelope. +SRC_IP = '192.0.2.1' +#: Destination address for every generated IPv4 envelope. +DST_IP = '198.51.100.1' +#: Source address for every generated IPv6 envelope. +SRC_IP6 = '2001:db8::1' +#: Destination address for every generated IPv6 envelope. +DST_IP6 = '2001:db8::2' + +#: Seconds a single case may take before :func:`roundtrip` gives up on it and +#: records ``'TIMEOUT'``. Generous next to a working case, which takes low +#: single-digit milliseconds, and short enough that a whole sweep still ends. +DEADLINE = 5 + +#: Fixed instant handed to any constructor that would otherwise read the clock. +#: Timezone-aware and in the past, so it is stable and unambiguous. +FIXED_TIME = datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc) + + +class Case(NamedTuple): + """One option-like code, and the arguments that construct it.""" + + #: Family label, e.g. ``'tcp-option'``. Names the registry the code came + #: from and, with :attr:`name`, forms the case's unique label. + family: 'str' + #: Registry key, i.e. the wire code. Usually an enumeration member; for + #: ``PCAPNG.__option__`` a ``(str, int)`` tuple, since the same numeric + #: option code means different things in different block types. + code: 'Any' + #: Human-readable code name, used in the case label and in test ids. + name: 'str' + #: Keyword arguments handed to the constructor. ``{}`` means "every + #: default", which is what most codes need. + kwargs: 'dict[str, Any]' + + @property + def label(self) -> 'str': + """``family/name``, unique across every family.""" + return f'{self.family}/{self.name}' + + +class Outcome(NamedTuple): + """What one construct -> parse -> construct cycle did.""" + + #: The case this describes. + case: 'Case' + #: The step that failed, or ``'OK'`` -- see :data:`STATUSES`. + status: 'str' + #: Exception type and message, or for ``'MISMATCH'`` the two hex strings. + #: Empty for ``'OK'``. + detail: 'str' + #: The octets construction produced, or :data:`None` if it never got that + #: far. Kept as :obj:`bytes` so a caller can put them straight in a frame. + octets: 'Optional[bytes]' + #: Warning messages raised anywhere in the cycle, deduplicated but in + #: order. A case can be ``'OK'`` and still warn, and that is worth seeing: + #: a silent ``SchemaWarning: packet length < 0`` means an option over-read + #: and the octets around it are not what they appear to be. + warnings: 'tuple[str, ...]' + + +#: Statuses whose octets may go into a capture, i.e. the ones whose *parse* +#: step ran to completion. +#: +#: This is a safety property, not a tidiness one. ``HOPOPT``'s ``SMF_DPD`` +#: constructs perfectly well and then spins forever on the way back in, so its +#: octets are available and putting them in a fixture would produce a capture +#: that wedges every reader of it -- including the rest of this suite. A +#: ``CONSTRUCT`` or ``PARSE`` failure is milder but no more useful: the frame +#: would be one no reader can get through. So a capture carries only frames +#: that are known to parse, and the cases that do not are recorded in +#: :file:`tests/protocols/test_option_roundtrip_unit.py` instead, where a +#: failure is an assertion rather than a hang. +CAPTURABLE = ('OK', 'RECONSTRUCT', 'MISMATCH') + +#: Every value :attr:`Outcome.status` can take. +#: +#: * ``'OK'`` -- constructed, parsed, reconstructed, and the octets matched. +#: * ``'CONSTRUCT'`` -- the constructor raised. Note that +#: :meth:`ProtocolBase.__post_init__` packs *and then* unpacks, so a +#: ``_read_*`` fault on a perfectly good pack also lands here. +#: * ``'PARSE'`` -- the octets would not parse back. +#: * ``'RECONSTRUCT'`` -- ``_make_*`` could not consume what ``_read_*`` +#: produced. This is the step a construct-then-parse test cannot see. +#: * ``'MISMATCH'`` -- both directions worked and the octets differed. +#: * ``'TIMEOUT'`` -- the case did not finish inside :data:`DEADLINE`. +STATUSES = ('OK', 'CONSTRUCT', 'PARSE', 'RECONSTRUCT', 'MISMATCH', 'TIMEOUT') + + +class Family(NamedTuple): + """A registry, and how to exercise the codes in it.""" + + #: Family label, copied into every :class:`Case` it yields. + label: 'str' + #: Zero-argument callable returning the dispatch registry, or a mapping + #: standing in for one. Deferred so that importing this module does not + #: import every protocol in the tree. + registry: 'Callable[[], Any]' + #: Zero-argument callable returning per-code keyword overrides, keyed by + #: registry key. A code absent from the mapping is constructed with no + #: arguments at all. Deferred for the same reason as :attr:`registry`. + overrides: 'Callable[[], dict[Any, dict[str, Any]]]' + #: ``build(code, kwargs)`` -> a constructed protocol carrying exactly one + #: item of ``code``. + build: 'Callable[[Any, dict[str, Any]], Any]' + #: ``parse(octets)`` -> the protocol, parsed back from its own octets. + parse: 'Callable[[bytes], Any]' + #: ``extract(parsed)`` -> the parsed collection, in whatever form + #: :attr:`rebuild` takes. + extract: 'Callable[[Any], Any]' + #: ``rebuild(collection)`` -> a protocol constructed from the *parsed* + #: collection rather than from keyword arguments. This is the step that + #: catches a ``_make_*`` unable to consume what its own ``_read_*`` made. + rebuild: 'Callable[[Any], Any]' + #: Name of the capture this family's frames go into, or :data:`None` for a + #: family that is exercised but not captured -- see :data:`FAMILIES`. + capture: 'Optional[str]' + #: How the octets are wrapped into a frame -- see :func:`_frame`. + envelope: 'Optional[str]' + #: Next-layer protocol number for the envelope, where it needs one. + proto: 'Optional[int]' = None + + +def sort_key(code: 'Any') -> 'tuple[int, str, int]': + """Total order over registry keys, stable across runs. + + Registry iteration order is insertion order, which is an implementation + detail of whichever module last registered a code -- so ordering emitted + frames by it would let an unrelated import rewrite every capture. This + orders by the numeric code where there is one, and falls back to the string + form for the ``(str, int)`` keys of ``PCAPNG.__option__``. + + Args: + code: Registry key: an enumeration member, an :obj:`int`, or a + ``(str, int)`` tuple. + + Returns: + A tuple safe to compare against this function's output for any other + key, with numeric-keyed codes sorted ahead of tuple-keyed ones. + + """ + if isinstance(code, tuple): + name, number = code + return (1, str(name), int(number)) + try: + return (0, '', int(code)) + except (TypeError, ValueError): # pragma: no cover + return (2, str(code), 0) + + +def handler(registry: 'Any', code: 'Any') -> 'Any': + """The handler a registry holds for ``code``, without recording a miss. + + Enumerating a registry never needs this -- iterating a + :class:`collections.defaultdict` touches nothing, which is why :func:`cases` + only ever iterates. Asking what a *particular* code resolves to does need + it, because ``registry[code]`` on a miss inserts the key and permanently + grows a registry shared by every instance of the class in the process. The + inserted value is whatever the default factory would have produced anyway, + so it buys nothing and costs a spurious "already registered" warning from + the next genuine ``register`` call for that code. + + Args: + registry: A dispatch registry, or a mapping standing in for one. + code: Registry key to resolve. + + Returns: + The handler registered for ``code``, or the registry's fallback. + + """ + return ProtocolBase._lookup_registry(registry, code) # pylint: disable=protected-access + + +def code_name(code: 'Any') -> 'str': + """Render a registry key as a name usable in a test id. + + Args: + code: Registry key. + + Returns: + ``code.name`` for an enumeration member, ``'_'`` for a + ``(str, int)`` key, and the plain string form for anything else. + + """ + if isinstance(code, tuple): + name, number = code + return f'{name}_{number}' + return getattr(code, 'name', None) or str(code) + + +def _named_registry(owner: 'type', enum: 'Any', prefix: 'str', + attribute: 'str') -> 'Any': + """A protocol's dispatch registry, or a stand-in derived from its handlers. + + ``IPv4`` and ``HIP`` were the last two protocols to dispatch their options + and parameters by attribute name rather than through a registry, and for + those there was nothing to enumerate. The equivalent source of truth is the + enumeration crossed with the presence of the handler, which is what this + builds when the registry is absent. + + GitHub pull request #434 gave both of them a real registry, so the fallback + is no longer taken for either -- it is kept because it costs nothing and is + the only thing that would notice a protocol added tomorrow without one. Both + shapes are a :class:`collections.defaultdict`, so :func:`cases` and + :func:`handler` cannot tell which one they were given -- and in particular + :meth:`ProtocolBase._lookup_registry + ` still has a + ``default_factory`` to fall back on. + + Args: + owner: Protocol class holding the handler methods. + enum: Enumeration of wire codes to consider. + prefix: Handler-name prefix, e.g. ``'_make_opt_'``. + attribute: Name the real registry would be held under, e.g. + ``'__option__'``. + + Returns: + The registry, real or derived. + + """ + existing = getattr(owner, attribute, None) + if existing is not None: + return existing + + found = collections.defaultdict(lambda: None) # type: Any + for code in enum: + name = code.name.lower() + if hasattr(owner, f'{prefix}{name}'): + found[code] = name + return found + + +############################################################################### +# TCP -- options, and Multipath TCP subtypes +############################################################################### + +#: Header fields shared by every constructed TCP segment. Only ``options`` +#: varies between cases, so a difference in the octets is a difference in the +#: option and nothing else. +TCP_BASE = { + 'srcport': 50000, 'dstport': 80, 'seq': 1, 'ack': 0, + 'ns': False, 'cwr': False, 'ece': False, 'urg': False, 'ack_flag': False, + 'psh': False, 'rst': False, 'syn': True, 'fin': False, + 'window': 8192, 'checksum': b'\x00\x00', 'urgent_pointer': 0, + 'payload': b'', +} + + +def _tcp_registry() -> 'Any': + from pcapkit.protocols.transport.tcp import TCP + return TCP.__option__ + + +def _tcp_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.tcp.option import Option as Enum_Option + return { + # Each of these four defaults to an empty variable-length field, which + # is the one shape that never runs the field's packing loop. + Enum_Option.SACK: {'sack': [(1, 2), (3, 4)]}, + Enum_Option.TCP_Alternate_Checksum_Data: {'data': b'\x01\x02'}, + Enum_Option.TCP_Authentication_Option: {'mac': b'\x01\x02\x03\x04'}, + Enum_Option.TCP_Fast_Open_Cookie: {'cookie': b'\x01\x02\x03\x04\x05\x06\x07\x08'}, + } + + +def _tcp_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.protocols.transport.tcp import TCP + return TCP(options=[(code, kwargs)], **TCP_BASE) + + +def _tcp_parse(octets: 'bytes') -> 'Any': + from pcapkit.protocols.transport.tcp import TCP + return TCP(octets, len(octets)) + + +def _tcp_extract(parsed: 'Any') -> 'Any': + return parsed.info.options + + +def _tcp_rebuild(options: 'Any') -> 'Any': + from pcapkit.protocols.transport.tcp import TCP + return TCP(options=options, **TCP_BASE) + + +def _mptcp_registry() -> 'Any': + from pcapkit.protocols.transport.tcp import TCP + return TCP.__mp_option__ + + +def _mptcp_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.tcp.mp_tcp_option import MPTCPOption as Enum_MPTCPOption + return { + Enum_MPTCPOption.MP_CAPABLE: {'skey': 0x0102030405060708}, + Enum_MPTCPOption.DSS: { + 'ack': 1, 'dsn': 2, 'ssn': 3, 'dl_len': 4, 'checksum': b'\x00\x00'}, + Enum_MPTCPOption.ADD_ADDR: {'addr_id': 1, 'addr': '192.0.2.1'}, + Enum_MPTCPOption.REMOVE_ADDR: {'addr_id': [1]}, + Enum_MPTCPOption.MP_PRIO: {'addr_id': 1}, + Enum_MPTCPOption.MP_FAIL: {'dsn': 7}, + Enum_MPTCPOption.MP_FASTCLOSE: {'key': 9}, + } + + +def _mptcp_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.const.tcp.option import Option as Enum_Option + from pcapkit.protocols.transport.tcp import TCP + + # The subtype travels as an argument of the enclosing Multipath TCP option + # rather than as a registry key of its own, so it is spliced in here rather + # than repeated in every entry of the table above. + args = dict(kwargs) + args['subtype'] = code + return TCP(options=[(Enum_Option.Multipath_TCP, args)], **TCP_BASE) + + +############################################################################### +# IPv4 -- options +############################################################################### + +#: Header fields shared by every constructed IPv4 packet. +IPV4_BASE = {'protocol': 6, 'src': '192.0.2.1', 'dst': '198.51.100.1', 'payload': b''} + + +def _ipv4_registry() -> 'Any': + from pcapkit.const.ipv4.option_number import OptionNumber + from pcapkit.protocols.internet.ipv4 import IPv4 + return _named_registry(IPv4, OptionNumber, '_make_opt_', '__option__') + + +def _ipv4_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.ipv4.option_number import OptionNumber + from pcapkit.const.ipv4.protection_authority import ProtectionAuthority + return { + # A single authority whose value is 0 makes ``_make_opt_sec`` compute a + # zero-octet bitmap and then index into it; two keeps it non-empty. + OptionNumber.SEC: {'authorities': [ProtectionAuthority.GENSER, + ProtectionAuthority.NSA]}, + # ``counts=10``, the default, needs 43 option octets, which overflows + # the 4-bit ``ihl``. Nine is the largest an IPv4 header can carry. + OptionNumber.LSR: {'counts': 9}, + OptionNumber.RR: {'counts': 9}, + OptionNumber.SSR: {'counts': 9}, + # ``timestamp=None``, the default, is rejected outright. + OptionNumber.TS: {'counts': 1, 'timestamp': [1]}, + OptionNumber.E_SEC: {'info': b'\x00'}, + } + + +def _ipv4_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.protocols.internet.ipv4 import IPv4 + return IPv4(options=[(code, kwargs)], **IPV4_BASE) + + +def _ipv4_parse(octets: 'bytes') -> 'Any': + from pcapkit.protocols.internet.ipv4 import IPv4 + return IPv4(octets, len(octets)) + + +def _ipv4_extract(parsed: 'Any') -> 'Any': + from pcapkit.corekit.multidict import OrderedMultiDict + + # EOOL and NOP are dropped by ``_make_ipv4_options`` as padding, so the + # header they produce has no options at all and ``info`` carries no + # ``options`` attribute. An empty collection is the honest reading of that, + # and it keeps the two padding codes from being reported as parse failures. + return getattr(parsed.info, 'options', OrderedMultiDict()) + + +def _ipv4_rebuild(options: 'Any') -> 'Any': + from pcapkit.protocols.internet.ipv4 import IPv4 + return IPv4(options=options, **IPV4_BASE) + + +############################################################################### +# IPv6 extension headers -- HOPOPT, IPv6-Opts, IPv6-Route +############################################################################### + + +def _hopopt_registry() -> 'Any': + from pcapkit.protocols.internet.hopopt import HOPOPT + return HOPOPT.__option__ + + +def _ipv6_opts_registry() -> 'Any': + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + return IPv6_Opts.__option__ + + +def _ipv6_option_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.ipv6.option import Option as Enum_Option + + # HOPOPT and IPv6-Opts key on the same enumeration and behave identically + # on every code, so one table serves both. + return { + # ``nonce=0`` gives a zero-octet nonce. Only widths 3, 5, 6 and 7 pack + # at all, because a callable-length NumberField never clears the flag + # that says "hand struct a bytes"; 0xFFFFFF is three octets. + Enum_Option.ILNP_Nonce: {'nonce': 0xFFFFFF}, + Enum_Option.Line_Identification_Option: {'id': b'line-1'}, + } + + +def _hopopt_base() -> 'dict[str, Any]': + from pcapkit.const.reg.transtype import TransType + return {'next': TransType.UDP, 'payload': b''} + + +def _hopopt_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.protocols.internet.hopopt import HOPOPT + return HOPOPT(options=[(code, kwargs)], **_hopopt_base()) + + +def _hopopt_parse(octets: 'bytes') -> 'Any': + from pcapkit.protocols.internet.hopopt import HOPOPT + return HOPOPT(octets, len(octets), extension=True) + + +def _ipv6_extract(parsed: 'Any') -> 'Any': + from pcapkit.corekit.multidict import OrderedMultiDict + return getattr(parsed.info, 'options', OrderedMultiDict()) + + +def _hopopt_rebuild(options: 'Any') -> 'Any': + from pcapkit.protocols.internet.hopopt import HOPOPT + return HOPOPT(options=options, **_hopopt_base()) + + +def _ipv6_opts_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + return IPv6_Opts(options=[(code, kwargs)], **_hopopt_base()) + + +def _ipv6_opts_parse(octets: 'bytes') -> 'Any': + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + return IPv6_Opts(octets, len(octets), extension=True) + + +def _ipv6_opts_rebuild(options: 'Any') -> 'Any': + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + return IPv6_Opts(options=options, **_hopopt_base()) + + +def _ipv6_route_registry() -> 'Any': + from pcapkit.protocols.internet.ipv6_route import IPv6_Route + return IPv6_Route.__routing__ + + +def _ipv6_route_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.ipv6.routing import Routing as Enum_Routing + return { + Enum_Routing.Source_Route: {'ip': ['2001:db8::1', '2001:db8::2']}, + Enum_Routing.Type_2_Routing_Header: {'ip': '2001:db8::2'}, + Enum_Routing.RPL_Source_Route_Header: {'ip': ['2001:db8::1', '2001:db8::2']}, + } + + +def _ipv6_route_base() -> 'dict[str, Any]': + from pcapkit.const.reg.transtype import TransType + return {'next': TransType.UDP, 'seg_left': 0, 'payload': b''} + + +def _ipv6_route_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.protocols.internet.ipv6_route import IPv6_Route + return IPv6_Route(type=code, data=dict(kwargs), **_ipv6_route_base()) + + +def _ipv6_route_parse(octets: 'bytes') -> 'Any': + from pcapkit.protocols.internet.ipv6_route import IPv6_Route + return IPv6_Route(octets, len(octets), extension=True) + + +def _ipv6_route_extract(parsed: 'Any') -> 'Any': + # A routing header carries one routing type, not a collection of them, so + # the "collection" handed on is the message itself. + return parsed.info + + +def _ipv6_route_rebuild(info: 'Any') -> 'Any': + from pcapkit.protocols.internet.ipv6_route import IPv6_Route + return IPv6_Route(type=info.type, data=info, next=info.next, + seg_left=info.seg_left, payload=b'') + + +############################################################################### +# SCTP -- chunks, parameters and error causes +############################################################################### + +#: Header fields shared by every constructed SCTP packet. ``chksum`` is pinned +#: rather than left to the CRC32c path, so a mismatch points at the chunk under +#: test and not at a recomputed checksum. +SCTP_BASE = { + 'srcport': 50000, 'dstport': 80, 'vtag': 0x11223344, + 'chksum': b'\x00\x00\x00\x00', +} + + +def _sctp_chunk_registry() -> 'Any': + from pcapkit.protocols.transport.sctp import SCTP + return SCTP.__chunk__ + + +def _sctp_chunk_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.sctp.cause_code import CauseCode as Enum_CauseCode + from pcapkit.const.sctp.chunk import Chunk as Enum_Chunk + from pcapkit.const.sctp.parameter import Parameter as Enum_Parameter + + # The three-octet payloads are deliberate: they make each item's length + # 3 (mod 4), so every case runs the PaddingField path instead of landing + # already aligned. + return { + # ``data=b''`` is rejected outright -- RFC 9260 s3.3.1 wants at least + # one octet of user data -- so this override is required, not cosmetic. + Enum_Chunk.Payload_Data: {'data': b'\x01\x02\x03'}, + Enum_Chunk.Initiation: { + 'parameters': [(Enum_Parameter.State_Cookie, {'cookie': b'\x01\x02\x03'})]}, + Enum_Chunk.Initiation_Acknowledgement: { + 'parameters': [(Enum_Parameter.State_Cookie, {'cookie': b'\x01\x02\x03'})]}, + Enum_Chunk.Selective_Acknowledgement: {'gap_blocks': [(1, 2)], 'dup_tsn': [3]}, + # RFC 9260 s3.3.5 and s3.3.6 mandate exactly one Heartbeat Info + # parameter, so the default -- none at all -- is structurally invalid. + Enum_Chunk.Heartbeat_Request: { + 'parameters': [(Enum_Parameter.Heartbeat_Info, {'info': b'\x01\x02\x03'})]}, + Enum_Chunk.Heartbeat_Acknowledgement: { + 'parameters': [(Enum_Parameter.Heartbeat_Info, {'info': b'\x01\x02\x03'})]}, + Enum_Chunk.Abort: { + 'error': [(Enum_CauseCode.User_Initiated_Abort, {'info': b'\x01\x02\x03'})]}, + # ``_make_chunk_error`` documents "one or more error causes"; the + # default gives zero. + Enum_Chunk.Operation_Error: { + 'error': [(Enum_CauseCode.Protocol_Violation, {'info': b'\x01\x02\x03'})]}, + Enum_Chunk.State_Cookie: {'cookie': b'\x01\x02\x03'}, + } + + +def _sctp_chunk_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.protocols.transport.sctp import SCTP + return SCTP(chunks=[(code, kwargs)], **SCTP_BASE) + + +def _sctp_parse(octets: 'bytes') -> 'Any': + from pcapkit.protocols.transport.sctp import SCTP + return SCTP(octets, len(octets)) + + +def _sctp_extract(parsed: 'Any') -> 'Any': + return parsed.info.chunks + + +def _sctp_rebuild(chunks: 'Any') -> 'Any': + from pcapkit.protocols.transport.sctp import SCTP + return SCTP(chunks=chunks, **SCTP_BASE) + + +def _sctp_param_registry() -> 'Any': + from pcapkit.protocols.transport.sctp import SCTP + return SCTP.__parameter__ + + +def _sctp_param_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.sctp.parameter import Parameter as Enum_Parameter + return { + Enum_Parameter.Heartbeat_Info: {'info': b'\x01\x02\x03'}, + Enum_Parameter.IPv4_Address: {'address': '192.0.2.1'}, + Enum_Parameter.IPv6_Address: {'address': '2001:db8::1'}, + Enum_Parameter.State_Cookie: {'cookie': b'\x01\x02\x03'}, + # A complete unrecognised TLV -- type 0xffff, length 5, one value + # octet -- rather than the empty default, which is not a TLV at all. + Enum_Parameter.Unrecognized_Parameter: {'value': b'\xff\xff\x00\x05\x01'}, + Enum_Parameter.Host_Name_Address: {'name': b'localhost\x00'}, + Enum_Parameter.Supported_Address_Types: { + 'types': [Enum_Parameter.IPv4_Address, Enum_Parameter.IPv6_Address]}, + } + + +def _sctp_param_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.const.sctp.chunk import Chunk as Enum_Chunk + from pcapkit.protocols.transport.sctp import SCTP + + # A parameter is not a top-level item: it travels inside a chunk, and + # HEARTBEAT is the smallest chunk that carries an arbitrary one. + return SCTP(chunks=[(Enum_Chunk.Heartbeat_Request, + {'parameters': [(code, kwargs)]})], **SCTP_BASE) + + +def _sctp_cause_registry() -> 'Any': + from pcapkit.protocols.transport.sctp import SCTP + return SCTP.__cause__ + + +def _sctp_cause_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.sctp.cause_code import CauseCode as Enum_CauseCode + from pcapkit.const.sctp.parameter import Parameter as Enum_Parameter + return { + Enum_CauseCode.Missing_Mandatory_Parameter: { + 'types': [Enum_Parameter.State_Cookie]}, + # The four causes below carry an embedded TLV, and each one's default + # is a zero-length value, i.e. not a TLV. + Enum_CauseCode.Unresolvable_Address: { + 'value': b'\x00\x0b\x00\x0elocalhost\x00'}, + Enum_CauseCode.Unrecognized_Chunk_Type: {'value': b'\xff\x00\x00\x05\x01'}, + Enum_CauseCode.Unrecognized_Parameters: {'value': b'\xff\xff\x00\x05\x01'}, + Enum_CauseCode.Restart_of_an_Association_with_New_Addresses: { + 'value': b'\x00\x05\x00\x08\xc0\x00\x02\x01'}, + Enum_CauseCode.User_Initiated_Abort: {'info': b'\x01\x02\x03'}, + Enum_CauseCode.Protocol_Violation: {'info': b'\x01\x02\x03'}, + } + + +def _sctp_cause_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.const.sctp.chunk import Chunk as Enum_Chunk + from pcapkit.protocols.transport.sctp import SCTP + + # An error cause travels inside an ERROR (or ABORT) chunk, under the + # keyword ``error`` rather than ``causes``. + return SCTP(chunks=[(Enum_Chunk.Operation_Error, + {'error': [(code, kwargs)]})], **SCTP_BASE) + + +############################################################################### +# Mobility Header -- messages, options and extensions +############################################################################### + +#: Header fields shared by every constructed Mobility Header. ``next`` is +#: IPv6-NoNxt, so nothing downstream has to be constructed as well. +MH_BASE = {'next': 59, 'chksum': b'\x00\x00', 'payload': b''} + + +def _mh_message_registry() -> 'Any': + from pcapkit.protocols.internet.mh import MH + return MH.__message__ + + +def _mh_message_overrides() -> 'dict[Any, dict[str, Any]]': + return {} + + +def _mh_message_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.protocols.internet.mh import MH + + # ``data`` has to be a mapping for ``make`` to dispatch on the message + # type; a Data object takes the other branch, which is what rebuild uses. + return MH(type=code, data=dict(kwargs), **MH_BASE) + + +def _mh_parse(octets: 'bytes') -> 'Any': + from pcapkit.protocols.internet.mh import MH + return MH(octets, len(octets), extension=True) + + +def _mh_message_extract(parsed: 'Any') -> 'Any': + return parsed.info + + +def _mh_message_rebuild(message: 'Any') -> 'Any': + from pcapkit.protocols.internet.mh import MH + return MH(type=message.type, data=message, **MH_BASE) + + +def _mh_option_registry() -> 'Any': + from pcapkit.protocols.internet.mh import MH + return MH.__option__ + + +def _mh_option_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.mh.option import Option as Enum_Option + import ipaddress + return { + # ``{}`` makes ``_make_opt_pad`` warn and silently emit a Pad1, so the + # PadN case would otherwise be a duplicate of the Pad1 one. + Enum_Option.PadN: {'length': 4}, + # The authenticator has to be a multiple of 8 octets. + Enum_Option.Authorization_Data: {'data': b'\xa5' * 8}, + Enum_Option.Mobility_Header_Link_Layer_Address_option: { + 'address': b'\x00\x11\x22\x33\x44\x55'}, + # The default identifier is the *string* ``'::'``, whose ``len()`` is 2 + # rather than the 16 octets the field packs -- so the declared length + # is 3 going out and 17 coming back. An address object avoids it. + Enum_Option.MN_ID_OPTION_TYPE: {'identifier': ipaddress.IPv6Address('::1')}, + # ``(len + 6) % 4`` has to be 0. + Enum_Option.AUTH_OPTION_TYPE: {'data': b'\xa5\xa5'}, + # Without this ``_make_opt_mesg_id`` reads the clock, and the capture + # would differ on every regeneration. + Enum_Option.MESG_ID_OPTION_TYPE: {'interval': FIXED_TIME}, + Enum_Option.Signature: {'signature': b'\xaa' * 4}, + Enum_Option.Permanent_Home_Keygen_Token: {'token': b'\xbb' * 8}, + Enum_Option.Experimental_Mobility_Option: {'data': b'\xcc' * 4}, + Enum_Option.Binding_Authorization_Data_for_FMIPv6: {'data': b'\xb5'}, + } + + +def _mh_carrier() -> 'Any': + from pcapkit.const.mh.packet import Packet as Enum_Packet + + # Binding Refresh Request is the smallest message that carries an + # arbitrary option list: two reserved octets and then the options. + return Enum_Packet.Binding_Refresh_Request + + +def _mh_option_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.protocols.internet.mh import MH + return MH(type=_mh_carrier(), data={'options': [(code, dict(kwargs))]}, **MH_BASE) + + +def _mh_option_extract(parsed: 'Any') -> 'Any': + from pcapkit.corekit.multidict import OrderedMultiDict + return getattr(parsed.info, 'options', OrderedMultiDict()) + + +def _mh_option_rebuild(options: 'Any') -> 'Any': + from pcapkit.protocols.internet.mh import MH + return MH(type=_mh_carrier(), data={'options': options}, **MH_BASE) + + +def _mh_extension_registry() -> 'Any': + from pcapkit.protocols.internet.mh import MH + return MH.__extension__ + + +def _mh_extension_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.mh.cga_extension import CGAExtension as Enum_CGAExtension + return {Enum_CGAExtension.Multi_Prefix: {'prefixes': [0x20010db800000001]}} + + +def _mh_extension_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.const.mh.cga_type import CGAType as Enum_CGAType + from pcapkit.const.mh.option import Option as Enum_Option + from pcapkit.protocols.data.internet.mh import CGAParameter as Data_CGAParameter + from pcapkit.protocols.internet.mh import MH + + # A CGA extension has exactly one carrier: the ``extensions`` list of a CGA + # parameter, inside a CGA Parameters option. It has to be a *Data* model + # rather than a schema, because ``_make_opt_cga_param`` keeps only the + # length of the extensions it builds for a schema and throws the schemas + # themselves away. + # The data model annotates ``modifier`` as a CGAType and ``extensions`` as an + # already-built OrderedMultiDict, but the constructor this feeds takes the + # tuple form and resolves both -- so the arguments go through an untyped + # mapping rather than being spelled as keywords a checker would reject. + fields = { + 'modifier': Enum_CGAType.Tag_086F_CA5E_10B2_00C9_9C8C_E001_6427_7C08, + 'prefix': 0x20010db800000000, + 'collision_count': 0, + 'public_key': b'\x30\x02\xaa\xbb', + 'extensions': [(code, dict(kwargs))], + } # type: dict[str, Any] + parameter = Data_CGAParameter(**fields) + return MH(type=_mh_carrier(), + data={'options': [(Enum_Option.CGA_Parameters, + {'parameters': [parameter]})]}, + **MH_BASE) + + +############################################################################### +# HIP -- parameters +############################################################################### + +#: Header fields shared by every constructed HIP packet. +HIP_BASE = { + 'next': 6, 'packet': 1, 'version': 2, 'checksum': b'\x00\x00', + 'controls_anonymous': False, 'shit': 0, 'rhit': 0, 'payload': b'', +} + +#: Codes that are HIPv1-only, and the version their constructor demands. +HIP_VERSION = {129: 2, 128: 1} + +#: How many copies of the parameter under test go in one packet. +#: +#: Two, not one, and this is not padding for its own sake. ``HIP.make`` +#: computes the header's ``len`` field as ``total_length // 8 + 4``, which is +#: only lossless when the parameter octets are a multiple of eight -- and the +#: parameter padding rule pads the *contents* to eight, ignoring the four-octet +#: type-and-length header, so a single parameter is always ``4 (mod 8)`` and +#: always loses those four octets. ``_read_hip_param`` then checks the value +#: exactly and rejects the packet. Two copies sum to a multiple of eight, so +#: the arithmetic is exact and the parameter constructors become reachable: +#: measured, this is the difference between 4 and 29 of the 49 codes +#: round-tripping. The single-parameter case is not lost -- it is what +#: ``hip-parameter-single`` in the test's expected-failure table records. +HIP_COPIES = 2 + + +def _hip_registry() -> 'Any': + from pcapkit.const.hip.parameter import Parameter + from pcapkit.protocols.internet.hip import HIP + return _named_registry(HIP, Parameter, '_make_param_', '__parameter__') + + +def _hip_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.hip.ecdsa_curve import ECDSACurve + from pcapkit.const.hip.parameter import Parameter + return { + # ``lifetime=0`` reaches ``math.log2(0)``. + Parameter.PUZZLE: {'lifetime': 1}, + Parameter.SOLUTION: {'lifetime': 1}, + # ``hi_curve=None`` falls through to an explicit raise. + Parameter.HOST_ID: {'hi_curve': ECDSACurve.NIST_P_256}, + } + + +def _hip_base(code: 'Any') -> 'dict[str, Any]': + base = dict(HIP_BASE) + base['version'] = HIP_VERSION.get(int(code), 2) + return base + + +def _hip_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.protocols.internet.hip import HIP + + # ``extension=True`` is not decoration. ``HIP.alias`` reads ``self._info``, + # which ``__post_init__`` only assigns after ``read`` returns, and the sole + # path that returns before touching it is the ``if extension`` early exit. + # Without it every one of the 49 codes fails with ``AttributeError: 'HIP' + # object has no attribute '_info'``. + return HIP(parameters=[(code, kwargs)] * HIP_COPIES, + extension=True, **_hip_base(code)) + + +def _hip_parse(octets: 'bytes') -> 'Any': + from pcapkit.protocols.internet.hip import HIP + return HIP(octets, len(octets), extension=True) + + +def _hip_extract(parsed: 'Any') -> 'Any': + from pcapkit.corekit.multidict import OrderedMultiDict + return getattr(parsed.info, 'parameters', OrderedMultiDict()) + + +def _hip_rebuild(parameters: 'Any') -> 'Any': + from pcapkit.protocols.internet.hip import HIP + + # ``parameters`` is the parsed collection, so the version has to come off + # one of its members rather than from the code being tested. + first = next(iter(parameters), None) + base = dict(HIP_BASE) + if first is not None: + base['version'] = HIP_VERSION.get(int(first), 2) + return HIP(parameters=parameters, extension=True, **base) + + +############################################################################### +# HTTP/2 -- frames +############################################################################### + + +def _httpv2_registry() -> 'Any': + from pcapkit.protocols.application.httpv2 import HTTP + return HTTP.__frame__ + + +def _httpv2_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.http.frame import Frame as Enum_Frame + from pcapkit.const.http.setting import Setting as Enum_Setting + return { + # ``settings=None``, the default, matches none of the accepted forms. + Enum_Frame.SETTINGS: {'settings': [(Enum_Setting.HEADER_TABLE_SIZE, 4096)]}, + Enum_Frame.GOAWAY: {'debug_data': b'\xde\xad'}, + } + + +def _httpv2_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.protocols.application.httpv2 import HTTP + + # SETTINGS and PING both reject a non-zero stream identifier, so every + # frame uses stream 0 rather than each one picking its own. + return HTTP(type=code, sid=0, frame=dict(kwargs)) + + +def _httpv2_parse(octets: 'bytes') -> 'Any': + from pcapkit.protocols.application.httpv2 import HTTP + return HTTP(octets, len(octets)) + + +def _httpv2_extract(parsed: 'Any') -> 'Any': + return parsed.info + + +def _httpv2_rebuild(frame: 'Any') -> 'Any': + from pcapkit.protocols.application.httpv2 import HTTP + + # ``flags`` is deliberately not passed on: ``HTTP.make`` rebinds it from + # the ``_make_http_*`` return whenever ``frame`` is a mapping or a Data + # object, so passing it would be silently ignored. + return HTTP(type=frame.type, sid=frame.sid, frame=frame) + + +############################################################################### +# PCAP-NG -- blocks, block options, name records and decryption secrets +############################################################################### +# +# These four families are exercised by :func:`roundtrip` but deliberately write +# no capture, and the reason is not that a PCAP-NG capture would be +# uninteresting -- it is that the construction API cannot produce a +# *reproducible* one. ``_make_block_shb`` takes the section's byte order from +# :data:`sys.byteorder` with no way to override it, and the TLS and WireGuard +# key-log writers stamp :meth:`datetime.datetime.now` into the block body with +# no keyword to pin it. A fixture built this way would differ between a +# little-endian and a big-endian host, and the two key-log types would differ +# between one run and the next. +# +# The PCAP-NG option space is not left uncovered by that decision: +# :file:`examples/generators/pcapng.py` already builds ``profile.pcapng``, +# ``test.pcapng`` and ``many_interfaces.pcapng`` over it, hand-rolling every +# octet with :func:`struct.pack` for exactly this reason. What is added here is +# the half that generator cannot reach, since it never calls the construction +# API at all: whether ``_make_block_*`` and ``_make_option_*`` can produce what +# ``_read_block_*`` and ``_read_option_*`` consume. + +#: Fixed PCAP-NG timestamp, the same instant :file:`pcapng.py` uses. Not +#: optional: leaving ``timestamp`` unset reaches ``self._info`` before +#: ``__post_init__`` has assigned it, so the "now" default is unreachable as +#: well as non-deterministic. +PCAPNG_TIME = 1500000000 + +#: A minimal Ethernet frame for the packet-carrying blocks: destination and +#: source address, an unassigned EtherType, and two octets of body. +#: +#: It has to be non-empty. A ``PayloadField`` whose computed length is +#: legitimately zero reads the whole remainder of the block instead of nothing, +#: which swallows the option area -- so an empty packet would silently take the +#: whole ``epb_*`` and ``pack_*`` option space with it. Sixteen octets rather +#: than the bare 14-octet header, so the Ethernet dissector does not under-run. +#: 6 octets of destination, 6 of source, the EtherType, then the body. +PCAPNG_PACKET = bytes.fromhex('02000000000102000000000288b50000') + +#: A systemd journal export entry whose length is a multiple of four. +#: ``_make_block_systemd`` adds no padding, unlike every other block, so an +#: unaligned entry makes the block length invalid. +PCAPNG_JOURNAL = (b'__REALTIME_TIMESTAMP=1500000000000000\n_TRANSPORT=journal\n' + b'PRIORITY=6\nMESSAGE=probe.....\n\n') + + +def _pcapng_context() -> 'tuple[Any, Any]': + """A one-section PCAP-NG context with one Ethernet interface in it. + + Every block other than a Section Header Block is read and written relative + to a section, and the packet-carrying blocks resolve their snap length + through the section's interface list, so there has to be one. + + Returns: + The :class:`~pcapkit.foundation.engines.pcapng.Context` and the + interface-description block registered in it. + + """ + from pcapkit.const.pcapng.block_type import BlockType as Enum_BlockType + from pcapkit.const.reg.linktype import LinkType as Enum_LinkType + from pcapkit.foundation.engines.pcapng import Context + from pcapkit.protocols.misc.pcapng import PCAPNG + + section = PCAPNG(num=0, sct=1, ctx=None, + type=Enum_BlockType.Section_Header_Block, block={}) + # ``Protocol.info`` is annotated as the generic data model for the protocol, + # so a checker cannot see that a Section Header Block's is the one ``Context`` + # wants. The engine does exactly this, at + # pcapkit/foundation/engines/pcapng.py:148. + section_info = section.info # type: Any + context = Context(section_info) + interface = PCAPNG(num=1, sct=1, ctx=context, + type=Enum_BlockType.Interface_Description_Block, + block={'linktype': Enum_LinkType.ETHERNET, 'snaplen': 0x40000}) + interface_info = interface.info # type: Any + context.interfaces.append(interface_info) + return context, interface_info + + +def _pcapng_block(code: 'Any', block: 'dict[str, Any] | Any') -> 'Any': + """Construct one PCAP-NG block of ``code``. + + A fresh context per call, deliberately. ``PCAPNG.__post_init__`` packs and + then re-reads *on the same instance*, and the two halves share the + per-instance option counter, so state left by one case would change the + next one's result. + + Args: + code: Block type. + block: Block body, as a mapping of constructor arguments or as the + parsed data model. + + Returns: + The constructed block. + + """ + from pcapkit.const.pcapng.block_type import BlockType as Enum_BlockType + from pcapkit.protocols.misc.pcapng import PCAPNG + + # A Section Header Block is what *starts* a section, so it is the one block + # the engine reads with no context at all. + context = None if code == Enum_BlockType.Section_Header_Block else _pcapng_context()[0] + return PCAPNG(num=2, sct=1, ctx=context, type=code, block=block) + + +#: Block type of a Section Header Block, as it appears on the wire. The value +#: is byte-order independent by design -- that is what lets a reader work out a +#: section's endianness -- so it can be recognised before the byte order is +#: known, which is exactly what :func:`_pcapng_parse` needs. +PCAPNG_SHB_MAGIC = bytes.fromhex('0a0d0d0a') + + +def _pcapng_parse(octets: 'bytes') -> 'Any': + """Parse one PCAP-NG block back from its own octets. + + The block type is read out of the octets rather than passed in, so that + this keeps the one-argument shape every other family's ``parse`` has. It is + the first four octets of any block, and a Section Header Block -- the one + block read with no section context, because it is what *starts* a + section -- is recognisable by :data:`PCAPNG_SHB_MAGIC`. + + Args: + octets: The constructed block. + + Returns: + The parsed block. + + """ + from pcapkit.protocols.misc.pcapng import PCAPNG + + context = None if octets[:4] == PCAPNG_SHB_MAGIC else _pcapng_context()[0] + return PCAPNG(octets, len(octets), num=2, sct=1, ctx=context) + + +def _pcapng_block_registry() -> 'Any': + from pcapkit.protocols.misc.pcapng import PCAPNG + return PCAPNG.__block__ + + +def _pcapng_block_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.pcapng.block_type import BlockType as Enum_BlockType + from pcapkit.const.pcapng.record_type import RecordType as Enum_RecordType + from pcapkit.const.pcapng.secrets_type import SecretsType as Enum_SecretsType + from pcapkit.const.reg.linktype import LinkType as Enum_LinkType + return { + Enum_BlockType.Interface_Description_Block: { + 'linktype': Enum_LinkType.ETHERNET, 'snaplen': 0x40000}, + Enum_BlockType.Enhanced_Packet_Block: { + 'timestamp': PCAPNG_TIME, 'packet_data': PCAPNG_PACKET}, + Enum_BlockType.Simple_Packet_Block: {'packet_data': PCAPNG_PACKET}, + # Without the terminating record the records field consumes the whole + # block and the option area disappears. + Enum_BlockType.Name_Resolution_Block: { + 'records': [(Enum_RecordType.nrb_record_end, {})]}, + Enum_BlockType.Interface_Statistics_Block: {'timestamp': PCAPNG_TIME}, + Enum_BlockType.systemd_Journal_Export_Block: {'entries': PCAPNG_JOURNAL}, + # ZigBee rather than the default TLS key log, whose writer stamps the + # current time into the body and so can never round-trip. + Enum_BlockType.Decryption_Secrets_Block: { + 'secrets_type': Enum_SecretsType.ZigBee_NWK_Key, + 'secrets_data': {'nwk_key': bytes(range(16)), 'pan_id': 0x1234}}, + Enum_BlockType.Custom_Block_that_rewriters_can_copy_into_new_files: { + 'pen': 32473, 'data': b'\x01\x02\x03\x04'}, + Enum_BlockType.Custom_Block_that_rewriters_should_not_copy_into_new_files: { + 'pen': 32473, 'data': b'\x01\x02\x03\x04'}, + Enum_BlockType.Packet_Block: { + 'timestamp': PCAPNG_TIME, 'packet_data': PCAPNG_PACKET}, + } + + +def _pcapng_block_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + return _pcapng_block(code, dict(kwargs)) + + +def _pcapng_block_extract(parsed: 'Any') -> 'Any': + return parsed.info + + +def _pcapng_block_rebuild(info: 'Any') -> 'Any': + return _pcapng_block(info.type, info) + + +#: Which block hosts each block-option namespace, and that host's own +#: arguments. The namespace is the part of the ``_option_key`` tuple's first +#: element before the first underscore. +PCAPNG_OPTION_HOSTS = { + 'opt': 'Section_Header_Block', + 'if': 'Interface_Description_Block', + 'epb': 'Enhanced_Packet_Block', + 'ns': 'Name_Resolution_Block', + 'isb': 'Interface_Statistics_Block', + 'pack': 'Packet_Block', +} + + +def _pcapng_option_registry() -> 'Any': + from pcapkit.protocols.misc.pcapng import PCAPNG + return PCAPNG.__option__ + + +def _pcapng_option_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.pcapng.option_type import OptionType as Enum_OptionType + from pcapkit.protocols.misc.pcapng import _option_key + return { + _option_key(Enum_OptionType.opt_comment): {'comment': 'probe'}, + _option_key(Enum_OptionType.opt_custom_2988): {'pen': 32473, 'data': b'\x01\x02\x03\x04'}, + _option_key(Enum_OptionType.opt_custom_2989): {'pen': 32473, 'data': b'\x01\x02\x03\x04'}, + _option_key(Enum_OptionType.opt_custom_19372): {'pen': 32473, 'data': b'\x01\x02\x03\x04'}, + _option_key(Enum_OptionType.opt_custom_19373): {'pen': 32473, 'data': b'\x01\x02\x03\x04'}, + _option_key(Enum_OptionType.if_name): {'name': 'eth0'}, + _option_key(Enum_OptionType.if_description): {'description': 'probe interface'}, + _option_key(Enum_OptionType.if_IPv4addr): {'interface': '10.0.0.1/255.255.255.0'}, + _option_key(Enum_OptionType.if_IPv6addr): { + 'interface': '2001:db8:85a3:8d3:1319:8a2e:370:7344/64'}, + _option_key(Enum_OptionType.if_MACaddr): {'interface': '02:00:00:00:00:01'}, + _option_key(Enum_OptionType.if_EUIaddr): {'interface': '02:00:00:FF:FE:00:00:01'}, + _option_key(Enum_OptionType.if_speed): {'speed': 1000000000}, + _option_key(Enum_OptionType.if_tsresol): {'resolution': 1000000}, + _option_key(Enum_OptionType.if_tzone): {'tzone': 0}, + _option_key(Enum_OptionType.if_filter): {'expression': b'udp port 67'}, + # ``if_os`` and ``if_hardware`` default to the live platform strings, so + # leaving them unset would make the case machine-dependent. + _option_key(Enum_OptionType.if_os): {'os': 'probe OS'}, + _option_key(Enum_OptionType.if_hardware): {'hardware': 'probe adapter'}, + _option_key(Enum_OptionType.if_fcslen): {'fcs_length': 4}, + _option_key(Enum_OptionType.if_tsoffset): {'offset': 0}, + _option_key(Enum_OptionType.if_txspeed): {'speed': 1000000000}, + _option_key(Enum_OptionType.if_rxspeed): {'speed': 1000000000}, + _option_key(Enum_OptionType.epb_hash): {'hash': b'\x01\x02\x03\x04'}, + _option_key(Enum_OptionType.epb_verdict): {'value': b'\x01'}, + _option_key(Enum_OptionType.ns_dnsname): {'name': 'resolver.example'}, + _option_key(Enum_OptionType.ns_dnsIP4addr): {'ip': '10.0.0.53'}, + # The default is the IPv4 literal ``'8.8.8.8'``, which the v6 field + # rejects outright. + _option_key(Enum_OptionType.ns_dnsIP6addr): {'ip': '2001:db8::35'}, + _option_key(Enum_OptionType.isb_starttime): {'timestamp': PCAPNG_TIME}, + _option_key(Enum_OptionType.isb_endtime): {'timestamp': PCAPNG_TIME}, + _option_key(Enum_OptionType.isb_ifrecv): {'packets': 4}, + _option_key(Enum_OptionType.isb_ifdrop): {'packets': 1}, + _option_key(Enum_OptionType.isb_filteraccept): {'packets': 4}, + _option_key(Enum_OptionType.isb_osdrop): {'packets': 0}, + _option_key(Enum_OptionType.isb_usrdeliv): {'packets': 3}, + _option_key(Enum_OptionType.pack_hash): {'hash': b'\x01\x02\x03\x04'}, + } + + +def _pcapng_option_enum(key: 'Any') -> 'Any': + """The enumeration member behind an ``_option_key`` tuple. + + ``PCAPNG.__option__`` is keyed by ``(name, value)`` tuples because the same + numeric option code means different things in different block types, but + the construction API takes the enumeration member. + + Args: + key: The ``(str, int)`` registry key. + + Returns: + The matching :class:`~pcapkit.const.pcapng.option_type.OptionType`. + + Raises: + KeyError: If no member matches, which would mean the registry and the + enumeration have diverged. + + """ + from pcapkit.const.pcapng.option_type import OptionType as Enum_OptionType + from pcapkit.protocols.misc.pcapng import _option_key + + # The enumeration is an :mod:`aenum` one, whose metaclass does not advertise + # ``__iter__`` to a type checker even though it iterates perfectly well. + members = Enum_OptionType # type: Any + for member in members: + if _option_key(member) == key: + return member + raise KeyError(f'no OptionType matches registry key {key!r}') + + +def _pcapng_option_host(key: 'Any') -> 'tuple[Any, dict[str, Any]]': + """The block type that carries option ``key``, and its own arguments.""" + from pcapkit.const.pcapng.block_type import BlockType as Enum_BlockType + + namespace = str(key[0]).split('_', 1)[0] + name = PCAPNG_OPTION_HOSTS[namespace] + code = getattr(Enum_BlockType, name) + return code, dict(_pcapng_block_overrides().get(code, {})) + + +def _pcapng_option_build(key: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + code, block = _pcapng_option_host(key) + block['options'] = [(_pcapng_option_enum(key), dict(kwargs))] + return _pcapng_block(code, block) + + +def _pcapng_option_rebuild(info: 'Any') -> 'Any': + """Rebuild an option's host block, taking only the options from the parse. + + The host's own fields come from :func:`_pcapng_block_overrides` rather than + from the parsed block, and that is the point. Reconstructing the whole block + from ``info`` would drop the captured payload of a packet-carrying block -- + ``_make_block_epb`` and its siblings never restore ``packet_data``, and the + data model has nowhere to keep it -- so every ``epb_*`` and ``pack_*`` + option would be reported as a mismatch on account of a defect that has + nothing to do with the option. Re-supplying the host fields isolates the + option, which is what this family is measuring. + + Args: + info: The parsed block. + + Returns: + The reconstructed block. + + """ + block = dict(_pcapng_block_overrides().get(info.type, {})) + block['options'] = info.options + return _pcapng_block(info.type, block) + + +def _pcapng_record_registry() -> 'Any': + from pcapkit.protocols.misc.pcapng import PCAPNG + return PCAPNG.__record__ + + +def _pcapng_record_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.pcapng.record_type import RecordType as Enum_RecordType + return { + Enum_RecordType.nrb_record_ipv4: {'ip': '10.0.0.1', 'names': ['gateway.example']}, + # The default is the IPv4 literal ``'127.0.0.1'``, which the v6 field + # rejects outright. + Enum_RecordType.nrb_record_ipv6: {'ip': '2001:db8::1', 'names': ['v6.example']}, + } + + +def _pcapng_record_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.const.pcapng.block_type import BlockType as Enum_BlockType + from pcapkit.const.pcapng.record_type import RecordType as Enum_RecordType + + records = [(code, dict(kwargs))] + if code != Enum_RecordType.nrb_record_end: + # A record list has to be terminated, or the records field eats the + # rest of the block. + records.append((Enum_RecordType.nrb_record_end, {})) + return _pcapng_block(Enum_BlockType.Name_Resolution_Block, {'records': records}) + + +def _pcapng_record_extract(parsed: 'Any') -> 'Any': + return parsed.info.records + + +def _pcapng_record_rebuild(records: 'Any') -> 'Any': + from pcapkit.const.pcapng.block_type import BlockType as Enum_BlockType + return _pcapng_block(Enum_BlockType.Name_Resolution_Block, {'records': records}) + + +def _pcapng_secrets_registry() -> 'Any': + from pcapkit.protocols.misc.pcapng import PCAPNG + return PCAPNG.__secrets__ + + +def _pcapng_secrets_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.pcapng.secrets_type import SecretsType as Enum_SecretsType + from pcapkit.corekit.multidict import OrderedMultiDict + from pcapkit.protocols.misc.pcapng import TLSKeyLabel, WireGuardKeyLabel + return { + Enum_SecretsType.TLS_Key_Log: { + 'entries': {TLSKeyLabel.CLIENT_RANDOM: OrderedMultiDict( + [(bytes(range(0x20, 0x40)), bytes(range(0x40, 0x70)))])}}, + Enum_SecretsType.WireGuard_Key_Log: { + 'entries': OrderedMultiDict( + [(WireGuardKeyLabel.LOCAL_STATIC_PRIVATE_KEY, bytes(range(32)))])}, + Enum_SecretsType.ZigBee_NWK_Key: {'nwk_key': bytes(range(16)), 'pan_id': 0x1234}, + Enum_SecretsType.ZigBee_APS_Key: { + 'aps_key': bytes(range(16)), 'pan_id': 0x1234, 'short_address': 0xABCD}, + } + + +def _pcapng_secrets_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.const.pcapng.block_type import BlockType as Enum_BlockType + return _pcapng_block(Enum_BlockType.Decryption_Secrets_Block, + {'secrets_type': code, 'secrets_data': dict(kwargs)}) + + +############################################################################### +# The families, and the cycle +############################################################################### + +#: Every family this module exercises, in the order their frames are emitted. +FAMILIES = ( + Family('tcp-option', _tcp_registry, _tcp_overrides, + _tcp_build, _tcp_parse, _tcp_extract, _tcp_rebuild, + capture='options-tcp.pcap', envelope='ipv4', proto=6), + Family('tcp-mptcp', _mptcp_registry, _mptcp_overrides, + _mptcp_build, _tcp_parse, _tcp_extract, _tcp_rebuild, + capture='options-tcp.pcap', envelope='ipv4', proto=6), + Family('ipv4-option', _ipv4_registry, _ipv4_overrides, + _ipv4_build, _ipv4_parse, _ipv4_extract, _ipv4_rebuild, + capture='options-ipv4.pcap', envelope='ethernet-ipv4'), + Family('hopopt-option', _hopopt_registry, _ipv6_option_overrides, + _hopopt_build, _hopopt_parse, _ipv6_extract, _hopopt_rebuild, + capture='options-ipv6.pcap', envelope='ipv6', proto=0), + Family('ipv6-opts-option', _ipv6_opts_registry, _ipv6_option_overrides, + _ipv6_opts_build, _ipv6_opts_parse, _ipv6_extract, _ipv6_opts_rebuild, + capture='options-ipv6.pcap', envelope='ipv6', proto=60), + Family('ipv6-route-type', _ipv6_route_registry, _ipv6_route_overrides, + _ipv6_route_build, _ipv6_route_parse, _ipv6_route_extract, + _ipv6_route_rebuild, + capture='options-ipv6.pcap', envelope='ipv6', proto=43), + Family('sctp-chunk', _sctp_chunk_registry, _sctp_chunk_overrides, + _sctp_chunk_build, _sctp_parse, _sctp_extract, _sctp_rebuild, + capture='options-transport.pcap', envelope='ipv4', proto=132), + Family('sctp-parameter', _sctp_param_registry, _sctp_param_overrides, + _sctp_param_build, _sctp_parse, _sctp_extract, _sctp_rebuild, + capture='options-transport.pcap', envelope='ipv4', proto=132), + Family('sctp-cause', _sctp_cause_registry, _sctp_cause_overrides, + _sctp_cause_build, _sctp_parse, _sctp_extract, _sctp_rebuild, + capture='options-transport.pcap', envelope='ipv4', proto=132), + Family('httpv2-frame', _httpv2_registry, _httpv2_overrides, + _httpv2_build, _httpv2_parse, _httpv2_extract, _httpv2_rebuild, + capture='options-transport.pcap', envelope='tcp'), + Family('mh-message', _mh_message_registry, _mh_message_overrides, + _mh_message_build, _mh_parse, _mh_message_extract, _mh_message_rebuild, + capture='options-internet.pcap', envelope='ipv6', proto=135), + Family('mh-option', _mh_option_registry, _mh_option_overrides, + _mh_option_build, _mh_parse, _mh_option_extract, _mh_option_rebuild, + capture='options-internet.pcap', envelope='ipv6', proto=135), + Family('mh-extension', _mh_extension_registry, _mh_extension_overrides, + _mh_extension_build, _mh_parse, _mh_option_extract, _mh_option_rebuild, + capture='options-internet.pcap', envelope='ipv6', proto=135), + Family('hip-parameter', _hip_registry, _hip_overrides, + _hip_build, _hip_parse, _hip_extract, _hip_rebuild, + capture='options-internet.pcap', envelope='ipv6', proto=139), + # The four PCAP-NG families write no capture -- see the section comment + # above ``PCAPNG_TIME`` for why the construction API cannot make a + # reproducible one, and what already covers that space instead. + Family('pcapng-block', _pcapng_block_registry, _pcapng_block_overrides, + _pcapng_block_build, _pcapng_parse, _pcapng_block_extract, + _pcapng_block_rebuild, capture=None, envelope=None), + # Both of these reconstruct from the parsed *block* rather than from the + # option or secrets collection alone, because neither collection records + # which block type carried it -- but the option family re-supplies the + # host's own fields on the way, for the reason ``_pcapng_option_rebuild`` + # gives. + Family('pcapng-option', _pcapng_option_registry, _pcapng_option_overrides, + _pcapng_option_build, _pcapng_parse, _pcapng_block_extract, + _pcapng_option_rebuild, capture=None, envelope=None), + Family('pcapng-record', _pcapng_record_registry, _pcapng_record_overrides, + _pcapng_record_build, _pcapng_parse, _pcapng_record_extract, + _pcapng_record_rebuild, capture=None, envelope=None), + Family('pcapng-secrets', _pcapng_secrets_registry, _pcapng_secrets_overrides, + _pcapng_secrets_build, _pcapng_parse, _pcapng_block_extract, + _pcapng_block_rebuild, capture=None, envelope=None), +) + +#: The families keyed by label, for a caller that wants just one. +FAMILY_MAP = {family.label: family for family in FAMILIES} + +#: Codes deliberately left out of the case list, with the reason. +#: +#: The padding options are dropped by every ``_make_*_options`` in the tree -- +#: it discards them and inserts its own padding to reach the alignment the +#: protocol wants -- so a case for one would assert only that construction +#: ignores its own input, which it does. They are kept for TCP, where the +#: dropping is symmetric and the case is a genuine round trip of an +#: option-less header, and skipped where a second padding code would merely +#: duplicate the first. +SKIP = { + ('tcp-option', 'End_of_Option_List'): 'padding, dropped by _make_tcp_options', + ('tcp-option', 'No_Operation'): 'padding, dropped by _make_tcp_options', + # Multipath TCP is the envelope for the whole tcp-mptcp family, so a case + # here would silently duplicate whichever subtype it defaulted to. + ('tcp-option', 'Multipath_TCP'): 'envelope for the tcp-mptcp family', + # Pad1 and PadN are both discarded by _make_hopopt_options, so the two + # cases emit byte-identical option-less headers. + ('hopopt-option', 'PadN'): 'padding, indistinguishable from Pad1 here', + ('ipv6-opts-option', 'PadN'): 'padding, indistinguishable from Pad1 here', +} + + +def cases(families: 'Optional[tuple[Family, ...]]' = None) -> 'list[Case]': + """Every code every family registry claims to support, as a case. + + The list comes from the registries rather than from a table, which is what + lets :file:`tests/protocols/test_option_roundtrip_unit.py` notice that a + newly registered code has no case yet. + + Args: + families: Families to enumerate; :data:`FAMILIES` if not given. + + Returns: + The cases, ordered by family and then by :func:`sort_key`, so two runs + produce the same list and therefore the same captures. + + """ + out = [] # type: list[Case] + for family in (FAMILIES if families is None else families): + registry = family.registry() + overrides = family.overrides() + for code in sorted(registry, key=sort_key): + name = code_name(code) + if (family.label, name) in SKIP: + continue + # A code missing from ``overrides`` means "every default", which is + # what most of them want. + out.append(Case(family.label, code, name, dict(overrides.get(code, {})))) + return out + + +def roundtrip(case: 'Case', deadline: 'int' = DEADLINE) -> 'Outcome': + """Construct ``case``, parse it back, construct it again, compare. + + The third step is the one that earns its keep. A ``_make_*`` that takes + only keyword arguments, and cannot consume the data model its own + ``_read_*`` produced, still passes a construct-then-parse test and fails + here. + + Args: + case: The case to exercise. + deadline: Whole seconds to allow, or ``0`` for no deadline. See the + module docstring for why there is one at all. + + Returns: + An :class:`Outcome` naming the step that failed, or ``'OK'``. + + """ + family = FAMILY_MAP[case.family] + # Bound inside the ``with`` below and read after it, so it is declared here. + # Non-optional, unlike the ``octets`` field of the ``Outcome`` it ends up in: + # every path that reaches the comparison at the bottom has been through a + # successful construction, and saying so keeps the ``.hex()`` there honest. + octets = b'' + again = b'' + + with _deadline(deadline), warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + + def seen() -> 'tuple[str, ...]': + """Warning messages so far, deduplicated but in order. + + :exc:`ResourceWarning` is dropped. It is raised by the garbage + collector rather than by the code under test, so whichever case + happens to be running when a file object from somewhere else is + finalised would otherwise be blamed for it -- which is exactly what + happened: a ``.pcapng`` handle left open by a sibling generator was + reported against ``hopopt-option/Router_Alert``. + """ + return tuple(dict.fromkeys( + str(entry.message) for entry in caught + if not issubclass(entry.category, ResourceWarning))) + + try: + octets = bytes(family.build(case.code, case.kwargs)) + except TimeoutError as exc: + return Outcome(case, 'TIMEOUT', str(exc), None, seen()) + except Exception as exc: # pylint: disable=broad-except + # Broad on purpose: the point is to find out *what* a constructor + # does when it is wrong, and narrowing this to ProtocolError would + # let an AttributeError or a KeyError -- which is what most of the + # current defects raise -- abort the sweep rather than be recorded + # as this case's result. + return Outcome(case, 'CONSTRUCT', f'{type(exc).__name__}: {exc}', None, seen()) + + try: + parsed = family.extract(family.parse(octets)) + except TimeoutError as exc: + return Outcome(case, 'TIMEOUT', str(exc), octets, seen()) + except Exception as exc: # pylint: disable=broad-except + return Outcome(case, 'PARSE', f'{type(exc).__name__}: {exc}', octets, seen()) + + try: + again = bytes(family.rebuild(parsed)) + except TimeoutError as exc: + return Outcome(case, 'TIMEOUT', str(exc), octets, seen()) + except Exception as exc: # pylint: disable=broad-except + return Outcome(case, 'RECONSTRUCT', f'{type(exc).__name__}: {exc}', + octets, seen()) + + warned = seen() + + if octets != again: + return Outcome(case, 'MISMATCH', f'{octets.hex()} != {again.hex()}', + octets, warned) + return Outcome(case, 'OK', '', octets, warned) + + +class _deadline: # pylint: disable=invalid-name + """Raise :exc:`TimeoutError` in the body after ``seconds`` seconds. + + Written as a class rather than a :func:`contextlib.contextmanager` so that + it can be entered alongside :func:`warnings.catch_warnings` in one ``with`` + statement without the generator machinery swallowing the alarm. + + A no-op where :data:`signal.SIGALRM` is missing (Windows) or where the + caller asked for no deadline. That is a deliberate degradation rather than + a hard failure: without an interval timer the only alternative would be to + refuse to generate the captures at all. + + """ + + def __init__(self, seconds: 'int') -> 'None': + #: Whole seconds to allow. :func:`signal.alarm` counts in whole + #: seconds, so this cannot usefully be fractional. + self.seconds = seconds + #: Whether the alarm was actually armed, and so needs disarming. + self.armed = False + #: Handler displaced by :meth:`__enter__`, restored by :meth:`__exit__`. + self.previous = None # type: Any + + def __enter__(self) -> '_deadline': + if self.seconds <= 0 or not hasattr(signal, 'SIGALRM'): + return self + + def expire(signum: 'int', frame: 'Any') -> 'None': + raise TimeoutError(f'did not finish within {self.seconds}s') + + self.previous = signal.signal(signal.SIGALRM, expire) + signal.alarm(self.seconds) + self.armed = True + return self + + def __exit__(self, *exc_info: 'Any') -> 'None': + if self.armed: + # Cancel before restoring, so an alarm firing between the two + # cannot be delivered to whatever handler was installed before. + signal.alarm(0) + signal.signal(signal.SIGALRM, self.previous) + self.armed = False + + +def outcomes(families: 'Optional[tuple[Family, ...]]' = None, + deadline: 'int' = DEADLINE) -> 'Iterator[Outcome]': + """Run :func:`roundtrip` over :func:`cases`. + + Args: + families: Families to enumerate; :data:`FAMILIES` if not given. + deadline: Whole seconds to allow each case. + + Yields: + One :class:`Outcome` per case, in case order. + + """ + for case in cases(families): + yield roundtrip(case, deadline) + + +def _frame(family: 'Family', octets: 'bytes', index: 'int') -> 'Any': + """Wrap constructed octets in the envelope that makes them decodable. + + Args: + family: The family the octets came from, which decides the envelope. + octets: The octets :func:`roundtrip` constructed. + index: Position within the capture, used as the IPv4 identification so + that two frames carrying identical options still differ somewhere. + + Returns: + A :mod:`scapy` packet, with its timestamp already pinned. + + Raises: + ValueError: If ``family`` names an envelope this function does not know. + + """ + from scapy.all import IP, IPv6, Ether, Raw, TCP # pylint: disable=no-name-in-module + + link = Ether(src=SRC_MAC, dst=DST_MAC) + if family.envelope == 'ethernet-ipv4': + # The octets *are* the IPv4 header, options and all, so nothing may be + # put between them and the link layer -- and because the payload is a + # bare ``Raw``, scapy has nothing to infer the EtherType from and falls + # back to its default of 0x9000. That decodes as Loopback, not IPv4, so + # the type is stated here rather than left to be guessed. + frame = link / Raw(octets) + frame.type = 0x0800 + elif family.envelope == 'ipv4': + frame = link / IP(src=SRC_IP, dst=DST_IP, proto=family.proto, + id=index, ttl=64) / Raw(octets) + elif family.envelope == 'ipv6': + frame = link / IPv6(src=SRC_IP6, dst=DST_IP6, nh=family.proto, + hlim=64) / Raw(octets) + elif family.envelope == 'tcp': + frame = (link / IP(src=SRC_IP, dst=DST_IP, id=index, ttl=64) + / TCP(sport=50000, dport=80, flags='A', seq=1, ack=1) + / Raw(octets)) + else: # pragma: no cover + raise ValueError(f'unknown envelope {family.envelope!r}') + + # A fixed epoch plus the frame's own index, rather than the clock. + frame.time = EPOCH + index + return frame + + +def generate(dest: 'pathlib.Path | None' = None) -> 'list[pathlib.Path]': + """Write the option-coverage sample captures. + + Args: + dest: Destination directory; ``examples/captures/`` under the + repository root, if not given. Created if it does not exist. + + Returns: + The paths written, in the order they were written. + + """ + from scapy.all import wrpcap # pylint: disable=no-name-in-module + + dest = SAMPLE if dest is None else pathlib.Path(dest) + dest.mkdir(parents=True, exist_ok=True) + + buckets = {} # type: dict[str, list[Any]] + unreachable = {} # type: dict[str, str] + warned = {} # type: dict[str, tuple[str, ...]] + + for outcome in outcomes(): + family = FAMILY_MAP[outcome.case.family] + if family.capture is None: + # Exercised, but deliberately not captured. + continue + if outcome.status not in CAPTURABLE: + unreachable[outcome.case.label] = f'{outcome.status}: {outcome.detail}' + continue + if outcome.warnings: + warned[outcome.case.label] = outcome.warnings + + octets = outcome.octets + if octets is None: # pragma: no cover + # Unreachable: every status in CAPTURABLE got past construction, so + # it has octets. Narrowed explicitly rather than asserted, because a + # generator should not abort a whole run over a bookkeeping slip. + continue + + frames = buckets.setdefault(family.capture, []) + frames.append(_frame(family, octets, len(frames))) + + written = [] # type: list[pathlib.Path] + for name in sorted(buckets): + path = dest / name + wrpcap(str(path), buckets[name]) + written.append(path) + + total = sum(len(frames) for frames in buckets.values()) + print(f'options: {total} capturable case(s) across {len(written)} capture(s); ' + f'{len(unreachable)} case(s) left out') + for label in sorted(unreachable): + print(f' [left out] {label}: {unreachable[label]}') + for label in sorted(warned): + print(f' [warned] {label}: {"; ".join(warned[label])}') + return written + + +if __name__ == '__main__': + for sample in generate(): + print(f'{str(sample.relative_to(ROOT)):<32s} {sample.stat().st_size:8d} octets') diff --git a/tests/protocols/test_option_coverage_runtime.py b/tests/protocols/test_option_coverage_runtime.py new file mode 100644 index 0000000000..13575143ab --- /dev/null +++ b/tests/protocols/test_option_coverage_runtime.py @@ -0,0 +1,224 @@ +# -*- coding: utf-8 -*- +"""The ``options-*.pcap`` fixtures decode, and the octets in them are pcapkit's own. + +:file:`examples/generators/options.py` constructs one item of every option-like +code the library registers, and writes the ones that survive a construct -> +parse -> construct cycle into five captures. This module reads them back through +the public extraction interface. + +The point is not to re-assert the round trip -- +:file:`tests/protocols/test_option_roundtrip_unit.py` does that, and does it +without needing a fixture. The point is that these captures go through the +*whole* stack: the pcap reader, the link and internet layer dispatch, the +option-area walk, and the dumpers. A defect that only shows up once an option is +inside a real frame -- an option area whose length disagrees with the header, a +next-layer lookup that fires on the wrong protocol number -- is invisible to a +direct round trip and lands here. + +This module is fixture tier, so it may read generated captures; it is named +``*_runtime.py`` for exactly that reason, and it handles the captures being +absent so that a checkout which has not run ``make samples`` skips rather than +fails. See :mod:`tests._tiers` for the rule. + +Every extraction is wrapped in :func:`tests._support.time_limit`. That is not +belt-and-braces: the generator deliberately leaves out the cases whose parse does +not terminate -- ``HOPOPT``'s ``SMF_DPD`` is one -- and this deadline is what +turns a regression that puts one back into a failure rather than a wedged CI run. + +""" + +from __future__ import annotations + +import importlib.util +import os +import tempfile +import unittest +from typing import TYPE_CHECKING + +from tests._support import close_extractor, purge_modules, time_limit +from tests._tiers import SAMPLE_ROOT + +if TYPE_CHECKING: + from typing import Any, Optional + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + +#: Whole seconds one capture may take to extract. The largest of them is a few +#: hundred frames of a few dozen octets, so this is orders of magnitude of slack; +#: it exists for the non-terminating case, not for slowness. +EXTRACT_TIMEOUT = 30 + +#: The captures, and the outermost protocol chain every frame in each should +#: decode to. Spelled as a prefix rather than the whole chain because the +#: interesting part is that the envelope dispatched to the right layer -- what +#: the option itself decodes to is the round-trip test's business. +CAPTURES = { + 'options-tcp.pcap': 'Ethernet:IPv4:TCP', + 'options-ipv4.pcap': 'Ethernet:IPv4', + 'options-ipv6.pcap': 'Ethernet:IPv6', + 'options-transport.pcap': 'Ethernet:IPv4', + 'options-internet.pcap': 'Ethernet:IPv6', +} + + +def _capture(name: str) -> 'Optional[str]': + """Path to a generated option capture, or :data:`None` if it is not there. + + Deliberately not :func:`tests._support.sample_path`: that raises + :exc:`FileNotFoundError` for a missing capture, and every caller here wants + to skip instead. The tier guard is satisfied either way, since this module is + fixture tier and may read a generated capture at all. + + Args: + name: Bare capture file name. + + Returns: + The absolute path, or :data:`None` if the capture has not been generated. + + """ + path = SAMPLE_ROOT / name + return str(path) if path.is_file() else None + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class OptionCoverageCaptureTests(unittest.TestCase): + """The generated option captures extract cleanly.""" + + def setUp(self) -> None: + purge_modules(['pcapkit']) + + def test_every_option_capture_extracts(self) -> None: + """Each capture reads back, with every frame reaching its expected layer. + + One subtest per capture, so a single broken fixture names itself rather + than stopping at the first one. + + """ + import pcapkit + + found = 0 + for name, chain in sorted(CAPTURES.items()): + path = _capture(name) + if path is None: + continue + found += 1 + + with self.subTest(capture=name): + with time_limit(EXTRACT_TIMEOUT): + extractor = pcapkit.extract(fin=path, nofile=True, format='tree', + store=True, ip=True, tcp=True) + self.addCleanup(close_extractor, extractor) + + self.assertGreater(extractor.length, 0, + f'{name} carries no frames') + self.assertEqual(len(extractor.frame), extractor.length) + + # ``Extractor.frame`` is annotated as a union that includes the + # raw ``(timestamp, bytes)`` tuple the ``pcap_ct``-style engines + # yield; the default engine always gives a protocol object. + frames = list(extractor.frame) # type: list[Any] + for index, frame in enumerate(frames): + protochain = str(frame.protochain) + self.assertTrue( + protochain.startswith(chain), + f'{name} frame {index}: expected a {chain} envelope, ' + f'got {protochain}' + ) + + if not found: + self.skipTest( + 'no options-*.pcap fixtures present; run ' + 'examples/generators/make_samples.py to build them' + ) + + def test_option_captures_dump_to_tree_and_json(self) -> None: + """Each capture survives both dumpers, with reassembly turned on. + + The dumpers walk every field of every parsed option, so this reaches + representation code that extraction alone does not -- a field whose value + cannot be rendered fails here and nowhere else. + + """ + import pcapkit + + found = 0 + for name in sorted(CAPTURES): + path = _capture(name) + if path is None: + continue + found += 1 + + for fmt, suffix in (('tree', 'txt'), ('json', 'json')): + with self.subTest(capture=name, format=fmt): + with tempfile.TemporaryDirectory() as tmpdir: + out = os.path.join(tmpdir, f'{name}.{suffix}') + # ``extract`` types ``format`` as a Literal, and a value + # read out of a loop is a plain ``str`` to a checker. + fmt_arg = fmt # type: Any + with time_limit(EXTRACT_TIMEOUT): + extractor = pcapkit.extract( + fin=path, fout=out, format=fmt_arg, store=False, + ip=True, tcp=True, reassembly=True) + self.addCleanup(close_extractor, extractor) + self.assertGreater(extractor.length, 0) + self.assertTrue(os.path.isfile(out), + f'{fmt} dump of {name} wrote no file') + self.assertGreater(os.path.getsize(out), 0, + f'{fmt} dump of {name} is empty') + + if not found: + self.skipTest('no options-*.pcap fixtures present') + + def test_option_captures_are_what_the_generator_says_they_are(self) -> None: + """Frame counts match the case table, so a stale fixture is caught. + + A capture regenerated from a different revision of the case table would + otherwise sit on disk indefinitely and quietly test the wrong thing. The + count is derived from the same table the generator uses, not written down + here. + + """ + from tests._tiers import ROOT + + spec = importlib.util.spec_from_file_location( + 'pcapkit_samples_options_runtime', + ROOT / 'examples' / 'generators' / 'options.py') + if spec is None or spec.loader is None: # pragma: no cover + self.skipTest('cannot load the option case table') + options = importlib.util.module_from_spec(spec) + spec.loader.exec_module(options) + + import pcapkit + + expected = {} # type: dict[str, int] + for outcome in options.outcomes(): + family = options.FAMILY_MAP[outcome.case.family] + if family.capture is None or outcome.status not in options.CAPTURABLE: + continue + expected[family.capture] = expected.get(family.capture, 0) + 1 + + found = 0 + for name, count in sorted(expected.items()): + path = _capture(name) + if path is None: + continue + found += 1 + with self.subTest(capture=name): + with time_limit(EXTRACT_TIMEOUT): + extractor = pcapkit.extract(fin=path, nofile=True, format='tree', + store=False) + self.addCleanup(close_extractor, extractor) + self.assertEqual( + extractor.length, count, + f'{name} holds {extractor.length} frames but the case table ' + f'yields {count}; regenerate it with ' + f'examples/generators/make_samples.py' + ) + + if not found: + self.skipTest('no options-*.pcap fixtures present') + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/test_option_roundtrip_unit.py b/tests/protocols/test_option_roundtrip_unit.py new file mode 100644 index 0000000000..f7830b7a65 --- /dev/null +++ b/tests/protocols/test_option_roundtrip_unit.py @@ -0,0 +1,915 @@ +# -*- coding: utf-8 -*- +"""Construct -> parse -> construct identity, over every option-like code. + +Every option, chunk, parameter, message type, frame type and block type this +library claims to support is enumerated *from the dispatch registries* and put +through one cycle: construct it through the public construction API, parse the +result back, construct it again from what was parsed, and require the octets to +be identical across the round trip. + +That third step is what this module exists for. A construct-then-parse test +passes for a ``_make_*`` that takes only keyword arguments and cannot consume +the data model its own ``_read_*`` produced -- which is a defect that ships, and +one this suite had no way to see. Sixteen HIP parameters are in exactly that +state today. + +The case list and the cycle both live in +:file:`examples/generators/options.py`, next to the generator that turns the +same cases into the ``options-*.pcap`` fixtures, so that the captures and these +assertions can never describe different things. See that module for why the +enumeration is registry-driven and how the per-code arguments were chosen. + +What this module adds is the *judgement*: :data:`EXPECTED_FAILURES` records, case +by case, which cycles do not close today and which defect stops each one. A case +absent from that table has to come back ``'OK'``. + +When the outcome depends on the interpreter, not the code +--------------------------------------------------------- + +One recorded expectation per case assumes the outcome is a property of the +library. Seven PCAP-NG name-resolution cases show it can be a property of the +*interpreter*: they round-trip from Python 3.11 on and fail to construct on 3.10 +and earlier, because :class:`~pcapkit.protocols.schema.schema.Schema` subclasses +share one ``_abc_impl`` there and an ``isinstance`` against a schema class +therefore answers by cache order rather than by type (GitHub issue #439). + +Those seven live in :data:`INTERPRETER_GAPS` rather than in +:data:`EXPECTED_FAILURES`, and the separation is deliberate. Behaving differently +on two supported Pythons is a defect and not a variant, so the mechanism is +asserted directly by +:meth:`OptionRoundTripTests.test_schema_isinstance_is_interpreter_dependent`: +the record is evidence about a named library bug, not an exemption keyed on a +version number. On a modern interpreter the seven are held to ``'OK'`` like +anything else, and fixing #439 turns 3.10 red. + +Why the table is asserted in both directions +-------------------------------------------- + +An expected failure that is only allowed to fail is a test that rots. So each +entry is checked to still fail, *and* to fail in the recorded way -- so fixing +one of these defects turns this module red, which is the reminder to delete the +entry. The reverse check matters as much: :meth:`test_expected_failures_name_real_cases` +fails if the table names a case that no longer exists, which is what happens when +a registry entry is renamed or removed. + +Nothing here is a workaround for the defects it records. Every ``_make_*`` +argument in the generator's tables is a legitimate value for that option, and +none of the assertions below has been loosened to make a failing case pass. The +one place where an argument was chosen to route *around* a defect rather than +into it is HIP's ``HIP_COPIES``, which puts two copies of each parameter in a +packet because one is unrepresentable; the defect that forces it is not lost, +:meth:`OptionRoundTripTests.test_a_single_hip_parameter_cannot_be_constructed` +pins it directly. + +This module is unit tier: it constructs its own octets and reads no capture, so +it runs on a fresh checkout with nothing generated. + +""" + +from __future__ import annotations + +import importlib.util +import sys +import types +import unittest +from typing import TYPE_CHECKING, NamedTuple + +from tests._support import purge_modules, time_limit +from tests._tiers import ROOT + +if TYPE_CHECKING: + from typing import Any, Optional + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + +#: Whole seconds one case may take. Generous next to a working case, which takes +#: low single-digit milliseconds; the point of the deadline is the cases that +#: never finish at all. +CASE_TIMEOUT = 10 + + +class Gap(NamedTuple): + """One cycle that does not close, and the defect that stops it.""" + + #: Expected :attr:`~examples.generators.options.Outcome.status`. + status: 'str' + #: Substring the failure detail must contain, or a tuple of substrings *all* + #: of which it must contain, or ``''`` to assert only the status. Empty is + #: used where the detail is a pair of long hex strings, or where it embeds a + #: timestamp and so is not stable between runs. + #: + #: Make it as specific as the message allows, because a fragment that matches + #: half the tree does not pin anything: ``'invalid format'`` alone occurs 205 + #: times across 13 modules (31 in :file:`internet/hip.py`, 26 in + #: :file:`transport/tcp.py`), so it is satisfied by a regression at any of + #: them. Prefer the alias and whatever bracketed code the message carries -- + #: ``'TCP: [OptNo 28] invalid format'`` narrows those 26 sites to the one + #: option that can print ``28``. The tuple form is for messages whose stable + #: parts are not contiguous, so that a fragment does not have to bake in a + #: rendering that is itself defective. + fragment: 'str | tuple[str, ...]' + #: The defect, named with the ``file:line`` that causes it. This is the + #: field that makes the entry worth keeping rather than just silencing. + defect: 'str' + + +#: Every case whose cycle does not close today. +#: +#: Grouped by defect rather than by family, because the failures cluster far more +#: tightly by cause than by protocol: four lines of ``schema/transport/tcp.py`` +#: account for four Multipath TCP subtypes, and one line of +#: ``pcapkit/protocols/misc/pcapng.py`` accounts for twenty-seven PCAP-NG +#: options. +EXPECTED_FAILURES = { + + # -- TCP ------------------------------------------------------------------ + + # ``_make_mode_timeout`` writes ``length=3`` into an option that packs to + # four octets (kind, length, and a two-octet bitfield), and + # ``_read_mode_timeout`` checks the length exactly. Measured: the schema + # packs ``1c03003c`` where a correct one is ``1c04003c``. + 'tcp-option/User_Timeout_Option': Gap( + 'CONSTRUCT', 'TCP: [OptNo 28] invalid format', + 'pcapkit/protocols/transport/tcp.py:2506 -- _make_mode_timeout sets ' + 'length=3 for a 4-octet option'), + + # ``_make_mode_qs`` computes ``rate_val`` as a floor of a logarithm that is + # negative for any rate under 40 kbps, and a negative value then fails to + # pack into the 4-bit field with ``ValueError: invalid literal for int() + # with base 2: b'0000-110'``. At the default ``rate=0`` it is guarded, and + # the option constructs -- and then will not parse back. + 'tcp-option/Quick_Start_Response': Gap( + 'PARSE', 'StructError', + 'pcapkit/protocols/transport/tcp.py:2465 -- _make_mode_qs; rate_val is ' + 'negative below 40 kbps, and the option it emits at rate=0 does not ' + 'parse back'), + + # Four lambdas read ``pkt['length']`` while packing, but the enclosing + # ``_make_mptcp_*`` never puts a ``length`` in the packet, so the key is + # simply absent on the construction path. + 'tcp-mptcp/MP_CAPABLE': Gap( + 'CONSTRUCT', "KeyError: 'length'", + 'pcapkit/protocols/schema/transport/tcp.py:658'), + 'tcp-mptcp/ADD_ADDR': Gap( + 'CONSTRUCT', "KeyError: 'length'", + 'pcapkit/protocols/schema/transport/tcp.py:790'), + 'tcp-mptcp/REMOVE_ADDR': Gap( + 'CONSTRUCT', "KeyError: 'length'", + 'pcapkit/protocols/schema/transport/tcp.py:807'), + 'tcp-mptcp/MP_PRIO': Gap( + 'CONSTRUCT', "KeyError: 'length'", + 'pcapkit/protocols/schema/transport/tcp.py:827'), + + # ``_read_tcp_options`` reads ``schema.kind`` off every option schema, but + # the nested Multipath TCP subtype schemas do not declare one -- the field + # belongs to the enclosing option. + 'tcp-mptcp/DSS': Gap( + 'CONSTRUCT', "no attribute 'kind'", + 'pcapkit/protocols/transport/tcp.py:668 -- MPTCPDSS declares no kind'), + 'tcp-mptcp/MP_FAIL': Gap( + 'CONSTRUCT', "no attribute 'kind'", + 'pcapkit/protocols/transport/tcp.py:668 -- MPTCPFallback declares no kind'), + 'tcp-mptcp/MP_FASTCLOSE': Gap( + 'CONSTRUCT', "no attribute 'kind'", + 'pcapkit/protocols/transport/tcp.py:668 -- MPTCPFastclose declares no kind'), + + # ``_make_mptcp_join`` branches on ``self._flags``, which only the parse + # path ever sets, so the constructor cannot be called at all. + 'tcp-mptcp/MP_JOIN': Gap( + 'CONSTRUCT', "no attribute '_flags'", + 'pcapkit/protocols/transport/tcp.py:2675 -- _make_mptcp_join reads ' + 'self._flags, which exists only while parsing'), + + # -- IPv4 ----------------------------------------------------------------- + + # ``_make_ipv4_options`` appends a bare enumeration member to the option + # list as its end-of-list padding, and ``OptionField.pack`` accepts only + # bytes or a Schema. It fires for every option whose packed length is not a + # multiple of four, which is what these four have in common. LSR, RR and SSR + # cannot be brought to a multiple of four by any argument: their length is + # ``3 + counts * 4``. + 'ipv4-option/LSR': Gap( + 'CONSTRUCT', 'Field options has invalid value', + 'pcapkit/protocols/internet/ipv4.py:1225 and :1252 -- a bare ' + 'Enum_OptionNumber.EOOL is appended to the option list'), + 'ipv4-option/RR': Gap( + 'CONSTRUCT', 'Field options has invalid value', + 'pcapkit/protocols/internet/ipv4.py:1225 and :1252'), + 'ipv4-option/SSR': Gap( + 'CONSTRUCT', 'Field options has invalid value', + 'pcapkit/protocols/internet/ipv4.py:1225 and :1252'), + # SID reaches the same padding branch for a second reason of its own: + # ``_make_opt_sid`` declares ``length=4`` while ``SIDOption.sid`` is a + # 32-bit field, so the option packs to six octets (``880400000000``). + 'ipv4-option/SID': Gap( + 'CONSTRUCT', 'Field options has invalid value', + 'pcapkit/protocols/internet/ipv4.py:1683 -- _make_opt_sid declares ' + 'length=4 but packs 6 octets, which then trips the :1225 padding branch'), + + # ``_make_opt_ts`` passes ``data=`` where the schema field is ``ts_data``. + # ``Schema.__init__`` only warns about an unknown field name and carries on, + # so the value is dropped and the attribute stays bound to the class-level + # descriptor -- which ``post_process`` then tries to iterate. + 'ipv4-option/TS': Gap( + 'CONSTRUCT', "'ListField' object is not iterable", + 'pcapkit/protocols/internet/ipv4.py:1488 -- data= should be ts_data=, ' + 'dropped with UnknownFieldWarning and surfacing at ' + 'pcapkit/protocols/schema/internet/ipv4.py:262'), + + # -- Quick-Start, in all three protocols that carry it -------------------- + + # ``_make_opt_qs`` returns a bare nested schema, but ``func`` is set only by + # ``_QSOption.post_process``, which runs on the parse path. Since + # ``__post_init__`` packs and then re-reads, construction fails. + # + # There is a second, independent defect in the same option that this case + # never reaches, and it is the more serious of the two: + # ``quick_start_data_selector`` hands the nested schema a hardcoded + # ``SchemaField(length=5)`` where the schema needs eight octets + # (type 1 + length 1 + flags 1 + ttl 1 + nonce 4). Measured on a + # hand-built, well-formed 8-octet IPv4 Quick-Start option + # ``1908002adeadbee0``: it parses "successfully" with + # ``SchemaWarning: packet length < 0: -3`` and decodes ``nonce`` as **55** + # instead of 933982136 -- silent corruption rather than a failure -- and the + # three unconsumed octets are then read as a further, fabricated option, + # which makes the enclosing IPv4 packet unparseable. The identical + # ``SchemaField(length=5)`` is at hopopt.py:224 and ipv6_opts.py:224, with + # the same measured nonce of 55. Fixing the ``func`` defect alone will not + # make these cases pass. + 'ipv4-option/QS': Gap( + 'CONSTRUCT', "no attribute 'func'", + 'pcapkit/protocols/internet/ipv4.py:1144 -- func is set only by ' + 'post_process; and separately ' + 'pcapkit/protocols/schema/internet/ipv4.py:128 -- SchemaField(length=5) ' + 'for an 8-octet option, which decodes nonce as 55'), + 'hopopt-option/Quick_Start': Gap( + 'CONSTRUCT', "no attribute 'func'", + 'pcapkit/protocols/internet/hopopt.py:869; and separately ' + 'pcapkit/protocols/schema/internet/hopopt.py:224 -- SchemaField(length=5)'), + 'ipv6-opts-option/Quick_Start': Gap( + 'CONSTRUCT', "no attribute 'func'", + 'pcapkit/protocols/internet/ipv6_opts.py:881; and separately ' + 'pcapkit/protocols/schema/internet/ipv6_opts.py:224 -- ' + 'SchemaField(length=5)'), + + # -- The non-progress loop, now half fixed -------------------------------- + + # Both of these used to *hang* rather than fail, which is why the cycle is + # run under a deadline at all. #432 landed the progress guard in + # ``OptionField``/``ListField`` and fixed the ``_SMFDPDOption`` sizing in + # *both* schema modules symmetrically -- 26 lines each, ``'len': (1, 8)`` to + # ``(8, 8)`` and the ``+ 2`` on the selector's ``SchemaField`` -- so + # ``hopopt-option/SMF_DPD`` now round-trips and has no entry here at all. + # + # ``IPv6-Opts`` still fails, and not because it missed that fix. The two + # modules differ in exactly one line of code, and it is older than #432: + # ``ipv6_opts.SMFIdentificationBasedDPDOption`` declares a second, redundant + # ``test`` ``ForwardMatchField`` that ``hopopt``'s does not. The enclosing + # ``_SMFDPDOption`` already has one, in both modules, and it is that outer + # field the selector reads -- nothing reads the nested copy. But a + # ``ForwardMatchField`` does not consume the stream while still occupying a + # slot in ``__buffer__``, so the nested schema over-reports its own size by + # one octet, and ``OptionField`` then mis-counts the option area against the + # header. Measured on the *same* octets, ``1100080100010100``: + # + # hopopt __fields__ = [type, len, info, tid, id] len(schema) = 3 + # ipv6_opts __fields__ = [type, len, test, info, tid, id] len(schema) = 4 + # + # HOPOPT(...) -> options=[SMF_DPD, PadN] + # IPv6_Opts(...) -> ProtocolError: IPv6-Opts: invalid format + # + # Identical on 3.10.20 and 3.14.7, so this one is not interpreter-dependent. + # + # The deadline in the sweep stays regardless. It is protection against the + # *next* non-progress defect, not against this one. + 'ipv6-opts-option/SMF_DPD': Gap( + 'PARSE', 'IPv6-Opts: invalid format', + 'pcapkit/protocols/schema/internet/ipv6_opts.py:434 -- a redundant second ' + "'test' ForwardMatchField that hopopt.py's equivalent does not have; it " + 'consumes nothing but is counted in __buffer__, so the nested schema ' + 'reports 4 octets where it read 3, and the threshold check at ' + 'pcapkit/protocols/internet/ipv6_opts.py:497 rejects the option area'), + + # -- IPv6-Route ----------------------------------------------------------- + + # ``IPv6_Route.make`` writes a raw octet count into ``length``, which is an + # 8-octet-unit field, on the dict and Data paths -- while the bytes and + # Schema paths divide by eight. The read-side guards then reject it, and + # those guards compare the unit field against octet counts too, so the + # RFC-correct value would fail as well. + # Both fragments are tuples rather than the whole message, because the + # message itself is defective: these two guards interpolate a bare ``type`` + # into an f-string in a method that has no ``type`` parameter, so the name + # resolves to the builtin and the detail reads ``[TypeNo ]`` + # (issue #442). Matching the literal rendering would pin that bug into this + # table and turn it red when #442 is fixed, which says nothing about whether + # the round trip closes. The stable parts either side of it do pin the site: + # ``[TypeNo`` occurs at exactly three lines of ``ipv6_route.py`` (:462, :506, + # :546), against 205 occurrences of ``'invalid format'`` tree-wide. + 'ipv6-route-type/Source_Route': Gap( + 'CONSTRUCT', ('IPv6-Route', '[TypeNo', 'invalid format'), + 'pcapkit/protocols/internet/ipv6_route.py:276 -- length in octets, not ' + 'in 8-octet units; guard at :461'), + 'ipv6-route-type/Type_2_Routing_Header': Gap( + 'CONSTRUCT', ('IPv6-Route', '[TypeNo', 'invalid format'), + 'pcapkit/protocols/internet/ipv6_route.py:276; guard at :505'), + # RPL fails earlier still: ``post_process`` assumes ``addresses`` is bytes, + # which is true after unpacking and false while packing, where it is still + # the list the constructor was handed. + 'ipv6-route-type/RPL_Source_Route_Header': Gap( + 'CONSTRUCT', 'does not appear to be an IPv4 or IPv6 address', + 'pcapkit/protocols/schema/internet/ipv6_route.py:156 -- post_process ' + 'assumes bytes; it runs on the pack path too, from schema.py:647'), + + # -- Mobility Header ------------------------------------------------------ + + # ``CGAParameter``'s nested length callback reads ``pkt['length']``, which + # is present while packing and absent while unpacking: ``SchemaField.unpack`` + # starts the nested schema with a fresh context whose parent is under + # ``__packet__``. A CGA extension has no other carrier, so the whole + # ``MH.__extension__`` registry is unreachable through the public API. + 'mh-extension/Multi_Prefix': Gap( + 'PARSE', "KeyError: 'length'", + 'pcapkit/protocols/schema/internet/mh.py:516 -- needs ' + "pkt['__packet__']['length'] on the unpack path"), + + # -- HIP ------------------------------------------------------------------ + + # ``_read_param_*`` and ``_make_param_*`` are found by enumeration member + # name, so both exist for code 128; but the *schema* registry is keyed by + # the ``code=`` of the class statement, and ``R1CounterParameter`` declares + # only 129. So code 128 parses as an ``UnassignedParameter``. + 'hip-parameter/R1_Counter': Gap( + 'PARSE', "no attribute 'counter'", + 'pcapkit/protocols/internet/hip.py:822 -- Parameter.registry[128] is ' + 'UnassignedParameter, because R1CounterParameter declares code=129 only'), + + # Sixteen parameters whose ``_read_param_*`` stores a list-valued field as a + # tuple, which ``_make_param_*`` passes straight back to a ``ListField`` + # that accepts only a list. This is the class of defect the reconstruct step + # exists to find: each one constructs and parses perfectly. + **{ + f'hip-parameter/{name}': Gap( + 'RECONSTRUCT', "unsupported type ", + 'pcapkit/protocols/schema/schema.py:624 -- _read_param_* returns a ' + 'tuple where _make_param_* needs a list') + for name in ( + 'ACK', 'DH_GROUP_LIST', 'HIP_CIPHER', 'NAT_TRAVERSAL_MODE', + 'HIT_SUITE_LIST', 'REG_INFO', 'REG_REQUEST', 'REG_RESPONSE', + 'REG_FAILED', 'TRANSPORT_FORMAT_LIST', 'ESP_TRANSFORM', 'ACK_DATA', + 'ROUTE_DST', 'HIP_TRANSPORT_MODE', 'ROUTE_VIA', 'VIA_RVS', + ) + }, + + # ``_make_param_encrypted`` passes ``cipher=``, which is not a field of + # ``EncryptedParameter`` -- so the cipher id is dropped with an + # ``UnknownFieldWarning`` and never reaches the wire. The mismatch itself + # comes from a second defect in the same parameter: the ``data`` length + # callback omits the four octets ``reserved`` already consumed out of + # ``len``, so ``len`` grows by four on every round trip (measured: 4 -> 8). + 'hip-parameter/ENCRYPTED': Gap( + 'MISMATCH', '', + 'pcapkit/protocols/internet/hip.py:3445 -- cipher= is not a field of ' + 'EncryptedParameter and is silently dropped; and ' + 'pcapkit/protocols/schema/internet/hip.py:463 -- the data length ' + "callback omits the 4 octets 'reserved' took out of len"), + + # Two parameters whose own packed length is not what the header arithmetic + # can represent even in pairs -- see HIP_COPIES in the generator for why the + # pair is used at all. + 'hip-parameter/HIP_TRANSFORM': Gap( + 'CONSTRUCT', 'HIPv2: [ParamNo 577] invalid parameter', + 'pcapkit/protocols/internet/hip.py:698 -- the len check; HIP_TRANSFORM ' + 'packs to a length the make-side arithmetic cannot express'), + 'hip-parameter/HOST_ID': Gap( + 'CONSTRUCT', 'HIPv2: invalid format', + 'pcapkit/protocols/internet/hip.py:698 -- HOST_ID packs to 14 octets ' + 'with len=8, so it is not even 4-aligned'), + + # -- HTTP/2 --------------------------------------------------------------- + + # ``SchemaField.pack`` gives a nested frame schema a fresh packet context + # whose only link to the parent is ``__packet__``, but six frame schemas + # reach for the HTTP/2 header's ``flags`` bitfield directly -- either from a + # ConditionalField test or from ``FrameType.post_process``. The three frames + # that pass are exactly the three declaring no flag members at all: + # RST_STREAM, GOAWAY and WINDOW_UPDATE. + **{ + f'httpv2-frame/{name}': Gap( + 'CONSTRUCT', "KeyError: 'flags'", + 'pcapkit/protocols/schema/application/httpv2.py:144 ' + '(FrameType.post_process) and the pad_len ConditionalField tests at ' + ':175, :204, :305 -- the nested context reaches for the parent ' + "header's flags") + for name in ('DATA', 'HEADERS', 'SETTINGS', 'PUSH_PROMISE', 'PING', + 'CONTINUATION') + }, + + # ``make`` writes ``length = payload + 9`` and a PRIORITY payload is five + # octets, so the constructed header always says 14 -- while the reader + # demands exactly 9. Its siblings all check payload+9 (RST_STREAM 13, + # WINDOW_UPDATE 13, PING 17), so 9 looks like the outlier. + 'httpv2-frame/PRIORITY': Gap( + 'CONSTRUCT', 'HTTP/2: [Type 2] invalid format', + 'pcapkit/protocols/application/httpv2.py:572 -- reads length != 9 for a ' + 'frame make() always builds with length 14'), + + # -- PCAP-NG -------------------------------------------------------------- + + # ``PCAPNG.__post_init__`` packs and then re-parses *on the same instance*, + # and both halves share the per-instance option counter: the make path + # increments it, the read path then trips its own "only one of these" + # guard on the option it just built. Twenty-seven options have such a + # guard; the ten that construct are the ones that do not. + **{ + f'pcapng-option/{name}': Gap( + 'CONSTRUCT', 'option must be only one', + 'pcapkit/protocols/misc/pcapng.py:1078 packs and :1088 re-parses on ' + 'one instance, sharing self._opt; incremented at :3941, checked at ' + ':2175') + for name in ( + 'if_name_2', 'if_description_3', 'if_MACaddr_6', 'if_EUIaddr_7', + 'if_speed_8', 'if_tsresol_9', 'if_tzone_10', 'if_filter_11', + 'if_os_12', 'if_fcslen_13', 'if_tsoffset_14', 'if_hardware_15', + 'if_txspeed_16', 'if_rxspeed_17', + 'epb_flags_2', 'epb_dropcount_4', 'epb_packetid_5', 'epb_queue_6', + 'ns_dnsname_2', 'ns_dnsIP4addr_3', 'ns_dnsIP6addr_4', + 'isb_ifrecv_4', 'isb_ifdrop_5', 'isb_filteraccept_6', + 'isb_osdrop_7', 'isb_usrdeliv_8', + 'pack_flags_2', + ) + }, + + # ``_make_option_if_ipv6`` hardcodes ``length=8``, copied from its IPv4 + # sibling where 8 is right, while ``IPv6InterfaceField`` is 17 octets. The + # option packs to 21, the block total becomes 41, and no argument the caller + # can pass changes it. + 'pcapng-option/if_IPv6addr_5': Gap( + 'CONSTRUCT', 'invalid length: 41', + 'pcapkit/protocols/misc/pcapng.py:4190 -- length=8 for a 17-octet ' + 'IPv6InterfaceField'), + + # ``_isb_interface_id`` is read by these two make-side constructors but + # assigned only by ``_read_block_isb``. + 'pcapng-option/isb_starttime_2': Gap( + 'CONSTRUCT', "no attribute '_isb_interface_id'", + 'pcapkit/protocols/misc/pcapng.py:4991 -- reads an attribute set only ' + 'at :1790, on the parse path'), + 'pcapng-option/isb_endtime_3': Gap( + 'CONSTRUCT', "no attribute '_isb_interface_id'", + 'pcapkit/protocols/misc/pcapng.py:5024 -- as above'), + + # The three packet-carrying blocks lose their payload on the way back: + # ``_make_block_*`` never restores ``packet_data``, and it could not, since + # the data model has no field to keep it in -- the octets go to the + # next-layer dissector and survive only as the decoded chain. The rebuilt + # block keeps ``captured_len`` while carrying no data, so it is malformed + # rather than merely shorter. + **{ + f'pcapng-block/{name}': Gap( + 'MISMATCH', '', + 'pcapkit/protocols/misc/pcapng.py:3514, :3565, :3851 -- ' + 'packet_data is not restored, and ' + 'pcapkit/protocols/data/misc/pcapng.py:442, :474, :895 have no ' + 'field to restore it from') + for name in ('Enhanced_Packet_Block', 'Simple_Packet_Block', 'Packet_Block') + }, + + # The two key-log secrets writers terminate each line with ``os.sep`` -- a + # forward slash on POSIX -- where a newline is meant, while the readers + # split on newlines. So the whole log parses as one comment line and every + # entry is lost. They also stamp the current time into the body with no way + # to override it, which is why these two are the only cases in the suite + # whose failure detail is not stable, and why the generator's + # Decryption Secrets Block case uses a ZigBee key instead. + 'pcapng-secrets/TLS_Key_Log': Gap( + 'MISMATCH', '', + 'pcapkit/protocols/misc/pcapng.py:5553 and :5556 -- os.sep as a line ' + 'terminator, against splitlines() at ' + 'pcapkit/protocols/schema/misc/pcapng.py:1480; and datetime.now() in ' + 'the payload'), + 'pcapng-secrets/WireGuard_Key_Log': Gap( + 'MISMATCH', '', + 'pcapkit/protocols/misc/pcapng.py:5585 and :5587, against ' + 'pcapkit/protocols/schema/misc/pcapng.py:1519; and datetime.now()'), +} + + +#: Whether this interpreter gives every :class:`Schema` subclass its own +#: ``_abc_impl``, and therefore answers ``isinstance`` against schema classes +#: correctly. +#: +#: On CPython 3.10 and earlier they all share :class:`Schema`'s, which is GitHub +#: issue #439. Measured on 3.10.20 and 3.14.7: +#: ``EndRecord._abc_impl is IPv4Record._abc_impl`` is :data:`True` on 3.10 and +#: :data:`False` on 3.14, for the schema classes in +#: :mod:`pcapkit.protocols.schema.misc.pcapng`. Note that the *data* models +#: (:class:`~pcapkit.corekit.infoclass.Info` subclasses) are **not** affected on +#: either: ``Data_EndRecord._abc_impl is Info._abc_impl`` is :data:`False` on +#: both, so it is specifically the schema hierarchy. +SCHEMA_ABC_IS_PER_CLASS = sys.version_info >= (3, 11) + +#: Cases whose outcome is a property of the *interpreter* rather than of the +#: option, applying only where :data:`SCHEMA_ABC_IS_PER_CLASS` is false. +#: +#: This table exists reluctantly, and it is worth being explicit about why it is +#: not the same thing as :data:`EXPECTED_FAILURES`. +#: +#: A case that behaves differently on two supported Pythons is a defect, not a +#: variant, and the honest thing is to surface it rather than to encode it as a +#: shrug. So it is surfaced twice over: the entries below name the library defect +#: and the ``file:line`` that causes it, exactly as the main table does, and +#: :meth:`OptionRoundTripTests.test_schema_isinstance_is_interpreter_dependent` +#: pins the *mechanism* directly so the record is evidence and not an assertion +#: about a version number. What the table does not do is let the cases pass +#: quietly: on a modern interpreter they are held to ``'OK'`` by the main table's +#: absence of an entry, and on an old one they are held to failing in exactly the +#: recorded way. Fixing #439 turns this red on 3.10, which is the point. +#: +#: An entry here **overrides** :data:`EXPECTED_FAILURES` for the same label +#: rather than being disjoint from it, and three of the seven need that. The +#: ``ns_dns*`` options fail on every interpreter, but for different reasons: from +#: 3.11 on it is the shared ``self._opt`` counter rejecting the option the same +#: instance just built, and on 3.10 the block never gets that far because +#: ``post_process`` dies first. Same status, different cause, so both are recorded +#: and the interpreter decides which one is checked. The remaining four have no +#: entry in the main table at all, because they round-trip cleanly from 3.11. +#: +#: All seven are PCAP-NG name resolution, and they share one cause on 3.10. +#: ``NameResolutionBlock.post_process`` asks +#: ``isinstance(record, (IPv4Record, IPv6Record))`` at +#: :file:`pcapkit/protocols/schema/misc/pcapng.py:1248` and then reads +#: ``record.names``. Because every schema class shares one ``_abc_impl`` on 3.10, +#: that question returns :data:`True` for the block's terminating ``EndRecord`` +#: -- measured in the real path, with ``type(record) is EndRecord`` :data:`True` +#: and ``isinstance(record, (IPv4Record, IPv6Record))`` also :data:`True` in the +#: same breath -- so ``post_process`` reads ``names`` off a record that has none. +#: Every NRB carries a terminating ``EndRecord``, which is why the block, its +#: three ``ns_dns*`` options and its three records all go together. +INTERPRETER_GAPS = { + label: Gap( + 'CONSTRUCT', "'EndRecord' object has no attribute 'names'", + 'pcapkit/protocols/schema/misc/pcapng.py:1248 -- isinstance against ' + 'schema classes, which share one _abc_impl on CPython <= 3.10 (#439), so ' + 'the terminating EndRecord tests True as an IPv4Record/IPv6Record') + for label in ( + 'pcapng-block/Name_Resolution_Block', + 'pcapng-option/ns_dnsname_2', + 'pcapng-option/ns_dnsIP4addr_3', + 'pcapng-option/ns_dnsIP6addr_4', + 'pcapng-record/nrb_record_end', + 'pcapng-record/nrb_record_ipv4', + 'pcapng-record/nrb_record_ipv6', + ) +} + + +def _abc_impl(cls: 'type') -> 'Any': + """The ABC implementation object a class dispatches ``isinstance`` through. + + ``_abc_impl`` is a CPython internal and is not in the type stubs, so it is + read by name. Present on both the C ``_abc`` and the pure-python ``_py_abc`` + backends; :func:`tests._support._reset_abc_caches` relies on the same + attribute. + + Args: + cls: The class to inspect. + + Returns: + The class's ``_abc_impl``, or :data:`None` if it has none. + + """ + return getattr(cls, '_abc_impl', None) + + +def _gap_for(label: 'str') -> 'Optional[Gap]': + """The recorded expectation for ``label``, or :data:`None` if it should pass. + + Args: + label: Case label, e.g. ``'tcp-mptcp/MP_JOIN'``. + + Returns: + The recorded :class:`Gap`, or :data:`None` when the case is expected to + round-trip on this interpreter. + + """ + # The interpreter-conditional record wins where there is one, because on an + # affected interpreter the case fails there before it can reach whatever the + # main table describes. + if not SCHEMA_ABC_IS_PER_CLASS: + gap = INTERPRETER_GAPS.get(label) + if gap is not None: + return gap + return EXPECTED_FAILURES.get(label) + + +def _load_generator() -> 'types.ModuleType': + """Load :file:`examples/generators/options.py` by path. + + That directory is not a package, and its module names are too generic to put + on :data:`sys.path` -- which is exactly why + :file:`examples/generators/make_samples.py` loads its siblings by path too. + This follows it. + + Returns: + The generator module, which exposes ``cases``, ``roundtrip``, ``FAMILIES`` + and ``SKIP``. + + Raises: + RuntimeError: If the module cannot be found or loaded. + + """ + path = ROOT / 'examples' / 'generators' / 'options.py' + spec = importlib.util.spec_from_file_location('pcapkit_samples_options', path) + if spec is None or spec.loader is None: # pragma: no cover + raise RuntimeError(f'cannot load the option case table from {path}') + + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class OptionRoundTripTests(unittest.TestCase): + """One construct -> parse -> construct cycle per registered code.""" + + #: The generator module, loaded once for the whole class. Loading it imports + #: :mod:`pcapkit`, so it must happen after :meth:`setUp` has purged the + #: previous test's copy -- hence a class attribute filled in + #: :meth:`setUpClass` rather than a module-level import. + options = None # type: Any + + @classmethod + def setUpClass(cls) -> None: + purge_modules(['pcapkit']) + cls.options = _load_generator() + + def _run(self, case: 'Any') -> 'Any': + """One cycle, under this suite's deadline rather than the generator's. + + The generator applies its own :func:`signal.alarm` deadline, because + ``make samples`` must not hang either. Letting both arm the alarm would + break the outer one: the inner context cancels the pending alarm on the + way out, which is the outer deadline's countdown. So the generator's is + switched off here and :func:`tests._support.time_limit` owns the clock, + which is what the rest of the suite uses. + + Args: + case: The case to exercise. + + Returns: + The cycle's outcome, with a timeout reported as ``'TIMEOUT'`` rather + than raised, so that it is asserted on like any other failure. + + """ + try: + with time_limit(CASE_TIMEOUT): + return self.options.roundtrip(case, deadline=0) + except TimeoutError as exc: + return self.options.Outcome(case, 'TIMEOUT', str(exc), None, ()) + + def test_every_registered_code_has_a_case(self) -> None: + """No registry entry is left without a case. + + This is what makes the coverage self-maintaining. Registering a new + option and forgetting to exercise it fails here, rather than going + unnoticed until the option turns out not to construct. + + """ + for family in self.options.FAMILIES: + with self.subTest(family=family.label): + registry = family.registry() + covered = {case.code for case in self.options.cases((family,))} + skipped = { + code for code in registry + if (family.label, self.options.code_name(code)) in self.options.SKIP + } + self.assertEqual( + set(registry), covered | skipped, + f'{family.label}: every code in the registry needs a case in ' + f'examples/generators/options.py, or an entry in its SKIP table' + ) + + def test_every_registered_code_resolves_without_growing_the_registry(self) -> None: + """Each code resolves to a handler, and asking does not mutate the registry. + + Two things at once, because they are the same lookup. + + A registry entry naming a handler that does not exist is a live bug that + nothing else catches: dispatch falls through to the fallback and the + option is quietly parsed as unassigned. Two of the ``MH`` registries' + own docstrings name prefixes that were never implemented + (``_read_option_*`` and ``_read_extension_*``, where the code uses + ``_read_opt_*`` and ``_read_ext_*``), so the hazard is not theoretical. + + And every one of these registries is a :class:`collections.defaultdict` + on a class attribute, so a lookup that misses *inserts*, permanently, for + the whole process. The sizes are compared either side to prove the + non-recording path was used. + + """ + for family in self.options.FAMILIES: + with self.subTest(family=family.label): + registry = family.registry() + before = len(registry) + + for code in list(registry): + resolved = self.options.handler(registry, code) + self.assertIsNotNone( + resolved, + f'{family.label}: {self.options.code_name(code)} resolves ' + f'to nothing' + ) + + self.assertEqual( + len(registry), before, + f'{family.label}: the registry grew while being read; a ' + f'lookup went through registry[code] instead of ' + f'ProtocolBase._lookup_registry' + ) + + def test_expected_failures_name_real_cases(self) -> None: + """Both tables name only cases that exist. + + Without this they rot silently: a renamed or removed registry entry + leaves behind an entry that can never be checked, and which then reads as + documentation of a defect nobody can find. + + """ + labels = {case.label for case in self.options.cases()} + for name, table in (('EXPECTED_FAILURES', EXPECTED_FAILURES), + ('INTERPRETER_GAPS', INTERPRETER_GAPS)): + with self.subTest(table=name): + stale = sorted(set(table) - labels) + self.assertEqual( + stale, [], + f'these entries of {name} name cases that no longer exist; ' + f'delete them, or fix the label' + ) + + # An override that records the same thing as the entry it overrides is + # noise, and would quietly outlive the reason it was added. + for label in sorted(set(EXPECTED_FAILURES) & set(INTERPRETER_GAPS)): + with self.subTest(overridden=label): + general, specific = EXPECTED_FAILURES[label], INTERPRETER_GAPS[label] + self.assertNotEqual( + (general.status, general.fragment), + (specific.status, specific.fragment), + f'{label} records the same status and fragment in both tables, ' + f'so the interpreter-conditional entry adds nothing; delete it' + ) + + def test_schema_isinstance_is_interpreter_dependent(self) -> None: + """Pin the mechanism behind :data:`INTERPRETER_GAPS`, so it is evidence. + + The seven cases in that table are excused on old interpreters, and an + excuse resting on a version comparison would be indistinguishable from + giving up. So the *cause* is asserted here directly: every + :class:`~pcapkit.protocols.schema.schema.Schema` subclass shares one + ``_abc_impl`` on CPython 3.10 and earlier, and has its own from 3.11 -- + which is what makes ``isinstance`` against a schema class answer by cache + order rather than by type. + + This is GitHub issue #439 and belongs to ``SchemaMeta``, not here. When it + is fixed this test fails on 3.10, which is the prompt to delete both it + and :data:`INTERPRETER_GAPS`. + + The data models are checked too, and asserted *unaffected* on both + interpreters -- that is the boundary of the defect, and getting it wrong + in either direction would send the next reader down the wrong path. + + """ + from pcapkit.corekit.infoclass import Info + from pcapkit.protocols.data.misc.pcapng import EndRecord as Data_EndRecord + from pcapkit.protocols.schema.misc.pcapng import EndRecord, IPv4Record + from pcapkit.protocols.schema.schema import Schema + + shared = _abc_impl(EndRecord) is _abc_impl(IPv4Record) + self.assertEqual( + shared, not SCHEMA_ABC_IS_PER_CLASS, + f'on Python {sys.version_info[0]}.{sys.version_info[1]}, schema ' + f'classes sharing one _abc_impl is {shared}; INTERPRETER_GAPS assumes ' + f'{not SCHEMA_ABC_IS_PER_CLASS}. If #439 is fixed, delete that table ' + f'and this test.' + ) + if shared: + self.assertIs( + _abc_impl(EndRecord), _abc_impl(Schema), + "the shared _abc_impl should be Schema's own" + ) + + # The boundary: Info subclasses are unaffected on every interpreter, so + # nothing that dispatches on a *data* model is implicated in #439. + self.assertIsNot( + _abc_impl(Data_EndRecord), _abc_impl(Info), + 'the data models were unaffected by #439 on both 3.10 and 3.14; if ' + 'that has changed, the scope of the defect is wider than recorded' + ) + + def test_round_trip_is_identity_or_a_recorded_gap(self) -> None: + """Every case either closes the cycle, or fails exactly as recorded. + + Both halves matter. A case absent from :data:`EXPECTED_FAILURES` must + come back ``'OK'``; a case present in it must still fail, and with the + recorded status and detail, so that fixing the defect turns this red and + the entry gets deleted rather than left behind. + + """ + cases = self.options.cases() + self.assertGreater(len(cases), 200, + 'the registries should yield a few hundred codes; a ' + 'much smaller number means the enumeration broke') + + for case in cases: + with self.subTest(case=case.label): + outcome = self._run(case) + gap = _gap_for(case.label) + + if gap is None: + self.assertEqual( + outcome.status, 'OK', + f'{case.label} no longer round-trips: {outcome.detail}. ' + f'If this is a newly found defect, add it to ' + f'EXPECTED_FAILURES with the file:line that causes it -- ' + f'do not change the case to avoid it.' + ) + continue + + self.assertEqual( + outcome.status, gap.status, + f'{case.label} was recorded as failing with ' + f'{gap.status} ({gap.defect}) but came back ' + f'{outcome.status}: {outcome.detail}. If the defect is fixed, ' + f'delete its EXPECTED_FAILURES entry.' + ) + fragments = ((gap.fragment,) if isinstance(gap.fragment, str) + else gap.fragment) + for fragment in fragments: + if not fragment: + continue + self.assertIn( + fragment, outcome.detail, + f'{case.label} still fails with {gap.status}, but not in ' + f'the recorded way ({gap.defect}); detail was ' + f'{outcome.detail!r}' + ) + + def test_a_single_hip_parameter_cannot_be_constructed(self) -> None: + """A HIP packet carrying exactly one parameter is rejected by its own reader. + + This is the defect the generator's ``HIP_COPIES = 2`` routes around, and + it is pinned here so that routing around it does not also bury it. + + ``HIP.make`` computes the header's ``len`` as ``total_length // 8 + 4``, + which is lossless only when the parameter octets are a multiple of eight. + The parameter padding rule pads the *contents* to eight and ignores the + four-octet type-and-length header, so one parameter is always + ``4 (mod 8)``; the floor division drops those four octets, and + ``_read_hip_param`` compares the recovered length exactly and raises. + + Two copies sum to a multiple of eight, so the same parameter that fails + alone succeeds in a pair -- which is the control that makes this a + statement about the header arithmetic rather than about ``SEQ``. + + """ + from pcapkit.const.hip.parameter import Parameter + from pcapkit.protocols.internet.hip import HIP + from pcapkit.utilities.exceptions import ProtocolError + + base = dict(self.options.HIP_BASE) + one = [(Parameter.SEQ, {})] # type: list[tuple[Any, dict[str, Any]]] + + with self.assertRaises(ProtocolError) as caught: + HIP(parameters=one, extension=True, **base) + self.assertIn('invalid format', str(caught.exception)) + + # The control: the identical parameter, twice, round-trips exactly. + paired = bytes(HIP(parameters=one * 2, extension=True, **base)) + reparsed = HIP(paired, len(paired), extension=True) + again = bytes(HIP(parameters=reparsed.info.parameters, extension=True, **base)) + self.assertEqual(paired, again) + + def test_recorded_gaps_are_a_minority(self) -> None: + """Most of the option space round-trips, and the rest is accounted for. + + A guard on the shape of the result rather than on any one case: if a + change makes the failures outnumber the successes, something systemic + broke and the per-case assertions above will be too noisy to read. + + """ + total = len(self.options.cases()) + recorded = len(EXPECTED_FAILURES) + if not SCHEMA_ABC_IS_PER_CLASS: + recorded += len(INTERPRETER_GAPS) + self.assertLess( + recorded, total // 2, + f'{recorded} of {total} cases are recorded as failing; that is more ' + f'than half, which suggests the harness rather than the library is at ' + f'fault' + ) + + +if __name__ == '__main__': + unittest.main()