diff --git a/README.md b/README.md index 1c2a20a..4848015 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,248 @@ # patchwork -A polymorphic multi-stage python obfuscator that mangles source via AST rewrites, encrypts each function's bytecode into a separate lazily-decrypted blob, and wraps everything in a multi-cipher anti-debug loader + +Multi-layer polymorphic Python source obfuscator. + +Takes a `.py` file and emits a single self-contained `.py` file that, when run, +executes the original program. Every build is unique — different mangled +identifiers, different encryption keys, different decryption-stub layout, and +different junk filler. + +## Features + +- **Identifier renaming** — every variable, function, class, import alias, and exception binding gets a confusable identifier (`_OoIl01…`). +- **String / bytes encryption** — every literal is XOR-encrypted with a per-literal random key, decrypted on demand. +- **Integer obfuscation** — every int literal becomes one of: XOR cancellation, sum displacement, bit-shift round-trip, `int.from_bytes`, affine identity, or split-sum. +- **Mixed Boolean-Arithmetic (MBA)** — bitwise operators rewritten with algebraic identities (`a^b → (a|b) - (a&b)`, etc.). +- **Opaque predicates** — `if`/`while` tests wrapped with runtime-true tautologies (`s*(s+1) % 2 == 0`, Fermat's little theorem) anchored on a runtime seed. +- **Junk dead branches** — realistic-looking unreachable blocks gated by always-false runtime predicates. +- **Lazy per-function code encryption** — each user function/method's compiled `code` object is encrypted into a separate blob; the marshaled module only contains stubs. Real code is decrypted on first call and the function's `__code__` is swapped in place. +- **Multi-cipher payload packing** — `marshal` → `zlib` → N (default 3) layers chosen at random from `{XOR-CTR with SHA256 keystream, byte permutation, bit rotation}`, each with its own random key. +- **Multi-stage loader** — outer Stage 1 decrypts and execs Stage 2 (compiled, marshaled, multi-cipher encrypted). Stage 2 holds the lazy blob table, the resolver, and the user payload. +- **Anti-debug** — `sys.gettrace`, `sys.getprofile`, `sys.modules` debugger probe, `PYTHONBREAKPOINT` env check, audit hook blocking debugger module imports / `sys.settrace` / `sys.setprofile`, full frame-stack walk for debugger frames. Probes are scattered through both stages. +- **Chunked bytes literals** — payloads >64 bytes are split into 4–9 shuffled chunks, reassembled at runtime via an inverse-permutation tuple. +- **Polymorphic** — every emitted name is mangled fresh; ordering of independent statements is randomized; junk filler is interspersed. + +## Requirements + +- Python 3.9+ +- Standard library only — **no third-party dependencies** + +The obfuscated output must be run on the same Python `major.minor` that built +it. `marshal` format is version-bound. + +## Install + +```sh +git clone patchwork +cd patchwork +``` + +That's it — no install needed. + +Optionally install editable to get the `patchwork` console script: + +```sh +pip install -e . +``` + +## Quick start + +CLI: + +```sh +python -m patchwork myscript.py +python myscript_obf.py +``` + +Custom output, reproducible build, more cipher layers: + +```sh +python -m patchwork myscript.py -o protected.py --seed 12345 --layers 5 +``` + +Python API: + +```python +from patchwork import Obfuscator, obfuscate_file + +obfuscate_file('myscript.py', 'protected.py', seed=12345) + +obf = Obfuscator(seed=12345, layers=4, stage2_layers=4) +protected_source = obf.obfuscate(open('myscript.py').read()) +``` + +## CLI reference + +``` +python -m patchwork INPUT [-o OUTPUT] [options] + + -o, --output PATH output path (default: _obf.py) + --seed INT RNG seed for reproducible output + --layers INT cipher layers wrapping the user payload (default 3) + --stage2-layers INT cipher layers wrapping stage 2 (default 3) + --keep NAME identifier to leave un-renamed (repeatable) + --no-rename disable identifier renaming + --no-encrypt-strings disable string/bytes literal encryption + --no-obfuscate-numbers disable integer literal obfuscation + --no-opaque disable opaque predicate injection + --no-mba disable Mixed Boolean-Arithmetic + --no-junk disable junk dead-branch injection + --no-lazy disable lazy per-function encryption + --no-anti-debug disable runtime anti-debug probes + -q, --quiet suppress informational output +``` + +## Python API + +```python +from patchwork import Obfuscator + +obf = Obfuscator( + seed=42, # int or None for fresh randomness + rename=True, + encrypt_strings=True, + obfuscate_numbers=True, + opaque_predicates=True, + mba=True, + junk_branches=True, + lazy_funcs=True, + anti_debug=True, + layers=3, # cipher layers around user payload + stage2_layers=3, # cipher layers around stage 2 + keep={'public_api_name'}, # names you don't want renamed +) +output_source = obf.obfuscate(input_source) +``` + +`obfuscate(src, **kwargs)` is a shortcut for `Obfuscator(**kwargs).obfuscate(src)`. +`obfuscate_file(in_path, out_path=None, **kwargs)` reads, obfuscates, writes, returns the output path. + +## How it works + +``` +input.py + │ + ├─ AST passes: + │ opaque-predicate injection + │ junk dead-branch injection + │ MBA on bitwise ops + │ integer-literal obfuscation + │ string/bytes encryption (helpers prepended) + │ identifier renaming + │ + ├─ compile to module code object + │ + ├─ lazy function encryption: + │ walks co_consts, replaces each user-defined function/method + │ code with a kind-matching stub (function / generator / coroutine + │ / async-generator); the original code goes into a per-function + │ encrypted blob in a side table + │ + ├─ marshal → zlib → N cipher layers ─────► user payload + │ + ├─ build stage 2 source: + │ cipher helpers, anti-debug, lazy blobs table, resolver + │ function, user-payload decryption + exec + │ + ├─ compile stage 2 → marshal → N cipher layers ─────► stage 2 payload + │ + └─ emit stage 1 source (the file you ship): + polymorphic mangled imports, anti-debug, audit hook, + frame-stack walk, cipher helpers, stage 2 payload (chunked), + decryption layers, exec + +output.py +``` + +At runtime: + +``` +python output.py + │ + ├─ stage 1: anti-debug, decrypt stage 2, exec it + │ + ├─ stage 2: anti-debug, decrypt user payload, build user namespace + │ with __pw_resolve_lazy__ injected, exec user code + │ + ├─ user code runs with stub function bodies. On first call, each + │ stub invokes __pw_resolve_lazy__, which: + │ 1. finds the calling function via sys._getframe(1).f_code + │ 2. decrypts that function's blob + │ 3. swaps the function's __code__ in place (and caches it) + │ 4. re-invokes for native semantics + │ + └─ subsequent calls use the now-real code directly +``` + +## Caveats + +- **Same Python version**: marshaled bytecode is tied to the major.minor that built it. Obfuscate on 3.11 → run on 3.11; obfuscate on 3.12 → run on 3.12. +- **Single-module scope**: patchwork obfuscates one file at a time. For a package, run it on each `.py` separately. Names that cross module boundaries (imports of one obfuscated module from another) keep their original spelling. +- **External attribute API**: `obj.x` attribute access is never rewritten — `x` might address something external. Class-body method names and class attributes are auto-kept. If you expose other names externally, use `--keep`. +- **Generators / coroutines / async generators**: fully supported with kind-matching stubs. +- **Decorators**: supported. Lazy resolution kicks in on first call, including the call made by `@decorator` at `def` time. +- **`sys.argv[0]`**: will be the obfuscated file path, not the original source path. +- **Not a cryptographic guarantee**: a determined reverse-engineer with debugger access can patch out anti-debug, dump unmarshaled code objects, and decompile. The defense is depth: every layer must be unwrapped before useful structure appears, and the decompiled output is heavily mangled. + +## Examples + +```sh +python -m patchwork examples/hello.py +python examples/hello_obf.py +# Hello, World! (from patchwork) +# number: 42 +# computed: 285 +``` + +`examples/stress.py` exercises decorators, generators, classes with `super()`, +`match` statements, walrus operator, exception handling, `*args`/`**kwargs`, +and globals/nonlocals — useful as a torture test. + +## Tests + +```sh +python tests/test_obfuscator.py +``` + +Runs every example through the obfuscator at multiple seeds, executes original +and obfuscated versions, and verifies `stdout` is byte-identical. Also checks +polymorphism (different seeds → different output) and reproducibility +(same seed → stable output). + +## Layout + +``` +patchwork/ +├── pyproject.toml +├── requirements.txt +├── README.md +├── .gitignore +├── examples/ smoke-test inputs +│ ├── hello.py +│ ├── fizzbuzz.py +│ ├── classes.py +│ └── stress.py +├── patchwork/ +│ ├── __init__.py +│ ├── __main__.py python -m patchwork +│ ├── cli.py argparse front-end +│ ├── core.py pipeline orchestrator +│ ├── crypto.py XOR-CTR / permutation / rotation primitives +│ ├── packer.py multi-cipher payload packing +│ ├── lazy.py per-function code-object encryption +│ ├── loader.py polymorphic 2-stage loader generator +│ ├── util.py confusable-name generator + RNG +│ └── transforms/ +│ ├── identifiers.py +│ ├── strings.py +│ ├── numbers.py +│ ├── opaque.py +│ ├── mba.py +│ └── junk.py +└── tests/ + └── test_obfuscator.py +``` + +## License + +MIT diff --git a/examples/classes.py b/examples/classes.py new file mode 100644 index 0000000..d3c997e --- /dev/null +++ b/examples/classes.py @@ -0,0 +1,47 @@ +from collections import deque + +class Shape: + + def __init__(self, name: str): + self.name = name + + def describe(self) -> str: + return f'shape<{self.name}>' + + def area(self) -> float: + raise NotImplementedError + +class Rect(Shape): + + def __init__(self, w: float, h: float): + super().__init__('rect') + self.w = w + self.h = h + + def area(self) -> float: + return self.w * self.h + +class Circle(Shape): + + def __init__(self, r: float): + super().__init__('circle') + self.r = r + + def area(self) -> float: + return 3.14159265 * self.r * self.r + +def total_area(shapes): + q = deque(shapes) + total = 0.0 + while q: + s = q.popleft() + try: + total += s.area() + except NotImplementedError: + continue + return total +if __name__ == '__main__': + shapes = [Rect(3, 4), Circle(5), Rect(1, 1), Circle(2)] + for s in shapes: + print(s.describe(), '->', round(s.area(), 4)) + print('total area:', round(total_area(shapes), 4)) diff --git a/examples/fizzbuzz.py b/examples/fizzbuzz.py new file mode 100644 index 0000000..c3af00d --- /dev/null +++ b/examples/fizzbuzz.py @@ -0,0 +1,16 @@ +def fizzbuzz(n: int) -> str: + if n % 15 == 0: + return 'FizzBuzz' + if n % 3 == 0: + return 'Fizz' + if n % 5 == 0: + return 'Buzz' + return str(n) + +def main() -> None: + out = [] + for i in range(1, 21): + out.append(fizzbuzz(i)) + print(' '.join(out)) +if __name__ == '__main__': + main() diff --git a/examples/hello.py b/examples/hello.py new file mode 100644 index 0000000..367268e --- /dev/null +++ b/examples/hello.py @@ -0,0 +1,8 @@ +GREETING = 'Hello, World!' + +def greet(name: str) -> str: + return f'{GREETING} (from {name})' +if __name__ == '__main__': + print(greet('patchwork')) + print('number:', 42) + print('computed:', sum((i * i for i in range(10)))) diff --git a/examples/stress.py b/examples/stress.py new file mode 100644 index 0000000..b8362c9 --- /dev/null +++ b/examples/stress.py @@ -0,0 +1,58 @@ +from functools import wraps +COUNTER = 0 + +def trace(fn): + + @wraps(fn) + def wrapper(*a, **kw): + global COUNTER + COUNTER += 1 + return fn(*a, **kw) + return wrapper + +@trace +def factorial(n: int) -> int: + if n < 0: + raise ValueError('negative') + return 1 if n <= 1 else n * factorial(n - 1) + +def squares(limit): + return [x * x for x in range(limit) if x % 2 == 0] + +def gen_pairs(seq): + for i, v in enumerate(seq): + yield (i, v) + +def make_adder(by=1): + + def add(x, *, scale=1): + return (x + by) * scale + return add + +def categorize(n): + match n: + case 0: + return 'zero' + case x if x < 0: + return 'negative' + case _: + return 'positive' + +def run() -> None: + print('factorials:', [factorial(i) for i in range(7)]) + print('squares:', squares(10)) + print('pairs:', list(gen_pairs('abcd'))) + add5 = make_adder(5) + print('adder:', add5(3, scale=2)) + print('category:', [categorize(x) for x in (-1, 0, 7)]) + if (n := sum(squares(8))) > 100: + print('walrus-large:', n) + else: + print('walrus-small:', n) + try: + factorial(-1) + except ValueError as e: + print('caught:', str(e)) + print('counter:', COUNTER) +if __name__ == '__main__': + run() diff --git a/patchwork/__init__.py b/patchwork/__init__.py new file mode 100644 index 0000000..2522499 --- /dev/null +++ b/patchwork/__init__.py @@ -0,0 +1,3 @@ +from .core import Obfuscator, obfuscate, obfuscate_file +__version__ = '0.1.0' +__all__ = ['Obfuscator', 'obfuscate', 'obfuscate_file'] diff --git a/patchwork/__main__.py b/patchwork/__main__.py new file mode 100644 index 0000000..daa0cfc --- /dev/null +++ b/patchwork/__main__.py @@ -0,0 +1,3 @@ +from .cli import main +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/patchwork/__pycache__/__init__.cpython-314.pyc b/patchwork/__pycache__/__init__.cpython-314.pyc new file mode 100644 index 0000000..ca981ae Binary files /dev/null and b/patchwork/__pycache__/__init__.cpython-314.pyc differ diff --git a/patchwork/__pycache__/__main__.cpython-314.pyc b/patchwork/__pycache__/__main__.cpython-314.pyc new file mode 100644 index 0000000..7acaec4 Binary files /dev/null and b/patchwork/__pycache__/__main__.cpython-314.pyc differ diff --git a/patchwork/__pycache__/cli.cpython-314.pyc b/patchwork/__pycache__/cli.cpython-314.pyc new file mode 100644 index 0000000..51dd60e Binary files /dev/null and b/patchwork/__pycache__/cli.cpython-314.pyc differ diff --git a/patchwork/__pycache__/core.cpython-314.pyc b/patchwork/__pycache__/core.cpython-314.pyc new file mode 100644 index 0000000..7c08d1a Binary files /dev/null and b/patchwork/__pycache__/core.cpython-314.pyc differ diff --git a/patchwork/__pycache__/crypto.cpython-314.pyc b/patchwork/__pycache__/crypto.cpython-314.pyc new file mode 100644 index 0000000..81baa96 Binary files /dev/null and b/patchwork/__pycache__/crypto.cpython-314.pyc differ diff --git a/patchwork/__pycache__/lazy.cpython-314.pyc b/patchwork/__pycache__/lazy.cpython-314.pyc new file mode 100644 index 0000000..46f99f7 Binary files /dev/null and b/patchwork/__pycache__/lazy.cpython-314.pyc differ diff --git a/patchwork/__pycache__/loader.cpython-314.pyc b/patchwork/__pycache__/loader.cpython-314.pyc new file mode 100644 index 0000000..90f1130 Binary files /dev/null and b/patchwork/__pycache__/loader.cpython-314.pyc differ diff --git a/patchwork/__pycache__/packer.cpython-314.pyc b/patchwork/__pycache__/packer.cpython-314.pyc new file mode 100644 index 0000000..2f4bd6d Binary files /dev/null and b/patchwork/__pycache__/packer.cpython-314.pyc differ diff --git a/patchwork/__pycache__/util.cpython-314.pyc b/patchwork/__pycache__/util.cpython-314.pyc new file mode 100644 index 0000000..f01848c Binary files /dev/null and b/patchwork/__pycache__/util.cpython-314.pyc differ diff --git a/patchwork/cli.py b/patchwork/cli.py new file mode 100644 index 0000000..033eeaf --- /dev/null +++ b/patchwork/cli.py @@ -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()) diff --git a/patchwork/core.py b/patchwork/core.py new file mode 100644 index 0000000..075f6b6 --- /dev/null +++ b/patchwork/core.py @@ -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, '', '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 diff --git a/patchwork/crypto.py b/patchwork/crypto.py new file mode 100644 index 0000000..7e3e523 --- /dev/null +++ b/patchwork/crypto.py @@ -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))) diff --git a/patchwork/lazy.py b/patchwork/lazy.py new file mode 100644 index 0000000..e5b0e0e --- /dev/null +++ b/patchwork/lazy.py @@ -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 = ' 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, '', '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) diff --git a/patchwork/loader.py b/patchwork/loader.py new file mode 100644 index 0000000..d468d98 --- /dev/null +++ b/patchwork/loader.py @@ -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, '', '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' diff --git a/patchwork/packer.py b/patchwork/packer.py new file mode 100644 index 0000000..af02d5c --- /dev/null +++ b/patchwork/packer.py @@ -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) diff --git a/patchwork/transforms/__init__.py b/patchwork/transforms/__init__.py new file mode 100644 index 0000000..44aaf91 --- /dev/null +++ b/patchwork/transforms/__init__.py @@ -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'] diff --git a/patchwork/transforms/__pycache__/__init__.cpython-314.pyc b/patchwork/transforms/__pycache__/__init__.cpython-314.pyc new file mode 100644 index 0000000..9959fc6 Binary files /dev/null and b/patchwork/transforms/__pycache__/__init__.cpython-314.pyc differ diff --git a/patchwork/transforms/__pycache__/identifiers.cpython-314.pyc b/patchwork/transforms/__pycache__/identifiers.cpython-314.pyc new file mode 100644 index 0000000..118f40c Binary files /dev/null and b/patchwork/transforms/__pycache__/identifiers.cpython-314.pyc differ diff --git a/patchwork/transforms/__pycache__/junk.cpython-314.pyc b/patchwork/transforms/__pycache__/junk.cpython-314.pyc new file mode 100644 index 0000000..730a425 Binary files /dev/null and b/patchwork/transforms/__pycache__/junk.cpython-314.pyc differ diff --git a/patchwork/transforms/__pycache__/mba.cpython-314.pyc b/patchwork/transforms/__pycache__/mba.cpython-314.pyc new file mode 100644 index 0000000..20709b3 Binary files /dev/null and b/patchwork/transforms/__pycache__/mba.cpython-314.pyc differ diff --git a/patchwork/transforms/__pycache__/numbers.cpython-314.pyc b/patchwork/transforms/__pycache__/numbers.cpython-314.pyc new file mode 100644 index 0000000..dfe3440 Binary files /dev/null and b/patchwork/transforms/__pycache__/numbers.cpython-314.pyc differ diff --git a/patchwork/transforms/__pycache__/opaque.cpython-314.pyc b/patchwork/transforms/__pycache__/opaque.cpython-314.pyc new file mode 100644 index 0000000..e146ca2 Binary files /dev/null and b/patchwork/transforms/__pycache__/opaque.cpython-314.pyc differ diff --git a/patchwork/transforms/__pycache__/strings.cpython-314.pyc b/patchwork/transforms/__pycache__/strings.cpython-314.pyc new file mode 100644 index 0000000..ac98b49 Binary files /dev/null and b/patchwork/transforms/__pycache__/strings.cpython-314.pyc differ diff --git a/patchwork/transforms/identifiers.py b/patchwork/transforms/identifiers.py new file mode 100644 index 0000000..f0b932e --- /dev/null +++ b/patchwork/transforms/identifiers.py @@ -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 diff --git a/patchwork/transforms/junk.py b/patchwork/transforms/junk.py new file mode 100644 index 0000000..2b8407c --- /dev/null +++ b/patchwork/transforms/junk.py @@ -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] diff --git a/patchwork/transforms/mba.py b/patchwork/transforms/mba.py new file mode 100644 index 0000000..af7c507 --- /dev/null +++ b/patchwork/transforms/mba.py @@ -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 diff --git a/patchwork/transforms/numbers.py b/patchwork/transforms/numbers.py new file mode 100644 index 0000000..4457769 --- /dev/null +++ b/patchwork/transforms/numbers.py @@ -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 diff --git a/patchwork/transforms/opaque.py b/patchwork/transforms/opaque.py new file mode 100644 index 0000000..02dde8a --- /dev/null +++ b/patchwork/transforms/opaque.py @@ -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] diff --git a/patchwork/transforms/strings.py b/patchwork/transforms/strings.py new file mode 100644 index 0000000..de64fa5 --- /dev/null +++ b/patchwork/transforms/strings.py @@ -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 diff --git a/patchwork/util.py b/patchwork/util.py new file mode 100644 index 0000000..7e62e91 --- /dev/null +++ b/patchwork/util.py @@ -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))) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..2eb72ca --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,25 @@ +[build-system] +requires = ["setuptools>=61", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "patchwork" +version = "0.1.0" +description = "Advanced multi-layer Python source obfuscator: AST mangling + marshaled bytecode + multi-cipher encryption + polymorphic anti-debug loader" +requires-python = ">=3.9" +license = {text = "MIT"} +classifiers = [ + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Topic :: Security", + "Topic :: Software Development :: Code Generators", +] + +[project.scripts] +patchwork = "patchwork.cli:main" + +[tool.setuptools.packages.find] +include = ["patchwork*"] diff --git a/tests/test_obfuscator.py b/tests/test_obfuscator.py new file mode 100644 index 0000000..d098f53 --- /dev/null +++ b/tests/test_obfuscator.py @@ -0,0 +1,58 @@ +from __future__ import annotations +import subprocess +import sys +from pathlib import Path +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT)) +from patchwork import Obfuscator +EXAMPLES = ROOT / 'examples' +EXAMPLE_NAMES = ['hello', 'fizzbuzz', 'classes', 'stress'] +SEEDS = [1, 7, 42, 1234, 999999] + +def run(path: Path) -> str: + res = subprocess.run([sys.executable, str(path)], capture_output=True, text=True) + if res.returncode != 0: + raise RuntimeError(f'{path} exited {res.returncode}\nstderr:\n{res.stderr}') + return res.stdout + +def check_one(name: str, seed: int) -> None: + src_path = EXAMPLES / f'{name}.py' + expected = run(src_path) + obf_src = Obfuscator(seed=seed).obfuscate(src_path.read_text(encoding='utf-8')) + obf_path = EXAMPLES / f'{name}_test_{seed}.py' + obf_path.write_text(obf_src, encoding='utf-8') + try: + actual = run(obf_path) + finally: + obf_path.unlink(missing_ok=True) + if actual != expected: + raise AssertionError(f'{name} seed={seed}: output mismatch\n--- expected ---\n{expected}--- actual ---\n{actual}') + +def main() -> int: + fails = 0 + for name in EXAMPLE_NAMES: + for seed in SEEDS: + try: + check_one(name, seed) + print(f'ok: {name:10s} seed={seed}') + except Exception as e: + fails += 1 + print(f'FAIL: {name:10s} seed={seed}: {e}') + src = (EXAMPLES / 'hello.py').read_text(encoding='utf-8') + a = Obfuscator(seed=1).obfuscate(src) + b = Obfuscator(seed=2).obfuscate(src) + c = Obfuscator(seed=1).obfuscate(src) + if a == b: + print('FAIL: polymorphism (seed=1 vs seed=2 produced identical output)') + fails += 1 + else: + print('ok: polymorphism (seeds 1 vs 2 differ)') + if a != c: + print('FAIL: reproducibility (same seed gave different output)') + fails += 1 + else: + print('ok: reproducibility (seed=1 stable)') + print(f'\n{fails} failures' if fails else '\nall good') + return 1 if fails else 0 +if __name__ == '__main__': + raise SystemExit(main())