| argument-hint | <Obsidian note name> | |||
|---|---|---|---|---|
| allowed-tools |
|
Read a Splunk query from an Obsidian note, execute it against a Splunk instance, and write the results back into the note with raw results and an enriched process tree.
$ARGUMENTS contains the Obsidian note name (without .md extension).
If no note name is provided, ask the user for one.
SPLUNK_URL -- from environment variable $SPLUNK_URL
SPLUNK_TOKEN -- from environment variable $SPLUNK_TOKEN
VAULT=Obsidian Vault
TREE_SCRIPT -- write splunk-tree.py to /tmp if it doesn't exist (source below)
If either env var is not set, ask the user.
Target: 2 Splunk calls, 1 Obsidian write.
Before running, check if /tmp/splunk-tree.py exists. If not, write it from the embedded source at the bottom of this file.
Read the note and parse the splunk code block:
obsidian read vault="{VAULT}" file="{NOTE_NAME}"Then execute it. Default time range is -1d. Prepend search if the query doesn't start with search or a generating command. Do NOT modify the user's query -- do not add filters like EventCode=1 or change the search terms. Only append a | table with the needed fields:
curl -s -k \
-H "Authorization: Bearer $SPLUNK_TOKEN" \
-d "search=search <QUERY> | table _time EventCode Image ParentImage CommandLine User Computer ProcessId ParentProcessId ProcessGuid ParentProcessGuid CurrentDirectory IntegrityLevel RuleName ParentCommandLine LogonId" \
-d "output_mode=json" \
-d "exec_mode=oneshot" \
-d "earliest_time=-1d" \
-d "latest_time=now" \
"$SPLUNK_URL/services/search/jobs/export" > /tmp/splunk_initial.jsonImportant: If the query already contains | table or | stats or other transforming commands, do NOT append the table command -- run it as-is and skip the process tree (Step 2). Only build a tree when working with raw Sysmon events.
Use splunk-tree.py to extract GUIDs, run one expanded query, then build the tree:
GUID_FILTER=$(python3 /tmp/splunk-tree.py /tmp/splunk_initial.json --guids-only)
curl -s -k \
-H "Authorization: Bearer $SPLUNK_TOKEN" \
-d "search=search index=sysmon ($GUID_FILTER) | table _time EventCode Image SourceImage TargetImage ParentImage CommandLine User ProcessId ProcessGuid ParentProcessId ParentProcessGuid ParentCommandLine DestinationIp DestinationPort DestinationHostname SourceIp SourcePort Protocol QueryName QueryResults TargetFilename Details RuleName CurrentDirectory IntegrityLevel TargetObject LogonId | sort _time" \
-d "output_mode=json" \
-d "exec_mode=oneshot" \
-d "earliest_time=-1d" \
-d "latest_time=now" \
"$SPLUNK_URL/services/search/jobs/export" > /tmp/splunk_expanded.json
TREE=$(python3 /tmp/splunk-tree.py /tmp/splunk_expanded.json)
STATS=$(python3 /tmp/splunk-tree.py /tmp/splunk_expanded.json --stats-only)The output has two sections: raw results table from the initial query, then the enriched process tree from the expanded query.
Format the initial results as a markdown table inside the callout. Use the most relevant fields from the initial query results (e.g., _time, Image, ParentImage, CommandLine, User, ProcessId). Cap at 50 rows.
Then append the process tree below it.
> [!success] Splunk Results -- {YYYY-MM-DD}
> **Query:** `{query}`
> **Time range:** -1d to now
> **Results:** {initial_count} events ({expanded_total} expanded across {types} Sysmon event types)
>
> | Time | Image | Parent | CommandLine | User | PID |
> |------|-------|--------|-------------|------|-----|
> | ... | ... | ... | ... | ... | ... |
### Process Tree
\`\`\`
{tree output}
\`\`\`Build the full output in a temp file, then append in a single obsidian append call. Use > [!success] for results with data, > [!warning] for zero results, > [!failure] for errors. Append below any existing results (don't replace).
Report to the user:
- Note name
- Query executed
- Total events and event types
- Key observations (ATT&CK techniques, network activity, suspicious patterns)
Write this to /tmp/splunk-tree.py at the start of each run if it doesn't exist:
#!/usr/bin/env python3
"""Parse Splunk Sysmon NDJSON and build an enriched process tree."""
import json, sys
from collections import defaultdict
EC = {
'1': 'ProcessCreate', '3': 'NetworkConnect', '5': 'ProcessTerminate',
'7': 'ImageLoad', '10': 'ProcessAccess', '11': 'FileCreate',
'12': 'RegAddDelete', '13': 'RegSetValue', '17': 'PipeCreate',
'22': 'DNSQuery', '23': 'FileDelete', '26': 'FileDeleteDetected',
'29': 'FileExecutableDetected',
}
def load_ndjson(path):
events = []
with open(path) as f:
for line in f:
if line.strip():
obj = json.loads(line)
if 'result' in obj:
events.append(obj['result'])
return events
def find_guid_fields(events):
all_keys = set()
for e in events:
all_keys.update(e.keys())
pg = next((k for k in all_keys if k.lower() == 'processguid'), None)
ppg = next((k for k in all_keys if k.lower() == 'parentprocessguid'), None)
return pg, ppg
def extract_guids(events, pg_field, ppg_field):
guids = set()
for e in events:
if pg_field and e.get(pg_field):
guids.add(e[pg_field])
if ppg_field and e.get(ppg_field):
guids.add(e[ppg_field])
return guids
def parse_tag(rule):
if not rule or rule == '-':
return ''
tids, tnames = [], []
for p in rule.split(','):
p = p.strip()
if '=' not in p:
continue
k, v = p.split('=', 1)
if 'technique_id' in k:
tids.append(v)
elif 'technique_name' in k:
tnames.append(v)
if tnames:
return '[' + ' '.join(tids + tnames) + ']'
return ''
def get_time(e):
t = e.get('_time', '')
parts = t.split(' ')
if len(parts) > 1:
t = parts[1]
if '.' in t:
t = t.split('.')[0]
return t
def basename(path):
return path.split('\\')[-1] if path else ''
def fmt_event(e):
ec = e.get('EventCode', '')
t = get_time(e)
tag = parse_tag(e.get('RuleName', ''))
if tag:
tag = ' ' + tag
if ec == '3':
src = '{}:{}'.format(e.get('SourceIp', ''), e.get('SourcePort', ''))
dst = '{}:{}'.format(e.get('DestinationIp', ''), e.get('DestinationPort', ''))
return ('net', 'NetConnect {} -- {} -> {} ({}){}'.format(t, src, dst, e.get('Protocol', ''), tag))
elif ec == '22':
qr = e.get('QueryResults', '').split(';')[0].replace('::ffff:', '')
return ('dns', 'DNSQuery {} -- {} -> {}{}'.format(t, e.get('QueryName', ''), qr, tag))
elif ec == '7':
return ('ImageLoad', 'ImageLoad {}{}'.format(t, tag))
elif ec == '10':
return ('ProcAccess', 'ProcAccess {}{}'.format(t, tag))
elif ec == '11':
return ('file', 'FileCreate {} -- {}{}'.format(t, basename(e.get('TargetFilename', '')), tag))
elif ec in ('23', '26'):
return ('file', 'FileDelete {} -- {}{}'.format(t, basename(e.get('TargetFilename', '')), tag))
elif ec == '17':
return ('pipe', 'PipeCreate {}{}'.format(t, tag))
elif ec == '5':
return ('term', 'Terminate {}'.format(t))
elif ec == '12':
return ('reg', 'RegChange {} -- {}{}'.format(t, e.get('TargetObject', ''), tag))
elif ec == '13':
return ('reg', 'RegSet {} -- {}{}'.format(t, e.get('TargetObject', ''), tag))
elif ec == '29':
return ('file', 'FileExeDet {} -- {}{}'.format(t, basename(e.get('TargetFilename', '')), tag))
else:
return ('other', 'Event{} {}{}'.format(ec, t, tag))
def collapse(formatted):
out = []
i = 0
while i < len(formatted):
etype, text = formatted[i]
if etype in ('ImageLoad', 'ProcAccess'):
count = 1
while i + count < len(formatted) and formatted[i + count][0] == etype:
count += 1
out.append('{} (x{})'.format(text, count) if count > 1 else text)
i += count
else:
out.append(text)
i += 1
return out
def build_tree(events, pg_field, ppg_field):
by_guid = defaultdict(list)
proc_info = {}
for e in events:
guid = e.get(pg_field, '') if pg_field else ''
if guid:
by_guid[guid].append(e)
if e.get('EventCode', '') == '1' and guid and guid not in proc_info:
proc_info[guid] = e
children = defaultdict(list)
for guid in proc_info:
ppguid = proc_info[guid].get(ppg_field, '') if ppg_field else ''
if ppguid:
children[ppguid].append(guid)
for ppg in children:
children[ppg].sort(key=lambda g: proc_info.get(g, {}).get('_time', ''))
roots = sorted(
[g for g in proc_info if (proc_info[g].get(ppg_field, '') if ppg_field else '') not in proc_info],
key=lambda g: proc_info[g].get('_time', ''))
if not roots:
return "No process creation events found.", {}
first = proc_info[roots[0]]
host = first.get('Computer', '')
user = first.get('User', '')
integrity = first.get('IntegrityLevel', '')
logon = first.get('LogonId', '')
cwd = first.get('CurrentDirectory', '')
lines = []
lines.append(host)
lines.append('{} -- {} Integrity -- Logon {}'.format(user, integrity, logon))
lines.append('CWD: {}'.format(cwd))
lines.append('')
def print_proc(guid, prefix='', is_last=True):
info = proc_info[guid]
img = basename(info.get('Image', ''))
pid = info.get('ProcessId', '?')
cmd = info.get('CommandLine', '')
t = get_time(info)
tag = parse_tag(info.get('RuleName', ''))
conn = '`--' if is_last else '|--'
ext = ' ' if is_last else '| '
lines.append('{}{}{} ({}) @ {} -- {}'.format(prefix, conn, img, pid, t, cmd))
if tag:
lines.append('{}{} {}'.format(prefix, ext, tag))
non_proc = [e for e in by_guid.get(guid, []) if e.get('EventCode', '') != '1']
if non_proc:
formatted = [fmt_event(e) for e in non_proc]
collapsed_lines = collapse(formatted)
if collapsed_lines:
lines.append('{}{}'.format(prefix, ext))
for el in collapsed_lines:
lines.append('{}{} {}'.format(prefix, ext, el))
kids = children.get(guid, [])
if kids:
lines.append('{}{}'.format(prefix, ext))
for i, cg in enumerate(kids):
print_proc(cg, prefix + ext, i == len(kids) - 1)
for ri, rg in enumerate(roots):
info = proc_info[rg]
pimg = basename(info.get('ParentImage', ''))
ppid = info.get('ParentProcessId', '?')
pcmd = info.get('ParentCommandLine', '')
lines.append('{} ({}) -- {}'.format(pimg, ppid, pcmd))
lines.append('|')
print_proc(rg, '', True)
if ri < len(roots) - 1:
lines.append('')
ec_counts = defaultdict(int)
for e in events:
ec_counts[EC.get(e.get('EventCode', ''), e.get('EventCode', ''))] += 1
return '\n'.join(lines), ec_counts
def main():
if len(sys.argv) < 2:
print("Usage: splunk-tree.py <ndjson_file> [--guids-only] [--stats-only]", file=sys.stderr)
sys.exit(1)
path = sys.argv[1]
events = load_ndjson(path)
if not events:
print("No events found.", file=sys.stderr)
sys.exit(1)
pg_field, ppg_field = find_guid_fields(events)
if '--guids-only' in sys.argv:
guids = extract_guids(events, pg_field, ppg_field)
field_name = pg_field or 'ProcessGuid'
parts = ['{}="{}"'.format(field_name, g) for g in sorted(guids)]
print(' OR '.join(parts))
sys.exit(0)
if '--stats-only' in sys.argv:
ec_counts = defaultdict(int)
for e in events:
ec_counts[EC.get(e.get('EventCode', ''), e.get('EventCode', ''))] += 1
print(json.dumps({'total': len(events), 'types': len(ec_counts), 'counts': dict(ec_counts)}))
sys.exit(0)
tree, ec_counts = build_tree(events, pg_field, ppg_field)
print(tree)
print('\n--- Stats ---', file=sys.stderr)
print('Total events: {}'.format(len(events)), file=sys.stderr)
print('Event types: {}'.format(len(ec_counts)), file=sys.stderr)
for k, v in sorted(ec_counts.items(), key=lambda x: -x[1]):
print(' {}: {}'.format(k, v), file=sys.stderr)
if __name__ == '__main__':
main()