Add 5 new cybersecurity skills: golden ticket detection, traffic baselining, sandbox evasion analysis, domain fronting hunting, SpiderFoot OSINT

skills/analyzing-malware-sandbox-evasion-techniques/scripts/agent.py

@@ -0,0 +1,205 @@
+#!/usr/bin/env python3
+"""Analyze malware sandbox evasion techniques from Cuckoo/AnyRun behavioral reports."""
+
+import json
+import argparse
+from datetime import datetime
+from collections import defaultdict
+
+TIMING_APIS = {
+    "GetTickCount", "GetTickCount64", "QueryPerformanceCounter",
+    "QueryPerformanceFrequency", "GetSystemTimeAsFileTime", "NtQuerySystemTime",
+    "timeGetTime", "GetLocalTime", "GetSystemTime",
+}
+
+SLEEP_APIS = {"Sleep", "SleepEx", "NtDelayExecution", "WaitForSingleObject"}
+
+VM_REGISTRY_KEYS = [
+    "HKLM\\SOFTWARE\\VMware", "HKLM\\SOFTWARE\\Oracle\\VirtualBox",
+    "HKLM\\HARDWARE\\ACPI\\DSDT\\VBOX", "HKLM\\SYSTEM\\CurrentControlSet\\Services\\VBoxGuest",
+    "HKLM\\SOFTWARE\\Microsoft\\Virtual Machine\\Guest\\Parameters",
+    "HKLM\\HARDWARE\\Description\\System\\SystemBiosVersion",
+]
+
+VM_PROCESSES = {
+    "vmtoolsd.exe", "vmwaretray.exe", "vboxservice.exe", "vboxtray.exe",
+    "qemu-ga.exe", "vmusrvc.exe", "prl_tools.exe", "xenservice.exe",
+    "windanr.exe", "vdagent.exe",
+}
+
+VM_MAC_PREFIXES = ["00:0C:29", "00:50:56", "08:00:27", "00:1C:42", "00:16:3E", "52:54:00"]
+
+USER_INTERACTION_APIS = {
+    "GetCursorPos", "GetAsyncKeyState", "GetForegroundWindow",
+    "GetLastInputInfo", "mouse_event", "keybd_event",
+}
+
+WMI_EVASION_QUERIES = [
+    "Win32_ComputerSystem", "Win32_BIOS", "Win32_DiskDrive",
+    "Win32_PhysicalMemory", "Win32_Processor",
+]
+
+
+def parse_cuckoo_report(filepath):
+    """Parse a Cuckoo Sandbox behavioral report JSON."""
+    with open(filepath) as f:
+        report = json.load(f)
+    behavior = report.get("behavior", {})
+    api_calls = []
+    for process in behavior.get("processes", []):
+        for call in process.get("calls", []):
+            api_calls.append({
+                "api": call.get("api", ""),
+                "category": call.get("category", ""),
+                "arguments": call.get("arguments", {}),
+                "return": call.get("return", ""),
+                "process_name": process.get("process_name", ""),
+                "pid": process.get("pid", 0),
+            })
+    return api_calls, report
+
+
+def detect_timing_checks(api_calls):
+    """Detect timing-based sandbox evasion via GetTickCount, QPC, etc."""
+    findings = []
+    timing_count = 0
+    for call in api_calls:
+        if call["api"] in TIMING_APIS:
+            timing_count += 1
+    if timing_count >= 3:
+        findings.append({
+            "technique": "Timing-Based Evasion",
+            "mitre_id": "T1497.003",
+            "api_count": timing_count,
+            "apis_used": list({c["api"] for c in api_calls if c["api"] in TIMING_APIS}),
+            "severity": "high",
+            "description": f"{timing_count} timing API calls detected; malware may be measuring execution time to detect sandbox acceleration",
+        })
+    return findings
+
+
+def detect_sleep_inflation(api_calls, min_sleep_ms=60000):
+    """Detect sleep calls with long durations used to evade sandbox time limits."""
+    findings = []
+    for call in api_calls:
+        if call["api"] not in SLEEP_APIS:
+            continue
+        ms = 0
+        args = call.get("arguments", {})
+        if isinstance(args, dict):
+            ms = int(args.get("Milliseconds", args.get("milliseconds", 0)))
+        elif isinstance(args, list):
+            for a in args:
+                if isinstance(a, dict) and a.get("name", "").lower() == "milliseconds":
+                    ms = int(a.get("value", 0))
+        if ms >= min_sleep_ms:
+            findings.append({
+                "technique": "Sleep Inflation",
+                "mitre_id": "T1497.003",
+                "api": call["api"],
+                "sleep_ms": ms,
+                "sleep_seconds": ms / 1000,
+                "process": call["process_name"],
+                "severity": "high",
+                "description": f"Sleep call of {ms / 1000:.0f}s detected; likely delaying execution to outlast sandbox analysis window",
+            })
+    return findings
+
+
+def detect_vm_artifact_checks(api_calls):
+    """Detect VM artifact queries (registry, processes, MAC addresses)."""
+    findings = []
+    for call in api_calls:
+        args_str = json.dumps(call.get("arguments", "")).lower()
+        for reg_key in VM_REGISTRY_KEYS:
+            if reg_key.lower() in args_str:
+                findings.append({
+                    "technique": "VM Registry Artifact Check",
+                    "mitre_id": "T1497.001",
+                    "registry_key": reg_key,
+                    "api": call["api"],
+                    "severity": "high",
+                })
+                break
+        for wmi_query in WMI_EVASION_QUERIES:
+            if wmi_query.lower() in args_str:
+                findings.append({
+                    "technique": "WMI Environment Fingerprinting",
+                    "mitre_id": "T1497.001",
+                    "wmi_class": wmi_query,
+                    "api": call["api"],
+                    "severity": "medium",
+                })
+                break
+    return findings
+
+
+def detect_user_interaction_checks(api_calls):
+    """Detect checks for user interaction (mouse, keyboard, foreground window)."""
+    interaction_apis = [c for c in api_calls if c["api"] in USER_INTERACTION_APIS]
+    if len(interaction_apis) >= 2:
+        return [{
+            "technique": "User Interaction Detection",
+            "mitre_id": "T1497.002",
+            "api_count": len(interaction_apis),
+            "apis_used": list({c["api"] for c in interaction_apis}),
+            "severity": "medium",
+            "description": "Malware checks for user input to determine if running in automated sandbox",
+        }]
+    return []
+
+
+def score_evasion_sophistication(all_findings):
+    """Score evasion sophistication based on technique diversity."""
+    technique_ids = {f["mitre_id"] for f in all_findings}
+    categories = {f["technique"].split()[0] for f in all_findings}
+    score = min(len(all_findings) * 10 + len(technique_ids) * 15 + len(categories) * 10, 100)
+    level = "low" if score < 30 else "medium" if score < 60 else "high"
+    return {"score": score, "level": level, "unique_techniques": len(technique_ids), "total_indicators": len(all_findings)}
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Sandbox Evasion Technique Analyzer")
+    parser.add_argument("--report", required=True, help="Path to Cuckoo/AnyRun behavioral report JSON")
+    parser.add_argument("--min-sleep-ms", type=int, default=60000, help="Minimum sleep duration to flag (ms)")
+    parser.add_argument("--output", default="evasion_analysis_report.json", help="Output report path")
+    args = parser.parse_args()
+
+    api_calls, raw_report = parse_cuckoo_report(args.report)
+    print(f"[+] Parsed {len(api_calls)} API calls from behavioral report")
+
+    timing = detect_timing_checks(api_calls)
+    sleep = detect_sleep_inflation(api_calls, args.min_sleep_ms)
+    vm_checks = detect_vm_artifact_checks(api_calls)
+    user_checks = detect_user_interaction_checks(api_calls)
+
+    all_findings = timing + sleep + vm_checks + user_checks
+    sophistication = score_evasion_sophistication(all_findings)
+
+    report = {
+        "analysis_time": datetime.utcnow().isoformat() + "Z",
+        "sample_sha256": raw_report.get("target", {}).get("file", {}).get("sha256", ""),
+        "total_api_calls": len(api_calls),
+        "evasion_findings": {
+            "timing_checks": timing,
+            "sleep_inflation": sleep,
+            "vm_artifact_checks": vm_checks,
+            "user_interaction_checks": user_checks,
+        },
+        "total_indicators": len(all_findings),
+        "sophistication": sophistication,
+        "mitre_techniques": ["T1497.001", "T1497.002", "T1497.003"],
+    }
+
+    with open(args.output, "w") as f:
+        json.dump(report, f, indent=2)
+    print(f"[+] Timing checks: {len(timing)}")
+    print(f"[+] Sleep inflation: {len(sleep)}")
+    print(f"[+] VM artifact checks: {len(vm_checks)}")
+    print(f"[+] User interaction checks: {len(user_checks)}")
+    print(f"[+] Evasion sophistication: {sophistication['level']} ({sophistication['score']}/100)")
+    print(f"[+] Report saved to {args.output}")
+
+
+if __name__ == "__main__":
+    main()