#!/usr/bin/env python3 """Analyze .NET (managed-code) malware samples. Detects .NET assemblies, identifies obfuscators, parses metadata tables, enumerates embedded resources, and maps capabilities via P/Invoke and namespace analysis. Produces a structured JSON or text report. """ from __future__ import annotations import argparse import hashlib import json import math import re import struct import sys from datetime import datetime, timezone from pathlib import Path from typing import Any try: import pefile HAS_PEFILE = True except ImportError: HAS_PEFILE = False try: import dnfile HAS_DNFILE = True except ImportError: HAS_DNFILE = False # --------------------------------------------------------------------------- # Hashing helpers # --------------------------------------------------------------------------- def compute_hashes(data: bytes) -> dict[str, str]: """Compute MD5, SHA-1, and SHA-256 hashes of raw bytes.""" return { "md5": hashlib.md5(data).hexdigest(), "sha1": hashlib.sha1(data).hexdigest(), "sha256": hashlib.sha256(data).hexdigest(), } def _entropy(data: bytes) -> float: """Calculate Shannon entropy of a byte sequence.""" if not data: return 0.0 freq = [0] * 256 for b in data: freq[b] += 1 length = len(data) return -sum( (c / length) * math.log2(c / length) for c in freq if c > 0 ) # --------------------------------------------------------------------------- # .NET detection # --------------------------------------------------------------------------- _MSCOREE = b"mscoree.dll" _CLI_HEADER_DIR_INDEX = 14 # IMAGE_DIRECTORY_ENTRY_COM_DESCRIPTOR def detect_dotnet(file_path: Path) -> dict[str, Any]: """Determine whether the file is a .NET assembly and extract CLR info.""" data = file_path.read_bytes() result: dict[str, Any] = { "is_dotnet": False, "clr_version": None, "flags": [], "entry_point_token": None, "is_mixed_mode": False, } # Quick signature check for mscoree.dll import if _MSCOREE not in data.lower(): return result if HAS_PEFILE: try: pe = pefile.PE(data=data) except pefile.PEFormatError: return result # Check for COM descriptor directory if not hasattr(pe, "OPTIONAL_HEADER"): return result if len(pe.OPTIONAL_HEADER.DATA_DIRECTORY) <= _CLI_HEADER_DIR_INDEX: return result cli_dir = pe.OPTIONAL_HEADER.DATA_DIRECTORY[_CLI_HEADER_DIR_INDEX] if cli_dir.VirtualAddress == 0: return result result["is_dotnet"] = True # Parse CLI header fields (72 bytes starting at RVA) try: cli_offset = pe.get_offset_from_rva(cli_dir.VirtualAddress) cli_data = data[cli_offset:cli_offset + 72] if len(cli_data) >= 24: _cb, _major, _minor, _meta_rva, _meta_size, flags = struct.unpack_from( " dict[str, Any]: """Detect obfuscator by scanning for known byte signatures.""" data = file_path.read_bytes() best_match: dict[str, Any] = {"name": "Unknown", "confidence": 0.0, "indicators": []} for sig in _OBFUSCATOR_SIGNATURES: indicators: list[str] = [] for pat in sig["patterns"]: if pat in data: indicators.append(f"Found signature: {pat.decode('ascii', errors='replace')}") if indicators: confidence = sig["weight"] * min(len(indicators) / len(sig["patterns"]), 1.0) if confidence > best_match["confidence"]: best_match = { "name": sig["name"], "confidence": round(confidence, 2), "indicators": indicators, } return best_match # --------------------------------------------------------------------------- # Metadata extraction (requires dnfile) # --------------------------------------------------------------------------- def extract_metadata(file_path: Path) -> dict[str, Any]: """Parse .NET metadata tables and return structural information.""" result: dict[str, Any] = { "assembly_name": None, "assembly_version": None, "mvid": None, "referenced_assemblies": [], "type_count": 0, "method_count": 0, "pinvoke_imports": [], "resources": [], } if not HAS_DNFILE: result["_note"] = "Install dnfile for full metadata extraction" return result try: dn = dnfile.dnPE(str(file_path)) except Exception: result["_note"] = "Failed to parse .NET metadata" return result # Assembly name and version if hasattr(dn.net, "mdtables") and dn.net.mdtables.Assembly: for row in dn.net.mdtables.Assembly: result["assembly_name"] = str(getattr(row, "Name", "")) ver_parts = [ getattr(row, "MajorVersion", 0), getattr(row, "MinorVersion", 0), getattr(row, "BuildNumber", 0), getattr(row, "RevisionNumber", 0), ] result["assembly_version"] = ".".join(str(v) for v in ver_parts) # MVID if hasattr(dn.net, "mdtables") and dn.net.mdtables.Module: for row in dn.net.mdtables.Module: mvid = getattr(row, "Mvid", None) if mvid: result["mvid"] = str(mvid) # Referenced assemblies if hasattr(dn.net, "mdtables") and dn.net.mdtables.AssemblyRef: for row in dn.net.mdtables.AssemblyRef: name = str(getattr(row, "Name", "")) ver_parts = [ getattr(row, "MajorVersion", 0), getattr(row, "MinorVersion", 0), getattr(row, "BuildNumber", 0), getattr(row, "RevisionNumber", 0), ] version = ".".join(str(v) for v in ver_parts) result["referenced_assemblies"].append(f"{name}, Version={version}") # Type and method counts if hasattr(dn.net, "mdtables"): if dn.net.mdtables.TypeDef: result["type_count"] = len(dn.net.mdtables.TypeDef) if dn.net.mdtables.MethodDef: result["method_count"] = len(dn.net.mdtables.MethodDef) # P/Invoke imports if hasattr(dn.net, "mdtables") and dn.net.mdtables.ImplMap: for row in dn.net.mdtables.ImplMap: import_name = str(getattr(row, "ImportName", "")) scope = getattr(row, "ImportScope", None) module_name = str(getattr(scope.row, "Name", "")) if scope and hasattr(scope, "row") else "" result["pinvoke_imports"].append({ "method": import_name, "module": module_name, }) # Embedded resources if hasattr(dn.net, "resources"): for res in dn.net.resources: name = str(getattr(res, "name", "")) size = getattr(res, "size", 0) entry: dict[str, Any] = {"name": name, "size": size} # Compute entropy if resource data is accessible res_data = getattr(res, "data", None) if res_data and isinstance(res_data, bytes): ent = _entropy(res_data) entry["entropy"] = round(ent, 2) entry["likely_encrypted"] = ent > 7.0 result["resources"].append(entry) return result # --------------------------------------------------------------------------- # Capability detection # --------------------------------------------------------------------------- _CAPABILITY_PATTERNS: dict[str, list[bytes]] = { "process_injection": [ b"VirtualAllocEx", b"WriteProcessMemory", b"CreateRemoteThread", b"NtCreateSection", b"NtMapViewOfSection", b"QueueUserAPC", ], "credential_theft": [ b"ProtectedData", b"Unprotect", b"CredentialCache", b"Login Data", b"logins.json", b"signons.sqlite", ], "keylogging": [ b"SetWindowsHookEx", b"GetAsyncKeyState", b"GetKeyState", b"LowLevelKeyboardProc", ], "screen_capture": [ b"CopyFromScreen", b"Screen.PrimaryScreen", b"PrintWindow", b"BitBlt", ], "persistence": [ b"CurrentVersion\\Run", b"CurrentVersion\\RunOnce", b"Startup", b"TaskScheduler", b"ServiceInstaller", b"SchTasks", ], "network_communication": [ b"HttpClient", b"WebClient", b"TcpClient", b"SmtpClient", b"FtpWebRequest", b"WebSocket", ], "anti_analysis": [ b"Debugger.IsAttached", b"IsDebuggerPresent", b"CheckRemoteDebuggerPresent", b"Win32_ComputerSystem", b"SandboxieControl", b"vmware", b"VirtualBox", ], "clipboard_theft": [ b"Clipboard.GetText", b"GetClipboardData", b"SetClipboardViewer", ], "crypto_operations": [ b"RijndaelManaged", b"Aes.Create", b"AesCryptoServiceProvider", b"RSACryptoServiceProvider", b"Rfc2898DeriveBytes", b"DESCryptoServiceProvider", ], } def detect_capabilities(file_path: Path) -> dict[str, bool]: """Scan binary for capability indicators based on string patterns.""" data = file_path.read_bytes() results: dict[str, bool] = {} for capability, patterns in _CAPABILITY_PATTERNS.items(): results[capability] = any(pat in data for pat in patterns) return results # --------------------------------------------------------------------------- # Strings of interest # --------------------------------------------------------------------------- _INTERESTING_PATTERNS = [ re.compile(rb"https?://[^\x00-\x1f\x7f-\xff\"'<>\s]{5,200}", re.ASCII), re.compile(rb"\b(?:\d{1,3}\.){3}\d{1,3}(?::\d{1,5})?\b"), re.compile(rb"HKCU\\[^\x00]{10,200}"), re.compile(rb"HKLM\\[^\x00]{10,200}"), re.compile(rb"[a-zA-Z0-9+/]{40,}={0,2}"), # Base64 blobs ] def extract_interesting_strings(file_path: Path) -> list[str]: """Extract URLs, IPs, registry paths, and Base64 blobs.""" data = file_path.read_bytes() found: list[str] = [] seen: set[str] = set() for pattern in _INTERESTING_PATTERNS: for match in pattern.finditer(data): s = match.group().decode("ascii", errors="replace") if s not in seen: seen.add(s) found.append(s) return found[:200] # Limit to prevent huge output # --------------------------------------------------------------------------- # MITRE ATT&CK mapping # --------------------------------------------------------------------------- _MITRE_MAP: dict[str, dict[str, str]] = { "process_injection": {"technique": "T1055", "name": "Process Injection"}, "credential_theft": {"technique": "T1555.003", "name": "Credentials from Web Browsers"}, "keylogging": {"technique": "T1056.001", "name": "Keylogging"}, "screen_capture": {"technique": "T1113", "name": "Screen Capture"}, "persistence": {"technique": "T1547.001", "name": "Registry Run Keys / Startup Folder"}, "clipboard_theft": {"technique": "T1115", "name": "Clipboard Data"}, "anti_analysis": {"technique": "T1497", "name": "Virtualization/Sandbox Evasion"}, } def map_mitre_attack(capabilities: dict[str, bool]) -> list[dict[str, str]]: """Map detected capabilities to MITRE ATT&CK technique IDs.""" techniques: list[dict[str, str]] = [] for cap, detected in capabilities.items(): if detected and cap in _MITRE_MAP: techniques.append(_MITRE_MAP[cap]) return techniques # --------------------------------------------------------------------------- # Main analysis orchestrator # --------------------------------------------------------------------------- def analyze(input_path: Path, output_format: str = "json") -> dict[str, Any]: """Run all analysis stages and return a combined result dictionary.""" data = input_path.read_bytes() hashes = compute_hashes(data) dotnet_info = detect_dotnet(input_path) obfuscator = identify_obfuscator(input_path) metadata = extract_metadata(input_path) capabilities = detect_capabilities(input_path) strings = extract_interesting_strings(input_path) mitre = map_mitre_attack(capabilities) return { "sample": str(input_path), "hashes": hashes, "is_dotnet": dotnet_info.pop("is_dotnet"), "dotnet_info": dotnet_info, "obfuscator": obfuscator, "metadata": metadata, "capabilities": capabilities, "strings_of_interest": strings, "mitre_attack": mitre, "timestamp": datetime.now(timezone.utc).isoformat(), } # --------------------------------------------------------------------------- # Text formatter # --------------------------------------------------------------------------- def format_text(report: dict[str, Any]) -> str: """Render the analysis report as human-readable text.""" lines: list[str] = [] lines.append(f".NET Malware Analysis Report: {report['sample']}") lines.append("=" * 60) lines.append(f"\n.NET Assembly: {'Yes' if report['is_dotnet'] else 'No'}") info = report.get("dotnet_info", {}) lines.append(f"CLR Version: {info.get('clr_version', 'unknown')}") lines.append(f"Flags: {', '.join(info.get('flags', []))}") lines.append(f"Entry Point: {info.get('entry_point_token', 'unknown')}") lines.append(f"Mixed-Mode: {info.get('is_mixed_mode', False)}") obf = report.get("obfuscator", {}) lines.append(f"\nObfuscator: {obf.get('name', 'Unknown')} " f"(confidence: {obf.get('confidence', 0):.0%})") for ind in obf.get("indicators", []): lines.append(f" - {ind}") meta = report.get("metadata", {}) lines.append(f"\nAssembly: {meta.get('assembly_name', '?')} " f"v{meta.get('assembly_version', '?')}") lines.append(f"Types: {meta.get('type_count', '?')} " f"Methods: {meta.get('method_count', '?')}") pinvoke = meta.get("pinvoke_imports", []) if pinvoke: lines.append(f"\nP/Invoke Imports ({len(pinvoke)}):") for p in pinvoke[:30]: lines.append(f" {p.get('module', '?')}!{p.get('method', '?')}") resources = meta.get("resources", []) if resources: lines.append(f"\nEmbedded Resources ({len(resources)}):") for r in resources: enc = " [ENCRYPTED?]" if r.get("likely_encrypted") else "" lines.append(f" {r.get('name', '?')} ({r.get('size', '?')} bytes, " f"entropy={r.get('entropy', '?')}){enc}") caps = report.get("capabilities", {}) active = [k for k, v in caps.items() if v] if active: lines.append(f"\nCapabilities ({len(active)}):") for c in active: lines.append(f" + {c.replace('_', ' ')}") mitre = report.get("mitre_attack", []) if mitre: lines.append(f"\nMITRE ATT&CK Mappings ({len(mitre)}):") for m in mitre: lines.append(f" {m['technique']}: {m['name']}") strings = report.get("strings_of_interest", []) if strings: lines.append(f"\nStrings of Interest ({len(strings)}):") for s in strings[:30]: lines.append(f" {s}") return "\n".join(lines) # --------------------------------------------------------------------------- # CLI entry point # --------------------------------------------------------------------------- def main() -> None: parser = argparse.ArgumentParser( description="Analyze .NET malware — detect obfuscators, parse metadata, " "map capabilities, and extract IOCs." ) parser.add_argument( "--input", type=Path, required=True, help="Path to the .NET PE file to analyze", ) parser.add_argument( "--output", type=Path, default=None, help="Path to write the report (stdout if omitted)", ) parser.add_argument( "--format", default="json", choices=["json", "text"], help="Output format (default: json)", ) args = parser.parse_args() if not args.input.exists(): print(f"Error: file not found: {args.input}", file=sys.stderr) sys.exit(1) report = analyze(args.input, args.format) if args.format == "text": output = format_text(report) else: output = json.dumps(report, indent=2, default=str) if args.output: args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(output, encoding="utf-8") print(f"[+] Report saved to {args.output}", file=sys.stderr) else: print(output) if __name__ == "__main__": main()