| """ |
| Reverse Engineering Agent |
| ======================== |
| Multi-stage AI agent for reverse engineering SOFTWARE of any kind: |
| - Binaries: ELF / PE / Mach-O / raw (default mode) |
| - Web apps / sites: --mode web |
| - Android apps: --mode apk |
| - iOS apps: --mode ipa |
| |
| Decompilation backends: auto | ghidra | radare2 | llm4decompile | none |
| |
| Usage: |
| python agent.py analyze ./target.bin --decomp llm4decompile |
| python agent.py analyze ./target.bin # binary |
| python agent.py analyze https://example.com --mode web # website |
| python agent.py analyze app.apk --mode apk # android |
| python agent.py analyze app.ipa --mode ipa # ios |
| """ |
|
|
| from __future__ import annotations |
| import os, sys, json, hashlib, subprocess, shutil, re, zipfile, urllib.request, urllib.parse |
| from pathlib import Path |
| from dataclasses import dataclass, field, asdict |
| from typing import List, Dict, Any, Optional |
| from collections import defaultdict |
|
|
| try: |
| import typer; TYPER = True |
| except ImportError: TYPER = False |
| try: |
| from rich.console import Console; from rich.table import Table; RICH = True |
| except ImportError: RICH = False |
| try: |
| import lief; LIEF = True |
| except ImportError: LIEF = False |
| try: |
| import capstone; CAPSTONE = True |
| except ImportError: CAPSTONE = False |
| try: |
| import openai; OPENAI = True |
| except ImportError: OPENAI = False |
| try: |
| from bs4 import BeautifulSoup; BS4 = True |
| except ImportError: BS4 = False |
|
|
| console = Console() if RICH else None |
|
|
| @dataclass |
| class Function: |
| address: int = 0 |
| name: str = "" |
| size: int = 0 |
| summary: str = "" |
| tags: List[str] = field(default_factory=list) |
| decompiled: str = "" |
|
|
| @dataclass |
| class AnalysisContext: |
| target: Path |
| mode: str = "binary" |
| sha256: str = "" |
| md5: str = "" |
| size: int = 0 |
| file_type: str = "" |
| format: str = "" |
| arch: str = "" |
| bits: int = 0 |
| endian: str = "little" |
| entrypoint: int = 0 |
| sections: List[Dict[str, Any]] = field(default_factory=list) |
| imports: List[Dict[str, Any]] = field(default_factory=list) |
| exports: List[Dict[str, Any]] = field(default_factory=list) |
| symbols: List[Dict[str, Any]] = field(default_factory=list) |
| strings: List[Dict[str, Any]] = field(default_factory=list) |
| functions: List[Function] = field(default_factory=list) |
| disassembly: Dict[int, List[str]] = field(default_factory=dict) |
| decompiled: Dict[int, str] = field(default_factory=dict) |
| callgraph: Dict[str, List[str]] = field(default_factory=dict) |
| report: str = "" |
| metadata: Dict[str, Any] = field(default_factory=dict) |
| errors: List[str] = field(default_factory=list) |
| endpoints: List[str] = field(default_factory=list) |
| secrets: List[Dict[str, str]] = field(default_factory=list) |
| assets: List[str] = field(default_factory=list) |
| permissions: List[str] = field(default_factory=list) |
| components: List[Dict[str, str]] = field(default_factory=list) |
|
|
| def to_dict(self): |
| d = asdict(self); d["target"] = str(self.target); return d |
|
|
| class Stage: |
| name: str = "stage" |
| def run(self, ctx): raise NotImplementedError |
| def log(self, msg): |
| if RICH: console.log(f"[cyan]{self.name}[/cyan]: {msg}") |
| else: print(f"[{self.name}] {msg}") |
|
|
| class FileIdentificationStage(Stage): |
| name = "file_identification" |
| def run(self, ctx): |
| if ctx.mode != "binary": return ctx |
| p = ctx.target; data = p.read_bytes() |
| ctx.size = len(data); ctx.sha256 = hashlib.sha256(data).hexdigest(); ctx.md5 = hashlib.md5(data).hexdigest() |
| magic = data[:16] |
| if magic[:4] == b"\x7fELF": ctx.format = "ELF" |
| elif magic[:2] == b"MZ": ctx.format = "PE" |
| elif magic[:4] in (b"\xfe\xed\xfa\xce", b"\xce\xfa\xed\xfe", b"\xfe\xed\xfa\xcf", b"\xcf\xfa\xed\xfe"): ctx.format = "Mach-O" |
| else: ctx.format = "raw" |
| if LIEF: |
| try: |
| binary = lief.parse(str(p)) |
| if binary is not None: |
| ctx.file_type = str(binary) |
| if ctx.format == "ELF": |
| ctx.arch = str(binary.header.machine_type) |
| ctx.bits = 64 if binary.header.identity_class == lief.ELF.ELF_CLASS.CLASS64 else 32 |
| ctx.endian = "big" if binary.header.identity_data == lief.ELF.ELF_DATA.DATA_MSB else "little" |
| ctx.entrypoint = binary.header.entrypoint |
| for s in binary.sections: ctx.sections.append({"name": s.name, "offset": s.offset, "size": s.size, "virtual_address": s.virtual_address, "type": str(s.type)}) |
| for imp in binary.imported_symbols: ctx.imports.append({"name": imp.name, "library": getattr(imp, "library", "")}) |
| for exp in binary.exported_symbols: ctx.exports.append({"name": exp.name, "address": exp.value}) |
| for sym in binary.symbols: ctx.symbols.append({"name": sym.name, "value": sym.value, "size": getattr(sym, "size", 0), "type": str(getattr(sym, "type", ""))}) |
| elif ctx.format == "PE": |
| ctx.arch = str(binary.header.machine) |
| ctx.bits = 64 if binary.header.optional_header.magic == 0x20b else 32 |
| ctx.entrypoint = binary.optional_header.addressof_entrypoint |
| for s in binary.sections: ctx.sections.append({"name": s.name, "offset": s.pointerto_raw_data, "size": s.sizeof_raw_data, "virtual_address": s.virtual_address}) |
| for imp in binary.imports: |
| for f in imp.entries: ctx.imports.append({"name": f.name, "library": imp.name}) |
| for exp in binary.exported_functions: ctx.exports.append({"name": exp.name, "address": exp.address}) |
| elif ctx.format == "Mach-O": |
| ctx.arch = str(binary.header.cpu_type); ctx.bits = 64 if binary.header.is_64 else 32 |
| ctx.entrypoint = getattr(binary, "entrypoint", 0) |
| except Exception as e: ctx.errors.append(f"lief parse failed: {e}") |
| if not ctx.file_type: |
| try: |
| out = subprocess.run(["file", "-b", str(p)], capture_output=True, text=True, timeout=30) |
| ctx.file_type = out.stdout.strip() |
| except Exception as e: ctx.errors.append(f"file(1) failed: {e}") |
| self.log(f"{ctx.format} {ctx.arch} {ctx.bits}-bit, {ctx.size} bytes, {len(ctx.sections)} sections") |
| return ctx |
|
|
| class StringExtractionStage(Stage): |
| name = "string_extraction" |
| def __init__(self, min_len=4): self.min_len = min_len |
| def run(self, ctx): |
| if ctx.mode != "binary": return ctx |
| data = ctx.target.read_bytes() |
| for m in re.finditer(rb"[\x20-\x7e]{" + str(self.min_len).encode() + rb",}", data): |
| ctx.strings.append({"addr": m.start(), "value": m.group().decode("ascii", "replace"), "type": "ascii"}) |
| for m in re.finditer(rb"(?:[\x20-\x7e]\x00){" + str(self.min_len).encode() + rb",}", data): |
| try: ctx.strings.append({"addr": m.start(), "value": m.group().decode("utf-16-le"), "type": "utf16le"}) |
| except Exception: pass |
| url_re = re.compile(r"https?://[^\s\"'<>]+", re.I) |
| path_re = re.compile(r"(?:[A-Za-z]:\\|/)[^\s\"'<>]+") |
| for s in ctx.strings: |
| v = s["value"]; tags = [] |
| if url_re.search(v): tags.append("url") |
| if path_re.search(v): tags.append("path") |
| if re.search(r"[A-Fa-f0-9]{32,}", v): tags.append("hexblob") |
| if any(k in v.lower() for k in ("password", "secret", "token", "api_key")): tags.append("credential-hint") |
| s["tags"] = tags |
| self.log(f"extracted {len(ctx.strings)} strings") |
| return ctx |
|
|
| class DisassemblyStage(Stage): |
| name = "disassembly" |
| def __init__(self, max_bytes=1 << 20): self.max_bytes = max_bytes |
| def run(self, ctx): |
| if ctx.mode != "binary" or not CAPSTONE: |
| if ctx.mode == "binary" and not CAPSTONE: ctx.errors.append("capstone not installed") |
| return ctx |
| data = ctx.target.read_bytes() |
| code_sections = [s for s in ctx.sections if "text" in s.get("name", "").lower() or "code" in s.get("name", "").lower()] |
| if not code_sections and ctx.sections: code_sections = [max(ctx.sections, key=lambda s: s.get("size", 0))] |
| if "x86" in ctx.arch.lower() or "amd" in ctx.arch.lower() or "i386" in ctx.arch.lower(): |
| archmode = (capstone.CS_ARCH_X86, capstone.CS_MODE_64 if ctx.bits == 64 else capstone.CS_MODE_32) |
| elif "arm" in ctx.arch.lower() and ctx.bits == 64: |
| archmode = (capstone.CS_ARCH_ARM64, capstone.CS_MODE_ARM) |
| elif "arm" in ctx.arch.lower(): |
| archmode = (capstone.CS_ARCH_ARM, capstone.CS_MODE_ARM) |
| else: |
| archmode = (capstone.CS_ARCH_X86, capstone.CS_MODE_64) |
| md = capstone.Cs(*archmode); md.detail = True |
| for sec in code_sections: |
| off = sec.get("offset", 0); size = min(sec.get("size", 0), self.max_bytes) |
| if size <= 0: continue |
| va = sec.get("virtual_address", off); chunk = data[off:off + size] |
| ctx.disassembly[va] = [f"0x{ins.address:x}: {ins.mnemonic:<8} {ins.op_str}" for ins in md.disasm(chunk, va)] |
| self.log(f"disassembled {len(ctx.disassembly)} sections") |
| return ctx |
|
|
| class DecompilationStage(Stage): |
| name = "decompilation" |
| def __init__(self, backend="auto"): self.backend = backend |
| def run(self, ctx): |
| backend = self.backend |
| if backend == "auto": |
| if shutil.which("analyzeHeadless"): backend = "ghidra" |
| elif shutil.which("r2"): backend = "radare2" |
| else: backend = "none" |
| if backend == "ghidra": self._ghidra(ctx) |
| elif backend == "radare2": self._radare2(ctx) |
| elif backend == "llm4decompile": self._llm4decompile(ctx) |
| else: ctx.errors.append("no decompiler backend found") |
| return ctx |
| def _ghidra(self, ctx): |
| work = Path("/tmp/ghidra_proj"); work.mkdir(parents=True, exist_ok=True) |
| script = work / "DecompileAll.py" |
| script.write_text('import ghidra.app.decompiler as Decomp\nfrom ghidra.util.task import ConsoleTaskMonitor\nfm=currentProgram.getFunctionManager()\nfor f in fm.getFunctions(True):\n d=Decomp.DecompInterface(); d.openProgram(currentProgram)\n r=d.decompileFunction(f,30,ConsoleTaskMonitor())\n if r and r.decompiledFunction:\n print("FUNC::"+f.getName()+"::"+hex(f.getEntryPoint().getOffset()))\n print(r.decompiledFunction.getC())\n') |
| try: |
| subprocess.run(["analyzeHeadless", str(work), "proj", "-import", str(ctx.target), "-postScript", str(script), "-deleteProject"], capture_output=True, text=True, timeout=600) |
| except Exception as e: ctx.errors.append(f"ghidra failed: {e}") |
| def _radare2(self, ctx): |
| if not shutil.which("r2"): return |
| try: |
| out = subprocess.run(["r2", "-q", "-c", "aaa; pdfj @ main", str(ctx.target)], capture_output=True, text=True, timeout=300) |
| ctx.decompiled[0] = out.stdout |
| except Exception as e: ctx.errors.append(f"radare2 failed: {e}") |
| def _llm4decompile(self, ctx): |
| |
| asm_path = Path("/tmp") / (ctx.target.stem + ".asm") |
| try: |
| subprocess.run(["objdump", "-d", str(ctx.target)], stdout=open(asm_path, "w"), check=True, timeout=300) |
| except Exception as e: |
| ctx.errors.append(f"objdump failed: {e}"); return |
| |
| try: |
| sys.path.insert(0, str(Path(__file__).parent)) |
| from llm4decompile_backend import LLM4DecompileBackend |
| except Exception as e: |
| ctx.errors.append(f"llm4decompile backend import failed: {e}"); return |
| if not LLM4DecompileBackend.available(): |
| ctx.errors.append("transformers/torch not installed"); return |
| model = os.environ.get("LLM4DECOMPILE_MODEL", "LLM4Binary/llm4decompile-6.7b-v2") |
| mode = "ref" if "v2" in model else "end" |
| try: |
| be = LLM4DecompileBackend(model=model, mode=mode) |
| asm = asm_path.read_text() |
| |
| code = be.decompile(asm) if mode == "end" else be.refine(asm) |
| ctx.decompiled[0] = code |
| ctx.metadata["llm4decompile_model"] = model |
| self.log(f"llm4decompile produced {len(code)} chars") |
| except Exception as e: |
| ctx.errors.append(f"llm4decompile failed: {e}") |
|
|
| class WebFetchStage(Stage): |
| name = "web_fetch" |
| def run(self, ctx): |
| if ctx.mode != "web": return ctx |
| url = str(ctx.target) |
| try: |
| req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0 (RE-Agent)"}) |
| with urllib.request.urlopen(req, timeout=30) as r: |
| html = r.read().decode("utf-8", "replace") |
| ctx.metadata["html"] = html |
| ctx.size = len(html.encode()) |
| ctx.sha256 = hashlib.sha256(html.encode()).hexdigest() |
| ctx.file_type = "text/html" |
| scripts = [] |
| if BS4: |
| soup = BeautifulSoup(html, "html.parser") |
| for tag in soup.find_all("script"): |
| src = tag.get("src") |
| if src: scripts.append(urllib.parse.urljoin(url, src)) |
| elif tag.string: scripts.append(("inline", tag.string)) |
| else: |
| for m in re.finditer(r'<script[^>]*src=["\']([^"\']+)["\']', html): |
| scripts.append(urllib.parse.urljoin(url, m.group(1))) |
| ctx.metadata["scripts"] = scripts |
| forms = [] |
| if BS4: |
| soup = BeautifulSoup(html, "html.parser") |
| for f in soup.find_all("form"): |
| forms.append({"action": f.get("action"), "method": f.get("method", "get"), |
| "inputs": [i.get("name") for i in f.find_all("input")]}) |
| ctx.metadata["forms"] = forms |
| self.log(f"fetched {len(html)} bytes, {len(scripts)} scripts, {len(forms)} forms") |
| except Exception as e: |
| ctx.errors.append(f"fetch failed: {e}") |
| return ctx |
|
|
| class WebAnalyzeStage(Stage): |
| name = "web_analyze" |
| def run(self, ctx): |
| if ctx.mode != "web": return ctx |
| html = ctx.metadata.get("html", "") |
| text = html |
| for s in ctx.metadata.get("scripts", []): |
| if isinstance(s, tuple): |
| text += "\n" + s[1]; continue |
| try: |
| req = urllib.request.Request(s, headers={"User-Agent": "Mozilla/5.0 (RE-Agent)"}) |
| with urllib.request.urlopen(req, timeout=20) as r: |
| text += "\n" + r.read().decode("utf-8", "replace") |
| except Exception: pass |
| ep_patterns = [ |
| r'["\'](/api/[A-Za-z0-9_\-/\.]+)["\']', |
| r'["\'](https?://[A-Za-z0-9.\-]+(?::\d+)?/[A-Za-z0-9_\-/\.]+)["\']', |
| r'fetch\(["\']([^"\']+)["\']', |
| r'axios\.(get|post|put|delete)\(["\']([^"\']+)["\']', |
| ] |
| eps = set() |
| for pat in ep_patterns: |
| for m in re.finditer(pat, text): |
| eps.add(m.group(m.lastindex)) |
| ctx.endpoints = sorted(eps) |
| secret_patterns = [ |
| ("api_key", r'(?i)(api[_-]?key|apikey)\s*[:=]\s*["\']([A-Za-z0-9_\-]{16,})["\']'), |
| ("token", r'(?i)(token|bearer)\s*[:=]\s*["\']([A-Za-z0-9_\-\.]{16,})["\']'), |
| ("aws_key", r'AKIA[0-9A-Z]{16}'), |
| ("jwt", r'eyJ[A-Za-z0-9_\-]+\.[A-Za-z0-9_\-]+\.[A-Za-z0-9_\-]+'), |
| ("private_key", r'-----BEGIN (RSA |EC )?PRIVATE KEY-----'), |
| ("password", r'(?i)password\s*[:=]\s*["\']([^"\']{4,})["\']'), |
| ] |
| for name, pat in secret_patterns: |
| for m in re.finditer(pat, text): |
| val = m.group(m.lastindex) if m.lastindex else m.group(0) |
| ctx.secrets.append({"type": name, "value": val[:80]}) |
| for m in re.finditer(r'//# sourceMappingURL=(\S+)', text): |
| ctx.assets.append("sourcemap: " + m.group(1)) |
| for m in re.finditer(r'["\']([^"\']{6,})["\']', text): |
| ctx.strings.append({"addr": 0, "value": m.group(1), "type": "js", "tags": []}) |
| self.log(f"endpoints={len(ctx.endpoints)} secrets={len(ctx.secrets)} strings={len(ctx.strings)}") |
| return ctx |
|
|
| class ApkUnzipStage(Stage): |
| name = "apk_unzip" |
| def run(self, ctx): |
| if ctx.mode not in ("apk", "ipa"): return ctx |
| p = ctx.target |
| try: |
| z = zipfile.ZipFile(p) |
| out = Path("/tmp/re_app") |
| if out.exists(): shutil.rmtree(out) |
| out.mkdir(parents=True) |
| z.extractall(out) |
| ctx.metadata["extracted"] = str(out) |
| ctx.assets = z.namelist() |
| ctx.size = p.stat().st_size |
| ctx.sha256 = hashlib.sha256(p.read_bytes()).hexdigest() |
| ctx.file_type = "zip/" + ctx.mode |
| self.log(f"extracted {len(ctx.assets)} entries") |
| except Exception as e: |
| ctx.errors.append(f"unzip failed: {e}") |
| return ctx |
|
|
| class ApkAnalyzeStage(Stage): |
| name = "apk_analyze" |
| def run(self, ctx): |
| if ctx.mode not in ("apk", "ipa"): return ctx |
| root = Path(ctx.metadata.get("extracted", "/tmp/re_app")) |
| if ctx.mode == "apk": |
| manifest = root / "AndroidManifest.xml" |
| if manifest.exists(): |
| try: |
| txt = manifest.read_text(errors="replace") |
| ctx.metadata["manifest_raw"] = txt |
| for m in re.finditer(r'android:permission="([^"]+)"', txt): |
| ctx.permissions.append(m.group(1)) |
| for m in re.finditer(r'<activity[^>]*android:name="([^"]+)"', txt): |
| ctx.components.append({"type": "activity", "name": m.group(1)}) |
| for m in re.finditer(r'<service[^>]*android:name="([^"]+)"', txt): |
| ctx.components.append({"type": "service", "name": m.group(1)}) |
| for m in re.finditer(r'<receiver[^>]*android:name="([^"]+)"', txt): |
| ctx.components.append({"type": "receiver", "name": m.group(1)}) |
| for m in re.finditer(r'<provider[^>]*android:name="([^"]+)"', txt): |
| ctx.components.append({"type": "provider", "name": m.group(1)}) |
| except Exception as e: ctx.errors.append(f"manifest parse: {e}") |
| for so in root.rglob("*.so"): |
| ctx.assets.append("native: " + str(so.relative_to(root))) |
| if shutil.which("jadx"): |
| for dex in root.rglob("*.dex"): |
| outdir = root / ("jadx_" + dex.stem) |
| subprocess.run(["jadx", "-d", str(outdir), str(dex)], capture_output=True, timeout=300) |
| for f in root.rglob("*"): |
| if f.is_file() and f.suffix in (".xml", ".json", ".txt", ".properties", ".smali"): |
| try: |
| txt = f.read_text(errors="replace") |
| for m in re.finditer(r'(?i)(api[_-]?key|secret|token|password)\s*[:=]\s*["\']?([A-Za-z0-9_\-]{8,})', txt): |
| ctx.secrets.append({"type": m.group(1), "value": m.group(2)[:80], "file": str(f.relative_to(root))}) |
| except Exception: pass |
| self.log(f"permissions={len(ctx.permissions)} components={len(ctx.components)} secrets={len(ctx.secrets)}") |
| else: |
| for f in root.rglob("*.app/*"): |
| if f.is_file(): ctx.assets.append(str(f.relative_to(root))) |
| plist = None |
| for f in root.rglob("Info.plist"): |
| plist = f; break |
| if plist: |
| ctx.metadata["plist"] = plist.read_text(errors="replace") |
| for m in re.finditer(r'<key>([^<]+)</key>\s*<string>([^<]+)</string>', ctx.metadata["plist"]): |
| if "URL" in m.group(1) or "Scheme" in m.group(1): |
| ctx.endpoints.append(m.group(2)) |
| for b in root.rglob("*.app/*"): |
| if b.is_file() and not b.suffix: |
| ctx.assets.append("binary: " + b.name) |
| self.log(f"ipa assets={len(ctx.assets)} endpoints={len(ctx.endpoints)}") |
| return ctx |
|
|
| class AIAugmentationStage(Stage): |
| name = "ai_augmentation" |
| def __init__(self, model="gpt-4o-mini", api_key=None): |
| self.model = model; self.api_key = api_key or os.environ.get("OPENAI_API_KEY") |
| def run(self, ctx): |
| if not OPENAI or not self.api_key: |
| self.log("openai not configured; heuristic mode"); self._heuristic(ctx); return ctx |
| client = openai.OpenAI(api_key=self.api_key) |
| sample = [s.get("value", "") for s in ctx.strings[:60]] + ctx.endpoints[:20] |
| prompt = "You are a reverse engineering assistant. Given extracted data from software, identify purpose, frameworks, crypto, networking, IOCs, permissions, risks. Return concise bullets.\n\nDATA:\n" + "\n".join(sample) |
| try: |
| resp = client.chat.completions.create(model=self.model, messages=[{"role": "user", "content": prompt}], temperature=0.2) |
| ctx.metadata["ai_analysis"] = resp.choices[0].message.content |
| except Exception as e: |
| ctx.errors.append(f"LLM call failed: {e}"); self._heuristic(ctx) |
| return ctx |
| def _heuristic(self, ctx): |
| ctx.metadata["heuristic"] = { |
| "strings": len(ctx.strings), "endpoints": len(ctx.endpoints), |
| "secrets": len(ctx.secrets), "permissions": len(ctx.permissions), |
| "components": len(ctx.components), |
| } |
|
|
| class ReportStage(Stage): |
| name = "report" |
| def run(self, ctx): |
| lines = ["# Reverse Engineering Report", "", "## Target", f"- Target: `{ctx.target}`", f"- Mode: {ctx.mode}", f"- Size: {ctx.size} bytes", f"- SHA256: `{ctx.sha256}`", ""] |
| if ctx.mode == "binary": |
| lines += [f"- Format: {ctx.format} ({ctx.arch}, {ctx.bits}-bit, {ctx.endian})", f"- Entrypoint: 0x{ctx.entrypoint:x}", "", f"## Sections ({len(ctx.sections)})"] |
| for s in ctx.sections: lines.append(f"- `{s.get('name')}` off={s.get('offset')} size={s.get('size')} va=0x{s.get('virtual_address',0):x}") |
| lines += ["", f"## Imports ({len(ctx.imports)})"] |
| for imp in ctx.imports[:80]: lines.append(f"- {imp.get('name')} ({imp.get('library','')})") |
| lines += ["", f"## Exports ({len(ctx.exports)})"] |
| for e in ctx.exports[:80]: lines.append(f"- {e.get('name')} @ 0x{e.get('address',0):x}") |
| if ctx.endpoints: |
| lines += ["", f"## Endpoints ({len(ctx.endpoints)})"] + [f"- {e}" for e in ctx.endpoints] |
| if ctx.secrets: |
| lines += ["", f"## Secrets / IOCs ({len(ctx.secrets)})"] |
| for s in ctx.secrets: lines.append(f"- [{s.get('type')}] {s.get('value')} {s.get('file','')}") |
| if ctx.permissions: |
| lines += ["", f"## Permissions ({len(ctx.permissions)})"] + [f"- {p}" for p in ctx.permissions] |
| if ctx.components: |
| lines += ["", f"## Components ({len(ctx.components)})"] |
| for c in ctx.components: lines.append(f"- [{c.get('type')}] {c.get('name')}") |
| if ctx.assets: |
| lines += ["", f"## Assets ({len(ctx.assets)})"] + [f"- {a}" for a in ctx.assets[:60]] |
| lines += ["", f"## Strings ({len(ctx.strings)})"] |
| for s in ctx.strings[:120]: lines.append(f"- {s.get('value','')[:120]}") |
| if ctx.metadata.get("ai_analysis"): lines += ["", "## AI Analysis", ctx.metadata["ai_analysis"]] |
| if ctx.metadata.get("llm4decompile_model"): lines += ["", f"## Decompiler: LLM4Decompile ({ctx.metadata['llm4decompile_model']})"] |
| if ctx.decompiled: |
| lines += ["", "## Decompiled (sample)"] |
| for k, v in list(ctx.decompiled.items())[:1]: |
| lines.append("```c") |
| lines.append(v[:3000]) |
| lines.append("```") |
| if ctx.errors: lines += ["", "## Errors"] + [f"- {e}" for e in ctx.errors] |
| ctx.report = "\n".join(lines) |
| self.log("report generated") |
| return ctx |
|
|
| class ReverseEngineeringAgent: |
| def __init__(self, mode="binary", model="gpt-4o-mini", min_str_len=4, decomp_backend="auto", api_key=None): |
| self.mode = mode |
| self.decomp_backend = decomp_backend |
| if mode == "binary": |
| self.stages = [FileIdentificationStage(), StringExtractionStage(min_len=min_str_len), DisassemblyStage(), DecompilationStage(backend=decomp_backend), AIAugmentationStage(model=model, api_key=api_key), ReportStage()] |
| elif mode == "web": |
| self.stages = [WebFetchStage(), WebAnalyzeStage(), AIAugmentationStage(model=model, api_key=api_key), ReportStage()] |
| elif mode in ("apk", "ipa"): |
| self.stages = [ApkUnzipStage(), ApkAnalyzeStage(), AIAugmentationStage(model=model, api_key=api_key), ReportStage()] |
| else: |
| raise ValueError(mode) |
| def analyze(self, target): |
| ctx = AnalysisContext(target=Path(target) if not str(target).startswith("http") else Path(target), mode=self.mode) |
| if self.mode == "web": ctx.target = Path(target) |
| for stage in self.stages: |
| if RICH: console.rule(f"[bold green]{stage.name}[/bold green]") |
| ctx = stage.run(ctx) |
| return ctx |
|
|
| def main(): |
| if not TYPER: |
| import argparse |
| ap = argparse.ArgumentParser() |
| sub = ap.add_subparsers(dest="cmd", required=True) |
| a = sub.add_parser("analyze") |
| a.add_argument("target") |
| a.add_argument("--mode", default="binary", choices=["binary", "web", "apk", "ipa"]) |
| a.add_argument("--out", default="report.md") |
| a.add_argument("--json", default=None) |
| a.add_argument("--model", default="gpt-4o-mini") |
| a.add_argument("--min-str", type=int, default=4) |
| a.add_argument("--decomp", default="auto", choices=["auto","ghidra","radare2","llm4decompile","none"]) |
| args = ap.parse_args() |
| if args.cmd == "analyze": |
| agent = ReverseEngineeringAgent(mode=args.mode, model=args.model, min_str_len=args.min_str, decomp_backend=args.decomp) |
| ctx = agent.analyze(args.target) |
| Path(args.out).write_text(ctx.report) |
| if args.json: Path(args.json).write_text(json.dumps(ctx.to_dict(), indent=2, default=str)) |
| print(f"report -> {args.out}") |
| return |
| app = typer.Typer(help="Reverse Engineering Agent (binary/web/apk/ipa)") |
| @app.command() |
| def analyze(target: str, mode: str = "binary", out: str = "report.md", json_out: Optional[str] = None, model: str = "gpt-4o-mini", min_str: int = 4, decomp: str = "auto"): |
| agent = ReverseEngineeringAgent(mode=mode, model=model, min_str_len=min_str, decomp_backend=decomp) |
| ctx = agent.analyze(target) |
| Path(out).write_text(ctx.report) |
| if json_out: Path(json_out).write_text(json.dumps(ctx.to_dict(), indent=2, default=str)) |
| typer.echo(f"report -> {out}") |
| app() |
|
|
| if __name__ == "__main__": |
| main() |
