threat hunting

Detecting WMI Persistence

Detect WMI event subscription persistence by analyzing Sysmon Event IDs 19, 20, and 21 for malicious EventFilter, EventConsumer, and FilterToConsumerBinding creation.

dfirmitre-attackpersistencesysmont1546.003threat-huntingwindowswmi
Install this skill
npx skills add mukul975/Anthropic-Cybersecurity-Skills
Framework mappings

When to Use

  • When hunting for WMI event subscription persistence (MITRE ATT&CK T1546.003)
  • After detecting suspicious WMI activity in endpoint telemetry
  • During incident response to identify attacker persistence mechanisms
  • When Sysmon alerts trigger on Event IDs 19, 20, or 21
  • During purple team exercises testing WMI-based persistence

Prerequisites

  • Sysmon v6.1+ deployed with WMI event logging enabled (Event IDs 19, 20, 21)
  • Windows Security Event Log forwarding configured
  • SIEM with Sysmon data ingested (Splunk, Elastic, Sentinel)
  • PowerShell access for WMI enumeration on endpoints
  • Sysinternals Autoruns for manual WMI subscription review

Workflow

  1. Collect Telemetry: Parse Sysmon Event IDs 19 (WmiEventFilter), 20 (WmiEventConsumer), 21 (WmiEventConsumerToFilter).
  2. Identify Suspicious Consumers: Flag CommandLineEventConsumer and ActiveScriptEventConsumer types executing code.
  3. Analyze Event Filters: Examine WQL queries in EventFilters for process start triggers or timer-based execution.
  4. Correlate Bindings: Match FilterToConsumerBindings linking suspicious filters to consumers.
  5. Check Persistence Locations: Query WMI namespaces root\subscription and root\default for active subscriptions.
  6. Validate Findings: Cross-reference with known-good WMI subscriptions (SCCM, AV products).
  7. Document and Remediate: Remove malicious subscriptions and update detection rules.

Key Concepts

Concept Description
Sysmon Event 19 WmiEventFilter creation detected
Sysmon Event 20 WmiEventConsumer creation detected
Sysmon Event 21 WmiEventConsumerToFilter binding detected
T1546.003 Event Triggered Execution: WMI Event Subscription
CommandLineEventConsumer Executes system commands when filter triggers
ActiveScriptEventConsumer Runs VBScript/JScript when filter triggers

Tools & Systems

Tool Purpose
Sysmon Windows event monitoring for WMI activity
WMI Explorer GUI tool for browsing WMI namespaces
Autoruns Sysinternals tool listing persistence mechanisms
PowerShell Get-WMIObject Enumerate WMI event subscriptions
Splunk SIEM analysis of Sysmon WMI events
Velociraptor Endpoint WMI artifact collection

Output Format

Hunt ID: TH-WMI-[DATE]-[SEQ]
Technique: T1546.003
Host: [Hostname]
Event Type: [EventFilter|EventConsumer|Binding]
Consumer Type: [CommandLine|ActiveScript]
WQL Query: [Filter query text]
Command: [Executed command or script]
Risk Level: [Critical/High/Medium/Low]
Recommended Action: [Remove subscription, investigate lateral movement]
Source materials

References and resources

Everything below is rendered for inspection. Script files are read-only and never run.

References 1

api-reference.md2.9 KB

WMI Persistence Detection Reference

Sysmon Event IDs

Event ID Type Description
19 WmiEventFilter Logs WMI EventFilter creation with WQL query
20 WmiEventConsumer Logs WMI EventConsumer creation (command/script)
21 WmiEventConsumerToFilter Logs binding of EventFilter to EventConsumer

Sysmon Configuration

Enable WMI event logging in sysmonconfig.xml:

<RuleGroup groupRelation="or">
  <WmiEvent onmatch="include">
    <Operation condition="is">Created</Operation>
  </WmiEvent>
</RuleGroup>

Install: sysmon64.exe -accepteula -i sysmonconfig.xml

PowerShell WMI Enumeration

# List all EventFilters
Get-WmiObject -Namespace root\subscription -Class __EventFilter
 
# List all EventConsumers
Get-WmiObject -Namespace root\subscription -Class __EventConsumer
 
# List all Bindings
Get-WmiObject -Namespace root\subscription -Class __FilterToConsumerBinding
 
# Remove specific subscription
Get-WmiObject -Namespace root\subscription -Class __EventFilter -Filter "Name='MalFilter'" | Remove-WmiObject
Get-WmiObject -Namespace root\subscription -Class CommandLineEventConsumer -Filter "Name='MalConsumer'" | Remove-WmiObject
Get-WmiObject -Namespace root\subscription -Class __FilterToConsumerBinding | Where-Object {$_.Filter -like '*MalFilter*'} | Remove-WmiObject

Suspicious Consumer Types

