admin/SuperMega
1
0
Fork 0
mirror of https://github.com/dobin/SuperMega synced 2026-06-02 17:27:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: move directories around 1/2

Browse Source
This commit is contained in:
Dobin
2024-03-29 18:02:16 +00:00
parent 0296537ec5
commit a6dbbe69ac
14 changed files with 103 additions and 180 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+8 -17
View File
@@ -1,21 +1,12 @@
*.exe.injected
*-verify.exe
*.infected.exe
app/upload/*
data/exes_more/
*.obj
*.lnk
/*.bin
__pycache__
bak/
build/
out/
tools/
doc/
*.pickle
logs/
app/projects/*
data_orig/
app/upload_orig/
data/source/payload/
data/exes_more/
data/source/payload/
log-*
*.verify.exe
*.infected.exe
app/data.pickle
BIN
View File
Binary file not shown.
app/views.py
+10 -17
View File
@@ -124,7 +124,8 @@ def dev_build_route(name):
@views.route("/project/<name>")
def project(name):
project = storage.get_project(name)
log_files = get_logfiles()
project.settings.prep()
log_files = get_logfiles(project.settings.main_dir)
exes = []
for file in os.listdir(PATH_EXES):
@@ -153,7 +154,7 @@ def project(name):
injectstyles=injectstyles,
log_files=log_files,
)
)
@views.route("/project_add", methods=['POST', 'GET'])
@@ -235,16 +236,6 @@ def supermega_thread(project: Project):
start(project.settings)
thread_running = False
# copy generated file to project folder
file_basename = os.path.basename(project.settings.inject_exe_out)
project.project_exe = file_basename
dest = PATH_WEB_PROJECT + "{}/{}".format(project.name, file_basename)
logger.info("Copy {} to project folder {}".format(project.settings.inject_exe_out, dest))
shutil.copy(
project.settings.inject_exe_out,
dest,
)
@views.route("/project/<project_name>/build", methods=['POST', 'GET'])
def build_project(project_name):
@@ -323,18 +314,20 @@ def start_project(project_name):
return redirect("/project/{}".format(project_name), code=302)
def get_logfiles():
def get_logfiles(directory):
log_files = []
id = 0
asm_a = "" # for diff
asm_b = ""
for file in os.listdir(f"{logs_dir}/"):
for file in os.listdir(f"{directory}/"):
if file.startswith("."):
continue
if not file.startswith("log-"):
continue
if file.endswith(".bin"):
continue
with open(os.path.join(f"{logs_dir}/", file), "r") as f:
if file.endswith(".bin"):
continue
with open(os.path.join(f"{directory}/", file), "r") as f:
data = f.read()
if 'main_c' in file:
data = highlight(data, CLexer(), HtmlFormatter(full=False))
helper.py
+9 -14
View File
@@ -13,25 +13,20 @@ logger = logging.getLogger("Helper")
SHC_VERIFY_SLEEP = 0.1
def clean_files():
def clean_files(settings):
logger.info("--( Remove old files")
files_to_clean = [
# compile artefacts in current dir
# compile artefacts in current working dir
"main-clean.obj",
"main.obj",
"mllink$.lnk",
#"out/7z-verify.exe",
#"out/wifiinfoview-verify.exe",
#"out/procexp64-verify.exe",
# out/ stuff
os.path.join(build_dir, "main.asm"),
os.path.join(build_dir, "main.bin"),
os.path.join(build_dir, "main.c"),
os.path.join(build_dir, "peb_lookup.h"),
os.path.join(build_dir, "main.exe"),
VerifyFilename,
# temporary files
settings.main_c_path,
settings.main_asm_path,
settings.main_shc_path,
settings.main_exe_path,
]
for file in files_to_clean:
pathlib.Path(file).unlink(missing_ok=True)
model/defs.py
+2 -12
View File
@@ -7,6 +7,7 @@ class FilePath(str):
# with data/shellcodes/createfile.bin
VerifyFilename: FilePath = r'C:\Temp\a'
# Directory structure
PATH_EXES = "data/binary/exes/"
PATH_SHELLCODES = "data/binary/shellcodes/"
PATH_PEB_WALK = "data/source/carrier/peb_walk/"
@@ -14,8 +15,7 @@ PATH_IAT_REUSE = "data/source/carrier/iat_reuse/"
PATH_PAYLOAD = "data/source/payload/"
PATH_DECODER = "data/source/carrier/decoder/"
PATH_WEB_PROJECT = "app/projects/"
PATH_WORKING = "working/"
PATH_WEB_PROJECT = "app/projects/" # web only
# Correlated with real template files
@@ -62,13 +62,3 @@ class IatEntry():
self.dll_name: str = dll_name
self.func_name: str = func_name
self.iat_vaddr: int = iat_vaddr
# no slash at end
build_dir = "working/build"
logs_dir = "working/logs"
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")
model/settings.py
+9
View File
@@ -27,3 +27,12 @@ class Settings():
self.generate_asm_from_c: bool = True
self.generate_shc_from_asm: bool = True
def prep(self):
self.main_dir = "data/source/carrier/" + self.source_style.value + "/"
self.template_path = self.main_dir + "template.c"
self.main_c_path = self.main_dir + "main.c"
self.main_asm_path = self.main_dir + "main.asm"
self.main_exe_path = self.main_dir + "main.exe"
self.main_shc_path = self.main_dir + "main.bin"
phases/compiler.py
-1
View File
@@ -14,7 +14,6 @@ from model.carrier import Carrier
from model.exehost import ExeHost
logger = logging.getLogger("Compiler")
use_templates = True
# NOTE: Mostly copy-pasted from compiler.py::compile()
def compile_dev(
phases/injector.py
+2 -2
View File
@@ -16,7 +16,7 @@ logger = logging.getLogger("Injector")
def inject_exe(
main_shc_file: FilePath,
main_shc_path: FilePath,
settings: Settings,
project: Project,
):
@@ -32,7 +32,7 @@ def inject_exe(
# Read prepared loader shellcode
# And check if it fits into the target code section
main_shc = file_readall_binary(main_shc_file)
main_shc = file_readall_binary(main_shc_path)
l = len(main_shc)
if l + 128 > project.exe_host.code_section.Misc_VirtualSize:
logger.error("Error: Shellcode {}+128 too small for target code section {}".format(
phases/templater.py
+34 -76
View File
@@ -1,98 +1,56 @@
from jinja2 import Template
import pprint
import shutil
import logging
from helper import *
from observer import observer
from model.defs import *
from model.settings import Settings
use_templates = True
logger = logging.getLogger("Assembler")
# INPUT:
# data/plugins/
# data/source/
#
# Output:
# build/main.c
# build/*.h
def create_c_from_template(
source_style: SourceStyle,
alloc_style: AllocStyle,
exec_style: ExecStyle,
decoder_style: DecoderStyle,
payload_len: int,
):
plugin_allocator = ""
plugin_decoder = ""
plugin_executor = ""
def create_c_from_template(settings: Settings, payload_len: int):
logger.info("--[ Create C from template")
plugin_decoder = ""
#filepath = "data/plugins/allocator/{}.c".format(alloc_style.value)
#with open(filepath, "r", encoding='utf-8') as file:
# plugin_allocator = file.read()
# plugin_allocator = Template(plugin_allocator).render({
# 'PAYLOAD_LEN': payload_len,
# })
filepath = PATH_DECODER + "{}.c".format(decoder_style.value)
with open(filepath, "r", encoding='utf-8') as file:
# Decoder
filepath_decoder = PATH_DECODER + "{}.c".format(settings.decoder_style.value)
with open(filepath_decoder, "r", encoding='utf-8') as file:
plugin_decoder = file.read()
plugin_decoder = Template(plugin_decoder).render({
'PAYLOAD_LEN': payload_len,
'XOR_KEY': config.xor_key,
})
#filepath = "data/plugins/executor/{}.c".format(exec_style.value)
#with open("data/plugins/executor/direct_1.c", "r", encoding='utf-8') as file:
# plugin_executor = file.read()
# plugin_executor = Template(plugin_executor).render({
# 'PAYLOAD_LEN': payload_len,
# })
if source_style == SourceStyle.peb_walk:
if use_templates:
with open(PATH_PEB_WALK + "template.c", 'r', encoding='utf-8') as file:
template_content = file.read()
observer.add_text_file("main_c_template", template_content)
# C Template: peb_walk
if settings.source_style == SourceStyle.peb_walk:
with open(settings.template_path, 'r', encoding='utf-8') as file:
template_content = file.read()
observer.add_text_file("main_c_template", template_content)
template = Template(template_content)
rendered_template = template.render({
'plugin_allocator': plugin_allocator,
'plugin_decoder': plugin_decoder,
'plugin_executor': plugin_executor,
'PAYLOAD_LEN': payload_len,
})
with open(main_c_file, "w", encoding='utf-8') as file:
file.write(rendered_template)
observer.add_text_file("main_c_rendered", rendered_template)
template = Template(template_content)
rendered_template = template.render({
'plugin_decoder': plugin_decoder,
'PAYLOAD_LEN': payload_len,
})
with open(settings.main_c_path, "w", encoding='utf-8') as file:
file.write(rendered_template)
observer.add_text_file("main_c_rendered", rendered_template)
# TODO PEB
shutil.copy(PATH_PEB_WALK + "peb_lookup.h", f"{build_dir}/peb_lookup.h")
else:
observer.add_text_file("main_c", file_readall_text(PATH_PEB_WALK + "main.c"))
shutil.copy(PATH_PEB_WALK + "main.c", main_c_file)
# TODO PEB
shutil.copy(PATH_PEB_WALK + "peb_lookup.h", f"{build_dir}/peb_lookup.h")
# C Template: iat_reuse
elif settings.source_style == SourceStyle.iat_reuse:
with open(PATH_IAT_REUSE + "template.c", 'r', encoding='utf-8') as file:
template_content = file.read()
observer.add_text_file("main_c_template", template_content)
template = Template(template_content)
rendered_template = template.render({
'plugin_decoder': plugin_decoder,
'PAYLOAD_LEN': payload_len,
})
with open(settings.main_c_path, "w", encoding='utf-8') as file:
file.write(rendered_template)
observer.add_text_file("main_c_rendered", rendered_template)
elif source_style == SourceStyle.iat_reuse:
if use_templates:
with open(PATH_IAT_REUSE + "template.c", 'r', encoding='utf-8') as file:
template_content = file.read()
observer.add_text_file("main_c_template", template_content)
template = Template(template_content)
rendered_template = template.render({
'plugin_allocator': plugin_allocator,
'plugin_decoder': plugin_decoder,
'plugin_executor': plugin_executor,
'PAYLOAD_LEN': payload_len,
})
with open(main_c_file, "w", encoding='utf-8') as file:
file.write(rendered_template)
observer.add_text_file("main_c_rendered", rendered_template)
else:
observer.add_text_file("main_c", file_readall_text(PATH_IAT_REUSE + "main.c"))
shutil.copy(PATH_IAT_REUSE + "main.c", main_c_file)
else:
raise Exception("Invalid source style: {}".format(settings.source_style))
sender.py
-9
View File
@@ -1,9 +1,7 @@
import requests as req
import logging
import brotli
import os
import time
import shutil
from config import config
@@ -32,10 +30,3 @@ def scannerDetectsBytes(data: bytes, filename: str, useBrotli=True, verify=False
raise Exception("Server error, aborting")
return jsonRes
def main():
with open("data/exes/7z-verify.exe", "rb") as f:
data = f.read()
res = scannerDetectsBytes(data, "test.exe")
print("Answer: {}".format(res))
supermega.py
+25 -28
View File
@@ -18,7 +18,6 @@ from model.project import Project
from model.settings import Settings
from model.defs import *
from log import setup_logging
from utils import delete_all_files_in_directory
def main():
@@ -104,25 +103,29 @@ def main():
def start(settings: Settings) -> int:
"""Main entry point for the application. Will handle log files and cleanup"""
settings.prep()
# Delete: all old files
if settings.cleanup_files_on_start:
clean_files()
delete_all_files_in_directory(f"{logs_dir}/")
clean_files(settings)
# And logs
observer.reset()
# Do the thing and catch errors
try:
start_real(settings)
except Exception as e:
logger.error(f'Error compiling: {e}')
write_logs()
write_logs(settings.main_dir)
return 1
# Cleanup files
if settings.cleanup_files_on_exit:
clean_files()
clean_files(settings)
write_logs()
# Write logs (on success)
write_logs(settings.main_dir)
return 0
@@ -142,19 +145,13 @@ def start_real(settings: Settings):
))
# Create: Carrier C source files from template (C->C)
phases.templater.create_c_from_template(
source_style = settings.source_style,
alloc_style = settings.alloc_style,
exec_style = settings.exec_style,
decoder_style= settings.decoder_style,
payload_len = project.payload.len,
)
phases.templater.create_c_from_template(settings, project.payload.len)
# Compile: Carrier to .asm (C -> ASM)
if settings.generate_asm_from_c:
phases.compiler.compile(
c_in = main_c_file,
asm_out = main_asm_file,
c_in = settings.main_c_path,
asm_out = settings.main_asm_path,
payload_len = project.payload.len,
carrier = project.carrier,
source_style = project.settings.source_style,
@@ -164,15 +161,15 @@ def start_real(settings: Settings):
# Assemble: Assemble .asm to .shc (ASM -> SHC)
if settings.generate_shc_from_asm:
phases.assembler.asm_to_shellcode(
asm_in = main_asm_file,
build_exe = main_exe_file,
shellcode_out = main_shc_file)
asm_in = settings.main_asm_path,
build_exe = settings.main_exe_path,
shellcode_out = settings.main_shc_path)
# Merge: shellcode/loader with payload (SHC + PAYLOAD -> SHC)
if settings.dataref_style == DataRefStyle.APPEND:
phases.assembler.merge_loader_payload(
shellcode_in = main_shc_file,
shellcode_out = main_shc_file,
shellcode_in = settings.main_shc_path,
shellcode_out = settings.main_shc_path,
payload_data = project.payload.payload_data,
decoder_style = settings.decoder_style)
@@ -181,12 +178,12 @@ def start_real(settings: Settings):
logger.info("--[ RWX section {} found. Will obfuscate loader+payload and inject into it".format(
project.exe_host.rwx_section.Name.decode().rstrip('\x00')
))
obfuscate_shc_loader(main_shc_file, main_shc_file + ".sgn")
observer.add_code_file("payload_sgn", file_readall_binary(main_shc_file + ".sgn"))
shutil.move(main_shc_file + ".sgn", main_shc_file)
obfuscate_shc_loader(settings.main_shc_path, settings.main_shc_path + ".sgn")
observer.add_code_file("payload_sgn", file_readall_binary(settings.main_shc_path + ".sgn"))
shutil.move(settings.main_shc_path + ".sgn", settings.main_shc_path)
# inject merged loader into an exe
phases.injector.inject_exe(main_shc_file, settings, project)
phases.injector.inject_exe(settings.main_shc_path, settings, project)
observer.add_code_file("exe_final", extract_code_from_exe_file_ep(settings.inject_exe_out, 300))
if config.get("avred_server") != "":
@@ -210,21 +207,21 @@ def start_real(settings: Settings):
], check=True)
def write_logs():
def write_logs(working_dir: str):
# Our log output
with open(f"{logs_dir}/supermega.log", "w") as f:
with open(f"{working_dir}log-supermega.log", "w") as f:
for line in observer.get_logs():
f.write(line + "\n")
# Stdout of executed commands
with open(f"{logs_dir}/cmdoutput.log", "w") as f:
with open(f"{working_dir}log-cmdoutput.log", "w") as f:
for line in observer.get_cmd_output():
f.write(line)
# Write all files
idx = 0
for name, data in observer.files:
with open(f"{logs_dir}/{idx}-{name}", "w") as f:
with open(f"{working_dir}log-{idx}-{name}", "w") as f:
f.write(data)
idx += 1
tester.py
+4 -4
View File
@@ -22,7 +22,7 @@ def main():
settings.source_style = SourceStyle.peb_walk
settings.inject_mode = InjectStyle.ChangeEntryPoint
settings.inject_exe_in = PATH_EXES + "7z.exe"
settings.inject_exe_out = PATH_EXES + "7z-verify.exe"
settings.inject_exe_out = PATH_EXES + "7z.verify.exe"
if start(settings) != 0:
print("Error")
return 1
@@ -31,7 +31,7 @@ def main():
settings.source_style = SourceStyle.peb_walk
settings.inject_mode = InjectStyle.BackdoorCallInstr
settings.inject_exe_in = PATH_EXES + "7z.exe"
settings.inject_exe_out = PATH_EXES + "7z-verify.exe"
settings.inject_exe_out = PATH_EXES + "7z.verify.exe"
if start(settings) != 0:
print("Error")
return 1
@@ -40,7 +40,7 @@ def main():
settings.source_style = SourceStyle.iat_reuse
settings.inject_mode = InjectStyle.ChangeEntryPoint
settings.inject_exe_in = PATH_EXES + "procexp64.exe"
settings.inject_exe_out = PATH_EXES + "procexp64-verify.exe"
settings.inject_exe_out = PATH_EXES + "procexp64.verify.exe"
if start(settings) != 0:
print("Error")
return 1
@@ -49,7 +49,7 @@ def main():
settings.source_style = SourceStyle.iat_reuse
settings.inject_mode = InjectStyle.ChangeEntryPoint
settings.inject_exe_in = PATH_EXES + "procexp64.exe"
settings.inject_exe_out = PATH_EXES + "procexp64-verify.exe"
settings.inject_exe_out = PATH_EXES + "procexp64.verify.exe"
if start(settings) != 0:
print("Error")
return 1
working/build/.gitkeep
View File
working/logs/.gitkeep
View File