CTF Writeup - Malware and snapshots

We got infected by a malware, it was writing and deleting files on our server, luckily we had a service that snapshotted the filesystem continuosly.

We need to know what was the malware doing, can you help us?

Challenge authors: ACN, Giotino

Table of contents
Preface
Challenge
malware.py
- dump_key(path, key, length=16)
Solving
Attachments

Preface

This challenge was solved around 4am, so bear with me if the thought process is a bit scattered.

Note: Most if not all zfs commands require root privileges, so I had to use sudo for most of the commands.

Challenge

The challenge provides a zip file containing a series of snapshots of the zfs file system, ordered sequentially.

snapshots.zip
├── snap-0-.zfs
├── snap-0-to-1.zfs
├── snap-1-to-2.zfs
├── ...

My first task was figuring out a way how to open these. First I tried on WSL, then on bare Windows, then gave up and tried on macOS. So I went ahead and installed the openzfs cask from homebrew.

After doing some doc-reading, I found that we can do so by creating a zpool, and then “receiving” the snapshots into the pool. For some weird reason, file redirection wasn’t working, and I wasted a lot of time on this.

sudo su -
truncate -s 1G /tmp/mypool
zpool create mypool /tmp/mypool
cat snap-0-.zfs | zfs receive -F mypool

After executing the above, we get Volume mypool on disk4s1 unmounted and thus we may start looking at the contents of the pool. Every snapshot has a similar structure to this:

/Volumes/mypool
├── challenge/
    ├── 11484.txt
    ├── malware.py
    ├── test_file_write.txt

malware.py

Immediately, malware.py looks suspicious. Let’s take a look at it.

#!/usr/bin/env python
# coding: utf-8

import base64
from Crypto.Cipher import AES
from datetime import datetime
from datetime import timezone
import json
import os
import re
import string
import random
import time


def test_file_write(path):
    test_string = "THIS IS A TEST\n"
    with open(os.path.join(path,"test_file_write.txt"),"w") as f:
        f.write(test_string)

    with open(os.path.join(path,"test_file_write.txt")) as f:
        data = f.read()

    if data == test_string:
        return True
    else:
        return False

def write_file(path, data):
    with open(path,"w") as f:
        f.write(data)
    time.sleep(1)
    os.system(f'sudo rm -f {path}')

def dump_key(path, key, lenght=16):
    now = datetime.now(timezone.utc)
    seed = int((now - datetime(1970, 1, 1, tzinfo=timezone.utc)).total_seconds())
    random.seed(seed)

    num_files = random.randint(100,200)

    for item in range(0, num_files):
        r = random.randint(0,65535)
        s = ''.join(random.choices(string.ascii_lowercase, k=lenght)).encode("ascii")
        write_file(os.path.join(path,str(r)+".txt"), base64.b64encode(s).decode('ascii'))

    r = random.randint(0,65535)
    key_filename = os.path.join(path,str(r)+".txt")
    nonce = ''.join(random.choices(string.ascii_lowercase, k=lenght)).encode("ascii")
    xored_key = bytes(a ^ b for a, b in zip(nonce, key))
    write_file(key_filename, base64.b64encode(xored_key).decode('ascii'))

    num_files = random.randint(100,200)

    for item in range(0, num_files):
        r = random.randint(0,65535)
        s = ''.join(random.choices(string.ascii_lowercase, k=lenght)).encode("ascii")
        write_file(os.path.join(path,str(r)+".txt"), base64.b64encode(s).decode('ascii'))

    return key_filename, nonce

def create_config(url):
    obj = {"url":url}
    return json.dumps(obj)

def encrypt_data(key, data):
    cipher = AES.new(key, AES.MODE_EAX)
    ciphertext = cipher.encrypt(data)
    return ciphertext, cipher.nonce


if __name__ == "__main__":
    url = "REDACTED"
    path = "/pool/challenge"
    if not os.path.exists(path):
        os.makedirs(path)

    if test_file_write(path):
        time.sleep(random.randint(1,10))
        key = ''.join(random.choices(string.ascii_lowercase, k=16)).encode('ascii')
        key_file, nonce = dump_key(path, key)
        config = create_config(url)
        chiphertext, nonce = encrypt_data(key, config.encode('ascii'))
        with open(os.path.join(path,'configuration.encrypted'),"wb") as f:
            f.write(nonce)
            f.write(chiphertext)
        print(key_file)
        print(nonce)
    else:
        print("Cannot write to directory")

There’s a lot going on here, let’s unpack it slowly. Looking at the main function, I saw that

we have some sort of url - this is likely the flag
the /challenge path is initialized
test_file_write.txt is written to the path
randomly wait 1-10s
a random 16-byte key is generated
dump_key is called, returning key_filename and nonce (we will look at this later)
the config object is created containing the url
the config object is encrypted, returning the ciphertext and the cipher.nonce (THIS IS NOT THE SAME AS THE nonce FROM dump_key, IT IS OVERWRITTEN)
the cipher.nonce and the ciphertext is written to configuration.encrypted (most likely will be in one of the last snapshots)

dump_key(path, key, length=16)

This function seems to be what generated the differences in the snapshots.

