building-threat-intelligence-enrichment-in-splunk

在 Splunk 中构建威胁情报富化

概述

Splunk Enterprise Security 中的威胁情报框架使 SOC 团队能够自动将失陷指标（IOC）与安全事件进行关联。该框架会摄取威胁源，将指标规范化为 KV Store 集合，并使用基于查找的关联搜索来标记匹配的事件。Splunk Threat Intelligence Management 集中管理来自多个来源的收集、规范化与富化，通过为分析师提供即时上下文来缩短分类处置时间。

何时使用

• 在您的环境中部署或配置在 Splunk 中构建威胁情报富化功能时
• 在建立符合合规要求的安全控制措施时
• 在为该领域构建或改进安全架构时
• 在开展需要此实现的安全评估时

先决条件

• Splunk Enterprise Security (ES) 7.x 或更高版本
• Threat Intelligence Management 附加组件或威胁情报框架
• 外部威胁情报源（MISP、OTX、VirusTotal、AbuseIPDB）的 API 密钥
• 已启用并正确配置 KV Store
• 用于模块化输入配置的管理员访问权限

威胁情报框架架构

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)

配置威胁情报源

STIX/TAXII 源集成

# 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 的威胁列表

# 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

用于基于 API 的源的自定义模块化输入

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


class OTXFeedInput(Script):
    def get_scheme(self):
        scheme = Scheme("OTX AlienVault Feed")
        scheme.description = "Collects IOCs from AlienVault OTX"
        scheme.use_external_validation = False
        scheme.streaming_mode = Scheme.streaming_mode_xml

        api_key_arg = Argument("api_key")
        api_key_arg.data_type = Argument.data_type_string
        api_key_arg.required_on_create = True
        scheme.add_argument(api_key_arg)

        pulse_days_arg = Argument("pulse_days")
        pulse_days_arg.data_type = Argument.data_type_number
        pulse_days_arg.required_on_create = False
        scheme.add_argument(pulse_days_arg)

        return scheme

    def stream_events(self, inputs, ew):
        for input_name, input_item in inputs.inputs.items():
            api_key = input_item["api_key"]
            pulse_days = int(input_item.get("pulse_days", 30))

            headers = {"X-OTX-API-KEY": api_key}
            url = f"https://otx.alienvault.com/api/v1/pulses/subscribed?modified_since={pulse_days}d"

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

                for pulse in data.get("results", []):
                    for indicator in pulse.get("indicators", []):
                        event = Event()
                        event.stanza = input_name
                        event.data = json.dumps({
                            "indicator": indicator["indicator"],
                            "type": indicator["type"],
                            "pulse_name": pulse["name"],
                            "pulse_id": pulse["id"],
                            "description": indicator.get("description", ""),
                            "created": indicator.get("created", ""),
                            "threat_source": "OTX",
                            "confidence": pulse.get("adversary", "unknown"),
                        })
                        ew.write_event(event)
            except requests.RequestException as e:
                ew.log("ERROR", f"OTX feed collection failed: {str(e)}")


if __name__ == "__main__":
    sys.exit(OTXFeedInput().run(sys.argv))

构建富化查找

KV Store 集合配置

# collections.conf
[ip_threat_intel]
field.ip = string
field.threat_type = string
field.confidence = number
field.source = string
field.description = string
field.first_seen = time
field.last_seen = time
field.severity = string

[domain_threat_intel]
field.domain = string
field.threat_type = string
field.confidence = number
field.source = string
field.whois_registrar = string
field.whois_created = string

[file_hash_intel]
field.file_hash = string
field.hash_type = string
field.malware_family = string
field.confidence = number
field.source = string
field.detection_names = string

查找表定义

# transforms.conf
[ip_threat_intel_lookup]
external_type = kvstore
collection = ip_threat_intel
fields_list = ip, threat_type, confidence, source, description, severity

[domain_threat_intel_lookup]
external_type = kvstore
collection = domain_threat_intel
fields_list = domain, threat_type, confidence, source

