Security Skills Atlas
soc operations

Performing Threat Hunting with Elastic SIEM

Performs proactive threat hunting in Elastic Security SIEM using KQL/EQL queries, detection rules, and Timeline investigation to identify threats that evade automated detection. Use when SOC teams need to hunt for specific ATT&CK techniques, investigate anomalous behaviors, or validate detection coverage gaps using Elasticsearch and Kibana Security.

elasticeqlkibanakqlmitre-attacksiemsocthreat-hunting
Install this skill
npx skills add mukul975/Anthropic-Cybersecurity-Skills
Framework mappings

MITRE ATT&CK

T1003 on the official MITRE ATT&CK siteT1021 on the official MITRE ATT&CK siteT1027 on the official MITRE ATT&CK siteT1053 on the official MITRE ATT&CK siteT1078 on the official MITRE ATT&CK siteT1105 on the official MITRE ATT&CK siteT1218 on the official MITRE ATT&CK siteT1547 on the official MITRE ATT&CK siteT1566 on the official MITRE ATT&CK siteT1685.002 on the official MITRE ATT&CK siteT1685.005 on the official MITRE ATT&CK site

NIST CSF 2.0

DE.AE-02 on the official NIST CSF 2.0 siteDE.AE-06 on the official NIST CSF 2.0 siteDE.CM-01 on the official NIST CSF 2.0 siteRS.MA-01 on the official NIST CSF 2.0 site

MITRE ATLAS

AML.T0066 on the official MITRE ATLAS siteAML.T0070 on the official MITRE ATLAS siteAML.T0082 on the official MITRE ATLAS site

MITRE D3FEND

Application Protocol Command Analysis on the official MITRE D3FEND siteClient-server Payload Profiling on the official MITRE D3FEND siteNetwork Isolation on the official MITRE D3FEND siteNetwork Traffic Analysis on the official MITRE D3FEND siteNetwork Traffic Community Deviation on the official MITRE D3FEND site

NIST AI RMF

MANAGE-2.4 on the official NIST AI RMF siteMAP-5.1 on the official NIST AI RMF siteMEASURE-2.7 on the official NIST AI RMF site

When to Use

Use this skill when:

  • SOC teams need to proactively search for threats not caught by existing detection rules
  • Threat intelligence reports describe new TTPs requiring validation against historical data
  • Red team exercises reveal detection gaps that need hunting query development
  • Periodic hunting cadence requires structured hypothesis-driven investigations

Do not use for real-time alert triage — that belongs in the Elastic Security Alerts queue with automated detection rules.

Prerequisites

  • Elastic Security 8.x+ with Security app enabled in Kibana
  • Data ingestion via Elastic Agent (Endpoint Security integration) or Beats (Winlogbeat, Filebeat, Packetbeat)
  • Data normalized to Elastic Common Schema (ECS) field mappings
  • User role with kibana_security_solution and read access to relevant indices
  • MITRE ATT&CK framework knowledge for hypothesis generation

Workflow

Step 1: Develop Hunting Hypothesis

Start with a hypothesis based on threat intelligence, ATT&CK technique, or anomaly:

Example Hypothesis: "Attackers are using living-off-the-land binaries (LOLBins) for execution, specifically certutil.exe for file downloads (T1105 — Ingress Tool Transfer)."

Define scope:

  • Data sources: logs-endpoint.events.process-*, logs-windows.sysmon_operational-*
  • Time range: Last 30 days
  • Expected indicators: certutil.exe with -urlcache, -split, or -decode flags

Step 2: Hunt Using KQL in Discover

Open Kibana Discover and query with KQL (Kibana Query Language):

process.name: "certutil.exe" and process.args: ("-urlcache" or "-split" or "-decode" or "-encode" or "-verifyctl")

Refine to exclude known legitimate use:

