"""Luau source -> Luraph v14.9-style prototype graph -> protected Luau. from compiler import compile_source protected = compile_source('print("hello")') Pipeline: 1. luau-compile -O0 -g0 (official front end: parsing, scoping, closures, upvalue analysis, register allocation, control-flow lowering) 2. parse the bytecode (luau_bc.py) 3. lower every Luau instruction to one or more recovered v14.9 instructions (isa.py), with captured locals boxed into cells 4. serialise the prototype table and wrap it in the VM runtime The emitted graph uses the recovered v14.9 prototype schema: prototypes are dicts keyed 1..11 (arity, upvalues, opcodes, C, B, frame, K3, K2, K1, A, debug names), with sparse per-field tables indexed by 1-based pc. """ from __future__ import annotations import json import os import random import re import struct import subprocess import sys import tempfile from pathlib import Path HERE = Path(__file__).resolve().parent sys.path.insert(0, str(HERE)) import handler_runtime import isa import luau_bc import macros LUAU_COMPILE = os.environ.get('LUAU_COMPILE') or ('/opt/ws-remote/bin/luau-compile' if Path('/opt/ws-remote/bin/luau-compile').is_file() else '/tmp/opencode/luau-src/build/luau-compile') LUAU = os.environ.get('LUAU') or ('/opt/ws-remote/bin/luau' if Path('/opt/ws-remote/bin/luau').is_file() else '/tmp/opencode/luau-src/build/luau') ROLE_FIELD = {'A': 'A', 'B': 'B', 'C': 'C', 'K1': 'K1', 'K2': 'K2', 'K3': 'K3'} class CompileError(Exception): pass # -------------------------------------------------------------------------- # IR class Ins: __slots__ = ('name', 'fields', 'target', 'pc', 'src_line') def __init__(self, name, fields, pc): self.name = name self.fields = fields self.target = None # target Luau pc, resolved in pass 2 self.pc = pc # source Luau word pc (diagnostics) self.src_line = 0 # source line (diagnostics; -g1/-g2 builds) def __repr__(self): bits = ', '.join('%s=%r' % kv for kv in sorted(self.fields.items())) t = '' if self.target is None else ' -> %d' % self.target return '%s(%s)%s' % (self.name, bits, t) class Lowerer: def __init__(self, module, proto, boxed, proto_ids): self.module = module self.proto = proto self.consts = proto.consts self.boxed = boxed self.proto_ids = proto_ids self.ir = [] self.labels = {} self.pc = 0 self.scratch = proto.maxstack + 2 self.allocated = 0 self.max_reg = proto.maxstack + 2 self.local_targets = {} self.next_local = 0 # -- emission ------------------------------------------------------- def emit(self, name, target=None, **roles): spec = isa.OPS[name] fields = {} for role, value in roles.items(): f = spec['roles'].get(role) if f is None: raise CompileError('%s has no role %r (pc %d)' % (name, role, self.pc)) fields[f] = value ins = Ins(name, fields, self.pc) ins.src_line = self.proto.lines.get(self.pc, 0) if target is not None: if 'target' not in spec['roles']: raise CompileError('%s is not a jump' % name) ins.fields.pop(spec['roles']['target'], None) ins.target = target self.ir.append(ins) return ins def alloc(self, count=1): base = self.scratch + self.allocated self.allocated += count self.max_reg = max(self.max_reg, base + count - 1) return base def reset_scratch(self): self.allocated = 0 def new_local(self): self.next_local += 1 return self.next_local def mark_local(self, ident): self.local_targets[ident] = len(self.ir) # -- boxed registers ------------------------------------------------ def rd(self, reg): if reg in self.boxed: s = self.alloc() self.emit('XGETCELL', dst=s, cell=reg) return s return reg def wr(self, reg, produce): if reg in self.boxed: s = self.alloc() produce(s) self.emit('XSETCELL', cell=reg, src=s) else: produce(reg) def copy(self, src, dst): v = self.rd(src) self.wr(dst, lambda d: self.emit('MOVE', dst=d, src=v)) # -- constants ------------------------------------------------------ def kconst(self, index): if index is None or index < 0 or index >= len(self.consts): raise CompileError('constant index out of range: %r' % (index,)) return self.consts[index] def import_names(self, did): count = did >> 30 ids = [(did >> 20) & 1023, (did >> 10) & 1023, did & 1023][:count] names = [] for i in ids: v = self.kconst(i) if not isinstance(v, str): raise CompileError('import component is not a string: %r' % (v,)) names.append(v) return names def load_import(self, dst, did): names = self.import_names(did) self.wr(dst, lambda d: self.emit('GETGLOBAL', dst=d, k=names[0])) for name in names[1:]: def prod(d, name=name): self.emit('GETTABLE', dst=d, obj=d, k=name) v = self.rd(dst) if dst in self.boxed else dst if dst in self.boxed: # chained lookup happens in the scratch register for name2 in names[1:]: self.emit('GETTABLE', dst=v, obj=v, k=name2) self.emit('XSETCELL', cell=dst, src=v) return self.emit('GETTABLE', dst=dst, obj=dst, k=name) # -- calls ---------------------------------------------------------- def lower_call(self, a, b, c): argc = b - 1 if b > 0 else -1 nres = c - 1 if c > 0 else -1 clean = (a not in self.boxed and all((a + 1 + i) not in self.boxed for i in range(max(argc, 0)))) # faithful single-result calls when the register window is not boxed if clean and nres == 1 and argc >= 0: if argc == 0: self.emit('CALL0', base=a) return if argc == 1: self.emit('CALL1', base=a) return if argc == 2: self.emit('CALL2', base=a) return self.emit('CALLN', base=a, argc=argc + 1) return # A call whose argument count comes from the runtime vararg top can # stay in the original register window. Moving it into a low scratch # window can overwrite the varargs themselves. if argc < 0 and a not in self.boxed: self.emit('XCALL', base=a, argc=-1, nres=nres) return # general path: build the call in a private scratch window size = 2 + (argc if argc > 0 else 1) cb = self.alloc(size + 2) self.copy(a, cb) if argc >= 0: for i in range(argc): self.copy(a + 1 + i, cb + 1 + i) eff = argc else: self.emit('XMOVETOP', dst=cb + 1, src=a + 1) eff = -1 self.emit('XCALL', base=cb, argc=eff, nres=nres) if nres < 0: self.emit('XMOVETOP', dst=a, src=cb) else: for i in range(nres): self.copy(cb + i, a + i) # -- the translator ------------------------------------------------- def run(self): code = self.proto.code boxed_at_entry = {r for r in self.boxed if r < self.proto.numparams} for r in sorted(boxed_at_entry): self.emit('XBOXCELL', reg=r) for ins in code: pc, name, a, b, c, d, e, aux = ins self.pc = pc self.labels[pc] = len(self.ir) self.reset_scratch() self.translate(name, a, b, c, d, e, aux, pc) # implicit fall-off-the-end return if not self.ir or self.ir[-1].name not in ('RET0', 'RET1', 'XRET', 'XRETMULTI'): self.emit('RET0', close=0) return self.finish() def finish(self): n = len(self.ir) code = [None] * (n + 1) for idx, ins in enumerate(self.ir): fields = dict(ins.fields) if ins.target is not None: if isinstance(ins.target, tuple): tgt = self.local_targets[ins.target[1]] else: if ins.target not in self.labels: raise CompileError('jump to missing pc %d' % ins.target) tgt = self.labels[ins.target] # encoded target - 1 spec = isa.OPS[ins.name] fields[spec['roles']['target']] = tgt code[idx + 1] = (isa.OPS[ins.name]['op'], fields) return code, self.max_reg def translate(self, name, a, b, c, d, e, aux, pc): emit = self.emit rd, wr, copy = self.rd, self.wr, self.copy nxt = pc + luau_bc.op_length(luau_bc.OP[name]) sd = d - 0x10000 if d >= 0x8000 else d # signed 16-bit offset se = e - 0x1000000 if e >= 0x800000 else e # signed 24-bit offset if name in ('NOP', 'BREAK', 'COVERAGE', 'PREPVARARGS'): return if name == 'CLOSEUPVALS': return if name.startswith('FASTCALL') or name == 'NATIVECALL': return if name == 'CALLFB': # Feedback-slot call is a normal call; the aux slot is irrelevant here. self.lower_call(a, b, c) return if name == 'LOADNIL': # Luau LOADNIL always writes one register; B is not a count. wr(a, lambda dst: emit('LOADNIL', dst=dst)) elif name == 'LOADB': wr(a, lambda dst: emit('LOADK', dst=dst, k=bool(b))) if c: emit('JMP', target=nxt + c) elif name == 'LOADN': wr(a, lambda dst: emit('LOADK', dst=dst, k=float(d))) elif name == 'LOADK': v = self.kconst(d) if isinstance(v, tuple) and v and v[0] == 'import': self.load_import(a, v[1]) elif isinstance(v, tuple) and v and v[0] == 'table': self.build_table(a, v[1]) elif isinstance(v, tuple) and v and v[0] == 'closure': emit('CLOSURE', dst=a, child=self.proto_ids[v[1]], caps=self.capture_list(pc)) else: wr(a, lambda dst: emit('LOADK', dst=dst, k=v)) elif name == 'LOADKX': v = self.kconst(aux) wr(a, lambda dst: emit('LOADK', dst=dst, k=v)) elif name == 'MOVE': copy(b, a) elif name == 'GETGLOBAL': v = self.kconst(aux) wr(a, lambda dst: emit('GETGLOBAL', dst=dst, k=v)) elif name == 'SETGLOBAL': v = self.kconst(aux) emit('XSETGLOBAL', k=v, src=rd(a)) elif name == 'GETUPVAL': wr(a, lambda dst: emit('GETUPVAL', dst=dst, cell=b)) elif name == 'SETUPVAL': emit('SETUPVAL', cell=b, src=rd(a)) elif name == 'GETIMPORT': self.load_import(a, aux) elif name == 'GETTABLE': k = rd(c) v = rd(b) wr(a, lambda dst: emit('GETTABLE_RR', dst=dst, obj=v, key=k)) elif name == 'GETTABLEKS': v = rd(b) key = self.kconst(aux) wr(a, lambda dst: emit('GETTABLE', dst=dst, obj=v, k=key)) elif name == 'GETTABLEN': v = rd(b) wr(a, lambda dst: emit('GETTABLE', dst=dst, obj=v, k=c + 1)) elif name == 'SETTABLE': val, obj, key = rd(a), rd(b), rd(c) emit('SETTABLE_RR', obj=obj, key=key, val=val) elif name == 'SETTABLEKS': obj, key = rd(b), self.kconst(aux) emit('SETTABLE', obj=obj, k=key, val=rd(a)) elif name == 'SETTABLEN': obj = rd(b) emit('SETTABLE', obj=obj, k=c + 1, val=rd(a)) elif name == 'NEWTABLE': wr(a, lambda dst: emit('NEWTABLE', dst=dst)) elif name == 'DUPTABLE': keys = self.kconst(d) wr(a, lambda dst: emit('NEWTABLE', dst=dst)) if isinstance(keys, tuple) and keys and keys[0] == 'table': for k in keys[1]: if not isinstance(k, tuple): key = self.kconst(k) emit('SETTABLE', obj=rd(a), k=key, val=0.0) elif name == 'SETLIST': # Luau: B = first source register, C = number of values + 1 # (0 = to top), aux = first array index. count = c - 1 if c > 0 else -1 source = b if b > 0 else a + 1 emit('XSETLIST', dst=a, **{'from': source, 'count': count, 'index': aux}) elif name == 'ADD': l, r = rd(b), rd(c) wr(a, lambda dst: emit('ADD', dst=dst, lhs=l, rhs=r)) elif name == 'SUB': l, r = rd(b), rd(c) wr(a, lambda dst: emit('SUB', dst=dst, lhs=l, rhs=r)) elif name == 'MUL': l, r = rd(b), rd(c) wr(a, lambda dst: emit('MUL', dst=dst, lhs=l, rhs=r)) elif name == 'DIV': l, r = rd(b), rd(c) wr(a, lambda dst: emit('DIV', dst=dst, lhs=l, rhs=r)) elif name == 'MOD': l, r = rd(b), rd(c) wr(a, lambda dst: emit('MOD', dst=dst, lhs=l, rhs=r)) elif name == 'POW': l, r = rd(b), rd(c) wr(a, lambda dst: emit('XPOW', dst=dst, lhs=l, rhs=r)) elif name == 'IDIV': l, r = rd(b), rd(c) wr(a, lambda dst: emit('IDIV_EXT', dst=dst, lhs=l, rhs=r)) elif name in ('ADDK', 'SUBK', 'MULK', 'DIVK', 'MODK', 'POWK', 'IDIVK'): k = self.kconst(c) opname = {'ADDK': 'ADD', 'SUBK': 'SUB', 'MULK': 'MUL', 'DIVK': 'DIV', 'MODK': 'MOD', 'POWK': 'XPOW', 'IDIVK': 'IDIV_EXT'}[name] if name == 'ADDK' and a not in self.boxed: emit('ADDK', dst=a, lhs=rd(b), k=k) return if name == 'SUBK' and a not in self.boxed: emit('SUBK', dst=a, lhs=rd(b), k=k) return l = rd(b) t = self.alloc() emit('LOADK', dst=t, k=k) wr(a, lambda dst: emit(opname, dst=dst, lhs=l, rhs=t)) elif name == 'CONCAT': # right-associative fold into a scratch accumulator acc = self.alloc() emit('MOVE', dst=acc, src=rd(c)) for i in range(c - 1, b - 1, -1): l = rd(i) t = self.alloc() emit('CONCAT', dst=t, lhs=l, rhs=acc) emit('MOVE', dst=acc, src=t) wr(a, lambda dst: emit('MOVE', dst=dst, src=acc)) elif name == 'NOT': v = rd(b) wr(a, lambda dst: emit('NOT', dst=dst, src=v)) elif name == 'MINUS': v = rd(b) wr(a, lambda dst: emit('UNM', dst=dst, src=v)) elif name == 'LENGTH': v = rd(b) wr(a, lambda dst: emit('LEN', dst=dst, src=v)) elif name in ('AND', 'OR', 'ANDK', 'ORK'): # R[a] = and/or if name in ('ANDK', 'ORK'): k = self.kconst(c) r = self.alloc() emit('LOADK', dst=r, k=k) l = rd(b) else: l, r = rd(b), rd(c) t = self.alloc() falsy = self.alloc() emit('NOT', dst=falsy, src=l) end_l = self.new_local() right_l = self.new_local() # AND keeps the left value when it is falsy; OR keeps it when truthy emit('JMPIFNOT' if name in ('AND', 'ANDK') else 'JMPIF', cond=falsy, target=('local', right_l)) emit('MOVE', dst=t, src=l) emit('JMP', target=('local', end_l)) self.mark_local(right_l) emit('MOVE', dst=t, src=r) self.mark_local(end_l) wr(a, lambda dst: emit('MOVE', dst=dst, src=t)) elif name == 'JUMP' or name == 'JUMPBACK': emit('JMP', target=pc + 1 + sd) elif name == 'JUMPX': emit('JMP', target=pc + 1 + se) elif name == 'JUMPIF': emit('JMPIF', cond=rd(a), target=pc + 1 + sd) elif name == 'JUMPIFNOT': emit('JMPIFNOT', cond=rd(a), target=pc + 1 + sd) elif name == 'JUMPIFEQ': t = self.alloc() emit('EQ', dst=t, lhs=rd(a), rhs=rd((aux or 0) & 0xFF)) emit('JMPIF', cond=t, target=pc + 1 + sd) elif name == 'JUMPIFNOTEQ': t = self.alloc() emit('NE', dst=t, lhs=rd(a), rhs=rd((aux or 0) & 0xFF)) emit('JMPIF', cond=t, target=pc + 1 + sd) elif name == 'JUMPIFLE': t = self.alloc() emit('LE', dst=t, lhs=rd(a), rhs=rd((aux or 0) & 0xFF)) emit('JMPIF', cond=t, target=pc + 1 + sd) elif name == 'JUMPIFNOTLE': t = self.alloc() emit('GT', dst=t, lhs=rd(a), rhs=rd((aux or 0) & 0xFF)) emit('JMPIF', cond=t, target=pc + 1 + sd) elif name == 'JUMPIFLT': t = self.alloc() emit('GT', dst=t, lhs=rd((aux or 0) & 0xFF), rhs=rd(a)) emit('JMPIF', cond=t, target=pc + 1 + sd) elif name == 'JUMPIFNOTLT': t = self.alloc() emit('GE', dst=t, lhs=rd(a), rhs=rd((aux or 0) & 0xFF)) emit('JMPIF', cond=t, target=pc + 1 + sd) elif name.startswith('JUMPXEQK'): kind = name[len('JUMPXEQK'):] negate = bool((aux or 0) >> 31) if kind in ('N', 'S', 'B') else bool((aux or 0) >> 31) t = self.alloc() if kind == 'NIL': emit('EQNIL', dst=t, src=rd(a)) else: if kind == 'B': k = bool((aux or 0) & 1) elif kind == 'N': raw = (aux or 0) & 0xFFFFFF if raw & 0x800000: raw -= 0x1000000 k = float(raw) else: k = self.kconst((aux or 0) & 0xFFFF) kv = self.alloc() emit('LOADK', dst=kv, k=k) emit('EQ', dst=t, lhs=rd(a), rhs=kv) emit('JMPIFNOT' if negate else 'JMPIF', cond=t, target=pc + 1 + sd) elif name == 'FORNPREP': emit('XFORNPREP', base=a, target=pc + 1 + sd) elif name == 'FORNLOOP': emit('XFORNLOOP', base=a, target=pc + 1 + sd) elif name in ('FORGPREP', 'FORGPREP_NEXT', 'FORGPREP_INEXT'): # FORGPREP jumps to the matching FORGLOOP; the step performs the # first iterator call and jumps back into the body. emit('JMP', target=pc + 1 + sd) elif name == 'FORGLOOP': emit('XFORGLOOP', base=a, nvars=aux & 0xFFFF, target=pc + 1 + sd) elif name == 'CALL': self.lower_call(a, b, c) elif name == 'RETURN': if b == 0: emit('XRETMULTI', **{'from': a}) elif b == 1: emit('RET0', close=0) elif b == 2: emit('RET1', src=rd(a)) else: vals = [rd(a + i) for i in range(b - 1)] # independent registers already; emit a range return if all(isinstance(v, int) for v in vals) and vals == list(range(a, a + b - 1)): emit('XRET', **{'from': a, 'count': b - 1}) else: base = self.alloc(b - 1) for i, v in enumerate(vals): emit('MOVE', dst=base + i, src=v) emit('XRET', **{'from': base, 'count': b - 1}) elif name == 'NEWCLOSURE': # D indexes this prototype's child list, not the global proto table. if d < 0 or d >= len(self.proto.children): raise CompileError('NEWCLOSURE child index out of range: %r' % (d,)) child = self.proto_ids[self.proto.children[d]] emit('CLOSURE', dst=a, child=child, caps=self.capture_list(pc)) elif name == 'DUPCLOSURE': v = self.kconst(d) if isinstance(v, tuple) and v[0] == 'closure': emit('CLOSURE', dst=a, child=self.proto_ids[v[1]], caps=self.capture_list(pc)) else: raise CompileError('DUPCLOSURE without closure constant') elif name == 'CAPTURE': return elif name == 'GETVARARGS': cnt = b - 1 if b > 0 else -1 emit('XGETVARARG', dst=a, count=cnt) elif name == 'NAMECALL': obj = rd(b) copy(b, a + 1) wr(a, lambda dst: emit('GETTABLE', dst=dst, obj=obj, k=self.kconst(aux))) else: raise CompileError('unsupported Luau opcode %s (pc %d)' % (name, pc)) # -- helpers -------------------------------------------------------- def capture_list(self, pc): """Collect the CAPTURE instructions that follow a closure op.""" caps = [] for ins in self.proto.code: if ins[0] <= pc: continue if ins[1] != 'CAPTURE': break caps.append([ins[2], ins[3]]) return caps def build_table(self, dst, entries): wr(dst, lambda d: self.emit('NEWTABLE', dst=d)) if isinstance(entries, list) and entries and isinstance(entries[0], tuple): for k, v in entries: key = self.kconst(k) val = self.kconst(v) if v >= 0 else 0.0 emit('SETTABLE', obj=rd(dst), k=key, val=val) else: for k in entries: key = self.kconst(k) emit('SETTABLE', obj=rd(dst), k=key, val=0.0) # -------------------------------------------------------------------------- # module -> graph class ModuleLowerer: def __init__(self, bc): self.bc = bc self.protos = bc['protos'] self.proto_ids = {p.index: p.index + 1 for p in self.protos} self.main = self.proto_ids[bc['main']] self.boxed = [self.find_boxed(p) for p in self.protos] @staticmethod def find_boxed(proto): """Registers captured by reference (CAPTURE kind 1).""" boxed = set() code = proto.code for i, ins in enumerate(code): if ins[1] not in ('NEWCLOSURE', 'DUPCLOSURE'): continue child = None if ins[1] == 'NEWCLOSURE': child = proto.children[ins[5]] if ins[5] < len(proto.children) else None else: v = proto.consts[ins[5]] if ins[5] < len(proto.consts) else None if isinstance(v, tuple) and v and v[0] == 'closure': child = v[1] if child is None: continue nups = 0 for j in range(i + 1, len(code)): if code[j][1] != 'CAPTURE': break nups += 1 if code[j][2] == 1: boxed.add(code[j][3]) return boxed def build(self): protos = {} for p in self.protos: pid = self.proto_ids[p.index] lower = Lowerer(self.bc, p, self.boxed[p.index], self.proto_ids) code, max_reg = lower.run() proto = {1: p.numparams, 2: {}, 3: {}, 4: {}, 5: {}, 6: max_reg + 1, 7: {}, 8: {}, 9: {}, 10: {}, 11: {}, 12: [self.proto_ids[c] for c in p.children], 13: sorted(lower.boxed), 14: bool(p.is_vararg)} for pc in range(1, len(code)): op, fields = code[pc] proto[3][pc] = op for name, value in fields.items(): f = {'A': 10, 'B': 5, 'C': 4, 'K1': 9, 'K2': 8, 'K3': 7}[name] proto[f][pc] = value # upvalue descriptors, informational (the closure site carries them) ups = {} for i, ins in enumerate(p.code): if ins[1] in ('NEWCLOSURE', 'DUPCLOSURE'): slot = 1 for j in range(i + 1, len(p.code)): if p.code[j][1] != 'CAPTURE': break ups[slot] = [p.code[j][2], p.code[j][3]] slot += 1 if ups: proto[2] = ups for later in proto[3]: pass protos[pid] = proto return {'format': 'luraph-v149-graph/1', 'root': self.main, 'prototypes': protos, 'state': {}, 'nodes': [], 'enc': False} # -------------------------------------------------------------------------- # serialisation def encode_string(s, key=0, sbox=None): raw = s.encode('utf-8', 'surrogateescape') seed = (57 + key * 17) % 256 out = bytearray() for b in raw: v = (b ^ seed ^ 92) & 0xFF out.append(sbox[v] if sbox is not None else v) seed = (141 * seed + 29) % 256 return b'\x01' + bytes(out) def luau_string(data): out = ['"'] for b in data: if b == 34: out.append('\\"') elif b == 92: out.append('\\\\') elif 32 <= b < 127: out.append(chr(b)) else: out.append('\\%03d' % b) out.append('"') return ''.join(out) def luau_value(v, enc, key, sbox=None): if v is None: return 'nil' if v is True: return 'true' if v is False: return 'false' if isinstance(v, str): return luau_string(encode_string(v, key, sbox) if enc else v.encode('utf-8', 'surrogateescape')) if isinstance(v, float): if v == int(v) and abs(v) < 1e15: return '%.1f' % v return repr(v) if isinstance(v, int): return str(v) if isinstance(v, list): return '{' + ','.join(luau_value(x, enc, key, sbox) for x in v) + '}' raise CompileError('cannot serialise %r' % (v,)) def field_names(key=0): """Per-build names for every exposed graph/prototype/instruction field.""" rng = random.Random((int(key) * 2654435761 + 0x9E3779B9) & 0xFFFFFFFF) names = {} for field in ('enc', 'key', 'opkey', 'root', 'protos', 'opmap', 'sbox', 'sbox_inv', 'regbase', 'params', 'vararg', 'stack', 'boxed', 'code', 'A', 'B', 'C', 'K1', 'K2', 'K3'): while True: name = '_' + ''.join(rng.choice('abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789') for _ in range(9)) if name not in names.values(): names[field] = name break return names def emit_payload(graph, enc=True, key=0, opkey=0, names=None, opmap=None, sbox=None, regshift=0): if names is None: names = field_names(key) f = names lines = ['{%s=%s,%s=%d,%s=%d,%s=%d,%s=%d,%s={' % ( f['enc'], 'true' if enc else 'false', f['key'], key, f['opkey'], opkey, f['regbase'], regshift, f['root'], graph['root'], f['protos'])] for pid in sorted(graph['prototypes']): p = graph['prototypes'][pid] lines.append('[%d]={%s=%d,%s=%s,%s=%d,%s={%s},%s={' % (pid, f['params'], p[1], f['vararg'], 'true' if p.get(14) else 'false', f['stack'], p[6] + regshift, f['boxed'], ','.join(str(x + regshift) for x in p.get(13, [])), f['code'])) code = {} for pc, op in p[3].items(): rec = {'op': opmap[op] if opmap is not None else (op ^ opkey)} for name, field in (('A', 10), ('B', 5), ('C', 4), ('K1', 9), ('K2', 8), ('K3', 7)): if pc in p[field]: rec[name] = p[field][pc] if regshift: roles = isa.BY_CODE.get(op, {}).get('roles', {}) field_roles = {v: k for k, v in roles.items()} reg_roles = {'dst', 'src', 'obj', 'key', 'val', 'lhs', 'rhs', 'cond', 'base', 'from', 'to', 'reg'} for field in ('A', 'B', 'C'): role = field_roles.get(field) if role in reg_roles or (role == 'cell' and op in (0x105, 0x106)): if field in rec: rec[field] += regshift if op == 49 and 'K2' in rec and isinstance(rec['K2'], list): shifted = [] for cap in rec['K2']: if isinstance(cap, list) and len(cap) >= 2: if cap[0] in (0, 1): shifted.append([cap[0], cap[1] + regshift]) else: shifted.append(cap) else: shifted.append(cap) rec['K2'] = shifted code[pc] = rec n = max(code) if code else 0 for pc in range(1, n + 1): rec = code.get(pc) if rec is None: lines.append('{0},') continue parts = [str(rec['op'])] for name in ('A', 'B', 'C', 'K1', 'K2', 'K3'): if name in rec: parts.append('%s=%s' % (f[name], luau_value(rec[name], enc, key, sbox))) lines.append('{' + ','.join(parts) + '},') lines.append('}},') extras = [] if opmap is not None: inv = ','.join('[%d]=%d' % (code, op) for op, code in sorted(opmap.items(), key=lambda kv: kv[1])) extras.append('%s={%s}' % (f['opmap'], inv)) if sbox is not None: inv = [0] * 256 for i, v in enumerate(sbox): inv[v] = i extras.append('%s={%s}' % (f['sbox'], ','.join('[%d]=%d' % (i, x) for i, x in enumerate(sbox)))) extras.append('%s={%s}' % (f['sbox_inv'], ','.join('[%d]=%d' % (i, x) for i, x in enumerate(inv)))) if extras: lines.append('},' + ','.join(extras) + '}') else: lines.append('}}') return '\n'.join(lines) def _xor_state(state): state = (state ^ ((state << 13) & 0xFFFFFFFF)) & 0xFFFFFFFF state = (state ^ (state >> 17)) & 0xFFFFFFFF state = (state ^ ((state << 5) & 0xFFFFFFFF)) & 0xFFFFFFFF return state def _xor_bytes(data, key, regshift): state = (key * 257 + regshift * 17 + 0x5EC20) & 0xFFFFFFFF out = bytearray(data) for i in range(len(out)): state = _xor_state(state) out[i] ^= state & 0xFF return bytes(out) def _b_u8(v): return bytes((int(v) & 0xFF,)) def _b_u32(v): return struct.pack('= 2 and cap[0] in (0, 1): out.append([cap[0], cap[1] + regshift]) else: out.append(cap) rec['K2'] = out return rec def emit_binary(graph, enc=True, key=0, opmap=None, sbox=None, regshift=0): """Compact binary graph. It is XOR-streamed, then the outer LZMA stage compresses it further. No Lua table for opcodes/operands is emitted.""" opkey = key & 0xFF payload = bytearray() payload += _b_u32(opkey) payload += _b_u32(graph['root']) payload += _b_u32(len(graph['prototypes'])) for pid in sorted(graph['prototypes']): p = graph['prototypes'][pid] payload += _b_u32(pid) payload += _b_u32(p[1]) payload += _b_u8(1 if p.get(14) else 0) payload += _b_u32(p[6] + regshift) boxed = [x + regshift for x in p.get(13, [])] payload += _b_u32(len(boxed)) for x in boxed: payload += _b_u32(x) code = {} for pc, op in p[3].items(): rec = {'op': opmap[op] if opmap is not None else (op ^ opkey)} for name, field in (('A', 10), ('B', 5), ('C', 4), ('K1', 9), ('K2', 8), ('K3', 7)): if pc in p[field]: rec[name] = p[field][pc] code[pc] = _shift_rec(op, rec, regshift) maxpc = max(code) if code else 0 payload += _b_u32(maxpc) for pc in range(1, maxpc + 1): rec = code.get(pc) if rec is None: payload += _b_u32(0) payload += _b_u8(0) continue payload += _b_u32(rec['op']) present = [(i, name) for i, name in enumerate(('A', 'B', 'C', 'K1', 'K2', 'K3')) if name in rec] payload += _b_u8(len(present)) for fid, name in present: payload += _b_u8(fid) payload += _b_value(rec[name]) # opcode map: encoded -> original if opmap is not None: items = sorted(opmap.items(), key=lambda kv: kv[1]) payload += _b_u32(len(items)) for original, encoded in items: payload += _b_u32(encoded) payload += _b_u32(original) else: payload += _b_u32(0) if sbox is not None: inv = [0] * 256 for i, v in enumerate(sbox): inv[v] = i payload += b'\x01' payload += bytes(sbox) payload += bytes(inv) else: payload += b'\x00' payload += _b_u32(_b_crc(payload)) enc_payload = _xor_bytes(bytes(payload), key, regshift) key_raw = key ^ ((regshift * 37 + 0xA5) & 0xFF) header = b'SGB2' + bytes((1, key_raw & 0xFF, regshift & 0xFF, 1 if enc else 0)) return (header + enc_payload).hex() VM_PRELUDE = '''local __VM = (function() ''' def _runtime_names(key=0): rng = random.Random((int(key) * 2246822519 + 0x85EBCA6B) & 0xFFFFFFFF) out = {} for old in ('V', 'M', 'ENV', 'truthy', 'K', 'cell', 'mkclos', 'callwith', 'decode', 'exec', 'run', 'setprotos', 'graph'): while True: new = '_' + ''.join(rng.choice('abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789') for _ in range(9)) if new not in out.values(): out[old] = new break return out def make_sbox(key=0): rng = random.Random((int(key) * 2654435761 + 0xAB11C3D5) & 0xFFFFFFFF) box = list(range(256)) rng.shuffle(box) return box def make_opmap(graph, key=0): """Assign a unique 32-bit encoded opcode to every opcode in the graph.""" used = sorted({op for p in graph['prototypes'].values() for op in p[3].values()}) rng = random.Random((int(key) * 3266489917 + 0x27D4EB2F) & 0xFFFFFFFF) mapping, seen = {}, set() for op in used: while True: code = rng.randrange(0x10000, 0x7FFFFFFF) if code not in seen: seen.add(code) mapping[op] = code break return mapping def shuffle_handlers(vm: str, key=0) -> str: """Shuffle the independent opcode branch order inside the VM dispatch.""" lines = vm.splitlines(True) start = None for i, line in enumerate(lines): if line.startswith('\t\tif op =='): start = i break if start is None: return vm else_at = None for i in range(start + 1, len(lines)): if lines[i].startswith('\t\telse\n') or lines[i].startswith('\t\telse '): else_at = i break if else_at is None: return vm starts = [] for i in range(start, else_at): if lines[i].startswith('\t\tif op ==') or lines[i].startswith('\t\telseif op =='): starts.append(i) if len(starts) < 2: return vm blocks = [] for n, at in enumerate(starts): end = starts[n + 1] if n + 1 < len(starts) else else_at blocks.append(lines[at:end]) # Keep the first branch as the `if`; shuffle the remaining `elseif`s. rng = random.Random((int(key) * 2654435761 + 0x165667B1) & 0xFFFFFFFF) tail = blocks[1:] rng.shuffle(tail) new_lines = lines[:start] + blocks[0] + sum(tail, []) + lines[else_at:] return ''.join(new_lines) def add_decoy_handlers(vm: str, key=0) -> str: """Insert unreachable handler branches with build-specific opcode values.""" lines = vm.splitlines(True) for i, line in enumerate(lines): if line.startswith('\t\telse\n') or line.startswith('\t\telse '): rng = random.Random((int(key) * 2654435761 + 0x7F4A7C15) & 0xFFFFFFFF) insert_at = i decoys = [] for _ in range(rng.randrange(2, 5)): value = rng.randrange(0x20000000, 0x7F000000) decoys.append('\t\telseif op == %d then\n' % value) decoys.append('\t\t\tpc = pc + 1\n') return ''.join(lines[:insert_at] + decoys + lines[insert_at:]) return vm def protected_source(graph, enc=True, key=0, vm_source=None, opkey=None, hardened=False, binary=False): vm = vm_source if vm_source is not None else (HERE / 'vm_runtime.luau').read_text() if opkey is None: opkey = key & 0xFF names = field_names(key) runtime = _runtime_names(key) opmap = make_opmap(graph, key) if hardened else None sbox = make_sbox(key) if hardened else None regshift = (random.Random((int(key) * 2246822519 + 0x9E3779B9) & 0xFFFFFFFF).randrange(4, 17) if hardened else 0) if binary: handler_mode = vm_source is None if handler_mode: vm = handler_runtime.build_handler_runtime(vm, key) vm = re.sub(r'--\[(=*)\[.*?\]\1\]', '', vm, flags=re.S) vm = re.sub(r'--[^\n]*', '', vm) if not handler_mode: for old, new in runtime.items(): vm = re.sub(r'\b' + re.escape(old) + r'\b', new, vm) if hardened: vm = shuffle_handlers(vm, key) vm = add_decoy_handlers(vm, key) vm_name = '_' + ''.join(random.Random((int(key) * 1103515245 + 24680) & 0xFFFFFFFF).choice( 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789') for _ in range(10)) blob = emit_binary(graph, enc=enc, key=key, opmap=opmap, sbox=sbox, regshift=regshift) prelude = 'local %s = (function()\n' % vm_name decode_name = 'decode' if handler_mode else runtime['decode'] run_name = 'run' if handler_mode else runtime['run'] return (prelude + vm + '\nend)()\nlocal __G=' + luau_string(blob.encode('ascii')) + '\nreturn %s.load(%s.%s(__G)).%s()\n' % ( vm_name, vm_name, decode_name, run_name)) # Remove comments, rename the VM internals, and remap every exposed graph # field. The graph serializer uses the same per-build map, so the VM no # longer exposes a fixed schema. vm = re.sub(r'--\[(=*)\[.*?\]\1\]', '', vm, flags=re.S) vm = re.sub(r'--[^\n]*', '', vm) for old, new in names.items(): vm = re.sub(r'\.' + re.escape(old) + r'\b', '.' + new, vm) vm = vm.replace('V = {protos = {}, enc = false, key = 0, cache = {}}', 'V = {%s = {}, %s = false, %s = 0, cache = {}}' % ( names['protos'], names['enc'], names['key'])) for old, new in runtime.items(): vm = re.sub(r'\b' + re.escape(old) + r'\b', new, vm) if hardened: vm = shuffle_handlers(vm, key) vm = add_decoy_handlers(vm, key) vm_name = '_' + ''.join(random.Random((int(key) * 1103515245 + 12345) & 0xFFFFFFFF).choice( 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789') for _ in range(10)) payload = emit_payload(graph, enc=enc, key=key, opkey=opkey, names=names, opmap=opmap, sbox=sbox, regshift=regshift) prelude = 'local %s = (function()\n' % vm_name return (prelude + vm + '\nend)()\nlocal __G = ' + payload + '\nreturn %s.load(__G).%s()\n' % (vm_name, runtime['run'])) # -------------------------------------------------------------------------- # front end def compile_to_bytecode(source: str, optimize: int = 0, debug: int = 0) -> bytes: with tempfile.TemporaryDirectory(prefix='luraphc-') as tmp: src = Path(tmp) / 'input.luau' src.write_text(source, encoding='utf-8') proc = subprocess.run([LUAU_COMPILE, '--binary', '-O%d' % optimize, '-g%d' % debug, str(src)], capture_output=True) if proc.returncode != 0: raise CompileError('luau-compile failed:\n' + proc.stderr.decode('utf-8', 'replace')) return proc.stdout def compile_source(source: str, enc: bool = True, key: int = 0, keep_graph: str | None = None, hardened: bool = False, binary: bool = False) -> str: """Compile Luau source to a self-contained protected Luau chunk.""" source = macros.expand_macros(source) bc = luau_bc.parse(compile_to_bytecode(source)) graph = ModuleLowerer(bc).build() if keep_graph: Path(keep_graph).write_text(json.dumps(graph, indent=1, default=str)) return protected_source(graph, enc=enc, key=key, hardened=hardened, binary=binary) def compile_graph(source: str) -> dict: source = macros.expand_macros(source) bc = luau_bc.parse(compile_to_bytecode(source)) return ModuleLowerer(bc).build() def main(argv=None): import argparse ap = argparse.ArgumentParser(description=__doc__) ap.add_argument('input') ap.add_argument('-o', '--output') ap.add_argument('--graph') ap.add_argument('--no-encode', action='store_true') ap.add_argument('--hardened', action='store_true') ap.add_argument('--binary-graph', action='store_true') ap.add_argument('--key', type=int, default=0) ap.add_argument('--run', action='store_true') args = ap.parse_args(argv) source = Path(args.input).read_text(encoding='utf-8') graph = compile_graph(source) if args.graph: Path(args.graph).write_text(json.dumps(graph, indent=1, default=str)) out = protected_source(graph, enc=not args.no_encode, key=args.key, hardened=args.hardened, binary=args.binary_graph) dest = Path(args.output or (args.input + '.protected.luau')) dest.write_text(out) print('wrote %s (%d bytes, %d prototypes, %d instructions)' % ( dest, len(out), len(graph['prototypes']), sum(len(p[3]) for p in graph['prototypes'].values()))) if args.run: proc = subprocess.run([LUAU, str(dest)], capture_output=True, text=True) sys.stdout.write(proc.stdout) sys.stderr.write(proc.stderr) return proc.returncode return 0 if __name__ == '__main__': raise SystemExit(main())