在 Splunk Enterprise Security 中构建威胁情报富化流水线IOC 摄取、KV Store 关联与 SOC 实战指南【免费下载链接】Anthropic-Cybersecurity-Skills817 structured cybersecurity skills for AI agents · Mapped to 6 frameworks: MITRE ATTCK, NIST CSF 2.0, MITRE ATLAS, D3FEND, NIST AI RMF MITRE F3 (Fight Fraud) · agentskills.io standard · Works with Claude Code, GitHub Copilot, Codex CLI, Cursor, Gemini CLI 20 platforms · 29 security domains · Apache 2.0项目地址: https://gitcode.com/GitHub_Trending/an/Anthropic-Cybersecurity-Skills导读本文基于开源仓库 Anthropic-Cybersecurity-Skills 中的 building-threat-intelligence-enrichment-in-splunk 技能文档系统讲解如何在 Splunk Enterprise SecurityES中搭建完整的威胁情报Threat Intelligence, TI富化流水线从 STIX/TAXII、CSV、API 等外部情报源摄取 IOCs经模块化输入Modular Inputs解析归一化后写入 KV Store 集合再通过查找表Lookup驱动的关联搜索对安全事件进行命中标记最终生成携带威胁上下文的 Notable Events显著缩短 SOC 分诊Triage时间。读完本文你将掌握威胁情报框架的架构设计、inputs.conf/collections.conf/transforms.conf三件套配置、三种典型 IOC 关联搜索IP / 域名 / 文件哈希、多源富化管道以及情报源健康度仪表盘的完整落地方法。威胁情报富化的核心价值Splunk Enterprise Security 内置的Threat Intelligence Framework威胁情报框架让 SOC 团队能够将失陷指标Indicators of Compromise, IOCs与安全事件自动关联。其工作方式可以概括为框架负责摄取威胁情报源将各类指标归一化后存入 KV Store 集合随后基于 Lookup 的关联搜索在事件流中标记 IOC 命中而Splunk Threat Intelligence Management则统一了来自多个情报源的采集、归一化与富化过程。对分析人员而言最终收益是事件被命中时立即获得威胁类型、置信度、来源、严重级别等上下文从而把精力集中到真正的告警上。该技能在仓库中的定位属于soc-operationsSOC 运营子域并映射到 NIST CSF 的 DE.CM-01持续监控、DE.AE-02异常与威胁活动分析、RS.MA-01危害评估与 DE.AE-06攻击指标关联同时映射到 MITRE ATTCK 的 T1071应用层协议、T1105工具传输与 T1041经 C2 信道外泄——即该流水线重点检测的攻击阶段集中在命令与控制与外泄环节。适用场景与前置条件适用场景在环境中部署或配置 Splunk 威胁情报富化能力需要建立与合规要求对齐的安全控制如 NIST CSF 的监测与响应类目正在构建或改进本领域的安全架构执行需要该能力落地的安全评估。前置条件条件说明Splunk Enterprise SecurityES7.x 及以上版本TI 组件Threat Intelligence Management 附加组件或 Threat Intelligence Framework外部情报源 API KeyMISP、AlienVault OTX、VirusTotal、AbuseIPDB 等KV Store已启用并正确配置IP / 域名 / 文件 / URL / 邮件五类集合权限具备模块化输入Modular Input配置的管理员访问权限威胁情报框架架构SKILL.md 给出了完整的分层架构这是整条流水线的骨架也是后续所有配置的落点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)与这条架构对应仓库中的 process.py 用 Python 对象模型把每一层都具象化了ThreatIndicator单个指标对象包含value、indicator_type、source、threat_type、confidence、severity、first_seen/last_seen等属性confidence在构造时会被钳制到 0–100_generate_id()用indicator_type:value:source的 SHA256 前 16 位生成唯一_key天然支持跨源的指标去重ThreatFeed情报源对象持有feed_typeapi / csv / stix / taxii 等、url、polling_interval、api_key并通过generate_splunk_input_conf()直接生成inputs.conf的[threatlist://...]段落EnrichmentPipeline流水线编排对象负责ingest_all_feeds()把活跃指标写入kv_store的各集合、simulate_correlation()模拟事件与 IOC 的关联命中与generate_spl_correlation()按指标类型生成 SPL 关联搜索模板。从源码结构看这套对象模型恰好对应 SKILL.md 中模块化输入 → KV Store → Lookup → 关联搜索的每一层可以作为理解 Splunk 配置背后数据流的最佳对照物。配置威胁情报数据源STIX/TAXII 情报源接入STIXStructured Threat Information eXpression2.1 是目前表示威胁情报的事实标准定义了 Indicator、Malware、Attack Pattern、Threat Actor、Campaign 等对象TAXIITrusted Automated eXchange of Indicator Information2.1 则是其传输机制通过 RESTful API 提供 Collection 与 Channel 两种共享模型。在 Splunk 侧接入一个 TAXII 情报源只需要在inputs.conf中声明一个 threatlist# 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关键参数说明type taxii声明该 threatlist 走 TAXII 协议解析urlcollection指向 TAXII 2.1 服务器与具体 Collectionpolling_interval 3600以秒为单位的轮询周期3600 即每小时拉取一次api_key建议存放 Splunk 的加密存储encrypted credential而非明文。CSV 情报列表对于内部运营产生的黑名单例如自家威胁情报团队维护的恶意 IP 列表直接以 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 falsepolling_interval 1800表示每 30 分钟刷新一次适合变动较快的内部阻断名单。无论是 TAXII 还是 CSV最终都会被解析并归一化进 KV Store。自定义模块化输入OTX API 情报收集器当标准 threatlist 无法覆盖某个 API 型情报源时可以编写 Splunk 自定义模块化输入Modular Input。SKILL.md 提供了一个完整的 AlienVault OTX 收集器示例使用splunklib.modularinput框架# 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 fhttps://otx.alienvault.com/api/v1/pulses/subscribed?modified_since{pulse_days}d try: response requests.get(url, headersheaders, timeout60) 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, fOTX feed collection failed: {str(e)}) if __name__ __main__: sys.exit(OTXFeedInput().run(sys.argv))实现要点get_scheme()声明了两个参数必填的api_key与可选的pulse_days默认 30 天并由 Splunk 自动生成配置 UIstream_events()中以X-OTX-API-KEY头调用pulses/subscribed接口把每个 pulse 内的指标展平成事件补充pulse_name、pulse_id、threat_source等富化字段异常处理通过ew.log(ERROR, ...)写入 Splunk 日志便于排障。仓库 api-reference.md 还给出了 OTX 的 curl 用法作为对照GET /api/v1/pulses/{PULSE_ID}/indicators取 pulse 指标GET /api/v1/search/pulses?qransomwarepage1加X-OTX-API-KEY头搜索 pulse。此外agent.py 中的fetch_otx_pulse_iocs()与convert_iocs_to_splunk_lookup()演示了把 OTX 指标转换为 KV Store 行记录映射IPv4/IPv6 → ip_intel、domain/hostname → domain_intel、FileHash-SHA256/MD5 → file_intelgenerate_splunk_lookup_csv()则可直接导出 CSV 供静态 Lookup 使用。构建富化查找表LookupKV Store 集合配置情报数据落库前需要在collections.conf中按指标类型声明集合与字段。SKILL.md 给出三组典型集合# 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字段类型string/number/time直接决定后续 SPL 中的比较与聚合行为例如confidence必须是number才能执行confidence 70的阈值过滤。域名集合额外纳入whois_registrar/whois_created用于注册信息富化文件哈希集合纳入hash_type、malware_family与detection_names用于恶意家族与杀软检出名。仓库 process.py 中to_kv_store_record()产出的记录字段_key、ip/domain/file_hash、threat_type、confidence、source、severity、first_seen、last_seen、tags与这里一一对应可视为集合字段的最小实现基准。查找表定义集合定义好后在transforms.conf中把集合暴露为 KV Store 型查找表# 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其中external_type kvstore表示该查找直接读写 KV Store 集合fields_list声明可用于匹配与 OUTPUT 的字段。仓库 agent.py 中定义了与 SKILL.md 互补的框架级集合清单SPLUNK_TI_COLLECTIONSip_intel / ip_intel_lookup、domain_intel / domain_intel_lookup、file_intel / file_intel_lookup、email_intel / email_intel_lookup字段统一为threat_key、description、source、weight、time模式——这是 ES 官方威胁情报框架SA-ThreatIntelligence 应用惯用的命名风格实际部署时可根据需要二选一或互相映射。富化关联搜索Correlation Searches关联搜索是整条流水线的触发器负责把事件字段与查找表中的 IOC 做匹配。基于 IP 的威胁情报关联| tstats summariesonlytrue count from datamodelNetwork_Traffic where All_Traffic.actionallowed by All_Traffic.src_ip, All_Traffic.dest_ip, All_Traffic.dest_port, _time span5m | 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 urgencycase( ti_severitycritical AND asset_prioritycritical, critical, ti_severityhigh OR asset_prioritycritical, high, ti_severitymedium, medium, true(), low ) | eval descriptionConnection from .src_ip. (.asset_name.) to known malicious IP .dest_ip. (.threat_type.) - Source: .ti_source要点拆解先用tstats从Network_Traffic数据模型按 5 分钟窗口聚合放行流量再lookup用dest_ip匹配ip_threat_intel_lookupOUTPUT重命名source as ti_source、severity as ti_severity避免与事件自身字段冲突二次lookup asset_lookup把源 IP 关联到资产信息asset_name/asset_owner/asset_priority这是 SKILL.md 强调的在 TI 之上叠加资产上下文eval urgency实现威胁严重级 × 资产优先级联合打分输出description直接作为 Notable Event 的摘要文本。基于域名的威胁情报关联indexdns sourcetypestream:dns query_typeA OR query_typeAAAA | 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 severitycase(domain_confidence 90, critical, domain_confidence 70, high, true(), medium) | eval descriptionDNS queries to malicious domain .query. from .unique_sources. hosts - Threat: .domain_threat针对 DNS 流将查询域名与域名情报库匹配用置信度做两级过滤domain_confidence 70才告警 90直接定级criticaldc(src_ip)统计解析恶意域名的去重主机数帮助判断是单点感染还是大规模内网中毒。基于文件哈希的关联indexendpoint sourcetypesysmon EventCode1 | 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 severitycritical | eval descriptionKnown malware .malware_family. executed on .Computer. by .User. - Binary: .Image针对 Sysmon 进程创建事件EventCode1用进程哈希命中已知恶意文件情报库命中即定级critical并汇总主机、用户、可执行文件路径与父进程命令行给出完整的执行上下文。仓库 process.py 的generate_spl_correlation()维护了与上述三类搜索一致的模板并在其中加入confidence 70条件而simulate_correlation()以纯 Python 方式模拟了 IP / 域名命中过程——把事件字典与 KV Store 记录做集合查找并产出match_type、matched_value、threat_type、confidence、severity、source等命中字段。你可以在本地直接运行该脚本观察整条流水线的模拟输出作为上线前验证关联逻辑的方式。多源富化管道Multi-Source Enrichment Pipeline生产环境中单一事件往往需要叠加多个情报维度。SKILL.md 给出了把 TI 命中、地理位置、WHOIS 三层富化合并到一次搜索的示例indexfirewall sourcetypepan:traffic actionallowed | eval indicatorsmvappend(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_contextThreat: .mvjoin(threat_types, , ). | Geo: .mvjoin(countries, , ). | Org: .mvjoin(organizations, , )mvappend(src_ip, dest_ip)mvexpand将源、目的 IP 展平为多值依次匹配一份搜索同时覆盖入站与出站方向对同一 IP 连续执行 TI、GeoIP、WHOIS 三个 Lookup命中后通过stats ... values()聚合去重最终enrichment_context把威胁类型、来源、国家、归属组织拼接成一句话上下文直接供告警与工单展示。结合 process.py 的示例数据OTX 与 AbuseIPDB 两个 feed、跨源重复指标203.0.113.50可以看到仓库在数据层面也印证了多源场景同一指标来自不同情报源时置信度与来源归属会被分别保留最终由下游规则决定取最高置信度还是合并展示。威胁情报仪表盘IOC 覆盖统计| inputlookup ip_threat_intel_lookup | stats count by source, threat_type | sort -count | head 20按来源与威胁类型统计当前情报库的指标分布用于回答各情报源贡献了多少指标、以哪些威胁类型为主。情报源新鲜度监控| inputlookup ip_threat_intel_lookup | eval age_daysround((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 statuscase(avg_age_days 30, STALE, avg_age_days 7, AGING, true(), FRESH)依据每条指标last_seen计算平均与最大存活天数把情报源标记为FRESH 7 天、AGING7–30 天或STALE 30 天驱动运维及时清理过期指标或排查断更的 feed。assets/template.md 提供了完整的运营看板模板建议纳入的指标包括活跃指标总数、feed 平均新鲜度目标 7 天、近 30 天命中率目标 0.5%、误报率目标 5%与 feed 重叠率目标 30%。情报生命周期管理与数据质量仓库 workflows.md 将整条流水线归纳为可复用的六步工作流1. Identify Relevant TI Sources - Commercial feeds (Recorded Future, Mandiant) - Open source (OTX, AbuseIPDB, VirusTotal) - Industry ISACs - Internal threat lists | v 2. Configure Modular Inputs - Set polling intervals - Configure authentication - Map feed fields to Splunk schema | v 3. Normalize to KV Store - Parse raw feed data - Map to standard field names - Set confidence scores - Add source attribution | v 4. Create Lookup Definitions - Define transforms.conf entries - Set field mappings - Enable automatic lookups where appropriate | v 5. Build Correlation Searches - Match events against IOC lookups - Add asset/identity enrichment - Set severity based on confidence | v 6. Monitor and Maintain - Track feed freshness - Remove stale indicators - Measure hit rates per sourceIOC 生命周期被建模为Ingestion → Validation → Active Use → Aging → Expiration → Removal即原始 feed 解析后经去重与置信度打分进入活跃匹配期随后按时间衰减置信度权重最终过期归档或删除。这与 process.py 中ThreatIndicator.is_expired(max_age_days90)及ThreatFeed.get_active_indicators()的实现一致指标超过 90 天未被更新即视为过期并从活跃集中剔除。feed 质量评估建议按以下阈值持续监控指标良好警告严重Feed 延迟 1 小时1–24 小时 24 小时误报率 5%5–15% 15%命中率 1%0.1–1% 0.1%覆盖重叠率 30%30–60% 60%指标新鲜度 7 天7–30 天 30 天通过 REST API 与 Python SDK 管理 KV Store除了配置文件api-reference.md 提供了自动化运维的另一种路径——直接调用 Splunk KV Store REST API# Create collection curl -k -u admin:pass -X POST \ https://localhost:8089/servicesNS/nobody/SA-ThreatIntelligence/storage/collections/config \ -d nameip_intel # Insert record curl -k -u admin:pass -X POST \ https://localhost:8089/servicesNS/nobody/SA-ThreatIntelligence/storage/collections/data/ip_intel \ -H Content-Type: application/json \ -d {ip:198.51.100.42,threat_key:c2_server,weight:3} # Batch insert curl -k -u admin:pass -X POST \ https://localhost:8089/servicesNS/nobody/SA-ThreatIntelligence/storage/collections/data/ip_intel/batch_save \ -H Content-Type: application/json \ -d [{ip:1.2.3.4,threat_key:malware},{ip:5.6.7.8,threat_key:c2}]注意 REST 端点位于SA-ThreatIntelligence应用命名空间下即 ES 威胁情报框架自带的 KV Storebatch_save用于批量写入。对应的 Python 侧实现使用 Splunk SDKimport splunklib.client as client service client.connect( hostlocalhost, port8089, usernameadmin, passwordchangeme ) # Access KV store collection collection service.kvstore[ip_intel] collection.data.insert(json.dumps({ ip: 198.51.100.42, threat_key: c2_server }))这套 API 非常适合把外部编排系统如 SOAR接到 Splunk第三方情报平台推送的新指标可以直接经 REST 写入 KV Store无需重启任何模块化输入。ES 框架下各集合与查找表、数据模型的对应关系总结如下CollectionLookup对应 Data Modelip_intelip_intel_lookupNetwork_Trafficdomain_inteldomain_intel_lookupNetwork_Resolutionfile_intelfile_intel_lookupEndpointemail_intelemail_intel_lookupEmailhttp_intelhttp_intel_lookupWeb指标类型、CIM 数据模型与置信度阈值standards.md 给出了与 Splunk CIMCommon Information Model对齐的关联字段与阈值基线是编写关联搜索时的重要参考CIM 数据模型与 TI 关联字段Data ModelTI Correlation FieldsNetwork_Trafficsrc_ip, dest_ip, dest_portWeburl, http_user_agent, domainEmailsrc_user, file_hash, urlEndpointprocess_hash, file_hash, destAuthenticationsrc_ip, user, appDNSquery, answer, src_ipIOC 类型与推荐置信度阈值IOC 类型Splunk 集合置信度阈值IP 地址ip_intel 70%域名domain_intel 70%文件哈希SHA256file_intel 80%URLurl_intel 75%邮箱地址email_intel 80%相关标准还包括OpenIOCMandiant 的 XML 格式 IOC 共享框架与 OCSFOpen Cybersecurity Schema Framework面向安全事件数据归一化的行业标准1.0 版本于 BlackHat 2023 发布。在配置关联搜索时建议按此阈值表在where条件中过滤低置信度命中以控制误报。参考本技能文档skills/building-threat-intelligence-enrichment-in-splunk/SKILL.mdAPI 参考KV Store REST API / OTX API / Python SDKreferences/api-reference.md标准参考STIX/TAXII/OpenIOC/OCSF、CIM 映射、置信度阈值references/standards.md工作流参考六步接入流程、IOC 生命周期、feed 质量评估references/workflows.md可运行脚本OTX 拉取、KV Store 转换、SPL 生成scripts/agent.py流水线模拟脚本ThreatIndicator / ThreatFeed / EnrichmentPipelinescripts/process.py部署与运营模板assets/template.md结语至此一条完整的 Splunk 威胁情报富化流水线已经具备可落地闭环外部情报源TAXII / CSV / 自定义 API 输入→ 模块化输入解析 → KV Store 集合归一化存储 → Lookup 定义 → 关联搜索命中富化 → Notable Events 输出上下文再配合 IOC 覆盖统计、feed 新鲜度监控与质量评估持续运营。SKILL.md 中的配置示例与 process.py 的模拟实现相互印证无论是首次在 ES 中接入威胁情报还是优化既有流水线的命中质量与数据新鲜度都可以直接以此为起点逐步落地。【免费下载链接】Anthropic-Cybersecurity-Skills817 structured cybersecurity skills for AI agents · Mapped to 6 frameworks: MITRE ATTCK, NIST CSF 2.0, MITRE ATLAS, D3FEND, NIST AI RMF MITRE F3 (Fight Fraud) · agentskills.io standard · Works with Claude Code, GitHub Copilot, Codex CLI, Cursor, Gemini CLI 20 platforms · 29 security domains · Apache 2.0项目地址: https://gitcode.com/GitHub_Trending/an/Anthropic-Cybersecurity-Skills创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
在 Splunk Enterprise Security 中构建威胁情报富化流水线：IOC 摄取、KV Store 关联与 SOC 实战指南
在 Splunk Enterprise Security 中构建威胁情报富化流水线IOC 摄取、KV Store 关联与 SOC 实战指南【免费下载链接】Anthropic-Cybersecurity-Skills817 structured cybersecurity skills for AI agents · Mapped to 6 frameworks: MITRE ATTCK, NIST CSF 2.0, MITRE ATLAS, D3FEND, NIST AI RMF MITRE F3 (Fight Fraud) · agentskills.io standard · Works with Claude Code, GitHub Copilot, Codex CLI, Cursor, Gemini CLI 20 platforms · 29 security domains · Apache 2.0项目地址: https://gitcode.com/GitHub_Trending/an/Anthropic-Cybersecurity-Skills导读本文基于开源仓库 Anthropic-Cybersecurity-Skills 中的 building-threat-intelligence-enrichment-in-splunk 技能文档系统讲解如何在 Splunk Enterprise SecurityES中搭建完整的威胁情报Threat Intelligence, TI富化流水线从 STIX/TAXII、CSV、API 等外部情报源摄取 IOCs经模块化输入Modular Inputs解析归一化后写入 KV Store 集合再通过查找表Lookup驱动的关联搜索对安全事件进行命中标记最终生成携带威胁上下文的 Notable Events显著缩短 SOC 分诊Triage时间。读完本文你将掌握威胁情报框架的架构设计、inputs.conf/collections.conf/transforms.conf三件套配置、三种典型 IOC 关联搜索IP / 域名 / 文件哈希、多源富化管道以及情报源健康度仪表盘的完整落地方法。威胁情报富化的核心价值Splunk Enterprise Security 内置的Threat Intelligence Framework威胁情报框架让 SOC 团队能够将失陷指标Indicators of Compromise, IOCs与安全事件自动关联。其工作方式可以概括为框架负责摄取威胁情报源将各类指标归一化后存入 KV Store 集合随后基于 Lookup 的关联搜索在事件流中标记 IOC 命中而Splunk Threat Intelligence Management则统一了来自多个情报源的采集、归一化与富化过程。对分析人员而言最终收益是事件被命中时立即获得威胁类型、置信度、来源、严重级别等上下文从而把精力集中到真正的告警上。该技能在仓库中的定位属于soc-operationsSOC 运营子域并映射到 NIST CSF 的 DE.CM-01持续监控、DE.AE-02异常与威胁活动分析、RS.MA-01危害评估与 DE.AE-06攻击指标关联同时映射到 MITRE ATTCK 的 T1071应用层协议、T1105工具传输与 T1041经 C2 信道外泄——即该流水线重点检测的攻击阶段集中在命令与控制与外泄环节。适用场景与前置条件适用场景在环境中部署或配置 Splunk 威胁情报富化能力需要建立与合规要求对齐的安全控制如 NIST CSF 的监测与响应类目正在构建或改进本领域的安全架构执行需要该能力落地的安全评估。前置条件条件说明Splunk Enterprise SecurityES7.x 及以上版本TI 组件Threat Intelligence Management 附加组件或 Threat Intelligence Framework外部情报源 API KeyMISP、AlienVault OTX、VirusTotal、AbuseIPDB 等KV Store已启用并正确配置IP / 域名 / 文件 / URL / 邮件五类集合权限具备模块化输入Modular Input配置的管理员访问权限威胁情报框架架构SKILL.md 给出了完整的分层架构这是整条流水线的骨架也是后续所有配置的落点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)与这条架构对应仓库中的 process.py 用 Python 对象模型把每一层都具象化了ThreatIndicator单个指标对象包含value、indicator_type、source、threat_type、confidence、severity、first_seen/last_seen等属性confidence在构造时会被钳制到 0–100_generate_id()用indicator_type:value:source的 SHA256 前 16 位生成唯一_key天然支持跨源的指标去重ThreatFeed情报源对象持有feed_typeapi / csv / stix / taxii 等、url、polling_interval、api_key并通过generate_splunk_input_conf()直接生成inputs.conf的[threatlist://...]段落EnrichmentPipeline流水线编排对象负责ingest_all_feeds()把活跃指标写入kv_store的各集合、simulate_correlation()模拟事件与 IOC 的关联命中与generate_spl_correlation()按指标类型生成 SPL 关联搜索模板。从源码结构看这套对象模型恰好对应 SKILL.md 中模块化输入 → KV Store → Lookup → 关联搜索的每一层可以作为理解 Splunk 配置背后数据流的最佳对照物。配置威胁情报数据源STIX/TAXII 情报源接入STIXStructured Threat Information eXpression2.1 是目前表示威胁情报的事实标准定义了 Indicator、Malware、Attack Pattern、Threat Actor、Campaign 等对象TAXIITrusted Automated eXchange of Indicator Information2.1 则是其传输机制通过 RESTful API 提供 Collection 与 Channel 两种共享模型。在 Splunk 侧接入一个 TAXII 情报源只需要在inputs.conf中声明一个 threatlist# 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关键参数说明type taxii声明该 threatlist 走 TAXII 协议解析urlcollection指向 TAXII 2.1 服务器与具体 Collectionpolling_interval 3600以秒为单位的轮询周期3600 即每小时拉取一次api_key建议存放 Splunk 的加密存储encrypted credential而非明文。CSV 情报列表对于内部运营产生的黑名单例如自家威胁情报团队维护的恶意 IP 列表直接以 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 falsepolling_interval 1800表示每 30 分钟刷新一次适合变动较快的内部阻断名单。无论是 TAXII 还是 CSV最终都会被解析并归一化进 KV Store。自定义模块化输入OTX API 情报收集器当标准 threatlist 无法覆盖某个 API 型情报源时可以编写 Splunk 自定义模块化输入Modular Input。SKILL.md 提供了一个完整的 AlienVault OTX 收集器示例使用splunklib.modularinput框架# 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 fhttps://otx.alienvault.com/api/v1/pulses/subscribed?modified_since{pulse_days}d try: response requests.get(url, headersheaders, timeout60) 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, fOTX feed collection failed: {str(e)}) if __name__ __main__: sys.exit(OTXFeedInput().run(sys.argv))实现要点get_scheme()声明了两个参数必填的api_key与可选的pulse_days默认 30 天并由 Splunk 自动生成配置 UIstream_events()中以X-OTX-API-KEY头调用pulses/subscribed接口把每个 pulse 内的指标展平成事件补充pulse_name、pulse_id、threat_source等富化字段异常处理通过ew.log(ERROR, ...)写入 Splunk 日志便于排障。仓库 api-reference.md 还给出了 OTX 的 curl 用法作为对照GET /api/v1/pulses/{PULSE_ID}/indicators取 pulse 指标GET /api/v1/search/pulses?qransomwarepage1加X-OTX-API-KEY头搜索 pulse。此外agent.py 中的fetch_otx_pulse_iocs()与convert_iocs_to_splunk_lookup()演示了把 OTX 指标转换为 KV Store 行记录映射IPv4/IPv6 → ip_intel、domain/hostname → domain_intel、FileHash-SHA256/MD5 → file_intelgenerate_splunk_lookup_csv()则可直接导出 CSV 供静态 Lookup 使用。构建富化查找表LookupKV Store 集合配置情报数据落库前需要在collections.conf中按指标类型声明集合与字段。SKILL.md 给出三组典型集合# 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字段类型string/number/time直接决定后续 SPL 中的比较与聚合行为例如confidence必须是number才能执行confidence 70的阈值过滤。域名集合额外纳入whois_registrar/whois_created用于注册信息富化文件哈希集合纳入hash_type、malware_family与detection_names用于恶意家族与杀软检出名。仓库 process.py 中to_kv_store_record()产出的记录字段_key、ip/domain/file_hash、threat_type、confidence、source、severity、first_seen、last_seen、tags与这里一一对应可视为集合字段的最小实现基准。查找表定义集合定义好后在transforms.conf中把集合暴露为 KV Store 型查找表# 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其中external_type kvstore表示该查找直接读写 KV Store 集合fields_list声明可用于匹配与 OUTPUT 的字段。仓库 agent.py 中定义了与 SKILL.md 互补的框架级集合清单SPLUNK_TI_COLLECTIONSip_intel / ip_intel_lookup、domain_intel / domain_intel_lookup、file_intel / file_intel_lookup、email_intel / email_intel_lookup字段统一为threat_key、description、source、weight、time模式——这是 ES 官方威胁情报框架SA-ThreatIntelligence 应用惯用的命名风格实际部署时可根据需要二选一或互相映射。富化关联搜索Correlation Searches关联搜索是整条流水线的触发器负责把事件字段与查找表中的 IOC 做匹配。基于 IP 的威胁情报关联| tstats summariesonlytrue count from datamodelNetwork_Traffic where All_Traffic.actionallowed by All_Traffic.src_ip, All_Traffic.dest_ip, All_Traffic.dest_port, _time span5m | 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 urgencycase( ti_severitycritical AND asset_prioritycritical, critical, ti_severityhigh OR asset_prioritycritical, high, ti_severitymedium, medium, true(), low ) | eval descriptionConnection from .src_ip. (.asset_name.) to known malicious IP .dest_ip. (.threat_type.) - Source: .ti_source要点拆解先用tstats从Network_Traffic数据模型按 5 分钟窗口聚合放行流量再lookup用dest_ip匹配ip_threat_intel_lookupOUTPUT重命名source as ti_source、severity as ti_severity避免与事件自身字段冲突二次lookup asset_lookup把源 IP 关联到资产信息asset_name/asset_owner/asset_priority这是 SKILL.md 强调的在 TI 之上叠加资产上下文eval urgency实现威胁严重级 × 资产优先级联合打分输出description直接作为 Notable Event 的摘要文本。基于域名的威胁情报关联indexdns sourcetypestream:dns query_typeA OR query_typeAAAA | 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 severitycase(domain_confidence 90, critical, domain_confidence 70, high, true(), medium) | eval descriptionDNS queries to malicious domain .query. from .unique_sources. hosts - Threat: .domain_threat针对 DNS 流将查询域名与域名情报库匹配用置信度做两级过滤domain_confidence 70才告警 90直接定级criticaldc(src_ip)统计解析恶意域名的去重主机数帮助判断是单点感染还是大规模内网中毒。基于文件哈希的关联indexendpoint sourcetypesysmon EventCode1 | 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 severitycritical | eval descriptionKnown malware .malware_family. executed on .Computer. by .User. - Binary: .Image针对 Sysmon 进程创建事件EventCode1用进程哈希命中已知恶意文件情报库命中即定级critical并汇总主机、用户、可执行文件路径与父进程命令行给出完整的执行上下文。仓库 process.py 的generate_spl_correlation()维护了与上述三类搜索一致的模板并在其中加入confidence 70条件而simulate_correlation()以纯 Python 方式模拟了 IP / 域名命中过程——把事件字典与 KV Store 记录做集合查找并产出match_type、matched_value、threat_type、confidence、severity、source等命中字段。你可以在本地直接运行该脚本观察整条流水线的模拟输出作为上线前验证关联逻辑的方式。多源富化管道Multi-Source Enrichment Pipeline生产环境中单一事件往往需要叠加多个情报维度。SKILL.md 给出了把 TI 命中、地理位置、WHOIS 三层富化合并到一次搜索的示例indexfirewall sourcetypepan:traffic actionallowed | eval indicatorsmvappend(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_contextThreat: .mvjoin(threat_types, , ). | Geo: .mvjoin(countries, , ). | Org: .mvjoin(organizations, , )mvappend(src_ip, dest_ip)mvexpand将源、目的 IP 展平为多值依次匹配一份搜索同时覆盖入站与出站方向对同一 IP 连续执行 TI、GeoIP、WHOIS 三个 Lookup命中后通过stats ... values()聚合去重最终enrichment_context把威胁类型、来源、国家、归属组织拼接成一句话上下文直接供告警与工单展示。结合 process.py 的示例数据OTX 与 AbuseIPDB 两个 feed、跨源重复指标203.0.113.50可以看到仓库在数据层面也印证了多源场景同一指标来自不同情报源时置信度与来源归属会被分别保留最终由下游规则决定取最高置信度还是合并展示。威胁情报仪表盘IOC 覆盖统计| inputlookup ip_threat_intel_lookup | stats count by source, threat_type | sort -count | head 20按来源与威胁类型统计当前情报库的指标分布用于回答各情报源贡献了多少指标、以哪些威胁类型为主。情报源新鲜度监控| inputlookup ip_threat_intel_lookup | eval age_daysround((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 statuscase(avg_age_days 30, STALE, avg_age_days 7, AGING, true(), FRESH)依据每条指标last_seen计算平均与最大存活天数把情报源标记为FRESH 7 天、AGING7–30 天或STALE 30 天驱动运维及时清理过期指标或排查断更的 feed。assets/template.md 提供了完整的运营看板模板建议纳入的指标包括活跃指标总数、feed 平均新鲜度目标 7 天、近 30 天命中率目标 0.5%、误报率目标 5%与 feed 重叠率目标 30%。情报生命周期管理与数据质量仓库 workflows.md 将整条流水线归纳为可复用的六步工作流1. Identify Relevant TI Sources - Commercial feeds (Recorded Future, Mandiant) - Open source (OTX, AbuseIPDB, VirusTotal) - Industry ISACs - Internal threat lists | v 2. Configure Modular Inputs - Set polling intervals - Configure authentication - Map feed fields to Splunk schema | v 3. Normalize to KV Store - Parse raw feed data - Map to standard field names - Set confidence scores - Add source attribution | v 4. Create Lookup Definitions - Define transforms.conf entries - Set field mappings - Enable automatic lookups where appropriate | v 5. Build Correlation Searches - Match events against IOC lookups - Add asset/identity enrichment - Set severity based on confidence | v 6. Monitor and Maintain - Track feed freshness - Remove stale indicators - Measure hit rates per sourceIOC 生命周期被建模为Ingestion → Validation → Active Use → Aging → Expiration → Removal即原始 feed 解析后经去重与置信度打分进入活跃匹配期随后按时间衰减置信度权重最终过期归档或删除。这与 process.py 中ThreatIndicator.is_expired(max_age_days90)及ThreatFeed.get_active_indicators()的实现一致指标超过 90 天未被更新即视为过期并从活跃集中剔除。feed 质量评估建议按以下阈值持续监控指标良好警告严重Feed 延迟 1 小时1–24 小时 24 小时误报率 5%5–15% 15%命中率 1%0.1–1% 0.1%覆盖重叠率 30%30–60% 60%指标新鲜度 7 天7–30 天 30 天通过 REST API 与 Python SDK 管理 KV Store除了配置文件api-reference.md 提供了自动化运维的另一种路径——直接调用 Splunk KV Store REST API# Create collection curl -k -u admin:pass -X POST \ https://localhost:8089/servicesNS/nobody/SA-ThreatIntelligence/storage/collections/config \ -d nameip_intel # Insert record curl -k -u admin:pass -X POST \ https://localhost:8089/servicesNS/nobody/SA-ThreatIntelligence/storage/collections/data/ip_intel \ -H Content-Type: application/json \ -d {ip:198.51.100.42,threat_key:c2_server,weight:3} # Batch insert curl -k -u admin:pass -X POST \ https://localhost:8089/servicesNS/nobody/SA-ThreatIntelligence/storage/collections/data/ip_intel/batch_save \ -H Content-Type: application/json \ -d [{ip:1.2.3.4,threat_key:malware},{ip:5.6.7.8,threat_key:c2}]注意 REST 端点位于SA-ThreatIntelligence应用命名空间下即 ES 威胁情报框架自带的 KV Storebatch_save用于批量写入。对应的 Python 侧实现使用 Splunk SDKimport splunklib.client as client service client.connect( hostlocalhost, port8089, usernameadmin, passwordchangeme ) # Access KV store collection collection service.kvstore[ip_intel] collection.data.insert(json.dumps({ ip: 198.51.100.42, threat_key: c2_server }))这套 API 非常适合把外部编排系统如 SOAR接到 Splunk第三方情报平台推送的新指标可以直接经 REST 写入 KV Store无需重启任何模块化输入。ES 框架下各集合与查找表、数据模型的对应关系总结如下CollectionLookup对应 Data Modelip_intelip_intel_lookupNetwork_Trafficdomain_inteldomain_intel_lookupNetwork_Resolutionfile_intelfile_intel_lookupEndpointemail_intelemail_intel_lookupEmailhttp_intelhttp_intel_lookupWeb指标类型、CIM 数据模型与置信度阈值standards.md 给出了与 Splunk CIMCommon Information Model对齐的关联字段与阈值基线是编写关联搜索时的重要参考CIM 数据模型与 TI 关联字段Data ModelTI Correlation FieldsNetwork_Trafficsrc_ip, dest_ip, dest_portWeburl, http_user_agent, domainEmailsrc_user, file_hash, urlEndpointprocess_hash, file_hash, destAuthenticationsrc_ip, user, appDNSquery, answer, src_ipIOC 类型与推荐置信度阈值IOC 类型Splunk 集合置信度阈值IP 地址ip_intel 70%域名domain_intel 70%文件哈希SHA256file_intel 80%URLurl_intel 75%邮箱地址email_intel 80%相关标准还包括OpenIOCMandiant 的 XML 格式 IOC 共享框架与 OCSFOpen Cybersecurity Schema Framework面向安全事件数据归一化的行业标准1.0 版本于 BlackHat 2023 发布。在配置关联搜索时建议按此阈值表在where条件中过滤低置信度命中以控制误报。参考本技能文档skills/building-threat-intelligence-enrichment-in-splunk/SKILL.mdAPI 参考KV Store REST API / OTX API / Python SDKreferences/api-reference.md标准参考STIX/TAXII/OpenIOC/OCSF、CIM 映射、置信度阈值references/standards.md工作流参考六步接入流程、IOC 生命周期、feed 质量评估references/workflows.md可运行脚本OTX 拉取、KV Store 转换、SPL 生成scripts/agent.py流水线模拟脚本ThreatIndicator / ThreatFeed / EnrichmentPipelinescripts/process.py部署与运营模板assets/template.md结语至此一条完整的 Splunk 威胁情报富化流水线已经具备可落地闭环外部情报源TAXII / CSV / 自定义 API 输入→ 模块化输入解析 → KV Store 集合归一化存储 → Lookup 定义 → 关联搜索命中富化 → Notable Events 输出上下文再配合 IOC 覆盖统计、feed 新鲜度监控与质量评估持续运营。SKILL.md 中的配置示例与 process.py 的模拟实现相互印证无论是首次在 ES 中接入威胁情报还是优化既有流水线的命中质量与数据新鲜度都可以直接以此为起点逐步落地。【免费下载链接】Anthropic-Cybersecurity-Skills817 structured cybersecurity skills for AI agents · Mapped to 6 frameworks: MITRE ATTCK, NIST CSF 2.0, MITRE ATLAS, D3FEND, NIST AI RMF MITRE F3 (Fight Fraud) · agentskills.io standard · Works with Claude Code, GitHub Copilot, Codex CLI, Cursor, Gemini CLI 20 platforms · 29 security domains · Apache 2.0项目地址: https://gitcode.com/GitHub_Trending/an/Anthropic-Cybersecurity-Skills创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
