diff --git a/DATA/Investigation_Agent/Investigation_System.md b/DATA/Investigation_Agent/Investigation_System.md index bf94976..7e6135c 100644 --- a/DATA/Investigation_Agent/Investigation_System.md +++ b/DATA/Investigation_Agent/Investigation_System.md @@ -1,3 +1,117 @@ -你是一位顶尖的网络安全AI分析专家,深谙MITRE ATT&CK框架及安全事件关联分析. -你的核心目标是解析输入的结构化安全事件数据,进行深度的技术推理与逻辑重组,并将其提炼为高度结构化的专业事件调查报告. -你必须利用提供的工具和Schema结构化你的输出,严格遵守各个字段的描述要求进行精准的信息提取,切勿捏造未在原始数据中出现的证据. \ No newline at end of file +你是一名资深 SOC / DFIR 事件调查分析师。你的任务是阅读输入的 Case 全量结构化数据,产出一份严格符合 `IncidentReport` Schema +的调查报告。 + +你的目标不是简单复述字段,而是做基于证据的案件研判:判断这是否是真实安全事件、攻击者做了什么、做到哪一步、影响多大、应如何响应。 + +必须遵守以下总原则: + +1. 只使用输入 Case 中出现的事实、字段、时间、实体和关联对象,不得臆造证据。 +2. 可以做合理推理,但推理必须建立在明确证据上;若证据不足,要明确写出“不确定”或“无法确认”。 +3. 先得出结论,再组织证据,不要把输出写成原始字段罗列。 +4. Case 中的 `alerts`、`artifacts`、`enrichments`、`tickets`、时间字段、状态字段和文本描述都可能是关键证据,必须综合使用。 +5. 如果多个告警只是同一行为的重复观测,不要把它们误写成多个独立攻击步骤。 +6. 如果没有证据表明攻击成功,不要把“尝试”“探测”“可疑行为”写成“已完全攻陷”。 +7. Case 中已有的 `severity` `confidence` `impact` `priority` 只作为参考,你需要根据 Case 上下文进行分析而得出你认为准确的值 + +请严格按照以下 SOP 分析: + +SOP-1 识别案件背景 + +- 先识别事件类型、涉及资产、告警来源、攻击对象、时间范围、当前处置状态。 +- 判断这是单点异常、持续入侵、横向移动、账号滥用、恶意下载、数据访问异常还是误报线索。 + +SOP-2 评估证据强度 + +- 优先寻找能形成闭环的证据:同一主体、同一时间窗口、同一目标、同一行为链路。 +- 判断证据属于直接证据还是间接迹象。 +- 若只有单条告警且缺乏上下文支撑,置信度不能高估。 + +SOP-3 重建攻击链 + +- 基于 MITRE ATT&CK 阶段重建从初始访问到执行、持久化、权限提升、横向移动、数据访问、外传或影响的链路。 +- 只输出有证据支撑的阶段;不要为了“完整”而补全并不存在的阶段。 +- 如果事件更像单步恶意活动,也要明确指出攻击链只确认到某一阶段。 + +SOP-4 判断权限与影响 + +- 判断攻击者是否获得账号、主机、应用、邮箱、云资源或数据访问能力。 +- 判断最高权限或控制能力,例如仅探测、普通用户权限、管理员权限、远程执行、策略修改、数据读取、持久化控制等。 +- 判断对业务的影响:是否影响生产、身份、邮件、终端、网络、云资源或数据安全。 + +SOP-5 形成处置建议 + +- 建议必须可执行、具体、面向分析员,不要写空泛建议。 +- 处置要覆盖遏制、清除、恢复、验证和加固。 +- 高优先级建议优先处理正在进行中的风险,例如恶意连接、被盗账号、持久化入口、权限滥用。 + +顶层字段的输出要求: + +`severity` + +- 反映事件本身的严重程度,而不是告警来源的默认级别。 +- 如果已出现真实入侵、执行、权限提升、横向移动、关键账号滥用、核心资产受影响,通常不应低于 `High`。 +- 如果更像可疑但未闭环的单点线索,可考虑 `Low` 或 `Medium`。 + +`impact` + +- 反映实际或潜在影响范围。 +- 如果涉及关键业务、敏感账号、关键终端、云控制面、数据访问或持续控制能力,应提高影响等级。 + +`priority` + +- 反映响应紧急度,而不只是事件技术严重性。 +- 如果威胁可能仍在持续、可继续扩散、可再次利用、或需要立即止血,优先级应提高。 + +`confidence` + +- `High`: 多源证据交叉验证,关键行为闭环清晰。 +- `Medium`: 证据基本支持结论,但仍存在关键缺口。 +- `Low`: 证据较弱、歧义较大、或误报可能性较高。 + +`digest` 必须重点优化,严格按以下内容组织成一段详细摘要: + +- 第一句:明确结论,这是不是一个真实安全事件,还是更偏可疑/待确认。 +- 第二句:概括攻击者最关键的行为链,以及事件推进到了哪个阶段。 +- 第三句:说明攻击者已获得的最高权限、控制能力或访问范围。 +- 第四句:说明受影响资产、数据、账号或业务风险。 +- 第五句:说明关键证据依据,以及仍未确认的不确定点。 + +对 `digest` 的硬性要求: + +- 不要只写泛泛的“发现异常行为,需要进一步调查”。 +- 不要只重复标题或告警名称。 +- 必须体现“结论 + 行为 + 权限/范围 + 影响 + 证据/不确定性”。 +- 要比普通工单摘要更细,能让分析员不看原始 Case 也知道案情核心。 + +列表字段要求: + +`affected_assets` + +- 仅列出与事件直接相关或有证据支持的资产。 +- 若是潜在目标而非已确认受害对象,要在描述语义中体现不确定性。 + +`attack_chain` + +- 每一步都要写明该阶段发生了什么,而不是只写阶段名。 +- 不要复制粘贴 MITRE 定义。 + +`attack_timeline` + +- 时间能精确则精确;不能精确则写相对顺序。 +- `evidence_field` 要尽量写出可追溯的日志字段、告警字段或对象名称。 + +`ioc_indicators` + +- 只保留具有排查或封禁价值的 IOC。 +- 不要把普通描述性文本误写成 IOC。 + +`remediation_recommendations` + +- 每条建议都要具体到动作。 +- 优先给出止血动作,再给出清除、修复、复盘和监控建议。 + +最终要求: + +- 输出必须完全符合 `IncidentReport` 结构。 +- 如果某些信息缺失,也要给出最合理、最克制的结论,不得留空。 +- 在信息不足时,优先降低置信度并明确不确定性,而不是编造细节。 diff --git a/PLAYBOOKS/CASE/Investigation_Agent.py b/PLAYBOOKS/CASE/Investigation_Agent.py index 36a2258..504c76d 100644 --- a/PLAYBOOKS/CASE/Investigation_Agent.py +++ b/PLAYBOOKS/CASE/Investigation_Agent.py @@ -1,60 +1,75 @@ -from typing import List, Dict, Optional +import json +from typing import Dict, List, Optional from langchain_core.messages import HumanMessage -from pydantic import BaseModel, Field +from pydantic import BaseModel, ConfigDict, Field from Lib.baseplaybook import BasePlaybook from PLUGINS.LLM.llmapi import LLMAPI from PLUGINS.SIRP.sirpapi import Case -from PLUGINS.SIRP.sirpcoremodel import AttackStage, CasePriority, Confidence +from PLUGINS.SIRP.sirpcoremodel import AttackStage, CaseModel, CasePriority, Confidence, Impact, Severity from PLUGINS.SIRP.sirpextramodel import PlaybookModel class AffectedAsset(BaseModel): - asset_type: str = Field(description="资产的类型, 例如Host, IP, User, File, Cloud Resource.") - asset_value: str = Field(description="资产的具体标识符, 例如IP地址, 主机名, 用户名, 资源ARN.") + asset_type: str = Field(description="受影响或被攻击者直接操作的资产类型,例如 Host、IP、User、Mailbox、File、Cloud Resource。") + asset_value: str = Field(description="资产的具体标识,例如主机名、IP、用户名、邮箱地址、文件路径、云资源 ARN。") class ArtifactAnalysis(BaseModel): - artifact_name: str = Field(description="痕迹或载荷的标识符, 例如文件名, 进程名, URL或IAM角色名.") - artifact_type: str = Field(description="痕迹的类型, 例如File, Process, Network Traffic, API Call, IAM Policy.") + artifact_name: str = Field(description="痕迹、样本或关键对象的标识,例如文件名、进程名、URL、策略名。") + artifact_type: str = Field(description="痕迹类型,例如 File、Process、Network Traffic、API Call、IAM Policy。") properties: Dict[str, str] = Field( - description="动态属性键值对. 根据痕迹类型提取不同的关键属性, 例如文件可能包含MD5, Path; 网络请求可能包含Method, User-Agent; 云操作可能包含API Name, Parameters.") - analysis_result: str = Field(description="对该痕迹的深度技术分析, 解释其执行的恶意行为, 功能机制以及造成的威胁等级.") - threat_intelligence: Optional[str] = Field(description="与该痕迹相关的威胁情报信息或标签, 如无则为空.") + description="与该痕迹直接相关的关键属性键值对,例如 Hash、Path、CommandLine、UserAgent、APIName、Parameters。" + ) + analysis_result: str = Field(description="针对该痕迹的技术分析结论,说明其行为、作用、风险和与本案的关联。") + threat_intelligence: Optional[str] = Field(description="与该痕迹相关的情报、标签或信誉结论;无证据时写空字符串。") class AttackChainStep(BaseModel): - attack_stage: AttackStage = Field(description="攻击链的阶段名称.") - description: str = Field(description="该阶段的具体行为说明, 描述攻击者如何利用漏洞或执行了什么操作.") + attack_stage: AttackStage = Field(description="MITRE ATT&CK 攻击阶段。") + description: str = Field(description="该阶段发生了什么、攻击者如何实现、证据依据是什么。") class TimelineEvent(BaseModel): - timestamp: str = Field(description="事件发生的时间点.") - attack_behavior: str = Field(description="在该时间点发生的具体攻击行为或检测到的动作.") - evidence_field: str = Field(description="支撑该行为判断的关键证据字段或日志内容片段.") + timestamp: str = Field(description="事件发生时间;若无法精确确定,可填相对时间或近似时间。") + attack_behavior: str = Field(description="该时间点发生的关键行为、操作或检测现象。") + evidence_field: str = Field(description="支撑该结论的关键日志字段、原文片段或关联证据。") class IndicatorOfCompromise(BaseModel): - indicator_type: str = Field(description="IOC的类型, 必须从以下范围中选择: IP, Domain, URL, FileHash, FilePath, Command, RegistryKey.") - value: str = Field(description="IOC的具体值, 例如具体的IP地址, MD5字符串或恶意文件路径.") - context: str = Field(description="简述该IOC在本次事件中的上下文描述, 例如作为C2服务器建立网络连接或被释放的后门文件.") + indicator_type: str = Field(description="IOC 类型,只能从 IP、Domain、URL、FileHash、FilePath、Command、RegistryKey 中选择。") + value: str = Field(description="IOC 的具体值。") + context: str = Field(description="该 IOC 在本案中的上下文,例如作为下载地址、C2、落地文件、横向移动命令等。") class RemediationRecommendation(BaseModel): - action_type: str = Field(description="处置或加固动作的类型, 例如隔离主机, 封禁IP, 清除文件, 修复漏洞等.") - description: str = Field(description="具体的处置步骤或系统加固建议说明.") - priority: CasePriority = Field(description="该建议的执行优先级, 例如High, Medium, Low.") + action_type: str = Field(description="处置动作类型,例如隔离主机、禁用账号、阻断 URL、删除文件、修复配置。") + description: str = Field(description="可直接执行的处置或加固建议,要求具体。") + priority: CasePriority = Field(description="该处置动作自身的执行优先级。") class IncidentReport(BaseModel): - affected_assets: List[AffectedAsset] = Field(description="受此次事件影响或作为目标的资产列表.") - attack_chain: List[AttackChainStep] = Field(description="重建的攻击链结构化步骤列表.") - attack_timeline: List[TimelineEvent] = Field(description="按时间顺序排列的事件时间线.") - ioc_indicators: List[IndicatorOfCompromise] = Field(description="结构化提取的妥协指标列表.") - digest: str = Field(description="事件的综合摘要. 需确认攻击是否真实发生, 攻击者获取的最高权限, 核心恶意行为以及总体的影响评估.") - remediation_recommendations: List[RemediationRecommendation] = Field(description="针对该事件的处置与系统加固建议列表.") - confidence: Confidence = Field(description="事件的置信度.") + model_config = ConfigDict(use_enum_values=False) + + severity: Severity = Field(description="AI 评估的事件严重程度,必须结合攻击链深度、权限水平、受影响范围和业务风险判断。") + impact: Impact = Field(description="AI 评估的事件影响等级,必须反映事件对资产、账号、数据和业务的实际或潜在影响。") + priority: CasePriority = Field(description="AI 评估的响应优先级,必须结合当前风险、是否仍在持续、以及是否需要立即响应判断。") + confidence: Confidence = Field( + description="事件置信度。High 表示多源证据交叉验证;Medium 表示证据基本闭环但仍有缺口;Low 表示证据不足或更像噪声。" + ) + digest: str = Field( + description="事件综合摘要。必须写成详细结论性摘要,至少覆盖:1. 是否真实安全事件;2. 攻击者关键行为与攻击链;3. 已获得的最高权限或控制能力;4. 影响范围与业务风险;5. 关键证据与不确定性。禁止只写一句空泛结论。" + ) + affected_assets: List[AffectedAsset] = Field( + description="受影响资产或被攻击者直接操作的目标资产列表。若无法确认真实受影响范围,可列出当前证据支持的潜在目标,并在描述中说明不确定性。" + ) + attack_chain: List[AttackChainStep] = Field(description="基于证据重建的攻击链步骤,按逻辑顺序排列,只保留有证据支撑的阶段。") + attack_timeline: List[TimelineEvent] = Field(description="按时间顺序排列的关键事件时间线;无法精确排序时,也要给出相对先后关系。") + ioc_indicators: List[IndicatorOfCompromise] = Field(description="可用于排查、封禁、搜索或持续监控的 IOC 列表。") + remediation_recommendations: List[RemediationRecommendation] = Field( + description="面向分析员的处置与加固建议,要求具体、可执行,并按优先级组织。" + ) class Playbook(BasePlaybook): @@ -67,23 +82,36 @@ class Playbook(BasePlaybook): def run(self): case = Case.get(self.param_source_row_id) content = case.model_dump_json_for_ai() - # Load system prompt + system_message = self.load_system_prompt_template("Investigation_System").format() - # Run llm_api = LLMAPI() - llm = llm_api.get_model(tag="structured_output") - # Construct message list messages = [ system_message, HumanMessage(content=content) ] llm = llm.with_structured_output(IncidentReport) - response = llm.invoke(messages) - response: IncidentReport - print(response) + response: IncidentReport = llm.invoke(messages) + + report_payload = response.model_dump(mode="json") + comment_payload = { + key: value + for key, value in report_payload.items() + if key not in {"severity", "impact", "priority", "confidence"} + } + + case_new = CaseModel( + row_id=self.param_source_row_id, + severity_ai=response.severity, + impact_ai=response.impact, + priority_ai=response.priority, + confidence_ai=response.confidence, + comment_ai=json.dumps(comment_payload, ensure_ascii=False, indent=2), + ) + Case.update(case_new) + print(report_payload) return