Skip to content

DCTwav

stegobox.codec.DCTwav

Bases: BaseCodec

This module implements simple DCT steganography in WAV audios.

the code by zhudehua

Source code in stegobox/codec/dct_wav_str.py 
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
class DCTwav(BaseCodec):
    """
    This module implements simple DCT steganography in WAV audios.

    the code by zhudehua
    """

    def __init__(self) -> None:
        super().__init__()

    def encode(
        self, carrier: wave.Wave_read, payload: str
    ) -> tuple[bytes, wave._wave_params]:
        """Encoder requires the carrier audio to be WAV and the payload to be a string.

        Args:
            carrier: Carrier audio in format WAV. Read with `stegobox.io.audio.read()`.
            payload: Payload (secret message) to be encoded.

        Returns:
            bytes: Frame information in bytes of the audio with the payload embedded.
            wave._wave_params: Parameters for audio with the payload embedded.
        """
        # make string to bit
        message_bits = []
        for pl in payload:
            message_bits += [(ord(pl) >> i) & 1 for i in range(8)]

        # get the basic info from carrier

        frame_num = carrier.getnframes()
        wave_head = carrier.getparams()
        frame_rate = carrier.getframerate()
        frame_data = carrier.readframes(frame_num)
        # calculate the length of dct data
        bin_data = bytearray(frame_data)
        block_size = len(bin_data) // 8
        zong_length = block_size * 8
        dct_data = bin_data[:zong_length]
        # deal the left data
        left_data = bin_data[zong_length:]
        left = []
        for i in range(len(left_data)):
            left.append(left_data[i])
            left.append(left_data[i])
        # make the carrier data to N*4*2 block for dct's input
        dct_data = np.array(dct_data).reshape(-1, 2)
        dct_data = dct_data.astype(np.float32)
        dct_data = np.vsplit(dct_data, block_size)
        temp_dct = np.zeros(shape=(block_size, 4, 2))
        # DCT
        for i in range(len(dct_data)):
            temp_dct[i] = cv2.dct(dct_data[i])

        out_audio_dct = []
        end = 0
        #   insert into secret info
        for i in range(len(message_bits)):
            end = end + 1
            out_audio_dct.append(temp_dct[i])
            te = temp_dct[i].copy()
            if message_bits[i] == 0:
                te[0][0] = te[0][0] - 10
            out_audio_dct.append(te)

        for j in range(end + 1, len(temp_dct)):
            out_audio_dct.append(temp_dct[j])
            out_audio_dct.append(temp_dct[j])

        outdata = np.zeros(shape=(2 * block_size, 4, 2))

        for i in range(len(out_audio_dct)):
            outdata[i] = cv2.idct(out_audio_dct[i])
        outdata = np.trunc(outdata).astype(np.uint8)

        for i in range(10):
            print(outdata[i])
        outdata = outdata.reshape(-1)
        outdata = bytes(outdata)
        outdata = outdata + bytes(left)
        wave_head = wave_head._replace(framerate=2 * frame_rate)
        wave_head = wave_head._replace(nframes=2 * frame_num)
        return outdata, wave_head

    def decode(self, carrier: wave.Wave_read, payload_length: int = 20) -> str:
        """Decode the secret payload from the carrier audio

        Args:
            carrier: Carrier audio in format WAV. Read with `stegobox.io.audio.read()`.
            payload_length: the length of secret message

        Returns:
            The decoded payload (secret message).
        """

        frame_data = carrier.readframes(carrier.getnframes())
        bin_data = bytearray(frame_data)
        # divide into  4*2 block
        block_size = len(bin_data) // 8
        zong_length = block_size * 8
        dct_data = bin_data[:zong_length]
        dct_data = np.array(dct_data).reshape(-1, 2)
        dct_data = dct_data.astype(np.float32)
        dct_data = np.vsplit(dct_data, block_size)
        # used to save data after dct
        temp_dct = np.zeros(shape=(block_size, 4, 2))
        for i in range(len(dct_data)):
            temp_dct[i] = cv2.dct(dct_data[i])

        # used to save extract data in format bit
        outdata = []
        for i in range(0, 2 * payload_length, 2):
            if temp_dct[i][0][0] == temp_dct[i + 1][0][0]:
                outdata.append(1)
            else:
                outdata.append(0)
        message_ex = []
        value = 0
        for i in range(len(outdata)):
            if i % 8 == 0 and i != 0:
                message_ex.append(value)
                value = 0
            value |= outdata[i] << i % 8
        payload = "".join([chr(L) for L in message_ex])[:payload_length]

        return payload

encode(carrier, payload)

Encoder requires the carrier audio to be WAV and the payload to be a string.

Parameters:

Name Type Description Default
carrier Wave_read

Carrier audio in format WAV. Read with stegobox.io.audio.read().

 required
