From df67d72d297b95563074bff6cc60f5769dc3c4bc Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Tue, 22 Sep 2026 15:37:13 -0400 Subject: [PATCH] fix(pcapng): bound an option's payload to the area its block declares (#594) A PCAP-NG option's length is a 16-bit wire field, so a four-octet option header can declare 65,535 octets of payload. Nothing bounded that against the option area the block frames, and the field layer pads every shortfall inside a 16-bit length unconditionally -- deliberately, so a snapshot-truncated capture still parses -- so repeating such an option across blocks amplified without limit. * `bounded_option()` clamps an option or record payload to the octets its area has left at that field, and warns (`SchemaWarning`) when it does. Applied to all 15 variable-width option and record payloads in the schema. * `bounded_area()` clamps a packet block's option area to the octets the block itself holds, less the trailing Block Total Length. Without it a block could declare 1,000,000 octets while holding 36, size its area from the lie, and synthesise 65,535 octets anyway -- 1,820x. A no-op on well-formed blocks, where the two are equal by construction. * The bound comes from this layer because the field layer cannot see it: what separates the crafted case from a legitimate one is inconsistency with the block's own declared framing, not the shortfall's size. * Clamping, not refusing: a block read has no catch point above `FieldBase.unpack`, so one refusal aborts the whole extraction. It reads only the block's own framing, so it is history-independent. * The negative-remainder guard is load-bearing on the unpack path, where `__length__` can already be past zero: without it a ten-octet area raises `struct.error` from a `'-2s'` template. * Block-level payload fields are left alone; #593 already budgets that 32-bit band, and the five non-packet option areas keep the framing assumption, which #678 is the general fix for. Crafted capture of 2,000 Enhanced Packet Blocks in 80,048 octets: 131,070,000 octets of padding and 1637.393x before, 0 octets and 0.000x after, with all 2,000 frames and 2,000 options still parsed. Worst ratio over five adversarial shapes after the fix is 0.862x, against 1637.393x/960.360x/224.067x before. Zero clamps fire across all six PCAP-NG fixtures (338 options), and 501 truncation levels of `dhcp.pcapng` are byte-identical. 117 tests and 742 subtests pass across the PCAP-NG and contract suites; 9 of the 15 new tests fail on `main` (exit 1 to 0). Fixes #594 --- pcapkit/protocols/schema/misc/pcapng.py | 175 +++++++++- tests/protocols/misc/test_pcapng_unit.py | 400 +++++++++++++++++++++++ 2 files changed, 557 insertions(+), 18 deletions(-) diff --git a/pcapkit/protocols/schema/misc/pcapng.py b/pcapkit/protocols/schema/misc/pcapng.py index b0e6418d4..56454f409 100644 --- a/pcapkit/protocols/schema/misc/pcapng.py +++ b/pcapkit/protocols/schema/misc/pcapng.py @@ -29,7 +29,7 @@ from pcapkit.protocols.schema.schema import EnumSchema, Schema, schema_final from pcapkit.utilities.exceptions import FieldValueError, ProtocolError, stacklevel from pcapkit.utilities.logging import SPHINX_TYPE_CHECKING -from pcapkit.utilities.warnings import ProtocolWarning, warn +from pcapkit.utilities.warnings import ProtocolWarning, SchemaWarning, warn __all__ = [ 'PCAPNG', @@ -207,6 +207,143 @@ def shb_byteorder_callback(field: 'NumberField', packet: 'dict[str, Any]') -> 'N packet['byteorder'] = field._byteorder +def bounded_option(length: 'Callable[[dict[str, Any]], int]') -> 'Callable[[dict[str, Any]], int]': + """Clamp an option or record payload to the octets its area still declares. + + Args: + length: Callback computing the payload's nominal length, as the option's + or record's own declared length field gives it. + + Returns: + A callback returning that length, never past the octets the enclosing + option or record area has left to give. + + An option's length is a 16-bit wire field, so it can declare up to 65,535 + octets of payload from four octets of option header. Nothing bounds that + against the area the option sits in: :meth:`OptionField.unpack + ` subtracts each + option's *parsed* size from the area it was given but never checks the + declared size against it, and :meth:`FieldBase.unpack + ` zero-pads any shortfall + within :data:`~pcapkit.corekit.fields.field._MAX_ZERO_PAD_SHORTFALL` + unconditionally -- deliberately, since that ceiling is the full span of a + 16-bit length and a capture cut short by its snapshot length must still + parse. Repeating such an option across many blocks therefore synthesised + padding without limit: 2,000 Enhanced Packet Blocks in 80,048 octets, each + with one option declaring 65,535 against none present, produced 131,070,000 + octets of zero padding, an amplification of 1,637x linear in the block + count. See `#594 `__, + and `#593 `__ for the + 32-bit band the field layer's own budget already covers. + + The bound has to come from this layer because the field layer cannot see + it. What distinguishes the crafted case from the legitimate one is not the + shortfall's size -- both are inside a 16-bit length, which is why #571's + ``len(buffer) < length`` rejection was declined -- but whether the option + is inconsistent with the framing the block itself declares. Block Total + Length is authoritative and cross-checked against its own trailing copy, so + the area is ``length`` less the fixed fields, ``captured_len`` and its + padding; an option declaring more payload than that area has left is + malformed however complete the file behind it is. A snapshot-truncated + capture says so through ``captured_len`` instead, and leaves its options + whole, so it never trips this. + + Clamping rather than refusing is what keeps the `#431 + `__ accommodation: a + block read has no catch point above :meth:`FieldBase.unpack + `, so one refusal aborts the + whole extraction rather than one block, and a truncated capture would stop + parsing at the cut instead of reporting the frames before it. The clamp is + also history-independent -- it reads only this block's own declared framing + -- which a running threshold would not be. + + Note: + The clamp is skipped when ``__length__`` is absent or negative, which is + what :meth:`Schema.pack ` + leaves it as when no length is known. That matters here in a way it does + not for :func:`pcapkit.protocols.schema.transport.sctp.bounded`, whose + list fields ignore their length while packing: these are + :class:`~pcapkit.corekit.fields.strings.BytesField` and + :class:`~pcapkit.corekit.fields.strings.StringField` payloads, and + :meth:`FieldBase.pack ` + packs them through ``struct.pack('s', ...)``, which truncates + silently. A clamp applied while packing would therefore shorten a + perfectly good option rather than reject it. + + """ + def callback(pkt: 'dict[str, Any]') -> 'int': + nominal = length(pkt) + + remaining = pkt.get('__length__') + if not isinstance(remaining, int) or remaining < 0 or nominal <= remaining: + return nominal + + warn(f'PCAP-NG: option declares {nominal} octet(s) of payload with ' + f'{remaining} octet(s) left in its area; reading {remaining}', + SchemaWarning, stacklevel=stacklevel()) + return remaining + return callback + + +def bounded_area(length: 'Callable[[dict[str, Any]], int]') -> 'Callable[[dict[str, Any]], int]': + """Clamp a packet block's option area to the octets the block itself holds. + + Args: + length: Callback computing the option area's nominal span, from the + block's own declared Block Total Length. + + Returns: + A callback returning that span, never past the octets left of the block. + + :func:`bounded_option` bounds a payload by the area, and the area by the + block's declared Block Total Length. That closes the band only while the + declared length is itself backed by real octets, and nothing checks that: + ``BlockType.post_process`` compares ``length`` against its own trailing copy + and never against the file. A block declaring 1,000,000 octets while holding + 36 therefore sizes its option area at 999,964, an option inside it declaring + 65,535 is under that and is not clamped, and 65,535 octets of zeros are + synthesised from 36 -- measured, 1,820x, with no warning. Clamping the area + to ``__length__`` as well removes the step: the payload is then bounded by + the octets the block was actually handed, whatever it declared. + + This is a no-op on every well-formed block rather than a second guess at the + area. At the option field the only field still to come is the trailing Block + Total Length, so ``__length__`` is exactly the area plus that field's four + octets, and subtracting them makes the two equal. Taking ``__length__`` whole + over-grants by exactly four, which is not academic: the area then reaches the + trailing length itself and reads it as option payload, measured as four + octets of payload on a block that holds none. + + Note: + The five non-packet blocks' option areas -- Section Header, Interface + Description, Name Resolution, Interface Statistics and Decryption + Secrets -- are deliberately left unclamped here, since each computes its + span with a different offset and the equality above has to be + re-established per block rather than assumed. The general fix is to stop + a declared length reaching a read at all, which is `#678 + `__. + + """ + def callback(pkt: 'dict[str, Any]') -> 'int': + nominal = length(pkt) + + remaining = pkt.get('__length__') + if not isinstance(remaining, int): + return nominal + + # The trailing Block Total Length follows the option area in both packet + # block types, so it is not the area's to read. + available = remaining - 4 + if available < 0 or nominal <= available: + return nominal + + warn(f'PCAP-NG: block declares an option area of {nominal} octet(s) with ' + f'{available} octet(s) left of the block; reading {available}', + SchemaWarning, stacklevel=stacklevel()) + return available + return callback + + def pcapng_block_selector(packet: 'dict[str, Any]') -> 'Field': """Selector function for :attr:`PCAPNG.block` field. @@ -491,7 +628,7 @@ class UnknownOption(_OPT_Option): """Header schema for unknown PCAP-NG file options.""" #: Option value. - data: 'bytes' = BytesField(length=lambda pkt: pkt['length']) + data: 'bytes' = BytesField(length=bounded_option(lambda pkt: pkt['length'])) #: Padding. padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) @@ -512,7 +649,7 @@ class CommentOption(_OPT_Option, code=Enum_OptionType.opt_comment): """Header schema for PCAP-NG file ``opt_comment`` options.""" #: Comment text. - comment: 'str' = StringField(length=lambda pkt: pkt['length'], encoding='utf-8') + comment: 'str' = StringField(length=bounded_option(lambda pkt: pkt['length']), encoding='utf-8') #: Padding. padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) @@ -530,7 +667,7 @@ class CustomOption(_OPT_Option, code=[Enum_OptionType.opt_custom_2988, #: Private enterprise number (PEN). pen: 'int' = UInt32Field(callback=byteorder_callback) #: Custom data. - data: 'bytes' = BytesField(length=lambda pkt: pkt['length'] - 4) + data: 'bytes' = BytesField(length=bounded_option(lambda pkt: pkt['length'] - 4)) #: Padding. padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) @@ -642,7 +779,7 @@ class IF_NameOption(_IF_Option, code=Enum_OptionType.if_name): """Header schema for PCAP-NG file ``if_name`` options.""" #: Interface name. - name: 'str' = StringField(length=lambda pkt: pkt['length'], encoding='utf-8') + name: 'str' = StringField(length=bounded_option(lambda pkt: pkt['length']), encoding='utf-8') #: Padding. padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) @@ -655,7 +792,7 @@ class IF_DescriptionOption(_IF_Option, code=Enum_OptionType.if_description): """Header schema for PCAP-NG file ``if_description`` options.""" #: Interface description. - description: 'str' = StringField(length=lambda pkt: pkt['length'], encoding='utf-8') + description: 'str' = StringField(length=bounded_option(lambda pkt: pkt['length']), encoding='utf-8') #: Padding. padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) @@ -781,7 +918,7 @@ class IF_FilterOption(_IF_Option, code=Enum_OptionType.if_filter): #: Filter code. code: 'Enum_FilterType' = EnumField(length=1, namespace=Enum_FilterType, callback=byteorder_callback) #: Capture filter. - filter: 'bytes' = BytesField(length=lambda pkt: pkt['length'] - 1) + filter: 'bytes' = BytesField(length=bounded_option(lambda pkt: pkt['length'] - 1)) #: Padding. padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) @@ -794,7 +931,7 @@ class IF_OSOption(_IF_Option, code=Enum_OptionType.if_os): """Header schema for PCAP-NG file ``if_os`` options.""" #: OS information. - os: 'str' = StringField(length=lambda pkt: pkt['length'], encoding='utf-8') + os: 'str' = StringField(length=bounded_option(lambda pkt: pkt['length']), encoding='utf-8') #: Padding. padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) @@ -833,7 +970,7 @@ class IF_HardwareOption(_IF_Option, code=Enum_OptionType.if_hardware): """Header schema for PCAP-NG file ``if_hardware`` options.""" #: Hardware information. - hardware: 'str' = StringField(length=lambda pkt: pkt['length'], encoding='utf-8') + hardware: 'str' = StringField(length=bounded_option(lambda pkt: pkt['length']), encoding='utf-8') #: Padding. padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) @@ -938,7 +1075,7 @@ class EPB_HashOption(_EPB_Option, code=Enum_OptionType.epb_hash): #: Hash algorithm. func: 'Enum_HashAlgorithm' = EnumField(length=1, namespace=Enum_HashAlgorithm, callback=byteorder_callback) #: Hash value. - data: 'bytes' = BytesField(length=lambda pkt: pkt['length'] - 1) + data: 'bytes' = BytesField(length=bounded_option(lambda pkt: pkt['length'] - 1)) #: Padding. padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) @@ -992,7 +1129,7 @@ class EPB_VerdictOption(_EPB_Option, code=Enum_OptionType.epb_verdict): #: Verdict type. verdict: 'Enum_VerdictType' = EnumField(length=1, namespace=Enum_VerdictType, callback=byteorder_callback) #: Verdict value. - value: 'bytes' = BytesField(length=lambda pkt: pkt['length'] - 1) + value: 'bytes' = BytesField(length=bounded_option(lambda pkt: pkt['length'] - 1)) #: Padding. padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) @@ -1028,7 +1165,8 @@ class EnhancedPacketBlock(BlockType, code=Enum_BlockType.Enhanced_Packet_Block): # ``padding_data``: a PaddingField is written straight into the schema # buffer while packing and never lands in the packet data, so its name # is not a key here on the packing path. - length=lambda pkt: pkt['length'] - 32 - pkt['captured_len'] - (4 - pkt['captured_len'] % 4) % 4, + length=bounded_area(lambda pkt: pkt['length'] - 32 - pkt['captured_len'] + - (4 - pkt['captured_len'] % 4) % 4), base_schema=_EPB_Option, type_name='type', registry=Option.registry['epb'], @@ -1086,7 +1224,7 @@ class UnknownRecord(NameResolutionRecord): """Header schema for PCAP-NG NRB unknown records.""" #: Unknown record data. - data: 'bytes' = BytesField(length=lambda pkt: pkt['length']) + data: 'bytes' = BytesField(length=bounded_option(lambda pkt: pkt['length'])) #: Padding. padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) @@ -1109,7 +1247,7 @@ class IPv4Record(NameResolutionRecord, code=Enum_RecordType.nrb_record_ipv4): #: IPv4 address. ip: 'IPv4Address' = IPv4AddressField() #: Name resolution data. - resol: 'str' = StringField(length=lambda pkt: pkt['length'] - 4) + resol: 'str' = StringField(length=bounded_option(lambda pkt: pkt['length'] - 4)) #: Padding. padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) @@ -1140,7 +1278,7 @@ class IPv6Record(NameResolutionRecord, code=Enum_RecordType.nrb_record_ipv6): #: IPv6 address. ip: 'IPv6Address' = IPv6AddressField() #: Name resolution data. - resol: 'str' = StringField(length=lambda pkt: pkt['length'] - 16) + resol: 'str' = StringField(length=bounded_option(lambda pkt: pkt['length'] - 16)) #: Padding. padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) @@ -1178,7 +1316,7 @@ class NS_DNSNameOption(_NS_Option, code=Enum_OptionType.ns_dnsname): """Header schema for PCAP-NG ``ns_dnsname`` option.""" #: DNS name. - name: 'str' = StringField(length=lambda pkt: pkt['length']) + name: 'str' = StringField(length=bounded_option(lambda pkt: pkt['length'])) #: Padding. padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) @@ -1689,7 +1827,7 @@ class PACK_HashOption(_PACK_Option, code=Enum_OptionType.pack_hash): #: Hash algorithm. func: 'Enum_HashAlgorithm' = EnumField(length=1, namespace=Enum_HashAlgorithm, callback=byteorder_callback) #: Hash value. - data: 'bytes' = BytesField(length=lambda pkt: pkt['length'] - 1) + data: 'bytes' = BytesField(length=bounded_option(lambda pkt: pkt['length'] - 1)) #: Padding. padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) @@ -1725,7 +1863,8 @@ class PacketBlock(BlockType, code=Enum_BlockType.Packet_Block): options: 'list[Option]' = OptionField( # NOTE: see EnhancedPacketBlock.options on why the padding is recomputed # here instead of being read back from ``padding_data``. - length=lambda pkt: pkt['length'] - 32 - pkt['captured_length'] - (4 - pkt['captured_length'] % 4) % 4, + length=bounded_area(lambda pkt: pkt['length'] - 32 - pkt['captured_length'] + - (4 - pkt['captured_length'] % 4) % 4), base_schema=_PACK_Option, type_name='type', registry=Option.registry['pack'], diff --git a/tests/protocols/misc/test_pcapng_unit.py b/tests/protocols/misc/test_pcapng_unit.py index 849381a6f..d186ba170 100644 --- a/tests/protocols/misc/test_pcapng_unit.py +++ b/tests/protocols/misc/test_pcapng_unit.py @@ -3180,5 +3180,405 @@ def test_sample_capture_timestamp_matches_its_own_raw_bytes(self) -> None: 'no tzdata installed, so this test proves nothing') +#: An option code registered in no PCAP-NG namespace, so it selects +#: ``UnknownOption``, whose payload is sized straight from its declared length. +UNKNOWN_OPT = 0x00FA + + +def epb_body(options: bytes, packet_data: bytes = bytes(4)) -> bytes: + """Build an Enhanced Packet Block body around ``options``. + + The returned buffer is what :func:`block_body` wraps, i.e. the EPB without + its block type or either block total length. ``EnhancedPacketBlock`` sizes + its option area as ``length - 32 - captured_len - padding``, so wrapping + this makes the area exactly ``len(options)``. + + """ + padding = -len(packet_data) % 4 + return (struct.pack(' bytes: + """An option whose length field claims ``declared`` octets of ``value``.""" + return struct.pack(' None: + purge_modules(['pcapkit']) + + def _unpack_epb(self, options: bytes, packet_data: bytes = bytes(4)): + """Unpack an EPB whose option area is exactly ``options``.""" + from pcapkit.protocols.schema.misc.pcapng import EnhancedPacketBlock + + raw = block_body(epb_body(options, packet_data)) + schema = EnhancedPacketBlock.unpack(raw, len(raw), {'byteorder': 'little'}) + + self.assertEqual(schema.length, schema.length2) + self.assertEqual(schema.length, len(raw) + 4) + return schema + + def _unpack_epb_declaring(self, options: bytes, declared: int, + packet_data: bytes = bytes(4)): + """Unpack an EPB whose Block Total Length lies about its own size.""" + from pcapkit.protocols.schema.misc.pcapng import EnhancedPacketBlock + + body = epb_body(options, packet_data) + raw = struct.pack(' int: + """Total octets the block's options report holding.""" + total = 0 + for option in schema.options: + for name in ('data', 'comment', 'name', 'description', 'os', + 'hardware', 'filter', 'value', 'resol'): + payload = getattr(option, name, None) + if isinstance(payload, (bytes, str)): + total += len(payload) + return total + + def test_an_option_declaring_more_than_its_area_reads_only_the_area(self) -> None: + """#594's vector, one block: 65,535 declared against a four-octet area. + + The area is exactly the option header, so there is nothing left for the + payload. Before the bound this read ``b''`` off an exhausted stream and + zero-padded it to 65,535 octets -- from a 40-octet block. + + """ + schema = self._unpack_epb(over_declared(UNKNOWN_OPT, b'', 0xFFFF)) + + self.assertEqual(len(schema.options), 1) + self.assertEqual(schema.options[0].length, 0xFFFF) + self.assertEqual(schema.options[0].data, b'') + self.assertEqual(self._payload_octets(schema), 0) + + def test_an_option_that_exactly_fills_its_area_is_untouched(self) -> None: + """Four octets declared and four present, in an eight-octet area. + + This is the input a bound applied one header-width off would break: the + payload is available in full, and clamping it to the area *including* + the four octets of option header the payload sits behind would read zero + octets instead of four. + + """ + schema = self._unpack_epb(tlv(UNKNOWN_OPT, b'\xaa\xbb\xcc\xdd')) + + self.assertEqual(len(schema.options), 1) + self.assertEqual(schema.options[0].length, 4) + self.assertEqual(schema.options[0].data, b'\xaa\xbb\xcc\xdd') + + def test_an_option_over_declaring_by_one_octet_loses_only_that_octet(self) -> None: + """Five declared and four present, in an eight-octet area. + + Before the bound the fifth octet was synthesised as a zero; the value + was five octets long for an option that held four. The bound reads the + four that are there. Paired with the exact-fit case above, this is what + separates "clamp to what the area has left" from either leaving the + shortfall padded or refusing the option outright. + + """ + schema = self._unpack_epb(over_declared(UNKNOWN_OPT, b'\xaa\xbb\xcc\xdd', 5)) + + self.assertEqual(len(schema.options), 1) + self.assertEqual(schema.options[0].length, 5) + self.assertEqual(schema.options[0].data, b'\xaa\xbb\xcc\xdd') + + def test_an_option_over_declaring_by_four_octets_is_clamped_to_the_remainder(self) -> None: + """Eight declared and four present, in an eight-octet area. + + The sharpest discriminator in this class. The option area is eight + octets and the option declares eight, so a bound taken against the + *area* leaves the declared length untouched and four zero octets are + still synthesised. The bound has to be taken against what the area has + left *at that field* -- four octets, the header having consumed the + other four -- which is what reads the four real octets and no more. + + """ + schema = self._unpack_epb(over_declared(UNKNOWN_OPT, b'\xaa\xbb\xcc\xdd', 8)) + + self.assertEqual(len(schema.options), 1) + self.assertEqual(schema.options[0].length, 8) + self.assertEqual(schema.options[0].data, b'\xaa\xbb\xcc\xdd') + + def test_a_whole_sixteen_bit_option_the_block_declares_room_for_is_untouched(self) -> None: + """65,535 octets declared, 65,535 present, and the block says so. + + This is the input that a constant ceiling on option payloads would + break, and the reason the bound is the area rather than a number: a + 65,535-octet option is legitimate when the block reserves room for it, + and nothing here is short. + + """ + value = b'\xcd' * 0xFFFF + schema = self._unpack_epb(tlv(UNKNOWN_OPT, value)) + + self.assertEqual(len(schema.options), 1) + self.assertEqual(schema.options[0].length, 0xFFFF) + self.assertEqual(schema.options[0].data, value) + + def test_a_tail_option_gets_only_what_the_options_before_it_left(self) -> None: + """Two options in a twelve-octet area, the second declaring 65,535. + + The first option consumes eight of the twelve octets honestly, so the + second has four -- its own header -- and no payload. A bound taken + against the area as declared, rather than against the area as the + earlier options left it, would hand the second option twelve octets and + synthesise eight of them. This is the input that separates the two. + + """ + options = (tlv(UNKNOWN_OPT, b'\xaa\xaa\xaa\xaa') + + over_declared(UNKNOWN_OPT + 1, b'', 0xFFFF)) + schema = self._unpack_epb(options) + + self.assertEqual(len(schema.options), 2) + self.assertEqual(schema.options[0].data, b'\xaa\xaa\xaa\xaa') + self.assertEqual(schema.options[1].length, 0xFFFF) + self.assertEqual(schema.options[1].data, b'') + self.assertEqual(self._payload_octets(schema), 4) + + def test_the_payload_never_exceeds_the_area_for_any_declared_length(self) -> None: + """The bound itself, over the whole interesting space rather than one input. + + For every option area size and every declared length, the octets the + area's options report holding must not exceed the area. That is the + property that bounds the amplification: a block's options can never + synthesise more than the block's own declared length, and the block's + declared length is what the reader advances the file by. + + """ + for area_options in (1, 2, 3): + for declared in (0, 1, 4, 5, 8, 12, 0x100, 0x1000, 0xFFFF): + with self.subTest(options=area_options, declared=declared): + options = b''.join( + over_declared(UNKNOWN_OPT + index, b'', declared) + for index in range(area_options) + ) + schema = self._unpack_epb(options) + + self.assertLessEqual(self._payload_octets(schema), len(options)) + self.assertLessEqual(self._payload_octets(schema), schema.length) + + def test_the_amplification_does_not_grow_with_the_block_count(self) -> None: + """A bound, not an example: the ratio is flat in the number of blocks. + + #594's crafted capture is 2,000 Enhanced Packet Blocks in 80,048 + octets, each with one option declaring 65,535 against none present, and + it synthesised 131,070,000 octets -- 1,637x, linear in the block count + and so unbounded in the input size. Unpacking the same block repeatedly + is that capture's option path without the file: the total has to stay + inside the octets the blocks themselves declare, at every count. + + """ + options = over_declared(UNKNOWN_OPT, b'', 0xFFFF) + raw = block_body(epb_body(options)) + + ratios = set() + for blocks in (1, 8, 64, 512): + with self.subTest(blocks=blocks): + total = 0 + for _ in range(blocks): + total += self._payload_octets(self._unpack_epb(options)) + + declared = blocks * (len(raw) + 4) + self.assertLessEqual(total, declared) + ratios.add(total / declared) + + self.assertEqual(len(ratios), 1, f'amplification varied with block count: {ratios}') + + def test_the_same_option_area_answers_the_same_whatever_preceded_it(self) -> None: + """No history dependence, which a running budget alone would not give. + + The bound reads only the block's own declared framing, so byte-identical + input produces byte-identical output regardless of what was parsed + before it. C.f. #593, whose own notes record a naive running threshold + answering four different ways across 40 byte-identical calls. + + """ + options = over_declared(UNKNOWN_OPT, b'', 0xFFFF) + + first = self._unpack_epb(options) + expected = (first.options[0].length, first.options[0].data) + + for index in range(200): + schema = self._unpack_epb(options) + self.assertEqual((schema.options[0].length, schema.options[0].data), expected, + f'call {index + 2} answered differently from call 1') + + def test_every_clamped_payload_is_bounded_and_none_was_missed(self) -> None: + """One subtest per option and record payload the bound is applied to. + + Each is handed an area holding its own header and fixed fields and + nothing more, while declaring 65,535 octets of payload. Every one must + read an empty payload rather than synthesise the shortfall. The table is + the list of sites, so a payload added later without the bound shows up + here as a subtest that was never written -- and one that lost the bound + shows up as a subtest that fails. + + """ + from pcapkit.protocols.schema.misc import pcapng as schema_module + + # (schema class, octets of fixed field between the header and the + # payload, payload attribute) + cases = [ + ('UnknownOption', b'', 'data'), + ('CommentOption', b'', 'comment'), + ('CustomOption', struct.pack(' None: + """A clamped option is reported; an option that fits is not.""" + from pcapkit.protocols.schema.misc.pcapng import UnknownOption + from pcapkit.utilities.warnings import SchemaWarning + + with mock.patch('pcapkit.protocols.schema.misc.pcapng.warn') as warn: + raw = struct.pack(' None: + """An option still packs when no length is known. + + :meth:`Schema.pack` leaves ``__length__`` at ``-1`` when no length is + known. This passes with or without the bound's negative-remainder guard, + and the cross-review of #676 is what established that: both + :class:`~pcapkit.corekit.fields.strings.BytesField` and + :class:`~pcapkit.corekit.fields.strings.StringField` repair a negative + width in ``pre_process``, resetting it to ``len(value)``, so the pack + path cannot see the guard at all. It is kept as a no-regression check + rather than as a discriminator; + :meth:`test_a_negative_remainder_is_left_alone_rather_than_clamped_to` is + the input that actually exercises the guard. + + """ + from pcapkit.protocols.schema.misc.pcapng import UnknownOption + + value = b'\xaa' * 8 + option = UnknownOption(type=UNKNOWN_OPT, length=len(value), data=value) + + packed = option.pack() + + self.assertEqual(packed, struct.pack(' None: + """A remainder already past zero is the field's problem, not the bound's. + + ``__length__`` can be negative by the time a payload sizes itself: + :meth:`Schema.unpack` warns and carries on when an earlier field has + over-consumed. A ten-octet area does it -- the first option takes eight, + leaving two, which is not a multiple of four, so the loop runs again with + two octets of area; the next option's type field takes both and its + length field short-reads to zero, putting ``__length__`` at -2 before the + payload's own callback is consulted. + + Clamping to that would hand ``-2`` to the field, whose struct template is + then ``'-2s'`` and whose unpack raises ``struct.error: bad char in struct + format`` -- not one of :mod:`pcapkit.utilities.exceptions`, and a new + failure on input that parses today. Measured: this input parses with the + guard and raises without it, while the eight-octet control parses either + way, so it isolates the guard rather than the clamp. + + """ + options = (tlv(UNKNOWN_OPT, b'\xaa\xbb') + + struct.pack(' None: + """The area is clamped too, so the payload cannot outrun the real block. + + Block Total Length is checked only against its own trailing copy, never + against the file, so a block may declare 1,000,000 octets while holding + 36. Its option area is then sized at 999,964, an option declaring 65,535 + is comfortably under that, and the payload bound alone does not fire: + before the area was clamped this synthesised 65,535 octets of zeros from + a 36-octet buffer -- 1,820x, with no warning at all. Found by the + cross-review of #676, which is why the area carries its own bound. + + The block still parses, and reports the length it declared; what it + cannot do is manufacture a payload the buffer never held. + + """ + options = over_declared(UNKNOWN_OPT, b'', 0xFFFF) + schema = self._unpack_epb_declaring(options, 1_000_000) + + self.assertEqual(schema.length, 1_000_000) + self.assertEqual(schema.options[0].length, 0xFFFF) + self.assertEqual(self._payload_octets(schema), 0) + + def test_the_area_bound_does_not_touch_a_well_formed_block(self) -> None: + """Clamping the area is a no-op whenever the declared length is honest. + + At the option field the only field still to come is the trailing Block + Total Length, so the octets left of the block are exactly the area plus + four and the minimum of the two is the area. Asserted over a range of + ``captured_len`` values, since the area expression subtracts + ``captured_len`` and its padding and the equality has to survive every + alignment. + + """ + for captured in range(1, 17): + with self.subTest(captured_len=captured): + value = bytes(range(captured)) + options = tlv(UNKNOWN_OPT, b'\xaa\xbb\xcc\xdd') + + schema = self._unpack_epb(options, packet_data=value) + + self.assertEqual(len(schema.options), 1) + self.assertEqual(schema.options[0].data, b'\xaa\xbb\xcc\xdd') + self.assertEqual(schema.captured_len, captured) + + if __name__ == '__main__': unittest.main()