diff --git a/Automation/Handle/playbook.py b/Automation/Handle/playbook.py index 73c6730..0e4789e 100644 --- a/Automation/Handle/playbook.py +++ b/Automation/Handle/playbook.py @@ -21,6 +21,10 @@ class Playbook(object): # 获取模块实例 if name is not None and type is not None: module_config = Xcache.get_module_config_by_name_and_type(type, name) + if module_config is None: + # try again to load all module config + Playbook.load_all_module_config() + module_config = Xcache.get_module_config_by_name_and_type(type, name) if module_config is None: context = data_return(305, {"status": "Failed", "job_id": None}, Playbook_MSG_ZH.get(305), Playbook_MSG_EN.get(305)) return context diff --git a/DATA/Case_L3_SOC_Analyst_Agent/L3_SOC_Analyst.md b/DATA/Case_L3_SOC_Analyst_Agent/L3_SOC_Analyst.md new file mode 100644 index 0000000..fa88423 --- /dev/null +++ b/DATA/Case_L3_SOC_Analyst_Agent/L3_SOC_Analyst.md @@ -0,0 +1,72 @@ +# 角色 (Role) + +你是一个高级网络安全运营中心（SOC）分析专家，拥有多年的NDR（网络检测和响应）、EDR（端点检测和响应）及XDR（扩展检测与响应）事件分析和应急响应经验。你的核心专长是处理 +**任何复杂类型**的攻击事件，尤其是通过关联多个异构告警（Alerts）来评估一个聚合案件（Case）的真实风险和严重性。 + +# 核心任务 (Core Task) + +你的唯一任务是：分析一个“案件”（Case）对象。这个Case刚刚**被挂载了一个新的告警（Alert）**。你需要智能地评估这个新告警对整个Case的影响，并以 +**严格的JSON格式**输出你的分析结论，包括对该Case的**最终严重性（Severity）**、**置信度（Confidence）**、**当前攻击阶段**和* +*建议行动**的判断。 + +# 背景上下文 (Context) + +1. **数据结构:** + * 一个 `Case` 是一个或多个异构 `Alert` 的集合。 + * 一个 `Alert` 包含一个或多个 `Artifact` (如IP地址, 域名, 文件哈希, 进程路径, 邮箱地址等)。 +2. **触发机制:** + * 当一个新的 `Alert` 产生时，系统规则会判断它是否应归属(挂载)到一个已有的 `Case`。 + * 你现在收到的 `Case` 对象，是**已经完成了新 `Alert` 挂载**的最新状态。 +3. **关键挑战:** + * Case的严重性是动态的。新挂载的 `Alert` 可能会**显著提升、降低或改变**Case的严重性。 + * 你必须判断新告警是“噪声”（重复或良性告警）还是“信号”（攻击升级或新证据）。 + +# 输入数据 (Input Data) + +你将收到一个 `Case` 对象（通常是JSON格式）。这个对象包含了该Case的**所有信息**： + +* Case ID +* 当前（旧的）Case严重性 (例如：`original_severity`) +* 包含在Case中的所有 `Alerts` 列表（**请注意，这个列表已经包含了那个新挂载的Alert**） +* 每个 `Alert` 的详细信息（包括告警来源、类型、时间戳和关联的`Artifacts`）。 + +**注意：** 你需要自行推断哪个是“最新”的Alert（通常是时间戳最晚的那个），并以此作为分析起点。 + +# 智能分析指南 (Intelligent Analysis Guidelines) + +在分析时，你必须像一个真人专家一样思考，重点关注以下几点，以生成JSON输出所需的所有字段： + +1. **识别新证据 (New Evidence):** 快速定位新挂载的 `Alert` 及其提供的新的、独特的 `Artifacts` 或事件类型。 +2. **上下文关联 (Contextualization):** + * 将新 `Alert` 与Case中**已有的所有 `Alerts`** 进行跨数据源（NDR, EDR, Email等）的比较。 + * 新 `Alert` 是否是已有 `Alert` 所揭示攻击的**逻辑下一步**（例如，在网络C2连接之后出现了端点的进程注入）？ +3. **攻击链 (Kill Chain) 分析:** + * 新 `Alert` 是否使整个Case的攻击链（应参考**MITRE ATT\&CK框架**的Tactic/Technique）**前进了一步**？这是判断严重性和攻击阶段的核心依据。 + * **示例（关键！）:** 如果现有证据是“初始访问”，新告警揭示了“凭据访问”，则 `current_attack_stage` 必须更新，且 + `new_severity` **应大幅提升**。 + * `current_attack_stage` 应使用最能代表当前整个案件状态的**MITRE ATT\&CK Tactic**名称或关键攻击行为。 +4. **置信度 (Confidence) 评估:** + * 如果新 `Alert` 只是Case中已有 `Alert` 的**重复或噪音**（例如，重复的扫描告警），`new_severity` 可能不变甚至降低（如果发现是良性行为），但 + `confidence` **通常应提升**（确信度更高）。 + * 如果新 `Alert` 提供了全新的、吻合攻击逻辑的**异构证据**（如网络告警被端点告警证实），`confidence` 应**大幅提升** + （例如，从"中"到"高"）。 +5. **确定建议行动 (Recommended Actions):** + * 基于你评估的 `new_severity` 和 `current_attack_stage`，推导出**最关键、最紧急**的响应动作。行动应直接、具体（例如， + `Isolate host X.X.X.X`，`Reset user Y password`）。 + +# 输出格式 (Output Format) + +你的**所有输出**必须是一个**单一、有效、可解析的JSON对象**。**不要**在JSON代码块之外添加任何解释性文本、开场白或总结。 + +**JSON结构定义:** + +```json +{{ + "original_severity": "string (Low/Medium/High/Critical)", + "new_severity": "string (Low/Medium/High/Critical)", + "analysis_rationale": "string", + "confidence": "string (Low/Medium/High)", + "current_attack_stage": "string (e.g., 'T1059 - Command and Control', 'Lateral Movement', 'Persistence')", + "recommended_actions": "string (e.g., 'Isolate host 10.1.1.5 and begin forensic analysis')" +}} +``` \ No newline at end of file diff --git a/Docker/Img/img.png b/Docker/Img/img.png new file mode 100644 index 0000000..8b606e4 Binary files /dev/null and b/Docker/Img/img.png differ diff --git a/Docker/Img/img_20.png b/Docker/Img/img_20.png new file mode 100644 index 0000000..7b82239 Binary files /dev/null and b/Docker/Img/img_20.png differ diff --git a/Docker/Img/img_21.png b/Docker/Img/img_21.png new file mode 100644 index 0000000..d960dbc Binary files /dev/null and b/Docker/Img/img_21.png differ diff --git a/Docker/Img/img_22.png b/Docker/Img/img_22.png new file mode 100644 index 0000000..0cec809 Binary files /dev/null and b/Docker/Img/img_22.png differ diff --git a/Docker/Img/img_23.png b/Docker/Img/img_23.png new file mode 100644 index 0000000..0f3baaf Binary files /dev/null and b/Docker/Img/img_23.png differ diff --git a/Docker/Img/img_24.png b/Docker/Img/img_24.png new file mode 100644 index 0000000..c5cee5e Binary files /dev/null and b/Docker/Img/img_24.png differ diff --git a/Docker/mock/SIRP_MOCK/nocoly_debug.py b/Docker/mock/SIRP_MOCK/nocoly_debug.py index 999486a..f02dd3f 100644 --- a/Docker/mock/SIRP_MOCK/nocoly_debug.py +++ b/Docker/mock/SIRP_MOCK/nocoly_debug.py @@ -1,12 +1,36 @@ from Docker.mock.SIRP_MOCK.alert import get_mock_alerts from Docker.mock.SIRP_MOCK.rule import rule_list from Lib.api import get_current_time_string, string_to_timestamp, generate_four_random_timestamps -from Lib.grouprule import GroupRule +from PLUGINS.SIRP.grouprule import GroupRule from PLUGINS.SIRP.nocolyapi import OptionSet -from PLUGINS.SIRP.sirpapi import Alert, Artifact, Case +from PLUGINS.SIRP.sirpapi import Alert, Artifact, Case, create_alert_with_group_rule -if __name__ == "__main__": +def import_alerts(): + alert_list = get_mock_alerts() + ALL_RULES = {} + for rule in rule_list: + ALL_RULES[rule.rule_id] = rule + + for alert in alert_list: + rule_def: GroupRule = ALL_RULES.get(alert["rule_id"]) + if rule_def is None: + print(f"未找到规则定义，跳过处理此告警: {alert['rule_id']}") + continue + + default_times = generate_four_random_timestamps() + alert_date = default_times["alert_date"] + created_date = default_times["created_date"] + acknowledged_date = default_times["acknowledged_date"] + closed_date = default_times["closed_date"] + + alert["name"] = alert["rule_name"] + alert["alert_date"] = alert_date + + case_row_id = create_alert_with_group_rule(alert, rule_def) + + +def old(): # print("使用自定义参数生成时间戳:") # custom_times = generate_four_random_timestamps( # days_ago_max=5, # T1 在当前时间前 5 天内 @@ -77,6 +101,8 @@ if __name__ == "__main__": row_id_list = Artifact.update_or_create(fields, artifact_filter) artifact_rowid_list.extend(row_id_list) + case_row_id = create_alert_with_group_rule(input_alert, rule) + alert_fields = [ {"id": "tags", "value": alert["tags"], "type": 2}, {"id": "severity", "value": alert["severity"]}, @@ -91,7 +117,7 @@ if __name__ == "__main__": {"id": "artifact", "value": artifact_rowid_list}, ] # alert - row_id_alert = Alert.create(alert_fields) + row_id_alert = Alert.create(alert) print(f"create alert: {row_id_alert}") # case @@ -160,3 +186,7 @@ if __name__ == "__main__": ] row_id_updated = Case.update(row_id_case, case_field) print(f"update case: {row_id_updated}") + + +if __name__ == "__main__": + import_alerts() diff --git a/Docker/mock/SIRP_MOCK/rule.py b/Docker/mock/SIRP_MOCK/rule.py index 9e47a68..37ebe37 100644 --- a/Docker/mock/SIRP_MOCK/rule.py +++ b/Docker/mock/SIRP_MOCK/rule.py @@ -1,4 +1,4 @@ -from Lib.grouprule import GroupRule +from PLUGINS.SIRP.grouprule import GroupRule rule_list = [ GroupRule( diff --git a/Forwarder/views.py b/Forwarder/views.py index 0923c14..afa8fff 100644 --- a/Forwarder/views.py +++ b/Forwarder/views.py @@ -20,7 +20,6 @@ class WebhookSplunkView(BaseView): app = request.data.get('app') owner = request.data.get('owner') results_link = request.data.get('results_link') - app = request.data.get('app') logger.debug(f"Splunk webhook: {request.data}") redis_stream_api = RedisStreamAPI() redis_stream_api.send_message(search_name, result) @@ -62,6 +61,7 @@ class WebhookNocolyMailView(BaseView): def create(self, request, **kwargs): try: data = request.data + logger.info(data) context = data_return(200, {}, CODE_MSG_ZH.get(200), CODE_MSG_EN.get(200)) return Response(context) except Exception as E: diff --git a/Lib/basemodule.py b/Lib/basemodule.py index 2dce81a..f130a88 100644 --- a/Lib/basemodule.py +++ b/Lib/basemodule.py @@ -43,7 +43,7 @@ class LanggraphModule(BaseModule): config = RunnableConfig() config["configurable"] = {"thread_id": self.module_name} if self.agent_state is None: - self.agent_state = AgentState(messages=[], alert_raw={}, temp_data={}, analyze_result={}) + self.agent_state = AgentState() for event in self.graph.stream(self.agent_state, config, stream_mode="values"): self.logger.debug(event) diff --git a/Lib/baseplaybook.py b/Lib/baseplaybook.py index 410a864..101ffba 100644 --- a/Lib/baseplaybook.py +++ b/Lib/baseplaybook.py @@ -37,7 +37,7 @@ class LanggraphPlaybook(BasePlaybook): config = RunnableConfig() config["configurable"] = {"thread_id": self.module_name} if self.agent_state is None: - self.agent_state = AgentState(messages=[], alert_raw={}, temp_data={}, analyze_result={}) + self.agent_state = AgentState() for event in self.graph.stream(self.agent_state, config, stream_mode="values"): self.logger.debug(event) diff --git a/Lib/engine.py b/Lib/engine.py index 9d9fdcd..6918b82 100644 --- a/Lib/engine.py +++ b/Lib/engine.py @@ -23,11 +23,14 @@ class Engine: logger.info("All modules have been stopped") def _load_initial_modules(self): + load_module_count = 0 for filename in os.listdir(self.modules_dir): if filename.endswith(".py") and not filename.startswith("_"): module_name = filename.replace(".py", "") file_path = os.path.join(self.modules_dir, filename) self.load_module(module_name, file_path) + load_module_count += 1 + logger.info(f"Loaded {load_module_count} modules from '{self.modules_dir}' directory") def run_loop(self, module_name, instance): while module_name in self.modules: @@ -43,7 +46,7 @@ class Engine: if module_name in self.modules: return - logger.info(f"Loading module: {module_name}") + logger.debug(f"Loading module: {module_name}") try: spec = importlib.util.spec_from_file_location(module_name, file_path) module = importlib.util.module_from_spec(spec) @@ -64,7 +67,7 @@ class Engine: for thread in self.modules[module_name]: thread.start() - logger.info(f"Module '{module_name}' started successfully") + logger.debug(f"Module '{module_name}' started successfully") except Exception as e: logger.error(f"Failed to load module: {e}") diff --git a/Lib/llmapi.py b/Lib/llmapi.py index f343950..cca88dd 100644 --- a/Lib/llmapi.py +++ b/Lib/llmapi.py @@ -5,7 +5,9 @@ from pydantic import BaseModel class AgentState(BaseModel): - messages: Annotated[List[Any], add_messages] - alert_raw: Dict[str, Any] - temp_data: Dict[str, Any] - analyze_result: Dict[str, Any] + messages: Annotated[List[Any], add_messages] = [] + case: Dict[str, Any] = {} + alert: Dict[str, Any] = {} + artifact: Dict[str, Any] = {} + temp_data: Dict[str, Any] = {} + analyze_result: Dict[str, Any] = {} diff --git a/MODULES/EDR-Rule-11-Suspicious-Process-Spawned-by-Office.py b/MODULES/EDR-Rule-11-Suspicious-Process-Spawned-by-Office.py index 98ca6b9..196256d 100644 --- a/MODULES/EDR-Rule-11-Suspicious-Process-Spawned-by-Office.py +++ b/MODULES/EDR-Rule-11-Suspicious-Process-Spawned-by-Office.py @@ -10,9 +10,9 @@ from pydantic import BaseModel, Field from Lib.api import string_to_string_time, get_current_time_string from Lib.basemodule import LanggraphModule -from Lib.grouprule import GroupRule from Lib.llmapi import AgentState from PLUGINS.LLM.llmapi import LLMAPI +from PLUGINS.SIRP.grouprule import GroupRule from PLUGINS.SIRP.sirpapi import create_alert_with_group_rule, InputAlert @@ -48,7 +48,7 @@ class Module(LanggraphModule): "Authentication-Results": headers["Authentication-Results"]} alert["headers"] = headers - state.alert_raw = alert + state.alert = alert return state # Define nodes @@ -124,7 +124,7 @@ class Module(LanggraphModule): messages = [ system_message, *few_shot_examples, - HumanMessage(content=json.dumps(state.alert_raw)), + HumanMessage(content=json.dumps(state.alert)), ] # Execute @@ -140,7 +140,7 @@ class Module(LanggraphModule): def alert_output_node(state: AgentState): """Process analysis results""" analyze_result: AnalyzeResult = AnalyzeResult(**state.analyze_result) - alert_raw = state.alert_raw + alert_raw = state.alert mail_to = alert_raw["headers"]["To"] mail_subject = alert_raw["headers"]["Subject"] @@ -151,7 +151,7 @@ class Module(LanggraphModule): severity = "Info" # splunk webhook timestamp format: 2024-10-10T12:34:56.789Z - # alert_date = string_to_string_time(alert_raw.get("@timestamp"), "%Y-%m-%dT%H:%M:%S.%fZ", "%Y-%m-%dT%H:%M:%SZ") + # alert_date = string_to_string_time(alert.get("@timestamp"), "%Y-%m-%dT%H:%M:%S.%fZ", "%Y-%m-%dT%H:%M:%SZ") alert_date = string_to_string_time(alert_raw.get("headers").get("Date"), "%a, %d %b %Y %H:%M:%S %z", "%Y-%m-%dT%H:%M:%SZ") description = f""" ## Analyze Result (AI) diff --git a/MODULES/ES-Rule-21-Phishing_user_report_mail.py b/MODULES/ES-Rule-21-Phishing_user_report_mail.py index 79f8cf4..63a44fc 100644 --- a/MODULES/ES-Rule-21-Phishing_user_report_mail.py +++ b/MODULES/ES-Rule-21-Phishing_user_report_mail.py @@ -10,9 +10,9 @@ from pydantic import BaseModel, Field from Lib.api import string_to_string_time, get_current_time_string from Lib.basemodule import LanggraphModule -from Lib.grouprule import GroupRule from Lib.llmapi import AgentState from PLUGINS.LLM.llmapi import LLMAPI +from PLUGINS.SIRP.grouprule import GroupRule from PLUGINS.SIRP.sirpapi import create_alert_with_group_rule, InputAlert @@ -55,7 +55,7 @@ class Module(LanggraphModule): alert["headers"] = headers # Store the preprocessed alert in workflow state for downstream nodes - state.alert_raw = alert + state.alert = alert return state def alert_analyze_node(state: AgentState): @@ -131,7 +131,7 @@ class Module(LanggraphModule): messages = [ system_message, *few_shot_examples, - HumanMessage(content=json.dumps(state.alert_raw)), + HumanMessage(content=json.dumps(state.alert)), ] # Invoke LLM with structured output schema for consistent analysis results @@ -151,7 +151,7 @@ class Module(LanggraphModule): Generate alert ticket and send to SIRP for incident tracking and investigation. """ analyze_result: AnalyzeResult = AnalyzeResult(**state.analyze_result) - alert_raw = state.alert_raw + alert_raw = state.alert mail_to = alert_raw["headers"]["To"] mail_subject = alert_raw["headers"]["Subject"] @@ -209,20 +209,17 @@ class Module(LanggraphModule): { "type": "mail_to", "value": mail_to, - # "deduplication_key": f"mail_to-{mail_to}", # Enrichment metadata (TI lookup results, asset info, etc.) "enrichment": {"update_time": get_current_time_string()} }, { "type": "mail_subject", "value": mail_subject, - # "deduplication_key": f"mail_subject-{mail_subject}", "enrichment": {"update_time": get_current_time_string()} }, { "type": "mail_from", "value": mail_from, - # "deduplication_key": f"mail_from-{mail_from}", "enrichment": {"update_time": get_current_time_string()} }, ], diff --git a/MODULES/ES-Rule-22-Phishing_user_report_mail.py b/MODULES/ES-Rule-22-Phishing_user_report_mail.py index 787317d..cbcf206 100644 --- a/MODULES/ES-Rule-22-Phishing_user_report_mail.py +++ b/MODULES/ES-Rule-22-Phishing_user_report_mail.py @@ -6,8 +6,8 @@ from pydantic import BaseModel, Field from Lib.api import string_to_string_time, get_current_time_string from Lib.basemodule import BaseModule -from Lib.grouprule import GroupRule from PLUGINS.Dify.dify import Dify +from PLUGINS.SIRP.grouprule import GroupRule from PLUGINS.SIRP.sirpapi import create_alert_with_group_rule, InputAlert @@ -46,7 +46,7 @@ class Module(BaseModule): client = Dify() api_key = client.get_dify_api_key(self.module_name) inputs = { - "alert_raw": json.dumps(self.agent_state.alert_raw) + "alert": json.dumps(self.agent_state.alert_raw) } result = client.run_workflow( @@ -96,19 +96,16 @@ class Module(BaseModule): { "type": "mail_to", "value": mail_to, - # "deduplication_key": f"mail_to-{mail_to}", "enrichment": {"update_time": get_current_time_string()} # just for test, no meaning, data should come from TI or other cmdb }, { "type": "mail_subject", "value": mail_subject, - # "deduplication_key": f"mail_subject-{mail_subject}", "enrichment": {"update_time": get_current_time_string()} }, { "type": "mail_from", "value": mail_from, - # "deduplication_key": f"mail_from-{mail_from}", "enrichment": {"update_time": get_current_time_string()} }, ], diff --git a/MODULES/NDR-Rule-05-Suspect-C2-Communication.py b/MODULES/NDR-Rule-05-Suspect-C2-Communication.py index 9e3bd5e..5ce2916 100644 --- a/MODULES/NDR-Rule-05-Suspect-C2-Communication.py +++ b/MODULES/NDR-Rule-05-Suspect-C2-Communication.py @@ -9,9 +9,9 @@ from pydantic import BaseModel, Field, ConfigDict from Lib.api import get_current_time_string from Lib.basemodule import LanggraphModule -from Lib.grouprule import GroupRule from Lib.llmapi import AgentState from PLUGINS.LLM.llmapi import LLMAPI +from PLUGINS.SIRP.grouprule import GroupRule from PLUGINS.SIRP.sirpapi import create_alert_with_group_rule, InputAlert, Case @@ -59,7 +59,7 @@ class Module(LanggraphModule): if alert is None: return - state.alert_raw = alert + state.alert = alert artifact: list = alert.get("artifact") alert_date: str = alert.get("alert_date") rule_name = "Suspicious command and control (C2) communication" diff --git a/PLAYBOOKS/Alert_Suggestion_Gen_By_LLM.py b/PLAYBOOKS/Alert_Suggestion_Gen_By_LLM.py index c7274b0..45ac679 100644 --- a/PLAYBOOKS/Alert_Suggestion_Gen_By_LLM.py +++ b/PLAYBOOKS/Alert_Suggestion_Gen_By_LLM.py @@ -22,9 +22,9 @@ class AgentState(BaseModel): class Playbook(LanggraphPlaybook): - RUN_AS_JOB = True - TYPE = "ALERT" - NAME = "Suggestion Generation by LLM" + RUN_AS_JOB = True # 异步模块 + TYPE = "ALERT" # 分类标签 + NAME = "Suggestion Generation by LLM" # 剧本名称 def __init__(self): super().__init__() # do not delete this code @@ -35,13 +35,11 @@ class Playbook(LanggraphPlaybook): """预处理数据""" # worksheet = self.param("worksheet") rowid = self.param("rowid") - alert = WorksheetRow.get(Alert.WORKSHEET_ID, rowid, include_system_fields=False) + worksheet = self.param("worksheet") + alert = WorksheetRow.get(worksheet, rowid, include_system_fields=False) artifacts = WorksheetRow.relations(Alert.WORKSHEET_ID, alert.get("rowId"), "artifact", relation_worksheet_id=Artifact.WORKSHEET_ID, include_system_fields=False) alert["artifact"] = artifacts - - Notice.send(self.param("user"), "Alert_Suggestion_Gen_By_LLM preprocess_node Finish", f"rowid：{self.param('rowid')}") - state.alert = alert return state @@ -82,8 +80,6 @@ class Playbook(LanggraphPlaybook): response = llm.invoke(messages) response = LLMAPI.extract_think(response) # langchain chatollama bug临时方案 state.suggestion = response.content - - Notice.send(self.param("user"), "Alert_Suggestion_Gen_By_LLM analyze_node Finish", f"rowid：{self.param('rowid')}") return state def output_node(state: AgentState): diff --git a/PLAYBOOKS/Case_L3_SOC_Analyst_Agent.py b/PLAYBOOKS/Case_L3_SOC_Analyst_Agent.py new file mode 100644 index 0000000..d25ab17 --- /dev/null +++ b/PLAYBOOKS/Case_L3_SOC_Analyst_Agent.py @@ -0,0 +1,164 @@ +import json +from enum import Enum +from typing import Optional, Union, Dict, Any + +from langchain_core.messages import HumanMessage +from langgraph.graph import StateGraph +from langgraph.graph.state import CompiledStateGraph +from pydantic import BaseModel, Field, ConfigDict + +from Lib.baseplaybook import LanggraphPlaybook +from Lib.llmapi import AgentState +from PLUGINS.LLM.llmapi import LLMAPI +from PLUGINS.SIRP.nocolyapi import WorksheetRow +from PLUGINS.SIRP.sirpapi import Alert, Artifact +from PLUGINS.SIRP.sirpapi import Case +from PLUGINS.SIRP.sirpapi import Notice +from PLUGINS.SIRP.sirpapi import Playbook as SIRPPlaybook + + +class ConfidenceLevel(str, Enum): + """置信度等级""" + LOW = "Low" + MEDIUM = "Medium" + HIGH = "High" + + +class Severity(str, Enum): + """置信度等级""" + INFO = "Info" + LOW = "Low" + MEDIUM = "Medium" + HIGH = "High" + CRITICAL = "Critical" + + +class AnalyzeResult(BaseModel): + """Structure for extracting user information from text""" + # config + model_config = ConfigDict(use_enum_values=True) + + original_severity: Severity = Field(description="Original alert severity") + new_severity: Severity = Field(description="Recommended new severity level") + confidence: ConfidenceLevel = Field(description="Confidence score, only one of 'Low', 'Medium', or 'High'") + analysis_rationale: str = Field(description="Analysis process and reasons", default=None) + current_attack_stage: Optional[Union[str, Dict[str, Any]]] = Field(description="e.g., 'T1059 - Command and Control', 'Lateral Movement'", default=None) + recommended_actions: Optional[Union[str, Dict[str, Any]]] = Field(description="e.g., 'Isolate host 10.1.1.5'", default=None) + + +class Playbook(LanggraphPlaybook): + RUN_AS_JOB = True + TYPE = "CASE" + NAME = "L3 SOC Analyst Agent" + + def __init__(self): + super().__init__() # do not delete this code + self.init() + + def init(self): + def preprocess_node(state: AgentState): + """预处理数据""" + # worksheet = self.param("worksheet") + rowid = self.param("rowid") + + case = WorksheetRow.get(Case.WORKSHEET_ID, rowid, include_system_fields=False) + + alerts = WorksheetRow.relations(Case.WORKSHEET_ID, rowid, "alert", relation_worksheet_id=Alert.WORKSHEET_ID, include_system_fields=False) + for alert in alerts: + artifacts = WorksheetRow.relations(Alert.WORKSHEET_ID, alert.get("rowId"), "artifact", relation_worksheet_id=Artifact.WORKSHEET_ID, + include_system_fields=False) + alert["artifact"] = artifacts + case["alert"] = alerts + state.case = case + return state + + # 定义node + def analyze_node(state: AgentState): + """AI分析Case数据""" + + # 加载system prompt + system_prompt_template = self.load_system_prompt_template("L3_SOC_Analyst") + + system_message = system_prompt_template.format() + + # 构建few-shot示例 + few_shot_examples = [ + # HumanMessage( + # content=json.dumps({ + # "requirement": ".", + # }) + # ), + # AIMessage( + # content=json.dumps({ + # "function": "the amount of pneumothorax", + # }) + # ), + ] + + # 运行 + llm_api = LLMAPI() + + llm = llm_api.get_model() + + # 构建消息列表 + messages = [ + system_message, + *few_shot_examples, + HumanMessage(content=json.dumps(state.case)) + ] + llm = llm.with_structured_output(AnalyzeResult) + response: AnalyzeResult = llm.invoke(messages) + state.analyze_result = response.model_dump() + + # response = llm.invoke(messages) + # response = LLMAPI.extract_think(response) # langchain chatollama bug临时方案 + # state.analyze_result = json.loads(response.content) + return state + + def output_node(state: AgentState): + """处理分析结果""" + + analyze_result: AnalyzeResult = AnalyzeResult(**state.analyze_result) + + case_row_id = self.param("rowid") + + case_field = [ + {"id": "severity", "value": analyze_result.new_severity}, + {"id": "confidence_ai", "value": analyze_result.confidence}, + {"id": "analysis_rationale_ai", "value": analyze_result.analysis_rationale}, + {"id": "attack_stage_ai", "value": analyze_result.current_attack_stage}, + {"id": "recommended_actions_ai", "value": analyze_result.recommended_actions}, + ] + + Case.update(case_row_id, case_field) + + Notice.send(self.param("user"), "Case_L3_SOC_Analyst_Agent Finish", f"rowid：{self.param('rowid')}") + + SIRPPlaybook.update_status_and_remark(self.param("playbook_rowid"), "Success", "Get suggestion by ai agent completed.") # Success/Failed + return state + + # 编译graph + workflow = StateGraph(AgentState) + + workflow.add_node("preprocess_node", preprocess_node) + workflow.add_node("analyze_node", analyze_node) + workflow.add_node("output_node", output_node) + + workflow.set_entry_point("preprocess_node") + workflow.add_edge("preprocess_node", "analyze_node") + workflow.add_edge("analyze_node", "output_node") + workflow.set_finish_point("output_node") + self.agent_state = AgentState() + self.graph: CompiledStateGraph = workflow.compile(checkpointer=self.get_checkpointer()) + return True + + def run(self): + self.run_graph() + return + + +if __name__ == "__main__": + params_debug = {'rowid': '27ca1468-5e3b-46ca-b238-3308539241e1', 'worksheet': 'case'} + module = Playbook() + module._params = params_debug + module.run() diff --git a/PLUGINS/SIRP/SIRP.mdy b/PLUGINS/SIRP/SIRP.mdy index f1f83e6..8cfc43c 100644 Binary files a/PLUGINS/SIRP/SIRP.mdy and b/PLUGINS/SIRP/SIRP.mdy differ diff --git a/Lib/grouprule.py b/PLUGINS/SIRP/grouprule.py similarity index 86% rename from Lib/grouprule.py rename to PLUGINS/SIRP/grouprule.py index 508262d..347f515 100644 --- a/Lib/grouprule.py +++ b/PLUGINS/SIRP/grouprule.py @@ -7,7 +7,11 @@ class GroupRule(object): 告警聚合规则,用于定义如何将多个告警(alert)聚合到同一个案件(case)中。 基本方法: 根据规则ID、告警中的凭据(artifact)以及时间窗口，生成去重指纹(deduplication key)。还有相同指纹的告警会被聚合到同一个案件中。 还可以定义案件标题的模板，以便生成更具描述性的案件标题。 +<<<<<<<< HEAD:Lib/grouprule.py 方法在心智成本和聚合效果之间取得平衡，适用于绝大多数常见的告警聚合场景。 +======== + 方法在心智成本和聚合效果之间取得平衡，适用于绝大多数常见地告警聚合场景。 +>>>>>>>> origin/master:PLUGINS/SIRP/grouprule.py """ def __init__(self, @@ -17,7 +21,13 @@ class GroupRule(object): case_title_template: str = None, deduplication_window: str = "24h", source: str = "Default", +<<<<<<<< HEAD:Lib/grouprule.py workbook: str = None +======== + workbook: str = None, + follow_alert_severity: bool = True, + append_alert_tags: bool = True, +>>>>>>>> origin/master:PLUGINS/SIRP/grouprule.py ): self.rule_id = rule_id @@ -26,6 +36,12 @@ class GroupRule(object): self.case_title_template = case_title_template self.source = source self.workbook = workbook +<<<<<<<< HEAD:Lib/grouprule.py +======== + self.follow_alert_severity = follow_alert_severity + self.append_alert_tags = append_alert_tags + +>>>>>>>> origin/master:PLUGINS/SIRP/grouprule.py valid_windows = ['10m', '30m', '1h', '8h', '12h', '24h'] if deduplication_window not in valid_windows: raise ValueError(f"'{deduplication_window}' 不是一个有效的时间窗口选项。请从 {valid_windows} 中选择。") diff --git a/PLUGINS/SIRP/sirpapi.py b/PLUGINS/SIRP/sirpapi.py index 9180d91..26ec1ee 100644 --- a/PLUGINS/SIRP/sirpapi.py +++ b/PLUGINS/SIRP/sirpapi.py @@ -5,30 +5,13 @@ from typing import TypedDict, List, Optional, Union, Dict, Any, NotRequired import requests from Lib.api import string_to_timestamp, get_current_time_string -from Lib.grouprule import GroupRule from Lib.log import logger from PLUGINS.Embeddings.embeddingsapi import EmbeddingsAPI from PLUGINS.SIRP.CONFIG import SIRP_NOTICE_WEBHOOK +from PLUGINS.SIRP.grouprule import GroupRule from PLUGINS.SIRP.nocolyapi import WorksheetRow, OptionSet -class InputAlert(TypedDict): - source: str - rule_id: str - rule_name: str - name: str - alert_date: str - created_date: str - tags: List[str] - severity: str - reference: NotRequired[str] - source_data_identifier: NotRequired[str] - description: str - summary_ai: NotRequired[Optional[Union[str, Dict[str, Any]]]] - artifact: List[Dict] - raw_log: NotRequired[Optional[Union[str, Dict[str, Any]]]] - - class InputCase(TypedDict): """ 需要和SIRP的Case表结构保持一致 @@ -53,6 +36,23 @@ class InputCase(TypedDict): alert: List[str] +class InputAlert(TypedDict): + source: str + rule_id: str + rule_name: str + name: str + alert_date: str + created_date: str + tags: List[str] + severity: str + reference: NotRequired[str] + source_data_identifier: NotRequired[str] + description: str + summary_ai: NotRequired[Optional[Union[str, Dict[str, Any]]]] + artifact: List[Dict] + raw_log: NotRequired[Optional[Union[str, Dict[str, Any]]]] + + class InputArtifact(TypedDict): type: str value: str @@ -112,15 +112,11 @@ class Alert(object): @staticmethod def create(alert: InputAlert): - # debug - - # debug - artifact_rowid_list = [] artifacts: list[dict] = alert.get("artifact", []) for artifact in artifacts: - if artifact.get("deduplication_key") is None: - artifact["deduplication_key"] = f"{artifact["type"]}-{artifact["value"]}" + # if artifact.get("deduplication_key") is None: + # artifact["deduplication_key"] = f"{artifact["type"]}-{artifact["value"]}" artifact_fields = [ {"id": "type", "value": artifact.get("type")}, @@ -150,6 +146,9 @@ class Alert(object): row_id_list = Artifact.update_or_create(artifact_fields, artifact_filter) artifact_rowid_list.extend(row_id_list) + if alert.get("created_date") is None: + alert["created_date"] = get_current_time_string() + alert_fields = [ {"id": "tags", "value": alert.get("tags"), "type": 2}, {"id": "severity", "value": alert.get("severity")}, @@ -333,11 +332,13 @@ def create_alert_with_group_rule(alert: InputAlert, rule_def: GroupRule) -> str: else: workbook = rule_def.workbook + case_status_new = OptionSet.get_option_key_by_name_and_value("case_status", "New") + case: InputCase = { "title": rule_def.generate_case_title(artifacts=artifacts), "deduplication_key": deduplication_key, "alert": [row_id_alert], - "case_status": "New", + "case_status": case_status_new, "created_date": get_current_time_string(), "tags": alert["tags"], "severity": alert["severity"], @@ -353,25 +354,30 @@ def create_alert_with_group_rule(alert: InputAlert, rule_def: GroupRule) -> str: if row_id_alert not in existing_alerts: existing_alerts.append(row_id_alert) - option_new_score = OptionSet.get_option_by_name_and_value("alert_case_severity", alert["severity"]).get("score", 0) - - severity_value_exist = row.get("severity") - option_exist_score = OptionSet.get_option_by_name_and_value("alert_case_severity", severity_value_exist).get("score", 0) - - if option_new_score > option_exist_score: - severity = alert["severity"] - else: - severity = severity_value_exist - - tags_exist = row.get("tags", []) - for tag in alert["tags"]: - if tag not in tags_exist: - tags_exist.append(tag) - case_field = [ {"id": "alert", "value": existing_alerts}, - {"id": "severity", "value": severity}, - {"id": "tags", "value": tags_exist, "type": 2} ] + + # change case severity if new alert severity is higher + if rule_def.follow_alert_severity: + option_new_score = OptionSet.get_option_by_name_and_value("alert_case_severity", alert["severity"]).get("score", 0) + + severity_value_exist = row.get("severity") + option_exist_score = OptionSet.get_option_by_name_and_value("alert_case_severity", severity_value_exist).get("score", 0) + + if option_new_score > option_exist_score: + severity = alert["severity"] + else: + severity = severity_value_exist + case_field.append({"id": "severity", "value": severity}) + + # append alert tags to case tags + if rule_def.append_alert_tags: + tags_exist = row.get("tags", []) + for tag in alert["tags"]: + if tag not in tags_exist: + tags_exist.append(tag) + case_field.append({"id": "tags", "value": tags_exist, "type": 2}) + row_id_updated = Case.update(row_id_case, case_field) return row_id_updated diff --git a/README.md b/README.md index 15a0bae..b229088 100644 --- a/README.md +++ b/README.md @@ -1,57 +1,61 @@ -# AI SOC Framework (ASF) +![cover-v5-optimized](Docker/Img/img.png) -基于 AI Agent 的 SOC 自动化框架. 开源, 灵活, 强大, 私有部署. +

