#!/usr/bin/env python3 """Analyze infostealer malware samples for family identification, credential harvesting techniques, exfiltration channels, and configuration extraction. Supports: AgentTesla, Rhadamanthys, RedLine, Raccoon, Lumma, Vidar """ from __future__ import annotations import argparse import hashlib import json import re import struct import sys from datetime import datetime from pathlib import Path try: import pefile HAS_PEFILE = True except ImportError: HAS_PEFILE = False try: import yara HAS_YARA = True except ImportError: HAS_YARA = False # Known family signatures for quick identification FAMILY_SIGNATURES = { "agenttesla": { "strings": [b"|Sun|", b"SmtpClient", b"NetworkCredential", b"Aborting Thread"], "dotnet": True, "description": "AgentTesla .NET infostealer", }, "redline": { "strings": [b"StringDecrypt", b"CommandLine", b"ScanAll", b"SOAP-ENV"], "dotnet": True, "description": "RedLine Stealer", }, "raccoon": { "strings": [b"machineId", b"configId", b"sstmnfo_", b"dllbit"], "dotnet": False, "description": "Raccoon Stealer", }, "lumma": { "strings": [b"/api", b"/c2sock", b".bmp", b"lid="], "dotnet": False, "description": "Lumma Stealer (LummaC2)", }, "vidar": { "strings": [b"hwid=", b"os=", b"profile=", b"/1,/2,/3"], "dotnet": False, "description": "Vidar Stealer", }, "rhadamanthys": { "strings": [b"rhada", b"rhadamanthys"], "dotnet": False, "description": "Rhadamanthys Stealer", }, } # Browser credential database paths BROWSER_TARGETS = { "Chrome": r"%LOCALAPPDATA%\Google\Chrome\User Data\Default\Login Data", "Firefox": r"%APPDATA%\Mozilla\Firefox\Profiles\*\logins.json", "Edge": r"%LOCALAPPDATA%\Microsoft\Edge\User Data\Default\Login Data", "Opera": r"%APPDATA%\Opera Software\Opera Stable\Login Data", "Brave": r"%LOCALAPPDATA%\BraveSoftware\Brave-Browser\User Data\Default\Login Data", } # Cryptocurrency wallet paths WALLET_TARGETS = { "Bitcoin Core": r"%APPDATA%\Bitcoin\wallet.dat", "Electrum": r"%APPDATA%\Electrum\wallets", "Exodus": r"%APPDATA%\Exodus\exodus.wallet", "Atomic": r"%APPDATA%\atomic\Local Storage\leveldb", "Jaxx": r"%APPDATA%\com.liberty.jaxx\IndexedDB", "Coinomi": r"%LOCALAPPDATA%\Coinomi\Coinomi\wallets", } # Browser extension IDs for crypto wallets EXTENSION_IDS = { "nkbihfbeogaeaoehlefnkodbefgpgknn": "MetaMask", "ibnejdfjmmkpcnlpebklmnkoeoihofec": "TronLink", "fhbohimaelbohpjbbldcngcnapndodjp": "Binance Wallet", "hnfanknocfeofbddgcijnmhnfnkdnaad": "Coinbase Wallet", "bfnaelmomeimhlpmgjnjophhpkkoljpa": "Phantom", "aiifbnbfobpmeekipheeijimdpnlpgpp": "Station Wallet", } def compute_hashes(file_path) -> dict: """Compute MD5, SHA1, and SHA256 hashes for a sample.""" data = Path(file_path).read_bytes() return { "md5": hashlib.md5(data).hexdigest(), "sha1": hashlib.sha1(data).hexdigest(), "sha256": hashlib.sha256(data).hexdigest(), "size": len(data), } def is_dotnet_binary(file_path) -> dict: """Check if the PE file is a .NET assembly.""" if not HAS_PEFILE: return None try: pe = pefile.PE(file_path) for entry in pe.DIRECTORY_ENTRY_IMPORT: if b"mscoree.dll" in entry.dll.lower(): return True return False except Exception: return None def extract_strings(data, min_length=6) -> dict: """Extract ASCII and Unicode strings from binary data.""" ascii_strings = re.findall(rb"[\x20-\x7e]{%d,}" % min_length, data) unicode_strings = re.findall( rb"(?:[\x20-\x7e]\x00){%d,}" % min_length, data ) decoded_unicode = [s.decode("utf-16-le", errors="ignore") for s in unicode_strings] decoded_ascii = [s.decode("ascii", errors="ignore") for s in ascii_strings] return decoded_ascii + decoded_unicode def identify_family(data, strings_list) -> tuple: """Identify the infostealer family from binary content and strings.""" scores = {} for family, sigs in FAMILY_SIGNATURES.items(): score = 0 matched = [] for sig in sigs["strings"]: if sig in data: score += 1 matched.append(sig.decode(errors="ignore")) scores[family] = {"score": score, "total": len(sigs["strings"]), "matched": matched} best = max(scores, key=lambda k: scores[k]["score"]) if scores[best]["score"] == 0: return "unknown", 0.0, {} confidence = scores[best]["score"] / scores[best]["total"] return best, confidence, scores def analyze_browser_targeting(strings_list) -> list: """Analyze which browsers the stealer targets for credential theft.""" findings = { "targeted_browsers": [], "database_files": [], "crypto_apis": [], "sqlite_queries": [], } browser_keywords = { "Chrome": ["chrome", "Google\\Chrome", "Login Data"], "Firefox": ["firefox", "Mozilla\\Firefox", "logins.json", "key4.db"], "Edge": ["microsoft\\edge", "Edge\\User Data"], "Opera": ["opera", "Opera Software"], "Brave": ["brave", "BraveSoftware"], } for browser, keywords in browser_keywords.items(): for kw in keywords: if any(kw.lower() in s.lower() for s in strings_list): if browser not in findings["targeted_browsers"]: findings["targeted_browsers"].append(browser) break # Check for credential database interactions db_patterns = ["Login Data", "logins.json", "cookies.sqlite", "Cookies", "Web Data", "key4.db", "key3.db", "signons.sqlite"] for pat in db_patterns: if any(pat.lower() in s.lower() for s in strings_list): findings["database_files"].append(pat) # Check for DPAPI usage (Chrome credential decryption) dpapi_markers = ["CryptUnprotectData", "encrypted_key", "os_crypt", "Local State"] for marker in dpapi_markers: if any(marker in s for s in strings_list): findings["crypto_apis"].append(marker) # Check for SQLite queries for s in strings_list: if re.search(r"SELECT.*FROM.*(logins|moz_cookies|autofill)", s, re.IGNORECASE): findings["sqlite_queries"].append(s[:200]) return findings def analyze_exfiltration(data, strings_list) -> dict: """Identify data exfiltration methods and extract channel configurations.""" exfil = { "methods": [], "smtp_config": {}, "telegram_config": {}, "discord_config": {}, "http_config": {}, } # SMTP exfiltration smtp_indicators = [s for s in strings_list if re.search( r"(smtp\.|:587|:465|SmtpClient|NetworkCredential)", s, re.IGNORECASE )] if smtp_indicators: exfil["methods"].append("SMTP") emails = re.findall(rb"[\w\.\-]+@[\w\.\-]+\.\w{2,}", data) exfil["smtp_config"]["emails_found"] = list(set( e.decode(errors="ignore") for e in emails )) # Extract SMTP server smtp_servers = [s for s in strings_list if re.match( r"(smtp|mail)\.\w+\.\w+", s, re.IGNORECASE )] exfil["smtp_config"]["servers"] = smtp_servers[:5] # Telegram bot API telegram_tokens = re.findall(rb"\d{8,10}:[A-Za-z0-9_-]{35}", data) chat_ids = re.findall(rb"chat_id[=:](-?\d+)", data) if telegram_tokens or any("telegram" in s.lower() for s in strings_list): exfil["methods"].append("Telegram") exfil["telegram_config"]["bot_tokens"] = [ t.decode(errors="ignore") for t in telegram_tokens ] exfil["telegram_config"]["chat_ids"] = [ c.decode(errors="ignore") for c in chat_ids ] # Discord webhooks discord_hooks = re.findall( rb"https?://(?:discord|discordapp)\.com/api/webhooks/\d+/[\w-]+", data ) if discord_hooks: exfil["methods"].append("Discord") exfil["discord_config"]["webhooks"] = [ h.decode(errors="ignore") for h in discord_hooks ] # HTTP POST exfiltration http_indicators = [s for s in strings_list if re.search( r"(POST |Content-Type|multipart/form-data|application/x-www-form)", s )] if http_indicators and "SMTP" not in exfil["methods"]: exfil["methods"].append("HTTP POST") if not exfil["methods"]: exfil["methods"].append("unknown") return exfil def analyze_crypto_targeting(strings_list) -> list: """Analyze cryptocurrency wallet targeting and clipboard hijacking.""" findings = { "desktop_wallets": [], "browser_extensions": [], "clipboard_hijacking": False, "replacement_addresses": {}, } # Desktop wallet detection for wallet_name, wallet_path in WALLET_TARGETS.items(): path_fragment = wallet_path.split("\\")[-1].lower() if any(path_fragment in s.lower() for s in strings_list): findings["desktop_wallets"].append(wallet_name) # Browser extension wallet detection for ext_id, ext_name in EXTENSION_IDS.items(): if any(ext_id in s for s in strings_list): findings["browser_extensions"].append(ext_name) # Clipboard hijacking clipboard_apis = ["SetClipboardData", "GetClipboardData", "AddClipboardFormatListener", "OpenClipboard"] if any(api in s for s in strings_list for api in clipboard_apis): findings["clipboard_hijacking"] = True # Look for crypto address regex patterns (used for clipboard replacement) btc_pattern = re.compile(r"[13][a-km-zA-HJ-NP-Z1-9]{25,34}|bc1[a-zA-HJ-NP-Z0-9]{39,59}") eth_pattern = re.compile(r"0x[0-9a-fA-F]{40}") for s in strings_list: btc_match = btc_pattern.search(s) if btc_match: findings["replacement_addresses"]["BTC"] = btc_match.group() eth_match = eth_pattern.search(s) if eth_match: findings["replacement_addresses"]["ETH"] = eth_match.group() return findings def analyze_capabilities(strings_list) -> dict: """Enumerate stealer capabilities from API imports and strings.""" capabilities = { "keylogging": False, "screenshot_capture": False, "clipboard_monitoring": False, "webcam_capture": False, "file_grabber": False, "email_client_theft": False, "ftp_client_theft": False, "vpn_client_theft": False, } keylog_apis = ["SetWindowsHookEx", "GetAsyncKeyState", "GetKeyState", "WH_KEYBOARD_LL", "LowLevelKeyboardProc"] if any(api in s for s in strings_list for api in keylog_apis): capabilities["keylogging"] = True screenshot_apis = ["BitBlt", "CopyFromScreen", "GetDesktopWindow"] if any(api in s for s in strings_list for api in screenshot_apis): capabilities["screenshot_capture"] = True clipboard_apis = ["GetClipboardData", "SetClipboardData", "ClipboardChanged"] if any(api in s for s in strings_list for api in clipboard_apis): capabilities["clipboard_monitoring"] = True webcam_apis = ["capCreateCaptureWindow", "avicap32", "webcam"] if any(api.lower() in s.lower() for s in strings_list for api in webcam_apis): capabilities["webcam_capture"] = True file_grabber_markers = ["*.txt", "*.doc", "Desktop\\", "Documents\\", "FileGrabber"] if any(m.lower() in s.lower() for s in strings_list for m in file_grabber_markers): capabilities["file_grabber"] = True email_markers = ["Outlook", "Thunderbird", "Foxmail", "The Bat!"] if any(m.lower() in s.lower() for s in strings_list for m in email_markers): capabilities["email_client_theft"] = True ftp_markers = ["FileZilla", "WinSCP", "CoreFTP", "recentservers.xml"] if any(m.lower() in s.lower() for s in strings_list for m in ftp_markers): capabilities["ftp_client_theft"] = True vpn_markers = ["NordVPN", "OpenVPN", "ProtonVPN", "ovpn"] if any(m.lower() in s.lower() for s in strings_list for m in vpn_markers): capabilities["vpn_client_theft"] = True return capabilities def extract_configuration(data, strings_list, family) -> dict: """Extract embedded configuration based on identified family.""" config = { "campaign_id": None, "mutex": None, "install_path": None, "persistence": None, "c2_urls": [], } # Extract URLs url_pattern = re.compile(rb"https?://[\w\.\-:]+[/\w\.\-\?\&\=]*") urls = [u.decode(errors="ignore") for u in url_pattern.findall(data)] config["c2_urls"] = list(set(urls))[:20] # Extract mutex names mutex_candidates = [s for s in strings_list if re.match( r"(Global\\|Local\\|AsyncMutex|Mutex|mtx)[A-Za-z0-9_\-\{\}]+", s )] if mutex_candidates: config["mutex"] = mutex_candidates[0] # Extract persistence indicators persistence_markers = ["SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Run", "Startup", "schtasks", "TaskScheduler"] for marker in persistence_markers: if any(marker.lower() in s.lower() for s in strings_list): config["persistence"] = marker break # Family-specific config extraction if family == "agenttesla": # Look for the "|Sun|" delimited log format config["log_format"] = "pipe_delimited" if any("|Sun|" in s for s in strings_list) else "unknown" elif family == "vidar": # Vidar uses numbered endpoint paths numbered = [s for s in strings_list if re.match(r"/\d+$", s)] if numbered: config["numbered_endpoints"] = numbered elif family == "raccoon": # Raccoon uses RC4 with machineId if any("machineId" in s for s in strings_list): config["config_encryption"] = "RC4" return config def analyze_sample(sample_path, family="auto") -> dict: """Perform comprehensive infostealer analysis on a sample.""" data = Path(sample_path).read_bytes() strings_list = extract_strings(data) hashes = compute_hashes(sample_path) # Family identification if family == "auto": detected_family, confidence, all_scores = identify_family(data, strings_list) else: detected_family = family.lower() confidence = 1.0 all_scores = {} dotnet = is_dotnet_binary(sample_path) result = { "sample": str(sample_path), "hashes": hashes, "family": detected_family, "confidence": round(confidence, 2), "is_dotnet": dotnet, "identification_scores": all_scores, "browser_targeting": analyze_browser_targeting(strings_list), "crypto_targeting": analyze_crypto_targeting(strings_list), "capabilities": analyze_capabilities(strings_list), "exfiltration": analyze_exfiltration(data, strings_list), "configuration": extract_configuration(data, strings_list, detected_family), "mitre_attack": [ "T1555.003", # Credentials from Web Browsers "T1056.001", # Keylogging "T1115", # Clipboard Data "T1539", # Steal Web Session Cookie "T1041", # Exfiltration Over C2 Channel ], "timestamp": datetime.now().isoformat(), "tool": "infostealer_analyzer", } return result def main() -> None: parser = argparse.ArgumentParser( description="Analyze infostealer malware samples for family ID, " "credential harvesting, exfiltration channels, and config extraction" ) parser.add_argument("--input", "--sample", required=True, help="Path to malware sample") parser.add_argument( "--family", default="auto", choices=["auto", "agenttesla", "redline", "raccoon", "lumma", "vidar", "rhadamanthys"], help="Specify family or 'auto' for detection (default: auto)", ) parser.add_argument("--output", "-o", help="Output file path (JSON)") parser.add_argument( "--format", choices=["json", "text"], default="json", help="Output format (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"[*] Analyzing sample: {args.sample}") print(f"[*] Family hint: {args.family}") result = analyze_sample(sample_path, family=args.family) if args.format == "json": output = json.dumps(result, indent=2, default=str) else: output = f"=== Infostealer Analysis: {args.sample} ===\n" output += f"Family: {result['family']} (confidence: {result['confidence']})\n" output += f"Is .NET: {result['is_dotnet']}\n" output += f"SHA256: {result['hashes']['sha256']}\n\n" output += f"Targeted Browsers: {', '.join(result['browser_targeting']['targeted_browsers']) or 'None detected'}\n" output += f"Exfil Methods: {', '.join(result['exfiltration']['methods'])}\n" caps = [k for k, v in result['capabilities'].items() if v] output += f"Capabilities: {', '.join(caps) or 'None detected'}\n" output += f"C2 URLs: {len(result['configuration']['c2_urls'])} found\n" for url in result["configuration"]["c2_urls"][:10]: output += f" - {url}\n" if args.output: Path(args.output).write_text(output) print(f"[+] Report saved to {args.output}") else: print(output) if __name__ == "__main__": main()