fix(corekit): signed NumberField decode, zero-length PayloadField, PCAP trailing bytes and empty Raw

pcapkit/corekit/fields/numbers.py

@@ -306,12 +306,23 @@ def post_process(self, value: 'int | bytes', packet: 'dict[str, Any]') -> 'int':
         Returns:
             Processed field value.
 
+        Notes:
+            Masking against :attr:`self._bit_mask <NumberField.bit_length>`
+            yields the unsigned pattern, so a signed field maps it back into
+            its signed range afterwards, mirroring :meth:`pre_process`. A
+            resolved signed field therefore decodes ``ffffffff`` as ``-1``
+            rather than ``4294967295``. See GitHub issue :issue:`1274`.
+
         """
         if not self._need_process:
-            return cast('int', value) & self._bit_mask
-        return int.from_bytes(
-            cast('bytes', value), self._byteorder, signed=self._signed
-        ) & self._bit_mask
+            value = cast('int', value) & self._bit_mask
+        else:
+            value = int.from_bytes(
+                cast('bytes', value), self._byteorder, signed=self._signed
+            ) & self._bit_mask
+        if self._signed and value > self._bit_mask >> 1:
+            value -= self._bit_mask + 1
+        return value
 
 
 class Int32Field(NumberField):

pcapkit/protocols/misc/pcap/frame.py

@@ -37,7 +37,7 @@
 from pcapkit.protocols.protocol import ProtocolBase
 from pcapkit.protocols.schema.misc.pcap.frame import Frame as Schema_Frame
 from pcapkit.utilities.compat import localcontext
-from pcapkit.utilities.exceptions import RegistryError, UnsupportedCall, stacklevel
+from pcapkit.utilities.exceptions import RegistryError, StreamEOFError, UnsupportedCall, stacklevel
 from pcapkit.utilities.warnings import ProtocolWarning, RegistryWarning, warn
 
 if TYPE_CHECKING:
@@ -213,6 +213,11 @@ def unpack(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_Frame
         """
         if cast('Optional[Schema_Frame]', self.__header__) is None:
             packet = kwargs.get('__packet__', {})  # packet data
+
+            # NOTE: Reached on the parsing path only; the construction path
+            # sets ``__header__`` before unpacking its own buffer.
+            self._check_record_floor(length)
+
             packet['byteorder'] = self._ghdr.magic_number.byteorder
             self.__header__ = cast('Schema_Frame', self.__schema__.unpack(self._file, length, packet))  # type: ignore[call-arg,misc]
         return self.read(length, **kwargs)
@@ -429,6 +434,35 @@ def __length_hint__(self) -> 'Literal[16]':
         """Return an estimated length for the object."""
         return 16
 
+    def _check_record_floor(self, length: 'Optional[int]') -> 'None':
+        """Refuse a tail too short to hold a record header.
+
+        Args:
+            length: Length of packet data, as the caller declared it, or
+                :obj:`None` to measure what is left in
+                :attr:`self._file <pcapkit.protocols.protocol.Protocol._file>`.
+
+        Raises:
+            StreamEOFError: If fewer than sixteen octets are left to read.
+
+        A PCAP record header is sixteen octets, which is what
+        :meth:`__length_hint__` reports. A shorter tail is not a record, so
+        parsing one would only invent fields from zero padding. Reporting the
+        end of the stream instead lets :meth:`Extractor.record_frames
+        <pcapkit.foundation.extraction.Extractor.record_frames>` stop with the
+        frames read so far, as the PCAP-NG reader does for a truncated block
+        (:issue:`678`). See GitHub issue :issue:`1265`.
+
+        """
+        if length is None:
+            current = self._file.tell()
+            length = self._file.seek(0, io.SEEK_END) - current
+            self._file.seek(current, io.SEEK_SET)
+
+        if length < 16:
+            raise StreamEOFError(f'PCAP: record truncated: {length} octet(s) left, '
+                                 'fewer than the 16 a record header needs', quiet=True)
+
     # NOTE: This is a hack to make the ``__index__`` method work both as a
     # class method and an instance method.
     def __index__(self: 'Optional[Frame]' = None) -> 'int':  # type: ignore[override]