+ Getting-started · + Documentation +

-## 功能 +

+ + Static Badge + + chat on Discord + + follow on X(Twitter) + + Commits last month + + Issues closed + + Release +

-- 基于 Webhook + Redis Stream 的告警流水线, 支持主流 SIEM 平台 -- 提供 Langchain/Langgraph/Dify 等 AI Agent 模块模板, 快速开发 Pre Automation/Post Automation 模块 -- 内置功能齐全的 SIRP 平台(AI SOAR), 支持 Artifact/Alert/Case 网络安全告警数据模型 -- AI SOAR 内置简单易用的告警聚合功能, 支持默认规则和自定义规则 -- 框架代码皆为 Python 编写, 易于二次开发和扩展 +

+ README in English + 简体中文版自述文件 +

-## 为什么使用 ASF & ASF 解决哪些问题 -- 预算 / 资源有限无法购买商用 SOAR 产品 +**AI SOC FRAMEWORK (ASF)** is a powerful, flexible, and open-source automated security operations platform. It integrates AI Agent capabilities with automation orchestration, supporting mainstream SIEM/SOAR scenarios to help enterprises efficiently build intelligent security operations systems. -> ASF 完全开源免费, 且支持对接社区版的 ELK(SIEM), 企业只需有基础的安全设备和日志采集能力即可构建完整的 SOC 基础设施 +## Core Features -- 所有网络安全相关数据不允许离开企业内网 +- 🧠 **AI-driven Intelligence**: Utilizes built-in AI Agent templates like Langgraph and Dify, supporting local LLMs to enhance alert analysis and automated response capabilities. +- 📊 **Built-in SIRP Platform**: Comes with a ready-to-use Security Incident Response Platform (SIRP) built on Nocoly, allowing for rapid customization of user interfaces, data models, reports, and workflows. +- ⚙️ **Powerful Automation Workflow**: Achieves efficient alert processing through Webhook + Redis Stream, natively supporting mainstream SIEM platforms such as Splunk and Kibana (ELK). +- 🛠️ **Highly Extensible**: Provides a rich library of modules and plugins. The entire framework is written in Python, facilitating secondary development and integration with various security devices and APIs. +- 🛡️ **Local Deployment & Data Control**: Supports complete local deployment. All data, models, and operations can be hosted within your own environment, ensuring enterprise data security and privacy. +- ⚡ **Streaming and Batch Processing**: Offers streaming processing (modules) for real-time alert analysis and event-driven automation (playbooks) for user-triggered tasks. -> ASF 所有组件 (AI SOAR/Redis Stack/Module Framework) 均可本地部署, -> 通过 vllm/ollama 等部署本地化的 LLM, 实现完全本地化的 AI Agent 能力 +## Architecture Overview -- 对于 SIRP 有大量定制化需求, 不限于个性化 UI, 定制化流程, 自定义数据模型, 数据报表等 +ASF processes security alerts and incidents through a simplified multi-stage process: -> AI SOAR 基于 [Nocoly](https://www.nocoly.com/) 构建, 无需编写代码即可实现定制化 UI 修改, 自定义工作流, 自定义报表等 +1. **SIEM/Alert Sources**: EDR, NDR, or other security tools send alerts to a SIEM (e.g., Splunk, Kibana). +2. **Webhook Forwarder**: The SIEM forwards these alerts via Webhook to the ASF's built-in Webhook receiver. +3. **Redis Stream**: The receiver pushes the alerts to the corresponding Redis Stream, serving as a persistent message queue. Each alert type has its own stream. +4. **Module Engine**: ASF **modules** consume alerts from their designated streams, perform analysis (often using AI Agents), enrich data, and determine outcomes. +5. **SIRP Platform**: The output of the modules (now formatted into standardized security records) is sent to the **SIRP** platform, where cases, alerts, and artifacts are created or updated. +6. **Playbook Engine**: Analysts can trigger **playbooks** from the SIRP user interface against cases, alerts, or artifacts to perform further automated actions, such as threat intelligence enrichment or remediation. -- 出于特定的安全业务需求或提高效率, 需要定制化的 AI Agent 分析告警 +![img_1.webp](Docker/Img/img_20.png) +![img_2.webp](Docker/Img/img_21.png) +![img_2.webp](Docker/Img/img_22.png) +![img_3.webp](Docker/Img/img_23.png) +![img_4.webp](Docker/Img/img_24.png) -> ASF 提供模块模板及样例模块, 用户可根据自身需求快速开发定制化的 AI Agent 模块, 支持多框架(Langchain/Langgraph/Dify 等) - -- 内部系统 / 设备接口众多, 需要额外的数据处理及转化, 主流的 SOAR(如 Swimlane/Splunk SOAR)或可视化编排产品 (n8n) 等无法满足需求 - -> ASF 的模块开发完全基于 Python, 用户可以使用任何 Python 库, 并且可以灵活地对接任何 API 或设备 - -- SOAR 的自动化剧本和自定义的自动化脚本无法调试 - -> ASF 中有用于调试的适配性代码, 用户可单独运行模块对指定告警 (Redis Insight 检索查看) 进行调试, 而无需启动整个框架 - -## 不适用于哪些场景 - -- 安全团队没有基础的 Python 开发能力 - -> ASF 不是开箱即用平台, 需要一定的 Python 开发能力来进行模块开发和定制化 - -- 企业使用单独厂商一揽子解决方案(如 XDR/MDR/MSS 等) - -> ASF 需要告警数据或日志来进行自动化分析,XDR/MDR/MSS 等封闭系统无法实现 - -- 企业没有基础的安全设备和日志采集能力 - -> 没有设备就没有日志和告警, 那也就不需要自动化 - -## 许可证 - -该项目采用 [MIT](https://choosealicense.com/licenses/mit/) 许可证。 +## Official Website +[https://asf.viperrtp.com/zh/](https://asf.viperrtp.com/zh/) diff --git a/README_ZH.md b/README_ZH.md new file mode 100644 index 0000000..4970308 --- /dev/null +++ b/README_ZH.md @@ -0,0 +1,64 @@ +![cover-v5-optimized](Docker/Img/img.png) + +

