building-threat-intelligence-enrichment-in-splunk

name	building-threat-intelligence-enrichment-in-splunk
description	Build automated threat intelligence enrichment pipelines in Splunk Enterprise Security using lookup tables, modular inputs, and the Threat Intelligence Framework.
domain	cybersecurity
subdomain	soc-operations
tags	["splunk","threat-intelligence","enrichment","ioc","lookup","siem","soc","enterprise-security"]
version	1.0
author	mahipal
license	Apache-2.0
nist_csf	["DE.CM-01","DE.AE-02","RS.MA-01","DE.AE-06"]

Building Threat Intelligence Enrichment in Splunk

Overview

Splunk's Threat Intelligence Framework in Enterprise Security enables SOC teams to automatically correlate indicators of compromise (IOCs) against security events. The framework ingests threat feeds, normalizes indicators into KV Store collections, and uses lookup-based correlation searches to flag matching events. Splunk Threat Intelligence Management centralizes collection, normalization, and enrichment from multiple sources, reducing triage time by providing analysts with immediate context.

When to Use

• When deploying or configuring building threat intelligence enrichment in splunk capabilities in your environment
• When establishing security controls aligned to compliance requirements
• When building or improving security architecture for this domain
• When conducting security assessments that require this implementation

Prerequisites

• Splunk Enterprise Security (ES) 7.x or later
• Threat Intelligence Management add-on or Threat Intelligence Framework
• API keys for external threat intelligence feeds (MISP, OTX, VirusTotal, AbuseIPDB)
• KV Store enabled and properly configured
• Admin access for modular input configuration

Threat Intelligence Framework Architecture

External TI Sources (STIX/TAXII, CSV, API)
    |
    v
Modular Inputs (download and parse feeds)
    |
    v
KV Store Collections (normalized IOC storage)
    |-- ip_intel
    |-- domain_intel
    |-- file_intel
    |-- url_intel
    |-- email_intel
    |
    v
Threat Intelligence Lookups
    |
    v
Correlation Searches (match events against IOCs)
    |
    v
Notable Events (enriched with TI context)

Configuring Threat Intelligence Sources

STIX/TAXII Feed Integration

# inputs.conf - TAXII feed configuration
[threatlist://taxii_feed_example]
description = TAXII 2.1 Threat Feed
type = taxii
url = https://threatfeed.example.com/taxii2/
collection = threat-indicators-v21
polling_interval = 3600
api_key = <encrypted_api_key>
disabled = false

CSV-Based Threat List

# inputs.conf - CSV threat list
[threatlist://custom_blocklist]
description = Internal threat blocklist
type = csv
url = https://internal.company.com/threat-feeds/blocklist.csv
polling_interval = 1800
disabled = false

Custom Modular Input for API-Based Feeds

# bin/threatfeed_otx.py - OTX AlienVault feed collector
import json
import sys
import requests
 splunklib.modularinput  Script, Scheme, Argument, Event


 ():
     ():
        scheme = Scheme()
        scheme.description = 
        scheme.use_external_validation = 
        scheme.streaming_mode = Scheme.streaming_mode_xml

        api_key_arg = Argument()
        api_key_arg.data_type = Argument.data_type_string
        api_key_arg.required_on_create = 
        scheme.add_argument(api_key_arg)

        pulse_days_arg = Argument()
        pulse_days_arg.data_type = Argument.data_type_number
        pulse_days_arg.required_on_create = 
        scheme.add_argument(pulse_days_arg)

         scheme

     ():
         input_name, input_item  inputs.inputs.items():
            api_key = input_item[]
            pulse_days = (input_item.get(, ))

            headers = {: api_key}
            url = 

            :
                response = requests.get(url, headers=headers, timeout=)
                response.raise_for_status()
                data = response.json()

                 pulse  data.get(, []):
                     indicator  pulse.get(, []):
                        event = Event()
                        event.stanza = input_name
                        event.data = json.dumps({
                            : indicator[],
                            : indicator[],
                            : pulse[],
                            : pulse[],
                            : indicator.get(, ),
                            : indicator.get(, ),
                            : ,
                            : pulse.get(, ),
                        })
                        ew.write_event(event)
             requests.RequestException  e:
                ew.log(, )


 __name__ == :
    sys.exit(OTXFeedInput().run(sys.argv))