payload str

Payload (secret message) to be encoded.

 required

Returns:

Name Type Description
bytes bytes

Frame information in bytes of the audio with the payload embedded.
_wave_params

wave._wave_params: Parameters for audio with the payload embedded.
Source code in stegobox/codec/dct_wav_str.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def encode(
    self, carrier: wave.Wave_read, payload: str
) -> tuple[bytes, wave._wave_params]:
    """Encoder requires the carrier audio to be WAV and the payload to be a string.

    Args:
        carrier: Carrier audio in format WAV. Read with `stegobox.io.audio.read()`.
        payload: Payload (secret message) to be encoded.

    Returns:
        bytes: Frame information in bytes of the audio with the payload embedded.
        wave._wave_params: Parameters for audio with the payload embedded.
    """
    # make string to bit
    message_bits = []
    for pl in payload:
        message_bits += [(ord(pl) >> i) & 1 for i in range(8)]

    # get the basic info from carrier

    frame_num = carrier.getnframes()
    wave_head = carrier.getparams()
    frame_rate = carrier.getframerate()
    frame_data = carrier.readframes(frame_num)
    # calculate the length of dct data
    bin_data = bytearray(frame_data)
    block_size = len(bin_data) // 8
    zong_length = block_size * 8
    dct_data = bin_data[:zong_length]
    # deal the left data
    left_data = bin_data[zong_length:]
    left = []
    for i in range(len(left_data)):
        left.append(left_data[i])
        left.append(left_data[i])
    # make the carrier data to N*4*2 block for dct's input
    dct_data = np.array(dct_data).reshape(-1, 2)
    dct_data = dct_data.astype(np.float32)
    dct_data = np.vsplit(dct_data, block_size)
    temp_dct = np.zeros(shape=(block_size, 4, 2))
    # DCT
    for i in range(len(dct_data)):
        temp_dct[i] = cv2.dct(dct_data[i])

    out_audio_dct = []
    end = 0
    #   insert into secret info
    for i in range(len(message_bits)):
        end = end + 1
        out_audio_dct.append(temp_dct[i])
        te = temp_dct[i].copy()
        if message_bits[i] == 0:
            te[0][0] = te[0][0] - 10
        out_audio_dct.append(te)

    for j in range(end + 1, len(temp_dct)):
        out_audio_dct.append(temp_dct[j])
        out_audio_dct.append(temp_dct[j])

    outdata = np.zeros(shape=(2 * block_size, 4, 2))

    for i in range(len(out_audio_dct)):
        outdata[i] = cv2.idct(out_audio_dct[i])
    outdata = np.trunc(outdata).astype(np.uint8)

    for i in range(10):
        print(outdata[i])
    outdata = outdata.reshape(-1)
    outdata = bytes(outdata)
    outdata = outdata + bytes(left)
    wave_head = wave_head._replace(framerate=2 * frame_rate)
    wave_head = wave_head._replace(nframes=2 * frame_num)
    return outdata, wave_head

decode(carrier, payload_length=20)

Decode the secret payload from the carrier audio

Parameters:

Name Type Description Default
carrier Wave_read

Carrier audio in format WAV. Read with stegobox.io.audio.read().

 required
payload_length int

the length of secret message

 20

Returns:

Type Description
str

The decoded payload (secret message).
Source code in stegobox/codec/dct_wav_str.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def decode(self, carrier: wave.Wave_read, payload_length: int = 20) -> str:
    """Decode the secret payload from the carrier audio

    Args:
        carrier: Carrier audio in format WAV. Read with `stegobox.io.audio.read()`.
        payload_length: the length of secret message

    Returns:
        The decoded payload (secret message).
    """

    frame_data = carrier.readframes(carrier.getnframes())
    bin_data = bytearray(frame_data)
    # divide into  4*2 block
    block_size = len(bin_data) // 8
    zong_length = block_size * 8
    dct_data = bin_data[:zong_length]
    dct_data = np.array(dct_data).reshape(-1, 2)
    dct_data = dct_data.astype(np.float32)
    dct_data = np.vsplit(dct_data, block_size)
    # used to save data after dct
    temp_dct = np.zeros(shape=(block_size, 4, 2))
    for i in range(len(dct_data)):
        temp_dct[i] = cv2.dct(dct_data[i])

    # used to save extract data in format bit
    outdata = []
    for i in range(0, 2 * payload_length, 2):
        if temp_dct[i][0][0] == temp_dct[i + 1][0][0]:
            outdata.append(1)
        else:
            outdata.append(0)
    message_ex = []
    value = 0
    for i in range(len(outdata)):
        if i % 8 == 0 and i != 0:
            message_ex.append(value)
            value = 0
        value |= outdata[i] << i % 8
    payload = "".join([chr(L) for L in message_ex])[:payload_length]

    return payload