Day 6: 连接数据之源 —— Splunk SDK for Python 实战今日目标: 验证Splunk 中的Python语句执行功能包括调取插件参数以及用SPL语句执行查询都是我们在后期需要的核心功能。今天我们将直接在 Add-on Builder (AOB) 强大的前端代码编辑器中利用 Splunk SDK for Python 获取session_key。动态读取用户配置的 Index执行带有特定时间窗口的防超载随机抽样 SPL拉取 5 条真实的底层日志。在 AOB 内完成测试无语法报错后回到 Splunk 搜索栏进行最终的后台调度与日志验证。️ Step 1: 回到 AOB 的代码实验室 (告别终端)我们今天全程在 Splunk 前端网页中操作。详细操作路径登录 Splunk: 打开浏览器输入你的 Splunk 地址通常是http://localhost:8000输入管理员账号密码登录。进入 AOB 应用: 在 Splunk 首页左侧的 Apps应用列表中找到并点击Splunk Add-on Builder。打开项目: 在 AOB 首页的“Created with Add-on Builder”列表中找到我们之前创建的PEAK-llm-analyzer项目点击它进入项目主页。进入数据收集配置页: 在项目主页的顶部导航栏点击Configure Data Collection配置数据收集。编辑任务模板: 在弹出的列表中找到我们在 Day 4 创建的PEAK AI Hunter。点击这行最右侧的Edit编辑按钮。直达代码编辑区: 此时会弹出向导窗口直接点击顶部的第三个标签页Define Test定义与测试。此时你左侧看到的是参数输入区右侧是 AOB 自带的Code Editor代码编辑器。 Step 2: 注入 SDK 抽样逻辑 (Define Test 页面)在右侧的Code Editor中我们需要做两处核心修改。第一处在文件最顶部导入 SDK滚动到代码的第一行在原有的import列表下方加入 Splunk 官方 SDK 库绑定AOB 环境已自带无需安装importosimportsysimporttimeimportdatetimeimportjson# NEW: Import Splunk SDK bindingsimportsplunklib.clientasclientimportsplunklib.resultsasresults第二处重写 collect_events 函数向下滚动找到def collect_events(helper, ew):这一行。将其下方的代码完全替换为以下逻辑⚠️ 极客铁律: 代码注释与日志严格使用纯英文。defcollect_events(helper,ew):Implement your data collection logic herehelper.log_info(PEAK AI Hunter started data collection cycle.)try:# 1. Robustly acquire the session_key from the modular input contextsession_keygetattr(helper,session_key,None)# Fallback for certain AOB versions: extract from internal input_definitionifnotsession_keyandhasattr(helper,_input_definition):session_keygetattr(helper._input_definition,metadata,{}).get(session_key)ifnotsession_key:helper.log_error(Failed to acquire session_key from context.)return# 2. Connect to the Splunk service using the SDKserviceclient.Service(tokensession_key)# 3. Dynamically hijack the user-selected Index# We read the index the user configured in the UI during Day 5target_indexhelper.get_output_index()ormain# 4. Construct the randomized baseline SPL with a specific time window# Geek Tip: We use earliest-24h to limit the time window, and head 10000 BEFORE random() to prevent timeouts.search_query(fsearch index{target_index}earliest-24h latestnow f| head 10000 | eval rand_valrandom() | sort 5 rand_val)# 5. Use jobs.oneshot for fast, synchronous retrieval in JSON formatkwargs_oneshot{output_mode:json}helper.log_info(fExecuting randomized oneshot SPL:{search_query})oneshot_search_resultsservice.jobs.oneshot(search_query,**kwargs_oneshot)# 6. Parse the results using the SDKs built-in JSON readerreaderresults.JSONResultsReader(oneshot_search_results)baseline_logs[]forresultinreader:ifisinstance(result,dict)and_rawinresult:baseline_logs.append(result[_raw])# 7. Log the extracted data length and print a sample to verify successiflen(baseline_logs)0:helper.log_info(fSuccessfully fetched{len(baseline_logs)}random raw logs from index {target_index}.)helper.log_info(fSample Random Log [1]:{baseline_logs[0][:200]}...)else:helper.log_warning(fSPL executed successfully, but returned 0 events from index {target_index}.)exceptExceptionase:helper.log_error(fExecution error occurred during Splunk SDK search:{str(e)})⚡ Step 3: 一键防崩测试与保存AOB 的代码编辑器自带了一个极客友好的防崩溃测试功能。详细操作路径在页面左侧的Data input parameters区域随便填入必须填的参数比如输入SDK Test。点击页面右上角绿色的Test按钮。立即向下滚动网页查看编辑器底部的Output (输出)面板。 测试成功标志此时AOB 会在后台静默运行你的代码。只要 Output 面板最后显示了一个绿色的Done并且没有弹出大段红色的报错代码Traceback这就证明你的 Python 语法完美无缺且 SDK 成功向系统发起了查询(注AOB 引擎会将helper.log_info直接写入底层的真实日志文件因此你在前端面板只会看到 Done这是企业级规范。)固化代码 (Save) —— 极其重要: 测试没报错后必须点击右上角绿色的Save按钮保存代码然后关闭向导窗口。 Step 4: 全局验证 —— 回到 Splunk 检索真正的样本数据我们在 AOB 里验证了代码“不会崩”现在要去系统的最底层去验收我们抓取到的“战利品”。点击左上角的Splunkenterprise徽标返回 Splunk 首页。进入Search Reporting搜索与报表应用。我们的代码被调度器接管后每隔 60 秒就会跑一次。在搜索框中输入以下极客搜索命令去内部日志里抓取结果index_internal PEAK AI Hunter started data collection cycle OR Sample Random Log将时间范围调整为Last 15 minutes点击搜索。 终极成功标志你会看到不仅有“开始采集”的心跳日志每隔 60 秒还会稳定地输出一次Successfully fetched 5 random raw logs...并且紧跟着一条Sample Random Log [1]: ...。由于我们加了random()函数你会发现每次打印出来的样本内容都是不一样的这证明你写的代码不仅极其强壮而且正在源源不断地为 AI 采集真实的、不同切面的随机数据样本明天Day 7我们将把这些动态抽取的真实随机日志配合极其严酷的 System Prompt系统提示词正式发往大模型。让 AI 化身资深安全专家它将观察这些正常的系统日志并像黑客一样逆向思考推测出系统可能遭受的攻击方式进入真正的威胁狩猎模式
06.Day 6：连接数据之源 —— Splunk SDK for Python 实战
Day 6: 连接数据之源 —— Splunk SDK for Python 实战今日目标: 验证Splunk 中的Python语句执行功能包括调取插件参数以及用SPL语句执行查询都是我们在后期需要的核心功能。今天我们将直接在 Add-on Builder (AOB) 强大的前端代码编辑器中利用 Splunk SDK for Python 获取session_key。动态读取用户配置的 Index执行带有特定时间窗口的防超载随机抽样 SPL拉取 5 条真实的底层日志。在 AOB 内完成测试无语法报错后回到 Splunk 搜索栏进行最终的后台调度与日志验证。️ Step 1: 回到 AOB 的代码实验室 (告别终端)我们今天全程在 Splunk 前端网页中操作。详细操作路径登录 Splunk: 打开浏览器输入你的 Splunk 地址通常是http://localhost:8000输入管理员账号密码登录。进入 AOB 应用: 在 Splunk 首页左侧的 Apps应用列表中找到并点击Splunk Add-on Builder。打开项目: 在 AOB 首页的“Created with Add-on Builder”列表中找到我们之前创建的PEAK-llm-analyzer项目点击它进入项目主页。进入数据收集配置页: 在项目主页的顶部导航栏点击Configure Data Collection配置数据收集。编辑任务模板: 在弹出的列表中找到我们在 Day 4 创建的PEAK AI Hunter。点击这行最右侧的Edit编辑按钮。直达代码编辑区: 此时会弹出向导窗口直接点击顶部的第三个标签页Define Test定义与测试。此时你左侧看到的是参数输入区右侧是 AOB 自带的Code Editor代码编辑器。 Step 2: 注入 SDK 抽样逻辑 (Define Test 页面)在右侧的Code Editor中我们需要做两处核心修改。第一处在文件最顶部导入 SDK滚动到代码的第一行在原有的import列表下方加入 Splunk 官方 SDK 库绑定AOB 环境已自带无需安装importosimportsysimporttimeimportdatetimeimportjson# NEW: Import Splunk SDK bindingsimportsplunklib.clientasclientimportsplunklib.resultsasresults第二处重写 collect_events 函数向下滚动找到def collect_events(helper, ew):这一行。将其下方的代码完全替换为以下逻辑⚠️ 极客铁律: 代码注释与日志严格使用纯英文。defcollect_events(helper,ew):Implement your data collection logic herehelper.log_info(PEAK AI Hunter started data collection cycle.)try:# 1. Robustly acquire the session_key from the modular input contextsession_keygetattr(helper,session_key,None)# Fallback for certain AOB versions: extract from internal input_definitionifnotsession_keyandhasattr(helper,_input_definition):session_keygetattr(helper._input_definition,metadata,{}).get(session_key)ifnotsession_key:helper.log_error(Failed to acquire session_key from context.)return# 2. Connect to the Splunk service using the SDKserviceclient.Service(tokensession_key)# 3. Dynamically hijack the user-selected Index# We read the index the user configured in the UI during Day 5target_indexhelper.get_output_index()ormain# 4. Construct the randomized baseline SPL with a specific time window# Geek Tip: We use earliest-24h to limit the time window, and head 10000 BEFORE random() to prevent timeouts.search_query(fsearch index{target_index}earliest-24h latestnow f| head 10000 | eval rand_valrandom() | sort 5 rand_val)# 5. Use jobs.oneshot for fast, synchronous retrieval in JSON formatkwargs_oneshot{output_mode:json}helper.log_info(fExecuting randomized oneshot SPL:{search_query})oneshot_search_resultsservice.jobs.oneshot(search_query,**kwargs_oneshot)# 6. Parse the results using the SDKs built-in JSON readerreaderresults.JSONResultsReader(oneshot_search_results)baseline_logs[]forresultinreader:ifisinstance(result,dict)and_rawinresult:baseline_logs.append(result[_raw])# 7. Log the extracted data length and print a sample to verify successiflen(baseline_logs)0:helper.log_info(fSuccessfully fetched{len(baseline_logs)}random raw logs from index {target_index}.)helper.log_info(fSample Random Log [1]:{baseline_logs[0][:200]}...)else:helper.log_warning(fSPL executed successfully, but returned 0 events from index {target_index}.)exceptExceptionase:helper.log_error(fExecution error occurred during Splunk SDK search:{str(e)})⚡ Step 3: 一键防崩测试与保存AOB 的代码编辑器自带了一个极客友好的防崩溃测试功能。详细操作路径在页面左侧的Data input parameters区域随便填入必须填的参数比如输入SDK Test。点击页面右上角绿色的Test按钮。立即向下滚动网页查看编辑器底部的Output (输出)面板。 测试成功标志此时AOB 会在后台静默运行你的代码。只要 Output 面板最后显示了一个绿色的Done并且没有弹出大段红色的报错代码Traceback这就证明你的 Python 语法完美无缺且 SDK 成功向系统发起了查询(注AOB 引擎会将helper.log_info直接写入底层的真实日志文件因此你在前端面板只会看到 Done这是企业级规范。)固化代码 (Save) —— 极其重要: 测试没报错后必须点击右上角绿色的Save按钮保存代码然后关闭向导窗口。 Step 4: 全局验证 —— 回到 Splunk 检索真正的样本数据我们在 AOB 里验证了代码“不会崩”现在要去系统的最底层去验收我们抓取到的“战利品”。点击左上角的Splunkenterprise徽标返回 Splunk 首页。进入Search Reporting搜索与报表应用。我们的代码被调度器接管后每隔 60 秒就会跑一次。在搜索框中输入以下极客搜索命令去内部日志里抓取结果index_internal PEAK AI Hunter started data collection cycle OR Sample Random Log将时间范围调整为Last 15 minutes点击搜索。 终极成功标志你会看到不仅有“开始采集”的心跳日志每隔 60 秒还会稳定地输出一次Successfully fetched 5 random raw logs...并且紧跟着一条Sample Random Log [1]: ...。由于我们加了random()函数你会发现每次打印出来的样本内容都是不一样的这证明你写的代码不仅极其强壮而且正在源源不断地为 AI 采集真实的、不同切面的随机数据样本明天Day 7我们将把这些动态抽取的真实随机日志配合极其严酷的 System Prompt系统提示词正式发往大模型。让 AI 化身资深安全专家它将观察这些正常的系统日志并像黑客一样逆向思考推测出系统可能遭受的攻击方式进入真正的威胁狩猎模式
