diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 0000000..2671538 --- /dev/null +++ b/.gitattributes @@ -0,0 +1,6 @@ +* text=auto +*.py text eol=lf +*.md text eol=lf +*.toml text eol=lf +*.yml text eol=lf +LICENSE text eol=lf diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..b00eff1 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,23 @@ +name: CI + +on: + push: + pull_request: + +jobs: + test: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + - name: Install package + run: python -m pip install -e . + - name: Run tests + run: | + python tests/test_audit.py + python tests/test_obfuscator.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..fa7a66a --- /dev/null +++ b/.gitignore @@ -0,0 +1,23 @@ +__pycache__/ +*.pyc +*.pyo +*.pyd +*.egg-info/ +build/ +dist/ +.eggs/ +.pytest_cache/ +.mypy_cache/ +.ruff_cache/ +.venv/ +venv/ +env/ +.env +*.swp +*.swo +.DS_Store +Thumbs.db +.idea/ +.vscode/ +examples/*_obf.py +examples/*_test_*.py diff --git a/README.md b/README.md index 0610810..1777c4e 100644 --- a/README.md +++ b/README.md @@ -4,6 +4,10 @@ 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.2 adds accountability features for legitimate software-protection +workflows: static source audits, build manifests, and strict audit refusal for +inputs that use sensitive APIs. + ## What it actually does Source-level rewrites first: @@ -39,7 +43,7 @@ git clone https://github.com/bikini/patchwork.git cd patchwork ``` -If you want the `patchwork` console script on your `PATH`: +That's it. If you want the `patchwork` console script on your `PATH`: ```sh pip install -e . @@ -87,9 +91,37 @@ python -m patchwork INPUT [-o OUTPUT] [options] --no-junk turn off junk dead-branch injection --no-lazy turn off lazy per-function encryption --no-anti-debug turn off runtime anti-debug probes + --audit-only analyze the input and exit without writing output + --audit-json PATH write static audit metadata as JSON + --manifest PATH write build manifest with hashes/options/audit data + --strict-audit refuse inputs that contain sensitive API indicators -q, --quiet quiet mode ``` +## Audit and Manifest Workflow + +Audit a file without generating an obfuscated output: + +```sh +python -m patchwork app.py --audit-only --audit-json evidence/app.audit.json +``` + +Generate a manifest alongside a reproducible build: + +```sh +python -m patchwork app.py --seed 12345 --manifest evidence/app.manifest.json +``` + +Refuse to transform files that contain review indicators such as dynamic +execution, process spawning, or sensitive standard-library imports: + +```sh +python -m patchwork app.py --strict-audit +``` + +The manifest records the input and output SHA-256 hashes, Python version, +Patchwork version, build seed, selected options, and static audit metadata. + ## Python API ```python @@ -145,6 +177,7 @@ python examples/hello_obf.py ```sh python tests/test_obfuscator.py +python tests/test_audit.py ``` Runs every example through the obfuscator at multiple seeds, executes original and obfuscated versions, and checks stdout matches byte-for-byte. Also confirms different seeds produce different output and the same seed produces stable output. @@ -184,4 +217,3 @@ patchwork/ ``` MIT. - diff --git a/patchwork/__init__.py b/patchwork/__init__.py index 2522499..71e8a42 100644 --- a/patchwork/__init__.py +++ b/patchwork/__init__.py @@ -1,3 +1,4 @@ -from .core import Obfuscator, obfuscate, obfuscate_file -__version__ = '0.1.0' -__all__ = ['Obfuscator', 'obfuscate', 'obfuscate_file'] +from .core import Obfuscator, obfuscate, obfuscate_file +from .audit import analyze_source, build_manifest +__version__ = '0.2.0' +__all__ = ['Obfuscator', 'analyze_source', 'build_manifest', 'obfuscate', 'obfuscate_file'] diff --git a/patchwork/audit.py b/patchwork/audit.py new file mode 100644 index 0000000..a4ca0fb --- /dev/null +++ b/patchwork/audit.py @@ -0,0 +1,214 @@ +from __future__ import annotations + +import ast +from datetime import datetime, timezone +import hashlib +from importlib.metadata import PackageNotFoundError, version +import json +import platform +from pathlib import Path +from typing import Any + + +SENSITIVE_IMPORTS = { + "ctypes", + "marshal", + "mmap", + "pickle", + "socket", + "subprocess", + "winreg", +} + +SENSITIVE_CALLS = { + "__import__", + "compile", + "eval", + "exec", + "marshal.loads", + "os.execl", + "os.execle", + "os.execlp", + "os.execlpe", + "os.execv", + "os.execve", + "os.execvp", + "os.execvpe", + "os.popen", + "os.spawnl", + "os.spawnle", + "os.spawnlp", + "os.spawnlpe", + "os.spawnv", + "os.spawnve", + "os.spawnvp", + "os.spawnvpe", + "os.system", + "pickle.loads", + "subprocess.call", + "subprocess.check_call", + "subprocess.check_output", + "subprocess.Popen", + "subprocess.run", +} + + +def sha256_text(value: str) -> str: + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + +def _utc_now() -> str: + return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + + +def _tool_version() -> str: + try: + return version("patchwork") + except PackageNotFoundError: + return "0.2.0" + + +def _call_name(node: ast.AST) -> str | None: + if isinstance(node, ast.Name): + return node.id + if isinstance(node, ast.Attribute): + parent = _call_name(node.value) + if parent: + return f"{parent}.{node.attr}" + return node.attr + return None + + +class AuditVisitor(ast.NodeVisitor): + def __init__(self) -> None: + self.imports: set[str] = set() + self.functions: list[str] = [] + self.classes: list[str] = [] + self.review_indicators: list[dict[str, object]] = [] + self.node_count = 0 + + def generic_visit(self, node: ast.AST) -> None: + self.node_count += 1 + super().generic_visit(node) + + def visit_Import(self, node: ast.Import) -> None: + for alias in node.names: + top_level = alias.name.split(".", 1)[0] + self.imports.add(alias.name) + if top_level in SENSITIVE_IMPORTS: + self.review_indicators.append( + { + "kind": "import", + "name": alias.name, + "line": node.lineno, + "reason": "sensitive standard-library capability", + } + ) + self.generic_visit(node) + + def visit_ImportFrom(self, node: ast.ImportFrom) -> None: + module = node.module or "" + if module: + self.imports.add(module) + top_level = module.split(".", 1)[0] + if top_level in SENSITIVE_IMPORTS: + self.review_indicators.append( + { + "kind": "import", + "name": module, + "line": node.lineno, + "reason": "sensitive standard-library capability", + } + ) + self.generic_visit(node) + + def visit_FunctionDef(self, node: ast.FunctionDef) -> None: + self.functions.append(node.name) + self.generic_visit(node) + + def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None: + self.functions.append(node.name) + self.generic_visit(node) + + def visit_ClassDef(self, node: ast.ClassDef) -> None: + self.classes.append(node.name) + self.generic_visit(node) + + def visit_Call(self, node: ast.Call) -> None: + name = _call_name(node.func) + if name in SENSITIVE_CALLS: + self.review_indicators.append( + { + "kind": "call", + "name": name, + "line": node.lineno, + "reason": "dynamic execution or process capability", + } + ) + self.generic_visit(node) + + +def analyze_source(source: str, *, path: str | Path | None = None) -> dict[str, Any]: + tree = ast.parse(source, filename=str(path or "")) + visitor = AuditVisitor() + visitor.visit(tree) + review_indicators = sorted(visitor.review_indicators, key=lambda item: (int(item["line"]), str(item["name"]))) + return { + "path": str(path) if path else None, + "sha256": sha256_text(source), + "lines": source.count("\n") + (1 if source else 0), + "bytes_utf8": len(source.encode("utf-8")), + "ast_nodes": visitor.node_count, + "imports": sorted(visitor.imports), + "functions": sorted(visitor.functions), + "classes": sorted(visitor.classes), + "review": { + "indicator_count": len(review_indicators), + "requires_human_review": bool(review_indicators), + "indicators": review_indicators, + }, + } + + +def build_manifest( + *, + input_path: str | Path, + output_path: str | Path, + input_source: str, + output_source: str, + seed: int, + options: dict[str, object], + audit: dict[str, Any], +) -> dict[str, Any]: + return { + "schema": "patchwork.manifest.v1", + "generated_at": _utc_now(), + "tool": { + "name": "patchwork", + "version": _tool_version(), + "python": platform.python_version(), + "platform": platform.platform(), + }, + "input": { + "path": str(input_path), + "sha256": sha256_text(input_source), + "bytes_utf8": len(input_source.encode("utf-8")), + }, + "output": { + "path": str(output_path), + "sha256": sha256_text(output_source), + "bytes_utf8": len(output_source.encode("utf-8")), + }, + "build": { + "seed": seed, + "options": options, + }, + "audit": audit, + } + + +def write_json(path: str | Path, data: dict[str, Any]) -> Path: + output_path = Path(path) + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n", encoding="utf-8") + return output_path diff --git a/patchwork/cli.py b/patchwork/cli.py index 033eeaf..3b86ba0 100644 --- a/patchwork/cli.py +++ b/patchwork/cli.py @@ -1,44 +1,93 @@ -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()) +from __future__ import annotations +import argparse +import json +import sys +from pathlib import Path +from .audit import analyze_source, build_manifest, write_json +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('--audit-only', action='store_true', help='analyze the input and exit without writing obfuscated output') + p.add_argument('--audit-json', metavar='PATH', help='write static audit metadata as JSON') + p.add_argument('--manifest', metavar='PATH', help='write build manifest with hashes, options, and audit metadata') + p.add_argument('--strict-audit', action='store_true', help='refuse to obfuscate when sensitive API indicators are present') + p.add_argument('--quiet', '-q', action='store_true') + return p + +def _options_dict(args: argparse.Namespace) -> dict[str, object]: + return { + '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': sorted(set(args.keep)), + } + +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') + audit = analyze_source(src, path=inp) + if args.audit_json: + write_json(args.audit_json, audit) + if args.audit_only: + if not args.quiet: + print(json.dumps(audit, indent=2, sort_keys=True)) + return 1 if args.strict_audit and audit['review']['requires_human_review'] else 0 + if args.strict_audit and audit['review']['requires_human_review']: + print('patchwork: strict audit refused to obfuscate input with review indicators', file=sys.stderr) + for indicator in audit['review']['indicators']: + print(f" line {indicator['line']}: {indicator['kind']} {indicator['name']} - {indicator['reason']}", file=sys.stderr) + return 3 + options = _options_dict(args) + obf = Obfuscator(seed=args.seed, keep=set(args.keep), **{key: value for key, value in options.items() if key != '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 args.manifest: + manifest = build_manifest( + input_path=inp, + output_path=out_path, + input_source=src, + output_source=out_src, + seed=obf.seed, + options=options, + audit=audit, + ) + write_json(args.manifest, manifest) + 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) + if args.manifest: + print(f' manifest: {args.manifest}', file=sys.stderr) + if audit['review']['requires_human_review']: + print(f" audit: {audit['review']['indicator_count']} review indicator(s)", file=sys.stderr) + return 0 +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/pyproject.toml b/pyproject.toml index 2eb72ca..0cc152b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,8 +4,8 @@ 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" +version = "0.2.0" +description = "Python source transformation tool with reproducible builds, static audit metadata, and manifest generation" requires-python = ">=3.9" license = {text = "MIT"} classifiers = [ diff --git a/tests/test_audit.py b/tests/test_audit.py new file mode 100644 index 0000000..b2c5528 --- /dev/null +++ b/tests/test_audit.py @@ -0,0 +1,64 @@ +from __future__ import annotations + +import subprocess +import sys +from pathlib import Path +import tempfile +import unittest + +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT)) + +from patchwork.audit import analyze_source, build_manifest + + +class AuditTests(unittest.TestCase): + def test_analyze_source_reports_review_indicators(self) -> None: + audit = analyze_source("import subprocess\nsubprocess.run(['echo', 'x'])\n") + + self.assertTrue(audit["review"]["requires_human_review"]) + names = {item["name"] for item in audit["review"]["indicators"]} + self.assertIn("subprocess", names) + self.assertIn("subprocess.run", names) + + def test_analyze_source_collects_basic_structure(self) -> None: + audit = analyze_source("class App:\n pass\n\ndef main():\n return 1\n") + + self.assertEqual(["App"], audit["classes"]) + self.assertEqual(["main"], audit["functions"]) + self.assertFalse(audit["review"]["requires_human_review"]) + + def test_build_manifest_contains_hashes_and_options(self) -> None: + audit = analyze_source("print('hello')\n") + manifest = build_manifest( + input_path="hello.py", + output_path="hello_obf.py", + input_source="print('hello')\n", + output_source="print('hello')\n", + seed=123, + options={"layers": 3}, + audit=audit, + ) + + self.assertEqual("patchwork.manifest.v1", manifest["schema"]) + self.assertEqual(123, manifest["build"]["seed"]) + self.assertEqual(manifest["input"]["sha256"], manifest["output"]["sha256"]) + + def test_cli_strict_audit_refuses_sensitive_input(self) -> None: + with tempfile.TemporaryDirectory() as raw: + path = Path(raw) / "sample.py" + path.write_text("import subprocess\nsubprocess.run(['echo', 'x'])\n", encoding="utf-8") + + result = subprocess.run( + [sys.executable, "-m", "patchwork", str(path), "--strict-audit"], + cwd=Path(__file__).resolve().parent.parent, + capture_output=True, + text=True, + ) + + self.assertEqual(3, result.returncode) + self.assertIn("strict audit refused", result.stderr) + + +if __name__ == "__main__": + unittest.main()