Pwn进你的心

from __future__ import print_function
from triton import *

code = [ # 每一项的结构是 (指令的地址， 指令的字节码)
    (0x400000, b"\x48\x8b\x05\xb8\x13\x00\x00"), # mov        rax, QWORD PTR [rip+0x13b8]
    (0x400007, b"\x48\x8d\x34\xc3"),             # lea        rsi, [rbx+rax*8]
    (0x40000b, b"\x67\x48\x8D\x74\xC3\x0A"),     # lea        rsi, [ebx+eax*8+0xa]
    (0x400011, b"\x66\x0F\xD7\xD1"),             # pmovmskb   edx, xmm1
    (0x400015, b"\x89\xd0"),                     # mov        eax, edx
    (0x400017, b"\x80\xf4\x99"),                 # xor        ah, 0x99
    (0x40001a, b"\xC5\xFD\x6F\xCA"),             # vmovdqa    ymm1, ymm2
]

if __name__ == '__main__':
    ctx = TritonContext()
    ctx.setArchitecture(ARCH.X86_64) # 设置模拟执行的代码架构

    for (addr, opcode) in code:
        inst = Instruction()   # 新建一个指令对象
        inst.setOpcode(opcode) # 传递字节码
        inst.setAddress(addr)  # 传递指令的地址
        ctx.processing(inst)   # 执行指令

        print(inst)	# 打印指令的信息
        print('    ---------------')
        print('    Is memory read :', inst.isMemoryRead())
        print('    Is memory write:', inst.isMemoryWrite())
        print('    ---------------')
        for op in inst.getOperands():
            print('    Operand:', op)
            if op.getType() == OPERAND.MEM:
                print('    - segment :', op.getSegmentRegister())
                print('    - base    :', op.getBaseRegister())
                print('    - index   :', op.getIndexRegister())
                print('    - scale   :', op.getScale())
                print('    - disp    :', op.getDisplacement())
            print('    ---------------')
        print()

from triton import *

def taint_analysis2(start, end):
    ctx = TritonContext()
    with open('./test', 'rb') as f:
        bin1 = f.read()

    ctx.setArchitecture(ARCH.X86_64)
    ctx.setConcreteMemoryAreaValue(0x400000, bin1) # 将二进制文件加载到0x400000处(这个地址由IDA分析得出,是程序的基地址)
    RBP_ADDR = 0x60000000
    RSP_ADDR = RBP_ADDR - 0x20000000
    INPUT_ADDR = 0x10000000

    ctx.setConcreteRegisterValue(ctx.registers.rip, start) 	  # 设置rip寄存器
    ctx.setConcreteRegisterValue(ctx.registers.rsp, RSP_ADDR) # 设置rsp寄存器
    ctx.setConcreteRegisterValue(ctx.registers.rbp, RBP_ADDR) # 设置rbp寄存器

    input = b"yhellow\x00"
    ctx.setConcreteMemoryAreaValue(INPUT_ADDR, input) # 将字符串加载到INPUT_ADDR
    pc = start
    
    while pc:
        inst = Instruction()
        opcode = ctx.getConcreteMemoryAreaValue(pc, 16) # 读取opcode
        inst.setOpcode(opcode)
        inst.setAddress(pc)
        ctx.processing(inst)
        print(str(inst))

        pc = ctx.getConcreteRegisterValue(ctx.registers.rip) # 读取rip寄存器

if __name__ == '__main__':
    taint_analysis2(0x401CD5, 0x401D74) # main的地址范围(在IDA中查看)

from triton import *

