diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b00eff1..c1ce52f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -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 diff --git a/README.md b/README.md index 1777c4e..0cdcd88 100644 --- a/README.md +++ b/README.md @@ -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: _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, diff --git a/patchwork/__init__.py b/patchwork/__init__.py index 71e8a42..8cc9ea5 100644 --- a/patchwork/__init__.py +++ b/patchwork/__init__.py @@ -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', +] diff --git a/patchwork/audit.py b/patchwork/audit.py index a4ca0fb..12d20bd 100644 --- a/patchwork/audit.py +++ b/patchwork/audit.py @@ -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) diff --git a/patchwork/cli.py b/patchwork/cli.py index 3b86ba0..317052a 100644 --- a/patchwork/cli.py +++ b/patchwork/cli.py @@ -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()) diff --git a/patchwork/config.py b/patchwork/config.py new file mode 100644 index 0000000..1d0ef6b --- /dev/null +++ b/patchwork/config.py @@ -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 diff --git a/patchwork/report.py b/patchwork/report.py new file mode 100644 index 0000000..4d83079 --- /dev/null +++ b/patchwork/report.py @@ -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( + "" + f"{_escape(indicator.get('line', ''))}" + f"{_escape(indicator.get('kind', ''))}" + f"{_escape(indicator.get('name', ''))}" + f"{_escape(indicator.get('reason', ''))}" + "" + ) + return "\n".join(rows) or 'No review indicators found.' + + +def _option_rows(options: dict[str, object]) -> str: + return "\n".join( + f"{_escape(key)}{_escape(value)}" + 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""" + + + + + Patchwork Build Report + + + +
+

Patchwork Build Report

+
+

Source Audit

+

Path: {_escape(audit.get('path'))}

+

SHA-256: {_escape(audit.get('sha256'))}

+

Lines: {_escape(audit.get('lines'))}

+

AST nodes: {_escape(audit.get('ast_nodes'))}

+

Review indicators: {_escape(review.get('indicator_count', 0))}

+
+
+

Output Stats

+

Output SHA-256: {_escape(stats.get('output_sha256', 'not generated'))}

+

Input bytes: {_escape(stats.get('input_bytes', 'not generated'))}

+

Output bytes: {_escape(stats.get('output_bytes', 'not generated'))}

+

Expansion ratio: {_escape(stats.get('ratio', 'not generated'))}

+
+
+

Options

+ + + {_option_rows(options)} +
OptionValue
+
+
+

Review Indicators

+ + + {_indicator_rows(audit)} +
LineKindNameReason
+
+
+ + +""" + + +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 diff --git a/pyproject.toml b/pyproject.toml index 0cc152b..4689087 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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", ] diff --git a/tests/test_audit.py b/tests/test_audit.py index b2c5528..fb24211 100644 --- a/tests/test_audit.py +++ b/tests/test_audit.py @@ -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("")) + 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()