pcapkit/protocols/misc/raw.py

@@ -137,6 +137,14 @@ def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None,
         self._data = _data
         #: io.BytesIO: Source packet stream.
         self._file = io.BytesIO(self._data)
+
+        # NOTE: The buffer above is all there is to parse, so its size is
+        # declared rather than left for :func:`~pcapkit.utilities.decorators.prepare`
+        # to derive: a derived zero is read as the end of the stream, whereas
+        # an empty raw payload is a payload all the same (:issue:`1285`).
+        if length is None:
+            length = len(self._data)
+
         #: pcapkit.protocols.data.misc.raw.Raw: Parsed packet data.
         self._info = self.unpack(length, **kwargs)
 

pcapkit/protocols/schema/schema.py

@@ -920,10 +920,15 @@ def unpack(cls, data: 'bytes | IO[bytes]',
             # NOTE: ``Field.length`` recomputes struct.calcsize() on every read,
             # so it is read once per field here rather than at each use.
             if isinstance(field, PayloadField):
+                # NOTE: A negative length is the unset default and takes what
+                # the declared length leaves, never less than nothing; a
+                # computed zero is an empty payload, not an unset one, or it
+                # would swallow the fields after it (:issue:`1275`).
                 length = field.length
-                payload_length = length or cast('int', packet['__length__'])
+                if length < 0:
+                    length = max(cast('int', packet['__length__']), 0)
 
-                payload = data.read(payload_length)
+                payload = data.read(length)
                 self.__buffer__[field.name] = payload
 
                 packet['__length__'] -= length

tests/corekit/test_fields_numbers_signed_decode_1274_unit.py

@@ -0,0 +1,56 @@
+# -*- coding: utf-8 -*-
+"""A resolved signed number field decodes into its signed range.
+
+GitHub issue #1274: :meth:`NumberField.post_process
+<pcapkit.corekit.fields.numbers.NumberField.post_process>` masked the
+decoded value with ``_bit_mask`` and returned it, so once ``__call__`` had set
+that mask -- as it does for every field inside a :class:`Schema` -- a signed
+field came back as its unsigned pattern: ``ffffffff`` read as ``4294967295``,
+and the PCAP ``thiszone`` of ``-3600`` as ``4294963696``.
+
+Every case builds its own octets in memory and reads no capture. Classes are
+imported inside each test, after :func:`~tests._support.reimport_once_per_class`.
+
+"""
+
+import io
+import unittest
+
+from tests._support import reimport_once_per_class
+
+
+class TestSignedNumberFieldDecode(unittest.TestCase):
+    """Pin the decoded value and the rebuilt octets of signed fields."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+
+    def test_schema_int32_decodes_negative(self) -> None:
+        from pcapkit.corekit.fields.numbers import Int32Field, UInt8Field
+        from pcapkit.protocols.schema.schema import Schema
+
+        class T(Schema):
+            z: 'int' = Int32Field()
+            u: 'int' = UInt8Field()
+
+        # big-endian, the field's default byte order
+        for raw, value in ((b'\xff\xff\xff\xff', -1), (b'\xff\xff\xf1\xf0', -3600),
+                           (b'\x80\x00\x00\x00', -2**31), (b'\x7f\xff\xff\xff', 2**31 - 1)):
+            with self.subTest(raw=raw.hex()):
+                data = raw + b'\xff'
+                schema = T.unpack(io.BytesIO(data), 5, {'__length__': 5})
+                self.assertEqual(schema.z, value)
+                self.assertEqual(schema.u, 255)
+                self.assertEqual(T(z=schema.z, u=schema.u).pack({}), data)
+
+    def test_pcap_thiszone_round_trips(self) -> None:
+        from pcapkit.protocols.misc.pcap.header import Header
+
+        data = bytes.fromhex('d4c3b2a1' '0200' '0400' 'f0f1ffff' '00000000' 'ffff0000' '01000000')
+        header = Header(data)
+        self.assertEqual(header.info.thiszone, -3600)
+        self.assertEqual(Header.from_data(header.info).data, data)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/protocols/misc/pcap/test_frame_trailing_bytes_1265_unit.py

@@ -0,0 +1,75 @@
+# -*- coding: utf-8 -*-
+"""A PCAP file ending in a partial record header stops cleanly at EOF.
+
+GitHub issue #1265: with fewer than sixteen octets left after the last record,
+:meth:`Frame.unpack <pcapkit.protocols.misc.pcap.frame.Frame.unpack>` parsed a
+record header out of zero padding, and the payload read got a negative length
+(``ValueError: read length must be non-negative or -1``). A tail that short now
+raises :exc:`~pcapkit.utilities.exceptions.StreamEOFError`, which the extractor
+treats as the end of the file, as it does for a truncated PCAP-NG block (#678).
+
+Every case builds its own octets and reads no capture. Classes are imported
+inside each test, after :func:`~tests._support.reimport_once_per_class`.
+
+"""
+
+import io
+import os
+import tempfile
+import unittest
+import warnings
+
+from tests._support import reimport_once_per_class
+
+#: Little-endian global header: v2.4, Ethernet.
+GLOBAL = bytes.fromhex('d4c3b2a1' '0200' '0400' '00000000' '00000000' 'ffff0000' '01000000')
+#: One record: ts 1.000002, incl_len = orig_len = 4, four payload octets.
+RECORD = bytes.fromhex('01000000' '02000000' '04000000' '04000000') + b'\xde\xad\xbe\xef'
+TRAIL = b'\x00\x01\x02'
+
+
+class TestFrameTrailingBytes(unittest.TestCase):
+    """Pin the end-of-file signal for a tail shorter than a record header."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+
+    def test_short_tail_raises_stream_eof(self) -> None:
+        from pcapkit.protocols.misc.pcap.header import Header
+        from pcapkit.protocols.misc.pcap.frame import Frame
+        from pcapkit.utilities.exceptions import StreamEOFError
+
+        header = Header(GLOBAL)
+        for size in range(16):
+            with self.subTest(size=size):
+                with self.assertRaises(StreamEOFError):
+                    Frame(io.BytesIO(RECORD[:size]), num=1, header=header.info)
+
+    def test_extract_keeps_frames_before_short_tail(self) -> None:
+        import pcapkit
+
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, 'trail.pcap')
+            with open(path, 'wb') as file:
+                file.write(GLOBAL + RECORD + TRAIL)
+
+            with warnings.catch_warnings():
+                warnings.simplefilter('ignore')
+                ext = pcapkit.extract(fin=path, nofile=True, store=True, auto=False,
+                                      engine='default')
+                frames = []
+                # bounded, so that a reader which never reaches EOF fails
+                # rather than spinning
+                for _ in range(5):
+                    try:
+                        frames.append(next(ext))
+                    except StopIteration:
+                        break
+                ext.close() if hasattr(ext, 'close') else None
+
+        self.assertEqual(len(frames), 1)
+        self.assertEqual(frames[0].info.packet, b'\xde\xad\xbe\xef')
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/protocols/misc/test_raw_empty_bytes_1285_unit.py

@@ -0,0 +1,46 @@
+# -*- coding: utf-8 -*-
+"""``Raw`` accepts an empty byte string on the parsing path too.
+
+GitHub issue #1285: ``Raw(b"")`` left the length for
+:func:`~pcapkit.utilities.decorators.prepare` to derive, and a derived zero
+is the end-of-stream signal, so it raised
+:exc:`~pcapkit.utilities.exceptions.StreamEOFError` while ``Raw(packet=b"")``
+worked.
+
+Classes are imported inside each test, after
+:func:`~tests._support.reimport_once_per_class`.
+
+"""
+
+import io
+import unittest
+
+from tests._support import reimport_once_per_class
+
+
+class TestRawEmptyBytes(unittest.TestCase):
+    """Pin that both construction forms accept an empty payload."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+
+    def test_empty_bytes_parse(self) -> None:
+        from pcapkit.protocols.misc.raw import Raw
+
+        for source in (b'', io.BytesIO(b'')):
+            with self.subTest(source=type(source).__name__):
+                raw = Raw(source)
+                self.assertEqual(raw.data, b'')
+                self.assertIsNone(raw.info.protocol)
+                self.assertEqual(Raw.from_data(raw.info).data, b'')
+
+    def test_both_forms_agree(self) -> None:
+        from pcapkit.protocols.misc.raw import Raw
+
+        for payload in (b'', b'abc'):
+            with self.subTest(payload=payload):
+                self.assertEqual(Raw(payload).data, Raw(packet=payload).data)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/protocols/test_schema_payload_zero_length_1275_unit.py

@@ -0,0 +1,91 @@
+# -*- coding: utf-8 -*-
+"""A computed ``PayloadField`` length of zero reads an empty payload.
+
+GitHub issue #1275: :meth:`Schema.unpack
+<pcapkit.protocols.schema.schema.Schema.unpack>` sized the payload as
+``length or packet['__length__']``, so a length that evaluated to zero was
+taken as "unset" and the payload swallowed every octet after it, leaving the
+following fields to be zero-filled.
+
+Every case builds its own octets in memory and reads no capture. Classes are
+imported inside each test, after :func:`~tests._support.reimport_once_per_class`.
+
+"""
+
+import io
+import unittest
+
+from tests._support import reimport_once_per_class
+
+
+class TestPayloadFieldZeroLength(unittest.TestCase):
+    """Pin the split between a zero-length payload and the fields after it."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+
+    def _schema(self) -> 'type':
+        from pcapkit.corekit.fields.misc import PayloadField
+        from pcapkit.corekit.fields.numbers import UInt8Field
+        from pcapkit.corekit.fields.strings import BytesField
+        from pcapkit.protocols.schema.schema import Schema
+
+        class S(Schema):
+            n: 'int' = UInt8Field()
+            body: 'bytes' = PayloadField(length=lambda p: p['n'])
+            tail: 'bytes' = BytesField(length=2)
+
+        return S
+
+    def test_zero_length_payload_is_empty(self) -> None:
+        S = self._schema()
+        for data, body in ((b'\x00ab', b''), (b'\x01xab', b'x')):
+            with self.subTest(data=data):
+                schema = S.unpack(io.BytesIO(data), len(data), {'__length__': len(data)})
+                self.assertEqual(schema.body, body)
+                self.assertEqual(schema.tail, b'ab')
+                self.assertEqual(bytes(schema), data)
+
+    def _unset_schema(self) -> 'type':
+        from pcapkit.corekit.fields.misc import PayloadField
+        from pcapkit.corekit.fields.numbers import UInt8Field
+        from pcapkit.protocols.schema.schema import Schema
+
+        class R(Schema):
+            n: 'int' = UInt8Field()
+            body: 'bytes' = PayloadField()
+
+        return R
+
+    def test_unset_length_takes_the_rest(self) -> None:
+        R = self._unset_schema()
+
+        data = b'\x07rest'
+        schema = R.unpack(io.BytesIO(data), len(data), {'__length__': len(data)})
+        self.assertEqual(schema.body, b'rest')
+        self.assertEqual(bytes(schema), data)
+
+    def test_unset_length_stops_at_the_declared_length(self) -> None:
+        """An unset payload is bounded by ``__length__``, never by the stream.
+
+        The declared length is the authority even when the stream holds more:
+        the octets past it belong to whatever encloses this schema, so reading
+        them is an over-read that also leaves ``__length__`` mis-decremented --
+        it used to come back *larger* than before the payload was read.
+
+        """
+        R = self._unset_schema()
+
+        data = b'\x07abcdefgh'
+        for declared, body, left in ((3, b'ab', 0), (1, b'', 0), (0, b'', -1)):
+            with self.subTest(declared=declared):
+                stream = io.BytesIO(data)
+                packet = {'__length__': declared}
+                schema = R.unpack(stream, declared, packet)
+                self.assertEqual(schema.body, body)
+                self.assertEqual(stream.tell(), 1 + len(body))
+                self.assertEqual(packet['__length__'], left)
+
+
+if __name__ == '__main__':
+    unittest.main()