fix(protocols)!: a declared length running past the data is accepted and zero-filled on rebuild

pcapkit/corekit/fields/collections.py

@@ -9,7 +9,7 @@
 from pcapkit.corekit.fields.numbers import NumberField
 from pcapkit.corekit.multidict import OrderedMultiDict
 from pcapkit.utilities.compat import List
-from pcapkit.utilities.exceptions import FieldValueError
+from pcapkit.utilities.exceptions import FieldValueError, ProtocolError
 
 __all__ = [
     'ListField', 'OptionField',
@@ -29,6 +29,44 @@
 _TS = TypeVar('_TS', bound='Schema')
 
 
+class _ShortReadTracker:
+    """Stream proxy recording whether a read came back short.
+
+    Args:
+        file: Stream to read from.
+
+    Only the ``read``, ``seek`` and ``tell`` calls that
+    :meth:`Schema.unpack <pcapkit.protocols.schema.schema.Schema.unpack>` and
+    :meth:`OptionField.unpack` make are delegated.
+
+    """
+
+    __slots__ = ('_file', 'short')
+
+    def __init__(self, file: 'IO[bytes]') -> 'None':
+        self._file = file
+        #: bool: Whether a read since the last reset returned fewer octets than
+        #: it asked for.
+        self.short = False
+
+    def read(self, size: 'Optional[int]' = -1) -> 'bytes':
+        """Read ``size`` octets, noting a short read."""
+        if size is None:
+            size = -1
+        data = self._file.read(size)
+        if 0 <= size and len(data) < size:
+            self.short = True
+        return data
+
+    def seek(self, offset: 'int', whence: 'int' = io.SEEK_SET) -> 'int':
+        """Seek the underlying stream."""
+        return self._file.seek(offset, whence)
+
+    def tell(self) -> 'int':
+        """Tell the underlying stream's position."""
+        return self._file.tell()
+
+
 class ListField(FieldBase[List[_TL]], Generic[_TL]):
     """Field list for protocol fields.
 
@@ -362,6 +400,8 @@ def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> 'list
         Raises:
             FieldValueError: If an option consumes nothing from ``buffer``, since
                 the loop below has then no way to get past it.
+            ProtocolError: If an option that begins with octets present reads
+                past the end of the data.
 
         """
         length = self._length
@@ -421,8 +461,24 @@ def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> 'list
         # number of octets to rewind to get back to the front of the option.
         type_field = self._type_field
 
+        # NOTE: An option whose declared length runs past the octets behind it
+        # reads short, and every field reading off the end pads the missing
+        # octets with zeros, so the option would rebuild octets that were never
+        # there: a HIP ``ACK`` declaring 256 octets with 4 present, or an MH
+        # ``PadN`` declaring 3 with none. Nothing downstream can tell, because
+        # the schema records the zeros as values. C.f. #1325.
+        #
+        # So the stream is read through a proxy that notes any short read, and
+        # an option that began with at least one octet present and then read
+        # short is rejected. An option that begins at the end of the data reads
+        # nothing at all and is left to the end-of-option-list and progress
+        # handling below, which is what an option area declared longer than its
+        # octets relies on.
+        tracker = file = _ShortReadTracker(file)  # type: ignore[assignment]
+
         temp = []  # type: list[_TS]
         while length > 0:
+            tracker.short = False
             if type_field is None:
                 # unpack option type using base schema
                 meta = self._base_schema.unpack(file, length, packet)  # type: ignore[call-arg,misc,var-annotated]
@@ -443,6 +499,13 @@ def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> 'list
 
             # unpack option using option schema
             data = schema.unpack(file, length, packet)  # type: ignore[call-arg,misc,var-annotated]
+            if tracker.short and consumed > 0:
+                raise ProtocolError(
+                    f'Field {self.name} has an option that runs past the end of '
+                    f'the data: {code!r} at offset {offset - start} of '
+                    f'{self._length}, with only {file.tell() - offset} octet(s) '
+                    f'of it present'
+                )
             new_packet[self.name].add(code, data)
             temp.append(data)
 

pcapkit/protocols/internet/ipv6_ext.py

@@ -121,7 +121,6 @@
 The house convention for a declared length that does not fit what remains
 is warn-and-clip: emit a warning naming what was declared against what is
 left, then read only what is left. See
-:func:`pcapkit.protocols.schema.misc.pcapng.bounded_option`,
 :func:`pcapkit.protocols.schema.misc.pcapng.bounded_area` and
 :meth:`pcapkit.protocols.misc.pcapng.PCAPNG.read`.
 

pcapkit/protocols/misc/pcapng.py

@@ -1143,6 +1143,17 @@ def read(self, length: 'Optional[int]' = None, *, _read: 'bool' = True,
             meth = name[0]
         block = meth(schema.block, header=schema)
 
+        # NOTE: An option or record area holding a malformed entry is kept as
+        # the octets captured, as one :obj:`bytes` item (see
+        # :class:`~pcapkit.protocols.schema.misc.pcapng.OptionAreaField`), and
+        # parses as no options. The octets are kept on the block as
+        # ``options_raw`` or ``records_raw``, which the block's ``make`` writes
+        # back verbatim (:issue:`1325`).
+        for name in ('records', 'options'):
+            area = getattr(schema.block, name, None)
+            if isinstance(area, list) and len(area) == 1 and isinstance(area[0], bytes):
+                block.__update__([(f'{name}_raw', area[0])])
+
         if not _read:
             # move backward to the beginning of the packet
             self._file.seek(0, io.SEEK_SET)
@@ -1626,6 +1637,21 @@ def _make_data(cls, data: 'Data_PCAPNG') -> 'dict[str, Any]':  # type: ignore[ov
             'block': data,
         }
 
+    @staticmethod
+    def _block_area(block: 'Data_PCAPNG', name: 'str') -> 'Any':
+        """Option or record area of a parsed block, for its ``make``.
+
+        Args:
+            block: Block data model.
+            name: ``'options'`` or ``'records'``.
+
+        Returns:
+            The area's octets if :meth:`read` kept it as captured, as
+            ``<name>_raw``; otherwise the parsed ``<name>``.
+
+        """
+        return getattr(block, f'{name}_raw', getattr(block, name))
+
     @staticmethod
     def _flags_enum(enum: 'Type[PacketDirection] | Type[PacketReception]', value: 'int',
                     option: 'Enum_OptionType') -> 'Any':
@@ -2383,6 +2409,10 @@ def _read_pcapng_options(self, options_schema: 'list[Schema_Option]') -> 'Option
         """
         options = OrderedMultiDict()  # type: Option
 
+        # an area kept as captured holds no options to read
+        if any(isinstance(schema, bytes) for schema in options_schema):
+            return options
+
         for schema in options_schema:
             type = schema.type
             name = self._lookup_registry(self.__option__, _option_key(type))
@@ -3478,6 +3508,10 @@ def _read_nrb_records(self, records_schema: 'list[Schema_NameResolutionRecord]')
         """
         records = OrderedMultiDict()  # type: Record
 
+        # an area kept as captured holds no records to read
+        if any(isinstance(schema, bytes) for schema in records_schema):
+            return records
+
         for schema in records_schema:
             type = schema.type
             name = self._lookup_registry(self.__record__, type)
@@ -3789,7 +3823,7 @@ def _make_block_shb(self, block: 'Optional[Data_SectionHeaderBlock]' = None, *,
             major_version = block.version.major
             minor_version = block.version.minor
             section_length = block.section_length
-            options = block.options
+            options = self._block_area(block, 'options')
         else:
             if major_version is None:
                 major_version = version[0]
@@ -3838,7 +3872,7 @@ def _make_block_idb(self, block: 'Optional[Data_InterfaceDescriptionBlock]' = No
         if block is not None:
             linktype_val = block.linktype
             snaplen = block.snaplen
-            options = block.options
+            options = self._block_area(block, 'options')
         else:
             linktype_val = self._make_index(linktype, linktype_default, namespace=linktype_namespace,  # type: ignore[assignment]
                                             reversed=linktype_reversed, pack=False)
@@ -3886,7 +3920,7 @@ def _make_block_epb(self, block: 'Optional[Data_EnhancedPacketBlock]' = None, *,
             captured_len = block.captured_len
             original_len = block.original_len
             packet_data = self._make_packet_data(block)
-            options = block.options
+            options = self._block_area(block, 'options')
 
         timestamp_high, timestamp_low = self._make_timestamp(timestamp, interface_id=interface_id)
         if captured_len is None:
@@ -3980,8 +4014,8 @@ def _make_block_nrb(self, block: 'Optional[Data_NameResolutionBlock]' = None, *,
 
         """
         if block is not None:
-            records = block.records
-            options = block.options
+            records = self._block_area(block, 'records')
+            options = self._block_area(block, 'options')
 
         if records is not None:
             records_value, records_length = self._make_nrb_records(records)
@@ -4021,7 +4055,7 @@ def _make_block_isb(self, block: 'Optional[Data_InterfaceStatisticsBlock]' = Non
         if block is not None:
             interface_id = block.interface_id
             timestamp = block.timestamp_epoch
-            options = block.options
+            options = self._block_area(block, 'options')
 
         timestamp_high, timestamp_low = self._make_timestamp(timestamp, interface_id=interface_id)
         # NOTE: ``isb_starttime`` and ``isb_endtime`` are scaled by the
@@ -4120,7 +4154,7 @@ def _make_block_dsb(self, block: 'Optional[Data_DecryptionSecretsBlock]' = None,
         if block is not None:
             secrets_type_val = block.secrets_type
             secrets_data = block.secrets_data
-            options = block.options
+            options = self._block_area(block, 'options')
         else:
             secrets_type_val = self._make_index(secrets_type, secrets_type_default, namespace=secrets_type_namespace,  # type: ignore[assignment]
                                                 reversed=secrets_type_reversed, pack=False)
@@ -4250,7 +4284,7 @@ def _make_block_packet(self, block: 'Optional[Data_PacketBlock]' = None, *,
             captured_len = block.captured_len
             original_len = block.original_len
             packet_data = self._make_packet_data(block)
-            options = block.options
+            options = self._block_area(block, 'options')
 
         timestamp_high, timestamp_low = self._make_timestamp(timestamp, interface_id=interface_id)
         if captured_len is None:
@@ -4287,7 +4321,7 @@ def _make_block_packet(self, block: 'Optional[Data_PacketBlock]' = None, *,
             length2=total_length + 32 + packet_len,
         )
 
-    def _make_pcapng_options(self, options: 'Option | list[Schema_Option | tuple[Enum_OptionType, dict[str, Any]] | bytes]',
+    def _make_pcapng_options(self, options: 'Option | list[Schema_Option | tuple[Enum_OptionType, dict[str, Any]] | bytes] | bytes',
                              namespace: 'str') -> 'tuple[list[Schema_Option | bytes], int]':
         """Make options for PCAP-NG.
 
@@ -4299,6 +4333,10 @@ def _make_pcapng_options(self, options: 'Option | list[Schema_Option | tuple[Enu
             Tuple of options and total length of options.
 
         """
+        # an option area kept as captured is written back verbatim
+        if isinstance(options, bytes):
+            return [options], len(options)
+
         has_endofopt = False
         total_length = 0
         if isinstance(options, list):
@@ -5725,7 +5763,7 @@ def _make_option_pack_hash(self, type: 'Enum_OptionType', option: 'Optional[Data
             data=hash,
         )
 
-    def _make_nrb_records(self, records: 'Record | list[Schema_NameResolutionRecord | tuple[Enum_RecordType, dict[str, Any]] | bytes]') -> 'tuple[list[Schema_NameResolutionRecord | bytes], int]':
+    def _make_nrb_records(self, records: 'Record | list[Schema_NameResolutionRecord | tuple[Enum_RecordType, dict[str, Any]] | bytes] | bytes') -> 'tuple[list[Schema_NameResolutionRecord | bytes], int]':
         """Make name resolution records for PCAP-NG.
 
         Args:
@@ -5735,6 +5773,10 @@ def _make_nrb_records(self, records: 'Record | list[Schema_NameResolutionRecord
             Tuple of name resolution records and total length of the records.
 
         """
+        # a record area kept as captured is written back verbatim
+        if isinstance(records, bytes):
+            return [records], len(records)
+
         has_record_end = False
         total_length = 0
         if isinstance(records, list):