process.name: "certutil.exe"
  and process.args: ("-urlcache" or "-split" or "-decode")
  and not process.parent.name: ("sccm*.exe" or "ccmexec.exe")
  and not user.name: "SYSTEM"

For PowerShell-based hunting with encoded commands (T1059.001):

process.name: "powershell.exe"
  and process.args: ("-enc" or "-encodedcommand" or "-e " or "frombase64string" or "iex" or "invoke-expression")
  and not process.parent.executable: "C:\\Windows\\System32\\svchost.exe"

Step 3: Use EQL for Sequence Detection

Elastic Event Query Language (EQL) enables hunting for multi-step attack sequences:

Detect parent-child process anomalies (T1055 — Process Injection):

sequence by host.name with maxspan=5m
  [process where event.type == "start" and process.name == "explorer.exe"]
  [process where event.type == "start" and process.parent.name == "explorer.exe"
    and process.name in ("cmd.exe", "powershell.exe", "rundll32.exe", "regsvr32.exe")]

Detect credential dumping sequence (T1003):

sequence by host.name with maxspan=2m
  [process where event.type == "start"
    and process.name in ("procdump.exe", "procdump64.exe", "rundll32.exe", "taskmgr.exe")
    and process.args : "*lsass*"]
  [file where event.type == "creation"
    and file.extension in ("dmp", "dump", "bin")]

Detect lateral movement via PsExec (T1021.002):

sequence by source.ip with maxspan=1m
  [authentication where event.outcome == "success" and winlog.logon.type == "Network"]
  [process where event.type == "start"
    and process.name == "psexesvc.exe"]

Step 4: Investigate with Elastic Security Timeline

Create a Timeline investigation in Elastic Security for collaborative analysis:

  1. Navigate to Security > Timelines > Create new timeline
  2. Add events from hunting queries using "Add to timeline" from Discover
  3. Pin critical events and add investigation notes
  4. Use the Timeline query bar for additional filtering:
host.name: "WORKSTATION-042" and event.category: ("process" or "network" or "file")

Add columns for key fields: @timestamp, event.action, process.name, process.args, user.name, source.ip, destination.ip

Step 5: Build Detection Rules from Findings

Convert successful hunting queries into Elastic detection rules:

{
  "name": "Certutil Download Activity",
  "description": "Detects certutil.exe used for file download, a common LOLBin technique",
  "risk_score": 73,
  "severity": "high",
  "type": "eql",
  "query": "process where event.type == \"start\" and process.name == \"certutil.exe\" and process.args : (\"-urlcache\", \"-split\", \"-decode\") and not process.parent.name : (\"ccmexec.exe\", \"sccm*.exe\")",
  "threat": [
    {
      "framework": "MITRE ATT&CK",
      "tactic": {
        "id": "TA0011",
        "name": "Command and Control"
      },
      "technique": [
        {
          "id": "T1105",
          "name": "Ingress Tool Transfer"
        }
      ]
    }
  ],
  "tags": ["Hunting", "LOLBins", "T1105"],
  "interval": "5m",
  "from": "now-6m",
  "enabled": true
}

Deploy via Elastic Security API:

curl -X POST "https://kibana:5601/api/detection_engine/rules" \
  -H "kbn-xsrf: true" \
  -H "Content-Type: application/json" \
  -H "Authorization: ApiKey YOUR_API_KEY" \
  -d @certutil_rule.json

Step 6: Aggregate and Visualize Findings

Create hunting dashboard with aggregations:

GET logs-endpoint.events.process-*/_search
{
  "size": 0,
  "query": {
    "bool": {
      "must": [
        {"term": {"process.name": "certutil.exe"}},
        {"range": {"@timestamp": {"gte": "now-30d"}}}
      ]
    }
  },
  "aggs": {
    "by_host": {
      "terms": {"field": "host.name", "size": 20},
      "aggs": {
        "by_user": {
          "terms": {"field": "user.name", "size": 10}
        },
        "by_args": {
          "terms": {"field": "process.args", "size": 10}
        }
      }
    }
  }
}

