Class RandomLSBInputStream

StegaPy/plugin/randlsb/random_lsb_input_stream.py:12–79

RandomLSB 输入流，使用与嵌入端相同的伪随机序列还原散布位。

class RandomLSBInputStream:
    """RandomLSB 输入流，使用与嵌入端相同的伪随机序列还原散布位。"""

    def __init__(self, image: Image.Image, config: LSBConfig, password: str = None):
        self.config = config
        self.pixels = np.array(image)
        self.height, self.width, self.channels = self.pixels.shape

        seed = CommonUtil.password_hash(password) if password else 0
        total_positions = self.width * self.height * self.channels
        rng = np.random.default_rng(seed % (2 ** 63))
        self.position_sequence = rng.permutation(total_positions)

        self._bit_offset = 0

        # 先读固定部分以获取 filename_len，再读变长文件名
        fixed_size = (len(LSBDataHeader.DATA_STAMP) + len(LSBDataHeader.HEADER_VERSION)
                      + LSBDataHeader.FIXED_HEADER_LENGTH + LSBDataHeader.CRYPT_ALGO_LENGTH)
        fixed_bytes = self._read_bytes(fixed_size)

        fixed_header_offset = len(LSBDataHeader.DATA_STAMP) + len(LSBDataHeader.HEADER_VERSION)
        filename_len = fixed_bytes[fixed_header_offset + 5]

        filename_bytes = self._read_bytes(filename_len) if filename_len > 0 else b''
        self.header = LSBDataHeader.from_bytes(fixed_bytes + filename_bytes, config)

    def _read_bytes(self, n_bytes: int) -> bytes:
        bits_per_ch = self.config.get_max_bits_used_per_channel()
        n_bits = n_bytes * 8

        if bits_per_ch == 1:
            positions = self.position_sequence[self._bit_offset:self._bit_offset + n_bits].astype(np.int64)
            pixel_indices = positions // self.channels
            channel_indices = positions % self.channels
            rows = pixel_indices // self.width
            cols = pixel_indices % self.width
            bits = (self.pixels[rows, cols, channel_indices] & 1).astype(np.uint8)
            self._bit_offset += n_bits
            return np.packbits(bits).tobytes()
        else:
            result = bytearray()
            for _ in range(n_bytes):
                byte_val = 0
                for bit_pos in range(8):
                    bit = self._read_one_bit()
                    byte_val = (byte_val << 1) | bit
                result.append(byte_val)
            return bytes(result)

    def _read_one_bit(self) -> int:
        bits_per_ch = self.config.get_max_bits_used_per_channel()
        ch_pos = self._bit_offset // bits_per_ch
        bit_slot = self._bit_offset % bits_per_ch
        pos = int(self.position_sequence[ch_pos])
        pixel_idx = pos // self.channels
        channel = pos % self.channels
        row = pixel_idx // self.width
        col = pixel_idx % self.width