diff --git a/pcapkit/corekit/io.py b/pcapkit/corekit/io.py index 5de8bdd36b..f77db6b37f 100644 --- a/pcapkit/corekit/io.py +++ b/pcapkit/corekit/io.py @@ -28,6 +28,10 @@ __all__ = ['SeekableReader', 'PeekableStream', 'NamedStream'] +#: Most octets :class:`SeekableReader` asks of its stream in one read. A read +#: allocates what it asks for up front, so a record's bogus length is read in pieces. +_READ_CHUNK = 1 << 20 + class SeekableReader(io.BufferedReader): """Seekable buffered reader. @@ -74,6 +78,9 @@ class SeekableReader(io.BufferedReader): _buffer_set: 'int' #: Buffer current position. _buffer_cur: 'int' + #: Octets :meth:`seek` reads past the position to find the end of the stream; + #: the buffer size by default. + _read_ahead_size: 'int' #: Path to save buffer. _buffer_path: 'str' @@ -109,6 +116,7 @@ def __init__(self, raw: 'IO[bytes]', buffer_size: 'int' = io.DEFAULT_BUFFER_SIZE self._buffer_view = self._buffer.getbuffer() self._buffer_size = buffer_size + self._read_ahead_size = buffer_size if buffer_save: if buffer_path is None: @@ -150,6 +158,82 @@ def _write_buffer(self, buf: 'bytes', /) -> 'None': else: self._buffer_view[old_ptr:self._buffer_cur] = buf + def _fill(self, size: 'int', minimum: 'Optional[int]' = None, /, *, + keep: 'Optional[int]' = None) -> 'bytes': + """Read up to ``size`` octets from the stream into the buffer, and return them. + + Args: + size: Octets to ask for; negative reads to the end of the stream. + minimum: Octets to keep reading for; ``size`` if omitted. + keep: Position the buffer must still hold afterwards; it grows rather than + slide past it. + + Returns: + The octets read, fewer than ``minimum`` only once the stream has ended. + + Note: + One read may return fewer octets than asked for at any point -- a pipe hands + over what has been written so far, a raw stream what one system call returned + -- and only an empty one means the stream has ended. Taking the first answer + for all there is ended a capture at a chunk boundary (:issue:`1586`), so this + reads until ``minimum`` octets are in hand. Past that it does not wait: + ``read1``, where the stream has it, returns what is already there. A + non-blocking stream's :obj:`None` stops the reading. No one read asks for + more than ``_READ_CHUNK`` octets, so a bogus length read from a record allocates + only what the stream actually holds. + + """ + if size < 0: + buf = self._stream.read(size) or b'' + self._write_buffer(buf) + return buf + + minimum = size if minimum is None else minimum + read = getattr(self._stream, 'read1', self._stream.read) + + chunks = [] # type: list[bytes] + count = 0 + while True: + chunk = read(min(size - count, _READ_CHUNK)) + if not chunk: + break + chunks.append(chunk) + count += len(chunk) + if count >= minimum: + break + + buf = b''.join(chunks) + if keep is not None: + span = self._buffer_set + self._buffer_cur + len(buf) - max(keep, self._buffer_set) + if span > self._buffer_size: + self._truncate_buffer(span) + self._write_buffer(buf) + return buf + + def _read_ahead(self) -> 'int': + """Read the stream to :attr:`_read_ahead_size` octets past the position. + + Returns: + How far the stream has been read: that many octets past the position, or the + end of the stream if it ended first. :meth:`seek` takes this as the end. + + Note: + The buffer slides forward to make the room, so after this only the last + ``_buffer_size`` less ``_read_ahead_size`` octets before the position can still + be sought back to -- none, at the default -- unless ``buffer_save`` keeps them. + It grows instead of sliding past the position itself, so a record longer than + the buffer is held whole, as its reader seeks back to its first octet; a bogus + length grows it only as far as the stream goes. A live stream holds the reader + here until the octets arrive, which is why the engines set + :attr:`_read_ahead_size` to the record they are about to read. + + """ + end = self._buffer_set + self._buffer_cur + want = self._tell + self._read_ahead_size - end + if want > 0: + end += len(self._fill(want, keep=self._tell)) + return end + def _seek_buffer(self) -> 'int': """Point the buffer at the current stream position, and say how much it can serve. @@ -222,7 +306,10 @@ def flush(self) -> 'None': read-only and non-blocking streams.""" if self._buffer_file is not None: self._buffer_file.flush() - self._stream.flush() + # NOTE: a stream its owner has closed already has nothing to flush, and asking + # raises -- out of :meth:`close` too, which a finaliser can only print. + if not getattr(self._stream, 'closed', False): + self._stream.flush() def isatty(self) -> 'bool': """Return :data:`True` if the stream is interactive (i.e., connected to a @@ -344,22 +431,31 @@ def seek(self, offset: 'int', whence: 'int' = io.SEEK_SET, /) -> 'int': the window, it holds with a saved buffer as well, which is exempt from the window check and would otherwise return a negative position as if the seek had worked. - """ - # NOTE: we mark the end of buffer content to the end of buffer - # so that it may trigger the IO to read more data to fill in - # the content. - buf_end = self._buffer_set + self._buffer_size - #buf_end = self._buffer_set + self._buffer_cur + A non-seekable stream has no end until it is over, so :data:`~io.SEEK_END` is + as far as :meth:`_read_ahead` reads past the position: ``_read_ahead_size`` + octets, a buffer's worth by default, or to the real end if that comes first. + The frame readers measure what is left this way, and the engines set that size + to the next record, so a record is measured whole, however long. The end used to + be the buffer's far edge, which once the buffer has filled is just how far the + stream has been read, so the first record ending there was taken for the end of + the capture (:issue:`1586`). The read slides the buffer, so a position before + the one :data:`~io.SEEK_END` was sought from may then be refused, as + :meth:`_read_ahead` says. + """ if whence == io.SEEK_SET: target = offset elif whence == io.SEEK_CUR: target = self._tell + offset elif whence == io.SEEK_END: - target = buf_end + offset + target = self._read_ahead() + offset else: raise SeekError(f'invalid whence ({whence}, should be {io.SEEK_SET}, {io.SEEK_CUR} or {io.SEEK_END})') + # NOTE: the buffer's far edge, read after :meth:`_read_ahead` has slid it, is how far + # a seek may go unwarned. It is not the end of the stream. + buf_end = self._buffer_set + self._buffer_size + if target < 0: raise SeekError(f'negative seek value {target}') # NOTE: both refusals read the target rather than ``_tell``, which is what lets them @@ -370,9 +466,7 @@ def seek(self, offset: 'int', whence: 'int' = io.SEEK_SET, /) -> 'int': self._tell = target if self._tell >= self._buffer_set: - if self._tell > buf_end: - warn(f'seek beyond the end of the buffer: {self._tell} > {buf_end}', - SeekWarning, stacklevel=stacklevel()) + ended = False if self._tell > (tmp_end := self._buffer_set + self._buffer_cur): # NOTE: if we do need to seek beyond the existing contents, # then we'll do a quick read to make up the contents; the @@ -380,11 +474,23 @@ def seek(self, offset: 'int', whence: 'int' = io.SEEK_SET, /) -> 'int': # the size of the content to be read, whichever is larger. # However, the length to fill must not be larger than the # buffer size itself. - tmp_len = min(max(self._tell - tmp_end, self._buffer_size // 4), self._buffer_size) - self._tell = tmp_end - - tmp_buf = self.read1(tmp_len) - self._tell = tmp_end + len(tmp_buf) + # + # The gap is waited for and the rest is not, and the position is + # the target, or the end of the stream if that comes first -- not + # wherever an over-asking read happened to stop (:issue:`1586`). + tmp_gap = min(self._tell - tmp_end, self._buffer_size) + tmp_len = min(max(tmp_gap, self._buffer_size // 4), self._buffer_size) + + tmp_buf = self._fill(tmp_len, tmp_gap) + ended = len(tmp_buf) < tmp_gap + self._tell = min(self._tell, tmp_end + len(tmp_buf)) + # NOTE: past a stream that ended inside the buffer, this is the silent seek past + # the end of a file, which a frame reader makes over a record the capture cut + # short. A seek that went past the buffer's edge, or stopped short of a target + # beyond it, is still warned of. + if target > buf_end and not (ended and self._tell <= buf_end): + warn(f'seek beyond the end of the buffer: {target} > {buf_end}', + SeekWarning, stacklevel=stacklevel()) self._buffer.seek(self._tell - self._buffer_set, io.SEEK_SET) else: # NOTE: only a saved buffer reaches here -- the refusal above has already @@ -518,9 +624,11 @@ def read(self, size: 'int | None' = -1, /) -> 'bytes': if size is None or size < 0: size = -1 + # NOTE: both reads from the stream go through :meth:`_fill`, which keeps reading + # past a short answer, as :meth:`io.BufferedIOBase.read` is specified to: a short + # result here means the end of the stream, and the frame readers take it so. if self._tell >= self._buffer_set + self._buffer_cur: - buf = self._stream.read(size) - self._write_buffer(buf) + buf = self._fill(size) else: if self._buffer_file is not None and self._tell < self._buffer_set: with open(self._buffer_path, 'rb') as temp_file: @@ -535,9 +643,7 @@ def read(self, size: 'int | None' = -1, /) -> 'bytes': size_rem = -1 if size < 0 or (size_rem := size - len(buf)) > 0: - buf_tmp = self._stream.read(size_rem) - self._write_buffer(buf_tmp) - buf += buf_tmp + buf += self._fill(size_rem) self._tell += len(buf) return buf diff --git a/pcapkit/foundation/engines/pcap.py b/pcapkit/foundation/engines/pcap.py index 4ef29d9e81..67e8399731 100644 --- a/pcapkit/foundation/engines/pcap.py +++ b/pcapkit/foundation/engines/pcap.py @@ -8,8 +8,10 @@ support, as is used by :class:`pcapkit.foundation.extraction.Extractor`. """ +import io from typing import TYPE_CHECKING +from pcapkit.corekit.io import SeekableReader from pcapkit.foundation.engines.engine import EngineBase from pcapkit.protocols.misc.pcap.frame import Frame from pcapkit.protocols.misc.pcap.header import Header @@ -157,6 +159,7 @@ def read_frame(self) -> 'Frame': ext = self._extractor # read frame header + self._bound_read_ahead() frame = Frame(ext._ifile, num=ext._frnum+1, header=self._gbhdr.info, layer=ext._exlyr, protocol=ext._exptl, nanosecond=self._nnsec, __context__=ext._exctx) @@ -207,3 +210,34 @@ def read_frame(self) -> 'Frame': # return frame record return frame + + ########################################################################## + # Utilities. + ########################################################################## + + def _bound_read_ahead(self) -> 'None': + """Have a non-seekable input read ahead to the end of the next record, no further. + + A record is measured by reading ahead (see :meth:`SeekableReader.seek + `), and then read again from its first + octet. A read past it waits, on a live capture, for the records after: a + buffer's worth held each frame back until that much more had arrived. A read + short of it lets a record longer than the buffer slide out of it before it is + read again (:issue:`1586`). So the read covers the record header and the + ``incl_len`` octets it declares, in the global header's byte order -- as far as + the stream goes, for a bogus length, which is as far as a file read goes too. + + """ + # pylint: disable=protected-access + ifile = self._extractor._ifile + if not isinstance(ifile, SeekableReader): + return + + position = ifile.tell() + head = ifile.read(16) + ifile.seek(position, io.SEEK_SET) + + size = 16 + if len(head) == 16: + size += int.from_bytes(head[8:12], self._gbhdr.byteorder) + ifile._read_ahead_size = size diff --git a/pcapkit/foundation/engines/pcapng.py b/pcapkit/foundation/engines/pcapng.py index c9eb73847d..dc79a38ca6 100644 --- a/pcapkit/foundation/engines/pcapng.py +++ b/pcapkit/foundation/engines/pcapng.py @@ -9,10 +9,12 @@ """ import collections +import os from typing import TYPE_CHECKING, cast from pcapkit.const.pcapng.block_type import BlockType as Enum_BlockType from pcapkit.corekit.infoclass import Info, info_final +from pcapkit.corekit.io import SeekableReader from pcapkit.dumpkit.pcap import PCAPIO from pcapkit.foundation.engines.engine import EngineBase from pcapkit.protocols.misc.pcapng import PCAPNG as P_PCAPNG @@ -27,7 +29,7 @@ logger = get_logger(__name__) if TYPE_CHECKING: - from typing import Optional + from typing import Literal, Optional from pcapkit.foundation.extraction import Extractor from pcapkit.protocols.data.misc.pcapng import PCAPNG as Data_PCAPNG @@ -106,6 +108,27 @@ class PCAPNG(EngineBase[P_PCAPNG]): b'\x0a\x0d\x0d\x0a', ) + #: A section header block's Byte-Order Magic, mapped to the byte order it + #: declares for its section. + BYTE_ORDER_MAGIC = { + b'\x1a\x2b\x3c\x4d': 'big', + b'\x4d\x3c\x2b\x1a': 'little', + } # type: dict[bytes, Literal['big', 'little']] + + #: Block types mapped to the octets their fixed fields span, Block Type and + #: both copies of Block Total Length included; any other block spans 12. + BLOCK_MINIMUM = { + Enum_BlockType.Section_Header_Block: 28, + Enum_BlockType.Interface_Description_Block: 20, + Enum_BlockType.Packet_Block: 32, + Enum_BlockType.Simple_Packet_Block: 16, + Enum_BlockType.Interface_Statistics_Block: 24, + Enum_BlockType.Enhanced_Packet_Block: 32, + Enum_BlockType.Decryption_Secrets_Block: 20, + Enum_BlockType.Custom_Block_that_rewriters_can_copy_into_new_files: 16, + Enum_BlockType.Custom_Block_that_rewriters_should_not_copy_into_new_files: 16, + } # type: dict[int, int] + #: Block types that carry a captured packet, mapped to the tag used when #: reporting them. Parsing any of them resolves the interface the packet was #: captured on, so the enclosing section must describe at least one @@ -156,6 +179,7 @@ def run(self) -> 'None': """ ext = self._extractor + self._bound_read_ahead() shb = P_PCAPNG(ext._ifile, num=0, sct=1, ctx=None) if shb.info.type != Enum_BlockType.Section_Header_Block: raise FormatError(f'PCAP-NG: [SHB] invalid block type: {shb.info.type!r}') @@ -202,6 +226,7 @@ def read_frame(self) -> 'P_PCAPNG': self._check_packet_block_context() # read next block + self._bound_read_ahead() block = P_PCAPNG(ext._ifile, num=ext._frnum+1, sct=len(self._ctx_list), ctx=self._ctx, layer=ext._exlyr, protocol=ext._exptl, __context__=ext._exctx, @@ -367,13 +392,75 @@ def _peek_block_type(self) -> 'Optional[int]': not have a whole block type field left to read. """ - ext = self._extractor - - buffer = ext._ifile.peek(4)[:4] + buffer = self._peek(4) if len(buffer) < 4: return None return int.from_bytes(buffer, self._ctx.section.byteorder) + def _peek(self, size: 'int') -> 'bytes': + """Read the next ``size`` octets of the input without consuming them. + + Args: + size: Number of octets wanted. + + Returns: + The octets, fewer than ``size`` only at the end of the input. + + ``peek`` makes at most one read, which a pipe may answer short of ``size`` + mid-capture, so a short answer is read again and the position put back + (:issue:`1586`). + + """ + ifile = self._extractor._ifile # pylint: disable=protected-access + + buffer = ifile.peek(size)[:size] + if len(buffer) < size: + position = ifile.tell() + buffer = ifile.read(size) + ifile.seek(position, os.SEEK_SET) + return buffer + + def _bound_read_ahead(self) -> 'None': + """Have a non-seekable input read ahead to the end of the next block, no further. + + A block is measured, and then parsed, from as far as + :meth:`SeekableReader.seek ` reads + ahead, so the read has to cover all the block's parse reads, however long -- + the buffer grows to hold them. A read past that waits, on a live capture, for + the blocks after: a buffer's worth held each block back until that much more + had arrived (:issue:`1586`). A bogus length is read as far as the stream goes, + which is as far as a file read goes too. + + How far is enough is the block's Total Length, in the section's byte order + or, for a section header block, the one its Byte-Order Magic declares. Two + kinds of block are parsed past it, and so are read as far as from a file: + one shorter than its own fixed fields, or not a whole number of words, + keeps the buffer's worth; a Decryption Secrets Block reads as far as its + Secrets Length runs. + + """ + # pylint: disable=protected-access + ifile = self._extractor._ifile + if not isinstance(ifile, SeekableReader): + return + size = ifile._buffer_size + + head = self._peek(12) + if head[:4] == self.MAGIC_NUMBER[0]: + byteorder = self.BYTE_ORDER_MAGIC.get(head[8:12]) + else: + byteorder = self._ctx.section.byteorder if self._ctx is not None else None + + if byteorder is not None and len(head) >= 8: + block_type = int.from_bytes(head[:4], byteorder) + length = int.from_bytes(head[4:8], byteorder) + if length >= self.BLOCK_MINIMUM.get(block_type, 12) and length % 4 == 0: + size = length + if block_type == Enum_BlockType.Decryption_Secrets_Block: + secrets = int.from_bytes(self._peek(16)[12:16], byteorder) + size = max(size, 20 + secrets + -secrets % 4) + ifile._read_ahead_size = size + def _check_packet_block_context(self) -> 'None': """Reject a packet block in a section that describes no interface. diff --git a/pcapkit/foundation/extraction.py b/pcapkit/foundation/extraction.py index 6337bd5edd..a8cac4892f 100644 --- a/pcapkit/foundation/extraction.py +++ b/pcapkit/foundation/extraction.py @@ -1389,8 +1389,13 @@ def __init__(self, # will not be moved after reading; however, the returned bytes object # may not be exactly 4 bytes, so we use [:4] to get the first 4 bytes self._magic = self._ifile.peek(4)[:4] - #self._magic = self._ifile.read(4) # magic number - #self._ifile.seek(0, os.SEEK_SET) + if len(self._magic) < 4: + # NOTE: ``peek`` makes at most one read of the stream, which a pipe or a + # raw stream may answer with fewer than four octets, so a short peek is + # not a short file. Read them instead, and seek back (:issue:`1586`). + position = self._ifile.tell() + self._magic = self._ifile.read(4) + self._ifile.seek(position, os.SEEK_SET) self.run() # start extraction diff --git a/tests/corekit/test_io.py b/tests/corekit/test_io.py index 76b998bc1d..992a4c77f3 100644 --- a/tests/corekit/test_io.py +++ b/tests/corekit/test_io.py @@ -257,7 +257,10 @@ def test_seek_variants_warnings_and_truncate_sizes(self) -> None: reader.seek(0, 99) reader.read(2) self.assertEqual(reader.seek(-1, io.SEEK_CUR), 1) - self.assertEqual(reader.seek(-1, io.SEEK_END), 3) + # NOTE: the end is a buffer's worth past the position, 1 + 4, so one before it is + # 4. It was the buffer's far edge, 0 + 4, until #1586 -- which a full buffer's + # edge reaches no further than the stream has been read. + self.assertEqual(reader.seek(-1, io.SEEK_END), 4) with mock.patch.object(self.io_module, 'warn') as warn: self.assertEqual(reader.seek(20), 6) warn.assert_called_once() diff --git a/tests/corekit/test_io_seekable_reader_short_reads_1586_unit.py b/tests/corekit/test_io_seekable_reader_short_reads_1586_unit.py new file mode 100644 index 0000000000..5a09e97a1a --- /dev/null +++ b/tests/corekit/test_io_seekable_reader_short_reads_1586_unit.py @@ -0,0 +1,204 @@ +# -*- coding: utf-8 -*- +""":class:`~pcapkit.corekit.io.SeekableReader` over a stream that answers short (#1586). + +One read of a pipe, or of a raw stream, may return fewer octets than were asked +for at any point; only an empty read means the stream has ended. The reader took +the first answer as all there was in three places, and the frame readers, which +measure what is left with ``seek(0, SEEK_END)``, took that for the end of the +capture: + +* ``read(n)`` made one read of the stream, so it came back short mid-stream. +* ``seek(0, SEEK_END)`` was the buffer's far edge. Once the buffer had filled, + that is exactly how far the stream has been read, so nothing more was read, + and the first record ending there measured nothing left after it. +* A forward ``seek`` made one read too, and landed wherever that read ended -- + short of the target, or past it when the stream answered more than the gap. + +Each test here fails on ``e9f3340f7``, the parent of the fix. + +""" + +import io +import unittest +import warnings + +from tests._support import reimport_once_per_class + +#: Octets the stream holds: more than a few buffers' worth. +DATA = bytes(range(256)) * 4 + + +class ShortReads(io.RawIOBase): + """A non-seekable raw stream answering at most ``chunk`` octets per read.""" + + def __init__(self, data: 'bytes', chunk: 'int') -> None: + super().__init__() + self._data = data + self._pos = 0 + self._chunk = chunk + + @property + def consumed(self) -> int: + """Octets read from the stream so far.""" + return self._pos + + def readable(self) -> bool: + return True + + def seekable(self) -> bool: + return False + + def readinto(self, buffer: 'bytearray | memoryview') -> int: # type: ignore[override] + size = min(len(buffer), self._chunk, len(self._data) - self._pos) + buffer[:size] = self._data[self._pos:self._pos + size] + self._pos += size + return size + + +class TestShortReads(unittest.TestCase): + """A short answer from the stream is not the end of it.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def reader(self, data: 'bytes' = DATA, chunk: 'int' = 3, size: 'int' = 16) -> 'object': + from pcapkit.corekit.io import SeekableReader # pylint: disable=import-outside-toplevel + + reader = SeekableReader(ShortReads(data, chunk), buffer_size=size, stream_closing=False) + self.addCleanup(reader.close) + return reader + + def test_read_returns_every_octet_asked_for(self) -> None: + """``read(10)`` over three-octet answers returned three octets.""" + reader = self.reader() + self.assertEqual(reader.read(10), DATA[:10]) + # half from the buffer, half from the stream + reader.seek(5) + self.assertEqual(reader.read(10), DATA[5:15]) + self.assertEqual(reader.tell(), 15) + + def test_readinto_fills_the_buffer_it_is_given(self) -> None: + reader = self.reader() + buffer = bytearray(10) + self.assertEqual(reader.readinto(buffer), 10) + self.assertEqual(bytes(buffer), DATA[:10]) + + def test_read_comes_back_short_only_at_the_end_of_the_stream(self) -> None: + reader = self.reader(DATA[:20]) + self.assertEqual(reader.read(100), DATA[:20]) + self.assertEqual(reader.read(1), b'') + + def test_end_of_stream_is_a_buffer_past_the_position(self) -> None: + """Before the buffer fills, the end was where one three-octet read left it.""" + reader = self.reader() + self.assertEqual(reader.seek(0, io.SEEK_END), 16) + self.assertEqual(reader.seek(0), 0) + self.assertEqual(reader.read(16), DATA[:16]) + + def test_end_of_stream_after_the_buffer_has_filled(self) -> None: + """The #1586 cut-off: a full buffer's edge measured nothing left to read. + + This is what a frame reader does at the first octet of every record, so + the record after the one that filled the buffer was taken for the end of + the capture -- 870 of the 1,117 frames of ``http.pcap``. + + """ + reader = self.reader(chunk=4096) + self.assertEqual(reader.read(40), DATA[:40]) + + current = reader.tell() + self.assertEqual(reader.seek(0, io.SEEK_END) - current, 16) # was 0 + self.assertEqual(reader.seek(current), current) + self.assertEqual(reader.read(16), DATA[40:56]) + + def test_end_of_stream_is_the_real_end_when_that_comes_first(self) -> None: + reader = self.reader(DATA[:50]) + self.assertEqual(reader.read(40), DATA[:40]) + self.assertEqual(reader.seek(0, io.SEEK_END), 50) + self.assertEqual(reader.read(1), b'') + + def test_forward_seek_lands_on_its_target_over_short_answers(self) -> None: + """A seek past the octets read stopped at the end of one three-octet read.""" + reader = self.reader() + self.assertEqual(reader.seek(10), 10) # was 3 + self.assertEqual(reader.read(2), DATA[10:12]) + + def test_forward_seek_does_not_overshoot_its_target(self) -> None: + """The read that filled the gap asked for more than it, and the seek landed there.""" + from pcapkit.corekit.io import SeekableReader # pylint: disable=import-outside-toplevel + + reader = SeekableReader(io.BytesIO(DATA), buffer_size=16, stream_closing=False) + self.addCleanup(reader.close) + self.assertEqual(reader.seek(1), 1) # was 4 + self.assertEqual(reader.read(2), DATA[1:3]) + + def test_end_of_stream_reads_ahead_only_as_far_as_asked(self) -> None: + """An engine that needs fewer octets measured does not wait for a buffer's worth. + + On a live stream a read past what has arrived waits for the writer, so the + engines set ``_read_ahead_size`` to the record they are about to read. + + """ + reader = self.reader(DATA[:30], chunk=4096) + reader._read_ahead_size = 10 # pylint: disable=protected-access + self.assertEqual(reader.seek(0, io.SEEK_END), 10) + self.assertEqual(reader.raw.consumed, 10) # was 16: the whole buffer's worth + self.assertEqual(reader.seek(0), 0) + self.assertEqual(reader.read(30), DATA[:30]) + + def test_reading_ahead_past_the_buffer_grows_it(self) -> None: + """A record longer than the buffer is held whole, from its first octet on. + + The frame readers measure a record from its first octet and then seek back to + it. Capped at the buffer, the read-ahead measured it short; read on, the record + slid out of the buffer, and the seek back failed with ``cannot seek before the + beginning of the buffer``. + + """ + reader = self.reader(chunk=7) + self.assertEqual(reader.read(8), DATA[:8]) + reader._read_ahead_size = 40 # pylint: disable=protected-access + self.assertEqual(reader.seek(0, io.SEEK_END), 48) # was 24: 8 + 16 + self.assertEqual(reader.seek(8), 8) + self.assertEqual(reader.read(40), DATA[8:48]) + + def test_reading_ahead_a_bogus_length_stops_at_the_real_end(self) -> None: + """A length read from a damaged record is read as far as the stream goes.""" + reader = self.reader(DATA[:50], chunk=4096) + reader._read_ahead_size = 1 << 40 # pylint: disable=protected-access + self.assertEqual(reader.seek(0, io.SEEK_END), 50) # was 16 + self.assertEqual(reader.seek(0), 0) + self.assertEqual(reader.read(), DATA[:50]) + + def test_close_after_the_stream_was_closed(self) -> None: + """A stream its owner closed first made :meth:`close` raise, from a finaliser too. + + Measured on a dpkt ``NeedData``: the extractor left with the reader, which was + collected after the caller closed its stream, and printed ``Exception ignored + ... ValueError: I/O operation on closed file``. + + """ + from pcapkit.corekit.io import SeekableReader # pylint: disable=import-outside-toplevel + + stream = ShortReads(DATA, 3) + reader = SeekableReader(stream, buffer_size=16, stream_closing=False) + self.assertEqual(reader.read(4), DATA[:4]) + stream.close() + reader.close() # was ValueError + self.assertTrue(reader.closed) + + def test_seek_past_a_stream_that_ended_inside_the_buffer_is_silent(self) -> None: + """As a seek past the end of a file is. A frame reader makes one over a cut record.""" + import pcapkit.utilities.warnings as pcapkit_warnings # pylint: disable=import-outside-toplevel + + reader = self.reader(DATA[:10]) + self.assertEqual(reader.read(10), DATA[:10]) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + self.assertEqual(reader.seek(20), 10) + self.assertEqual( + [w for w in caught if issubclass(w.category, pcapkit_warnings.SeekWarning)], []) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/foundation/test_extraction_non_seekable_stream_1586_unit.py b/tests/foundation/test_extraction_non_seekable_stream_1586_unit.py new file mode 100644 index 0000000000..2581e81a80 --- /dev/null +++ b/tests/foundation/test_extraction_non_seekable_stream_1586_unit.py @@ -0,0 +1,465 @@ +# -*- coding: utf-8 -*- +"""Extracting from a pipe or another non-seekable stream drops no frame (#1586). + +``examples/captures/http.pcap`` holds 1,117 frames. Read by path the default +engine gets all of them; piped in -- ``cat http.pcap | ...`` with ``fin`` set to +``sys.stdin.buffer``, an :func:`os.pipe` written in 4,096-octet chunks, or a raw +stream answering short -- it got 870, silently, with ``no_eof`` or without. +Frame 870 is the first to end past the window +:class:`~pcapkit.corekit.io.SeekableReader` keeps over such a stream, 131,072 +octets on CPython 3.14. Cut into seven-octet reads it got none at all, and +one-octet reads failed on the magic number. + +The frame readers measure what is left with ``seek(0, SEEK_END)``, and the +reader answered that with its window's far edge -- once the window has filled, +exactly how far the stream has been read, so the next record measured nothing +left. ``read`` and a forward ``seek`` took one short answer from the stream as +all there was, and ``Extractor`` took a short ``peek`` for a short file. See +:mod:`tests.corekit.test_io_seekable_reader_short_reads_1586_unit` for the +reader itself. Reading a buffer's worth ahead to measure each record then held a +live capture's frames back until that much more had arrived, so the engines read +ahead only as far as the record they are about to parse. + +The captures here are built in memory, past two windows of the default buffer, +so the window fills and slides whatever :data:`io.DEFAULT_BUFFER_SIZE` the +interpreter has. Each case compares the stream's frames, octet for octet and +warning for warning, with the same capture read by path. Each fails on +``e9f3340f7``, the parent of the fix. + +""" + +from __future__ import annotations + +import importlib.util +import io +import os +import pathlib +import struct +import subprocess # nosec: B404 +import sys +import tempfile +import threading +import unittest +import warnings + +from tests._support import reimport_once_per_class, scale_timeout +from tests._tiers import ROOT + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) +HAS_DPKT = importlib.util.find_spec('dpkt') is not None + +#: The chunkings the stream is cut into: single octets, an odd size that splits +#: every header, a pipe's page, and more than a pipe holds. +CHUNKS = (1, 7, 4096, 65536) + +#: Size of each built capture: past two windows of the default buffer. +SIZE = 2 * io.DEFAULT_BUFFER_SIZE + 3000 + +#: Seconds allowed to a CLI run, which takes about one. +TIMEOUT = 120 + + +def packet(index: 'int') -> 'bytes': + """An Ethernet frame of 60 to 1,514 octets, with an experimental EtherType.""" + size = 60 + (index * 397) % 1455 + body = bytes((index + offset) & 0xFF for offset in range(size - 14)) + return b'\x02\x00\x00\x00\x00\x01\x02\x00\x00\x00\x00\x02\x88\xb5' + body + + +def make_pcap(size: 'int' = SIZE) -> 'bytes': + """A little-endian PCAP capture of at least ``size`` octets.""" + out = [struct.pack(' 'bytes': + """An enhanced packet block on interface 0, in byte order ``order``.""" + data = packet(index) + pad = -len(data) % 4 + block = 32 + len(data) + pad + stamp = (1_700_000_000 + index) * 1_000_000 + return (struct.pack(f'{order}IIIIIII', 6, block, 0, stamp >> 32, stamp & 0xFFFFFFFF, + len(data), len(data)) + + data + bytes(pad) + struct.pack(f'{order}I', block)) + + +def make_pcapng(size: 'int' = SIZE, order: 'str' = '<') -> 'bytes': + """A PCAP-NG section of a header, an interface, and enhanced packet blocks.""" + out = [ + struct.pack(f'{order}IIIHHqI', 0x0A0D0D0A, 28, 0x1A2B3C4D, 1, 0, -1, 28), + struct.pack(f'{order}IIHHII', 1, 20, 1, 0, 65535, 20), + ] + length = 48 + index = 0 + while length < size: + out.append(epb(index, order)) + length += len(out[-1]) + index += 1 + return b''.join(out) + + +def two_sections() -> 'tuple[bytes, bytes]': + """A little-endian section and a big-endian one, and their packet blocks.""" + first, second = make_pcapng(SIZE // 2), make_pcapng(SIZE // 2, '>') + return first + second, first[48:] + second[48:] + + +#: Length of a record longer than the default buffer, so it cannot be held in one. +OVERSIZE = io.DEFAULT_BUFFER_SIZE + 9000 + + +def oversize(kind: 'str') -> 'bytes': + """A capture whose second record, or whose section header, is :data:`OVERSIZE` long. + + Args: + kind: ``'pcap'`` for a PCAP record, ``'epb'`` for a PCAP-NG enhanced packet + block, ``'shb'`` for a PCAP-NG section header block, made long by comments. + + """ + long = packet(0)[:14] + bytes(range(256)) * (OVERSIZE // 256) + bytes(OVERSIZE % 256 - 14) + if kind == 'pcap': + out = [struct.pack(' None: + super().__init__() + self._data = data + self._pos = 0 + self._chunk = chunk + + def readable(self) -> bool: + return True + + def seekable(self) -> bool: + return False + + def readinto(self, buffer: 'bytearray | memoryview') -> int: # type: ignore[override] + size = min(len(buffer), self._chunk, len(self._data) - self._pos) + buffer[:size] = self._data[self._pos:self._pos + size] + self._pos += size + return size + + +def feed(fd: 'int', data: 'bytes', chunk: 'int') -> 'threading.Thread': + """Write ``data`` to ``fd`` in ``chunk``-octet writes, then close it, from a thread.""" + def write() -> None: + try: + with os.fdopen(fd, 'wb', buffering=0) as stream: + for offset in range(0, len(data), chunk): + stream.write(data[offset:offset + chunk]) + except OSError: # the reader stopped early and closed its end + pass + + thread = threading.Thread(target=write, daemon=True) + thread.start() + return thread + + +@unittest.skipUnless(HAS_RUNTIME, 'pcapkit runtime dependencies are not installed') +class TestNonSeekableStream(unittest.TestCase): + """Every frame from a pipe or a short-answering stream, as from a path.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + tmp = tempfile.TemporaryDirectory() # pylint: disable=consider-using-with + self.addCleanup(tmp.cleanup) + self.tmp_path = pathlib.Path(tmp.name) + + def extract(self, fin: 'object', **kwargs: 'object') -> 'tuple[list[bytes], set[tuple[str, str]]]': + """The frames' octets, and the warnings raised on the way.""" + import pcapkit # pylint: disable=import-outside-toplevel + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + extractor = pcapkit.extract(fin=fin, nofile=True, **kwargs) + return ([bytes(frame) for frame in extractor.frame], # type: ignore[attr-defined] + {(w.category.__name__, str(w.message)) for w in caught}) + + def by_path(self, data: 'bytes', **kwargs: 'object') -> 'tuple[list[bytes], set[tuple[str, str]]]': + path = self.tmp_path / 'capture.pcap' + path.write_bytes(data) + return self.extract(str(path), **kwargs) + + def by_pipe(self, data: 'bytes', chunk: 'int', **kwargs: 'object') -> 'tuple[list[bytes], set[tuple[str, str]]]': + read_fd, write_fd = os.pipe() + thread = feed(write_fd, data, chunk) + with os.fdopen(read_fd, 'rb', buffering=0) as stream: + result = self.extract(stream, **kwargs) + thread.join(scale_timeout(30)) + return result + + def streams(self, data: 'bytes', chunk: 'int') -> 'dict[str, object]': + return { + 'raw': ShortReads(data, chunk), + 'buffered': io.BufferedReader(ShortReads(data, chunk)), + } + + def assert_every_chunking(self, data: 'bytes', body: 'bytes', **kwargs: 'object') -> None: + """Every chunking and stream kind, with ``no_eof`` and without, against the path.""" + for no_eof in (False, True): + expected, expected_warnings = self.by_path(data, no_eof=no_eof, **kwargs) + self.assertEqual(b''.join(expected), body, 'the path read is the reference') + + for chunk in CHUNKS: + with self.subTest(stream='pipe', chunk=chunk, no_eof=no_eof): + self.assertEqual(self.by_pipe(data, chunk, no_eof=no_eof, **kwargs), + (expected, expected_warnings)) + for name, stream in self.streams(data, chunk).items(): + with self.subTest(stream=name, chunk=chunk, no_eof=no_eof): + self.assertEqual(self.extract(stream, no_eof=no_eof, **kwargs), + (expected, expected_warnings)) + + def test_pcap_every_chunking(self) -> None: + data = make_pcap() + self.assert_every_chunking(data, data[24:]) + + def test_pcapng_every_chunking(self) -> None: + """Two sections, the second big-endian, as its Byte-Order Magic declares.""" + self.assert_every_chunking(*two_sections()) + + @unittest.skipUnless(HAS_DPKT, 'dpkt is not installed') + def test_dpkt_engine_every_chunking(self) -> None: + """The dpkt engine reads the same wrapped stream, through ``read``.""" + for name, data in (('pcap', make_pcap()), ('pcapng', make_pcapng())): + with self.subTest(format=name): + expected = self.by_path(data, engine='dpkt') + for chunk in (7, 4096): + with self.subTest(chunk=chunk): + self.assertEqual(self.extract(ShortReads(data, chunk), engine='dpkt'), expected) + + def test_no_eof_over_a_pipe_whose_writer_closed(self) -> None: + """The ``pcapkit -`` case, which sets ``no_eof``: every frame, and it returns.""" + data = make_pcap() + expected = self.by_path(data, no_eof=True) + self.assertEqual(self.by_pipe(data, 4096, no_eof=True), expected) + + def test_a_live_pipe_hands_over_each_frame_as_it_arrives(self) -> None: + """A frame does not wait for the stream to go on past it. + + ``pcapkit -`` reads a live capture. Measuring a record by reading a buffer's + worth past it held each frame back until that much more had arrived; and + before the fix, the second measurement of each record waited for at least one + more octet. The writer here sends one record, then waits for the reader to + have it before sending the rest. + + """ + for name, data, first in ( + ('pcap', make_pcap(), 24 + 16 + len(packet(0))), + ('pcapng', make_pcapng(), 48 + len(epb(0))), + ): + with self.subTest(format=name): + frames, early = self.live(data, first) + self.assertTrue(early, 'the first frame waited for the rest of the capture') + self.assertEqual(frames, self.by_path(data)[0]) + + def live(self, data: 'bytes', first: 'int') -> 'tuple[list[bytes], bool]': + """Read ``data`` off a pipe whose writer holds back all but ``first`` octets. + + Returns: + The frames, and whether the first arrived while the writer was holding. + + """ + import pcapkit # pylint: disable=import-outside-toplevel + + received = threading.Event() + resumed = threading.Event() + read_fd, write_fd = os.pipe() + + def write() -> None: + with os.fdopen(write_fd, 'wb', buffering=0) as stream: + stream.write(data[:first]) + received.wait(scale_timeout(10)) + resumed.set() + stream.write(data[first:]) + + thread = threading.Thread(target=write, daemon=True) + thread.start() + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + with os.fdopen(read_fd, 'rb', buffering=0) as stream: + extractor = pcapkit.extract(fin=stream, nofile=True, auto=False, no_eof=True) + frames = [bytes(next(extractor))] + early = not resumed.is_set() + received.set() + frames.extend(bytes(frame) for frame in extractor) + thread.join(scale_timeout(30)) + return frames, early + + def test_a_record_longer_than_the_buffer_reads_as_from_a_file(self) -> None: + """A record the buffer cannot hold is read whole, as it is from a file. + + The read-ahead was capped at the buffer, so a PCAP-NG block longer than it was + measured short and parsed as cut, and a PCAP record slid out of the buffer + before its reader sought back to it, which failed with ``cannot seek before + the beginning of the buffer``. + + """ + for kind in ('pcap', 'epb', 'shb'): + data = oversize(kind) + with self.subTest(record=kind): + expected = self.by_path(data) + self.assertEqual(self.by_pipe(data, 65536), expected) + self.assertEqual(self.extract(ShortReads(data, 7)), expected) + + @unittest.skipUnless(HAS_DPKT, 'dpkt is not installed') + def test_dpkt_reads_a_section_header_longer_than_the_buffer(self) -> None: + """The dpkt engine seeks back to the first octet once the section header is read. + + That is :meth:`Extractor.record_header + `, which failed with + ``cannot seek before the beginning of the buffer: 0`` once the section header + had slid out of it. + + """ + data = oversize('shb') + expected = self.by_path(data, engine='dpkt') + self.assertEqual(self.by_pipe(data, 65536, engine='dpkt'), expected) + + def test_pcapng_blocks_parsed_past_their_end_read_as_from_a_file(self) -> None: + """A block whose parse runs past its Total Length sees what a file read sees. + + The PCAP-NG engine reads a stream ahead only as far as the next block, but two + kinds of block are parsed past that: one shorter than its own fixed fields, + and a Decryption Secrets Block whose Secrets Length runs past it. Read only as + far as the block, the first warned ``packet length < 0`` where the file did + not, and the second was kept as cut where the file read the blocks after it + as secrets and failed. + + """ + import pcapkit.utilities.exceptions as errors # pylint: disable=import-outside-toplevel + + data = make_pcapng() + middle = 48 + sum(len(epb(index)) for index in range(100)) + short = struct.pack(' None: + """The engine's look at the next block type was a ``peek``, short on a pipe.""" + import pcapkit.utilities.exceptions as errors # pylint: disable=import-outside-toplevel + + data = make_pcapng() + capture = data[:28] + data[48:] # the interface description dropped + with self.assertRaises(errors.FormatError) as path: + self.by_path(capture) + with self.assertRaises(errors.FormatError) as stream: + self.extract(ShortReads(capture, 1)) + self.assertEqual(str(stream.exception), str(path.exception)) + + def test_a_truncated_stream_ends_as_the_truncated_file_does(self) -> None: + """Cut inside a record past the first window: same frames, same warnings, no others. + + For PCAP-NG, inside a block's header too, where its Total Length cannot be read. + + """ + pcap, (pcapng, _) = make_pcap(), two_sections() + header = 48 + len(epb(0)) + 6 + for name, data, cut in (('pcap', pcap, len(pcap) - 5), ('pcap', pcap, len(pcap) - 1000), + ('pcapng', pcapng, len(pcapng) - 5), ('pcapng', pcapng, header)): + with self.subTest(format=name, cut=cut): + expected, expected_warnings = self.by_path(data[:cut]) + self.assertIn(('ExtractionWarning', 'EOF reached'), expected_warnings) + self.assertNotIn('SeekWarning', {category for category, _ in expected_warnings}) + for chunk in (7, 4096): + self.assertEqual(self.extract(ShortReads(data[:cut], chunk)), + (expected, expected_warnings)) + + def test_cli_reads_every_frame_from_stdin(self) -> None: + """``pcapkit -`` writes the same report from a pipe as from the file. + + A PCAP capture to a PCAP report, which is the capture again, and a PCAP-NG one, + two sections of it, to JSON, which has a key per frame. + + """ + pcapng, _ = two_sections() + for name, data, fmt in (('pcap', make_pcap(), 'pcap'), ('pcapng', pcapng, 'json')): + with self.subTest(format=name): + from_path, from_stdin = self.cli(data, name, fmt) + self.assertEqual(from_stdin, from_path) + if fmt == 'pcap': + self.assertEqual(from_path, data) + else: + frames = self.by_path(data)[0] + self.assertIn(f'"Frame {len(frames)}"'.encode(), from_path) + self.assertNotIn(f'"Frame {len(frames) + 1}"'.encode(), from_path) + + def cli(self, data: 'bytes', name: 'str', fmt: 'str') -> 'tuple[bytes, bytes]': + """Run ``pcapkit`` on ``data`` from a file and from stdin, and return both reports.""" + capture = self.tmp_path / f'capture.{name}' + capture.write_bytes(data) + script = ( + 'import sys\n' + f'sys.path.insert(0, {str(ROOT)!r})\n' + 'import pcapkit\n' + f'assert pcapkit.__file__.startswith({str(ROOT)!r}), pcapkit.__file__\n' + 'from pcapkit.__main__ import main\n' + "sys.argv[0] = 'pcapkit-cli'\n" + 'sys.exit(main())\n' + ) + + def run(fin: 'str', out: 'str', stdin: 'int | None' = None) -> 'bytes': + report = self.tmp_path / f'{out}.{fmt}' + completed = subprocess.run( # nosec: B603 + [sys.executable, '-c', script, fin, '-o', str(report), '-f', fmt], + stdin=stdin, cwd=str(self.tmp_path), capture_output=True, + timeout=scale_timeout(TIMEOUT), check=False, + ) + self.assertEqual(completed.returncode, 0, completed.stderr.decode(errors='replace')) + return report.read_bytes() + + from_path = run(str(capture), f'{name}-path') + read_fd, write_fd = os.pipe() + thread = feed(write_fd, data, 4096) + try: + from_stdin = run('-', f'{name}-stdin', stdin=read_fd) + finally: + os.close(read_fd) + thread.join(scale_timeout(30)) + return from_path, from_stdin + + +if __name__ == '__main__': + unittest.main()