Example adaptive response action

This example implements an adaptive response action using the Have I been pwned API. For step-by-step instructions on creating an adaptive response action like this one, see Create an adaptive response action.

This action:

  • Creates a logging instance
  • Handles validation
  • Retrieves the payload
  • Retrieves results
  • Iterates on the results
  • Creates introspection events using the message() method
  • Creates information gathering events using the addevent() and writeevents() methods
  • Handles exceptions

This response action has one global setting and three parameters that a user needs to fill in when she selects and configures the response action. It also has some parameters that are required by the API but not exposed to end users.

For more information about the API, see https://haveibeenpwned.com/API/v2.


File structure for the example






Note: In this example, the developer of the action chose to predetermine an index to store response events created by the action. As a result of this decision, the action contains a default/indexes.conf file. To install and use this response action, the Splunk admin needs to set up this index on the search heads and indexers for type-ahead functionality and configure the storage, retention, and role access.


Python file for the example

haveibeenpwned.py implements a response action.

## Minimal set of standard modules to import
import csv      ## Result set is in CSV format
import gzip     ## Result set is gzipped
import json     ## Payload comes in JSON format
import logging  ## For specifying log levels
import sys      ## For appending the library path

## Standard modules specific to this action
import requests ## For making http based API calls
import urllib   ## For url encoding
import time     ## For rate limiting

## Importing the cim_actions.py library
## A.  Import make_splunkhome_path
## B.  Append your library path to sys.path
## C.  Import ModularAction from cim_actions
from splunk.clilib.bundle_paths import make_splunkhome_path
sys.path.append(make_splunkhome_path(["etc", "apps", "TA-haveibeenpwned", "lib"]))
from cim_actions import ModularAction

## Retrieve a logging instance from ModularAction
## It is required that this endswith _modalert
logger = ModularAction.setup_logger('haveibeenpwned_modalert')

