mirror of
https://github.com/bikini/patchwork
synced 2026-06-27 08:08:41 +00:00
Add files via upload
This commit is contained in:
bikini
@@ -0,0 +1,3 @@
|
||||
from .core import Obfuscator, obfuscate, obfuscate_file
|
||||
__version__ = '0.1.0'
|
||||
__all__ = ['Obfuscator', 'obfuscate', 'obfuscate_file']
|
||||
@@ -0,0 +1,3 @@
|
||||
from .cli import main
|
||||
if __name__ == '__main__':
|
||||
raise SystemExit(main())
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,44 @@
|
||||
from __future__ import annotations
|
||||
import argparse
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from .core import Obfuscator
|
||||
|
||||
def _parser() -> argparse.ArgumentParser:
|
||||
p = argparse.ArgumentParser(prog='patchwork')
|
||||
p.add_argument('input')
|
||||
p.add_argument('-o', '--output')
|
||||
p.add_argument('--seed', type=int, default=None)
|
||||
p.add_argument('--layers', type=int, default=3)
|
||||
p.add_argument('--stage2-layers', type=int, default=3, dest='stage2_layers')
|
||||
p.add_argument('--keep', action='append', default=[], metavar='NAME')
|
||||
p.add_argument('--no-rename', action='store_true')
|
||||
p.add_argument('--no-encrypt-strings', action='store_true')
|
||||
p.add_argument('--no-obfuscate-numbers', action='store_true')
|
||||
p.add_argument('--no-opaque', action='store_true')
|
||||
p.add_argument('--no-mba', action='store_true')
|
||||
p.add_argument('--no-junk', action='store_true')
|
||||
p.add_argument('--no-lazy', action='store_true')
|
||||
p.add_argument('--no-anti-debug', action='store_true')
|
||||
p.add_argument('--quiet', '-q', action='store_true')
|
||||
return p
|
||||
|
||||
def main(argv: list[str] | None=None) -> int:
|
||||
args = _parser().parse_args(argv)
|
||||
inp = Path(args.input)
|
||||
if not inp.is_file():
|
||||
print(f'patchwork: input file not found: {inp}', file=sys.stderr)
|
||||
return 2
|
||||
src = inp.read_text(encoding='utf-8')
|
||||
obf = Obfuscator(seed=args.seed, rename=not args.no_rename, encrypt_strings=not args.no_encrypt_strings, obfuscate_numbers=not args.no_obfuscate_numbers, opaque_predicates=not args.no_opaque, mba=not args.no_mba, junk_branches=not args.no_junk, lazy_funcs=not args.no_lazy, anti_debug=not args.no_anti_debug, layers=args.layers, stage2_layers=args.stage2_layers, keep=set(args.keep))
|
||||
out_src = obf.obfuscate(src)
|
||||
out_path = Path(args.output) if args.output else inp.with_name(inp.stem + '_obf.py')
|
||||
out_path.write_text(out_src, encoding='utf-8')
|
||||
if not args.quiet:
|
||||
in_size = len(src)
|
||||
out_size = len(out_src)
|
||||
ratio = out_size / max(in_size, 1)
|
||||
print(f'patchwork: {inp} -> {out_path}\n input: {in_size:,} bytes\n output: {out_size:,} bytes ({ratio:.1f}x)\n seed: {obf.seed}\n layers: {obf.layers}', file=sys.stderr)
|
||||
return 0
|
||||
if __name__ == '__main__':
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,79 @@
|
||||
from __future__ import annotations
|
||||
import ast
|
||||
from pathlib import Path
|
||||
from .lazy import encrypt_user_functions
|
||||
from .loader import build_loader
|
||||
from .packer import pack
|
||||
from .transforms import IdentifierRenamer, NumberObfuscator, OpaquePredicateInjector, StringEncryptor, build_decrypt_helper, collect_skip_names
|
||||
from .transforms.junk import JunkBranchInjector
|
||||
from .transforms.mba import MBATransformer
|
||||
from .transforms.opaque import build_seed_stmt
|
||||
from .util import make_rng
|
||||
_SEED_NAME = '_pw_seed_value'
|
||||
_DEC_S_NAME = '_pw_dec_str'
|
||||
_DEC_B_NAME = '_pw_dec_bytes'
|
||||
_RESOLVE_NAME = '__pw_resolve_lazy__'
|
||||
|
||||
class Obfuscator:
|
||||
|
||||
def __init__(self, *, seed: int | None=None, rename: bool=True, encrypt_strings: bool=True, obfuscate_numbers: bool=True, opaque_predicates: bool=True, mba: bool=True, junk_branches: bool=True, lazy_funcs: bool=True, anti_debug: bool=True, layers: int=3, stage2_layers: int=3, keep: set[str] | None=None):
|
||||
self.rng, self.seed = make_rng(seed)
|
||||
self.rename = rename
|
||||
self.encrypt_strings = encrypt_strings
|
||||
self.obfuscate_numbers = obfuscate_numbers
|
||||
self.opaque_predicates = opaque_predicates
|
||||
self.mba = mba
|
||||
self.junk_branches = junk_branches
|
||||
self.lazy_funcs = lazy_funcs
|
||||
self.anti_debug = anti_debug
|
||||
self.layers = max(1, int(layers))
|
||||
self.stage2_layers = max(1, int(stage2_layers))
|
||||
self.keep_extra = set(keep or ())
|
||||
|
||||
def obfuscate(self, source: str) -> str:
|
||||
tree = ast.parse(source)
|
||||
skip = collect_skip_names(tree)
|
||||
seed_used = False
|
||||
if self.opaque_predicates:
|
||||
opi = OpaquePredicateInjector(self.rng, _SEED_NAME)
|
||||
tree = opi.visit(tree)
|
||||
seed_used = seed_used or opi.injected
|
||||
if self.junk_branches:
|
||||
jbi = JunkBranchInjector(self.rng, _SEED_NAME)
|
||||
tree = jbi.visit(tree)
|
||||
seed_used = seed_used or jbi.injected
|
||||
if seed_used:
|
||||
tree.body.insert(0, build_seed_stmt(_SEED_NAME))
|
||||
if self.mba:
|
||||
tree = MBATransformer(self.rng).visit(tree)
|
||||
if self.obfuscate_numbers:
|
||||
tree = NumberObfuscator(self.rng).visit(tree)
|
||||
if self.encrypt_strings:
|
||||
enc = StringEncryptor(self.rng, _DEC_S_NAME, _DEC_B_NAME)
|
||||
tree = enc.visit(tree)
|
||||
tree.body[:0] = build_decrypt_helper(_DEC_S_NAME, _DEC_B_NAME)
|
||||
if self.rename:
|
||||
renamer = IdentifierRenamer(self.rng, keep=skip | self.keep_extra | {_RESOLVE_NAME})
|
||||
tree = renamer.visit(tree)
|
||||
ast.fix_missing_locations(tree)
|
||||
user_code = compile(tree, '<patchwork>', 'exec')
|
||||
if self.lazy_funcs:
|
||||
user_code, lazy_blobs = encrypt_user_functions(user_code, self.rng, _RESOLVE_NAME)
|
||||
else:
|
||||
lazy_blobs = []
|
||||
user_payload, user_keys = pack(user_code, self.rng, layers=self.layers)
|
||||
return build_loader(user_payload, user_keys, lazy_blobs, resolver_name=_RESOLVE_NAME, rng=self.rng, anti_debug=self.anti_debug, stage2_layers=self.stage2_layers)
|
||||
|
||||
def obfuscate(source: str, **kwargs) -> str:
|
||||
return Obfuscator(**kwargs).obfuscate(source)
|
||||
|
||||
def obfuscate_file(input_path: str | Path, output_path: str | Path | None=None, **kwargs) -> Path:
|
||||
inp = Path(input_path)
|
||||
src = inp.read_text(encoding='utf-8')
|
||||
out = obfuscate(src, **kwargs)
|
||||
if output_path is None:
|
||||
outp = inp.with_name(inp.stem + '_obf.py')
|
||||
else:
|
||||
outp = Path(output_path)
|
||||
outp.write_text(out, encoding='utf-8')
|
||||
return outp
|
||||
@@ -0,0 +1,48 @@
|
||||
from __future__ import annotations
|
||||
import hashlib
|
||||
import random
|
||||
|
||||
def keystream(key: bytes, length: int) -> bytes:
|
||||
out = bytearray()
|
||||
ctr = 0
|
||||
while len(out) < length:
|
||||
out.extend(hashlib.sha256(key + ctr.to_bytes(8, 'big')).digest())
|
||||
ctr += 1
|
||||
return bytes(out[:length])
|
||||
|
||||
def stream_xor(data: bytes, key: bytes) -> bytes:
|
||||
ks = keystream(key, len(data))
|
||||
return bytes((a ^ b for a, b in zip(data, ks)))
|
||||
|
||||
def make_perm(key: bytes) -> bytes:
|
||||
seed = int.from_bytes(hashlib.sha256(key).digest(), 'big')
|
||||
rng = random.Random(seed)
|
||||
perm = list(range(256))
|
||||
rng.shuffle(perm)
|
||||
return bytes(perm)
|
||||
|
||||
def invert_perm(perm: bytes) -> bytes:
|
||||
inv = bytearray(256)
|
||||
for i, p in enumerate(perm):
|
||||
inv[p] = i
|
||||
return bytes(inv)
|
||||
|
||||
def perm_apply(data: bytes, key: bytes) -> bytes:
|
||||
perm = make_perm(key)
|
||||
return bytes((perm[b] for b in data))
|
||||
|
||||
def perm_invert(data: bytes, key: bytes) -> bytes:
|
||||
inv = invert_perm(make_perm(key))
|
||||
return bytes((inv[b] for b in data))
|
||||
|
||||
def _rot_left(b: int, n: int) -> int:
|
||||
n &= 7
|
||||
if not n:
|
||||
return b
|
||||
return (b << n | b >> 8 - n) & 255
|
||||
|
||||
def rot_apply(data: bytes, key: bytes) -> bytes:
|
||||
return bytes((_rot_left(b, key[i % len(key)]) for i, b in enumerate(data)))
|
||||
|
||||
def rot_invert(data: bytes, key: bytes) -> bytes:
|
||||
return bytes((_rot_left(b, -key[i % len(key)] & 7) for i, b in enumerate(data)))
|
||||
@@ -0,0 +1,88 @@
|
||||
from __future__ import annotations
|
||||
import marshal
|
||||
import random
|
||||
import types
|
||||
from .crypto import stream_xor
|
||||
from .util import gen_bytes
|
||||
LAZY_PREFIX = '<pw'
|
||||
LAZY_SUFFIX = '>'
|
||||
_CO_GENERATOR = 0x20
|
||||
_CO_COROUTINE = 0x100
|
||||
_CO_ASYNC_GENERATOR = 0x200
|
||||
|
||||
|
||||
def _classify_code(code: types.CodeType) -> str:
|
||||
if not code.co_flags & 1:
|
||||
return 'class'
|
||||
if code.co_name.startswith('<') and code.co_name.endswith('>'):
|
||||
return 'special'
|
||||
return 'function'
|
||||
|
||||
|
||||
def _stub_source(kind: str, idx: int, resolver_name: str) -> str:
|
||||
if kind == 'async_gen':
|
||||
return (
|
||||
f'async def _stub(*_a, **_k):\n'
|
||||
f' async for _v in {resolver_name}({idx}, _a, _k):\n'
|
||||
f' yield _v\n'
|
||||
)
|
||||
if kind == 'coroutine':
|
||||
return (
|
||||
f'async def _stub(*_a, **_k):\n'
|
||||
f' return await {resolver_name}({idx}, _a, _k)\n'
|
||||
)
|
||||
if kind == 'generator':
|
||||
return (
|
||||
f'def _stub(*_a, **_k):\n'
|
||||
f' yield from {resolver_name}({idx}, _a, _k)\n'
|
||||
)
|
||||
return (
|
||||
f'def _stub(*_a, **_k):\n'
|
||||
f' return {resolver_name}({idx}, _a, _k)\n'
|
||||
)
|
||||
|
||||
|
||||
def _kind_for(code: types.CodeType) -> str:
|
||||
f = code.co_flags
|
||||
if f & _CO_ASYNC_GENERATOR:
|
||||
return 'async_gen'
|
||||
if f & _CO_COROUTINE:
|
||||
return 'coroutine'
|
||||
if f & _CO_GENERATOR:
|
||||
return 'generator'
|
||||
return 'function'
|
||||
|
||||
|
||||
def _make_stub_code(idx: int, original: types.CodeType, resolver_name: str) -> types.CodeType:
|
||||
kind = _kind_for(original)
|
||||
src = _stub_source(kind, idx, resolver_name)
|
||||
mod = compile(src, '<pwstub>', 'exec')
|
||||
inner = next((c for c in mod.co_consts if isinstance(c, types.CodeType)))
|
||||
return inner.replace(
|
||||
co_name=f'{LAZY_PREFIX}{idx:x}{LAZY_SUFFIX}',
|
||||
co_freevars=original.co_freevars,
|
||||
co_cellvars=original.co_cellvars,
|
||||
)
|
||||
|
||||
|
||||
def encrypt_user_functions(top: types.CodeType, rng: random.Random, resolver_name: str, key_size: int=24) -> tuple[types.CodeType, list[tuple[bytes, bytes]]]:
|
||||
blobs: list[tuple[bytes, bytes]] = []
|
||||
|
||||
def transform(code: types.CodeType, parent_kind: str) -> types.CodeType:
|
||||
new_consts: list = []
|
||||
for c in code.co_consts:
|
||||
if isinstance(c, types.CodeType):
|
||||
ck = _classify_code(c)
|
||||
if parent_kind in ('module', 'class') and ck == 'function':
|
||||
idx = len(blobs)
|
||||
key = gen_bytes(rng, key_size)
|
||||
enc = stream_xor(marshal.dumps(c), key)
|
||||
blobs.append((key, enc))
|
||||
new_consts.append(_make_stub_code(idx, c, resolver_name))
|
||||
else:
|
||||
new_consts.append(transform(c, ck))
|
||||
else:
|
||||
new_consts.append(c)
|
||||
return code.replace(co_consts=tuple(new_consts))
|
||||
new_top = transform(top, 'module')
|
||||
return (new_top, blobs)
|
||||
@@ -0,0 +1,143 @@
|
||||
from __future__ import annotations
|
||||
import base64
|
||||
import marshal
|
||||
import random
|
||||
import zlib
|
||||
from . import crypto
|
||||
from .packer import pack
|
||||
from .util import gen_bytes, gen_name
|
||||
_MOD_KEYS = ('sys', 'os', 'marshal', 'zlib', 'base64', 'hashlib', 'rand_mod')
|
||||
_FUNC_KEYS = ('kstream', 'xor_dec', 'perm_inv', 'perm_dec', 'rot_dec')
|
||||
_VAR_KEYS = ('data', 'code_obj', 'ns', 'blobs', 'tmp', 'i', 'k', 'p', 'g')
|
||||
|
||||
def _bytes_repr(b: bytes) -> str:
|
||||
return f'bytes.fromhex({b.hex()!r})'
|
||||
|
||||
def _chunked_bytes(b: bytes, rng: random.Random) -> str:
|
||||
if len(b) < 64:
|
||||
return _bytes_repr(b)
|
||||
nchunks = rng.randint(4, 9)
|
||||
chunk_size = (len(b) + nchunks - 1) // nchunks
|
||||
chunks: list[bytes] = []
|
||||
pos = 0
|
||||
while pos < len(b):
|
||||
chunks.append(b[pos:pos + chunk_size])
|
||||
pos += chunk_size
|
||||
order = list(range(len(chunks)))
|
||||
rng.shuffle(order)
|
||||
shuffled = [chunks[i] for i in order]
|
||||
inv = [0] * len(order)
|
||||
for i, o in enumerate(order):
|
||||
inv[o] = i
|
||||
parts_src = ', '.join((_bytes_repr(c) for c in shuffled))
|
||||
perm_src = ', '.join((str(x) for x in inv))
|
||||
return f"b''.join((({parts_src},)[__pwi] for __pwi in ({perm_src},)))"
|
||||
|
||||
def _junk(rng: random.Random) -> str:
|
||||
target = gen_name(rng)
|
||||
n = rng.randint(0, 1 << 16)
|
||||
forms = [f'{target} = (lambda _x: _x ^ _x)({n})', f'{target} = sum(({n},))', f'{target} = ({n} & {n}) | 0', f'{target} = pow({n} & 0xFF, 1, 1) + 0', f'{target} = len(()) + {n} - {n}', f'{target} = ({n} * 0) ^ 0', f'{target} = (~{n} ^ ~{n}) & 0xFF']
|
||||
return rng.choice(forms)
|
||||
|
||||
def _name_pool(rng: random.Random, keys: tuple[str, ...]) -> dict[str, str]:
|
||||
used: set[str] = set()
|
||||
out: dict[str, str] = {}
|
||||
for k in keys:
|
||||
n = gen_name(rng, used=used)
|
||||
used.add(n)
|
||||
out[k] = n
|
||||
return out
|
||||
|
||||
def _decrypt_call(N: dict[str, str], op: str, var: str, key: bytes) -> str:
|
||||
kbytes = _bytes_repr(key)
|
||||
if op == 'xor':
|
||||
return f"{var} = {N['xor_dec']}({var}, {kbytes})"
|
||||
if op == 'perm':
|
||||
return f"{var} = {N['perm_dec']}({var}, {kbytes})"
|
||||
if op == 'rot':
|
||||
return f"{var} = {N['rot_dec']}({var}, {kbytes})"
|
||||
raise ValueError(op)
|
||||
|
||||
def _emit_helpers(N: dict[str, str]) -> list[str]:
|
||||
return [f"def {N['kstream']}(k, n):\n r = bytearray()\n c = 0\n while len(r) < n:\n r += {N['hashlib']}.sha256(k + c.to_bytes(8, 'big')).digest()\n c += 1\n return bytes(r[:n])", f"def {N['xor_dec']}(d, k):\n s = {N['kstream']}(k, len(d))\n return bytes(a ^ b for a, b in zip(d, s))", f"def {N['perm_inv']}(k):\n g = {N['rand_mod']}.Random(int.from_bytes({N['hashlib']}.sha256(k).digest(), 'big'))\n p = list(range(256))\n g.shuffle(p)\n inv = bytearray(256)\n for i, v in enumerate(p):\n inv[v] = i\n return bytes(inv)", f"def {N['perm_dec']}(d, k):\n inv = {N['perm_inv']}(k)\n return bytes(inv[b] for b in d)", f"def {N['rot_dec']}(d, k):\n out = bytearray()\n L = len(k)\n for i, b in enumerate(d):\n n = (-k[i % L]) & 7\n if n:\n out.append(((b << n) | (b >> (8 - n))) & 0xFF)\n else:\n out.append(b)\n return bytes(out)"]
|
||||
|
||||
def _emit_imports(N: dict[str, str], rng: random.Random) -> list[str]:
|
||||
lines = [f"{N['sys']} = __import__('sys')", f"{N['os']} = __import__('os')", f"{N['marshal']} = __import__('marshal')", f"{N['zlib']} = __import__('zlib')", f"{N['base64']} = __import__('base64')", f"{N['hashlib']} = __import__('hashlib')", f"{N['rand_mod']} = __import__('random')"]
|
||||
rng.shuffle(lines)
|
||||
return lines
|
||||
|
||||
def _emit_anti_debug(N: dict[str, str], rng: random.Random) -> list[str]:
|
||||
checks = [f"({N['sys']}.gettrace() is None) or {N['os']}._exit(1)", f"({N['sys']}.getprofile() is None) or {N['os']}._exit(1)", f"all(_m not in {N['sys']}.modules for _m in ('pdb','bdb','pydevd','debugpy','pydevd_bundle','pydevd_bundle.pydevd_resolver','ptvsd','rpdb','wdb')) or {N['os']}._exit(1)", f"({N['os']}.environ.get('PYTHONBREAKPOINT', '') in ('', '0', 'pdb.set_trace')) or {N['os']}._exit(1)"]
|
||||
rng.shuffle(checks)
|
||||
return checks
|
||||
|
||||
def _emit_audit_hook(N: dict[str, str]) -> list[str]:
|
||||
return [f"def __pw_audit(event, args):\n if event == 'import':\n _n = args[0] if args else ''\n if _n in ('pdb','bdb','pydevd','debugpy','pydevd_bundle','ptvsd','rpdb','wdb'):\n {N['os']}._exit(1)\n elif event in ('sys.settrace','sys.setprofile'):\n {N['os']}._exit(1)", f"{N['sys']}.addaudithook(__pw_audit)"]
|
||||
|
||||
def _emit_frame_check(N: dict[str, str]) -> list[str]:
|
||||
return [f"_f = {N['sys']}._getframe(0)\nwhile _f is not None:\n _gn = _f.f_globals.get('__name__', '')\n if _gn.startswith(('pdb', 'bdb', 'pydevd', 'debugpy', '_pydev')):\n {N['os']}._exit(1)\n _f = _f.f_back"]
|
||||
|
||||
def _build_stage2_source(user_payload: bytes, user_layer_keys: list[tuple[str, bytes]], lazy_blobs: list[tuple[bytes, bytes]], resolver_name: str, mod_names: dict[str, str], rng: random.Random, *, anti_debug: bool=True) -> str:
|
||||
N = mod_names
|
||||
F = _name_pool(rng, _FUNC_KEYS)
|
||||
V = _name_pool(rng, _VAR_KEYS)
|
||||
parts: list[str] = []
|
||||
parts.append("_pw_types = __import__('types')")
|
||||
parts.extend(_emit_helpers({**N, **F}))
|
||||
if anti_debug:
|
||||
parts.extend(_emit_anti_debug({**N, **F}, rng))
|
||||
parts.append(_junk(rng))
|
||||
if lazy_blobs:
|
||||
blob_lines = []
|
||||
for k, e in lazy_blobs:
|
||||
blob_lines.append(f'({_chunked_bytes(k, rng)}, {_chunked_bytes(e, rng)})')
|
||||
parts.append(f"__pw_blobs = [{', '.join(blob_lines)}]")
|
||||
else:
|
||||
parts.append('__pw_blobs = []')
|
||||
parts.append(f"_pw_code_cache = {{}}\n_pw_done = set()\ndef _pw_find_func(g, target_code):\n seen = set()\n stack = list(g.values())\n while stack:\n v = stack.pop()\n oid = id(v)\n if oid in seen:\n continue\n seen.add(oid)\n if isinstance(v, _pw_types.FunctionType) and v.__code__ is target_code:\n return v\n if isinstance(v, (classmethod, staticmethod)):\n inner = v.__func__\n if isinstance(inner, _pw_types.FunctionType) and inner.__code__ is target_code:\n return inner\n stack.append(inner)\n continue\n if isinstance(v, property):\n for fn in (v.fget, v.fset, v.fdel):\n if fn is not None:\n stack.append(fn)\n continue\n if isinstance(v, type):\n for vv in v.__dict__.values():\n stack.append(vv)\n return None\ndef {resolver_name}(idx, args, kw):\n fr = {N['sys']}._getframe(1)\n g = fr.f_globals\n stub_code = fr.f_code\n if idx not in _pw_code_cache:\n key, enc = __pw_blobs[idx]\n _pw_code_cache[idx] = {N['marshal']}.loads({F['xor_dec']}(enc, key))\n real = _pw_code_cache[idx]\n target = _pw_find_func(g, stub_code)\n if target is not None and id(target) not in _pw_done:\n target.__code__ = real\n _pw_done.add(id(target))\n return target(*args, **kw)\n fn = _pw_types.FunctionType(real, g)\n return fn(*args, **kw)")
|
||||
user_encoded = base64.b85encode(user_payload)
|
||||
parts.append(f"{V['data']} = {N['base64']}.b85decode({_chunked_bytes(user_encoded, rng)})")
|
||||
for op, key in reversed(user_layer_keys):
|
||||
parts.append(_decrypt_call({**N, **F}, op, V['data'], key))
|
||||
parts.append(f"{V['data']} = {N['zlib']}.decompress({V['data']})")
|
||||
parts.append(f"{V['code_obj']} = {N['marshal']}.loads({V['data']})")
|
||||
parts.append(_junk(rng))
|
||||
if anti_debug:
|
||||
parts.append(f"({N['sys']}.gettrace() is None) or {N['os']}._exit(1)")
|
||||
parts.append(f"{V['ns']} = {{'__name__': '__main__', '__builtins__': __builtins__, '__file__': __file__, '__package__': None, '__doc__': None, '{resolver_name}': {resolver_name}}}")
|
||||
parts.append(f"exec({V['code_obj']}, {V['ns']})")
|
||||
return '\n'.join(parts) + '\n'
|
||||
|
||||
def build_loader(user_payload: bytes, user_layer_keys: list[tuple[str, bytes]], lazy_blobs: list[tuple[bytes, bytes]], resolver_name: str, rng: random.Random, *, anti_debug: bool=True, stage2_layers: int=3, junk_count: int=8) -> str:
|
||||
N = _name_pool(rng, _MOD_KEYS)
|
||||
F = _name_pool(rng, _FUNC_KEYS)
|
||||
V = _name_pool(rng, _VAR_KEYS)
|
||||
stage2_src = _build_stage2_source(user_payload, user_layer_keys, lazy_blobs, resolver_name=resolver_name, mod_names=N, rng=rng, anti_debug=anti_debug)
|
||||
stage2_code = compile(stage2_src, '<patchwork:stage2>', 'exec')
|
||||
stage2_payload, stage2_keys = pack(stage2_code, rng, layers=stage2_layers)
|
||||
parts: list[str] = []
|
||||
parts.extend(_emit_imports(N, rng))
|
||||
parts.append(_junk(rng))
|
||||
if anti_debug:
|
||||
parts.extend(_emit_anti_debug({**N, **F}, rng))
|
||||
parts.extend(_emit_audit_hook(N))
|
||||
parts.extend(_emit_frame_check(N))
|
||||
parts.append(_junk(rng))
|
||||
parts.extend(_emit_helpers({**N, **F}))
|
||||
parts.append(_junk(rng))
|
||||
stage2_encoded = base64.b85encode(stage2_payload)
|
||||
parts.append(f"{V['data']} = {N['base64']}.b85decode({_chunked_bytes(stage2_encoded, rng)})")
|
||||
for op, key in reversed(stage2_keys):
|
||||
parts.append(_decrypt_call({**N, **F}, op, V['data'], key))
|
||||
parts.append(f"{V['data']} = {N['zlib']}.decompress({V['data']})")
|
||||
if anti_debug:
|
||||
parts.append(f"({N['sys']}.gettrace() is None) or {N['os']}._exit(1)")
|
||||
parts.append(f"{V['code_obj']} = {N['marshal']}.loads({V['data']})")
|
||||
parts.append(_junk(rng))
|
||||
parts.append(f"{V['ns']} = {{'__builtins__': __builtins__, '__file__': __file__, '__name__': __name__, '{N['sys']}': {N['sys']}, '{N['os']}': {N['os']}, '{N['marshal']}': {N['marshal']}, '{N['zlib']}': {N['zlib']}, '{N['base64']}': {N['base64']}, '{N['hashlib']}': {N['hashlib']}, '{N['rand_mod']}': {N['rand_mod']}}}")
|
||||
parts.append(f"exec({V['code_obj']}, {V['ns']})")
|
||||
body = list(parts)
|
||||
for _ in range(junk_count):
|
||||
idx = rng.randint(0, len(body))
|
||||
body.insert(idx, _junk(rng))
|
||||
return '\n'.join(body) + '\n'
|
||||
@@ -0,0 +1,23 @@
|
||||
from __future__ import annotations
|
||||
import marshal
|
||||
import random
|
||||
import zlib
|
||||
from . import crypto
|
||||
from .util import gen_bytes
|
||||
CIPHER_OPS = ('xor', 'perm', 'rot')
|
||||
|
||||
def pack(code_obj, rng: random.Random, layers: int=3) -> tuple[bytes, list[tuple[str, bytes]]]:
|
||||
data = marshal.dumps(code_obj)
|
||||
data = zlib.compress(data, 9)
|
||||
layer_keys: list[tuple[str, bytes]] = []
|
||||
for _ in range(layers):
|
||||
op = rng.choice(CIPHER_OPS)
|
||||
key = gen_bytes(rng, rng.randint(20, 40))
|
||||
if op == 'xor':
|
||||
data = crypto.stream_xor(data, key)
|
||||
elif op == 'perm':
|
||||
data = crypto.perm_apply(data, key)
|
||||
elif op == 'rot':
|
||||
data = crypto.rot_apply(data, key)
|
||||
layer_keys.append((op, key))
|
||||
return (data, layer_keys)
|
||||
@@ -0,0 +1,5 @@
|
||||
from .identifiers import IdentifierRenamer, collect_skip_names
|
||||
from .strings import StringEncryptor, build_decrypt_helper
|
||||
from .numbers import NumberObfuscator
|
||||
from .opaque import OpaquePredicateInjector
|
||||
__all__ = ['IdentifierRenamer', 'collect_skip_names', 'StringEncryptor', 'build_decrypt_helper', 'NumberObfuscator', 'OpaquePredicateInjector']
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,167 @@
|
||||
from __future__ import annotations
|
||||
import ast
|
||||
import builtins
|
||||
import keyword
|
||||
import random
|
||||
from ..util import gen_name
|
||||
_DUNDERS = {'__name__', '__main__', '__file__', '__doc__', '__builtins__', '__package__', '__path__', '__loader__', '__spec__', '__cached__', '__all__', '__init__', '__new__', '__del__', '__init_subclass__', '__subclasshook__', '__set_name__', '__class_getitem__', '__enter__', '__exit__', '__aenter__', '__aexit__', '__await__', '__aiter__', '__anext__', '__call__', '__repr__', '__str__', '__bytes__', '__format__', '__hash__', '__bool__', '__len__', '__length_hint__', '__sizeof__', '__dir__', '__eq__', '__ne__', '__lt__', '__le__', '__gt__', '__ge__', '__add__', '__sub__', '__mul__', '__truediv__', '__floordiv__', '__mod__', '__divmod__', '__pow__', '__lshift__', '__rshift__', '__and__', '__or__', '__xor__', '__radd__', '__rsub__', '__rmul__', '__rtruediv__', '__rfloordiv__', '__rmod__', '__rdivmod__', '__rpow__', '__rlshift__', '__rrshift__', '__rand__', '__ror__', '__rxor__', '__iadd__', '__isub__', '__imul__', '__itruediv__', '__ifloordiv__', '__imod__', '__ipow__', '__ilshift__', '__irshift__', '__iand__', '__ior__', '__ixor__', '__matmul__', '__rmatmul__', '__imatmul__', '__neg__', '__pos__', '__abs__', '__invert__', '__int__', '__float__', '__complex__', '__round__', '__index__', '__trunc__', '__floor__', '__ceil__', '__getattr__', '__setattr__', '__delattr__', '__getattribute__', '__getitem__', '__setitem__', '__delitem__', '__missing__', '__iter__', '__next__', '__reversed__', '__contains__', '__copy__', '__deepcopy__', '__reduce__', '__reduce_ex__', '__getstate__', '__setstate__', '__getnewargs__', '__getnewargs_ex__', '__class__', '__instancecheck__', '__subclasscheck__', '__slots__', '__dict__', '__weakref__', '__module__', '__qualname__', '__defaults__', '__kwdefaults__', '__annotations__', '__wrapped__', '__signature__', '__match_args__', '__post_init__', '__fspath__', '__index__', '__buffer__', '__release_buffer__'}
|
||||
RESERVED: set[str] = set(dir(builtins)) | set(keyword.kwlist) | _DUNDERS | {'self', 'cls', 'metaclass'}
|
||||
|
||||
def collect_skip_names(tree: ast.AST) -> set[str]:
|
||||
skip: set[str] = set()
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Assign):
|
||||
for target in node.targets:
|
||||
if isinstance(target, ast.Name) and target.id == '__all__':
|
||||
if isinstance(node.value, (ast.List, ast.Tuple, ast.Set)):
|
||||
for elt in node.value.elts:
|
||||
if isinstance(elt, ast.Constant) and isinstance(elt.value, str):
|
||||
skip.add(elt.value)
|
||||
elif isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and (node.func.id in ('getattr', 'setattr', 'hasattr', 'delattr')):
|
||||
if len(node.args) >= 2:
|
||||
arg = node.args[1]
|
||||
if isinstance(arg, ast.Constant) and isinstance(arg.value, str):
|
||||
skip.add(arg.value)
|
||||
elif isinstance(node, ast.ClassDef):
|
||||
for stmt in node.body:
|
||||
if isinstance(stmt, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
skip.add(stmt.name)
|
||||
elif isinstance(stmt, ast.Assign):
|
||||
for tgt in stmt.targets:
|
||||
if isinstance(tgt, ast.Name):
|
||||
skip.add(tgt.id)
|
||||
elif isinstance(tgt, (ast.Tuple, ast.List)):
|
||||
for elt in tgt.elts:
|
||||
if isinstance(elt, ast.Name):
|
||||
skip.add(elt.id)
|
||||
elif isinstance(stmt, ast.AnnAssign) and isinstance(stmt.target, ast.Name):
|
||||
skip.add(stmt.target.id)
|
||||
elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda)):
|
||||
args = node.args
|
||||
for a in args.args + args.posonlyargs + args.kwonlyargs:
|
||||
skip.add(a.arg)
|
||||
if args.vararg:
|
||||
skip.add(args.vararg.arg)
|
||||
if args.kwarg:
|
||||
skip.add(args.kwarg.arg)
|
||||
return skip
|
||||
|
||||
class IdentifierRenamer(ast.NodeTransformer):
|
||||
|
||||
def __init__(self, rng: random.Random, keep: set[str] | None=None):
|
||||
self.rng = rng
|
||||
self.mapping: dict[str, str] = {}
|
||||
self.reserved = RESERVED | (keep or set())
|
||||
self._used_outputs: set[str] = set()
|
||||
|
||||
def _new(self) -> str:
|
||||
n = gen_name(self.rng, used=self._used_outputs)
|
||||
self._used_outputs.add(n)
|
||||
return n
|
||||
|
||||
def _rename(self, name: str) -> str:
|
||||
if not name or name in self.reserved:
|
||||
return name
|
||||
if name.startswith('__') and name.endswith('__') and (len(name) >= 4):
|
||||
return name
|
||||
if name not in self.mapping:
|
||||
self.mapping[name] = self._new()
|
||||
return self.mapping[name]
|
||||
|
||||
def visit_Name(self, node: ast.Name) -> ast.AST:
|
||||
node.id = self._rename(node.id)
|
||||
return node
|
||||
|
||||
def visit_FunctionDef(self, node: ast.FunctionDef) -> ast.AST:
|
||||
node.name = self._rename(node.name)
|
||||
node.decorator_list = [self.visit(d) for d in node.decorator_list]
|
||||
if node.returns is not None:
|
||||
node.returns = self.visit(node.returns)
|
||||
node.args = self.visit(node.args)
|
||||
node.body = [self.visit(stmt) for stmt in node.body]
|
||||
return node
|
||||
visit_AsyncFunctionDef = visit_FunctionDef
|
||||
|
||||
def visit_ClassDef(self, node: ast.ClassDef) -> ast.AST:
|
||||
node.name = self._rename(node.name)
|
||||
node.bases = [self.visit(b) for b in node.bases]
|
||||
for kw in node.keywords:
|
||||
kw.value = self.visit(kw.value)
|
||||
node.decorator_list = [self.visit(d) for d in node.decorator_list]
|
||||
node.body = [self.visit(stmt) for stmt in node.body]
|
||||
return node
|
||||
|
||||
def visit_Lambda(self, node: ast.Lambda) -> ast.AST:
|
||||
node.args = self.visit(node.args)
|
||||
node.body = self.visit(node.body)
|
||||
return node
|
||||
|
||||
def visit_arg(self, node: ast.arg) -> ast.AST:
|
||||
node.arg = self._rename(node.arg)
|
||||
if node.annotation is not None:
|
||||
node.annotation = self.visit(node.annotation)
|
||||
return node
|
||||
|
||||
def visit_arguments(self, node: ast.arguments) -> ast.AST:
|
||||
node.args = [self.visit(a) for a in node.args]
|
||||
node.posonlyargs = [self.visit(a) for a in node.posonlyargs]
|
||||
node.kwonlyargs = [self.visit(a) for a in node.kwonlyargs]
|
||||
if node.vararg is not None:
|
||||
node.vararg = self.visit(node.vararg)
|
||||
if node.kwarg is not None:
|
||||
node.kwarg = self.visit(node.kwarg)
|
||||
node.defaults = [self.visit(d) for d in node.defaults]
|
||||
node.kw_defaults = [self.visit(d) if d is not None else None for d in node.kw_defaults]
|
||||
return node
|
||||
|
||||
def visit_Attribute(self, node: ast.Attribute) -> ast.AST:
|
||||
node.value = self.visit(node.value)
|
||||
return node
|
||||
|
||||
def visit_keyword(self, node: ast.keyword) -> ast.AST:
|
||||
if node.value is not None:
|
||||
node.value = self.visit(node.value)
|
||||
return node
|
||||
|
||||
def visit_Global(self, node: ast.Global) -> ast.AST:
|
||||
node.names = [self._rename(n) for n in node.names]
|
||||
return node
|
||||
|
||||
def visit_Nonlocal(self, node: ast.Nonlocal) -> ast.AST:
|
||||
node.names = [self._rename(n) for n in node.names]
|
||||
return node
|
||||
|
||||
def visit_ExceptHandler(self, node: ast.ExceptHandler) -> ast.AST:
|
||||
if node.type is not None:
|
||||
node.type = self.visit(node.type)
|
||||
if node.name:
|
||||
node.name = self._rename(node.name)
|
||||
node.body = [self.visit(s) for s in node.body]
|
||||
return node
|
||||
|
||||
def visit_Import(self, node: ast.Import) -> ast.AST:
|
||||
for alias in node.names:
|
||||
if '.' in alias.name:
|
||||
if alias.asname:
|
||||
new = self._rename(alias.asname)
|
||||
alias.asname = new
|
||||
continue
|
||||
if alias.asname:
|
||||
alias.asname = self._rename(alias.asname)
|
||||
else:
|
||||
if alias.name not in self.mapping:
|
||||
self.mapping[alias.name] = self._new()
|
||||
alias.asname = self.mapping[alias.name]
|
||||
return node
|
||||
|
||||
def visit_ImportFrom(self, node: ast.ImportFrom) -> ast.AST:
|
||||
for alias in node.names:
|
||||
if alias.name == '*':
|
||||
continue
|
||||
if alias.asname:
|
||||
alias.asname = self._rename(alias.asname)
|
||||
else:
|
||||
if alias.name not in self.mapping:
|
||||
self.mapping[alias.name] = self._new()
|
||||
alias.asname = self.mapping[alias.name]
|
||||
return node
|
||||
@@ -0,0 +1,64 @@
|
||||
from __future__ import annotations
|
||||
import ast
|
||||
import random
|
||||
_JUNK_NAMES = ('_v', '_t', '_acc', '_state', '_buf', '_h', '_z', '_w', '_q', '_n')
|
||||
_JUNK_STMT_TEMPLATES = ('{v} = ({a} ^ {b}) + ({c} & {d})', '{v} = ((({a} * 0x01000193) ^ {b}) & 0xFFFFFFFF)', '{v} = sum(((({a} >> _k) & 0xFF) for _k in range(0, 32, 8)))', '{v} = ({a} | {b}) - (({c} & {d}) >> 1)', '{v} = pow({a} | 1, 1, ({b} | 0x10001))', '{v} = (({a} + {b}) ^ ({c} - {d})) & 0x7FFFFFFF', '{v} = bytes(((_i * {a}) ^ {b}) & 0xFF for _i in range(8))', '{v} = (~({a}) | ({b} & {c})) & 0xFFFFFF')
|
||||
|
||||
def _rint(rng: random.Random) -> int:
|
||||
return rng.randrange(1, 1 << 30)
|
||||
|
||||
class JunkBranchInjector(ast.NodeTransformer):
|
||||
|
||||
def __init__(self, rng: random.Random, seed_name: str, prob: float=0.4, max_depth: int=0):
|
||||
self.rng = rng
|
||||
self.seed_name = seed_name
|
||||
self.prob = prob
|
||||
self.injected = False
|
||||
self._depth = 0
|
||||
self._max_depth = max_depth
|
||||
|
||||
def visit_FunctionDef(self, node: ast.FunctionDef) -> ast.AST:
|
||||
return self._inject_into_body(node)
|
||||
visit_AsyncFunctionDef = visit_FunctionDef
|
||||
|
||||
def visit_Module(self, node: ast.Module) -> ast.AST:
|
||||
return self._inject_into_body(node)
|
||||
|
||||
def _inject_into_body(self, node):
|
||||
for child in ast.iter_child_nodes(node):
|
||||
if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef, ast.Module)):
|
||||
self.visit(child)
|
||||
if not getattr(node, 'body', None):
|
||||
return node
|
||||
if self.rng.random() > self.prob:
|
||||
return node
|
||||
block = self._build_block()
|
||||
idx = self.rng.randint(0, len(node.body))
|
||||
node.body.insert(idx, block)
|
||||
self.injected = True
|
||||
return node
|
||||
|
||||
def visit_ClassDef(self, node: ast.ClassDef) -> ast.AST:
|
||||
for child in node.body:
|
||||
if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
self.visit(child)
|
||||
return node
|
||||
|
||||
def _build_block(self) -> ast.stmt:
|
||||
s = self.seed_name
|
||||
kind = self.rng.choice(('parity', 'flt3', 'flt5'))
|
||||
if kind == 'parity':
|
||||
cond_src = f'({s} * ({s} + 1) + 1) % 2 == 0'
|
||||
elif kind == 'flt3':
|
||||
cond_src = f'({s} * {s} * {s} - {s} + 1) % 3 == 0'
|
||||
else:
|
||||
cond_src = f'({s} * {s} * {s} * {s} * {s} - {s} + 1) % 5 == 0'
|
||||
cond = ast.parse(cond_src, mode='eval').body
|
||||
body = [self._random_stmt() for _ in range(self.rng.randint(2, 4))]
|
||||
return ast.If(test=cond, body=body, orelse=[])
|
||||
|
||||
def _random_stmt(self) -> ast.stmt:
|
||||
v = self.rng.choice(_JUNK_NAMES)
|
||||
tmpl = self.rng.choice(_JUNK_STMT_TEMPLATES)
|
||||
src = tmpl.format(v=v, a=_rint(self.rng), b=_rint(self.rng), c=_rint(self.rng), d=_rint(self.rng))
|
||||
return ast.parse(src).body[0]
|
||||
@@ -0,0 +1,93 @@
|
||||
from __future__ import annotations
|
||||
import ast
|
||||
import copy
|
||||
import random
|
||||
|
||||
def _is_pure(node: ast.AST) -> bool:
|
||||
if isinstance(node, ast.Name):
|
||||
return True
|
||||
if isinstance(node, ast.Constant):
|
||||
return isinstance(node.value, (int, bool))
|
||||
if isinstance(node, ast.Attribute):
|
||||
return _is_pure(node.value)
|
||||
if isinstance(node, ast.Subscript):
|
||||
return _is_pure(node.value) and _is_pure(node.slice)
|
||||
if isinstance(node, ast.UnaryOp) and isinstance(node.op, (ast.USub, ast.UAdd, ast.Invert)):
|
||||
return _is_pure(node.operand)
|
||||
return False
|
||||
|
||||
def _dup(node: ast.AST) -> ast.AST:
|
||||
return copy.deepcopy(node)
|
||||
|
||||
class MBATransformer(ast.NodeTransformer):
|
||||
|
||||
def __init__(self, rng: random.Random, prob: float=0.7):
|
||||
self.rng = rng
|
||||
self.prob = prob
|
||||
|
||||
def visit_BinOp(self, node: ast.BinOp) -> ast.AST:
|
||||
self.generic_visit(node)
|
||||
if not (_is_pure(node.left) and _is_pure(node.right)):
|
||||
return node
|
||||
if self.rng.random() > self.prob:
|
||||
return node
|
||||
a, b = (node.left, node.right)
|
||||
op = node.op
|
||||
if isinstance(op, ast.BitXor):
|
||||
return self._xor(a, b)
|
||||
if isinstance(op, ast.BitOr):
|
||||
return self._or(a, b)
|
||||
if isinstance(op, ast.BitAnd):
|
||||
return self._and(a, b)
|
||||
if isinstance(op, ast.LShift) and isinstance(b, ast.Constant) and isinstance(b.value, int):
|
||||
return self._lshift(a, b.value)
|
||||
if isinstance(op, ast.RShift) and isinstance(b, ast.Constant) and isinstance(b.value, int):
|
||||
return self._rshift(a, b.value)
|
||||
return node
|
||||
|
||||
def visit_UnaryOp(self, node: ast.UnaryOp) -> ast.AST:
|
||||
self.generic_visit(node)
|
||||
if not _is_pure(node.operand):
|
||||
return node
|
||||
if self.rng.random() > self.prob:
|
||||
return node
|
||||
if isinstance(node.op, ast.Invert):
|
||||
return self._invert(node.operand)
|
||||
return node
|
||||
|
||||
def _xor(self, a: ast.AST, b: ast.AST) -> ast.AST:
|
||||
choice = self.rng.choice(('subor', 'ornotand', 'addsub'))
|
||||
if choice == 'subor':
|
||||
return ast.BinOp(left=ast.BinOp(left=_dup(a), op=ast.BitOr(), right=_dup(b)), op=ast.Sub(), right=ast.BinOp(left=_dup(a), op=ast.BitAnd(), right=_dup(b)))
|
||||
if choice == 'ornotand':
|
||||
return ast.BinOp(left=ast.BinOp(left=_dup(a), op=ast.BitOr(), right=_dup(b)), op=ast.BitAnd(), right=ast.UnaryOp(op=ast.Invert(), operand=ast.BinOp(left=_dup(a), op=ast.BitAnd(), right=_dup(b))))
|
||||
return ast.BinOp(left=ast.BinOp(left=_dup(a), op=ast.Add(), right=_dup(b)), op=ast.Sub(), right=ast.BinOp(left=ast.Constant(value=2), op=ast.Mult(), right=ast.BinOp(left=_dup(a), op=ast.BitAnd(), right=_dup(b))))
|
||||
|
||||
def _or(self, a: ast.AST, b: ast.AST) -> ast.AST:
|
||||
if self.rng.random() < 0.5:
|
||||
return ast.BinOp(left=ast.BinOp(left=_dup(a), op=ast.Add(), right=_dup(b)), op=ast.Sub(), right=ast.BinOp(left=_dup(a), op=ast.BitAnd(), right=_dup(b)))
|
||||
return ast.BinOp(left=ast.BinOp(left=_dup(a), op=ast.BitXor(), right=_dup(b)), op=ast.BitXor(), right=ast.BinOp(left=_dup(a), op=ast.BitAnd(), right=_dup(b)))
|
||||
|
||||
def _and(self, a: ast.AST, b: ast.AST) -> ast.AST:
|
||||
if self.rng.random() < 0.5:
|
||||
return ast.BinOp(left=ast.BinOp(left=_dup(a), op=ast.Add(), right=_dup(b)), op=ast.Sub(), right=ast.BinOp(left=_dup(a), op=ast.BitOr(), right=_dup(b)))
|
||||
return ast.UnaryOp(op=ast.Invert(), operand=ast.BinOp(left=ast.UnaryOp(op=ast.Invert(), operand=_dup(a)), op=ast.BitOr(), right=ast.UnaryOp(op=ast.Invert(), operand=_dup(b))))
|
||||
|
||||
def _invert(self, a: ast.AST) -> ast.AST:
|
||||
return ast.BinOp(left=ast.UnaryOp(op=ast.USub(), operand=_dup(a)), op=ast.Sub(), right=ast.Constant(value=1))
|
||||
|
||||
def _lshift(self, a: ast.AST, n: int) -> ast.AST:
|
||||
if n < 0 or n > 60:
|
||||
return ast.BinOp(left=_dup(a), op=ast.LShift(), right=ast.Constant(value=n))
|
||||
return ast.BinOp(left=_dup(a), op=ast.Mult(), right=ast.Constant(value=1 << n))
|
||||
|
||||
def _rshift(self, a: ast.AST, n: int) -> ast.AST:
|
||||
if n < 0 or n > 60:
|
||||
return ast.BinOp(left=_dup(a), op=ast.RShift(), right=ast.Constant(value=n))
|
||||
return ast.BinOp(left=_dup(a), op=ast.FloorDiv(), right=ast.Constant(value=1 << n))
|
||||
|
||||
def visit_match_case(self, node: ast.match_case) -> ast.AST:
|
||||
if node.guard is not None:
|
||||
node.guard = self.visit(node.guard)
|
||||
node.body = [self.visit(stmt) for stmt in node.body]
|
||||
return node
|
||||
@@ -0,0 +1,70 @@
|
||||
from __future__ import annotations
|
||||
import ast
|
||||
import random
|
||||
|
||||
class NumberObfuscator(ast.NodeTransformer):
|
||||
|
||||
def __init__(self, rng: random.Random, prob: float=0.85):
|
||||
self.rng = rng
|
||||
self.prob = prob
|
||||
|
||||
def visit_Constant(self, node: ast.Constant) -> ast.AST:
|
||||
v = node.value
|
||||
if isinstance(v, bool):
|
||||
return node
|
||||
if isinstance(v, int):
|
||||
if self.rng.random() <= self.prob:
|
||||
return self._obfuscate(v)
|
||||
return node
|
||||
|
||||
def visit_match_case(self, node: ast.match_case) -> ast.AST:
|
||||
if node.guard is not None:
|
||||
node.guard = self.visit(node.guard)
|
||||
node.body = [self.visit(stmt) for stmt in node.body]
|
||||
return node
|
||||
|
||||
def _obfuscate(self, n: int) -> ast.AST:
|
||||
choices = ['xor', 'sum', 'shift', 'frombytes', 'affine', 'split']
|
||||
kind = self.rng.choice(choices)
|
||||
try:
|
||||
return getattr(self, f'_t_{kind}')(n)
|
||||
except Exception:
|
||||
return ast.Constant(value=n)
|
||||
|
||||
def _t_xor(self, n: int) -> ast.AST:
|
||||
k = self.rng.randrange(1, 1 << 32)
|
||||
return self._parse_expr(f'(({n ^ k}) ^ {k})')
|
||||
|
||||
def _t_sum(self, n: int) -> ast.AST:
|
||||
a = self.rng.randint(-(1 << 16), 1 << 16)
|
||||
return self._parse_expr(f'(({n - a}) + ({a}))')
|
||||
|
||||
def _t_shift(self, n: int) -> ast.AST:
|
||||
if abs(n) >= 1 << 60:
|
||||
return self._t_xor(n)
|
||||
s = self.rng.randint(1, 8)
|
||||
return self._parse_expr(f'(({n} << {s}) >> {s})')
|
||||
|
||||
def _t_frombytes(self, n: int) -> ast.AST:
|
||||
signed = n < 0
|
||||
bit_len = max(1, n.bit_length() + (1 if signed else 0))
|
||||
byte_len = (bit_len + 7) // 8
|
||||
try:
|
||||
data = n.to_bytes(byte_len, 'big', signed=signed)
|
||||
except OverflowError:
|
||||
return self._t_xor(n)
|
||||
signed_arg = 'True' if signed else 'False'
|
||||
return self._parse_expr(f"int.from_bytes({data!r}, 'big', signed={signed_arg})")
|
||||
|
||||
def _t_affine(self, n: int) -> ast.AST:
|
||||
a = self.rng.randint(2, 64)
|
||||
b = self.rng.randint(-(1 << 12), 1 << 12)
|
||||
return self._parse_expr(f'((({n * a + b}) - ({b})) // {a})')
|
||||
|
||||
def _t_split(self, n: int) -> ast.AST:
|
||||
p = self.rng.randint(-(1 << 16), 1 << 16)
|
||||
q = n - p
|
||||
return self._parse_expr(f'({p} + ({q}))')
|
||||
|
||||
def _parse_expr(self, src: str) -> ast.AST:
|
||||
return ast.parse(src, mode='eval').body
|
||||
@@ -0,0 +1,42 @@
|
||||
from __future__ import annotations
|
||||
import ast
|
||||
import random
|
||||
|
||||
class OpaquePredicateInjector(ast.NodeTransformer):
|
||||
|
||||
def __init__(self, rng: random.Random, seed_name: str, prob: float=0.5):
|
||||
self.rng = rng
|
||||
self.seed_name = seed_name
|
||||
self.prob = prob
|
||||
self.injected = False
|
||||
|
||||
def visit_If(self, node: ast.If) -> ast.AST:
|
||||
self.generic_visit(node)
|
||||
if self.rng.random() < self.prob:
|
||||
node.test = self._wrap(node.test)
|
||||
self.injected = True
|
||||
return node
|
||||
|
||||
def visit_While(self, node: ast.While) -> ast.AST:
|
||||
self.generic_visit(node)
|
||||
if self.rng.random() < self.prob * 0.4:
|
||||
node.test = self._wrap(node.test)
|
||||
self.injected = True
|
||||
return node
|
||||
|
||||
def _wrap(self, test: ast.expr) -> ast.expr:
|
||||
return ast.BoolOp(op=ast.And(), values=[test, self._tautology()])
|
||||
|
||||
def _tautology(self) -> ast.expr:
|
||||
s = self.seed_name
|
||||
kind = self.rng.choice(('parity', 'flt3', 'flt5'))
|
||||
if kind == 'parity':
|
||||
src = f'({s} * ({s} + 1)) % 2 == 0'
|
||||
elif kind == 'flt3':
|
||||
src = f'({s} * {s} * {s} - {s}) % 3 == 0'
|
||||
else:
|
||||
src = f'({s} * {s} * {s} * {s} * {s} - {s}) % 5 == 0'
|
||||
return ast.parse(src, mode='eval').body
|
||||
|
||||
def build_seed_stmt(seed_name: str) -> ast.stmt:
|
||||
return ast.parse(f'{seed_name} = id(int) & 0xFFFFFFFF').body[0]
|
||||
@@ -0,0 +1,84 @@
|
||||
from __future__ import annotations
|
||||
import ast
|
||||
import random
|
||||
|
||||
class StringEncryptor(ast.NodeTransformer):
|
||||
|
||||
def __init__(self, rng: random.Random, decrypt_str_name: str, decrypt_bytes_name: str):
|
||||
self.rng = rng
|
||||
self.dec_s = decrypt_str_name
|
||||
self.dec_b = decrypt_bytes_name
|
||||
|
||||
def _strip_docstring(self, body: list[ast.stmt]) -> list[ast.stmt]:
|
||||
if body and isinstance(body[0], ast.Expr) and isinstance(body[0].value, ast.Constant) and isinstance(body[0].value.value, str):
|
||||
return body[1:] or [ast.Pass()]
|
||||
return body
|
||||
|
||||
def visit_Module(self, node: ast.Module) -> ast.AST:
|
||||
node.body = self._strip_docstring(node.body)
|
||||
self.generic_visit(node)
|
||||
return node
|
||||
|
||||
def visit_FunctionDef(self, node: ast.FunctionDef) -> ast.AST:
|
||||
node.body = self._strip_docstring(node.body)
|
||||
self.generic_visit(node)
|
||||
return node
|
||||
visit_AsyncFunctionDef = visit_FunctionDef
|
||||
|
||||
def visit_ClassDef(self, node: ast.ClassDef) -> ast.AST:
|
||||
node.body = self._strip_docstring(node.body)
|
||||
self.generic_visit(node)
|
||||
return node
|
||||
|
||||
def visit_JoinedStr(self, node: ast.JoinedStr) -> ast.AST:
|
||||
new_values: list[ast.AST] = []
|
||||
for v in node.values:
|
||||
if isinstance(v, ast.FormattedValue):
|
||||
v.value = self.visit(v.value)
|
||||
if v.format_spec is not None:
|
||||
v.format_spec = self.visit(v.format_spec)
|
||||
new_values.append(v)
|
||||
else:
|
||||
new_values.append(v)
|
||||
node.values = new_values
|
||||
return node
|
||||
|
||||
def visit_match_case(self, node: ast.match_case) -> ast.AST:
|
||||
if node.guard is not None:
|
||||
node.guard = self.visit(node.guard)
|
||||
node.body = [self.visit(stmt) for stmt in node.body]
|
||||
return node
|
||||
|
||||
def visit_Constant(self, node: ast.Constant) -> ast.AST:
|
||||
v = node.value
|
||||
if isinstance(v, bool):
|
||||
return node
|
||||
if isinstance(v, str):
|
||||
return self._encrypt_str(v) if v else node
|
||||
if isinstance(v, bytes):
|
||||
return self._encrypt_bytes(v) if v else node
|
||||
return node
|
||||
|
||||
def _make_key(self, n_min: int=4, n_max: int=24) -> bytes:
|
||||
return bytes((self.rng.randrange(256) for _ in range(self.rng.randint(n_min, n_max))))
|
||||
|
||||
def _xor(self, data: bytes, key: bytes) -> bytes:
|
||||
return bytes((b ^ key[i % len(key)] for i, b in enumerate(data)))
|
||||
|
||||
def _encrypt_str(self, s: str) -> ast.AST:
|
||||
try:
|
||||
data = s.encode('utf-8')
|
||||
except UnicodeEncodeError:
|
||||
return ast.Constant(value=s)
|
||||
key = self._make_key()
|
||||
enc = self._xor(data, key)
|
||||
return ast.Call(func=ast.Name(id=self.dec_s, ctx=ast.Load()), args=[ast.Constant(value=enc), ast.Constant(value=key)], keywords=[])
|
||||
|
||||
def _encrypt_bytes(self, b: bytes) -> ast.AST:
|
||||
key = self._make_key()
|
||||
enc = self._xor(b, key)
|
||||
return ast.Call(func=ast.Name(id=self.dec_b, ctx=ast.Load()), args=[ast.Constant(value=enc), ast.Constant(value=key)], keywords=[])
|
||||
|
||||
def build_decrypt_helper(decrypt_str_name: str, decrypt_bytes_name: str) -> list[ast.stmt]:
|
||||
src = f'def {decrypt_str_name}(d, k):\n return bytes(b ^ k[i % len(k)] for i, b in enumerate(d)).decode()\ndef {decrypt_bytes_name}(d, k):\n return bytes(b ^ k[i % len(k)] for i, b in enumerate(d))\n'
|
||||
return ast.parse(src).body
|
||||
@@ -0,0 +1,25 @@
|
||||
from __future__ import annotations
|
||||
import random
|
||||
import secrets
|
||||
CONFUSE_CHARS = 'OoIl01'
|
||||
|
||||
def make_rng(seed: int | None=None) -> tuple[random.Random, int]:
|
||||
if seed is None:
|
||||
seed = secrets.randbits(63)
|
||||
return (random.Random(seed), seed)
|
||||
|
||||
def gen_name(rng: random.Random, length: int | None=None, used: set[str] | None=None) -> str:
|
||||
if length is None:
|
||||
length = rng.randint(12, 20)
|
||||
used = used or set()
|
||||
for _ in range(64):
|
||||
name = '_' + ''.join((rng.choice(CONFUSE_CHARS) for _ in range(length)))
|
||||
if name not in used:
|
||||
return name
|
||||
while True:
|
||||
name = '_' + ''.join((rng.choice(CONFUSE_CHARS) for _ in range(length + rng.randint(1, 4))))
|
||||
if name not in used:
|
||||
return name
|
||||
|
||||
def gen_bytes(rng: random.Random, length: int) -> bytes:
|
||||
return bytes((rng.randrange(256) for _ in range(length)))
|
||||
Reference in New Issue
Block a user