#!/usr/bin/env python3 """Cryptominer analyzer — identify mining software, extract configs, and analyze evasion.""" from __future__ import annotations import argparse 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 identify_miner(filepath: str) -> dict: """Identify the mining software and cryptocurrency.""" data = Path(filepath).read_bytes() miners = { "XMRig": { "markers": [b"xmrig", b"XMRig", b"randomx", b"RandomX", b"cryptonight", b"CryptoNight"], "crypto": "Monero (XMR)", "algorithm": "RandomX/CryptoNight", }, "T-Rex": { "markers": [b"t-rex", b"T-Rex", b"trex"], "crypto": "Various (ETH, RVN, etc.)", "algorithm": "Ethash/KawPow", }, "PhoenixMiner": { "markers": [b"PhoenixMiner", b"phoenixminer"], "crypto": "Ethereum (ETH)", "algorithm": "Ethash", }, "NBMiner": { "markers": [b"nbminer", b"NBMiner"], "crypto": "Various", "algorithm": "Ethash/various", }, } result = { "miner": "unknown", "cryptocurrency": "unknown", "algorithm": "unknown", "confidence": "low", "markers_found": [], } for miner, info in miners.items(): found = [m.decode() for m in info["markers"] if m in data] if found: result["miner"] = miner result["cryptocurrency"] = info["crypto"] result["algorithm"] = info["algorithm"] result["confidence"] = "high" if len(found) >= 2 else "medium" result["markers_found"] = found break # Check for generic mining indicators if result["miner"] == "unknown": generic = [b"stratum+tcp", b"stratum+ssl", b"mining.subscribe", b"mining.authorize"] found = [m.decode() for m in generic if m in data] if found: result["miner"] = "generic/custom" result["confidence"] = "medium" result["markers_found"] = found return result def extract_wallet_addresses(filepath: str) -> dict: """Extract cryptocurrency wallet addresses from the binary.""" data = Path(filepath).read_bytes() text = data.decode("utf-8", errors="ignore") wallets = { "monero": [], "bitcoin": [], "ethereum": [], "pool_urls": [], "worker_names": [], } # Monero addresses (95 chars, starts with 4) xmr_pattern = r'4[0-9AB][1-9A-HJ-NP-Za-km-z]{93}' wallets["monero"] = list(set(re.findall(xmr_pattern, text))) # Bitcoin addresses btc_pattern = r'(?:bc1[a-zA-HJ-NP-Z0-9]{25,39}|[13][a-km-zA-HJ-NP-Z1-9]{25,34})' wallets["bitcoin"] = list(set(re.findall(btc_pattern, text))) # Ethereum addresses eth_pattern = r'0x[0-9a-fA-F]{40}' wallets["ethereum"] = list(set(re.findall(eth_pattern, text))) # Pool URLs pool_pattern = r'stratum\+(?:tcp|ssl)://[^\s"\'<>]{5,100}' wallets["pool_urls"] = list(set(re.findall(pool_pattern, text))) # Known pool domains pool_domains = [ "supportxmr.com", "nanopool.org", "hashvault.pro", "moneroocean.stream", "unmineable.com", "minexmr.com", "2miners.com", "herominers.com", ] for domain in pool_domains: if domain.encode() in data: if f"pool://{domain}" not in str(wallets["pool_urls"]): wallets["pool_urls"].append(f"(detected: {domain})") return wallets def analyze_evasion_techniques(filepath: str) -> dict: """Analyze evasion and stealth techniques.""" data = Path(filepath).read_bytes() evasion = { "cpu_throttling": False, "idle_only_mining": False, "process_injection": False, "name_spoofing": False, "task_manager_detection": False, "indicators": [], } checks = { "cpu_throttling": [b"max-cpu-usage", b"cpu-priority", b"SetProcessAffinityMask", b"threads"], "idle_only_mining": [b"GetLastInputInfo", b"LASTINPUTINFO", b"idle"], "process_injection": [b"VirtualAllocEx", b"WriteProcessMemory", b"CreateRemoteThread", b"NtCreateThreadEx", b"process hollowing"], "name_spoofing": [b"svchost", b"csrss", b"dwm.exe", b"conhost"], "task_manager_detection": [b"taskmgr", b"TaskMgr", b"ProcessHacker", b"procexp"], } for technique, patterns in checks.items(): found = [p.decode() for p in patterns if p in data] if found: evasion[technique] = True evasion["indicators"].extend(found) return evasion def analyze_stratum_traffic(pcap_path: str) -> dict: """Analyze Stratum protocol in PCAP capture.""" # This would integrate with pyshark/dpkt in production return { "stratum_detected": False, "note": "Requires pyshark for PCAP analysis. Use tshark manually.", "manual_command": f'tshark -r {pcap_path} -Y "tcp.payload" -T fields -e tcp.payload | xxd -r -p | strings | grep "mining\\."', } def analyze_web_miner(filepath: str) -> dict: """Analyze JavaScript-based web miners.""" content = Path(filepath).read_text(errors="ignore") indicators = { "web_miner_detected": False, "libraries": [], "wasm_usage": False, "web_worker": False, "obfuscated": False, } known_libs = { "Coinhive": ["coinhive", "CoinHive.Anonymous", "CoinHive.Token"], "CryptoLoot": ["cryptoloot", "CryptoLoot.Anonymous"], "deepMiner": ["deepMiner", "deepMiner.Anonymous"], "WebMinePool": ["webminepool", "WMP.Anonymous"], } for lib, patterns in known_libs.items(): if any(p in content for p in patterns): indicators["web_miner_detected"] = True indicators["libraries"].append(lib) if "WebAssembly" in content or ".wasm" in content: indicators["wasm_usage"] = True if "new Worker" in content or "SharedWorker" in content: indicators["web_worker"] = True if "eval(" in content or "\\x" in content[:1000]: indicators["obfuscated"] = True return indicators def main() -> None: parser = argparse.ArgumentParser(description="Cryptominer Analyzer") parser.add_argument("--sample", help="Binary sample file path") parser.add_argument("--input", help="Input file (HTML for web miners)") parser.add_argument("--pcap", help="PCAP file for Stratum analysis") parser.add_argument( "--mode", choices=["identify", "config", "evasion", "injection", "stratum", "web-miner", "iocs", "full"], default="identify", help="Analysis mode", ) parser.add_argument("--output", default="miner_analysis.json", help="Output file path") parser.add_argument("--format", choices=["json", "csv", "markdown"], default="json") args = parser.parse_args() filepath = args.sample or args.input or args.pcap if not filepath: parser.error("One of --sample, --input, or --pcap is required") if not Path(filepath).exists(): print(f"[!] File not found: {filepath}", file=sys.stderr) sys.exit(1) print(f"[*] Cryptominer Analysis — mode: {args.mode}") results = {"source": filepath, "mode": args.mode} if args.sample: hashes = compute_hashes(args.sample) results["hashes"] = hashes print(f"[*] SHA-256: {hashes['sha256']}") if args.mode == "identify" and args.sample: results["identification"] = identify_miner(args.sample) elif args.mode == "config" and args.sample: results["identification"] = identify_miner(args.sample) results["wallets"] = extract_wallet_addresses(args.sample) elif args.mode == "evasion" and args.sample: results["evasion"] = analyze_evasion_techniques(args.sample) elif args.mode == "stratum" and args.pcap: results["stratum"] = analyze_stratum_traffic(args.pcap) elif args.mode == "web-miner" and args.input: results["web_miner"] = analyze_web_miner(args.input) elif args.mode == "full" and args.sample: results["identification"] = identify_miner(args.sample) results["wallets"] = extract_wallet_addresses(args.sample) results["evasion"] = analyze_evasion_techniques(args.sample) elif args.mode == "iocs" and args.sample: results["identification"] = identify_miner(args.sample) results["wallets"] = extract_wallet_addresses(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()