| argument-hint | [index=<index>] [days=30] | |||
|---|---|---|---|---|
| allowed-tools |
|
Cross-reference MITRE ATT&CK data sources against what's actually flowing into Splunk — at the event ID level. Maps specific EventCodes, CloudTrail eventNames, and audit types to ATT&CK data components and techniques.
$ARGUMENTS optionally contains:
index=<name>to scope to a specific Splunk index (default: all indexes viaindex=*)days=Nfor the lookback window (default: 30)
If no arguments are given, run against all indexes with a 30-day window.
SPLUNK_URL -- from $SPLUNK_URL
SPLUNK_TOKEN -- from $SPLUNK_TOKEN
ATTACK_JSON=/home/anton/Desktop/claudeforblueteam/enterprise-attack.json
ANALYSIS_SCRIPT=/home/anton/Desktop/claudeforblueteam/attack-coverage.py
VAULT=Obsidian Vault
Verify $SPLUNK_URL and $SPLUNK_TOKEN exist before proceeding.
Target: 1 sourcetype discovery query, 1-4 event ID queries (parallel), 1 Python analysis, 1 Obsidian write.
curl -s -k \
-H "Authorization: Bearer $SPLUNK_TOKEN" \
-d "search=search index=<INDEX> earliest=-<DAYS>d latest=now | stats count by sourcetype source | sort -count" \
-d "output_mode=json" \
-d "exec_mode=oneshot" \
-d "earliest_time=-<DAYS>d" \
-d "latest_time=now" \
-d "count=0" \
"$SPLUNK_URL/services/search/jobs/export" > /tmp/splunk_discovery.jsonParse the NDJSON — take the last result set (streaming export sends progressive results; only the final batch has complete counts). Extract unique (sourcetype, source) pairs.
Based on what Step 1 discovered, run the applicable queries in parallel:
If XmlWinEventLog present (Windows events):
curl -s -k \
-H "Authorization: Bearer $SPLUNK_TOKEN" \
-d "search=search index=<INDEX> sourcetype=XmlWinEventLog earliest=-<DAYS>d latest=now | stats count by source EventCode | sort -count" \
-d "output_mode=json" \
-d "exec_mode=oneshot" \
-d "earliest_time=-<DAYS>d" \
-d "latest_time=now" \
-d "count=0" \
"$SPLUNK_URL/services/search/jobs/export" > /tmp/splunk_winevt.jsonIf aws:cloudtrail present:
curl -s -k \
-H "Authorization: Bearer $SPLUNK_TOKEN" \
-d "search=search index=<INDEX> sourcetype=\"aws:cloudtrail\" earliest=-<DAYS>d latest=now | stats count by eventName | sort -count" \
-d "output_mode=json" \
-d "exec_mode=oneshot" \
-d "earliest_time=-<DAYS>d" \
-d "latest_time=now" \
-d "count=0" \
"$SPLUNK_URL/services/search/jobs/export" > /tmp/splunk_cloudtrail.jsonIf Azure sourcetypes present (mscs:azure:, azure:):
curl -s -k \
-H "Authorization: Bearer $SPLUNK_TOKEN" \
-d "search=search index=<INDEX> sourcetype=\"mscs:azure:*\" earliest=-<DAYS>d latest=now | stats count by operationName | sort -count" \
-d "output_mode=json" \
-d "exec_mode=oneshot" \
-d "earliest_time=-<DAYS>d" \
-d "latest_time=now" \
-d "count=0" \
"$SPLUNK_URL/services/search/jobs/export" > /tmp/splunk_azure.jsonIf auditd/audit- present (Linux audit):*
curl -s -k \
-H "Authorization: Bearer $SPLUNK_TOKEN" \
-d "search=search index=<INDEX> sourcetype=\"audit*\" earliest=-<DAYS>d latest=now | eval audit_type=coalesce(type, SYSCALL) | stats count by audit_type | sort -count" \
-d "output_mode=json" \
-d "exec_mode=oneshot" \
-d "earliest_time=-<DAYS>d" \
-d "latest_time=now" \
-d "count=0" \
"$SPLUNK_URL/services/search/jobs/export" > /tmp/splunk_auditd.jsonParse all NDJSON result files. For each, take the final result set (last occurrence of each unique key).
Build a single JSON array of {source, event_id, count} objects:
- Windows events:
source= thesourcefield (e.g.,XmlWinEventLog:Microsoft-Windows-Sysmon/Operational),event_id=EventCode - CloudTrail:
source=aws:cloudtrail,event_id=eventName - Azure:
source=mscs:azure:eventhub,event_id=operationName - Auditd:
source=audit,event_id=audit_type
Write to /tmp/splunk_events.json.
Use a Python one-liner or small script. Example:
import json, sys
results = []
# Parse Windows events
for line in open('/tmp/splunk_winevt.json'):
try:
obj = json.loads(line.strip())
if 'result' in obj:
r = obj['result']
results.append({'source': r.get('source',''), 'event_id': r.get('EventCode',''), 'count': int(r.get('count',0))})
except: pass
# Parse CloudTrail
for line in open('/tmp/splunk_cloudtrail.json'):
try:
obj = json.loads(line.strip())
if 'result' in obj:
r = obj['result']
results.append({'source': 'aws:cloudtrail', 'event_id': r.get('eventName',''), 'count': int(r.get('count',0))})
except: pass
# ... repeat for Azure, auditd ...
# Deduplicate — keep highest count per (source, event_id)
deduped = {}
for r in results:
key = (r['source'], str(r['event_id']))
if key not in deduped or r['count'] > deduped[key]['count']:
deduped[key] = r
results = list(deduped.values())
with open('/tmp/splunk_events.json', 'w') as f:
json.dump(results, f, indent=2)
print(f"Wrote {len(results)} event entries")Adapt the script to include only the source types that actually exist in the Splunk environment.
First, write the analysis script to /tmp/attack_coverage.py if it doesn't already exist:
attack_coverage.py (click to expand)
#!/usr/bin/env python3
"""ATT&CK Coverage Analyzer — maps SIEM event IDs to ATT&CK data components and techniques.
Usage:
python3 attack_coverage.py --attack-json <path> --splunk-events <path> [--date <YYYY-MM-DD>] [--index <name>] [--days <N>]
--splunk-events: JSON file with an array of {source, event_id, count} objects from Splunk.
source = Splunk source field (e.g. "XmlWinEventLog:Microsoft-Windows-Sysmon/Operational")
event_id = EventCode, eventName, operation, etc. depending on log type
count = event count
Outputs Obsidian-formatted Markdown to stdout.
"""
import argparse
import json
import re
import sys
from collections import defaultdict
def parse_args():
p = argparse.ArgumentParser()
p.add_argument('--attack-json', required=True)
p.add_argument('--splunk-events', required=True)
p.add_argument('--date', default='2026-01-01')
p.add_argument('--index', default='*')
p.add_argument('--days', type=int, default=30)
return p.parse_args()
def load_attack(path):
with open(path) as f:
return json.load(f)
def load_splunk_events(path):
with open(path) as f:
return json.load(f)
def build_mappings(attack_data):
"""Build the full ATT&CK v18 chain: data component -> analytics -> detection strategies -> techniques."""
objects = attack_data['objects']
# Data sources (for grouping names)
datasources = {o['id']: o['name'] for o in objects if o.get('type') == 'x-mitre-data-source'}
ds_names = sorted(datasources.values())
# Active data components
components = {}
for o in objects:
if o.get('type') == 'x-mitre-data-component' and o.get('x_mitre_deprecated') == False:
components[o['id']] = {
'id': o['id'],
'name': o['name'],
'log_sources': o.get('x_mitre_log_sources', []),
}
# Active techniques
techniques = {}
for o in objects:
if o.get('type') == 'attack-pattern' and not o.get('x_mitre_deprecated') and not o.get('revoked'):
ext_id = ''
for ref in o.get('external_references', []):
if ref.get('source_name') == 'mitre-attack':
ext_id = ref.get('external_id', '')
break
techniques[o['id']] = {'id': o['id'], 'name': o['name'], 'ext_id': ext_id}
# Analytics -> data component refs (via x_mitre_log_source_references)
analytics = {}
for o in objects:
if o.get('type') == 'x-mitre-analytic' and not o.get('x_mitre_deprecated'):
comp_refs = set()
for ls_ref in o.get('x_mitre_log_source_references', []):
cr = ls_ref.get('x_mitre_data_component_ref', '')
if cr:
comp_refs.add(cr)
analytics[o['id']] = comp_refs
# Detection strategies -> analytics refs, and detects -> technique
det_strategies = {}
for o in objects:
if o.get('type') == 'x-mitre-detection-strategy' and not o.get('x_mitre_deprecated'):
det_strategies[o['id']] = set(o.get('x_mitre_analytic_refs', []))
# Detects relationships: detection-strategy -> technique
ds_to_tech = {}
for o in objects:
if (o.get('type') == 'relationship' and o.get('relationship_type') == 'detects'
and not o.get('x_mitre_deprecated')):
src = o['source_ref']
tgt = o['target_ref']
if tgt in techniques:
ds_to_tech.setdefault(src, set()).add(tgt)
# Chain: data_component -> techniques
# component <- analytics <- detection_strategy -> technique
comp_to_techs = defaultdict(set)
for ds_id, analytic_ids in det_strategies.items():
tech_ids = ds_to_tech.get(ds_id, set())
for a_id in analytic_ids:
comp_ids = analytics.get(a_id, set())
for c_id in comp_ids:
if c_id in components:
comp_to_techs[c_id].update(tech_ids)
# Assign parent data source to each component by name prefix matching
def get_parent(comp_name):
best = None
for ds in ds_names:
if comp_name.startswith(ds):
if best is None or len(ds) > len(best):
best = ds
return best or "Other"
for cid, comp in components.items():
comp['parent'] = get_parent(comp['name'])
comp['techniques'] = comp_to_techs.get(cid, set())
comp['technique_count'] = len(comp['techniques'])
return components, techniques, ds_names
def normalize_source(source):
"""Extract the log type from a Splunk source field."""
s = source.lower()
if 'sysmon' in s:
return 'sysmon'
elif 'security' in s and ('wineventlog' in s or 'xmlwineventlog' in s):
return 'winsecurity'
elif 'powershell' in s:
return 'powershell'
elif 'system' in s and ('wineventlog' in s or 'xmlwineventlog' in s):
return 'winsystem'
elif 'cloudtrail' in s:
return 'cloudtrail'
elif 'azure' in s or 'eventhub' in s:
return 'azure'
elif 'audit' in s:
return 'auditd'
return source.lower()
def match_log_source(ls_name, ls_channel, splunk_events_by_source):
"""Check if an ATT&CK log source (name + channel) matches any Splunk event data.
Returns list of matching (splunk_source, event_id, count) tuples.
"""
matches = []
name_lower = ls_name.lower()
channel = ls_channel.strip() if ls_channel else ''
# Extract EventCode from channel if present
ec_match = re.match(r'EventCode=(\d+(?:,\s*\d+)*)', channel)
event_codes = []
if ec_match:
event_codes = [ec.strip() for ec in ec_match.group(1).split(',')]
for source_key, events in splunk_events_by_source.items():
src_type = normalize_source(source_key)
# WinEventLog:Sysmon -> sysmon source
if 'sysmon' in name_lower and src_type == 'sysmon':
if event_codes:
for ec in event_codes:
if ec in events:
matches.append((source_key, ec, events[ec]))
elif not channel or channel == 'None':
if events:
matches.append((source_key, '*', sum(events.values())))
# WinEventLog:Security -> winsecurity source
elif name_lower == 'wineventlog:security' and src_type == 'winsecurity':
if event_codes:
for ec in event_codes:
if ec in events:
matches.append((source_key, ec, events[ec]))
elif not channel or channel == 'None':
if events:
matches.append((source_key, '*', sum(events.values())))
# WinEventLog:PowerShell
elif 'powershell' in name_lower and src_type == 'powershell':
if event_codes:
for ec in event_codes:
if ec in events:
matches.append((source_key, ec, events[ec]))
# WinEventLog:System
elif name_lower == 'wineventlog:system' and src_type == 'winsystem':
if event_codes:
for ec in event_codes:
if ec in events:
matches.append((source_key, ec, events[ec]))
# AWS CloudTrail
elif ('cloudtrail' in name_lower or name_lower.startswith('aws:')) and src_type == 'cloudtrail':
if channel and channel != 'None':
channel_events = [e.strip() for e in re.split(r'[,;]', channel) if e.strip()]
for ce in channel_events:
if ce in events:
matches.append((source_key, ce, events[ce]))
else:
for eid in events:
if eid.lower().startswith(ce.lower().rstrip('*')):
matches.append((source_key, eid, events[eid]))
break
else:
if events:
matches.append((source_key, '*', sum(events.values())))
# Azure
elif ('azure' in name_lower or 'entra' in name_lower) and src_type == 'azure':
if channel and channel != 'None':
channel_ops = [e.strip() for e in re.split(r'[,;]', channel) if e.strip()]
for co in channel_ops:
if co in events:
matches.append((source_key, co, events[co]))
else:
if events:
matches.append((source_key, '*', sum(events.values())))
# auditd
elif 'auditd' in name_lower and src_type == 'auditd':
if channel and channel != 'None':
for eid in events:
if channel.lower().split('/')[0] in eid.lower():
matches.append((source_key, eid, events[eid]))
break
else:
if events:
matches.append((source_key, '*', sum(events.values())))
return matches
def make_bar(pct, width=20):
filled = round(pct / 100 * width)
return '\u2588' * filled + '\u2591' * (width - filled)
def fmt_num(n):
if n >= 1_000_000:
return f"{n/1_000_000:.1f}M"
if n >= 1_000:
return f"{n/1_000:.1f}K"
return str(n)
def main():
args = parse_args()
# Load data
attack_data = load_attack(args.attack_json)
splunk_events = load_splunk_events(args.splunk_events)
# Build ATT&CK mappings
components, techniques, ds_names = build_mappings(attack_data)
# Organize Splunk events: source -> {event_id: count}
splunk_by_source = defaultdict(dict)
for ev in splunk_events:
splunk_by_source[ev['source']][str(ev['event_id'])] = int(ev['count'])
# Cross-reference each component
for cid, comp in components.items():
comp['covered'] = False
comp['matched_events'] = []
comp['missing_channels'] = []
if not comp['log_sources']:
comp['unmapped'] = True
continue
comp['unmapped'] = False
for ls in comp['log_sources']:
ls_name = ls.get('name', '')
ls_channel = ls.get('channel', '')
matches = match_log_source(ls_name, ls_channel, splunk_by_source)
if matches:
comp['covered'] = True
for m in matches:
comp['matched_events'].append({
'source': m[0], 'event_id': m[1], 'count': m[2],
'attack_log_source': ls_name, 'channel': ls_channel
})
else:
ec_match = re.match(r'EventCode=(\d+)', ls_channel)
if ec_match or ('cloudtrail' in ls_name.lower() and ls_channel and ls_channel != 'None'):
comp['missing_channels'].append({'name': ls_name, 'channel': ls_channel})
# Calculate technique coverage
all_tech_ids = set(techniques.keys())
covered_tech_ids = set()
partial_tech_data = {}
for tid in all_tech_ids:
detecting_comps = [c for c in components.values() if tid in c.get('techniques', set())]
total = len(detecting_comps)
covered = sum(1 for c in detecting_comps if c.get('covered'))
partial_tech_data[tid] = {'total': total, 'covered': covered}
if covered > 0:
covered_tech_ids.add(tid)
fully_covered = sum(1 for t in partial_tech_data.values() if t['total'] > 0 and t['covered'] == t['total'])
partially_covered = sum(1 for t in partial_tech_data.values() if t['total'] > 0 and 0 < t['covered'] < t['total'])
no_coverage = sum(1 for t in partial_tech_data.values() if t['total'] > 0 and t['covered'] == 0)
no_detection = sum(1 for t in partial_tech_data.values() if t['total'] == 0)
uncovered_tech_ids = {tid for tid, d in partial_tech_data.items() if d['covered'] == 0}
# Data source grouping
ds_coverage = defaultdict(lambda: {'total': 0, 'covered': 0, 'techs': set(), 'components': []})
for comp in components.values():
parent = comp.get('parent', 'Other')
ds = ds_coverage[parent]
ds['total'] += 1
if comp.get('covered'):
ds['covered'] += 1
ds['techs'].update(comp.get('techniques', set()))
ds['components'].append(comp)
# Priority gaps
missing_comps = [c for c in components.values() if not c.get('covered') and not c.get('unmapped')]
for mc in missing_comps:
mc['uncovered_added'] = len(mc.get('techniques', set()) & uncovered_tech_ids)
missing_comps.sort(key=lambda x: x['uncovered_added'], reverse=True)
# Matched event detail
matched_events_flat = []
for comp in components.values():
if comp.get('covered'):
for me in comp.get('matched_events', []):
matched_events_flat.append({
'source': me['source'], 'event_id': me['event_id'],
'count': me['count'], 'component': comp['name'],
'parent': comp.get('parent', ''), 'technique_count': comp['technique_count'],
})
# Stats
total_comps = len(components)
covered_comps = sum(1 for c in components.values() if c.get('covered'))
total_techs = len(all_tech_ids)
covered_techs = len(covered_tech_ids)
coverage_pct = round(covered_techs / total_techs * 100) if total_techs else 0
comp_pct = round(covered_comps / total_comps * 100) if total_comps else 0
callout = 'success' if coverage_pct > 60 else ('warning' if coverage_pct > 30 else 'failure')
splunk_sources = list(splunk_by_source.keys())
total_events = sum(sum(v.values()) for v in splunk_by_source.values())
# === Generate Obsidian note ===
lines = []
lines.append('---')
lines.append(f'scan_date: {args.date}')
lines.append(f'splunk_index: "{args.index}"')
lines.append(f'lookback_days: {args.days}')
lines.append(f'total_techniques: {total_techs}')
lines.append(f'covered_techniques: {covered_techs}')
lines.append(f'coverage_pct: {coverage_pct}%')
lines.append(f'total_data_components: {total_comps}')
lines.append(f'matched_data_components: {covered_comps}')
lines.append(f'splunk_sources: {len(splunk_sources)}')
lines.append('---')
lines.append('')
lines.append(f'# ATT&CK Coverage - {args.date}')
lines.append('')
lines.append(f'> [!{callout}] Coverage Summary')
lines.append(f'> **Index:** `{args.index}` | **Lookback:** {args.days} days | **Sources:** {len(splunk_sources)} | **Events:** {fmt_num(total_events)}')
lines.append(f'> **Technique coverage:** {covered_techs}/{total_techs} ({coverage_pct}%)')
lines.append(f'> **Data component coverage:** {covered_comps}/{total_comps} ({comp_pct}%)')
lines.append('')
lines.append('## Overall Coverage')
lines.append('')
lines.append('```')
lines.append(f'Data Components [{make_bar(comp_pct, 40)}] {comp_pct}% {covered_comps}/{total_comps} present')
lines.append(f'Techniques [{make_bar(coverage_pct, 40)}] {coverage_pct}% {covered_techs}/{total_techs} detectable')
lines.append('```')
lines.append('')
lines.append('## Technique Coverage Breakdown')
lines.append('')
lines.append('```')
lines.append('Technique Coverage')
lines.append('\u2500' * 35)
lines.append(f' \u2588\u2588 Fully covered: {fully_covered:>4d} ({round(fully_covered/total_techs*100) if total_techs else 0}%)')
lines.append(f' \u2593\u2593 Partially covered: {partially_covered:>4d} ({round(partially_covered/total_techs*100) if total_techs else 0}%)')
lines.append(f' \u2591\u2591 No coverage: {no_coverage:>4d} ({round(no_coverage/total_techs*100) if total_techs else 0}%)')
lines.append(f' \u00b7\u00b7 No detects: {no_detection:>4d} ({round(no_detection/total_techs*100) if total_techs else 0}%)')
lines.append('\u2500' * 35)
lines.append(f' Total: {total_techs:>4d}')
lines.append('```')
lines.append('')
lines.append('## Data Source Coverage')
lines.append('')
lines.append('```')
lines.append(f'{"Data Source":<25s} {"Bar":20s} {"Pct":>4s} {"Components":<16s} \u2502 Techniques')
lines.append('\u2500' * 85)
sorted_ds = sorted(ds_coverage.items(), key=lambda x: len(x[1]['techs']), reverse=True)
for ds_name, d in sorted_ds:
if not d['techs'] and d['total'] <= 1:
continue
pct = round(d['covered'] / d['total'] * 100) if d['total'] else 0
bar = make_bar(pct, 20)
gap = ' \u2190 GAP' if d['covered'] == 0 and d['total'] > 0 else ''
lines.append(f'{ds_name:<25s} {bar} {pct:>3d}% {d["covered"]}/{d["total"]} components \u2502 {len(d["techs"]):>3d} techniques{gap}')
lines.append('```')
lines.append('')
lines.append('## Event ID Coverage Detail')
lines.append('')
lines.append('Specific event IDs in your SIEM mapped to ATT&CK data components:')
lines.append('')
events_by_source = defaultdict(list)
for me in matched_events_flat:
events_by_source[me['source']].append(me)
for source in sorted(events_by_source.keys()):
events = events_by_source[source]
lines.append(f'### {source}')
lines.append('')
lines.append('| Event ID | Events | Data Component | Techniques |')
lines.append('|----------|--------|---------------|------------|')
seen = set()
for e in sorted(events, key=lambda x: str(x['event_id'])):
key = (e['event_id'], e['component'])
if key in seen:
continue
seen.add(key)
lines.append(f'| {e["event_id"]} | {fmt_num(e["count"])} | {e["component"]} | {e["technique_count"]} |')
lines.append('')
lines.append('## Missing Event IDs')
lines.append('')
lines.append('Event IDs referenced by ATT&CK but not found in your SIEM:')
lines.append('')
lines.append('| Log Source | Event ID / Channel | Data Component | Techniques |')
lines.append('|-----------|-------------------|---------------|------------|')
missing_shown = set()
for comp in sorted(components.values(), key=lambda c: c.get('technique_count', 0), reverse=True):
if comp.get('covered') or comp.get('unmapped'):
continue
for mc in comp.get('missing_channels', []):
key = (mc['name'], mc['channel'], comp['name'])
if key not in missing_shown:
missing_shown.add(key)
lines.append(f'| {mc["name"]} | {mc["channel"]} | {comp["name"]} | {comp["technique_count"]} |')
lines.append('')
lines.append('## Priority Gaps - Biggest Bang for Buck')
lines.append('')
lines.append('Data components not covered that would detect the most currently-undetectable techniques:')
lines.append('')
lines.append('| # | Data Component | Data Source | +Techniques | Example Log Sources |')
lines.append('|---|---------------|-------------|-------------|-------------------|')
for i, mc in enumerate(missing_comps[:15]):
if mc['uncovered_added'] == 0 and mc['technique_count'] == 0:
continue
example_ls = ', '.join(ls.get('name', '') for ls in mc['log_sources'][:3])
lines.append(f'| {i+1} | {mc["name"]} | {mc.get("parent","")} | +{mc["uncovered_added"]} | {example_ls} |')
lines.append('')
unmapped_list = [c for c in components.values() if c.get('unmapped')]
if unmapped_list:
lines.append('## Unmapped Components')
lines.append('')
lines.append('Data components with no log sources defined in ATT&CK (cannot be automatically mapped):')
lines.append('')
lines.append('| Data Component | Data Source | Techniques |')
lines.append('|---------------|-------------|------------|')
for c in sorted(unmapped_list, key=lambda x: x.get('technique_count', 0), reverse=True):
lines.append(f'| {c["name"]} | {c.get("parent","")} | {c["technique_count"]} |')
lines.append('')
print('\n'.join(lines))
if __name__ == '__main__':
main()Then run:
python3 /tmp/attack_coverage.py \
--attack-json /home/anton/Desktop/claudeforblueteam/enterprise-attack.json \
--splunk-events /tmp/splunk_events.json \
--date "$(date +%Y-%m-%d)" \
--index "<INDEX>" \
--days <DAYS> \
> /tmp/attack_coverage_note.mdRead /tmp/attack_coverage_note.md and write it to Obsidian:
obsidian create vault="{VAULT}" name="ATT&CK Coverage — $(date +%Y-%m-%d)" overwrite content="$(cat /tmp/attack_coverage_note.md)"Report to the user:
- Technique coverage: X/Y (Z%)
- Data component coverage: X/Y (Z%)
- Splunk sources found and event ID counts
- Top 3 priority gaps (missing data components that would add the most coverage)
- Confirm note written to Obsidian