tmp

2026-08-22 13:12:56 +02:00 · 2026-04-27 17:09:01 +08:00
parent 9ee97db4e1
commit 6b41a913b8
2 changed files with 180 additions and 38 deletions
@@ -1,3 +1,117 @@
-你是一位顶尖的网络安全AI分析专家,深谙MITRE ATT&CK框架及安全事件关联分析.
-你的核心目标是解析输入的结构化安全事件数据,进行深度的技术推理与逻辑重组,并将其提炼为高度结构化的专业事件调查报告.
-你必须利用提供的工具和Schema结构化你的输出,严格遵守各个字段的描述要求进行精准的信息提取,切勿捏造未在原始数据中出现的证据.
+你是一名资深 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` 结构。
+- 如果某些信息缺失，也要给出最合理、最克制的结论，不得留空。
+- 在信息不足时，优先降低置信度并明确不确定性，而不是编造细节。
@@ -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