cli.py

#!/usr/bin/env python3
"""
cage CLI — General C/C++ memory-error patching system.
Subcommands:
  cage run     --project <path> --build <compile_commands.json> [--out out] [--policy linux_kernel] [--no-clang]
  cage analyze --project <path> --build <compile_commands.json> [--out ctx.json] [--no-clang]
  cage synth   --ctx ctx.json [--out patches/] [--policy linux_kernel] [--profiles config] [--dry-run]
  cage verify  --project <path> --patches patches/ [--asan] [--ubsan]
Common:
  --verbose / --quiet; nonzero exit on failure; colored logs when TTY.
"""
from __future__ import annotations
import argparse, sys, json, os, shutil, signal
from pathlib import Path
from typing import Any, Dict
from builddb import load_compile_db
from ast_analyzer import Analyzer
from synthesizer import Synthesizer
from rewriter import Rewriter
from verifier import Verifier

VERSION = "0.3.0"

# ---------- logging ----------
def _isatty() -> bool: return sys.stderr.isatty()
C = {
    "Y": "\033[33m" if _isatty() else "",
    "G": "\033[32m" if _isatty() else "",
    "R": "\033[31m" if _isatty() else "",
    "X": "\033[0m"  if _isatty() else "",
}
def info(msg: str) -> None: 
    if not os.environ.get("CAGE_QUIET"): print(f"{C['Y']}[cage]{C['X']} {msg}", file=sys.stderr)
def ok(msg: str) -> None: 
    if not os.environ.get("CAGE_QUIET"): print(f"{C['G']}[ok]{C['X']} {msg}", file=sys.stderr)
def err(msg: str) -> None: print(f"{C['R']}[error]{C['X']} {msg}", file=sys.stderr)

# ---------- guards ----------
def _require_file(p: str, what: str) -> None:
    if not Path(p).is_file(): err(f"{what} not found: {p}"); sys.exit(2)
def _require_dir(p: str, what: str) -> None:
    if not Path(p).is_dir(): err(f"{what} not found: {p}"); sys.exit(2)

# ---------- commands ----------
def cmd_analyze(args: argparse.Namespace) -> None:
    _require_dir(args.project, "project")
    _require_file(args.build, "compile_commands.json")
    compdb = load_compile_db(args.build)
    analyzer = Analyzer(project_root=args.project, compile_db=compdb, prefer_clang=not args.no_clang)
    info(f"analyzing {args.project} (clang={'on' if not args.no_clang else 'off'})")
    ctx = analyzer.run()
    out = Path(args.out or "cage_context.json"); out.parent.mkdir(parents=True, exist_ok=True)
    out.write_text(json.dumps(ctx, indent=2, ensure_ascii=False), encoding="utf-8")
    ok(f"wrote {out} with {len(ctx.get('files', []))} files")

def cmd_synth(args: argparse.Namespace) -> None:
    _require_file(args.ctx, "context json")
    profiles_dir = Path(args.profiles or "config")
    if not profiles_dir.exists(): info(f"profiles dir not found, using defaults: {profiles_dir}")
    ctx = json.loads(Path(args.ctx).read_text(encoding="utf-8"))
    synth = Synthesizer(policy=args.policy, profiles_dir=str(profiles_dir))
    plan = synth.generate(ctx)
    if args.dry_run:
        print(json.dumps({"preview": plan.get("candidates", [])[:10], "total": len(plan.get("candidates", []))}, indent=2))
        return
    outd = Path(args.out or "patches"); outd.mkdir(parents=True, exist_ok=True)
    rew = Rewriter()
    index = rew.emit(ctx, plan, outd)
    (outd / "index.json").write_text(json.dumps(index, indent=2), encoding="utf-8")
    ok(f"patches: {len(index.get('patches', []))} -> {outd}")

def cmd_run(args: argparse.Namespace) -> None:
    out_root = Path(args.out or "out"); out_root.mkdir(parents=True, exist_ok=True)
    ctx_path = out_root / "ctx.json"
    cmd_analyze(argparse.Namespace(project=args.project, build=args.build, out=str(ctx_path), no_clang=args.no_clang))
    cmd_synth(argparse.Namespace(ctx=str(ctx_path), out=str(out_root / "patches"), policy=args.policy, profiles=args.profiles, dry_run=False))

def cmd_verify(args: argparse.Namespace) -> None:
    _require_dir(args.project, "project")
    _require_dir(args.patches, "patches dir")
    ver = Verifier(project_root=args.project)
    ok_ = ver.run(patches_dir=args.patches, use_asan=args.asan, use_ubsan=args.ubsan)
    sys.exit(0 if ok_ else 1)

# ---------- cli ----------
def main(argv: list[str] | None = None) -> None:
    signal.signal(signal.SIGINT, lambda *_: sys.exit(130))
    p = argparse.ArgumentParser("cage", description="C/C++ memory-error patching pipeline")
    p.add_argument("--version", action="version", version=f"%(prog)s {VERSION}")
    p.add_argument("--quiet", action="store_true", help="suppress info logs")
    sub = p.add_subparsers(dest="cmd", required=True)

    pa = sub.add_parser("analyze", help="analyze project to produce context JSON")
    pa.add_argument("--project", required=True); pa.add_argument("--build", required=True)
    pa.add_argument("--out"); pa.add_argument("--no-clang", action="store_true")
    pa.set_defaults(func=cmd_analyze)

    ps = sub.add_parser("synth", help="synthesize guards and generate patches")
    ps.add_argument("--ctx", required=True); ps.add_argument("--out")
    ps.add_argument("--policy", default=os.environ.get("CAGE_POLICY", "linux_kernel"))
    ps.add_argument("--profiles", default=os.environ.get("CAGE_PROFILES", "config"))
    ps.add_argument("--dry-run", action="store_true", help="preview candidates only")
    ps.set_defaults(func=cmd_synth)

    pr = sub.add_parser("run", help="one-shot analyze → synthesize")
    pr.add_argument("--project", required=True); pr.add_argument("--build", required=True)
    pr.add_argument("--out"); pr.add_argument("--policy", default=os.environ.get("CAGE_POLICY", "linux_kernel"))
    pr.add_argument("--profiles", default=os.environ.get("CAGE_PROFILES", "config"))
    pr.add_argument("--no-clang", action="store_true")
    pr.set_defaults(func=cmd_run)

    pv = sub.add_parser("verify", help="compile-smoke verify with sanitizers")
    pv.add_argument("--project", required=True); pv.add_argument("--patches", required=True)
    pv.add_argument("--asan", action="store_true"); pv.add_argument("--ubsan", action="store_true")
    pv.set_defaults(func=cmd_verify)

    args = p.parse_args(argv)
    if args.quiet: os.environ["CAGE_QUIET"] = "1"
    try:
        args.func(args)
    except FileNotFoundError as e:
        err(str(e)); sys.exit(2)
    except KeyboardInterrupt:
        err("interrupted"); sys.exit(130)
    except Exception as e:
        if os.environ.get("CAGE_VERBOSE"): raise
        err(f"unhandled error: {e}"); sys.exit(1)

if __name__ == "__main__":
    main()