09.Day 9：成果落地——Act 阶段战报生成与大屏数据落盘

🚀 Day 9：成果落地 —— Act 阶段战报生成与大屏数据落盘

今日目标：完成整个 PEAK 框架的最核心闭环：Act with Knowledge（基于知识采取行动）。
此时，Day 8 的代码已经执行完毕，我们的 Python 内存中拥有了一个名为all_hunt_evidence的列表，里面装满了诸如round_1_hit_count: 150这样的硬核数据。接下来的任务是：将这些数据转化为 LLM 能看懂的 Prompt，触发二次推理，并将最终的 JSON 战报连同前面的证据一起，打包成一个巨大的“企业级 Payload”写入 Splunk 索引。

💻 Step 1: Act 阶段 (喂入证据、二次触发 LLM 与防御性解析)

动作解析：
这一步的代码必须严格粘贴在 Day 8 的for循环结束之后。

1. 数据序列化：把 Python 里的字典数组转成 JSON 字符串，因为大模型只能读懂文本。
2. 构建 Prompt：赋予 AI “安全总监”的人设，要求它根据命中数（Hit Counts）的逻辑关系（如 R1 很高但 R2 为 0，说明是误报；R1 和 R2 都很高，说明是真实攻击），给出一个 0-100 的确切风险评分。
3. 防弹解析：通过try...except json.JSONDecodeError，防止大模型因为抽风输出非 JSON 格式而导致整个 5 分钟的调度任务崩溃。

📝 注入核心逻辑代码 (接在 Day 8 代码下方)：

# ==========================================# STEP 4: The Act Phase (Day 9 AI Final Qualification)# ==========================================helper.log_info("Initiating Act Phase: Triggering LLM API for Final Assessment...")# 1. Serialize the collected evidence list into a JSON string to feed the LLMevidence_payload=json.dumps(all_hunt_evidence,ensure_ascii=False)# 2. In a production environment, you would construct your messages here# and execute the second API call. For testing the pipeline, we mock the response.# This mock represents the LLM analyzing the 'evidence_payload' and making a decision.mock_act_response=""" { "executive_summary": "通过两轮下钻分析，假设1(暴力破解)在R2深度关联中发现连续高频请求，证实存在凭证填充攻击风险。", "threat_qualification": "Confirmed Threat", "risk_score": 92, "recommended_alert_spl": "search index=main CIp=* | bin _time span=1s | stats count by _time, CIp | where count > 10" } """# 3. Defensive parsing block to handle potential AI format hallucinationsfinal_report={}try:# Strip whitespace to prevent JSONDecodeError from trailing newlinesfinal_report=json.loads(mock_act_response.strip())helper.log_info(f"AI Assessment Complete. Final Risk Score calculated:{final_report.get('risk_score')}")exceptjson.JSONDecodeErrorase:helper.log_error(f"Critical error: Failed to parse final report JSON. Detail:{str(e)}")# Fallback mechanism to ensure the data pipeline does not breakfinal_report={"executive_summary":"Error parsing AI response during Act Phase.","threat_qualification":"Unknown","risk_score":0,"recommended_alert_spl":""}

💻 Step 2: 数据落盘 (组装上帝视角 Payload 并写入 Splunk)

动作解析：
很多新手写 Splunk 插件，习惯用print()或者helper.log_info()输出结果，这是大错特错的！打印在后台的日志（_internal索引）根本无法用于企业级大屏展示。

1. 组装 Master Payload：我们将 Day 7 的蓝图（ai_hunting_plan）、Day 8 的证据（execution_metrics）和刚才 Day 9 得到的终极定性（final_assessment）合并到一个字典中。这就相当于一份包含了起因、经过、结果的完整卷宗。
2. 强制指定 Sourcetype：使用helper.new_event()时，必须将sourcetype显式声明为_json。这会触发 Splunk 底层的自动字段提取机制（Auto KV Extraction），让复杂的 JSON 树在前端直接变成可被搜索的独立字段。
3. 提交写入：使用ew.write_event(event)将这条史诗级事件永久打入业务索引（如main索引）。

📝 注入落盘逻辑代码：

# ==========================================# STEP 5: Data Ingestion (Writing the Master Payload to Splunk)# ==========================================helper.log_info("Assembling Master Hunt Report for index ingestion...")# 1. Assemble the ultimate JSON document containing all three phases of PEAKmaster_payload={"event_type":"PEAK_Hunting_Report",# CRITICAL: Anchor field for Dashboard searches"timestamp":datetime.datetime.utcnow().isoformat(),"target_index":target_index,"hunting_plan":ai_hunting_plan,# The original blueprint from Prepare Phase"execution_metrics":all_hunt_evidence,# The hits and duration from Execute Phase"final_assessment":final_report# The ultimate AI qualification from Act Phase}# 2. Create a new Splunk event object using the Add-on Builder helper# Setting sourcetype to "_json" enables native Splunk JSON syntax highlighting and field extractionevent=helper.new_event(source=helper.get_input_type(),index=target_index,sourcetype="_json",data=json.dumps(master_payload,ensure_ascii=False))# 3. Commit the event to the user-specified Splunk data storeew.write_event(event)# Stop the master stopwatch and calculate total execution timetotal_cycle_time=round(time.time()-cycle_start_time,2)helper.log_info(f"SUCCESS: Master Hunt Report completely written to Splunk! Total cycle time:{total_cycle_time}seconds.")# 4. Catch-all exception block for the entire Agentic WorkflowexceptExceptionase:helper.log_error(f"Critical Failure during Agentic Execution Workflow:{str(e)}")

🔍 Step 3: 终极极客验证 (全自动闭环的结果确认)

代码写完，真正的极客绝不相信“理论上能跑”，我们必须用 SPL 在前端验证数据是否已完美结构化落地。

详细验证操作：

1. 在 AOB 代码编辑器的右上角点击绿色的Test按钮。
2. 观察底部 Output 面板，等待打印出绿色的Done。
3. 点击右上角 Save 按钮保存代码！(不点 Save 代码不会生效)
4. 打开一个新的浏览器标签页，进入 Splunk 的Search & Reporting(搜索与报表) 应用。
5. 在搜索框中输入以下极客指令（将时间范围调整为 Last 15 minutes）：

index=main sourcetype="_json" event_type="PEAK_Hunting_Report" | rename final_assessment.risk_score as Risk_Score, final_assessment.threat_qualification as Threat_Level, final_assessment.executive_summary as Summary | table _time, target_index, Threat_Level, Risk_Score, Summary | sort - Risk_Score

🎉 终极成功标志与意义：

• 你将看到什么：表格中清晰展现出Risk_Score = 92,Threat_Level = Confirmed Threat，以及详细的中文战报摘要。
• 背后的架构意义：由于我们在 Step 5 中设置了sourcetype="_json"，Splunk 极其聪明地自动解析了深达三层嵌套的 JSON 结构（它自动识别出了final_assessment.risk_score这个字段路径）。这种无缝的结构化数据落盘，是 Day 16 我们能够用几行 SPL 就画出极其绚丽的高管安全态势大屏的核心保障！