Python bit flips when writing a file

Ziv Ofir 0
Viewed 46

Background

This is a steganography project I am working on. The code is supposed to do the following:

  1. Read a mp3 file (in bytes)
  2. Find the frame headers (via the frame sync bits 0xFF)
  3. Get and convert a secret message (string) into bits
  4. Insert the bits into different parts of the frame header, such as bit 24, 29, 30. (with reference to this)
  5. Write the data back into a new file
  6. Try and retrieve the data from this new file

Errors encountered

Currently, after writing back to the file, when trying to retrieve the data from the newly written file, a few parts of the text will become garbled. Upon further analysis, it seems that there were some bit flips that occurred. In the following code, only options for 2 bits have been increasing the number of bits just adds a few more function call per bit. This issue occurs once more than one bit is used.

The code

steg.py:

import binascii
import utility


def encode_message(list_of_hex: list, message: str, num_bits: int = 1):
    """Function takes in a list of hexadecimal values in byte format, a message in string format and number of bits
    to replace (defaults to 1). Replaces the bytes in place so there is no return"""
    message_counter = 0
    message = ''.join(f"{ord(i):08b}" for i in message)
    length_of_message = len(message)
    if length_of_message % num_bits != 0:
        message += (num_bits - (length_of_message % num_bits)) * '0'
    for i in range(len(list_of_hex)):
        if message_counter < length_of_message:
            if list_of_hex[i] == b'ff' and list_of_hex[i + 1][:1] == b'f':
                if num_bits == 1:
                    current_byte = utility.convert_byte_hex_to_bin(list_of_hex[i + 2])  # Gets the next 8 bits (17-24)
                    replaced_byte = utility.encode_bit_0(current_byte, message[message_counter])  # Replace bit 24
                    message_counter += 1
                    list_of_hex[i + 2] = utility.convert_bin_to_hex(replaced_byte)
                elif num_bits == 2:
                    current_byte = utility.convert_byte_hex_to_bin(list_of_hex[i + 2])  # Gets the next 8 bits (17-24)
                    replaced_byte = utility.encode_bit_0(current_byte, message[message_counter])  # Replace bit 24
                    message_counter += 1
                    list_of_hex[i + 2] = utility.convert_bin_to_hex(replaced_byte)
                    # Encodes the next part of the header (Bits 25-32).
                    current_byte = utility.convert_byte_hex_to_bin(list_of_hex[i + 3])  # Gets the next 8 bits (25-32)
                    replaced_byte = utility.encode_bit_0(current_byte, message[message_counter])  # Replace bit 32
                    message_counter += 1
                    list_of_hex[i + 3] = utility.convert_bin_to_hex(replaced_byte)
                elif num_bits == 3:
                    ...
                elif num_bits == 4:
                    ...
                elif num_bits == 5:
                    ...
                elif num_bits == 6:  # Expect audio difference from here on out
                    ...
                elif num_bits == 7:  # Expect audio difference from here on out
                    ...
        else:
            break


def decode_message(list_of_hex: list, len_message: int, num_bits: int = 1):
    """Function takes in a list of hexadecimal values in byte format, length of the message (in terms of
    the number of characters) and number of bits per message (defaults to 1).
     Returns a list of binary values (1's and 0's)"""
    binary_list = []
    current_len = 0
    for i in range(len(list_of_hex)):
        if list_of_hex[i] == b'ff' and list_of_hex[i + 1][:1] == b'f':
            if current_len < len_message * 8:
                if num_bits == 1:
                    current_byte = utility.convert_byte_hex_to_bin(list_of_hex[i + 2])  # Gets the next 8 bits (17-24)
                    binary_list.append(current_byte[-1])  # inserts bit 24 into list
                    current_len += 1
                elif num_bits == 2:
                    current_byte = utility.convert_byte_hex_to_bin(list_of_hex[i + 2])  # Gets the next 8 bits (17-24)
                    binary_list.append(current_byte[-1])  # inserts bit 24 into list
                    current_len += 1
                    current_byte = utility.convert_byte_hex_to_bin(list_of_hex[i + 3])  # Gets the next 8 bits (25-32)
                    binary_list.append(current_byte[-1])  # inserts bit 32 into list
                    current_len += 1
                elif num_bits == 3:
                    ...
                elif num_bits == 4:
                    ...
                elif num_bits == 5:
                    ...
                elif num_bits == 6:
                    ...
                elif num_bits == 7:
                    ...
            else:
                break
    return binary_list