def taint_analysis2(start, end):
    ctx = TritonContext()
    with open('./test', 'rb') as f:
        bin1 = f.read()

    ctx.setArchitecture(ARCH.X86_64)
    ctx.setConcreteMemoryAreaValue(0x400000, bin1)
    RBP_ADDR = 0x7ffffffde000
    RSP_ADDR = RBP_ADDR - 0x21000
    INPUT_ADDR = 0x4C2290

    ctx.setConcreteRegisterValue(ctx.registers.rip, start)
    ctx.setConcreteRegisterValue(ctx.registers.rsp, RSP_ADDR)
    ctx.setConcreteRegisterValue(ctx.registers.rbp, RBP_ADDR)

    input = b"./flag\x00"
    ctx.setConcreteMemoryAreaValue(INPUT_ADDR, input)
    
    pc = start
    nop_addrs = []

    while pc:
        inst = Instruction()
        opcode = ctx.getConcreteMemoryAreaValue(pc, 16)
        
        inst.setOpcode(opcode)
        inst.setAddress(pc)
        ctx.processing(inst)
    
        print(str(inst))
        if pc == 0x401cf5:
            print("--------------")
            print("taint target: "+hex(ctx.getConcreteRegisterValue(ctx.registers.rdi))) # 获取寄存器的数据
            print("--------------")
            ctx.taintRegister(ctx.registers.rdi) # 将rdi中的地址数据标记为污点

        if inst.isTainted(): # 检测该指令是否被污染
            nop_addrs.append(hex(pc))
            print("--------------")
            if inst.isMemoryRead():
                for op in inst.getOperands():
                    if op.getType() == OPERAND.MEM:
                        print("read:0x{:08x}, size:{}".format(
                            op.getAddress(), op.getSize()))

            if inst.isMemoryWrite():
                for op in inst.getOperands():
                    if op.getType() == OPERAND.MEM:
                        print("write:0x{:08x}, size:{}".format(
                            op.getAddress(), op.getSize()))
                    
        pc = ctx.getConcreteRegisterValue(ctx.registers.rip)

    print(nop_addrs)

if __name__ == '__main__':
    taint_analysis2(0x401CD5, 0x401D3F)

from triton import *
import string
import os
import time
import lief

BASE_GOT   = 0x10000000
BASE_ARGV  = 0x20000000
BASE_STACK = 0x7ffffffde000

heap_addr = 0x555555559000
heap_size = 0
is_free   = False

def hookingHandler(ctx):
    pc = ctx.getConcreteRegisterValue(ctx.registers.rip)
    for rel in customRelocation:
        if rel[2] == pc:
            ret_value = rel[1](ctx) # 调用hook函数,并获取其返回值
            if ret_value is not None:
                ctx.setConcreteRegisterValue(ctx.registers.rax, ret_value) # 返回值将会被设置到ctx.registers.rax中
            ret_addr = ctx.getConcreteMemoryValue(MemoryAccess(ctx.getConcreteRegisterValue(ctx.registers.rsp), CPUSIZE.QWORD)) # 返回地址位于栈顶
            ctx.setConcreteRegisterValue(ctx.registers.rip, ret_addr) # 设置rip
            ctx.setConcreteRegisterValue(ctx.registers.rsp, ctx.getConcreteRegisterValue(ctx.registers.rsp)+CPUSIZE.QWORD) # 调整rsp
    return

def mymalloc(ctx):
    global heap_size 
    print('[+] malloc hooked')
    heap_size = ctx.getConcreteRegisterValue(ctx.registers.rdi)
    return heap_addr

def myfree(ctx):
    global is_free
    print('[+] free hooked')
    if is_free:
        print('[-] free BUG')
    is_free = True
    return 0

def mylibc(ctx):
    print('[+] __libc_start_main hooked')
    return 0

def myopen(ctx):
    print("[+] open hooked")
    arg1 = ctx.getConcreteRegisterValue(ctx.registers.rdi)
    arg2 = ctx.getConcreteRegisterValue(ctx.registers.rsi)
    return 0

def myread(ctx):
    global heap_size
    print("[+] read hooked")
    read_size = ctx.getConcreteRegisterValue(ctx.registers.rdx)
    if read_size > heap_size:
        print("[-] read BUG")
    return 0

def mywrite(ctx):
    global heap_size
    print("[+] write hooked")
    read_size = ctx.getConcreteRegisterValue(ctx.registers.rdx)
    if read_size > heap_size:
        print("[-] write BUG")
    return 0

def myprintf(ctx):
    print("[+] printf hooked")
    return 0

