CrowdSentinel Skills - Threat Hunting & Incident Response
When to Use This Skill
Use this skill when:
- Conducting threat hunting on Elasticsearch/OpenSearch logs
- Running detection rules (EQL, ES|QL, Lucene)
- Investigating security incidents
- Analysing Windows/Linux logs for threats
- Debugging Elasticsearch connection issues
- Handling field mapping mismatches (non-ECS data)
- Presenting investigation findings to users
The Iron Law
┌─────────────────────────────────────────────────────────────────────┐
│ NO INVESTIGATION IS COMPLETE WITHOUT ANALYSIS TOOLS │
│ │
│ If you have collected data but haven't used analysis tools, │
│ the investigation is INCOMPLETE. │
└─────────────────────────────────────────────────────────────────────┘
Decision Tree: What Should I Do?
START: User Request
│
├─→ "Hunt for threats" / "Find suspicious activity"
│ └─→ Go to: THREAT HUNTING WORKFLOW
│
├─→ "Run detection rule" / "Execute rule"
│ └─→ Go to: DETECTION RULE WORKFLOW
│
├─→ "Investigate incident" / "Analyse this host"
│ └─→ Go to: INCIDENT RESPONSE WORKFLOW
│
├─→ "Connection error" / "Query failed"
│ └─→ Go to: DEBUGGING WORKFLOW
│
├─→ "Field not found" / "Query returns empty"
│ └─→ Go to: FIELD MAPPING WORKFLOW
│
└─→ "Analyse PCAP" / "Network traffic"
└─→ Go to: WIRESHARK WORKFLOW (see wireshark-mcp-ir-th skill)
THREAT HUNTING WORKFLOW
Phase 1: Data Collection
| Tool | When to Use | Token Efficiency | |
|---|
threat_hunt_search | IR-focused search with auto IoC extraction | High (PREFERRED) | |
smart_search | Quick counts and summarisation | Very High | |
execute_detection_rule | Run curated EQL/Lucene rules | Medium | |
execute_esql_hunt | Run curated ES | QL hunting queries | Medium | |
esql_query | Ad-hoc ES | QL queries | Low | |
eql_search | Ad-hoc EQL queries | Low | |
search_with_lucene | Flexible Lucene queries | Low | |
Decision: Which search tool?
Is this an initial triage?
│
├─→ YES: Use smart_search or quick_count first
│ └─→ Then threat_hunt_search for details
│
└─→ NO: Do you need specific detection logic?
│
├─→ YES: Use execute_detection_rule or execute_esql_hunt
│
└─→ NO: Use threat_hunt_search (auto IoC extraction)
Phase 2: Analysis (MANDATORY - NEVER SKIP)
After EVERY search/hunt query, you MUST use analysis tools:
python
# Step 1: Analyse search results
analysis = analyze_search_results(
search_results=results,
context="Description of what you searched for"
)
# Returns: IoCs, MITRE ATT&CK mapping, severity, recommendations
# Step 2: Position in kill chain
kill_chain = analyze_kill_chain_stage(
iocs=[{"type": "ip", "value": "..."}, {"type": "process", "value": "..."}],
include_hunting_suggestions=True
)
# Returns: Kill chain stage, previous/next stage hunting queries
Phase 3: Investigation State (Multi-Query)
For investigations spanning multiple queries:
python
# Start tracking
create_investigation(
name="Suspicious PowerShell Activity",
description="Investigating encoded commands on DESKTOP-001",
severity="high"
)
# Store discovered IoCs
add_iocs_to_investigation(iocs=[...])
# Retrieve for cross-correlation
shared_iocs = get_shared_iocs(ioc_types=["ip", "domain"])
Phase 4: Reporting (BEFORE Concluding)
python
report = generate_investigation_report(
analysis_results=[analysis1, analysis2, ...],
investigation_context="Summary of investigation"
)
# Returns: Executive summary, all IoCs, affected hosts, recommendations
DETECTION RULE WORKFLOW
Decision: Which rule type?
What log source are you hunting in?
│
├─→ Windows Event Logs (winlogbeat-*, logs-windows.*)
│ │
│ └─→ What behaviour are you detecting?
│ │
│ ├─→ Sequences/Correlations → Use EQL rules
│ │ execute_detection_rule(rule_id="..._eql", index="winlogbeat-*")
│ │
│ └─→ Single events → Use Lucene rules
│ execute_detection_rule(rule_id="..._lucene", index="winlogbeat-*")
│
├─→ Linux Audit Logs (auditbeat-*, logs-linux.*)
│ └─→ Use ES|QL hunts or Lucene rules
│ execute_esql_hunt(rule_id="...", index="auditbeat-*")
│
└─→ Network Logs (packetbeat-*, logs-network.*)
└─→ Use Lucene rules or smart_search
Finding Rules
python
# List rules by platform and tactic
rules = list_detection_rules(
platform="windows",
mitre_tactic="credential_access",
limit=20
)
# Search by keyword
rules = list_detection_rules(search_term="mimikatz")
# Get rule details before execution
details = get_rule_details(rule_id="windows_powershell_...")
Executing Rules
python
# Execute single rule
result = execute_detection_rule(
rule_id="windows_builtin_win_security_susp_logon_eql",
index="winlogbeat-*",
timeframe_minutes=60
)
# Execute multiple rules (batch hunt)
results = execute_multiple_rules(
rule_ids=["rule1", "rule2", "rule3"],
index="winlogbeat-*",
timeframe_minutes=1440
)
# Hunt by category
results = hunt_with_rule_category(
platform="windows",
category="powershell",
index="winlogbeat-*",
timeframe_minutes=60
)
INCIDENT RESPONSE WORKFLOW
NIST IR Framework Integration
┌──────────────────────────────────────────────────────────────────────┐
│ NIST IR LIFECYCLE │
├──────────────────────────────────────────────────────────────────────┤
│ │
│ 1. PREPARATION │ Create investigation, set severity │
│ └─→ create_investigation(name, severity) │
│ │
│ 2. DETECTION │ Run detection rules, threat hunts │
│ └─→ execute_detection_rule(), threat_hunt_search() │
│ │
│ 3. ANALYSIS │ Analyse results, map kill chain │
│ └─→ analyze_search_results(), analyze_kill_chain_stage() │
│ │
│ 4. CONTAINMENT │ Identify scope, affected hosts │
│ └─→ get_host_activity_timeline(), hunt_for_ioc() │
│ │
│ 5. ERADICATION │ Track IoCs for removal │
│ └─→ export_iocs(format="values"), add_iocs_to_investigation() │
│ │
│ 6. RECOVERY │ Generate report for stakeholders │
│ └─→ generate_investigation_report() │
│ │
│ 7. LESSONS LEARNED │ Document findings │
│ └─→ close_investigation(resolution="...") │
│ │
└──────────────────────────────────────────────────────────────────────┘
Pyramid of Pain: IoC Prioritisation
┌───────────┐
│ TTPs │ ← HIGHEST VALUE: Behaviours (hunt_by_kill_chain_stage)
┌┴───────────┴┐
│ Tools │ ← Tool names, filenames (hunt_for_ioc type="process")
┌┴─────────────┴┐
│ Artifacts │ ← Registry keys, mutexes (analyze_search_results)
┌┴───────────────┴┐
│ Domains │ ← Domain names (hunt_for_ioc type="domain")
┌┴─────────────────┴┐
│ IP Addresses │ ← IPs (hunt_for_ioc type="ip")
┌┴───────────────────┴┐
│ Hash Values │ ← File hashes (hunt_for_ioc type="hash")
└─────────────────────┘
LOWEST VALUE
Tool selection by IoC type:
| IoC Type | CrowdSentinel Tool | Priority | |
|---|
| Hash | hunt_for_ioc(ioc_type="hash") | 1 (Trivial) | |
| IP | hunt_for_ioc(ioc_type="ip") | 2 (Easy) | |
| Domain | hunt_for_ioc(ioc_type="domain") | 3 (Simple) | |
| Artifact | analyze_search_results → extract | 4 (Annoying) | |
| Tool | hunt_for_ioc(ioc_type="process") | 5 (Challenging) | |
| TTP | hunt_by_kill_chain_stage, execute_detection_rule | 6 (Tough) | |
Diamond Model Analysis
ADVERSARY
│
(Who is attacking?)
analyze_kill_chain_stage()
│
┌─────────────────┼─────────────────┐
│ │ │
CAPABILITY ───────────┼─────────── INFRASTRUCTURE
│ │ │
(What tools?) │ (What IPs/Domains?)
hunt_for_ioc() │ hunt_for_ioc()
(type="process") │ (type="ip/domain")
│ │ │
└─────────────────┼─────────────────┘
│
VICTIM
│
(Who is targeted?)
get_host_activity_timeline()
Kill Chain Hunting
python
# Hunt specific kill chain stage
results = hunt_by_kill_chain_stage(
index="winlogbeat-*",
stage="INSTALLATION", # or RECONNAISSANCE, DELIVERY, EXPLOITATION, etc.
timeframe_minutes=1440
)
# Hunt adjacent stages (before and after)
results = hunt_adjacent_stages(
index="winlogbeat-*",
current_stage="COMMAND_AND_CONTROL",
hunt_previous=True, # Find how they persisted
hunt_next=True # Find what actions they took
)
# Analyse IoCs against kill chain
analysis = analyze_kill_chain_stage(
iocs=[
{"type": "ip", "value": "203.0.113.42"},
{"type": "process", "value": "mimikatz.exe"}
],
include_hunting_suggestions=True
)
DEBUGGING WORKFLOW
Connection Issues
Query failed or empty results?
│
├─→ "Connection refused" / "Timeout"
│ └─→ Run: scripts/debugging/check_connection.sh
│
├─→ "Authentication failed" / "401"
│ └─→ Run: scripts/debugging/check_auth.py
│
├─→ "Index not found" / "404"
│ └─→ Run: scripts/debugging/list_indices.sh
│
└─→ "Empty results but index exists"
└─→ Go to: FIELD MAPPING WORKFLOW
Quick Diagnostic Commands
bash
# Check cluster health
curl -k -u "$ES_USER:$ES_PASS" "$ES_HOST/_cluster/health?pretty"
# List all indices with document counts
curl -k -u "$ES_USER:$ES_PASS" "$ES_HOST/_cat/indices?v&s=docs.count:desc"
# Check index mapping
curl -k -u "$ES_USER:$ES_PASS" "$ES_HOST/winlogbeat-*/_mapping?pretty"
# Test simple query
curl -k -u "$ES_USER:$ES_PASS" "$ES_HOST/winlogbeat-*/_count"
Using MCP Tools for Debugging
python
# Check cluster health
health = get_cluster_health()
# List indices
indices = list_indices()
# Get index details
details = get_index(index="winlogbeat-*")
# Discover all assets
assets = discover_all_assets()
FIELD MAPPING WORKFLOW
The Problem
Detection rules use ECS field names, but your data may use different schemas: