From 879ba62db17199d59b174b022abbc618d0eecb8a Mon Sep 17 00:00:00 2001 From: ashton <63224111+bikini@users.noreply.github.com> Date: Thu, 4 Jun 2026 04:46:22 -0500 Subject: [PATCH] Strengthen literal obfuscation --- README.md | 19 ++- examples/literals.py | 22 +++ patchwork/__init__.py | 2 +- patchwork/audit.py | 2 +- patchwork/transforms/strings.py | 231 ++++++++++++++++++++------------ pyproject.toml | 2 +- tests/test_audit.py | 2 +- tests/test_obfuscator.py | 9 +- tests/test_stress_matrix.py | 9 +- 9 files changed, 201 insertions(+), 97 deletions(-) create mode 100644 examples/literals.py diff --git a/README.md b/README.md index b8d4e49..dd0888f 100644 --- a/README.md +++ b/README.md @@ -4,6 +4,9 @@ Python source obfuscator. Give it a `.py` file, get back a single self-contained Each build is unique by default. Pass `--seed N` if you want a reproducible one. +Version 0.4 improves literal obfuscation with split/shuffled encrypted string +and bytes payloads while preserving deterministic builds under `--seed`. + Version 0.3 adds advanced accountability and reproducibility workflows for legitimate software-protection use: config files, effective-config dumps, keep-name files, dry runs, stats JSON, HTML reports, manifest verification, @@ -28,7 +31,9 @@ Python versions. Source-level rewrites first: - Identifiers get renamed to confusable junk like `_OoIl01l1OOIlO0`. Variables, function names, classes, import aliases, exception bindings. -- String and bytes literals get XOR'd with random per-literal keys and replaced with calls to a tiny decrypt helper. +- String and bytes literals get XOR'd with random per-literal keys, rotated, + split into shuffled chunks, and replaced with calls to a tiny reconstructing + decrypt helper. - Integer literals get rewritten as random equivalent expressions: XOR cancellations, sum displacements, shift round-trips, `int.from_bytes` decodes, affine identities. - Bitwise ops get pushed through MBA identities (`a^b` becomes `(a|b) - (a&b)` and similar). - `if` and `while` tests get wrapped with always-true tautologies anchored on a runtime seed (`s*(s+1) % 2 == 0` and friends). @@ -230,6 +235,10 @@ closures, and `nonlocal`. `examples/future_annotations.py` covers `from __future__ import annotations` and verifies that observable annotation strings are preserved. +`examples/literals.py` covers Unicode strings, long strings, byte payloads, and +marker strings. The regression suite checks both behavior and that marker +strings do not appear as cleartext in generated output. + ## Tests ```sh @@ -244,10 +253,10 @@ different seeds produce different output and the same seed produces stable output. The stress matrix generates deterministic Python programs with arithmetic, -bitwise operations, pattern matching, closures, comprehensions, strings, bytes, -and future annotations. It then runs them through multiple obfuscation seeds and -option combinations, including disabled lazy loading, disabled renaming, -disabled string encryption, and disabled opaque/junk transforms. +bitwise operations, pattern matching, closures, comprehensions, Unicode strings, +bytes, and future annotations. It then runs them through multiple obfuscation +seeds and option combinations, including disabled lazy loading, disabled +renaming, disabled string encryption, and disabled opaque/junk transforms. ## Layout diff --git a/examples/literals.py b/examples/literals.py new file mode 100644 index 0000000..521f8d7 --- /dev/null +++ b/examples/literals.py @@ -0,0 +1,22 @@ +MARKER = 'PATCHWORK_LITERAL_MARKER_9f8b6b7d' +UNICODE = 'snowman=\u2603 rocket=\U0001f680 accents=naive cafe' +LONG_TEXT = 'alpha:' + ('0123456789abcdef' * 12) + ':omega' +PAYLOAD = bytes(range(32)) + b'PATCHWORK_BYTES_MARKER' + + +def build() -> tuple[str, bytes]: + joined = '|'.join([MARKER, UNICODE, LONG_TEXT]) + masked = bytes((b ^ 0x5A) for b in PAYLOAD) + return joined, masked + + +def run() -> None: + text, blob = build() + print('text-len:', len(text)) + print('text-head:', text[:42]) + print('blob-len:', len(blob)) + print('blob-tail:', blob[-12:].hex()) + + +if __name__ == '__main__': + run() diff --git a/patchwork/__init__.py b/patchwork/__init__.py index 8cc9ea5..9e024a0 100644 --- a/patchwork/__init__.py +++ b/patchwork/__init__.py @@ -1,7 +1,7 @@ from .core import Obfuscator, obfuscate, obfuscate_file from .audit import analyze_source, build_manifest, build_stats, verify_manifest -__version__ = '0.3.0' +__version__ = '0.4.0' __all__ = [ 'Obfuscator', diff --git a/patchwork/audit.py b/patchwork/audit.py index 12d20bd..91da75d 100644 --- a/patchwork/audit.py +++ b/patchwork/audit.py @@ -77,7 +77,7 @@ def _tool_version() -> str: try: return version("patchwork") except PackageNotFoundError: - return "0.3.0" + return "0.4.0" def _call_name(node: ast.AST) -> str | None: diff --git a/patchwork/transforms/strings.py b/patchwork/transforms/strings.py index de64fa5..ffab5b2 100644 --- a/patchwork/transforms/strings.py +++ b/patchwork/transforms/strings.py @@ -1,84 +1,147 @@ -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 +from __future__ import annotations +import ast +import random +from collections.abc import Sequence + +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 _rotate_left(self, data: bytes, amount: int) -> bytes: + if not data: + return data + amount %= len(data) + return data[amount:] + data[:amount] + + def _split_chunks(self, data: bytes) -> list[bytes]: + if len(data) <= 2: + return [data] + max_chunks = min(8, len(data)) + min_chunks = 2 if len(data) > 4 else 1 + count = self.rng.randint(min_chunks, max_chunks) + if count <= 1: + return [data] + cuts = sorted(self.rng.sample(range(1, len(data)), count - 1)) + positions = [0, *cuts, len(data)] + return [data[positions[i]:positions[i + 1]] for i in range(len(positions) - 1)] + + def _shuffle_chunks(self, chunks: Sequence[bytes]) -> tuple[list[bytes], list[int]]: + if len(chunks) <= 1: + return list(chunks), [0] + shuffled_indexes = list(range(len(chunks))) + self.rng.shuffle(shuffled_indexes) + shuffled = [chunks[i] for i in shuffled_indexes] + order = [shuffled_indexes.index(i) for i in range(len(chunks))] + return shuffled, order + + def _bytes_tuple(self, values: Sequence[bytes]) -> ast.Tuple: + return ast.Tuple(elts=[ast.Constant(value=value) for value in values], ctx=ast.Load()) + + def _int_tuple(self, values: Sequence[int]) -> ast.Tuple: + return ast.Tuple(elts=[ast.Constant(value=value) for value in values], ctx=ast.Load()) + + def _encoded_args(self, data: bytes, key: bytes) -> list[ast.AST]: + enc = self._xor(data, key) + rotation = self.rng.randrange(len(enc)) if enc else 0 + rotated = self._rotate_left(enc, rotation) + data_chunks, data_order = self._shuffle_chunks(self._split_chunks(rotated)) + key_chunks, key_order = self._shuffle_chunks(self._split_chunks(key)) + return [ + self._bytes_tuple(data_chunks), + self._bytes_tuple(key_chunks), + self._int_tuple(data_order), + self._int_tuple(key_order), + ast.Constant(value=rotation), + ] + + 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() + return ast.Call(func=ast.Name(id=self.dec_s, ctx=ast.Load()), args=self._encoded_args(data, key), keywords=[]) + + def _encrypt_bytes(self, b: bytes) -> ast.AST: + key = self._make_key() + return ast.Call(func=ast.Name(id=self.dec_b, ctx=ast.Load()), args=self._encoded_args(b, key), keywords=[]) + +def build_decrypt_helper(decrypt_str_name: str, decrypt_bytes_name: str) -> list[ast.stmt]: + src = f''' +def _pw_join_chunks(_chunks, _order): + return b''.join(_chunks[_i] for _i in _order) + +def _pw_unrotate(_data, _amount): + if not _data: + return _data + _amount %= len(_data) + return _data[-_amount:] + _data[:-_amount] if _amount else _data + +def {decrypt_bytes_name}(_d, _k, _do, _ko, _rot): + _data = _pw_unrotate(_pw_join_chunks(_d, _do), _rot) + _key = _pw_join_chunks(_k, _ko) + return bytes(_b ^ _key[_i % len(_key)] for _i, _b in enumerate(_data)) + +def {decrypt_str_name}(_d, _k, _do, _ko, _rot): + return {decrypt_bytes_name}(_d, _k, _do, _ko, _rot).decode() +''' + return ast.parse(src).body diff --git a/pyproject.toml b/pyproject.toml index 4689087..1a2a280 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "patchwork" -version = "0.3.0" +version = "0.4.0" description = "Python source transformation tool with reproducible builds, static audit metadata, and manifest generation" requires-python = ">=3.10" license = {text = "MIT"} diff --git a/tests/test_audit.py b/tests/test_audit.py index fb24211..5e2e16f 100644 --- a/tests/test_audit.py +++ b/tests/test_audit.py @@ -68,7 +68,7 @@ class AuditTests(unittest.TestCase): result = self.run_cli("--version") self.assertEqual(0, result.returncode) - self.assertIn("patchwork 0.3.0", result.stdout) + self.assertIn("patchwork 0.4.0", result.stdout) def test_config_dump_and_keep_file_merge_options(self) -> None: with tempfile.TemporaryDirectory() as raw: diff --git a/tests/test_obfuscator.py b/tests/test_obfuscator.py index 76b5759..d9e55cc 100644 --- a/tests/test_obfuscator.py +++ b/tests/test_obfuscator.py @@ -6,7 +6,7 @@ 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', 'modern', 'future_annotations'] +EXAMPLE_NAMES = ['hello', 'fizzbuzz', 'classes', 'stress', 'modern', 'future_annotations', 'literals'] SEEDS = [1, 7, 42, 1234, 999999] def run(path: Path) -> str: @@ -52,6 +52,13 @@ def main() -> int: fails += 1 else: print('ok: reproducibility (seed=1 stable)') + literals_src = (EXAMPLES / 'literals.py').read_text(encoding='utf-8') + literals_obf = Obfuscator(seed=2026).obfuscate(literals_src) + if 'PATCHWORK_LITERAL_MARKER_9f8b6b7d' in literals_obf or 'PATCHWORK_BYTES_MARKER' in literals_obf: + print('FAIL: literal markers were present in obfuscated output') + fails += 1 + else: + print('ok: literal markers absent from obfuscated output') future_src = (EXAMPLES / 'future_annotations.py').read_text(encoding='utf-8') for kwargs in ({'encrypt_strings': False}, {'opaque_predicates': False, 'junk_branches': False}): try: diff --git a/tests/test_stress_matrix.py b/tests/test_stress_matrix.py index 209aab0..3761f07 100644 --- a/tests/test_stress_matrix.py +++ b/tests/test_stress_matrix.py @@ -1,6 +1,7 @@ from __future__ import annotations import random +import os import subprocess import sys from pathlib import Path @@ -18,7 +19,9 @@ OBFUSCATION_SEEDS = [5, 17, 257] def run(path: Path) -> str: - result = subprocess.run([sys.executable, str(path)], capture_output=True, text=True, timeout=20) + env = dict(os.environ) + env["PYTHONIOENCODING"] = "utf-8" + result = subprocess.run([sys.executable, str(path)], capture_output=True, text=True, timeout=20, env=env) if result.returncode != 0: raise RuntimeError(f"{path} exited {result.returncode}\nstderr:\n{result.stderr}") return result.stdout @@ -27,7 +30,7 @@ def run(path: Path) -> str: def generated_program(seed: int) -> str: rng = random.Random(seed) numbers = [rng.randint(-5000, 5000) for _ in range(16)] - strings = [f"s{idx}_{rng.randrange(1 << 20):x}" for idx in range(6)] + strings = [f"s{idx}_{rng.randrange(1 << 20):x}_\u2603_\U0001f680" for idx in range(6)] pairs = [(rng.randint(-20, 20), rng.randint(-20, 20)) for _ in range(8)] return textwrap.dedent( f""" @@ -73,7 +76,7 @@ def generated_program(seed: int) -> str: print("strings", "|".join(s[::-1] for s in STRINGS)) print("pairs", [classify(pair) for pair in PAIRS]) print("closure", [fn(x) for x in range(6)]) - print("bytes", bytes(range(8)).hex()) + print("bytes", (bytes(range(8)) + b"literal-marker").hex()) if __name__ == "__main__": run()