Consumer Class Risk Description
CommandLineEventConsumer Critical Executes arbitrary system commands
ActiveScriptEventConsumer Critical Runs embedded VBScript or JScript
LogFileEventConsumer Low Writes to log file
NTEventLogEventConsumer Low Creates Windows event log entry
SMTPEventConsumer Medium Sends email notification

Splunk Detection Query

index=sysmon EventCode IN (19, 20, 21)
| eval event_type=case(EventCode=19, "EventFilter", EventCode=20, "EventConsumer", EventCode=21, "Binding")
| where Consumer_Type IN ("CommandLineEventConsumer", "ActiveScriptEventConsumer")
| stats count by Computer, event_type, Consumer_Type, Destination, User
| where count > 0

Elastic Detection Rule

{
  "rule": {
    "name": "WMI Persistence via Event Subscription",
    "query": "event.code:(\"19\" OR \"20\" OR \"21\") AND winlog.event_data.EventType:\"WmiConsumerEvent\" AND winlog.event_data.Type:(\"CommandLineEventConsumer\" OR \"ActiveScriptEventConsumer\")",
    "severity": "high",
    "risk_score": 73,
    "tags": ["ATT&CK T1546.003"]
  }
}

MITRE ATT&CK Mapping

  • Technique: T1546.003 - Event Triggered Execution: WMI Event Subscription
  • Tactic: Persistence, Privilege Escalation
  • Data Sources: WMI Objects (WMI Creation), Command Execution, Process Creation

Autoruns WMI Tab

autorunsc64.exe -accepteula -w -nobanner -c

Output includes WMI subscriptions under "WMI" category with filter name, consumer, and command details.

Scripts 1

