Strengthen literal obfuscation · bikini/patchwork@879ba62 · GitHub
Skip to content

Commit 879ba62

Browse files
bikinibikini
committed
Strengthen literal obfuscation
1 parent 0b0cda6 commit 879ba62

9 files changed

Lines changed: 201 additions & 97 deletions

File tree

‎README.md‎

Lines changed: 14 additions & 5 deletions

‎examples/literals.py‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,22 @@
1+
MARKER = 'PATCHWORK_LITERAL_MARKER_9f8b6b7d'
2+
UNICODE = 'snowman=\u2603 rocket=\U0001f680 accents=naive cafe'
3+
LONG_TEXT = 'alpha:' + ('0123456789abcdef' * 12) + ':omega'
4+
PAYLOAD = bytes(range(32)) + b'PATCHWORK_BYTES_MARKER'
5+
6+
7+
def build() -> tuple[str, bytes]:
8+
joined = '|'.join([MARKER, UNICODE, LONG_TEXT])
9+
masked = bytes((b ^ 0x5A) for b in PAYLOAD)
10+
return joined, masked
11+
12+
13+
def run() -> None:
14+
text, blob = build()
15+
print('text-len:', len(text))
16+
print('text-head:', text[:42])
17+
print('blob-len:', len(blob))
18+
print('blob-tail:', blob[-12:].hex())
19+
20+
21+
if __name__ == '__main__':
22+
run()

‎patchwork/__init__.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
from .core import Obfuscator, obfuscate, obfuscate_file
22
from .audit import analyze_source, build_manifest, build_stats, verify_manifest
33

4-
__version__ = '0.3.0'
4+
__version__ = '0.4.0'
55

