Detections

• Classify, Do Not Just Threshold
• Score with Reasons Attached
• Enrich While You Detect
• Sequences and State
• Suppression That Survives Review
• Decode What Is Hiding
• Turning It into an Alert
• Testing a Detection

Patterns for writing detection rules where the logic lives in Python and the data reduction stays in SPL.

Every example runs against the babysoarus_demo index from scripts/seed-demo-data.sh, and every one ships with the app: they are the built-in entries under detections/ in the editor's file tree, there to read, run and copy.

Classify, Do Not Just Threshold

A threshold tells you something crossed a line. A classification tells you what happened, which is what the analyst needs.

index=babysoarus_demo sourcetype=babysoarus:auth action=failure
| stats dc(user) as users_targeted count as attempts values(user) as targets by src_ip
| exec inline="for e in events:
    users = int(e['users_targeted'])
    attempts = int(e['attempts'])
    ratio = attempts / users if users else attempts

    e['attempts_per_user'] = round(ratio, 2)
    if users >= 8 and ratio < 6:
        e['pattern'] = 'password spray'
        e['severity'] = 'high'
        e['action'] = 'block source, force reset for targeted accounts'
    elif ratio >= 6:
        e['pattern'] = 'brute force'
        e['severity'] = 'medium'
        e['action'] = 'lock the targeted account, check for success events'
    else:
        e['pattern'] = 'noise'

events = [e for e in events if e['pattern'] != 'noise']"
| table src_ip pattern severity users_targeted attempts attempts_per_user action

Many accounts with few attempts each is spraying. One account with many attempts is brute force. Same source data, different response, and the rule says which.

Score with Reasons Attached

Numeric scores age badly because nobody remembers what went into them. Emit the reasons alongside.

index=babysoarus_demo sourcetype=babysoarus:proxy
| stats sum(bytes_out) as bytes_out count as requests
        values(http_user_agent) as agents by host, user, dest_host
| exec inline="import math

def entropy(text):
    counts = {}
    for ch in text:
        counts[ch] = counts.get(ch, 0) + 1
    return -sum((c/len(text)) * math.log2(c/len(text)) for c in counts.values())

SUSPICIOUS_AGENTS = ('python-requests', 'curl/', 'powershell')

out = []
for e in events:
    score, reasons = 0, []
    domain = e['dest_host']

    h = entropy(domain)
    if h > 3.6:
        score += 30; reasons.append(f'entropy {h:.2f}')

    longest = max(len(l) for l in domain.split('.'))
    if longest > 15:
        score += 25; reasons.append(f'{longest}-character label')

    mb = int(e['bytes_out']) / 1048576
    if mb > 50:
        score += 30; reasons.append(f'{mb:.0f} MB outbound')

    agents = e['agents'] if isinstance(e['agents'], list) else [e['agents']]
    if any(a.lower().startswith(SUSPICIOUS_AGENTS) for a in agents):
        score += 20; reasons.append('scripted user agent')

    if score >= 50:
        e.update(risk_score=score, reasons=reasons,
                 entropy=round(h, 2), megabytes_out=round(mb, 1))
        out.append(e)

events = out"
| sort - risk_score
| table host user dest_host risk_score reasons megabytes_out entropy

The reasons field arrives in Splunk as a multivalue field, so it renders as a readable list in the results and in the alert email.

Enrich While You Detect

Fetch context in the same search rather than maintaining a separate lookup refresh job. Call the API once per batch, never once per event.

index=babysoarus_demo sourcetype=babysoarus:proxy
| stats count by dest_host
| exec inline="
domains = [e['dest_host'] for e in events]

# One request for the whole batch.
r = http_request(
    'POST',
    'https://intel.example.internal/api/v2/bulk',
    headers={'Authorization': 'Bearer ' + context.get('intel_token', ''),
             'Content-Type': 'application/json'},
    body={'indicators': domains},
    timeout=15,
)

verdicts = {}
if r['status'] == 200:
    verdicts = {v['indicator']: v for v in json.loads(r['body'])['results']}

for e in events:
    verdict = verdicts.get(e['dest_host'], {})
    e['intel_verdict'] = verdict.get('verdict', 'unknown')
    e['intel_confidence'] = verdict.get('confidence', 0)
    e['first_seen'] = verdict.get('first_seen', '')

events = [e for e in events if e['intel_verdict'] in ('malicious', 'suspicious')]"

