mirror of
https://github.com/bikini/patchwork
synced 2026-06-27 08:08:41 +00:00
ashton
244 lines
10 KiB
Python
244 lines
10 KiB
Python
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("--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 _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",
|
|
"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)
|
|
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"],
|
|
"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)) # type: ignore[union-attr]
|
|
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", []) # type: ignore[union-attr]
|
|
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())
|