#!/usr/bin/env python3 """RAT analyzer — identify, decompile, and extract configurations from Remote Access Trojans.""" from __future__ import annotations import argparse import base64 import hashlib import json import re import sys from pathlib import Path def compute_hashes(filepath: str) -> dict: """Compute file hashes.""" data = Path(filepath).read_bytes() return { "md5": hashlib.md5(data).hexdigest(), "sha256": hashlib.sha256(data).hexdigest(), "size_bytes": len(data), } def is_dotnet_binary(filepath: str) -> bool: """Check if a PE file is a .NET assembly.""" try: import pefile pe = pefile.PE(filepath) clr_dir = pe.OPTIONAL_HEADER.DATA_DIRECTORY[14] return clr_dir.Size > 0 except Exception: return False def identify_rat_family(filepath: str) -> dict: """Identify the RAT family based on binary markers.""" data = Path(filepath).read_bytes() families = { "AsyncRAT": [b"AsyncClient", b"AsyncRAT", b"Async_RAT", b"pastebin.com/raw/"], "NjRAT": [b"njq8", b"|'\\|", b"njRAT", b"Njrat"], "QuasarRAT": [b"Client.Config", b"Quasar.Client", b"QuasarRAT"], "Remcos": [b"Remcos", b"SETTINGS", b"breakingsecurity", b"IPLK"], "DarkComet": [b"DC_MUTEX-", b"DarkComet", b"#KCMDDC", b"DCLIB"], "Warzone": [b"AVE_MARIA", b"WarzoneRAT", b"Warzone"], "DcRAT": [b"dcrat", b"DcRAT", b"AsyncMutex"], "Venom_RAT": [b"VenomRAT", b"Venom-RAT"], } results = {"family": "unknown", "confidence": "low", "markers_found": []} best_match = "" best_count = 0 for family, markers in families.items(): found = [m.decode("utf-8", errors="ignore") for m in markers if m in data] if len(found) > best_count: best_count = len(found) best_match = family results["markers_found"] = found if best_count >= 2: results["family"] = best_match results["confidence"] = "high" elif best_count == 1: results["family"] = best_match results["confidence"] = "medium" results["is_dotnet"] = is_dotnet_binary(filepath) return results def extract_config_strings(filepath: str) -> dict: """Extract potential configuration strings from the binary.""" data = Path(filepath).read_bytes() config = {"c2_hosts": [], "ports": [], "mutexes": [], "keys": [], "paths": []} # Extract potential base64-encoded configs b64_pattern = rb'[A-Za-z0-9+/]{20,}={0,2}' for match in re.finditer(b64_pattern, data): try: decoded = base64.b64decode(match.group()) text = decoded.decode("utf-8", errors="ignore") if re.match(r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}', text): config["c2_hosts"].append(text) except Exception: pass # Extract mutex patterns mutex_patterns = [ rb'(?:Global\\|Local\\)?[A-Za-z0-9_]{5,50}Mutex[A-Za-z0-9_]*', rb'DC_MUTEX-[A-Za-z0-9]+', rb'njq8[A-Za-z0-9]*', rb'AVE_MARIA[A-Za-z0-9]*', rb'AsyncMutex_[A-Za-z0-9]+', rb'IPLK[A-Za-z0-9]*', ] for pattern in mutex_patterns: for match in re.finditer(pattern, data): config["mutexes"].append(match.group().decode("utf-8", errors="ignore")) # Extract IP:port patterns ip_port = rb'(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}):(\d{1,5})' for match in re.finditer(ip_port, data): config["c2_hosts"].append(match.group(1).decode()) config["ports"].append(match.group(2).decode()) # Extract domain patterns domain_pattern = rb'[a-z0-9][-a-z0-9]{0,62}\.[a-z]{2,6}' for match in re.finditer(domain_pattern, data): domain = match.group().decode() if domain.count(".") >= 1 and not domain.endswith((".dll", ".exe", ".sys")): config["c2_hosts"].append(domain) # Deduplicate config["c2_hosts"] = list(set(config["c2_hosts"])) config["ports"] = list(set(config["ports"])) config["mutexes"] = list(set(config["mutexes"])) return config def enumerate_capabilities(filepath: str) -> list[str]: """Enumerate RAT capabilities based on API imports and strings.""" data = Path(filepath).read_bytes() capability_indicators = { "keylogger": [b"GetAsyncKeyState", b"SetWindowsHookEx", b"keylog", b"KeyboardHook"], "screen_capture": [b"CopyFromScreen", b"BitBlt", b"GetDesktopWindow", b"screenshot"], "webcam": [b"avicap32", b"capCreateCaptureWindow", b"webcam", b"camera"], "audio_recording": [b"mciSendString", b"waveInOpen", b"microphone"], "file_manager": [b"GetFiles", b"Upload", b"Download", b"FileManager"], "remote_shell": [b"cmd.exe", b"Process.Start", b"ShellExecute", b"/c "], "browser_credentials": [b"Login Data", b"logins.json", b"chrome", b"firefox"], "clipboard": [b"GetClipboardData", b"ClipboardChanged", b"SetClipboardViewer"], "process_manager": [b"GetProcesses", b"TerminateProcess", b"ProcessManager"], "reverse_proxy": [b"SOCKS", b"proxy", b"ProxyClient", b"PortForward"], "ddos": [b"UDPFlood", b"TCPFlood", b"HTTPFlood", b"SlowLoris"], "usb_spread": [b"RemovableDrive", b"autorun.inf", b"USB"], "hvnc": [b"HVNC", b"HiddenDesktop", b"CreateDesktop"], "ransomware": [b"encrypt", b"ransom", b"AES", b"RSA"], } capabilities = [] for capability, indicators in capability_indicators.items(): if any(ind in data for ind in indicators): capabilities.append(capability) return capabilities def analyze_persistence(filepath: str) -> dict: """Analyze persistence mechanisms in the binary.""" data = Path(filepath).read_bytes() persistence = {"methods": [], "details": []} indicators = { "registry_run_key": [b"CurrentVersion\\Run", b"CurrentVersion\\\\Run"], "startup_folder": [b"Startup", b"Start Menu\\Programs\\Startup"], "scheduled_task": [b"schtasks", b"TaskScheduler", b"Register-ScheduledTask"], "wmi_subscription": [b"EventFilter", b"CommandLineEventConsumer", b"__InstanceCreation"], "service_install": [b"CreateService", b"ServiceMain", b"StartService"], } for method, patterns in indicators.items(): if any(p in data for p in patterns): persistence["methods"].append(method) return persistence def main() -> None: parser = argparse.ArgumentParser(description="RAT Analyzer") parser.add_argument("--input", "--sample", required=True, help="Sample file path") parser.add_argument( "--mode", choices=["identify", "config", "capabilities", "c2-protocol", "persistence", "plugins", "iocs"], default="identify", help="Analysis mode", ) parser.add_argument("--output", default="rat_analysis.json", help="Output file path") parser.add_argument("--format", choices=["json", "csv", "markdown"], default="json") args = parser.parse_args() sample_path = Path(args.sample) if not sample_path.exists(): print(f"[!] Sample not found: {args.sample}", file=sys.stderr) sys.exit(1) print(f"[*] RAT Analysis — mode: {args.mode}") hashes = compute_hashes(args.sample) print(f"[*] SHA-256: {hashes['sha256']}") results = {"sample": args.sample, "hashes": hashes, "mode": args.mode} if args.mode == "identify": results["identification"] = identify_rat_family(args.sample) elif args.mode == "config": results["identification"] = identify_rat_family(args.sample) results["config"] = extract_config_strings(args.sample) elif args.mode == "capabilities": results["capabilities"] = enumerate_capabilities(args.sample) elif args.mode == "persistence": results["persistence"] = analyze_persistence(args.sample) elif args.mode == "iocs": results["identification"] = identify_rat_family(args.sample) results["config"] = extract_config_strings(args.sample) results["capabilities"] = enumerate_capabilities(args.sample) results["persistence"] = analyze_persistence(args.sample) output_path = Path(args.output) output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_text(json.dumps(results, indent=2)) print(f"[*] Results written to {args.output}") if __name__ == "__main__": main()