Expr [Binary]

Solve for flag checker that uses multithreading for flag check routine.

Challenge Description

(Lost it :P)

expr_dist.zip

Investigation

This program spawns over 100 threads to run different subroutines that does some condition checking. Doing this by hand is just pain, so I wanted to use Angr to solve this symbolicly. Since Angr doesn’t have multithreading support, I’ll export the decompilation and override the pthread_create to run the functions sequentially instaed of in threads.

All threads have to pass their respective conditions (and release their mutex) to pass the flag check as a whole.

Solution

I exported the program in Ghidra by File > Export Program....

I cleaned up the code by removing unused definitions and redefinitions. The overriding definition of pthread_create is

/*
 * Usage:
 *   iVar1 = pthread_create(&pStack_2b0,(ulong *)0x0,
 *                          FUN_00101d60,pcVar2);
 */
int pthread_create(pthread_t *thread, ulong *attr, undefined8 (*func_ptr)(long), char *arg) {
    return (*func_ptr)((long) arg);
}

And the subroutine functions are changed by search and replace:

undefined8 FUN_001011d0(long param_1)

{
  uint uVar1;
  
  // pthread_mutex_lock((pthread_mutex_t *)&DAT_0010c070);
  uVar1 = *(uint *)(param_1 + 0x1b) >> 2 & 0x7ff;
  if (((uVar1 >> 2 ^ 0x557) + (uVar1 * 0x40 + 0x10c ^ (*(uint *)(param_1 + 0x14) & 0x7ff) >> 6) &
      0x7ff) == 0x79d) {
    // pthread_mutex_unlock((pthread_mutex_t *)&DAT_0010c070);
    return 0;
  }
  // return 0;
  return 1;
}

Comments show the original code that was removed.

Finally, we just run the good-ol Angr script from angr_ctf.

Final Script

import angr
import claripy
import sys

def main(argv):
  path_to_binary = argv[1]
  project = angr.Project(path_to_binary)

  initial_state = project.factory.entry_state(
    add_options = { angr.options.SYMBOL_FILL_UNCONSTRAINED_MEMORY,
                    angr.options.SYMBOL_FILL_UNCONSTRAINED_REGISTERS}
  )

  simulation = project.factory.simgr(initial_state)

  def is_successful(state):
    stdout_output = state.posix.dumps(sys.stdout.fileno())
    return 'pass'.encode() in stdout_output

  def should_abort(state):
    stdout_output = state.posix.dumps(sys.stdout.fileno())
    return 'failed'.encode() in stdout_output

  simulation.explore(find=is_successful, avoid=should_abort)

  if simulation.found:
    solution_state = simulation.found[0]
    print(solution_state.posix.dumps(sys.stdin.fileno()).decode())
  else:
    raise Exception('Could not find the solution')

if __name__ == '__main__':
  main(sys.argv)