customRelocation = [
    ['__libc_start_main',   mylibc,   None],
    ['malloc',              mymalloc, None],
    ['free',                myfree,   None],
    ['open',                myopen,   None],
    ['read',                myread,   None],
    ['write',               mywrite,  None],
    ['printf',              myprintf, None],
]

def makeRelocation(ctx, binary):
    for pltIndex in range(len(customRelocation)):
        customRelocation[pltIndex][2] = BASE_GOT + pltIndex # 设置自定义got地址

    relocations = [x for x in binary.pltgot_relocations] # 读取pltgot重定位符号
    relocations.extend([x for x in binary.dynamic_relocations]) # 读取dynamic重定位符号

    for rel in relocations:
        symbolName = rel.symbol.name
        symbolRelo = rel.address
        for crel in customRelocation:
            if symbolName == crel[0]:
                print('Hooking %-10s:0x%x' %(symbolName,symbolRelo))
                ctx.setConcreteMemoryValue(MemoryAccess(symbolRelo, CPUSIZE.QWORD), crel[2]) # 将模拟执行的got表修改为基于BASE_GOT的自定义got表
                break
    return

def loadBinary(ctx, binary):
    phdrs = binary.segments # 读取所有segment
    for phdr in phdrs:
        size   = phdr.physical_size
        vaddr  = phdr.virtual_address
        print('[+] Loading 0x%06x - 0x%06x' %(vaddr, vaddr+size))
        ctx.setConcreteMemoryAreaValue(vaddr, list(phdr.content))
    return


if __name__ == '__main__':
    ctx = TritonContext(ARCH.X86_64)
    ctx.setMode(MODE.ALIGNED_MEMORY, True)
    ctx.setMode(MODE.CONSTANT_FOLDING, True)

    binary = lief.ELF.parse("./test") 
    
    loadBinary(ctx, binary)
    makeRelocation(ctx, binary)
    ctx.setConcreteRegisterValue(ctx.registers.rbp, BASE_STACK)
    ctx.setConcreteRegisterValue(ctx.registers.rsp, BASE_STACK)

    pc = 0x11E9
    while pc:
        inst = Instruction()
        opcode = ctx.getConcreteMemoryAreaValue(pc, 16)
        
        inst.setOpcode(opcode)
        inst.setAddress(pc)
        ctx.processing(inst)
        print(str(inst))

        hookingHandler(ctx) # 处理hook
        pc = ctx.getConcreteRegisterValue(ctx.registers.rip)

        if pc == 0x11ff:
            print("[+] rax: "+hex(ctx.getConcreteRegisterValue(ctx.registers.rax)))

from triton import *
import lief
import os
import time

def AnalysisBinary(path):
    binary = lief.ELF.parse(path) # 使用lief加载二进制文件

    sections = binary.sections
    for section in sections: # 读取所有section
        name = section.name
        size = section.size
        vaddr = section.virtual_address
        if name != "": 
            print('[+] %-28s 0x%06x - 0x%06x' % (name+":", vaddr, vaddr+size))
            data = bytes(section.content[0:0x20])
            for i in range(len(data)):
                print(f"{data[i]:02x}", end=" ")
                if i % 0x10 == 0xf: print()
            if i % 0x10 != 0xf: print()
            print("-------------------------------------------------")
            ctx.setConcreteMemoryAreaValue(vaddr, bytes(section.content)) # 加载所有section

    symbols = binary.symbols # 读取所有symbol
    for symbol in symbols:
        if symbol.name == 'main':
            main_address = symbol.value
            print()
            print('[+] Address of main function: 0x{:x}'.format(main_address))
            print()
            break
    else:
        print("'main' function not found")