[file_hash_intel_lookup]
external_type = kvstore
collection = file_hash_intel
fields_list = file_hash, hash_type, malware_family, confidence, source

富化关联搜索

基于 IP 的威胁情报关联

| tstats summariesonly=true count from datamodel=Network_Traffic
    where All_Traffic.action=allowed
    by All_Traffic.src_ip, All_Traffic.dest_ip, All_Traffic.dest_port, _time span=5m
| rename "All_Traffic.*" as *
| lookup ip_threat_intel_lookup ip as dest_ip OUTPUT threat_type, confidence, source as ti_source, severity as ti_severity
| where isnotnull(threat_type)
| lookup asset_lookup ip as src_ip OUTPUT asset_name, asset_owner, asset_priority
| eval urgency=case(
    ti_severity=="critical" AND asset_priority=="critical", "critical",
    ti_severity=="high" OR asset_priority=="critical", "high",
    ti_severity=="medium", "medium",
    true(), "low"
)
| eval description="Connection from ".src_ip." (".asset_name.") to known malicious IP ".dest_ip." (".threat_type.") - Source: ".ti_source

基于域名的威胁情报关联

index=dns sourcetype=stream:dns query_type=A OR query_type=AAAA
| lookup domain_threat_intel_lookup domain as query OUTPUT threat_type as domain_threat, confidence as domain_confidence, source as ti_source
| where isnotnull(domain_threat) AND domain_confidence > 70
| stats count dc(src_ip) as unique_sources values(src_ip) as source_ips by query, domain_threat, ti_source
| eval severity=case(domain_confidence > 90, "critical", domain_confidence > 70, "high", true(), "medium")
| eval description="DNS queries to malicious domain ".query." from ".unique_sources." hosts - Threat: ".domain_threat

文件哈希关联

index=endpoint sourcetype=sysmon EventCode=1
| lookup file_hash_intel_lookup file_hash as Hashes OUTPUT malware_family, confidence as hash_confidence, source as ti_source
| where isnotnull(malware_family)
| stats count values(ParentCommandLine) as parent_commands by Computer, User, Image, malware_family, ti_source
| eval severity="critical"
| eval description="Known malware ".malware_family." executed on ".Computer." by ".User." - Binary: ".Image

多源富化管道

index=firewall sourcetype=pan:traffic action=allowed
| eval indicators=mvappend(src_ip, dest_ip)
| mvexpand indicators
| lookup ip_threat_intel_lookup ip as indicators OUTPUT threat_type as ip_threat, confidence as ip_confidence, source as ip_ti_source
| lookup geo_ip_lookup ip as indicators OUTPUT country, city, latitude, longitude
| lookup whois_lookup ip as indicators OUTPUT org as ip_org, asn as ip_asn
| where isnotnull(ip_threat)
| stats count
    values(ip_threat) as threat_types
    values(ip_ti_source) as intel_sources
    values(country) as countries
    values(ip_org) as organizations
    latest(_time) as last_seen
    earliest(_time) as first_seen
    by src_ip, dest_ip, dest_port
| eval enrichment_context="Threat: ".mvjoin(threat_types, ", ")." | Geo: ".mvjoin(countries, ", ")." | Org: ".mvjoin(organizations, ", ")

威胁情报仪表板

IOC 覆盖统计

| inputlookup ip_threat_intel_lookup
| stats count by source, threat_type
| sort -count
| head 20

源新鲜度监控

| inputlookup ip_threat_intel_lookup
| eval age_days=round((now() - strptime(last_seen, "%Y-%m-%dT%H:%M:%S")) / 86400, 0)
| stats count avg(age_days) as avg_age_days max(age_days) as max_age_days by source
| eval status=case(avg_age_days > 30, "STALE", avg_age_days > 7, "AGING", true(), "FRESH")

参考资料

• Splunk 威胁情报框架文档
• Splunk Lantern - 威胁情报富化
• 集成情报富化 - Splunk 博客
• Splunk 中的 Cisco Talos 威胁情报