from __future__ import annotations import argparse import json import sys from pathlib import Path from . import __version__ from .audit import analyze_source, build_manifest, build_stats, verify_manifest, write_json from .config import ConfigError, DEFAULT_CONFIG, load_config, normalize_config, read_keep_file, write_config from .core import Obfuscator from .report import write_report def _parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser(prog="patchwork") p.add_argument("input", nargs="?") p.add_argument("--version", action="version", version=f"patchwork {__version__}") p.add_argument("-o", "--output") p.add_argument("--config", metavar="PATH", help="load JSON build options") p.add_argument("--dump-config", metavar="PATH", help="write the effective config JSON") p.add_argument("--verify-manifest", metavar="PATH", help="verify input/output hashes in a manifest and exit") p.add_argument("--seed", type=int, default=None) p.add_argument("--layers", type=int, default=None) p.add_argument("--stage2-layers", type=int, default=None, dest="stage2_layers") p.add_argument("--keep", action="append", default=None, metavar="NAME") p.add_argument("--keep-file", metavar="PATH", help="read names to preserve from a newline/comma separated file") p.add_argument("--no-rename", action="store_false", dest="rename", default=None) p.add_argument("--no-encrypt-strings", action="store_false", dest="encrypt_strings", default=None) p.add_argument("--no-obfuscate-numbers", action="store_false", dest="obfuscate_numbers", default=None) p.add_argument("--no-opaque", action="store_false", dest="opaque_predicates", default=None) p.add_argument("--no-mba", action="store_false", dest="mba", default=None) p.add_argument("--no-junk", action="store_false", dest="junk_branches", default=None) p.add_argument("--no-lazy", action="store_false", dest="lazy_funcs", default=None) p.add_argument("--no-anti-debug", action="store_false", dest="anti_debug", default=None) p.add_argument("--abyss", action="store_true", default=None, help="virtualize eligible functions into encrypted Abyss VM assets") p.add_argument("--abyss-functions", action="append", default=None, metavar="NAME[,NAME]", help="only virtualize the named function(s); repeatable or comma-separated") p.add_argument("--no-lower-fstrings", action="store_false", dest="lower_fstrings", default=None) p.add_argument("--no-lower-match", action="store_false", dest="lower_match", default=None) 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("--report", metavar="PATH", help="write an HTML audit/build report") p.add_argument("--stats-json", metavar="PATH", help="write input/output size and hash stats as JSON") p.add_argument("--strict-audit", action="store_true", help="refuse to obfuscate when sensitive API indicators are present") p.add_argument("--dry-run", action="store_true", help="show audit/config plan without writing obfuscated output") p.add_argument("--max-output-bytes", type=int, default=None, help="refuse to write output larger than this size") p.add_argument("--max-ratio", type=float, default=None, help="refuse to write output above this expansion ratio") p.add_argument("--max-review-indicators", type=int, default=None, help="refuse when audit indicators exceed this count") p.add_argument("--quiet", "-q", action="store_true") return p def _split_name_options(values: list[str] | None) -> list[str]: names: list[str] = [] for value in values or []: names.extend(part.strip() for part in value.split(",") if part.strip()) return names def _effective_config(args: argparse.Namespace) -> dict[str, object]: config = dict(DEFAULT_CONFIG) config.update(load_config(args.config)) for key in ( "seed", "layers", "stage2_layers", "rename", "encrypt_strings", "obfuscate_numbers", "opaque_predicates", "mba", "junk_branches", "lazy_funcs", "anti_debug", "abyss", "lower_fstrings", "lower_match", "max_output_bytes", "max_ratio", "max_review_indicators", ): value = getattr(args, key) if value is not None: config[key] = value keep = set(config.get("keep") or []) keep.update(args.keep or []) keep.update(read_keep_file(args.keep_file)) config["keep"] = sorted(keep) abyss_functions = set(config.get("abyss_functions") or []) abyss_functions.update(_split_name_options(args.abyss_functions)) config["abyss_functions"] = sorted(abyss_functions) return normalize_config(config) def _obfuscator_options(config: dict[str, object]) -> dict[str, object]: return { "rename": config["rename"], "encrypt_strings": config["encrypt_strings"], "obfuscate_numbers": config["obfuscate_numbers"], "opaque_predicates": config["opaque_predicates"], "mba": config["mba"], "junk_branches": config["junk_branches"], "lazy_funcs": config["lazy_funcs"], "anti_debug": config["anti_debug"], "abyss": config["abyss"], "abyss_functions": config["abyss_functions"], "lower_fstrings": config["lower_fstrings"], "lower_match": config["lower_match"], "layers": config["layers"], "stage2_layers": config["stage2_layers"], "keep": sorted(config["keep"]), } def _build_obfuscator(config: dict[str, object]) -> Obfuscator: options = _obfuscator_options(config) keep = set(options.pop("keep")) return Obfuscator(seed=config["seed"], keep=keep, **options) def _gate_errors(config: dict[str, object], audit: dict[str, object], stats: dict[str, object]) -> list[str]: errors: list[str] = [] review_count = int(audit.get("review", {}).get("indicator_count", 0)) max_review = config.get("max_review_indicators") if max_review is not None and review_count > int(max_review): errors.append(f"review indicators {review_count} exceed max_review_indicators {max_review}") output_bytes = stats.get("output_bytes") max_output = config.get("max_output_bytes") if output_bytes is not None and max_output is not None and int(output_bytes) > int(max_output): errors.append(f"output bytes {output_bytes} exceed max_output_bytes {max_output}") ratio = stats.get("ratio") max_ratio = config.get("max_ratio") if ratio is not None and max_ratio is not None and float(ratio) > float(max_ratio): errors.append(f"output ratio {ratio} exceeds max_ratio {max_ratio}") return errors def _print_audit_refusal(audit: dict[str, object]) -> None: print("patchwork: strict audit refused to obfuscate input with review indicators", file=sys.stderr) indicators = audit.get("review", {}).get("indicators", []) if isinstance(indicators, list): for indicator in indicators: if isinstance(indicator, dict): print( f" line {indicator['line']}: {indicator['kind']} {indicator['name']} - {indicator['reason']}", file=sys.stderr, ) def _write_auxiliary_outputs( args: argparse.Namespace, *, audit: dict[str, object], config: dict[str, object], stats: dict[str, object], ) -> None: if args.audit_json: write_json(args.audit_json, audit) if args.dump_config: write_config(args.dump_config, config) if args.stats_json: write_json(args.stats_json, stats) if args.report: write_report(args.report, audit=audit, options=_obfuscator_options(config), stats=stats) def main(argv: list[str] | None = None) -> int: parser = _parser() args = parser.parse_args(argv) try: if args.verify_manifest: result = verify_manifest(args.verify_manifest) if not args.quiet: print(json.dumps(result, indent=2, sort_keys=True)) return 0 if result["valid"] else 5 if not args.input: parser.error("the following arguments are required: input") config = _effective_config(args) 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) dry_stats = build_stats(src, None, audit) if args.audit_only or args.dry_run: _write_auxiliary_outputs(args, audit=audit, config=config, stats=dry_stats) if not args.quiet: payload = {"audit": audit, "config": config, "dry_run": bool(args.dry_run)} print(json.dumps(payload, indent=2, sort_keys=True)) if args.strict_audit and audit["review"]["requires_human_review"]: return 1 return 0 if args.strict_audit and audit["review"]["requires_human_review"]: _write_auxiliary_outputs(args, audit=audit, config=config, stats=dry_stats) _print_audit_refusal(audit) return 3 obf = _build_obfuscator(config) out_src = obf.obfuscate(src) stats = build_stats(src, out_src, audit) config["seed"] = obf.seed gate_errors = _gate_errors(config, audit, stats) if gate_errors: _write_auxiliary_outputs(args, audit=audit, config=config, stats=stats) for error in gate_errors: print(f"patchwork: {error}", file=sys.stderr) return 4 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=_obfuscator_options(config), audit=audit, ) write_json(args.manifest, manifest) _write_auxiliary_outputs(args, audit=audit, config=config, stats=stats) 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" f" input: {in_size:,} bytes\n" f" output: {out_size:,} bytes ({ratio:.1f}x)\n" f" seed: {obf.seed}\n" f" layers: {obf.layers}", file=sys.stderr, ) if args.manifest: print(f" manifest: {args.manifest}", file=sys.stderr) if args.report: print(f" report: {args.report}", file=sys.stderr) if audit["review"]["requires_human_review"]: print(f" audit: {audit['review']['indicator_count']} review indicator(s)", file=sys.stderr) return 0 except (ConfigError, ValueError) as exc: print(f"patchwork: {exc}", file=sys.stderr) return 2 if __name__ == "__main__": raise SystemExit(main())