+ Getting-started · + Documentation +

+ +

+ + Static Badge + + chat on Discord + + follow on X(Twitter) + + Commits last month + + Issues closed + + Release +

+ +

+ README in English + 简体中文版自述文件 +

+ + +**AI SOC FRAMEWORK (ASF)** 是一个功能强大、灵活且开源的自动化安全运营平台。它集成了 AI Agent 能力与自动化编排，支持主流 +SIEM/SOAR 场景，帮助企业高效构建智能化安全运营体系。 + +## 核心功能 + +- 🧠 **AI 驱动智能**: 利用内置的 Langgraph 和 Dify 等 AI Agent 模板，支持本地 LLM，增强告警分析和自动化响应能力。 +- 📊 **内置 SIRP 平台**: 内置安全事件响应平台 (SIRP)，可快速定制开发用户界面、数据模型、报告和工作流。 +- ⚙️ **强大的自动化流程**: 通过 Webhook + Redis Stream 实现高效的告警处理流程，原生支持 Splunk 和 Kibana (ELK) 等主流 + SIEM 平台。 +- 🛠️ **高度可扩展性**: 提供丰富的模块和插件库。整个框架用 Python 编写，便于二次开发和与各类安全设备及 API 集成。 +- 🛡️ **本地部署与数据控制**: 支持完全本地化部署。所有数据、模型和操作都可以在您自己的环境中托管，确保企业数据安全和隐私。 +- ⚡ **流式与批量处理**: 提供用于实时告警分析的流式处理（模块）和用于用户触发任务（剧本）的事件驱动自动化。 + +## 架构概览 + +ASF 通过简化的多阶段流程处理安全告警和事件： + +1. **SIEM/告警源**: EDR、NDR 或其他安全工具将告警发送到 SIEM（例如 Splunk、Kibana）。 +2. **Webhook 转发器**: SIEM 通过 Webhook 将这些告警转发到 ASF 内置的 Webhook 接收器。 +3. **Redis Stream**: 接收器将告警推送到相应的 Redis Stream 中，作为持久化消息队列。每种告警类型都有自己的流。 +4. **模块引擎**: ASF **模块** 从其指定的流中消费告警，执行分析（通常使用 AI Agent），丰富数据，并确定结果。 +5. **SIRP 平台**: 模块的输出（现在已格式化为标准化的安全记录）被发送到 **SIRP** 平台，在那里创建或更新案例、告警和 Artifact。 +6. **剧本引擎**: 分析师可以从 SIRP 用户界面触发针对案例、告警或 Artifact 的 **剧本**，以执行进一步的自动化操作，例如威胁情报丰富或修复。 + + +![img_1.webp](Docker/Img/img_20.png) +![img_2.webp](Docker/Img/img_21.png) +![img_2.webp](Docker/Img/img_22.png) +![img_3.webp](Docker/Img/img_23.png) +![img_4.webp](Docker/Img/img_24.png) + +## 官方网站 + +[https://asf.viperrtp.com/zh/](https://asf.viperrtp.com/zh/)