Step 7: Document Hunt and Close Loop

Record findings in a structured hunt report and update detection coverage:

  • Hypothesis validated or refuted
  • IOCs and affected hosts discovered
  • Detection rules created or updated
  • ATT&CK Navigator layer updated with new coverage
  • Recommendations for security control improvements

Key Concepts

Term Definition
KQL Kibana Query Language — simplified query syntax for filtering data in Kibana Discover and dashboards
EQL Event Query Language — Elastic's sequence-aware query language for detecting multi-step attack patterns
ECS Elastic Common Schema — standardized field naming convention enabling cross-source correlation
Timeline Elastic Security investigation workspace for collaborative event analysis and annotation
Hypothesis-Driven Hunting Structured approach starting with a theory about attacker behavior, tested against telemetry data
LOLBins Living Off the Land Binaries — legitimate Windows tools (certutil, mshta, rundll32) abused by attackers

Tools & Systems

  • Elastic Security: SIEM platform built on Elasticsearch with detection rules, Timeline, and case management
  • Elastic Agent: Unified data collection agent replacing Beats for endpoint and network telemetry
  • Elastic Endpoint Security: EDR capabilities integrated into Elastic Agent for process, file, and network monitoring
  • ATT&CK Navigator: MITRE tool for tracking detection and hunting coverage across the ATT&CK matrix

Common Scenarios

  • LOLBin Abuse: Hunt for mshta.exe, regsvr32.exe, rundll32.exe, certutil.exe with suspicious arguments
  • Persistence Mechanisms: Query for scheduled task creation, registry run key modification, WMI subscriptions
  • C2 Beaconing: Analyze network flow data for periodic outbound connections with consistent intervals
  • Data Staging: Hunt for large file compression (7z, rar, zip) followed by outbound transfers
  • Account Manipulation: Search for net.exe user creation, group membership changes, or password resets by non-admin users

Output Format

THREAT HUNT REPORT — TH-2024-012
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Hypothesis:   Attackers using certutil.exe for tool download (T1105)
Period:       2024-02-15 to 2024-03-15
Data Sources: Elastic Endpoint (process events), Sysmon
 
Findings:
  Total certutil executions:     342
  With -urlcache flag:           12 (3.5%)
  Suspicious (non-SCCM):        3 confirmed anomalous
 
Affected Hosts:
  WORKSTATION-042 (Finance)  — certutil downloading payload.exe from external IP
  SERVER-DB-03 (Database)    — certutil decoding base64 encoded binary
  LAPTOP-EXEC-07 (Executive) — certutil downloading script from Pastebin
 
Actions Taken:
  [DONE] 3 hosts isolated for forensic investigation
  [DONE] Detection rule "Certutil Download Activity" deployed (ID: elastic-th012)
  [DONE] ATT&CK Navigator updated: T1105 coverage = GREEN
 
Verdict:      HYPOTHESIS CONFIRMED — 3 true positive findings escalated to IR
Source materials

References and resources

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

References 1

api-reference.md2.4 KB

API Reference: Threat Hunting with Elastic SIEM Agent

Overview

Performs proactive threat hunting against Elasticsearch indices using structured queries for LOLBin abuse, credential dumping, lateral movement, and persistence mechanisms mapped to MITRE ATT&CK.

Dependencies

Package Version Purpose
elasticsearch >= 8.0 Elasticsearch Python client for queries

Core Functions

get_es_client(host, api_key, verify_certs)

Creates an authenticated Elasticsearch client.

  • Parameters: host (str), api_key (str, optional), verify_certs (bool)
  • Returns: Elasticsearch client instance

hunt_lolbins(es, index, days)

Hunts for LOLBin abuse (certutil, mshta, regsvr32, etc.) with suspicious arguments.

  • ATT&CK: T1105 (Ingress Tool Transfer), T1218 (Signed Binary Proxy Execution)
  • Returns: dict with hunt, total_hits, findings