## Subclass ModularAction for purposes of implementing
## a script specific dowork() method
class PwnedModularAction(ModularAction):
    ## Define a list of valid services
    VALID_SERVICES = ['breachedaccount', 'breach', 'pasteaccount']

    ## This method will initialize PwnedModularAction
    def __init__(self, settings, logger, action_name=None):
        ## Call ModularAction.__init__
        super(PwnedModularAction, self).__init__(settings, logger, action_name)
        ## Initialize param.limit
            self.limit = int(self.configuration.get('limit', 1))
            if self.limit<1 or self.limit>30:
                self.limit = 30
            self.limit = 1

    ## This method will handle validation
    def validate(self, result):
        ## outer validation
        if len(self.rids)<=1:
            ## Validate param.url
            if not self.configuration.get('url'):
                raise Exception('Invalid URL requested')
            ## Validate param.service
            if (self.configuration.get('service', '')
               not in PwnedModularAction.VALID_SERVICES):
                raise Exception('Invalid service requested')
            ## Validate param.parameter_field
            if self.configuration.get('parameter_field', '') not in result:
                raise Exception('Parameter field does not exist in result')

    ## This method will do the actual work itself
    def dowork(self, result):
        ## get parameter value
        parameter  = result[self.configuration.get('parameter_field')]
        ## get service
        service    = self.configuration.get('service', '')
        ## build sourcetype
        sourcetype = 'haveibeenpwned:' + service 
        ## build url
        url        = '%s/%s/%s' % (self.configuration.get('url'),
        ## set user-agent
        ua        = 'Pwnage-Checker-For-Splunk'
        ## build headers
        headers   = {'user-agent': ua}
        ## make request
        r = requests.get(url, headers=headers)
        ## process successful request
        if r.status_code==200:
            ## haveibeenpwned returns an array of pwnage objects
            ## one splunk event per object will be created
            ## also inject query_parameter
                json.dumps(dict(x, **{'query_parameter': parameter})),
             for x in json.loads(r.text)]
            self.message('Successfully queried for pwnage', status='success')
        ## process unsuccessful requests
            self.message('Failed to query for pwnage',

if __name__ == "__main__":
    ## This is standard chrome for validating that
    ## the script is being executed by splunkd accordingly
    if len(sys.argv) < 2 or sys.argv[1] != "--execute":
        print >> sys.stderr, "FATAL Unsupported execution mode (expected --execute flag)"

    ## The entire execution is wrapped in an outer try/except
        ## Retrieve an instanced of PwnedModularAction and name it modaction
        ## pass the payload (sys.stdin) and logging instance
        modaction = PwnedModularAction(sys.stdin.read(), logger, 'haveibeenpwned')
        ## Process the result set by opening results_file with gzip
        with gzip.open(modaction.results_file, 'rb') as fh:
            ## Iterate the result set using a dictionary reader
            ## We also use enumerate which provides "num" which
            ## can be used as the result ID (rid)
            for num, result in enumerate(csv.DictReader(fh)):
                ## results limiting
                if num>=modaction.limit:
                ## Set rid to row # (0->n) if unset
                result.setdefault('rid', str(num))
                ## Update the ModularAction instance
                ## with the current result.  This sets
                ## orig_sid/rid/orig_rid accordingly.
                ## Generate an invocation message for each result.
                ## Tells splunkd that we are about to perform the action
                ## on said result.
                ## Validate the invocation
                ## This is where we do the actual work.  In this case
                ## we are calling out to an external API and creating
                ## events based on the information returned
                ## rate limiting
        ## Once we're done iterating the result set and making 
        ## the appropriate API calls we will write out the events
        modaction.writeevents(index='haveibeenpwned', source='haveibeenpwned')

    ## This is standard chrome for outer exception handling
    except Exception as e:
        ## adding additional logging since adhoc search invocations do not write to stderr
            modaction.message(e, status='failure', level=logging.CRITICAL)
        print >> sys.stderr, "ERROR: %s" % e


Configuration files for the example

The haveibeenpwned example response action contains the following configuration files in the default directory.

File Description
alert_actions.conf Defines the parameters of the action.
app.conf Provides package and UI information about the add-on.
eventtypes.conf Defines an event type for the results produced by the action.
tags.conf Tags the results produced by the action.
restmap.conf Defines validation for the per-action parameters declared in savedsearches.conf.
setup.xml Defines the setup UI for the global parameter.
indexes.conf Defines an index for the result events produced by the action.


is_custom             = 1
label                 = haveibeenpwned
description           = Queries the haveibeenpwned API
icon_path             = haveibeenpwned.png
payload_format        = json

param._cam            = {\
    "category":   ["Information Gathering"],\
    "task":       ["scan"],\
    "subject":    ["user", "site"],\
    "technology": [{"vendor": "haveibeenpwned.com", "product": "API", "version": "2"}],\
    "supports_adhoc": true\

param.url             = https://haveibeenpwned.com/api/v2/
param.service         = breachedaccount
param.parameter_field =
param.limit           = 1
param.verbose         = false

ttl                   = 240
command               = sendalert $action_name$ results_file="$results.file$" results_link="$results.url$" param.action_name=$action_name$ | stats count


is_configured = 0
state = enabled
build = 1

author = Someone
version = 1.0.0
description = Adaptive response action using the Have I been pwned REST API. 

is_visible = 0
label = Haveibeenpwned response action

id = TA-haveibeenpwned


search = index=haveibeenpwned sourcetype=haveibeenpwned:*


modaction_result = enabled


action.haveibeenpwned.param.service = validate( 'action.haveibeenpwned.param.service'="breachedaccount" OR 'action.haveibeenpwned.param.service'="breach" OR 'action.haveibeenpwned.param.service'="pasteaccount", "Pwned service is invalid")
action.haveibeenpwned.param.limit = validate( isint('action.haveibeenpwned.param.limit') AND 'action.haveibeenpwned.param.limit'>=1 AND 'action.haveibeenpwned.param.limit'<=30, "Pwned limit is invalid")


homePath   = $SPLUNK_DB/haveibeenpwneddb/db
coldPath   = $SPLUNK_DB/haveibeenpwneddb/colddb
thawedPath = $SPLUNK_DB/haveibeenpwneddb/thaweddb

Spec files for the example

The haveibeenpwned example response action contains the following spec files in the README directory.



param.url             = <string>
   * The API url.
   * See "Via the URL" in API docs.
   * Required.
   * Defaults to "https://haveibeenpwned.com/api/v2/".

param.service         = <string>
   * The API service.
   * One of breachedaccount, breach, pasteaccount
   * Required.
   * Defaults to "breachedaccount".

param.parameter_field = <string>
   * The field which houses the parameter value.
   * Required.
   * Defaults to None.

param.limit           = <int>
   * The maximum number of requests to make.
   * Required.
   * Must be >=1 and <=30
   * Defaults to One.

param.verbose         = <bool>
   * Set modular alert action logger to verbose mode
   * Defaults to "false"


param.service         = <string>
   * The API service.
   * One of breachedaccount, breach, pasteaccount
   * Required.
   * Defaults to "breachedaccount".

param.parameter_field = <string>
   * The field which houses the parameter value.
   * Required.
   * Defaults to None.

param.limit           = <int>
   * The maximum number of requests to make.
   * Required.
   * Must be >=1 and <=30
   * Defaults to 1.

HTML file for the response action form

<form class="form-horizontal form-complex">
<div class="control-group">
        <label class="control-label" for="service">Pwned Service</label>

        <div class="controls">
            <select class="" name="action.haveibeenpwned.param.service" id="service">
                <option value="breachedaccount">Breached Account</option>
                <option value="breach">Breach</option>
                <option value="pasteaccount">Paste Account</option>
    <div class="control-group">
        <label class="control-label" for="parameter_field">Parameter Field</label>

        <div class="controls">
            <input type="text" class="input-xlarge" name="action.haveibeenpwned.param.parameter_field" id="parameter_field" placeholder="i.e. user" />
    <div class="control-group">
        <label class="control-label" for="limit">Limit</label>

        <div class="controls">
            <input type="text" class="input-xlarge" name="action.haveibeenpwned.param.limit" id="limit" placeholder="i.e. 1-30" />


Searches to test the response action

Test the adaptive response action directly, run this search.

| makeresults | eval user=<insert a valid email address here> | sendalert haveibeenpwned param.parameter_field=user

To test ad hoc invocation, run this search to create a notable event.

| makeresults | eval user=<insert a valid email address here> | sendalert notable

Then, go to Incident Review to trigger the response action from the notable event that you just created.


Sample correlation search designed to assign risk scores based on results from the haveibeenpwned action

The following is a sample event generated by haveibeenpwned.py when the action returns a result.

	{	[-]	
   AddedDate: 2016-05-21T21:35:40Z	
   BreachDate: 2012-05-05	
   DataClasses: [	[+]	
   Description: In May 2016, LinkedIn had 164 million email addresses and passwords exposed. Originally hacked in 2012, the data remained out of sight until being offered for sale on a dark market site 4 years later. The passwords in the breach were stored as SHA1 hashes without salt, the vast majority of which were quickly cracked in the days following the release of the data.	
   Domain: linkedin.com	
   IsActive: true	
   IsRetired: false	
   IsSensitive: false	
   IsVerified: true	
   LogoType: svg	
   Name: LinkedIn	
   PwnCount: 164611595	
   Title: LinkedIn	
   query_parameter: sometestemail@example.com	

You can design a correlation search to search for the expected content of this response and trigger additional actions. For example, the following correlation search adjusts the risk score of the object queried in the original action conditionally based on how many results the API returned from the Breached Account service. If only a single breach result is associated with the object, the risk score goes up by 10 points. If six breach results are associated with the same object, the risk score goes up by 60 points.

## Threat - Pwnage Detected - Rule breakdown
## This example illustrates the creation of risk modifiers
## based on haveibeenpwned events
## 1  - search index=haveibeenpwned
## 2  - set up risk_object and risk_object_type fields
## 2a - set risk_object as query_parameter
## 2b - set risk_object_type based on sourcetype
## 3  - perform a count based on orig_sid,orig_rid,risk_object,risk_object_type
## 3a - persisting orig_sid/orig_rid allows us to link the risk and haveibeenpwned
##      invocations
## 4  - calculate risk score based on 10x the number of events
## 5  - remove any other fields we don't need
[Threat - Pwnage Detected - Rule]
action.risk                         = 1
## action.risk.param._risk attributes are irrelevent here
## since we are using the search language to set these fields
action.risk.param._risk_score       = 1
action.risk.param._risk_object      = nosuchfield
action.risk.param._risk_object_type = other
action.risk.param.verbose           = 0
alert.suppress                      = 0
alert.track                         = 0
cron_schedule                       = */30 * * * *
dispatch.earliest_time              = -35m@m
dispatch.latest_time                = -5m@m
dispatch.rt_backfill                = 1
enableSched                         = 1
search                              = index="haveibeenpwned" | eval risk_object=query_parameter,risk_object_type=if(sourcetype LIKE "%account","user","system") | stats count by orig_sid,orig_rid,risk_object,risk_object_type | eval risk_score=count*10 | fields risk*,orig_sid,orig_rid