if __name__ == '__main__':
    ctx = TritonContext()
    ctx.setArchitecture(ARCH.X86_64)
    ast = ctx.getAstContext()

    AnalysisBinary(os.path.join(os.path.dirname(__file__), './test'))

    start = 0x401DE0 # 执行关键匹配函数之前的某个地址(从IDA中得出)
    RBP_ADDR = 0x7ffffffde000
    RSP_ADDR = RBP_ADDR - 0x21000

    ctx.setAstRepresentationMode(AST_REPRESENTATION.PYTHON)

    ctx.setConcreteRegisterValue(ctx.registers.rip, start)
    ctx.setConcreteRegisterValue(ctx.registers.rsp, RSP_ADDR)
    ctx.setConcreteRegisterValue(ctx.registers.rbp, RBP_ADDR)

    for i in range(5):
        input_addr = ctx.getConcreteRegisterValue(ctx.registers.rbp) - 0x20
        ctx.setConcreteMemoryValue(MemoryAccess(input_addr + i, CPUSIZE.BYTE), ord('a'))
        ctx.symbolizeMemory(MemoryAccess(input_addr + i, CPUSIZE.BYTE)) # 将目标地址的数据设置为符号变量

    pc = start

    while pc:
        inst = Instruction()
        opcode = ctx.getConcreteMemoryAreaValue(pc, 16)
        
        inst.setOpcode(opcode)
        inst.setAddress(pc)
        ctx.processing(inst)
    
        print(str(inst))
        if inst.getAddress() == 0x401DEC:
            rdata = ctx.getRegisterAst(ctx.registers.rax) # 获取rax寄存器的AST结构,可以添加到后续的约束条件中
            
            sv1 = ast.variable(ctx.getSymbolicVariable(0)) # 读取设置的符号变量
            sv2 = ast.variable(ctx.getSymbolicVariable(1))
            sv3 = ast.variable(ctx.getSymbolicVariable(2))
            sv4 = ast.variable(ctx.getSymbolicVariable(3))
            sv5 = ast.variable(ctx.getSymbolicVariable(4))

            cstr = ast.land([rdata == 0xAD6D] # 添加约束条件
            + [sv1 >= ord(b'A'), sv1 <= ord(b'z')]
            + [sv2 >= ord(b'A'), sv2 <= ord(b'z')]
            + [sv3 >= ord(b'A'), sv3 <= ord(b'z')]
            + [sv4 >= ord(b'A'), sv4 <= ord(b'z')]
            + [sv5 >= ord(b'A'), sv5 <= ord(b'z')]
            )

            model = ctx.getModel(cstr) # 进行约束求解
            answer = ""
            for k, v in sorted(model.items()):
                value = v.getValue()
                answer += chr(value)
                
            if len(answer)==5:
                print("answer: {}".format(answer))  

            break 

        pc = ctx.getConcreteRegisterValue(ctx.registers.rip)

from triton import *
import lief
import sys

def AnalysisBinary(path):
    binary = lief.ELF.parse(path)
    sections = binary.sections
    for section in sections:
        name = section.name
        size = section.size
        vaddr = section.virtual_address
        if name != "": 
            print('[+] %-28s 0x%06x - 0x%06x' % (name+":", vaddr, vaddr+size))
            data = bytes(section.content[0:0x20])
            for i in range(len(data)):
                print(f"{data[i]:02x}", end=" ")
                if i % 0x10 == 0xf: print()
            if i % 0x10 != 0xf: print()
            print("-------------------------------------------------")
            ctx.setConcreteMemoryAreaValue(vaddr, bytes(section.content))

