refactor: project -> settings and model/

This commit is contained in:
Dobin
2024-02-25 08:47:25 +00:00
parent b1dd3481ed
commit d61f358ae4
18 changed files with 185 additions and 157 deletions
+42
View File
@@ -0,0 +1,42 @@
from enum import Enum
import os
class FilePath(str):
pass
# with shellcodes/createfile.bin
VerifyFilename: FilePath = r'C:\Temp\a'
# Correlated with real template files
# in plugins/
class AllocStyle(Enum):
RWX = "rwx_1"
#RW_X = "rw_x"
#REUSE = "reuse"
class DecoderStyle(Enum):
PLAIN_1 = "plain_1"
XOR_1 = "xor_1"
class ExecStyle(Enum):
CALL = "direct_1"
#JMP = 2,
#FIBER = 3,
class DataRefStyle(Enum):
APPEND = 1
#class InjectStyle(Enum):
class SourceStyle(Enum):
peb_walk = 1
iat_reuse = 2
build_dir = "build"
main_c_file = os.path.join(build_dir, "main.c")
main_asm_file = os.path.join(build_dir, "main.asm")
main_exe_file = os.path.join(build_dir, "main.exe")
main_shc_file = os.path.join(build_dir, "main.bin")
+118
View File
@@ -0,0 +1,118 @@
from typing import Dict
import logging
import pefile
from model.defs import *
import pehelper
from peparser.misc import get_physical_address
logger = logging.getLogger("ExeHost")
class IatResolve():
def __init__(self, name: str, placeholder: bytes, addr: int):
self.name: str = name # Function Name, like "VirtualAlloc"
self.id: bytes = placeholder # Random bytes
self.addr: int = addr # The address of the IAT entry (incl. image_base)
def __str__(self) -> str:
return "0x{:X}: {} ({})".format(
self.addr,
self.name,
self.id
)
class ExeHost():
def __init__(self, filepath: FilePath):
self.filepath: FilePath = filepath
self.iat_resolves: Dict[str, IatResolve] = {}
self.image_base = 0
self.dynamic_base = False
self.code_virtaddr = 0
self.code_size = 0
self.code_section = None
self.iat = {}
self.base_relocs = []
self.rwx_section = None
self.ep = None
self.ep_raw = None
def add_iat_resolve(self, func_name, placeholder):
self.iat_resolves[func_name] = IatResolve(
func_name, placeholder, pehelper.get_addr_for(self.iat, func_name))
def init(self):
logger.info("--[ Analyzing: {}".format(self.filepath))
pe = pefile.PE(self.filepath)
if pe.FILE_HEADER.Machine != 0x8664:
raise Exception("Binary is not 64bit: {}".format(self.filepath))
self.ep = pe.OPTIONAL_HEADER.AddressOfEntryPoint
self.ep_raw = get_physical_address(pe, self.ep)
# image base
self.image_base = pe.OPTIONAL_HEADER.ImageBase
# dynamic base / ASLR
if pe.OPTIONAL_HEADER.DllCharacteristics & pefile.DLL_CHARACTERISTICS['IMAGE_DLLCHARACTERISTICS_DYNAMIC_BASE']:
self.dynamic_base = True
else:
self.dynamic_base = False
# .text virtual address
self.code_section = pehelper.get_code_section(pe)
self.code_virtaddr = self.code_section.VirtualAddress
self.code_size = self.code_section.Misc_VirtualSize
logger.info("---[ Injectable: Chosen code section: {} at 0x{:x} size: {}".format(
self.code_section.Name.decode().rstrip('\x00'),
self.code_virtaddr,
self.code_size))
# iat
self.iat = pehelper.extract_iat(pe)
# relocs
if hasattr(pe, 'DIRECTORY_ENTRY_BASERELOC'):
for base_reloc in pe.DIRECTORY_ENTRY_BASERELOC:
for entry in base_reloc.entries:
rva = entry.rva
base_rva = entry.base_rva
reloc_type = pefile.RELOCATION_TYPE[entry.type][0]
self.base_relocs.append({
'rva': rva,
'base_rva': base_rva,
'type': reloc_type,
})
# rwx
self.rwx_section = pehelper.get_rwx_section(pe)
def get_all_iat_resolvs(self) -> Dict[str, IatResolve]:
return self.iat_resolves
def has_all_functions(self, needs):
is_ok = True
for func_name in needs:
addr = pehelper.get_addr_for(self.iat, func_name)
if addr == 0:
logging.info("---( Function not available as import: {}".format(func_name))
is_ok = False
return is_ok
def print(self):
logger.info("--( Required IAT Resolves: ")
for _, cap in self.iat_resolves.items():
if cap.addr == 0:
logger.info(" {:28} {}".format(cap.name, "N/A"))
else:
logger.info(" {:28} 0x{:x}".format(cap.name, cap.addr))
+20
View File
@@ -0,0 +1,20 @@
import logging
from model import *
from model.defs import *
logger = logging.getLogger("Payload")
class Payload():
def __init__(self, filepath: FilePath):
self.payload_path: FilePath = filepath
self.payload_data: bytes = b""
self.len: int = 0
def init(self):
logging.info("--( Load payload: {}".format(self.payload_path))
with open(self.payload_path, 'rb') as f:
self.payload_data = f.read()
self.len = len(self.payload_data)
+21
View File
@@ -0,0 +1,21 @@
import logging
from model import *
from model.defs import *
from model.payload import Payload
from model.exehost import ExeHost
logger = logging.getLogger("Project")
class Project():
def __init__(self, settings):
self.settings = settings
self.payload = Payload(self.settings.payload_path)
self.exe_host = ExeHost(self.settings.inject_exe_in)
def init(self):
self.payload.init()
self.exe_host.init()
+32
View File
@@ -0,0 +1,32 @@
from model import *
from model.defs import *
class Settings():
def __init__(self):
self.payload_path: FilePath = ""
# Settings
self.source_style: SourceStyle = SourceStyle.peb_walk
self.alloc_style: AllocStyle = AllocStyle.RWX
self.exec_style: ExecStyle = ExecStyle.CALL
self.decoder_style: DecoderStyle = DecoderStyle.XOR_1
self.dataref_style: DataRefStyle = DataRefStyle.APPEND
self.short_call_patching: bool = False
# Injectable
self.inject: bool = False
self.inject_mode: int = 2
self.inject_exe_in: FilePath = ""
self.inject_exe_out: FilePath = ""
# Debug
self.show_command_output = False
self.verify: bool = False
self.try_start_loader_shellcode: bool = False
self.try_start_final_shellcode: bool = False
self.try_start_final_infected_exe: bool = False
self.cleanup_files_on_start: bool = True
self.cleanup_files_on_exit: bool = True
self.generate_asm_from_c: bool = True
self.generate_shc_from_asm: bool = True