[5주] 08장 퍼징

퍼징은 소프트웨어에 있는 버그를 찾아내는 가장 효과적인 기술중 하나로, 애플리케이션이 에러를 발생하게 만들기 위해 비정상적인 데이터를 만들어 애플리케이션에 전달하는 방법이다.

퍼저에는 제너레이션 퍼저와 뮤테이션 퍼저 두 가지가 있는데,

제너레이션 퍼저는 새로운 데이터를 생성해 전달하는 반면에 뮤테이션 퍼저는 기존의 데이터를 생성해 대상 애플리케이션에 전달한다.

효과적인 퍼저를 만들기 위해선 먼저 버그의 유형을 간단히 살펴볼 필요가 있다.

버그의 유형으로는 가장 흔한 형태의 소프트웨어 취약점인 버퍼오버플로우로 분류되는 스택기반의 오버플로우, 힙 기반의 오버플로우다. 또 흔하게 볼 수 있는 유형인 정수 오버플로우와 포맷스트링 공격도 있다.

파일 포맷 취약점을 이용하는 공격 벡터는 파일 포맷 파서 자체의 버그를 찾아내는 방향으로 관심을 기울여야 한다. 목적 달성을 위해서는 모든 종류의 다양한 파일 포맷에대한 변형을 만들 수 있어야 한다.

이제 퍼저를 구현해보자

#file_fuzzer.py
from pydbg import *
from pydbg.defines import *
 
import utils
import random
import sys
import struct
import threading
import os
import shutil
import time
import getopt
 
class file_fuzzer:
    def __init__(self, exe_path, ext, notify):
 
        self.exe_path             = exe_path
        self.ext                 = ext
        self.notify_crash         = notify
        self.orig_file             = None
        self.mutated_file         = None
        self.iteration             = 0
        self.exe_path             = exe_path
        self.orig_file             = None
        self.mutated_file         = None
        self.iteration             = 0
        self.crash                 = None
        self.send_notify         = False
        self.pid                 = None
        self.in_accessv_handler = False
        self.dbg                 = None
        self.running             = False
        self.ready                 = False
 
        #option
        self.smtpserver = 'mail.nostarch.com'
        self.recipients = ['jms@bughunter.ca',]
        self.sender = 'jms@bughunter.ca'
 
        self.test_cases = ["%s%n%s%n%s%n", "\xff", "\x00", "A"]
 
    def file_picker(self):
 
        file_list = os.listdir("examples/")
        list_length = len(file_list)
        file = file_list[random.randint(0, list_length - 1)]
        shutil.copy("examples\\%s" % file, "test.%s" % self.ext)
        
        return file
 
    def fuzz(self):
        while 1:
        if not self.running:
 
            self.test_file = self.file_picker()
            self.mutate_file()
 
            pydbg_thread = threading.Thread(target = self.start_debugger)
            pydbg_thread.setDaemon(0)
            pydbg_thread.start()
 
            while self.pid == None:
                time.sleep(1)
 
            monitor_thread = threading.Thread(target = self.monitor_debugger)
            monitor_thread.setDaemon(0)
            monitor_thread.start()
            self.iteration += 1
 
        else:
            time.sleep(1)
 
    def start_debugger(self):
 
        print "[*] Starting debugger for iteration: %d" % self.iteration
        self.running = True
        self.dbg = pydbg()
 
        self.dbg.set_callback(EXCEPTION_ACCESS_VIOLATION, self.check_accessv)
        pid = self.dbg.load(self.exe_path, "test.%s" % self.ext)
        
        self.pid = self.dbg.pid
        self.dbg.run()
 
    def check_accessv(self, dbg):
 
        if dbg.dbg.u.Exception.dwFirstChance:
            return DBG_CONTINUE
 
        print "[*] Woot! Handling an access violation!"
        self.in_accessv_handler = True
        crash_bin = utils.crash_binning.crash_binning()
        crash_bin.record_crash(dbg)
        self.crash = crash_bin.crash_synopsis()
 
        crash_fd = open("crashes\\crash-%d" % self.iteration, "w")
        crash_fd.write(self.crash)
 
        shutil.copy("test.%s" % self.ext, "crashes\\%d.%s" % 
                        (self.iteration, self.ext))
        shutil.copy("examples\\%s" % self.test_file, "crashes\\%d_orig.%s" % 
                        (self.iteration, self.ext))
        
        self.dbg.terminate_process()
        self.in_accessv_handler = False
        self.running = False
        
        return DBG_EXCEPTION_NOT_HANDLED
 
    def monitor_debugger(self):
        counter = 0
        print "[*] Monitor thread for pid: %d waiting." % self.pid,
        
        while counter < 3:
            time.sleep(1)
            print counter,
            counter += 1
        
        if self.in_accessv_handler != True:
            time.sleep(1)
            self.dbg.terminate_process()
            self.pid = None
            self.running = False
        else:
            print "[*] The access violation handler is doing its business. Waiting."
            while self.running:
                time.sleep(1)
 
    def notify(self):
 
        crash_message = "From:%s\r\n\r\nTo:\r\n\r\nIteration:%d\n\nOutput:\n\n %s" % 
                            (self.sender, self.iteration, self.crash)
 
        session = smtplib.SMTP(smtpserver)
        session.sendmail(sender, recipients, crash_message)
        session.quit()
        
        return
 
    def mutate_file(self):
 
        fd = open("test.%s" % self.ext, "rb")
        stream = fd.read()
        fd.close()
 
        test_case = self.test_cases[random.randint(0, len(self.test_cases) - 1)]
        stream_length = len(stream)
        rand_offset = random.randint(0, stream_length - 1)
        rand_len = random.randint(1, 1000)
 
        test_case = test_case * rand_len
 
        fuzz_file = stream[0:rand_offset]
        fuzz_file += str(test_case)
        fuzz_file += stream[rand_offset]
 
        fd = open("test.%s" % self.ext, "wb")
        fd.write(fuzz_file)
        fd.close()
 
        return
    def print_usage():
        
        print "[*]"
        print "[*] file_fuzzer.py -e  -x "
        print "[*]"
        
        sys.exit(0)
        
        if __name__ == "__main__":
            print "[*] Generic File Fuzzer."
 
        try:
            opts, argo = getopt.getopt(sys.argv[1:], "e:x:n")
        
        except getopt.GetoptError:
            
            print_usage()
 
        exe_path = None
        ext = None
        notify = False
 
        for o,a in opts:
        
            if o == "-e":
                exe_path = a
            elif o == "-x":
                ext = a
            elif o == "-n":
                notify = True
 
        if exe_path is not None and ext is not None:
            fuzzer = file_fuzzer(exe_path, ext, notify)
            fuzzer.fuzz()
        else:
            print_usage()