#!/usr/bin/env python3 """Classify observed malware behaviors into tactical categories and map them to MITRE ATT&CK techniques. Reads behavioral logs (JSON) from dynamic analysis and produces a structured classification covering persistence, C2, lateral movement, evasion, data theft, and other tactical categories. """ from __future__ import annotations import argparse import json import sys from pathlib import Path from typing import Any # Behavioral category definitions with associated ATT&CK tactics and indicators BEHAVIOR_CATEGORIES: dict[str, dict[str, Any]] = { "persistence": { "tactic": "TA0003", "tactic_name": "Persistence", "indicators": { "registry_run_key": { "technique": "T1547.001", "name": "Registry Run Keys / Startup Folder", "patterns": [ r"HKLM\Software\Microsoft\Windows\CurrentVersion\Run", r"HKCU\Software\Microsoft\Windows\CurrentVersion\Run", r"HKLM\Software\Microsoft\Windows\CurrentVersion\RunOnce", ], }, "scheduled_task": { "technique": "T1053.005", "name": "Scheduled Task", "patterns": ["schtasks", "ITaskService", "Schedule.Service"], }, "service_creation": { "technique": "T1543.003", "name": "Windows Service", "patterns": ["CreateService", "sc create", "New-Service"], }, "startup_folder": { "technique": "T1547.001", "name": "Startup Folder", "patterns": ["Startup", "Start Menu\\Programs\\Startup"], }, }, }, "c2": { "tactic": "TA0011", "tactic_name": "Command and Control", "indicators": { "http_c2": { "technique": "T1071.001", "name": "Web Protocols", "patterns": ["http://", "https://", "POST /", "GET /"], }, "dns_c2": { "technique": "T1071.004", "name": "DNS", "patterns": ["dns query", "TXT record", "nslookup"], }, "encrypted_channel": { "technique": "T1573", "name": "Encrypted Channel", "patterns": ["TLS", "SSL", "certificate", "encrypt"], }, }, }, "defense_evasion": { "tactic": "TA0005", "tactic_name": "Defense Evasion", "indicators": { "process_injection": { "technique": "T1055", "name": "Process Injection", "patterns": [ "CreateRemoteThread", "NtWriteVirtualMemory", "WriteProcessMemory", "VirtualAllocEx", ], }, "obfuscation": { "technique": "T1027", "name": "Obfuscated Files or Information", "patterns": ["base64", "xor", "encode", "decrypt", "deobfuscate"], }, "masquerading": { "technique": "T1036", "name": "Masquerading", "patterns": ["rename", "svchost", "explorer.exe"], }, }, }, "discovery": { "tactic": "TA0007", "tactic_name": "Discovery", "indicators": { "system_info": { "technique": "T1082", "name": "System Information Discovery", "patterns": [ "GetSystemInfo", "systeminfo", "hostname", "GetComputerName", ], }, "process_discovery": { "technique": "T1057", "name": "Process Discovery", "patterns": [ "CreateToolhelp32Snapshot", "Process32First", "tasklist", "EnumProcesses", ], }, "network_discovery": { "technique": "T1016", "name": "System Network Configuration Discovery", "patterns": ["ipconfig", "ifconfig", "netstat", "arp"], }, }, }, "lateral_movement": { "tactic": "TA0008", "tactic_name": "Lateral Movement", "indicators": { "remote_services": { "technique": "T1021", "name": "Remote Services", "patterns": ["SMB", "RDP", "WinRM", "SSH", "port 445", "port 3389"], }, "psexec": { "technique": "T1021.002", "name": "SMB/Windows Admin Shares", "patterns": ["PsExec", "ADMIN$", "IPC$", "C$"], }, }, }, "collection": { "tactic": "TA0009", "tactic_name": "Collection", "indicators": { "keylogging": { "technique": "T1056.001", "name": "Keylogging", "patterns": [ "SetWindowsHookEx", "GetAsyncKeyState", "keylog", "GetKeyState", ], }, "screen_capture": { "technique": "T1113", "name": "Screen Capture", "patterns": ["BitBlt", "screenshot", "PrintWindow", "GetDC"], }, "clipboard": { "technique": "T1115", "name": "Clipboard Data", "patterns": ["GetClipboardData", "OpenClipboard", "clipboard"], }, }, }, "exfiltration": { "tactic": "TA0010", "tactic_name": "Exfiltration", "indicators": { "exfil_c2": { "technique": "T1041", "name": "Exfiltration Over C2 Channel", "patterns": ["upload", "exfil", "send_data", "POST"], }, "exfil_web": { "technique": "T1567", "name": "Exfiltration Over Web Service", "patterns": ["pastebin", "discord", "telegram", "dropbox", "mega.nz"], }, }, }, "impact": { "tactic": "TA0040", "tactic_name": "Impact", "indicators": { "data_encrypted": { "technique": "T1486", "name": "Data Encrypted for Impact", "patterns": [ "CryptEncrypt", "ransom", ".locked", ".encrypted", "AES", "RSA", ], }, "data_destruction": { "technique": "T1485", "name": "Data Destruction", "patterns": ["wipe", "shred", "DeleteFile", "format"], }, }, }, "execution": { "tactic": "TA0002", "tactic_name": "Execution", "indicators": { "powershell": { "technique": "T1059.001", "name": "PowerShell", "patterns": ["powershell", "pwsh", "Invoke-Expression", "IEX"], }, "cmd": { "technique": "T1059.003", "name": "Windows Command Shell", "patterns": ["cmd.exe", "cmd /c", "command.com"], }, "scripting": { "technique": "T1059.005", "name": "Visual Basic", "patterns": ["wscript", "cscript", "VBScript", "JScript"], }, }, }, "credential_access": { "tactic": "TA0006", "tactic_name": "Credential Access", "indicators": { "lsass_dump": { "technique": "T1003.001", "name": "LSASS Memory", "patterns": ["lsass", "MiniDump", "sekurlsa", "mimikatz"], }, "credential_files": { "technique": "T1552.001", "name": "Credentials in Files", "patterns": ["password", "credential", "login_data", "web data"], }, }, }, } SEVERITY_THRESHOLDS = {"low": 1, "medium": 3, "high": 5} def load_events(input_path: Path) -> list[dict[str, Any]]: """Load behavioral event logs from a JSON file.""" with open(input_path, "r", encoding="utf-8") as f: data = json.load(f) if isinstance(data, list): return data if isinstance(data, dict) and "events" in data: return data["events"] return [data] def classify_event(event: dict[str, Any]) -> list[dict[str, Any]]: """Classify a single behavioral event against known indicator patterns.""" matches: list[dict[str, Any]] = [] event_str = json.dumps(event).lower() for category, cat_info in BEHAVIOR_CATEGORIES.items(): for indicator_key, indicator in cat_info["indicators"].items(): for pattern in indicator["patterns"]: if pattern.lower() in event_str: matches.append( { "category": category, "tactic": cat_info["tactic"], "tactic_name": cat_info["tactic_name"], "technique": indicator["technique"], "technique_name": indicator["name"], "indicator": indicator_key, "matched_pattern": pattern, "event_summary": _summarize_event(event), } ) break # One match per indicator per event is sufficient return matches def _summarize_event(event: dict[str, Any]) -> str: """Create a brief summary of a behavioral event.""" if "description" in event: return str(event["description"])[:200] if "api" in event: return f"API call: {event['api']}" if "type" in event: return f"Event type: {event['type']}" return json.dumps(event)[:200] def compute_severity(match_count: int) -> str: """Determine severity based on the number of indicator matches.""" if match_count >= SEVERITY_THRESHOLDS["high"]: return "high" if match_count >= SEVERITY_THRESHOLDS["medium"]: return "medium" return "low" def classify_behaviors( events: list[dict[str, Any]], category_filter: str | None = None ) -> dict[str, Any]: """Classify all behavioral events and produce a structured result.""" all_matches: list[dict[str, Any]] = [] for event in events: all_matches.extend(classify_event(event)) if category_filter: all_matches = [m for m in all_matches if m["category"] == category_filter] # Group by category by_category: dict[str, list[dict[str, Any]]] = {} for match in all_matches: cat = match["category"] if cat not in by_category: by_category[cat] = [] by_category[cat].append(match) # Build classification result classification: dict[str, Any] = {} for cat, matches in by_category.items(): techniques = {} for m in matches: tid = m["technique"] if tid not in techniques: techniques[tid] = { "technique_id": tid, "technique_name": m["technique_name"], "tactic": m["tactic_name"], "confidence": "medium", "evidence": [], } techniques[tid]["evidence"].append( {"pattern": m["matched_pattern"], "event": m["event_summary"]} ) # Upgrade confidence for techniques with multiple evidence items for tid, info in techniques.items(): if len(info["evidence"]) >= 3: info["confidence"] = "high" elif len(info["evidence"]) == 1: info["confidence"] = "low" classification[cat] = { "tactic": BEHAVIOR_CATEGORIES[cat]["tactic"], "tactic_name": BEHAVIOR_CATEGORIES[cat]["tactic_name"], "severity": compute_severity(len(matches)), "match_count": len(matches), "techniques": list(techniques.values()), } return classification def format_attack_matrix(classification: dict[str, Any]) -> dict[str, Any]: """Format classification as an ATT&CK matrix mapping.""" matrix: list[dict[str, Any]] = [] for cat, info in classification.items(): for tech in info["techniques"]: matrix.append( { "tactic": info["tactic"], "tactic_name": info["tactic_name"], "technique_id": tech["technique_id"], "technique_name": tech["technique_name"], "confidence": tech["confidence"], "evidence_count": len(tech["evidence"]), } ) return {"attack_mapping": matrix, "total_techniques": len(matrix)} def generate_report( events: list[dict[str, Any]], classification: dict[str, Any] ) -> dict[str, Any]: """Generate a comprehensive behavioral profile report.""" # Determine likely malware type based on categories present malware_type = "unknown" categories = set(classification.keys()) if "impact" in categories and any( t["technique_id"] == "T1486" for t in classification.get("impact", {}).get("techniques", []) ): malware_type = "ransomware" elif "c2" in categories and "collection" in categories: malware_type = "RAT/spyware" elif "c2" in categories and "credential_access" in categories: malware_type = "stealer" elif "lateral_movement" in categories: malware_type = "worm" elif "c2" in categories: malware_type = "trojan" total_techniques = sum( len(info["techniques"]) for info in classification.values() ) high_severity = [ cat for cat, info in classification.items() if info["severity"] == "high" ] return { "malware_type": malware_type, "total_events_analyzed": len(events), "categories_detected": list(classification.keys()), "total_techniques_mapped": total_techniques, "high_severity_categories": high_severity, "behavioral_severity": ( "high" if high_severity else ("medium" if total_techniques > 3 else "low") ), "classification": classification, "attack_mapping": format_attack_matrix(classification)["attack_mapping"], "recommended_actions": _generate_recommendations(classification), } def _generate_recommendations(classification: dict[str, Any]) -> list[str]: """Generate response recommendations based on classified behaviors.""" recommendations: list[str] = [] if "persistence" in classification: recommendations.append( "Remove persistence mechanisms (registry keys, scheduled tasks, services)" ) if "c2" in classification: recommendations.append( "Block C2 communication channels and add network IOCs to detection" ) if "lateral_movement" in classification: recommendations.append( "Isolate affected hosts and scan network for lateral movement indicators" ) if "credential_access" in classification: recommendations.append( "Reset credentials for compromised accounts and enable MFA" ) if "exfiltration" in classification: recommendations.append( "Assess data exposure and initiate incident response for data breach" ) if "impact" in classification: recommendations.append( "Restore affected systems from clean backups and preserve forensic evidence" ) if not recommendations: recommendations.append("Continue monitoring for additional malicious activity") return recommendations def analyze(input_path: Path, output_format: str = "json", category_filter: str | None = None) -> dict[str, Any]: """Analyze behavioral logs and return classified results.""" events = load_events(input_path) classification = classify_behaviors(events, category_filter) if output_format == "attack-matrix": return format_attack_matrix(classification) if output_format == "report": return generate_report(events, classification) return {"classification": classification, "event_count": len(events)} def main() -> None: """Entry point for the behavior classifier CLI.""" parser = argparse.ArgumentParser( description="Classify malware behaviors and map to MITRE ATT&CK techniques." ) parser.add_argument( "--input", type=Path, required=True, help="Path to behavioral events JSON file" ) parser.add_argument( "--output", type=Path, help="Path to output file (stdout if omitted)" ) parser.add_argument( "--format", default="json", choices=["json", "attack-matrix", "report", "text"], help="Output format (default: json)", ) parser.add_argument( "--filter", choices=[ "persistence", "c2", "defense_evasion", "discovery", "lateral_movement", "collection", "exfiltration", "impact", "execution", "credential_access", ], help="Filter results to a specific behavioral category", ) args = parser.parse_args() if not args.input.exists(): print(f"Error: Input file not found: {args.input}", file=sys.stderr) sys.exit(1) result = analyze(args.input, output_format=args.format, category_filter=args.filter) if args.format == "text": output_text = _format_text(result) else: output_text = json.dumps(result, indent=2) if args.output: args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(output_text, encoding="utf-8") print(f"Results written to {args.output}", file=sys.stderr) else: print(output_text) def _format_text(result: dict[str, Any]) -> str: """Format results as human-readable text.""" lines: list[str] = ["=== Behavioral Classification Report ===", ""] classification = result.get("classification", result) for category, info in classification.items(): if isinstance(info, dict) and "techniques" in info: lines.append(f"[{info.get('severity', 'unknown').upper()}] {category}") for tech in info["techniques"]: lines.append( f" - {tech['technique_id']} {tech['technique_name']} " f"(confidence: {tech['confidence']})" ) lines.append("") return "\n".join(lines) if __name__ == "__main__": main()