agent.py8.7 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""WMI Persistence Detection Agent - hunts for malicious WMI event subscriptions via Sysmon and WMI queries."""

import json
import argparse
import logging
import subprocess
import re
import xml.etree.ElementTree as ET
from datetime import datetime

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

REPLICATION_GUIDS = {
    "1131f6aa-9c07-11d1-f79f-00c04fc2dcd2": "DS-Replication-Get-Changes",
    "1131f6ad-9c07-11d1-f79f-00c04fc2dcd2": "DS-Replication-Get-Changes-All",
}

SUSPICIOUS_CONSUMERS = ["CommandLineEventConsumer", "ActiveScriptEventConsumer"]

KNOWN_GOOD_FILTERS = [
    "SCM Event Log Filter",
    "BVTFilter",
    "TSLogonFilter",
]


def query_sysmon_wmi_events(evtx_path=None, hours_back=72):
    """Query Sysmon Event IDs 19, 20, 21 for WMI persistence."""
    events = []
    for event_id in [19, 20, 21]:
        cmd = [
            "wevtutil", "qe", "Microsoft-Windows-Sysmon/Operational",
            "/q:*[System[EventID={}]]".format(event_id),
            "/f:xml", "/c:500",
        ]
        if evtx_path:
            cmd = ["wevtutil", "qe", evtx_path, "/lf:true",
                   "/q:*[System[EventID={}]]".format(event_id), "/f:xml", "/c:500"]
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
        for event_xml in re.findall(r"<Event.*?</Event>", result.stdout, re.DOTALL):
            try:
                root = ET.fromstring(event_xml)
                ns = {"s": "http://schemas.microsoft.com/win/2004/08/events/event"}
                data = {}
                for el in root.findall(".//s:Data", ns):
                    data[el.get("Name", "")] = el.text or ""
                events.append({
                    "event_id": event_id,
                    "timestamp": root.findtext(".//s:TimeCreated/@SystemTime", "", ns),
                    "computer": root.findtext(".//s:Computer", "", ns),
                    "operation": data.get("Operation", ""),
                    "event_type": data.get("EventType", ""),
                    "consumer_type": data.get("Type", ""),
                    "name": data.get("Name", ""),
                    "destination": data.get("Destination", ""),
                    "query": data.get("Query", ""),
                    "user": data.get("User", ""),
                    "raw_data": data,
                })
            except ET.ParseError:
                continue
    logger.info("Parsed %d Sysmon WMI events (IDs 19/20/21)", len(events))
    return events


def enumerate_wmi_subscriptions():
    """Enumerate active WMI event subscriptions via PowerShell."""
    subscriptions = {"filters": [], "consumers": [], "bindings": []}
    ps_commands = {
        "filters": "Get-WmiObject -Namespace root\\subscription -Class __EventFilter | Select Name, Query, QueryLanguage | ConvertTo-Json",
        "consumers": "Get-WmiObject -Namespace root\\subscription -Class __EventConsumer | Select __CLASS, Name, CommandLineTemplate, ScriptText | ConvertTo-Json",
        "bindings": "Get-WmiObject -Namespace root\\subscription -Class __FilterToConsumerBinding | Select Filter, Consumer | ConvertTo-Json",
    }
    for category, ps_cmd in ps_commands.items():
        cmd = ["powershell", "-Command", ps_cmd]
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
        if result.stdout.strip():
            try:
                data = json.loads(result.stdout)
                if isinstance(data, dict):
                    data = [data]
                subscriptions[category] = data
            except json.JSONDecodeError:
                pass
    return subscriptions


def analyze_suspicious_subscriptions(subscriptions):
    """Identify suspicious WMI subscriptions."""
    findings = []
    for consumer in subscriptions.get("consumers", []):
        consumer_class = consumer.get("__CLASS", "")
        name = consumer.get("Name", "")
        if consumer_class in SUSPICIOUS_CONSUMERS:
            severity = "critical"
            cmd_template = consumer.get("CommandLineTemplate", "")
            script_text = consumer.get("ScriptText", "")
            payload = cmd_template or script_text
            if any(kw in payload.lower() for kw in ["powershell", "cmd.exe", "wscript", "cscript", "mshta", "certutil", "bitsadmin"]):
                severity = "critical"
            elif payload:
                severity = "high"
            findings.append({
                "type": "suspicious_consumer",
                "consumer_class": consumer_class,
                "name": name,
                "payload": payload[:500],
                "severity": severity,
                "mitre_technique": "T1546.003",
            })
    for filt in subscriptions.get("filters", []):
        name = filt.get("Name", "")
        query = filt.get("Query", "")
        if name not in KNOWN_GOOD_FILTERS:
            if any(kw in query.lower() for kw in ["win32_processstarttr", "__instancecreationevent", "win32_logonsession"]):
                findings.append({
                    "type": "suspicious_filter",
                    "name": name,
                    "wql_query": query,
                    "severity": "high",
                    "mitre_technique": "T1546.003",
                })
    return findings


def analyze_sysmon_events(events):
    """Analyze Sysmon WMI events for suspicious patterns."""
    findings = []
    for event in events:
        eid = event["event_id"]
        if eid == 20 and event.get("consumer_type") in SUSPICIOUS_CONSUMERS:
            destination = event.get("destination", "")
            suspicious_cmds = ["powershell", "cmd.exe", "wscript", "mshta", "certutil", "regsvr32"]
            if any(cmd in destination.lower() for cmd in suspicious_cmds):
                findings.append({
                    "type": "sysmon_suspicious_consumer",
                    "event_id": eid,
                    "consumer_type": event["consumer_type"],
                    "destination": destination[:500],
                    "computer": event["computer"],
                    "timestamp": event["timestamp"],
                    "user": event["user"],
                    "severity": "critical",
                })
        if eid == 19:
            query = event.get("query", "")
            if "__instancecreationevent" in query.lower() or "win32_processstarttr" in query.lower():
                findings.append({
                    "type": "sysmon_suspicious_filter",
                    "event_id": eid,
                    "wql_query": query,
                    "computer": event["computer"],
                    "timestamp": event["timestamp"],
                    "severity": "high",
                })
    return findings


def generate_report(sysmon_events, live_subscriptions, sysmon_findings, subscription_findings):
    """Generate comprehensive WMI persistence hunt report."""
    all_findings = sysmon_findings + subscription_findings
    report = {
        "timestamp": datetime.utcnow().isoformat(),
        "hunt_type": "WMI Event Subscription Persistence (T1546.003)",
        "sysmon_events_analyzed": len(sysmon_events),
        "live_subscriptions": {
            "filters": len(live_subscriptions.get("filters", [])),
            "consumers": len(live_subscriptions.get("consumers", [])),
            "bindings": len(live_subscriptions.get("bindings", [])),
        },
        "total_findings": len(all_findings),
        "critical_findings": sum(1 for f in all_findings if f.get("severity") == "critical"),
        "high_findings": sum(1 for f in all_findings if f.get("severity") == "high"),
        "findings": all_findings,
    }
    return report


def main():
    parser = argparse.ArgumentParser(description="WMI Persistence Detection Agent")
    parser.add_argument("--evtx", help="Path to exported Sysmon .evtx file (optional)")
    parser.add_argument("--skip-live", action="store_true", help="Skip live WMI enumeration")
    parser.add_argument("--output", default="wmi_persistence_report.json")
    args = parser.parse_args()

    sysmon_events = query_sysmon_wmi_events(args.evtx)
    sysmon_findings = analyze_sysmon_events(sysmon_events)

    live_subs = {}
    sub_findings = []
    if not args.skip_live:
        live_subs = enumerate_wmi_subscriptions()
        sub_findings = analyze_suspicious_subscriptions(live_subs)

    report = generate_report(sysmon_events, live_subs, sysmon_findings, sub_findings)
    with open(args.output, "w") as f:
        json.dump(report, f, indent=2, default=str)
    logger.info("WMI hunt: %d events, %d findings (%d critical)",
                len(sysmon_events), report["total_findings"], report["critical_findings"])
    print(json.dumps(report, indent=2, default=str))


if __name__ == "__main__":
    main()
Keep exploring