building-threat-intelligence-enrichment-in-splunk
使用查询表、模块化输入和威胁情报框架，在 Splunk Enterprise Security 中构建自动化威胁情报富化流水线
9 stars
bykillvxk
Best use case
building-threat-intelligence-enrichment-in-splunk is best used when you need a repeatable AI agent workflow instead of a one-off prompt.
使用查询表、模块化输入和威胁情报框架，在 Splunk Enterprise Security 中构建自动化威胁情报富化流水线
Teams using building-threat-intelligence-enrichment-in-splunk should expect a more consistent output, faster repeated execution, less prompt rewriting.
When to use this skill
- You want a reusable workflow that can be run more than once with consistent structure.
When not to use this skill
- You only need a quick one-off answer and do not need a reusable workflow.
- You cannot install or maintain the underlying files, dependencies, or repository context.
Installation
Claude Code / Cursor / Codex
$curl -o ~/.claude/skills/building-threat-intelligence-enrichment-in-splunk/SKILL.md --create-dirs "https://raw.githubusercontent.com/killvxk/cybersecurity-skills-zh/main/skills/building-threat-intelligence-enrichment-in-splunk/SKILL.md"
Manual Installation
- Download SKILL.md from GitHub
- Place it in
.claude/skills/building-threat-intelligence-enrichment-in-splunk/SKILL.mdinside your project - Restart your AI agent — it will auto-discover the skill
How building-threat-intelligence-enrichment-in-splunk Compares
| Feature / Agent | building-threat-intelligence-enrichment-in-splunk | Standard Approach |
|---|---|---|
| Platform Support | Not specified | Limited / Varies |
| Context Awareness | High | Baseline |
| Installation Complexity | Unknown | N/A |
Frequently Asked Questions
What does this skill do?
使用查询表、模块化输入和威胁情报框架，在 Splunk Enterprise Security 中构建自动化威胁情报富化流水线
Where can I find the source code?
You can find the source code on GitHub using the link provided at the top of the page.
SKILL.md Source
# 在 Splunk 中构建威胁情报富化
## 概述
Splunk Enterprise Security 中的威胁情报（Threat Intelligence）框架使 SOC 团队能够自动将失陷指标（IOC）与安全事件进行关联。该框架摄取威胁情报源，将指标规范化存储到 KV Store 集合中，并通过基于查询表的关联搜索标记匹配事件。Splunk 威胁情报管理集中整合来自多个来源的收集、规范化和富化流程，为分析师提供即时上下文，从而缩短分诊时间。
## 前置条件
- Splunk Enterprise Security（ES）7.x 或更高版本
- 威胁情报管理插件或威胁情报框架
- 外部威胁情报源的 API 密钥（MISP、OTX、VirusTotal、AbuseIPDB）
- KV Store 已启用并正确配置
- 用于模块化输入配置的管理员权限
## 威胁情报框架架构
```
外部 TI 来源（STIX/TAXII、CSV、API）
|
v
模块化输入（下载并解析情报源）
|
v
KV Store 集合（规范化 IOC 存储）
|-- ip_intel
|-- domain_intel
|-- file_intel
|-- url_intel
|-- email_intel
|
v
威胁情报查询表
|
v
关联搜索（将事件与 IOC 匹配）
|
v
Notable 事件（已富化 TI 上下文）
```
## 配置威胁情报来源
### STIX/TAXII 情报源集成
```conf
# inputs.conf - TAXII 情报源配置
[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 的威胁列表
```conf
# inputs.conf - CSV 威胁列表
[threatlist://custom_blocklist]
description = 内部威胁封锁列表
type = csv
url = https://internal.company.com/threat-feeds/blocklist.csv
polling_interval = 1800
disabled = false
```
### 基于 API 的情报源自定义模块化输入
```python
# bin/threatfeed_otx.py - OTX AlienVault 情报源采集器
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 情报源")
scheme.description = "从 AlienVault OTX 采集 IOC"
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 情报源采集失败：{str(e)}")
if __name__ == "__main__":
sys.exit(OTXFeedInput().run(sys.argv))
```
## 构建富化查询表
### KV Store 集合配置
```conf
# 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
```
### 查询表定义
```conf
# 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 的威胁情报关联
```spl
| 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="来自 ".src_ip." (".asset_name.") 向已知恶意 IP ".dest_ip." (".threat_type.") 发起连接 - 来源：".ti_source
```
### 基于域名的威胁情报关联
```spl
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="来自 ".unique_sources." 台主机的 DNS 查询指向恶意域名 ".query." - 威胁类型：".domain_threat
```
### 文件哈希关联
```spl
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="已知恶意软件 ".malware_family." 在 ".Computer." 上由 ".User." 执行 - 二进制：".Image
```
## 多来源富化流水线
```spl
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="威胁：".mvjoin(threat_types, ", ")." | 地理位置：".mvjoin(countries, ", ")." | 机构：".mvjoin(organizations, ", ")
```
## 威胁情报仪表板
### IOC 覆盖率统计
```spl
| inputlookup ip_threat_intel_lookup
| stats count by source, threat_type
| sort -count
| head 20
```
### 情报源新鲜度监控
```spl
| 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, "过期", avg_age_days > 7, "老化中", true(), "新鲜")
```
## 参考资料
- [Splunk 威胁情报框架文档](https://help.splunk.com/en/splunk-enterprise-security-8/administer/8.2/threat-intelligence/overview-of-threat-intelligence-in-splunk-enterprise-security)
- [Splunk Lantern - 威胁情报富化](https://lantern.splunk.com/Security/UCE/Guided_Insights/Threat_intelligence)
- [集成智能富化 - Splunk 博客](https://www.splunk.com/en_us/blog/security/integrated-intelligence-enrichment-with-threat-intelligence-management.html)
- [Cisco Talos 威胁情报在 Splunk 中的应用](https://www.splunk.com/en_us/blog/security/cisco-talos-threat-intelligence-splunk-security.html)Related Skills
triaging-security-alerts-in-splunk
9
from killvxk/cybersecurity-skills-zh
在 Splunk Enterprise Security 中对安全告警进行分类，通过 SPL 查询和事件审查（Incident Review） 仪表板对重要事件进行严重性分类、调查、关联相关遥测并做出升级或关闭决策。 适用于 SOC 分析师需要处理关联搜索产生的告警队列、确定调查优先级， 或需要为交接给二/三级分析师记录分类决策时。
tracking-threat-actor-infrastructure
9
from killvxk/cybersecurity-skills-zh
威胁行为者基础设施追踪涉及使用被动 DNS、证书透明度日志、Shodan/Censys 扫描、WHOIS 分析和网络指纹技术，对对手控制的 C2 服务器、钓鱼域名和暂存服务器等资产进行监控、映射和持续追踪
profiling-threat-actor-groups
9
from killvxk/cybersecurity-skills-zh
通过聚合 TTP 文档、历史活动数据、工具指纹和来自多个情报源的归因指标，为 APT 组织、犯罪组织和黑客活动组织开发全面的威胁行为者画像。适用于就行业特定威胁向管理层汇报、更新威胁模型假设，或针对特定对手优先部署防御控制措施。当涉及 MITRE ATT&CK 组织、Mandiant APT 画像、CrowdStrike 对手命名或行业特定威胁简报时激活。
performing-threat-modeling-with-owasp-threat-dragon
9
from killvxk/cybersecurity-skills-zh
使用 OWASP Threat Dragon 创建数据流图，运用 STRIDE 和 LINDDUN 方法论识别威胁，并生成威胁模型报告用于安全设计审查。
performing-threat-landscape-assessment-for-sector
9
from killvxk/cybersecurity-skills-zh
通过分析威胁行为者定向攻击模式、常见攻击向量和行业特定漏洞，开展行业特定威胁态势评估，为组织风险管理提供决策依据
performing-threat-intelligence-sharing-with-misp
9
from killvxk/cybersecurity-skills-zh
使用 PyMISP 在 MISP 平台上创建、丰富和共享威胁情报事件，包括 IOC 管理、情报源集成、STIX 导出及社区共享工作流
performing-threat-hunting-with-yara-rules
9
from killvxk/cybersecurity-skills-zh
使用 YARA 模式匹配规则在文件系统和内存转储中狩猎恶意软件、可疑文件和入侵指标。 涵盖规则编写、yara-python 扫描以及与威胁情报源的集成。
performing-threat-hunting-with-elastic-siem
9
from killvxk/cybersecurity-skills-zh
使用 KQL/EQL 查询、检测规则和 Timeline 调查在 Elastic Security SIEM 中执行主动威胁狩猎， 识别绕过自动检测的威胁。适用于 SOC 团队针对特定 ATT&CK 技术进行狩猎、调查异常行为， 或使用 Elasticsearch 和 Kibana Security 验证检测覆盖缺口。
performing-threat-emulation-with-atomic-red-team
9
from killvxk/cybersecurity-skills-zh
使用 atomic-operator Python 框架执行 Atomic Red Team 测试，进行 MITRE ATT&CK 技术验证。 从 YAML 原子测试加载测试定义、运行攻击模拟并验证检测覆盖率。适用于测试 SIEM 检测规则、 验证 EDR 覆盖率或开展紫队演练。
performing-open-source-intelligence-gathering
9
from killvxk/cybersecurity-skills-zh
开源情报（OSINT）收集是红队演练的第一个主动阶段，操作员收集关于目标组织的公开可用信息，以识别攻击面、社会工程学目标、技术栈和凭据泄露情况。
performing-malware-hash-enrichment-with-virustotal
9
from killvxk/cybersecurity-skills-zh
使用 VirusTotal API 富化恶意软件文件哈希，获取检测率、行为分析、YARA 匹配和上下文威胁情报，用于事件分类和 IOC 验证。
performing-ioc-enrichment-automation
9
from killvxk/cybersecurity-skills-zh
通过编排 VirusTotal、AbuseIPDB、Shodan、MISP 和其他情报源的查询， 自动化入侵指标（IOC）丰富化，提供上下文评分和处置建议。 适用于 SOC 分析师在告警分诊或事件调查期间需要对 IP、域名、URL 和文件哈希 进行快速多源丰富化时。