hunt_credential_dumping(es, index, days)

Detects procdump targeting lsass, mimikatz execution, sekurlsa PowerShell commands.

  • ATT&CK: T1003 (OS Credential Dumping)
  • Returns: dict with hunt results

hunt_lateral_movement(es, index, days)

Identifies PsExec, Invoke-Command, and SMB/WinRM network flows.

  • ATT&CK: T1021 (Remote Services)
  • Returns: dict with hunt results

hunt_persistence(es, index, days)

Detects scheduled task creation and registry Run key modifications.

  • ATT&CK: T1053 (Scheduled Task), T1547 (Boot/Logon Autostart)
  • Returns: dict with hunt results

create_detection_rule(es, kibana_url, name, query, severity, risk_score)

Generates a detection rule payload for Elastic Security API deployment.

  • Returns: dict - rule configuration ready for POST to /api/detection_engine/rules

run_all_hunts(es, days)

Executes all hunt queries and aggregates results.

Elasticsearch Indices Used

Index Pattern Data Source
logs-endpoint.events.process-* Elastic Agent process events
logs-endpoint.events.* All endpoint event types
logs-windows.sysmon_operational-* Sysmon via Winlogbeat

Environment Variables

Variable Required Description
ES_HOST No Elasticsearch URL (default: https://localhost:9200)
ES_API_KEY No API key for authentication

Usage

python agent.py https://elastic.corp.local:9200

Scripts 1

agent.py8.3 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""Threat hunting agent for Elastic SIEM using elasticsearch-py."""

import os
import sys
from datetime import datetime

try:
    from elasticsearch import Elasticsearch
except ImportError:
    print("Install: pip install elasticsearch")
    sys.exit(1)


def get_es_client(host=None, api_key=None, verify_certs=True):
    host = host or os.environ.get("ES_HOSTS", "https://localhost:9200")
    kwargs = {"hosts": [host], "verify_certs": verify_certs}
    if api_key:
        kwargs["api_key"] = api_key
    return Elasticsearch(**kwargs)


def hunt_lolbins(es, index="logs-endpoint.events.process-*", days=30):
    """Hunt for living-off-the-land binary abuse."""
    lolbins = [
        "certutil.exe", "mshta.exe", "regsvr32.exe", "rundll32.exe",
        "cscript.exe", "wscript.exe", "bitsadmin.exe",
    ]
    suspicious_args = [
        "-urlcache", "-split", "-decode", "-encode", "javascript:",
        "scrobj.dll", "/transfer", "-encodedcommand", "-enc",
    ]
    query = {
        "size": 100,
        "query": {
            "bool": {
                "must": [
                    {"terms": {"process.name": lolbins}},
                    {"range": {"@timestamp": {"gte": f"now-{days}d"}}},
                ],
                "should": [{"match_phrase": {"process.args": arg}} for arg in suspicious_args],
                "minimum_should_match": 1,
                "must_not": [
                    {"terms": {"process.parent.name": ["ccmexec.exe", "sccm.exe"]}},
                ],
            }
        },
        "sort": [{"@timestamp": "desc"}],
    }
    result = es.search(index=index, body=query)
    hits = []
    for hit in result["hits"]["hits"]:
        src = hit["_source"]
        hits.append({
            "timestamp": src.get("@timestamp"),
            "host": src.get("host", {}).get("name"),
            "user": src.get("user", {}).get("name"),
            "process": src.get("process", {}).get("name"),
            "args": src.get("process", {}).get("args"),
            "parent": src.get("process", {}).get("parent", {}).get("name"),
        })
    return {"hunt": "LOLBin Abuse", "total_hits": result["hits"]["total"]["value"], "findings": hits}


def hunt_credential_dumping(es, index="logs-endpoint.events.process-*", days=30):
    """Hunt for credential dumping activity (T1003)."""
    query = {
        "size": 50,
        "query": {
            "bool": {
                "must": [{"range": {"@timestamp": {"gte": f"now-{days}d"}}}],
                "should": [
                    {"bool": {"must": [
                        {"terms": {"process.name": ["procdump.exe", "procdump64.exe", "rundll32.exe"]}},
                        {"match_phrase": {"process.args": "lsass"}},
                    ]}},
                    {"bool": {"must": [
                        {"term": {"process.name": "mimikatz.exe"}},
                    ]}},
                    {"bool": {"must": [
                        {"term": {"process.name": "powershell.exe"}},
                        {"match_phrase": {"process.args": "sekurlsa"}},
                    ]}},
                ],
                "minimum_should_match": 1,
            }
        },
    }
    result = es.search(index=index, body=query)
    hits = []
    for hit in result["hits"]["hits"]:
        src = hit["_source"]
        hits.append({
            "timestamp": src.get("@timestamp"),
            "host": src.get("host", {}).get("name"),
            "process": src.get("process", {}).get("name"),
            "args": src.get("process", {}).get("args"),
        })
    return {"hunt": "Credential Dumping (T1003)", "total_hits": result["hits"]["total"]["value"], "findings": hits}


def hunt_lateral_movement(es, index="logs-endpoint.events.*", days=14):
    """Hunt for lateral movement patterns (T1021)."""
    query = {
        "size": 50,
        "query": {
            "bool": {
                "must": [{"range": {"@timestamp": {"gte": f"now-{days}d"}}}],
                "should": [
                    {"term": {"process.name": "psexesvc.exe"}},
                    {"bool": {"must": [
                        {"term": {"process.name": "powershell.exe"}},
                        {"match_phrase": {"process.args": "invoke-command"}},
                    ]}},
                    {"bool": {"must": [
                        {"term": {"event.action": "network_flow"}},
                        {"terms": {"destination.port": [445, 135, 5985, 5986]}},
                    ]}},
                ],
                "minimum_should_match": 1,
            }
        },
    }
    result = es.search(index=index, body=query)
    hits = []
    for hit in result["hits"]["hits"]:
        src = hit["_source"]
        hits.append({
            "timestamp": src.get("@timestamp"),
            "host": src.get("host", {}).get("name"),
            "process": src.get("process", {}).get("name"),
            "source_ip": src.get("source", {}).get("ip"),
            "dest_ip": src.get("destination", {}).get("ip"),
            "dest_port": src.get("destination", {}).get("port"),
        })
    return {"hunt": "Lateral Movement (T1021)", "total_hits": result["hits"]["total"]["value"], "findings": hits}


def hunt_persistence(es, index="logs-endpoint.events.*", days=30):
    """Hunt for persistence mechanisms (T1053, T1547)."""
    query = {
        "size": 50,
        "query": {
            "bool": {
                "must": [{"range": {"@timestamp": {"gte": f"now-{days}d"}}}],
                "should": [
                    {"bool": {"must": [
                        {"term": {"process.name": "schtasks.exe"}},
                        {"match_phrase": {"process.args": "/create"}},
                    ]}},
                    {"bool": {"must": [
                        {"term": {"process.name": "reg.exe"}},
                        {"match_phrase": {"process.args": "CurrentVersion\\Run"}},
                    ]}},
                    {"bool": {"must": [
                        {"term": {"event.action": "registry_value_set"}},
                        {"wildcard": {"registry.path": "*CurrentVersion\\Run*"}},
                    ]}},
                ],
                "minimum_should_match": 1,
            }
        },
    }
    result = es.search(index=index, body=query)
    hits = []
    for hit in result["hits"]["hits"]:
        src = hit["_source"]
        hits.append({
            "timestamp": src.get("@timestamp"),
            "host": src.get("host", {}).get("name"),
            "process": src.get("process", {}).get("name"),
            "args": src.get("process", {}).get("args"),
        })
    return {"hunt": "Persistence (T1053/T1547)", "total_hits": result["hits"]["total"]["value"], "findings": hits}


def create_detection_rule(es, kibana_url, name, query, severity="high", risk_score=73):
    """Deploy a detection rule to Elastic Security via API."""
    rule = {
        "name": name,
        "description": f"Detection rule created from threat hunt: {name}",
        "risk_score": risk_score,
        "severity": severity,
        "type": "query",
        "query": query,
        "index": ["logs-endpoint.events.process-*"],
        "interval": "5m",
        "from": "now-6m",
        "enabled": True,
        "tags": ["Hunting", "Auto-generated"],
    }
    return rule


def run_all_hunts(es, days=30):
    hunts = []
    hunts.append(hunt_lolbins(es, days=days))
    hunts.append(hunt_credential_dumping(es, days=days))
    hunts.append(hunt_lateral_movement(es, days=min(days, 14)))
    hunts.append(hunt_persistence(es, days=days))
    return hunts


def print_hunt_report(hunts):
    print("THREAT HUNT REPORT")
    print("=" * 50)
    print(f"Date: {datetime.now().isoformat()}")
    total_findings = sum(h["total_hits"] for h in hunts)
    print(f"Total Findings: {total_findings}\n")
    for hunt in hunts:
        print(f"--- {hunt['hunt']} ---")
        print(f"Hits: {hunt['total_hits']}")
        for f in hunt["findings"][:5]:
            print(f"  {f.get('timestamp', 'N/A')} | {f.get('host', 'N/A')} | "
                  f"{f.get('process', 'N/A')} | {f.get('args', '')}")
        if hunt["total_hits"] > 5:
            print(f"  ... and {hunt['total_hits'] - 5} more")
        print()


if __name__ == "__main__":
    host = sys.argv[1] if len(sys.argv) > 1 else os.environ.get("ES_HOSTS", "https://localhost:9200")
    es = get_es_client(host=host, verify_certs=False)
    results = run_all_hunts(es)
    print_hunt_report(results)
Keep exploring
Building Detection Rules with SigmaBuilds vendor-agnostic detection rules using the Sigma rule format for threat detection across SIEM platforms including Splunk, Elastic, and Microsoft Sentinel. Use when creating portable detection logic from threat intelligence, mapping rules to MITRE ATT&CK techniques, or converting community Sigma rules into platform-specific queries using sigmac or pySigma backends.Implementing SIEM Use Cases for DetectionImplements SIEM detection use cases by designing correlation rules, threshold alerts, and behavioral analytics mapped to MITRE ATT&CK techniques across Splunk, Elastic, and Sentinel. Use when SOC teams need to expand detection coverage, formalize use case lifecycle management, or build a detection library aligned to organizational threat profile.Performing Alert Triage with Elastic SIEMPerform systematic alert triage in Elastic Security SIEM to rapidly classify, prioritize, and investigate security alerts for SOC operations.Analyzing Windows Event Logs in SplunkAnalyzes Windows Security, System, and Sysmon event logs in Splunk to detect authentication attacks, privilege escalation, persistence mechanisms, and lateral movement using SPL queries mapped to MITRE ATT&CK techniques. Use when SOC analysts need to investigate Windows-based threats, build detection queries, or perform forensic timeline analysis of Windows endpoints and domain controllers.Building Detection Rules with Splunk SPLBuild effective detection rules using Splunk Search Processing Language (SPL) correlation searches to identify security threats in SOC environments.Building SOC Playbook for RansomwareBuilds a structured SOC incident response playbook for ransomware attacks covering detection, containment, eradication, and recovery phases with specific SIEM queries, isolation procedures, and decision trees. Use when SOC teams need formalized response procedures for ransomware incidents aligned to NIST SP 800-61 and MITRE ATT&CK ransomware techniques.