Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

Response Plans

A response plan is a sequence: gather, decide, contain, record, notify. Most teams either run it by hand from a wiki page, or buy a platform to run it for them.

There is a middle option. A response plan is a function, and BabySOARus runs functions.

The Shape of a Plan

# 1. Gather
context = enrich(alert)

# 2. Decide
decision = classify(context)

# 3. Act
if decision.contain:
    isolate(context.host)

# 4. Record
case = open_case(context, decision)

# 5. Notify
notify(case, decision)

That is the whole thing, and it fits in an alert action.

A Worked Plan: Suspected Credential Compromise

Triggered by a scheduled search that finds a successful login following sustained failures.

import datetime

# --- configuration -------------------------------------------------------
IDENTITY_API = 'https://identity.example.internal/api/v1'
EDR_API = 'https://edr.example.internal/api/v3'
CASES_API = 'https://cases.example.internal/api/v1'
NEVER_DISABLE = {'svc_backup', 'svc_monitoring', 'breakglass'}

# Secrets come from Splunk's own store, never from the snippet.
def secret(name):
    r = splunk_request(
        'GET',
        f'/servicesNS/nobody/babysoarus/storage/passwords/%3A{name}%3A?output_mode=json')
    return json.loads(r['body'])['entry'][0]['content']['clear_password']

TOKENS = {'identity': secret('identity_token'), 'edr': secret('edr_token'),
          'cases': secret('cases_token')}

def call(api, method, path, token, body=None, timeout=20):
    return http_request(method, f'{api}{path}',
                        headers={'Authorization': f'Bearer {token}',
                                 'Content-Type': 'application/json'},
                        body=body, timeout=timeout)

# --- 1. gather -----------------------------------------------------------
timeline = []
for e in events:
    user, src = e.get('user', ''), e.get('src_ip', '')

    identity = {}
    r = call(IDENTITY_API, 'GET', f'/users/{user}', TOKENS['identity'])
    if r['status'] == 200:
        identity = json.loads(r['body'])

    e['department'] = identity.get('department', 'unknown')
    e['privileged'] = bool(identity.get('privileged'))
    e['manager'] = identity.get('manager_email', '')

    # Where has this account been today?
    r = splunk_request(
        'GET',
        '/services/search/jobs/export?output_mode=json&search='
        f'search index=auth user={user} earliest=-24h '
        '| stats dc(src_ip) as sources values(dest) as systems')
    if r['status'] == 200:
        for line in r['body'].decode().splitlines():
            if line.strip():
                try:
                    result = json.loads(line)['result']
                    e['sources_24h'] = int(result.get('sources', 0))
                    e['systems_touched'] = result.get('systems', [])
                except (KeyError, ValueError):
                    pass

    timeline.append(e)

# --- 2. decide -----------------------------------------------------------
for e in timeline:
    reasons = []
    severity = 'medium'

    if e['privileged']:
        severity = 'critical'
        reasons.append('privileged account')
    if int(e.get('preceding_failures', 0)) > 20:
        reasons.append(f"{e['preceding_failures']} failures before success")
    if int(e.get('sources_24h', 0)) > 5:
        severity = 'critical'
        reasons.append(f"{e['sources_24h']} distinct source addresses in 24h")

    e['severity'] = severity
    e['reasons'] = reasons
    e['contain'] = severity == 'critical' and e['user'] not in NEVER_DISABLE

# --- 3. act --------------------------------------------------------------
for e in timeline:
    e['actions_taken'] = []

    if not e['contain']:
        e['actions_taken'].append('no automatic containment')
        continue

    r = call(IDENTITY_API, 'POST', f"/users/{e['user']}/sessions/revoke",
             TOKENS['identity'])
    e['actions_taken'].append(
        'sessions revoked' if r['status'] < 300 else f"session revoke failed ({r['status']})")

    r = call(IDENTITY_API, 'POST', f"/users/{e['user']}/require-password-reset",
             TOKENS['identity'])
    e['actions_taken'].append(
        'password reset required' if r['status'] < 300 else f"reset failed ({r['status']})")

    for system in e.get('systems_touched', [])[:5]:
        r = call(EDR_API, 'POST', f'/hosts/{system}/scan',
                 TOKENS['edr'], body={'type': 'quick'})
        e['actions_taken'].append(
            f'scan queued on {system}' if r['status'] < 300 else f'scan failed on {system}')