if __name__ == '__main__':
    ctx = TritonContext(ARCH.X86_64)
    ctx.setMode(MODE.ALIGNED_MEMORY, True)
    ctx.setMode(MODE.ONLY_ON_SYMBOLIZED, True)

    AnalysisBinary("./test")

    start = 0x400c4b
    end   = 0x400C7C
    input = 0x002000 # 输入字符串的地址
    for i in range(31):
        ctx.setConcreteMemoryValue(MemoryAccess(input + i, CPUSIZE.BYTE), 0x61)
        ctx.symbolizeMemory(MemoryAccess(input + i, CPUSIZE.BYTE)) # 将输入字符串设置为符号变量

    rbp = 0x7fffffffe460
    ctx.setConcreteRegisterValue(ctx.registers.rbp, rbp)
    ctx.setConcreteRegisterValue(ctx.registers.rip, start)
    ctx.setConcreteMemoryValue(MemoryAccess(rbp - 96, CPUSIZE.QWORD), input) # 初始化"rbp - 96"为输入字符串的地址 
    ctx.setConcreteMemoryValue(MemoryAccess(rbp - 20, CPUSIZE.DWORD), 0) # 初始化"rbp - 20"为"0"(这里是解密输入值时使用的索引)

    for count in range(31):
        pc = start

        while pc: # 进行模拟执行,直到遇到约束条件
            inst = Instruction()
            opcode = ctx.getConcreteMemoryAreaValue(pc, 16)
            inst.setOpcode(opcode)
            inst.setAddress(pc)
            ctx.processing(inst)
            pc = ctx.getConcreteRegisterValue(ctx.registers.rip)
            print(inst)
            if pc == end:
                print("------------------------------------------")
                break

        zf  = ctx.getRegisterAst(ctx.registers.zf)
        ctx.pushPathConstraint(zf == 1) # 添加约束条件

        ctx.setConcreteMemoryValue(MemoryAccess(rbp - 20, CPUSIZE.DWORD), count + 1) # 更新索引
        ctx.setConcreteMemoryValue(MemoryAccess(rbp - 96, CPUSIZE.DWORD), input + count + 1) # 更新输入值

	mod = ctx.getModel(ctx.getPathPredicate()) # 进行约束求解
    if not mod:
        print('[-] Failed')
        sys.exit(-1)    
        
    flag = "" 
    for k, v in sorted(mod.items()):
        ctx.setConcreteVariableValue(ctx.getSymbolicVariable(v.getId()), v.getValue())
        value = v.getValue()
        flag += chr(value)

    print("flag: {}".format(flag))  

from pwn import *
import time
from Exrop import Exrop

binname = "/lib/x86_64-linux-gnu/libc.so.6"
libc = ELF(binname, checksec=False)
binsh = next(libc.search(b"/bin/sh"))
open_libc = libc.symbols['open']
read_libc = libc.symbols['read']
write_libc = libc.symbols['write']
bss = libc.bss()

t = time.mktime(time.gmtime())
rop = Exrop(binname)
rop.find_gadgets(cache=True)

print("execve('/bin/sh', 0, 0)")
chain = rop.syscall(59, ("/bin/sh", 0, 0), bss) # 字符串会写到第3个参数上
chain.set_base_addr(0)
chain.dump()

print("execve('/bin/sh', 0, 0)")
chain = rop.syscall(59, (binsh, 0, 0)) # 如果没设置第3个参数,则不能直接写入字符串
chain.set_base_addr(0) # 设置基地址
print(chain.payload_str()) # 将ROP链转化为bytes字符串

print("open('/etc/passwd', 0)")
chain = rop.func_call(open_libc, ("/etc/passwd", 0), bss)
chain.set_base_addr(0)
chain.dump()

print("read('rax', bss, 0x100)") 
chain = rop.func_call(read_libc, ('rax', bss, 0x100)) # 可以直接使用寄存器
chain.set_base_addr(0)
chain.dump()

print("write(1, bss, 0x100)")
chain = rop.func_call(write_libc, (1, bss, 0x100))
chain.set_base_addr(0)
chain.dump()

print("done in {}s".format(time.mktime(time.gmtime()) - t))

def parseRopGadget(filename, opt=""):
    from subprocess import Popen, PIPE, STDOUT
    import re

    cmd = ['ROPgadget', '--binary', filename, '--multibr', '--only',
            'pop|xchg|add|sub|xor|mov|ret|jmp|call|syscall|leave', '--dump']
    if opt:
        cmd.append(opt)
    process = Popen(cmd, stdout=PIPE, stderr=STDOUT)
    stdout, _ = process.communicate()
    ......

cmd = ['ROPgadget', '--binary', filename, '--multibr', '--only',
        'pop|mov|ret|call|syscall', '--dump']

import logging
import sys

from triton import ARCH

