mirror of
https://github.com/bikini/patchwork
synced 2026-06-27 08:08:41 +00:00
Add advanced workflow controls
This commit is contained in:
@@ -9,7 +9,7 @@ jobs:
|
||||
runs-on: ubuntu-latest
|
||||
strategy:
|
||||
matrix:
|
||||
python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"]
|
||||
python-version: ["3.10", "3.11", "3.12", "3.13"]
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
- uses: actions/setup-python@v5
|
||||
|
||||
@@ -4,9 +4,24 @@ 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.
|
||||
Version 0.3 adds advanced accountability and reproducibility workflows for
|
||||
legitimate software-protection use: config files, effective-config dumps,
|
||||
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.3 Improvements
|
||||
|
||||
- `--version` for scriptable tool identification.
|
||||
- `--config PATH` for JSON build profiles.
|
||||
- `--dump-config PATH` for saving the effective merged configuration.
|
||||
- `--keep-file PATH` for larger preserve-name lists.
|
||||
- `--dry-run` for audit/config planning without writing obfuscated output.
|
||||
- `--stats-json PATH` for machine-readable size/hash/build stats.
|
||||
- `--report PATH` for standalone HTML audit/build reports.
|
||||
- `--verify-manifest PATH` for checking manifest input/output hashes.
|
||||
- `--max-output-bytes`, `--max-ratio`, and `--max-review-indicators` gates.
|
||||
- Python support metadata and CI aligned to Python 3.10+.
|
||||
|
||||
## What it actually does
|
||||
|
||||
@@ -32,7 +47,7 @@ Then the loader:
|
||||
|
||||
## Requirements
|
||||
|
||||
Python 3.9 or newer. Stdlib only, no third-party packages.
|
||||
Python 3.10 or newer. Stdlib only, no third-party packages.
|
||||
|
||||
The output has to run on the same Python `major.minor` you built it on. `marshal` format is version-bound.
|
||||
|
||||
@@ -78,11 +93,16 @@ out = obf.obfuscate(open('myscript.py').read())
|
||||
```
|
||||
python -m patchwork INPUT [-o OUTPUT] [options]
|
||||
|
||||
--version print Patchwork version and exit
|
||||
-o, --output PATH output path (default: <input>_obf.py)
|
||||
--config PATH load JSON build options
|
||||
--dump-config PATH write the effective merged config JSON
|
||||
--verify-manifest PATH verify input/output hashes in a manifest and exit
|
||||
--seed INT RNG seed for reproducible builds
|
||||
--layers INT cipher layers around the user payload (default 3)
|
||||
--stage2-layers INT cipher layers around stage 2 (default 3)
|
||||
--keep NAME identifier to leave un-renamed (repeatable)
|
||||
--keep-file PATH read names to preserve from a text file
|
||||
--no-rename turn off identifier renaming
|
||||
--no-encrypt-strings turn off string/bytes literal encryption
|
||||
--no-obfuscate-numbers turn off integer literal obfuscation
|
||||
@@ -94,7 +114,13 @@ python -m patchwork INPUT [-o OUTPUT] [options]
|
||||
--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
|
||||
--report PATH write a standalone HTML audit/build report
|
||||
--stats-json PATH write input/output stats JSON
|
||||
--strict-audit refuse inputs that contain sensitive API indicators
|
||||
--dry-run show audit/config plan without writing output
|
||||
--max-output-bytes INT refuse output larger than this size
|
||||
--max-ratio FLOAT refuse output above this expansion ratio
|
||||
--max-review-indicators N refuse if review indicators exceed this count
|
||||
-q, --quiet quiet mode
|
||||
```
|
||||
|
||||
@@ -109,7 +135,28 @@ 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
|
||||
python -m patchwork app.py --seed 12345 --manifest evidence/app.manifest.json --stats-json evidence/app.stats.json --report evidence/app.report.html
|
||||
```
|
||||
|
||||
Verify a manifest later:
|
||||
|
||||
```sh
|
||||
python -m patchwork --verify-manifest evidence/app.manifest.json
|
||||
```
|
||||
|
||||
Use a JSON build profile and dump the effective merged config:
|
||||
|
||||
```json
|
||||
{
|
||||
"layers": 4,
|
||||
"stage2_layers": 4,
|
||||
"rename": true,
|
||||
"keep": ["public_api"]
|
||||
}
|
||||
```
|
||||
|
||||
```sh
|
||||
python -m patchwork app.py --config patchwork.json --keep-file keep-names.txt --dump-config evidence/effective-config.json
|
||||
```
|
||||
|
||||
Refuse to transform files that contain review indicators such as dynamic
|
||||
@@ -120,12 +167,14 @@ 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.
|
||||
Patchwork version, build seed, selected options, size/hash stats, and static
|
||||
audit metadata.
|
||||
|
||||
## Python API
|
||||
|
||||
```python
|
||||
from patchwork import Obfuscator
|
||||
from patchwork.audit import analyze_source, verify_manifest
|
||||
|
||||
obf = Obfuscator(
|
||||
seed=42,
|
||||
|
||||
+13
-3
@@ -1,4 +1,14 @@
|
||||
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']
|
||||
from .audit import analyze_source, build_manifest, build_stats, verify_manifest
|
||||
|
||||
__version__ = '0.3.0'
|
||||
|
||||
__all__ = [
|
||||
'Obfuscator',
|
||||
'analyze_source',
|
||||
'build_manifest',
|
||||
'build_stats',
|
||||
'obfuscate',
|
||||
'obfuscate_file',
|
||||
'verify_manifest',
|
||||
]
|
||||
|
||||
+85
-1
@@ -57,6 +57,18 @@ def sha256_text(value: str) -> str:
|
||||
return hashlib.sha256(value.encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
def sha256_file(path: str | Path) -> str:
|
||||
digest = hashlib.sha256()
|
||||
with Path(path).open("rb") as handle:
|
||||
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
||||
digest.update(chunk)
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def sha256_text_file(path: str | Path) -> str:
|
||||
return sha256_text(Path(path).read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def _utc_now() -> str:
|
||||
return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
|
||||
|
||||
@@ -65,7 +77,7 @@ def _tool_version() -> str:
|
||||
try:
|
||||
return version("patchwork")
|
||||
except PackageNotFoundError:
|
||||
return "0.2.0"
|
||||
return "0.3.0"
|
||||
|
||||
|
||||
def _call_name(node: ast.AST) -> str | None:
|
||||
@@ -170,6 +182,21 @@ def analyze_source(source: str, *, path: str | Path | None = None) -> dict[str,
|
||||
}
|
||||
|
||||
|
||||
def build_stats(input_source: str, output_source: str | None, audit: dict[str, Any]) -> dict[str, object]:
|
||||
input_bytes = len(input_source.encode("utf-8"))
|
||||
output_bytes = len(output_source.encode("utf-8")) if output_source is not None else None
|
||||
return {
|
||||
"input_bytes": input_bytes,
|
||||
"output_bytes": output_bytes,
|
||||
"ratio": round(output_bytes / max(input_bytes, 1), 4) if output_bytes is not None else None,
|
||||
"input_lines": input_source.count("\n") + (1 if input_source else 0),
|
||||
"output_lines": output_source.count("\n") + (1 if output_source else 0) if output_source is not None else None,
|
||||
"input_sha256": sha256_text(input_source),
|
||||
"output_sha256": sha256_text(output_source) if output_source is not None else None,
|
||||
"review_indicator_count": audit.get("review", {}).get("indicator_count", 0),
|
||||
}
|
||||
|
||||
|
||||
def build_manifest(
|
||||
*,
|
||||
input_path: str | Path,
|
||||
@@ -200,13 +227,70 @@ def build_manifest(
|
||||
"bytes_utf8": len(output_source.encode("utf-8")),
|
||||
},
|
||||
"build": {
|
||||
"cwd": str(Path.cwd()),
|
||||
"seed": seed,
|
||||
"options": options,
|
||||
},
|
||||
"stats": build_stats(input_source, output_source, audit),
|
||||
"audit": audit,
|
||||
}
|
||||
|
||||
|
||||
def verify_manifest(path: str | Path) -> dict[str, Any]:
|
||||
manifest_path = Path(path)
|
||||
try:
|
||||
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
||||
except FileNotFoundError as exc:
|
||||
raise ValueError(f"manifest file not found: {manifest_path}") from exc
|
||||
except json.JSONDecodeError as exc:
|
||||
raise ValueError(f"invalid manifest JSON: {exc}") from exc
|
||||
if not isinstance(manifest, dict):
|
||||
raise ValueError("manifest must contain a JSON object")
|
||||
|
||||
errors: list[str] = []
|
||||
if manifest.get("schema") != "patchwork.manifest.v1":
|
||||
errors.append("unsupported manifest schema")
|
||||
|
||||
build = manifest.get("build", {})
|
||||
cwd = Path(build.get("cwd") or Path.cwd()) if isinstance(build, dict) else Path.cwd()
|
||||
checks: dict[str, dict[str, object]] = {}
|
||||
for section in ("input", "output"):
|
||||
item = manifest.get(section, {})
|
||||
if not isinstance(item, dict):
|
||||
errors.append(f"{section} section missing or invalid")
|
||||
continue
|
||||
raw_path = item.get("path")
|
||||
expected = item.get("sha256")
|
||||
if not isinstance(raw_path, str) or not isinstance(expected, str):
|
||||
errors.append(f"{section} path or sha256 missing")
|
||||
continue
|
||||
candidate = Path(raw_path)
|
||||
if not candidate.is_absolute():
|
||||
candidate = cwd / candidate
|
||||
exists = candidate.exists()
|
||||
try:
|
||||
actual = sha256_text_file(candidate) if exists else None
|
||||
except UnicodeDecodeError:
|
||||
actual = None
|
||||
ok = actual == expected
|
||||
if not ok:
|
||||
errors.append(f"{section} hash mismatch or file missing: {candidate}")
|
||||
checks[section] = {
|
||||
"path": str(candidate),
|
||||
"exists": exists,
|
||||
"expected_sha256": expected,
|
||||
"actual_sha256": actual,
|
||||
"ok": ok,
|
||||
}
|
||||
|
||||
return {
|
||||
"valid": not errors,
|
||||
"errors": errors,
|
||||
"checks": checks,
|
||||
"manifest": str(manifest_path),
|
||||
}
|
||||
|
||||
|
||||
def write_json(path: str | Path, data: dict[str, Any]) -> Path:
|
||||
output_path = Path(path)
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
+226
-76
@@ -1,93 +1,243 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from .audit import analyze_source, build_manifest, write_json
|
||||
|
||||
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')
|
||||
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')
|
||||
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 _options_dict(args: argparse.Namespace) -> dict[str, object]:
|
||||
|
||||
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': 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)),
|
||||
"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 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)
|
||||
|
||||
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.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.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:
|
||||
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__':
|
||||
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())
|
||||
|
||||
@@ -0,0 +1,128 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
DEFAULT_CONFIG: dict[str, Any] = {
|
||||
"seed": None,
|
||||
"layers": 3,
|
||||
"stage2_layers": 3,
|
||||
"rename": True,
|
||||
"encrypt_strings": True,
|
||||
"obfuscate_numbers": True,
|
||||
"opaque_predicates": True,
|
||||
"mba": True,
|
||||
"junk_branches": True,
|
||||
"lazy_funcs": True,
|
||||
"anti_debug": True,
|
||||
"keep": [],
|
||||
"max_output_bytes": None,
|
||||
"max_ratio": None,
|
||||
"max_review_indicators": None,
|
||||
}
|
||||
|
||||
BOOLEAN_KEYS = {
|
||||
"rename",
|
||||
"encrypt_strings",
|
||||
"obfuscate_numbers",
|
||||
"opaque_predicates",
|
||||
"mba",
|
||||
"junk_branches",
|
||||
"lazy_funcs",
|
||||
"anti_debug",
|
||||
}
|
||||
|
||||
INTEGER_KEYS = {"seed", "layers", "stage2_layers", "max_output_bytes", "max_review_indicators"}
|
||||
FLOAT_KEYS = {"max_ratio"}
|
||||
LIST_KEYS = {"keep"}
|
||||
|
||||
|
||||
class ConfigError(ValueError):
|
||||
"""Raised for invalid Patchwork configuration files."""
|
||||
|
||||
|
||||
def load_config(path: str | Path | None) -> dict[str, Any]:
|
||||
if path is None:
|
||||
return {}
|
||||
config_path = Path(path)
|
||||
try:
|
||||
raw = json.loads(config_path.read_text(encoding="utf-8"))
|
||||
except FileNotFoundError as exc:
|
||||
raise ConfigError(f"config file not found: {config_path}") from exc
|
||||
except json.JSONDecodeError as exc:
|
||||
raise ConfigError(f"invalid config JSON: {exc}") from exc
|
||||
if not isinstance(raw, dict):
|
||||
raise ConfigError("config file must contain a JSON object")
|
||||
unknown = sorted(set(raw) - set(DEFAULT_CONFIG))
|
||||
if unknown:
|
||||
raise ConfigError(f"unknown config option(s): {', '.join(unknown)}")
|
||||
return dict(raw)
|
||||
|
||||
|
||||
def read_keep_file(path: str | Path | None) -> list[str]:
|
||||
if path is None:
|
||||
return []
|
||||
keep_path = Path(path)
|
||||
try:
|
||||
lines = keep_path.read_text(encoding="utf-8").splitlines()
|
||||
except FileNotFoundError as exc:
|
||||
raise ConfigError(f"keep file not found: {keep_path}") from exc
|
||||
names: list[str] = []
|
||||
for line in lines:
|
||||
value = line.split("#", 1)[0].strip()
|
||||
if not value:
|
||||
continue
|
||||
names.extend(part.strip() for part in value.split(",") if part.strip())
|
||||
return names
|
||||
|
||||
|
||||
def normalize_config(config: dict[str, Any]) -> dict[str, Any]:
|
||||
normalized = dict(DEFAULT_CONFIG)
|
||||
normalized.update(config)
|
||||
|
||||
for key in BOOLEAN_KEYS:
|
||||
if not isinstance(normalized[key], bool):
|
||||
raise ConfigError(f"{key} must be a boolean")
|
||||
|
||||
for key in INTEGER_KEYS:
|
||||
value = normalized[key]
|
||||
if value is None:
|
||||
continue
|
||||
if not isinstance(value, int) or isinstance(value, bool):
|
||||
raise ConfigError(f"{key} must be an integer")
|
||||
|
||||
for key in FLOAT_KEYS:
|
||||
value = normalized[key]
|
||||
if value is None:
|
||||
continue
|
||||
if not isinstance(value, (int, float)) or isinstance(value, bool):
|
||||
raise ConfigError(f"{key} must be a number")
|
||||
normalized[key] = float(value)
|
||||
|
||||
for key in LIST_KEYS:
|
||||
value = normalized[key]
|
||||
if not isinstance(value, list) or not all(isinstance(item, str) for item in value):
|
||||
raise ConfigError(f"{key} must be a list of strings")
|
||||
normalized[key] = sorted(set(item for item in value if item))
|
||||
|
||||
if normalized["layers"] < 1:
|
||||
raise ConfigError("layers must be at least 1")
|
||||
if normalized["stage2_layers"] < 1:
|
||||
raise ConfigError("stage2_layers must be at least 1")
|
||||
if normalized["max_output_bytes"] is not None and normalized["max_output_bytes"] < 1:
|
||||
raise ConfigError("max_output_bytes must be at least 1")
|
||||
if normalized["max_ratio"] is not None and normalized["max_ratio"] <= 0:
|
||||
raise ConfigError("max_ratio must be positive")
|
||||
if normalized["max_review_indicators"] is not None and normalized["max_review_indicators"] < 0:
|
||||
raise ConfigError("max_review_indicators cannot be negative")
|
||||
|
||||
return normalized
|
||||
|
||||
|
||||
def write_config(path: str | Path, config: dict[str, Any]) -> Path:
|
||||
output_path = Path(path)
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
output_path.write_text(json.dumps(normalize_config(config), indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
||||
return output_path
|
||||
@@ -0,0 +1,131 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import html
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
def _escape(value: object) -> str:
|
||||
return html.escape(str(value), quote=True)
|
||||
|
||||
|
||||
def _indicator_rows(audit: dict[str, Any]) -> str:
|
||||
indicators = audit.get("review", {}).get("indicators", [])
|
||||
rows: list[str] = []
|
||||
if isinstance(indicators, list):
|
||||
for indicator in indicators:
|
||||
if not isinstance(indicator, dict):
|
||||
continue
|
||||
rows.append(
|
||||
"<tr>"
|
||||
f"<td>{_escape(indicator.get('line', ''))}</td>"
|
||||
f"<td>{_escape(indicator.get('kind', ''))}</td>"
|
||||
f"<td>{_escape(indicator.get('name', ''))}</td>"
|
||||
f"<td>{_escape(indicator.get('reason', ''))}</td>"
|
||||
"</tr>"
|
||||
)
|
||||
return "\n".join(rows) or '<tr><td colspan="4">No review indicators found.</td></tr>'
|
||||
|
||||
|
||||
def _option_rows(options: dict[str, object]) -> str:
|
||||
return "\n".join(
|
||||
f"<tr><td>{_escape(key)}</td><td><code>{_escape(value)}</code></td></tr>"
|
||||
for key, value in sorted(options.items())
|
||||
)
|
||||
|
||||
|
||||
def render_report(*, audit: dict[str, Any], options: dict[str, object], stats: dict[str, object] | None = None) -> str:
|
||||
review = audit.get("review", {})
|
||||
stats = stats or {}
|
||||
return f"""<!doctype html>
|
||||
<html lang="en">
|
||||
<head>
|
||||
<meta charset="utf-8">
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1">
|
||||
<title>Patchwork Build Report</title>
|
||||
<style>
|
||||
:root {{
|
||||
font-family: ui-sans-serif, system-ui, -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif;
|
||||
line-height: 1.45;
|
||||
color: #15181f;
|
||||
background: #f5f7fa;
|
||||
}}
|
||||
body {{
|
||||
margin: 0;
|
||||
}}
|
||||
main {{
|
||||
max-width: 1080px;
|
||||
margin: 0 auto;
|
||||
padding: 32px 20px 48px;
|
||||
}}
|
||||
section {{
|
||||
background: #ffffff;
|
||||
border: 1px solid #dce2ea;
|
||||
border-radius: 8px;
|
||||
margin: 16px 0;
|
||||
padding: 18px;
|
||||
}}
|
||||
table {{
|
||||
border-collapse: collapse;
|
||||
width: 100%;
|
||||
background: #ffffff;
|
||||
}}
|
||||
th, td {{
|
||||
border-bottom: 1px solid #e7ebf0;
|
||||
padding: 9px;
|
||||
text-align: left;
|
||||
vertical-align: top;
|
||||
font-size: 14px;
|
||||
}}
|
||||
th {{
|
||||
background: #edf1f6;
|
||||
}}
|
||||
code {{
|
||||
font-family: Consolas, "Liberation Mono", monospace;
|
||||
overflow-wrap: anywhere;
|
||||
}}
|
||||
</style>
|
||||
</head>
|
||||
<body>
|
||||
<main>
|
||||
<h1>Patchwork Build Report</h1>
|
||||
<section>
|
||||
<h2>Source Audit</h2>
|
||||
<p><strong>Path:</strong> {_escape(audit.get('path'))}</p>
|
||||
<p><strong>SHA-256:</strong> <code>{_escape(audit.get('sha256'))}</code></p>
|
||||
<p><strong>Lines:</strong> {_escape(audit.get('lines'))}</p>
|
||||
<p><strong>AST nodes:</strong> {_escape(audit.get('ast_nodes'))}</p>
|
||||
<p><strong>Review indicators:</strong> {_escape(review.get('indicator_count', 0))}</p>
|
||||
</section>
|
||||
<section>
|
||||
<h2>Output Stats</h2>
|
||||
<p><strong>Output SHA-256:</strong> <code>{_escape(stats.get('output_sha256', 'not generated'))}</code></p>
|
||||
<p><strong>Input bytes:</strong> {_escape(stats.get('input_bytes', 'not generated'))}</p>
|
||||
<p><strong>Output bytes:</strong> {_escape(stats.get('output_bytes', 'not generated'))}</p>
|
||||
<p><strong>Expansion ratio:</strong> {_escape(stats.get('ratio', 'not generated'))}</p>
|
||||
</section>
|
||||
<section>
|
||||
<h2>Options</h2>
|
||||
<table>
|
||||
<thead><tr><th>Option</th><th>Value</th></tr></thead>
|
||||
<tbody>{_option_rows(options)}</tbody>
|
||||
</table>
|
||||
</section>
|
||||
<section>
|
||||
<h2>Review Indicators</h2>
|
||||
<table>
|
||||
<thead><tr><th>Line</th><th>Kind</th><th>Name</th><th>Reason</th></tr></thead>
|
||||
<tbody>{_indicator_rows(audit)}</tbody>
|
||||
</table>
|
||||
</section>
|
||||
</main>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
|
||||
|
||||
def write_report(path: str | Path, *, audit: dict[str, Any], options: dict[str, object], stats: dict[str, object] | None = None) -> Path:
|
||||
output_path = Path(path)
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
output_path.write_text(render_report(audit=audit, options=options, stats=stats), encoding="utf-8")
|
||||
return output_path
|
||||
+3
-3
@@ -4,16 +4,16 @@ build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "patchwork"
|
||||
version = "0.2.0"
|
||||
version = "0.3.0"
|
||||
description = "Python source transformation tool with reproducible builds, static audit metadata, and manifest generation"
|
||||
requires-python = ">=3.9"
|
||||
requires-python = ">=3.10"
|
||||
license = {text = "MIT"}
|
||||
classifiers = [
|
||||
"Programming Language :: Python :: 3",
|
||||
"Programming Language :: Python :: 3.9",
|
||||
"Programming Language :: Python :: 3.10",
|
||||
"Programming Language :: Python :: 3.11",
|
||||
"Programming Language :: Python :: 3.12",
|
||||
"Programming Language :: Python :: 3.13",
|
||||
"Topic :: Security",
|
||||
"Topic :: Software Development :: Code Generators",
|
||||
]
|
||||
|
||||
+104
-7
@@ -5,14 +5,23 @@ import sys
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
import unittest
|
||||
import json
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from patchwork.audit import analyze_source, build_manifest
|
||||
from patchwork.audit import analyze_source, build_manifest, verify_manifest
|
||||
|
||||
|
||||
class AuditTests(unittest.TestCase):
|
||||
def run_cli(self, *args: str) -> subprocess.CompletedProcess[str]:
|
||||
return subprocess.run(
|
||||
[sys.executable, "-m", "patchwork", *args],
|
||||
cwd=ROOT,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
|
||||
def test_analyze_source_reports_review_indicators(self) -> None:
|
||||
audit = analyze_source("import subprocess\nsubprocess.run(['echo', 'x'])\n")
|
||||
|
||||
@@ -43,22 +52,110 @@ class AuditTests(unittest.TestCase):
|
||||
self.assertEqual("patchwork.manifest.v1", manifest["schema"])
|
||||
self.assertEqual(123, manifest["build"]["seed"])
|
||||
self.assertEqual(manifest["input"]["sha256"], manifest["output"]["sha256"])
|
||||
self.assertIn("stats", manifest)
|
||||
|
||||
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,
|
||||
)
|
||||
result = self.run_cli(str(path), "--strict-audit")
|
||||
|
||||
self.assertEqual(3, result.returncode)
|
||||
self.assertIn("strict audit refused", result.stderr)
|
||||
|
||||
def test_cli_version_flag_does_not_require_input(self) -> None:
|
||||
result = self.run_cli("--version")
|
||||
|
||||
self.assertEqual(0, result.returncode)
|
||||
self.assertIn("patchwork 0.3.0", result.stdout)
|
||||
|
||||
def test_config_dump_and_keep_file_merge_options(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as raw:
|
||||
directory = Path(raw)
|
||||
source = directory / "sample.py"
|
||||
config = directory / "config.json"
|
||||
keep_file = directory / "keep.txt"
|
||||
dumped = directory / "effective.json"
|
||||
source.write_text("def public():\n return 1\n", encoding="utf-8")
|
||||
config.write_text(json.dumps({"layers": 2, "rename": False, "keep": ["public"]}), encoding="utf-8")
|
||||
keep_file.write_text("external_name\n# ignored\n", encoding="utf-8")
|
||||
|
||||
result = self.run_cli(str(source), "--config", str(config), "--keep-file", str(keep_file), "--dump-config", str(dumped), "--dry-run", "--quiet")
|
||||
|
||||
self.assertEqual(0, result.returncode)
|
||||
effective = json.loads(dumped.read_text(encoding="utf-8"))
|
||||
self.assertEqual(2, effective["layers"])
|
||||
self.assertFalse(effective["rename"])
|
||||
self.assertEqual(["external_name", "public"], effective["keep"])
|
||||
|
||||
def test_dry_run_does_not_write_obfuscated_output(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as raw:
|
||||
source = Path(raw) / "sample.py"
|
||||
output = Path(raw) / "sample_obf.py"
|
||||
source.write_text("print('dry run')\n", encoding="utf-8")
|
||||
|
||||
result = self.run_cli(str(source), "-o", str(output), "--dry-run", "--quiet")
|
||||
|
||||
self.assertEqual(0, result.returncode)
|
||||
self.assertFalse(output.exists())
|
||||
|
||||
def test_cli_writes_stats_report_manifest_and_verifies_manifest(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as raw:
|
||||
directory = Path(raw)
|
||||
source = directory / "sample.py"
|
||||
output = directory / "sample_obf.py"
|
||||
stats = directory / "stats.json"
|
||||
report = directory / "report.html"
|
||||
manifest = directory / "manifest.json"
|
||||
source.write_text("print('manifest')\n", encoding="utf-8")
|
||||
|
||||
result = self.run_cli(
|
||||
str(source),
|
||||
"-o",
|
||||
str(output),
|
||||
"--seed",
|
||||
"123",
|
||||
"--stats-json",
|
||||
str(stats),
|
||||
"--report",
|
||||
str(report),
|
||||
"--manifest",
|
||||
str(manifest),
|
||||
"--quiet",
|
||||
)
|
||||
|
||||
self.assertEqual(0, result.returncode)
|
||||
self.assertTrue(output.exists())
|
||||
self.assertTrue(report.read_text(encoding="utf-8").startswith("<!doctype html>"))
|
||||
stats_doc = json.loads(stats.read_text(encoding="utf-8"))
|
||||
self.assertGreater(stats_doc["output_bytes"], stats_doc["input_bytes"])
|
||||
self.assertTrue(verify_manifest(manifest)["valid"])
|
||||
verify_result = self.run_cli("--verify-manifest", str(manifest), "--quiet")
|
||||
self.assertEqual(0, verify_result.returncode)
|
||||
|
||||
def test_budget_gate_refuses_to_write_large_output(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as raw:
|
||||
source = Path(raw) / "sample.py"
|
||||
output = Path(raw) / "sample_obf.py"
|
||||
source.write_text("print('too large')\n", encoding="utf-8")
|
||||
|
||||
result = self.run_cli(str(source), "-o", str(output), "--max-output-bytes", "1", "--quiet")
|
||||
|
||||
self.assertEqual(4, result.returncode)
|
||||
self.assertFalse(output.exists())
|
||||
self.assertIn("exceed max_output_bytes", result.stderr)
|
||||
|
||||
def test_review_indicator_gate_refuses_sensitive_input(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as raw:
|
||||
source = Path(raw) / "sample.py"
|
||||
source.write_text("import subprocess\n", encoding="utf-8")
|
||||
|
||||
result = self.run_cli(str(source), "--max-review-indicators", "0", "--quiet")
|
||||
|
||||
self.assertEqual(4, result.returncode)
|
||||
self.assertIn("review indicators", result.stderr)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user