def text_from_bits(bits, encoding='utf8', errors='surrogatepass'):
    """Function takes in a string of bits.
    Parameters that can be overriden: Encoding and errors
    Returns text from a list of binary values (1's and 0's)"""
    n = int(bits, 2)
    return n.to_bytes((n.bit_length() + 7) // 8, 'big').decode(encoding, errors) or '\0'


def write_secret_to_file(filename: str, secret: str, num_bits: int = 1):
    """Function takes in the filename of mp3 file, secret to be encoded inside and number of bits to use.
    Writes the secret message into a new file called secret.mp3 which is created in the same directory"""

    encodelist = utility.read_file_into_hex_list(filename)
    encode_message(encodelist, secret, num_bits)
    with open('secret.mp3', 'wb') as filewrite:
        filewrite.write(binascii.unhexlify(b''.join(encodelist)))


def get_secret_from_file(filename: str, msg_len: int, num_bits: int = 1):
    """Function takes in the filename of the mp3 file which has a secret inside, the length of the secret message
    (in number of characters) and number of bits to use. Returns the decoded secret"""

    decodelist = utility.read_file_into_hex_list(filename)
    decoded_byte = decode_message(decodelist, msg_len, num_bits)
    decoded_byte_list = [decoded_byte[i:i + 8] for i in range(0, len(decoded_byte), 8)]
    char_list = []
    for i in decoded_byte_list:
        try:
            char_list.append(text_from_bits(''.join(i)))
        except UnicodeDecodeError:
            break
    return ''.join(char_list)

utility.py:

import binascii


def encode_bit_0(bin_str: str, bit: str):
    """Function takes in the byte (in terms of binary) to manipulate and a bit to encode (1 or 0). Changes bit 0.
     Returns the encoded value as a string """
    replaced_byte = bin_str[:-1] + bit  # Replaces bit 0 aka LSB
    return replaced_byte  # padding to ensure 2 digits


def encode_bit_1(bin_str: str, bit: str):
    """Function takes in the byte (in terms of binary) to manipulate and a bit to encode (1 or 0). Changes bit 1.
     Returns the encoded value as a string """
    replaced_byte = bin_str[:-2] + bit + bin_str[-1]  # Encodes bit 1
    return replaced_byte

...

def convert_bin_to_hex(bin_str: str):
    """Takes in a binary string (string of 1's and 0's) and formats it into the hexadecimal value"""
    return bytes(f'{int(bin_str, 2):x}', 'utf8').zfill(2)  # padding to ensure 2 digits


def convert_byte_hex_to_bin(hex_str: bytes):
    """Takes in a hex string in byte form and formats it into a binary string"""
    return bin(int(hex_str, 16))[2:].zfill(8)


def read_file_into_hex_list(file: str):
    with open(file, 'rb') as fileread:
        file_hex_str = binascii.hexlify(fileread.read())

    return [file_hex_str[i:i + 2] for i in range(0, len(file_hex_str), 2)]

main.py

import steg

num_bits = 1
msg = 'Test'

steg.write_secret_to_file('audio.mpeg', msg, num_bits)

print(steg.get_secret_from_file('secret.mp3', len(msg), num_bits))
byte file python
1 Answers

I think using binascii library is unnecessary and seems to overly complicate things.

On the mp3 I used `b'\xff\xfb' seemed good way to search for the header.

I used a list of booleans to store the values of the characters to be hidden in the file.

Here is the way I solved the problem:

from pathlib import Path
import re
from typing import List, Tuple

sync_word = b'\xff\xfb'


def max_msg_len(data: bytes) -> int:
    header_count = data.count(sync_word)
    return (header_count * 3) // 8


def get_header_locations(data: bytes) -> List[int]:
    result = [_.start() for _ in re.finditer(sync_word, data)]
    return result


def text_to_bits(txt: str) -> List[bool]:
    data = txt.encode()
    bin_str = ''.join([f"{x:08b}" for x in data])
    bits = [x == '1' for x in bin_str]
    return bits


def bits_to_text(bits: List[bool]) -> bytearray:
    data = bytearray()
    for x in range(0, (len(bits) // 8 * 8), 8):
        char = int(''.join(['1' if i else '0' for i in bits[x:x+8]]), 2)
        data.append(char)
    return data


def show_header(header: bytes) -> None:
    bin_str = ''.join([f"{x:08b}" for x in header])
    print("Show header:", bin_str)


def show_bits(header: bytes) -> None:
    sig_bits = extract_bits(header)
    print("Hidden bits", sig_bits)


def extract_bits(header: bytes) -> Tuple[bool, bool, bool]:
    bit_24 = get_bit(header[2], 0)  # 24
    bit_29 = get_bit(header[3], 3)  # 29
    bit_30 = get_bit(header[3], 2)  # 30
    return bit_24, bit_29, bit_30


def read_file(filename: Path) -> bytes:
    return filename.read_bytes()


def write_file(filename: Path, data: bytes) -> None:
    filename.write_bytes(data)


def extract_message(data: bytes) -> str:
    header_locs = get_header_locations(data)
    extracted_bits = []
    for idx in header_locs:
        header = data[idx:idx + 4]
        found_bits = extract_bits(header)
        extracted_bits.extend(found_bits)
    words = bits_to_text(extracted_bits)
    return_str = ''.join([x for x in words.decode() if x.isprintable()])
    return return_str


def write_message(data: bytes, bits: List[bool]) -> bytes:
    data = bytearray(data)
    header_locs = get_header_locations(data)
    header_pointer = 0
    for bit_pos, bit_val in enumerate(bits):
        header_pos = bit_pos % 3
        idx = header_locs[header_pointer]
        header = data[idx:idx + 4]
        if header_pos == 0:
            header[2] = set_bit(header[2], 0, bit_val)  # 24
        elif header_pos == 1:
            header[3] = set_bit(header[3], 3, bit_val)  # 29
        elif header_pos == 2:
            header[3] = set_bit(header[3], 2, bit_val)  # 30
            header_pointer += 1
        data[idx:idx + 4] = header
    return data


def set_bit(word, bit, value):
    mask = 1 << bit
    word &= ~mask
    if value:
        word |= mask
    return word


def get_bit(word, bit):
    value = word >> bit
    value &= 1
    return value


def main(mp3_in: Path, mp3_out: Path, msg: str) -> None:
    print('Message to hide:', msg)
    result = text_to_bits(msg)
    # Read mp3 file
    mp3_content = read_file(mp3_in)
    # Check message will fit in mp3 file
    if len(msg) > max_msg_len(mp3_content):
        print(f"Message too long for mp3 file. "
              f"Can only have {max_msg_len(mp3_content)} characters")
        exit()
    # Modify data
    new_data = write_message(mp3_content, result)
    # Check message can be extracted
    extracted_msg = extract_message(new_data)
    print("Message in new mp3:", extracted_msg)
    # Save to file
    write_file(mp3_out, new_data)
    # Read from secret file
    mp3_content = read_file(mp3_out)
    extracted_msg = extract_message(mp3_content)
    print(f"From {mp3_out.name} read: {extracted_msg}")


if __name__ == '__main__':
    # Files
    original_mp3 = Path.home().joinpath('Desktop', 'warning.mp3')
    secret_mp3 = Path.home().joinpath('Desktop', 'secret.mp3')
    # Message
    txt_to_hide = 'short test message'
    main(original_mp3, secret_mp3, txt_to_hide)

Which gave the following output:

Message to hide: short test message
Message in new mp3: short test message
From secret.mp3 read: short test message
ukBaz0
Related