from qsynthesis import SimpleSymExec, TopDownSynthesizer, InputOutputOracleLevelDB
import qsynthesis
import logging
import pkg_resources
logging.basicConfig(level=logging.DEBUG)

qsynthesis.enable_logging()

RIP_ADDR = 0x40B160
RSP_ADDR = 0x800000

INSTRUCTIONS = [b'U', b'H\x89\xe5', b'H\x89}\xf8', b'H\x89u\xf0', b'H\x89U\xe8', b'H\x89M\xe0', b'L\x89E\xd8',
                b'H\x8bE\xf0', b'H#E\xe0', b'H\x89\xc2', b'H\x8bE\xf0', b'H\x0bE\xe0', b'H\x0f\xaf\xd0', b'H\x8bE\xe0',
                b'H\xf7\xd0', b'H#E\xf0', b'H\x89\xc1', b'H\x8bE\xf0', b'H\xf7\xd0', b'H#E\xe0', b'H\x0f\xaf\xc1',
                b'H\x01\xc2', b'H\x8bE\xe0', b'H\x0f\xaf\xc0', b'H\x89\xd6', b'H!\xc6', b'H\x8bE\xf0', b'H#E\xe0',
                b'H\x89\xc2', b'H\x8bE\xf0', b'H\x0bE\xe0', b'H\x0f\xaf\xd0', b'H\x8bE\xe0', b'H\xf7\xd0', b'H#E\xf0',
                b'H\x89\xc1', b'H\x8bE\xf0', b'H\xf7\xd0', b'H#E\xe0', b'H\x0f\xaf\xc1', b'H\x01\xc2', b'H\x8bE\xe0',
                b'H\x0f\xaf\xc0', b'H\t\xd0', b'H)\xc6', b'H\x89\xf0', b'H\x83\xe8\x01', b'H3E\xf0', b'H\x89\xc2',
                b'H\x8bE\xf0', b'H#E\xe0', b'H\x89\xc1', b'H\x8bE\xf0', b'H\x0bE\xe0', b'H\x0f\xaf\xc8', b'H\x8bE\xe0',
                b'H\xf7\xd0', b'H#E\xf0', b'H\x89\xc6', b'H\x8bE\xf0', b'H\xf7\xd0', b'H#E\xe0', b'H\x0f\xaf\xc6',
                b'H\x01\xc1', b'H\x8bE\xe0', b'H\x0f\xaf\xc0', b'H1\xc8', b'H#E\xf0', b'H\x01\xc0', b'H)\xc2',
                b'H\x89\xd0', b']', b'\xc3']

qsynthesis_version = pkg_resources.get_distribution("qsynthesis").version # 读取qsynthesis的版本
print(f"The version of qsynthesis is: {qsynthesis_version}")

def test(oracle_file):
    # Perform symbolic execution of the instructions
    symexec = SimpleSymExec(ARCH.X86_64) # 使用预期的体系结构对其进行初始化
    symexec.initialize_register('rip', RIP_ADDR) # 初始化寄存器
    symexec.initialize_register('rsp', RSP_ADDR)
    for opcode in INSTRUCTIONS:
       symexec.execute(opcode) # 执行给定的操作码
    rax = symexec.get_register_ast("rax") # 执行指令后检索rax AST

    # Load lookup tables
    ltm = InputOutputOracleLevelDB.load(oracle_file) # 加载查找表数据库

    # Perform Synthesis of the expression
    synthesizer = TopDownSynthesizer(ltm) # 实例化表的综合大小
    synt_rax, simp = synthesizer.synthesize(rax) # 触发rax表达式的合成

    # Print synthesis results
    print(f"simplified: {simp}")
    print(f"synthesized expression: {synt_rax.pp_str}")
    sz, nsz = rax.node_count, synt_rax.node_count
    print(f"size: {rax.node_count} -> {synt_rax.node_count}\nsize reduction:{((sz-nsz)*100)/sz:.2f}%")
    return symexec, rax, synt_rax

if __name__ == "__main__":
    sx, rax, srax = test("my_oracle_table/")