# --- 4. record -----------------------------------------------------------
r = call(CASES_API, 'POST', '/cases', TOKENS['cases'], body={
    'title': f"Credential compromise: {', '.join(sorted({e['user'] for e in timeline}))}",
    'severity': 'critical' if any(e['severity'] == 'critical' for e in timeline) else 'medium',
    'opened': datetime.datetime.utcnow().isoformat() + 'Z',
    'source': 'splunk/BabySOARus',
    'splunk_sid': context.get('sid', ''),
    'findings': [
        {'user': e['user'], 'src_ip': e.get('src_ip'),
         'severity': e['severity'], 'reasons': e['reasons'],
         'actions': e['actions_taken']}
        for e in timeline
    ],
})
case = json.loads(r['body']) if r['status'] < 300 else {'id': 'unfiled', 'url': ''}

# --- 5. notify -----------------------------------------------------------
critical = [e for e in timeline if e['severity'] == 'critical']
if critical:
    http_request('POST', secret('pager_webhook'), body={
        'severity': 'critical',
        'summary': f"{len(critical)} account(s) contained: "
                   f"{', '.join(e['user'] for e in critical)}",
        'links': [{'href': case.get('url', ''), 'text': 'case'}],
    }, timeout=10)

for e in timeline:
    e['case_id'] = case.get('id', '')
    e['case_url'] = case.get('url', '')

events = timeline

Long, but every line is doing something an analyst would otherwise do by hand, and the whole plan is visible at once.

Design Notes

Never contain without an exclusion list. NEVER_DISABLE exists because the first time an automated plan disables a service account, automation gets switched off permanently. Make the list explicit and put it near the top.

Record what you tried, not what you intended. Every action appends its real outcome to actions_taken, including failures. When someone asks what happened at 3am, the answer is in the case.

Degrade rather than fail. An enrichment API being down should not stop containment. Check status codes, carry on with what you have.

Keep secrets out of the snippet. The secret() helper reads from Splunk's password store using the search's own credentials, so the code is safe to commit.

Staged Rollout

Automated containment is a trust exercise. Stage it.

Stage 1: observe. Run the plan with all actions replaced by logging. Compare its decisions against what analysts actually did.

DRY_RUN = True

def act(description, request):
    if DRY_RUN:
        return f'[dry run] would {description}'
    r = request()
    return description if r['status'] < 300 else f'{description} FAILED ({r["status"]})'

Stage 2: reversible actions only. Revoke sessions and require a password reset. Both are annoying and neither is destructive.

Stage 3: containment, with a whitelist. Isolate hosts, but only ones on a list you maintain deliberately.

Stage 4: containment by policy. Isolate anything meeting the criteria, with the exclusion list as the guard.

Most teams stop at stage 2 or 3, and that is a perfectly good place to stop.

Approval Gates

Some steps should need a human. prompt.ask asks and returns immediately -- the plan does not block waiting for an answer, and finishes as if it had not asked at all:

from babysoarus_exec import prompt

def process(events):
    # A resumed run carries the decision in the first event, not a fresh batch.
    if events and 'prompt_response' in events[0]:
        e = events[0]
        approved = e['prompt_response'] == 'yes'
        e['contained'] = approved
        if approved:
            http_request('POST', f'{PROXY}/api/isolate', body={'host': e['host']})
        return [e]

    for e in events:
        if e['severity'] == 'critical' and not e['contain']:
            prompt.ask(f"isolate {e['host']}?", ['yes', 'no'],
                       ['soc-lead'], timeout=1800)
            e['status'] = 'awaiting approval'
    return events

Answering happens later, outside this run, from the Activity view any Splunk user with BabySOARus access already has -- not a second callback endpoint of your own. Once someone in approvers answers, this same function runs again automatically, as a background task, with a fresh batch of one event carrying prompt_response/prompt_responder. timeout (seconds; 0 means none) refuses a late answer rather than accepting it.

This is a real approval gate, not a pattern you assemble from http_request and a second saved search. approvers is a list of usernames (not yet roles or capabilities), checked server side against who is actually answering, never trusted from the browser. The full chain -- who was asked, who answered, when, and what ran as a result -- is recorded and visible in Activity's own Pending/History tabs, not something to build a dashboard for separately.

Testing

Test the plan the way you test a detection: give it known input and check what it decides.

| makeresults
| eval user="d.harrington", src_ip="185.199.110.9", preceding_failures="34"
| exec inline="DRY_RUN = True
..."
| table user severity contain actions_taken

With DRY_RUN = True the plan runs end to end, calls nothing, and prints what it would have done. Run that on a schedule against synthetic input and you have a regression test for your response process.