A seed is generated from the current time

my brain at 2am: can we find seed hmmm we have file system snapshots, ooga boga this is police forensic imaging bs AAAAA they surely must have hidden the time which they created the images, exiftool that shit, no wait, do it in python

create random number (100-200) of files with random names and random fake base64 encoded keys
create a random file name, create a nonce, xor the key with the nonce, store b64 encoded xored key
create more random files with random names and random keys
return xor nonce and file name

It’s clear what we need to do: find the right xored key file, recover the nonce, xor to retrieve the original key, decrypt the configuration.encrypted, profit.

Solving

If we find the seed, the snapshot that contains the configuration.encrypted file, and the snapshot that contains the file with the xored key, we can regenerate the nonce, get the key back, and decrypt the config object.

I wrote a script to automate the entire challenge. Let’s walk slowly through it. The snapshots are in ./snapshots.

Here, I will just sort the snapshots, so they’re processed in the order they were generated, i.e. snap-0-, snap-0-to-1, snap-1-to-2, and not lexicographically.

import base64
import os
import random
import string

import Crypto.Cipher.AES as AES

# List and naively sort snapshots (not order we want)
snapshots = sorted(os.listdir("./snapshots"))
# Separate base snapshot
base, snapshots = snapshots[0], snapshots[1:]
# Sort remaining snapshots by extracted integer value
snapshots.sort(key=lambda x: int(x.split("snap-")[1].split("-to-")[0]))
snapshots = [base] + snapshots

Next, I just made sure my dump folder is clean, and the pool I’m working in isn’t polluted.

print("Cleaning up the dumps folder...")
if not os.path.exists("./dumps"):
    os.mkdir("./dumps")
else:
    os.system("rm -rf ./dumps/*")
os.system("sudo zfs destroy -r mypool")

Next up, I just iterate over every snapshot, load it, copy over the files to MY OWN file system in the ./dumps folder for further analysis.

for snapshot in snapshots:
    snap_name = snapshot.split(".")[0].split("snap-")[1]
    print(f"Snapshot {snapshot} is being processed...")
    os.system(
        f"""
            cat snapshots/{snapshot} | sudo zfs receive -F mypool \
            && sleep 0.1 \
            && cp -r /Volumes/mypool/challenge ./dumps/snap-{snap_name}
        """
    )

Ok, it’s GO TIME. We need to find the seed. The first thing that happens after the seed is set, is the generation of the random number used for the filename of the first decoy xored key file. We note that in .dumps/snap-0-/ there is 11484.txt. So we can try going “back in time”, starting from date of creation/last access of this snapshot, and generate a number every time until 11484 is found.

earliest_atime = int(os.path.getatime(f"snapshots/{snapshots[0]}"))
for i in range(1000):
    random.seed(earliest_atime - i)
    num_files = random.randint(100, 200)
    r = random.randint(0, 65535)
    if r == 11484:  # First random name
        print("Found seed:", earliest_atime - i)
        seed = earliest_atime - i

Now we can copy over the relevant part of the malware script, and generate all the meaningful values that we need, which are:

Variable	Description
`num_files`	The number of random iterations before the real xored key file is generated
`key_filename`	By the above, we actually find the `abcxyz.txt` that contains the xored key
`r`	The one outside of the for loop, the actual file name of the xored key file (11484)
`nonce`	The random bytes used to xor the key with

Let’s do this.

random.seed(seed) # omg we found the seed

num_files = random.randint(100, 200) # number of files before the key file
for i in range(0, num_files):
    random.randint(0, 65535)
    "".join(random.choices(string.ascii_lowercase, k=16)).encode("ascii")

r = random.randint(0, 65535) # the actual file name
key_filename = os.path.join(str(r) + ".txt")
nonce = "".join(random.choices(string.ascii_lowercase, k=16)).encode("ascii")

Done. I recovered the original key by xoring, \(\underbrace{(\text{key} \oplus \text{nonce})}_{\text{the xored key in the file}} \oplus \text{nonce} = \text{key}.\)

for dump in os.listdir("./dumps"):
    for file in os.listdir(f"./dumps/{dump}"):
        if file == key_filename:
            print(f"Found key file in {dump} dump...")
            b64_xored_key = base64.b64decode(open(f"./dumps/{dump}/{file}", "rb").read())
            break

key = bytes(a ^ b for a, b in zip(nonce, b64_xored_key))

Then I opened the configuration.encrypted file as binary

AND REALIZE that the cipher nonce is not the same as the one used to xor the key, totally didn’t spend 30 mins debugging this before i found out i had both

and parsed it, splitting the first 16 bytes as the nonce, and the rest as the ciphertext.

encrypted_file = os.path.join("dumps", "snap-243-to-244", "configuration.encrypted")
encrypted_data = open(encrypted_file, "rb").read()
cipher_nonce, ciphertext = encrypted_data[:16], encrypted_data[16:]

print("Decrypting the configuration...")
cipher = AES.new(key, AES.MODE_EAX, nonce=cipher_nonce)
config = cipher.decrypt(ciphertext)
print(f"Decrypted configuration: {config}")

Profit.

Attachments

solve.py - the solution script

Table of contents