If the API is down, r['status'] is not 200, verdicts stays empty, and everything is marked unknown rather than the search failing. Decide deliberately whether an enrichment outage should fail the detection or degrade it.

Sequences and State

Some detections need to remember what came before. The namespace persists across micro-batches within a search, so it can.

index=babysoarus_demo sourcetype=babysoarus:auth
| sort 0 _time
| exec inline="
try:
    history
except NameError:
    history = {}
    findings = []

for e in events:
    user = e.get('user', '')
    state = history.setdefault(user, {'failures': 0, 'sources': set()})

    if e.get('action') == 'failure':
        state['failures'] += 1
        state['sources'].add(e.get('src_ip', ''))
    elif e.get('action') == 'success' and state['failures'] >= 10:
        findings.append({
            'user': user,
            'src_ip': e.get('src_ip', ''),
            'preceding_failures': state['failures'],
            'distinct_sources': len(state['sources']),
            'pattern': 'successful login after sustained failures',
            'severity': 'critical',
        })
        state['failures'] = 0

events = findings"

sort 0 _time matters: the events must reach the snippet in order for the sequence to mean anything.

Under distributed search this state is per indexer. When the sequence must span the whole estate, aggregate with stats first and run the snippet on the search head.

Suppression That Survives Review

Allow-lists written as SPL NOT clauses become unreadable. Written as data, they stay legible.

| exec inline="
SUPPRESSIONS = [
    # Vulnerability scanner, expected to touch everything.
    {'field': 'src_ip', 'equals': '10.20.0.51', 'owner': 'infra', 'expires': '2026-12-31'},
    # Backup service, high volume by design.
    {'field': 'user', 'equals': 'svc_backup', 'owner': 'platform', 'expires': '2026-09-30'},
    # First-party CDN.
    {'field': 'dest_host', 'endswith': '.cdn.example.com', 'owner': 'web', 'expires': '2027-01-31'},
]

import datetime
today = datetime.date.today().isoformat()

def suppressed(event):
    for rule in SUPPRESSIONS:
        if rule['expires'] < today:
            continue
        value = event.get(rule['field'], '')
        if 'equals' in rule and value == rule['equals']:
            return rule
        if 'endswith' in rule and value.endswith(rule['endswith']):
            return rule
    return None

kept = []
for e in events:
    rule = suppressed(e)
    if rule is None:
        kept.append(e)
    else:
        e['suppressed_by'] = rule['owner']
events = kept"

Each suppression has an owner and an expiry, and expired ones stop applying on their own. That is difficult to express in SPL and trivial here.

Decode What Is Hiding

Attackers encode. Base64 in a command line, hex in a DNS label, URL encoding in a proxy log.

index=babysoarus_demo sourcetype=babysoarus:proxy
| exec inline="import base64, binascii, urllib.parse, re

B64 = re.compile(r'[A-Za-z0-9+/]{20,}={0,2}')
KEYWORDS = ('powershell', 'invoke-', 'downloadstring', '/bin/sh', 'certutil')

out = []
for e in events:
    url = urllib.parse.unquote(e.get('uri', ''))
    decoded = []
    for candidate in B64.findall(url):
        try:
            text = base64.b64decode(candidate + '===').decode('utf-8', 'ignore')
        except (binascii.Error, ValueError):
            continue
        if any(k in text.lower() for k in KEYWORDS):
            decoded.append(text[:200])
    if decoded:
        e['decoded_payloads'] = decoded
        e['severity'] = 'critical'
        out.append(e)

events = out"

Turning It into an Alert

Save the search, schedule it, and attach the alert action if you want code to run over the results too:

[Exfiltration to high-entropy domain]
search = index=proxy | stats ... | exec inline="..."
cron_schedule = */15 * * * *
dispatch.earliest_time = -15m
dispatch.latest_time = now
enableSched = 1
counttype = number of events
relation = greater than
quantity = 0
action.execalert = 1
action.execalert.param.inline = for e in events: e['case_priority'] = 'P1' if int(e['risk_score']) > 70 else 'P2'

Testing a Detection

Feed it known-bad data and assert on the verdict:

| makeresults
| eval dest_host="kq3n8vhs2wpxl4td.exfil-node.top", bytes_out="119681422", requests="18"
| exec inline="..."
| eval test_passed = if(risk_score >= 50, "pass", "FAIL")

Put a handful of those in a saved search and you have a regression test that runs on a schedule and tells you when a detection has stopped working.