From 56c466aa088b24ff3adc32e3275734d49f4de9ce Mon Sep 17 00:00:00 2001 From: ashton <63224111+bikini@users.noreply.github.com> Date: Wed, 17 Jun 2026 20:16:25 -0500 Subject: [PATCH] Harden Abyss assets with sealed packets --- README.md | 18 ++++++ patchwork/__init__.py | 2 +- patchwork/abyss.py | 143 +++++++++++++++++++++++++++++++++++++----- pyproject.toml | 2 +- tests/test_abyss.py | 27 ++++++++ tests/test_audit.py | 2 +- 6 files changed, 175 insertions(+), 19 deletions(-) diff --git a/README.md b/README.md index d0343ef..7d2c9c9 100644 --- a/README.md +++ b/README.md @@ -4,6 +4,11 @@ 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.7 hardens Abyss with a second sealed-packet layer: protected VM +instructions, constants, globals, locals, and entry offsets are individually +masked inside the encrypted Abyss asset, and the runtime decodes one tagged +packet at a time instead of loading a clean VM bytecode stream. + Version 0.6 adds optional Abyss VM protection for selected functions. Protected function bodies are lowered into encrypted VM assets before CPython compilation, then executed by a generated per-build runtime instead of being restored as @@ -22,6 +27,19 @@ keep-name files, dry runs, stats JSON, HTML reports, manifest verification, output budget gates, a version flag, and CI metadata that matches the supported Python versions. +## 0.7 Improvements + +- Abyss assets now use a second internal sealing layer after the outer asset + cipher: every VM instruction is stored as a masked packet with per-build, + per-position byte transforms and split shares. +- Protected function bytecode is merged into one packet stream with sealed entry + offsets, so decrypted Abyss assets no longer expose one clean `b` list per + function. +- Constants, globals, locals, and packet operands are sealed separately; a + decrypted Abyss JSON asset does not expose protected strings or opcode names. +- The generated runtime decodes and verifies one packet at a time with per-build + tags, then discards the decoded instruction after dispatch. + ## 0.6 Improvements - `--abyss` and `--abyss-functions` add an opt-in virtualized protection tier diff --git a/patchwork/__init__.py b/patchwork/__init__.py index 6ab1599..fe037f5 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.6.0' +__version__ = '0.7.0' __all__ = [ 'Obfuscator', diff --git a/patchwork/abyss.py b/patchwork/abyss.py index 9448505..d0042fb 100644 --- a/patchwork/abyss.py +++ b/patchwork/abyss.py @@ -89,6 +89,24 @@ class EncodedAbyssAssets: opcodes: dict[str, int] +_JUMP_OPS = { + "JUMP", + "JUMP_IF_FALSE", + "JUMP_IF_TRUE_KEEP", + "JUMP_IF_FALSE_KEEP", + "FOR_ITER", +} + +_PACKET_LAYOUTS = ( + (0, 1, 2), + (0, 2, 1), + (1, 0, 2), + (1, 2, 0), + (2, 0, 1), + (2, 1, 0), +) + + class _Label: def __init__(self) -> None: self.index: int | None = None @@ -237,7 +255,7 @@ class AbyssCompiler: return { "name": node.name, "code": self.emitter.resolve(), - "consts": [_encode_const(value) for value in self.constants], + "consts": list(self.constants), "globals": sorted(self.global_names), "locals": sorted(self.local_names), "externals": sorted(self.external_names), @@ -628,22 +646,78 @@ def _encode_const(value: Any) -> dict[str, Any]: raise UnsupportedAbyssNode(f"unsupported constant type {type(value).__name__}") +def _json_bytes(value: Any) -> bytes: + return json.dumps(value, separators=(",", ":"), sort_keys=True).encode("utf-8") + + +def _seal_value(value: Any, rng: random.Random, salt: int) -> list[Any]: + raw = _json_bytes(value) + key = gen_bytes(rng, rng.randint(9, 24)) + add = rng.randrange(256) + step = rng.randrange(1, 256, 2) + encoded = bytes(((byte ^ key[index % len(key)]) + add + ((index + salt) * step)) & 255 for index, byte in enumerate(raw)) + share = gen_bytes(rng, len(encoded)) + other = bytes(left ^ right for left, right in zip(encoded, share)) + return [ + base64.b85encode(share).decode("ascii"), + base64.b85encode(other).decode("ascii"), + base64.b85encode(key).decode("ascii"), + add, + step, + ] + + +def _tag_for(opcode: int, argc: int, index: int, salt: int) -> int: + return ((opcode * 2654435761) ^ (argc * 2246822519) ^ ((index + salt) * 3266489917)) & 0xFFFFFFFF + + +def _encode_packet(inst: list[Any], index: int, rng: random.Random, salt: int) -> list[Any]: + opcode = inst[0] + args = inst[1:] + tag = _tag_for(opcode, len(args), index, salt) + fields = [ + _seal_value(opcode, rng, salt + index * 5 + 1), + _seal_value(args, rng, salt + index * 5 + 2), + _seal_value(tag, rng, salt + index * 5 + 3), + ] + layout_id = rng.randrange(len(_PACKET_LAYOUTS)) + layout = _PACKET_LAYOUTS[layout_id] + return [layout_id, fields[layout[0]], fields[layout[1]], fields[layout[2]]] + + +def _rebased_instruction(inst: list[Any], base: int) -> list[Any]: + op = inst[0] + if op in _JUMP_OPS: + return [op, base + inst[1], *inst[2:]] + return list(inst) + + def encode_assets(assets: list[dict[str, Any]], rng: random.Random) -> EncodedAbyssAssets: values = rng.sample(range(1, 251), len(_OPS)) opcodes = dict(zip(_OPS, values)) encoded_funcs: list[dict[str, Any]] = [] - for asset in assets: - encoded_code = [[opcodes[inst[0]], *inst[1:]] for inst in asset["code"]] + merged_code: list[list[Any]] = [] + salt = rng.randrange(1 << 30) + for fn_index, asset in enumerate(assets): + entry = len(merged_code) + rebased_code = [_rebased_instruction(inst, entry) for inst in asset["code"]] + merged_code.extend(rebased_code) encoded_funcs.append( { - "n": asset["name"], - "c": asset["consts"], - "g": asset["globals"], - "l": asset["locals"], - "b": encoded_code, + "c": [ + _seal_value(_encode_const(value), rng, salt + (fn_index + 1) * 100000 + const_index) + for const_index, value in enumerate(asset["consts"]) + ], + "e": _seal_value(entry, rng, salt + 200000 + fn_index), + "g": _seal_value(asset["globals"], rng, salt + 300000 + fn_index), + "l": _seal_value(asset["locals"], rng, salt + 400000 + fn_index), } ) - document = json.dumps({"v": 1, "f": encoded_funcs}, separators=(",", ":"), sort_keys=True).encode("utf-8") + encoded_packets = [ + _encode_packet([opcodes[inst[0]], *inst[1:]], index, rng, salt) + for index, inst in enumerate(merged_code) + ] + document = _json_bytes({"v": 2, "m": {"s": salt}, "f": encoded_funcs, "p": encoded_packets}) key = gen_bytes(rng, rng.randint(24, 48)) encrypted = bytes(byte ^ key[index % len(key)] for index, byte in enumerate(document)) return EncodedAbyssAssets( @@ -657,10 +731,42 @@ def build_runtime_stmts(encoded: EncodedAbyssAssets, rng: random.Random) -> list blocks = _runtime_dispatch_blocks(encoded.opcodes) rng.shuffle(blocks) dispatch_blocks = "\n".join(blocks) + layouts_src = repr(_PACKET_LAYOUTS) source = f''' {ASSETS_NAME} = ({encoded.payload!r}, {encoded.key!r}) __pw_ab_cache__ = None +def __pw_ab_open__(_box, _salt): + _base64 = __import__('base64') + _json = __import__('json') + _share = _base64.b85decode(_box[0].encode('ascii')) + _other = _base64.b85decode(_box[1].encode('ascii')) + _key = _base64.b85decode(_box[2].encode('ascii')) + _add = _box[3] + _step = _box[4] + _encoded = bytes(_a ^ _b for _a, _b in zip(_share, _other)) + _raw = bytearray(len(_encoded)) + for _i, _b in enumerate(_encoded): + _raw[_i] = ((_b - _add - ((_i + _salt) * _step)) & 255) ^ _key[_i % len(_key)] + return _json.loads(bytes(_raw).decode('utf-8')) + +def __pw_ab_tag__(_opcode, _argc, _index, _salt): + return ((_opcode * 2654435761) ^ (_argc * 2246822519) ^ ((_index + _salt) * 3266489917)) & 4294967295 + +def __pw_ab_packet__(_packet, _index, _meta): + _layouts = {layouts_src} + _layout = _layouts[_packet[0] % len(_layouts)] + _fields = [None, None, None] + for _pos, _slot in enumerate(_layout): + _fields[_slot] = _packet[_pos + 1] + _salt = _meta['s'] + _opcode = __pw_ab_open__(_fields[0], _salt + _index * 5 + 1) + _args = __pw_ab_open__(_fields[1], _salt + _index * 5 + 2) + _tag = __pw_ab_open__(_fields[2], _salt + _index * 5 + 3) + if _tag != __pw_ab_tag__(_opcode, len(_args), _index, _salt): + raise RuntimeError('invalid abyss packet') + return [_opcode, *_args] + def __pw_ab_const__(_x): _t = _x['t'] if _t == 'none': @@ -689,8 +795,12 @@ def __pw_ab_load__(): _raw_key = _base64.b85decode(_key.encode('ascii')) _raw = bytes(_b ^ _raw_key[_i % len(_raw_key)] for _i, _b in enumerate(_enc)) _doc = _json.loads(_raw.decode('utf-8')) - for _fn in _doc['f']: - _fn['c'] = [__pw_ab_const__(_item) for _item in _fn['c']] + _meta = _doc['m'] + for _fi, _fn in enumerate(_doc['f']): + _fn['c'] = [__pw_ab_const__(__pw_ab_open__(_item, _meta['s'] + (_fi + 1) * 100000 + _ci)) for _ci, _item in enumerate(_fn['c'])] + _fn['e'] = __pw_ab_open__(_fn['e'], _meta['s'] + 200000 + _fi) + _fn['g'] = __pw_ab_open__(_fn['g'], _meta['s'] + 300000 + _fi) + _fn['l'] = __pw_ab_open__(_fn['l'], _meta['s'] + 400000 + _fi) __pw_ab_cache__ = _doc return __pw_ab_cache__ @@ -790,7 +900,7 @@ def __pw_ab_format__(_value, _conversion, _has_spec, _stack): _value = ascii(_value) return format(_value, _spec) -def __pw_ab_exec__(_fn, _initial_locals, _globals): +def __pw_ab_exec__(_doc, _fn, _initial_locals, _globals): _builtins = _globals.get('__builtins__', __builtins__) if not isinstance(_builtins, dict): _builtins = _builtins.__dict__ @@ -798,18 +908,19 @@ def __pw_ab_exec__(_fn, _initial_locals, _globals): _declared_locals = set(_fn.get('l', ())) _locals = dict(_initial_locals) _consts = _fn['c'] - _code = _fn['b'] + _code = _doc['p'] + _meta = _doc['m'] _stack = [] - _ip = 0 + _ip = _fn['e'] while True: - _inst = _code[_ip] + _inst = __pw_ab_packet__(_code[_ip], _ip, _meta) _op = _inst[0] {dispatch_blocks} raise RuntimeError('invalid abyss opcode') def {DISPATCH_NAME}(_fid, _env): _doc = __pw_ab_load__() - return __pw_ab_exec__(_doc['f'][_fid], _env, globals()) + return __pw_ab_exec__(_doc, _doc['f'][_fid], _env, globals()) ''' return ast.parse(source).body diff --git a/pyproject.toml b/pyproject.toml index ef9e4d1..7dd7915 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "patchwork" -version = "0.6.0" +version = "0.7.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_abyss.py b/tests/test_abyss.py index 9c46b1a..b47ae8a 100644 --- a/tests/test_abyss.py +++ b/tests/test_abyss.py @@ -1,8 +1,10 @@ from __future__ import annotations import ast +import base64 import contextlib import io +import json import random import subprocess import sys @@ -60,6 +62,31 @@ print(secret(7)) self.assertIn("secret", transformer.protected) self.assertEqual(expected, run_source(transformed_source)) + def test_encoded_assets_have_second_layer_sealing(self) -> None: + source = """ +def secret(limit): + marker = "ABYSS_SECOND_LAYER_MARKER" + total = 0 + for item in range(limit): + total += item + return f"{marker}:{total}" +""" + rng = random.Random(7007) + tree = ast.parse(source) + transformer = AbyssTransformer(rng, targets={"secret"}) + transformer.protect(tree) + encoded = encode_assets(transformer.assets, rng) + outer = base64.b85decode(encoded.payload.encode("ascii")) + key = base64.b85decode(encoded.key.encode("ascii")) + decoded = bytes(byte ^ key[index % len(key)] for index, byte in enumerate(outer)) + document = json.loads(decoded.decode("utf-8")) + + self.assertEqual(2, document["v"]) + self.assertIn("p", document) + self.assertNotIn("ABYSS_SECOND_LAYER_MARKER", decoded.decode("utf-8")) + self.assertNotIn("CONST", decoded.decode("utf-8")) + self.assertNotIn("JUMP", decoded.decode("utf-8")) + def test_obfuscator_abyss_function_can_call_preserved_global(self) -> None: source = """ def helper(value): diff --git a/tests/test_audit.py b/tests/test_audit.py index 622822c..024a5a7 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.6.0", result.stdout) + self.assertIn("patchwork 0.7.0", result.stdout) def test_config_dump_and_keep_file_merge_options(self) -> None: with tempfile.TemporaryDirectory() as raw: