:py:mod:`dissect.util.stream` ============================= .. py:module:: dissect.util.stream Module Contents --------------- Classes ~~~~~~~ .. autoapisummary:: dissect.util.stream.AlignedStream dissect.util.stream.RangeStream dissect.util.stream.RelativeStream dissect.util.stream.BufferedStream dissect.util.stream.MappingStream dissect.util.stream.RunlistStream dissect.util.stream.OverlayStream dissect.util.stream.ZlibStream dissect.util.stream.BitStream Attributes ~~~~~~~~~~ .. autoapisummary:: dissect.util.stream.STREAM_BUFFER_SIZE .. py:data:: STREAM_BUFFER_SIZE .. py:class:: AlignedStream(size: int | None = None, align: int = STREAM_BUFFER_SIZE) Bases: :py:obj:`io.RawIOBase` Basic buffered stream that provides aligned reads. Must be subclassed for various stream implementations. Subclasses can implement: - :meth:`~AlignedStream._read` - :meth:`~AlignedStream._seek` The offset and length for ``_read`` are guaranteed to be aligned for streams of a known size. If your stream has an unknown size (i.e. ``size == None``), reads of length ``-1`` (i.e. read until EOF) will be passed through to your implementation of ``_read``. The only time that overriding ``_seek`` would make sense is if there's no known size of your stream, but still want to provide ``SEEK_END`` functionality. Most subclasses of ``AlignedStream`` take one or more file-like objects as source. Operations on these subclasses, like reading, will modify the source file-like object as a side effect. :param size: The size of the stream. This is used in read and seek operations. ``None`` if unknown. :param align: The alignment size. Read operations are aligned on this boundary. Also determines buffer size. .. automethod:: _read .. automethod:: _seek .. py:attribute:: size :value: None .. py:attribute:: align .. py:method:: readable() -> bool Indicate that the stream is readable. .. py:method:: seekable() -> bool Indicate that the stream is seekable. .. py:method:: seek(pos: int, whence: int = io.SEEK_SET) -> int Seek the stream to the specified position. :returns: The new stream position after seeking. .. py:method:: tell() -> int Return current stream position. .. py:method:: read(n: int = -1) -> bytes Read and return up to ``n`` bytes, or read to the end of the stream if ``n`` is ``-1``. Returns an empty bytes object on EOF. .. py:method:: readinto(b: bytearray) -> int Read bytes into a pre-allocated bytes-like object b. Returns an int representing the number of bytes read (0 for EOF). .. py:method:: readall() -> bytes Read until end of stream. .. py:method:: readoffset(offset: int, length: int) -> bytes Convenience method to read from a given offset. :param offset: The offset in the stream to read from. :param length: The number of bytes to read. .. py:method:: peek(n: int) -> bytes Convenience method to peek from the current offset without advancing the stream position. :param n: The number of bytes to peek. .. py:method:: close() -> None Close the stream. Does nothing by default. .. py:class:: RangeStream(fh: BinaryIO, offset: int, size: int | None, align: int = STREAM_BUFFER_SIZE) Bases: :py:obj:`AlignedStream` Create a stream with a specific range from another file-like object. ASCII representation:: Source file-like object |................................................| RangeStream with offset and size |............................| :param fh: The source file-like object. :param offset: The offset the stream should start from on the source file-like object. :param size: The size the stream should be. :param align: The alignment size. .. py:attribute:: offset .. py:class:: RelativeStream(fh: BinaryIO, offset: int, size: int | None = None, align: int = STREAM_BUFFER_SIZE) Bases: :py:obj:`RangeStream` Create a relative stream from another file-like object. ASCII representation:: Source file-like object |................................................| RelativeStream with offset |........................................| :param fh: The source file-like object. :param offset: The offset the stream should start from on the source file-like object. :param size: Optional size the stream should be. :param align: The alignment size. .. py:class:: BufferedStream(fh: BinaryIO, offset: int = 0, size: int | None = None, align: int = STREAM_BUFFER_SIZE) Bases: :py:obj:`RelativeStream` Create a buffered stream from another file-like object. Optionally start from a specific offset. :param fh: The source file-like object. :param offset: The offset the stream should start from. :param size: The size the stream should be. :param align: The alignment size. .. py:class:: MappingStream(size: int | None = None, align: int = STREAM_BUFFER_SIZE) Bases: :py:obj:`AlignedStream` Create a stream from multiple mapped file-like objects. :param size: The size the stream should be. :param align: The alignment size. .. py:method:: add(offset: int, size: int, fh: BinaryIO, file_offset: int = 0) -> None Add a file-like object to the stream. :param offset: The offset in the stream this fh maps to. :param size: The size that this mapped fh spans in the stream. :param fh: The file-like object to map. :param file_offset: The offset in the fh to start from. Note that there is no check on overlapping offsets and/or sizes. .. py:class:: RunlistStream(fh: BinaryIO, runlist: list[tuple[int, int]], size: int, block_size: int, align: int | None = None) Bases: :py:obj:`AlignedStream` Create a stream from multiple runs on another file-like object. This is common in filesystems, where file data information is stored in "runs". A run is a ``(block_offset, block_count)`` tuple, meaning the amount of consecutive blocks from a specific starting block. A block_offset of ``None`` represents a sparse run, meaning it must simply return all ``\x00`` bytes. :param fh: The source file-like object. :param runlist: The runlist for this stream in block units. :param size: The size of the stream. This can be smaller than the total sum of blocks (to account for slack space). :param block_size: The block size in bytes. :param align: Optional alignment that differs from the block size, otherwise ``block_size`` is used as alignment. .. py:property:: runlist :type: list[tuple[int, int]] .. py:attribute:: block_size .. py:class:: OverlayStream(fh: BinaryIO, size: int | None = None, align: int = STREAM_BUFFER_SIZE) Bases: :py:obj:`AlignedStream` Create a stream from another file-like object with the ability to overlay other streams or bytes. Useful for patching large file-like objects without having to cache the entire contents. First wrap the original stream in this class, and then call ``add()`` with the offset and data to overlay. :param fh: The source file-like object. :param size: The size the stream should be. :param align: The alignment size. .. py:attribute:: overlays :type: dict[int, tuple[int, BinaryIO]] .. py:method:: add(offset: int, data: bytes | BinaryIO, size: int | None = None) -> None Add an overlay at the given offset. :param offset: The offset in bytes to add an overlay at. :param data: The bytes or file-like object to overlay. :param size: Optional size specification of the overlay, if it can't be inferred. .. py:class:: ZlibStream(fh: BinaryIO, size: int | None = None, align: int = STREAM_BUFFER_SIZE, **kwargs) Bases: :py:obj:`AlignedStream` Create a zlib stream from another file-like object. Basically the same as ``gzip.GzipFile`` but for raw zlib streams. Due to the nature of zlib streams, seeking backwards requires resetting the decompression context. :param fh: The source file-like object. :param size: The size the stream should be. .. py:method:: readall() -> bytes Read until end of stream. .. py:class:: BitStream(fh: BinaryIO) Bit-level stream reader. :param fh: File-like object to read bits from. .. py:attribute:: fh .. py:attribute:: buffer :value: 0 .. py:attribute:: bits :value: 0 .. py:method:: readable() -> bool Indicate that the stream is readable. .. py:method:: seekable() -> bool Indicate that the stream is seekable. .. py:method:: writable() -> bool Indicate that the stream is not writable. .. py:method:: seek(pos: int, whence: int = io.SEEK_SET) -> int Seek the stream to the specified position in bits. :returns: The new stream position after seeking. .. py:method:: tell() -> int Get the current position in the stream in bits. .. py:method:: read(n: int) -> int Read n bits from the stream. :param n: Number of bits to read. .. py:method:: peek(n: int) -> int Peek n bits from the stream without advancing. :param n: Number of bits to peek. .. py:method:: remove(n: int) -> None Remove n bits from the stream. :param n: Number of bits to remove.