66
__all__ = [
77
'Obfuscator',

‎patchwork/audit.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -77,7 +77,7 @@ def _tool_version() -> str:
7777
try:
7878
return version("patchwork")
7979
except PackageNotFoundError:
80-
return "0.3.0"
80+
return "0.4.0"
8181

8282

8383
def _call_name(node: ast.AST) -> str | None:

‎patchwork/transforms/strings.py‎

Lines changed: 147 additions & 84 deletions
Original file line numberDiff line numberDiff line change
@@ -1,84 +1,147 @@
1-
from __future__ import annotations
2-
import ast
3-
import random
4-
5-
class StringEncryptor(ast.NodeTransformer):
6-
7-
def __init__(self, rng: random.Random, decrypt_str_name: str, decrypt_bytes_name: str):
8-
self.rng = rng
9-
self.dec_s = decrypt_str_name
10-
self.dec_b = decrypt_bytes_name
11-
12-
def _strip_docstring(self, body: list[ast.stmt]) -> list[ast.stmt]:
13-
if body and isinstance(body[0], ast.Expr) and isinstance(body[0].value, ast.Constant) and isinstance(body[0].value.value, str):
14-
return body[1:] or [ast.Pass()]
15-
return body
16-
17-
def visit_Module(self, node: ast.Module) -> ast.AST:
18-
node.body = self._strip_docstring(node.body)
19-
self.generic_visit(node)
20-
return node
21-
22-
def visit_FunctionDef(self, node: ast.FunctionDef) -> ast.AST:
23-
node.body = self._strip_docstring(node.body)
24-
self.generic_visit(node)
25-
return node
26-
visit_AsyncFunctionDef = visit_FunctionDef
27-
28-
def visit_ClassDef(self, node: ast.ClassDef) -> ast.AST:
29-
node.body = self._strip_docstring(node.body)
30-
self.generic_visit(node)
31-
return node
32-
33-
def visit_JoinedStr(self, node: ast.JoinedStr) -> ast.AST:
34-
new_values: list[ast.AST] = []
35-
for v in node.values:
36-
if isinstance(v, ast.FormattedValue):
37-
v.value = self.visit(v.value)
38-
if v.format_spec is not None:
39-
v.format_spec = self.visit(v.format_spec)
40-
new_values.append(v)
41-
else:
42-
new_values.append(v)
43-
node.values = new_values
44-
return node
45-
46-
def visit_match_case(self, node: ast.match_case) -> ast.AST:
47-
if node.guard is not None:
48-
node.guard = self.visit(node.guard)
49-
node.body = [self.visit(stmt) for stmt in node.body]
50-
return node
51-
52-
def visit_Constant(self, node: ast.Constant) -> ast.AST:
53-
v = node.value
54-
if isinstance(v, bool):
55-
return node
56-
if isinstance(v, str):
57-
return self._encrypt_str(v) if v else node
58-
if isinstance(v, bytes):
59-
return self._encrypt_bytes(v) if v else node
60-
return node
61-
62-
def _make_key(self, n_min: int=4, n_max: int=24) -> bytes:
63-
return bytes((self.rng.randrange(256) for _ in range(self.rng.randint(n_min, n_max))))
64-
65-
def _xor(self, data: bytes, key: bytes) -> bytes:
66-
return bytes((b ^ key[i % len(key)] for i, b in enumerate(data)))
67-
68-
def _encrypt_str(self, s: str) -> ast.AST:
69-
try:
70-
data = s.encode('utf-8')
71-
except UnicodeEncodeError:
72-
return ast.Constant(value=s)
73-
key = self._make_key()
74-
enc = self._xor(data, key)
75-
return ast.Call(func=ast.Name(id=self.dec_s, ctx=ast.Load()), args=[ast.Constant(value=enc), ast.Constant(value=key)], keywords=[])
76-
77-
def _encrypt_bytes(self, b: bytes) -> ast.AST:
78-
key = self._make_key()
79-
enc = self._xor(b, key)
80-
return ast.Call(func=ast.Name(id=self.dec_b, ctx=ast.Load()), args=[ast.Constant(value=enc), ast.Constant(value=key)], keywords=[])
81-
82-
def build_decrypt_helper(decrypt_str_name: str, decrypt_bytes_name: str) -> list[ast.stmt]:
83-
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'
84-
return ast.parse(src).body
1+
from __future__ import annotations
2+
import ast
3+
import random
4+
from collections.abc import Sequence
5+
6+
class StringEncryptor(ast.NodeTransformer):
7+
8+
def __init__(self, rng: random.Random, decrypt_str_name: str, decrypt_bytes_name: str):
9+
self.rng = rng
10+
self.dec_s = decrypt_str_name
11+
self.dec_b = decrypt_bytes_name
12+
13+
def _strip_docstring(self, body: list[ast.stmt]) -> list[ast.stmt]:
14+
if body and isinstance(body[0], ast.Expr) and isinstance(body[0].value, ast.Constant) and isinstance(body[0].value.value, str):
15+
return body[1:] or [ast.Pass()]
16+
return body
17+
18+
def visit_Module(self, node: ast.Module) -> ast.AST:
19+
node.body = self._strip_docstring(node.body)
20+
self.generic_visit(node)
21+
return node
22+
23+
def visit_FunctionDef(self, node: ast.FunctionDef) -> ast.AST:
24+
node.body = self._strip_docstring(node.body)
25+
self.generic_visit(node)
26+
return node
27+
visit_AsyncFunctionDef = visit_FunctionDef
28+
29+
def visit_ClassDef(self, node: ast.ClassDef) -> ast.AST:
30+
node.body = self._strip_docstring(node.body)
31+
self.generic_visit(node)
32+
return node
33+
34+
def visit_JoinedStr(self, node: ast.JoinedStr) -> ast.AST:
35+
new_values: list[ast.AST] = []
36+
for v in node.values:
37+
if isinstance(v, ast.FormattedValue):
38+
v.value = self.visit(v.value)
39+
if v.format_spec is not None:
40+
v.format_spec = self.visit(v.format_spec)
41+
new_values.append(v)
42+
else:
43+
new_values.append(v)
44+
node.values = new_values
45+
return node
46+
47+
def visit_match_case(self, node: ast.match_case) -> ast.AST:
48+
if node.guard is not None:
49+
node.guard = self.visit(node.guard)
50+
node.body = [self.visit(stmt) for stmt in node.body]
51+
return node
52+
53+
def visit_Constant(self, node: ast.Constant) -> ast.AST:
54+
v = node.value
55+
if isinstance(v, bool):
56+
return node
57+
if isinstance(v, str):
58+
return self._encrypt_str(v) if v else node
59+
if isinstance(v, bytes):
60+
return self._encrypt_bytes(v) if v else node
61+
return node
62+
63+
def _make_key(self, n_min: int=4, n_max: int=24) -> bytes:
64+
return bytes((self.rng.randrange(256) for _ in range(self.rng.randint(n_min, n_max))))
65+
66+
def _xor(self, data: bytes, key: bytes) -> bytes:
67+
return bytes((b ^ key[i % len(key)] for i, b in enumerate(data)))
68+
69+
def _rotate_left(self, data: bytes, amount: int) -> bytes:
70+
if not data:
71+
return data
72+
amount %= len(data)
73+
return data[amount:] + data[:amount]
74+
75+
def _split_chunks(self, data: bytes) -> list[bytes]:
76+
if len(data) <= 2:
77+
return [data]
78+
max_chunks = min(8, len(data))
79+
min_chunks = 2 if len(data) > 4 else 1
80+
count = self.rng.randint(min_chunks, max_chunks)
81+
if count <= 1:
82+
return [data]
83+
cuts = sorted(self.rng.sample(range(1, len(data)), count - 1))
84+
positions = [0, *cuts, len(data)]
85+
return [data[positions[i]:positions[i + 1]] for i in range(len(positions) - 1)]
86+
87+
def _shuffle_chunks(self, chunks: Sequence[bytes]) -> tuple[list[bytes], list[int]]:
88+
if len(chunks) <= 1:
89+
return list(chunks), [0]
90+
shuffled_indexes = list(range(len(chunks)))
91+
self.rng.shuffle(shuffled_indexes)
92+
shuffled = [chunks[i] for i in shuffled_indexes]
93+
order = [shuffled_indexes.index(i) for i in range(len(chunks))]
94+
return shuffled, order
95+
96+
def _bytes_tuple(self, values: Sequence[bytes]) -> ast.Tuple:
97+
return ast.Tuple(elts=[ast.Constant(value=value) for value in values], ctx=ast.Load())
98+
99+
def _int_tuple(self, values: Sequence[int]) -> ast.Tuple:
100+
return ast.Tuple(elts=[ast.Constant(value=value) for value in values], ctx=ast.Load())
101+
102+
def _encoded_args(self, data: bytes, key: bytes) -> list[ast.AST]:
103+
enc = self._xor(data, key)
104+
rotation = self.rng.randrange(len(enc)) if enc else 0
105+
rotated = self._rotate_left(enc, rotation)
106+
data_chunks, data_order = self._shuffle_chunks(self._split_chunks(rotated))
107+
key_chunks, key_order = self._shuffle_chunks(self._split_chunks(key))
108+
return [
109+
self._bytes_tuple(data_chunks),
110+
self._bytes_tuple(key_chunks),
111+
self._int_tuple(data_order),
112+
self._int_tuple(key_order),
113+
ast.Constant(value=rotation),
114+
]
115+
116+
def _encrypt_str(self, s: str) -> ast.AST:
117+
try:
118+
data = s.encode('utf-8')
119+
except UnicodeEncodeError:
120+
return ast.Constant(value=s)
121+
key = self._make_key()
122+
return ast.Call(func=ast.Name(id=self.dec_s, ctx=ast.Load()), args=self._encoded_args(data, key), keywords=[])
123+
124+
def _encrypt_bytes(self, b: bytes) -> ast.AST:
125+
key = self._make_key()
126+
return ast.Call(func=ast.Name(id=self.dec_b, ctx=ast.Load()), args=self._encoded_args(b, key), keywords=[])
127+
128+
def build_decrypt_helper(decrypt_str_name: str, decrypt_bytes_name: str) -> list[ast.stmt]:
129+
src = f'''
130+
def _pw_join_chunks(_chunks, _order):
131+
return b''.join(_chunks[_i] for _i in _order)
132+
133+
def _pw_unrotate(_data, _amount):
134+
if not _data:
135+
return _data
136+
_amount %= len(_data)
137+
return _data[-_amount:] + _data[:-_amount] if _amount else _data
138+
139+
def {decrypt_bytes_name}(_d, _k, _do, _ko, _rot):
140+
_data = _pw_unrotate(_pw_join_chunks(_d, _do), _rot)
141+
_key = _pw_join_chunks(_k, _ko)
142+
return bytes(_b ^ _key[_i % len(_key)] for _i, _b in enumerate(_data))
143+
144+
def {decrypt_str_name}(_d, _k, _do, _ko, _rot):
145+
return {decrypt_bytes_name}(_d, _k, _do, _ko, _rot).decode()
146+
'''
147+
return ast.parse(src).body

‎pyproject.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
44

55
[project]
66
name = "patchwork"
7-
version = "0.3.0"
7+
version = "0.4.0"
88
description = "Python source transformation tool with reproducible builds, static audit metadata, and manifest generation"
99
requires-python = ">=3.10"
1010
license = {text = "MIT"}

‎tests/test_audit.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -68,7 +68,7 @@ def test_cli_version_flag_does_not_require_input(self) -> None:
6868
result = self.run_cli("--version")
6969

7070
self.assertEqual(0, result.returncode)
71-
self.assertIn("patchwork 0.3.0", result.stdout)
71+
self.assertIn("patchwork 0.4.0", result.stdout)
7272

7373
def test_config_dump_and_keep_file_merge_options(self) -> None:
7474
with tempfile.TemporaryDirectory() as raw:

‎tests/test_obfuscator.py‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
sys.path.insert(0, str(ROOT))
77
from patchwork import Obfuscator
88
EXAMPLES = ROOT / 'examples'
9-
EXAMPLE_NAMES = ['hello', 'fizzbuzz', 'classes', 'stress', 'modern', 'future_annotations']
9+
EXAMPLE_NAMES = ['hello', 'fizzbuzz', 'classes', 'stress', 'modern', 'future_annotations', 'literals']
1010
SEEDS = [1, 7, 42, 1234, 999999]
1111

1212
def run(path: Path) -> str:
@@ -52,6 +52,13 @@ def main() -> int:
5252
fails += 1
5353
else:
5454
print('ok: reproducibility (seed=1 stable)')
55+
literals_src = (EXAMPLES / 'literals.py').read_text(encoding='utf-8')
56+
literals_obf = Obfuscator(seed=2026).obfuscate(literals_src)
57+
if 'PATCHWORK_LITERAL_MARKER_9f8b6b7d' in literals_obf or 'PATCHWORK_BYTES_MARKER' in literals_obf:
58+
print('FAIL: literal markers were present in obfuscated output')
59+
fails += 1
60+
else:
61+
print('ok: literal markers absent from obfuscated output')
5562
future_src = (EXAMPLES / 'future_annotations.py').read_text(encoding='utf-8')
5663
for kwargs in ({'encrypt_strings': False}, {'opaque_predicates': False, 'junk_branches': False}):
5764
try:

‎tests/test_stress_matrix.py‎

Lines changed: 6 additions & 3 deletions

0 commit comments

Comments
 (0)