External/agentic-soc-platform
1
0
Fork 0
mirror of https://github.com/FunnyWolf/agentic-soc-platform.git synced 2026-08-22 13:12:56 +02:00
Code Issues Packages Projects Releases Wiki Activity

arch and todo

Browse Source
This commit is contained in:
rootkit
2026-03-19 04:43:15 +08:00
parent 9ded724cb2
commit 18fbf79c86
2 changed files with 46 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
PLUGINS/MCP/llmfunc.py
+33 -45
View File
@@ -11,11 +11,17 @@ from PLUGINS.SIRP.sirpmodel import ArtifactReputationScore, ArtifactRole, Artifa
AttackStage, KnowledgeAction, KnowledgeSource, PlaybookJobStatus, PlaybookType, TicketStatus, TicketType
def _build_filter_group(conditions: list[Condition]) -> Group:
return Group(logic="AND", children=conditions or [])
def _dump_models_for_ai(models, limit: int) -> list[str]:
return [model.model_dump_json_for_ai() for model in models[:limit]]
# Case
def list_cases(
case_id: Annotated[str, "Case ID, e.g. case_000005"] = None,
case_id: Annotated[Optional[str], "Case ID, e.g. case_000005"] = None,
status: Annotated[Optional[list[CaseStatus]], "Case status filter"] = None,
severity: Annotated[Optional[list[Severity]], "Case severity filter"] = None,
confidence: Annotated[Optional[list[Confidence]], "Case confidence filter"] = None,
@@ -44,13 +50,9 @@ def list_cases(
if tags:
conditions.append(Condition(field="tags", operator=Operator.CONTAINS, value=tags))
filter_model = Group(logic="AND", children=conditions) if conditions else Group(logic="AND", children=[])
filter_model = _build_filter_group(conditions)
models = Case.list(filter_model, lazy_load=True)
result = []
for model in models[:limit]:
result.append(model.model_dump_json_for_ai())
return result
return _dump_models_for_ai(models, limit)
def get_case_discussions(
@@ -103,8 +105,9 @@ def update_case(
return Case.update(case_new)
# Alert
def list_alerts(
alert_id: Annotated[str, "Alert ID, e.g. alert_000001"] = None,
alert_id: Annotated[Optional[str], "Alert ID, e.g. alert_000001"] = None,
status: Annotated[Optional[list[AlertStatus]], "Alert status filter"] = None,
severity: Annotated[Optional[list[Severity]], "Alert severity filter"] = None,
confidence: Annotated[Optional[list[Confidence]], "Alert confidence filter"] = None,
@@ -125,13 +128,9 @@ def list_alerts(
if correlation_uid:
conditions.append(Condition(field="correlation_uid", operator=Operator.EQ, value=correlation_uid))
filter_model = Group(logic="AND", children=conditions) if conditions else Group(logic="AND", children=[])
filter_model = _build_filter_group(conditions)
models = Alert.list(filter_model, lazy_load=True)
result = []
for model in models[:limit]:
result.append(model.model_dump_json_for_ai())
return result
return _dump_models_for_ai(models, limit)
def get_alert_discussions(
@@ -159,6 +158,8 @@ def update_alert(
)
# Artifact
def append_artifact(
alert_id: Annotated[str, "Target alert ID to append the artifact to"],
name: Annotated[str, "Artifact name"] = "",
@@ -183,7 +184,7 @@ def append_artifact(
def list_artifacts(
artifact_id: Annotated[str, "Artifact ID, e.g. artifact_000001"] = None,
artifact_id: Annotated[Optional[str], "Artifact ID, e.g. artifact_000001"] = None,
type: Annotated[Optional[list[ArtifactType]], "Artifact type filter"] = None,
role: Annotated[Optional[list[ArtifactRole]], "Artifact role filter"] = None,
reputation_score: Annotated[Optional[list[ArtifactReputationScore]], "Artifact reputation filter"] = None,
@@ -206,15 +207,12 @@ def list_artifacts(
if value:
conditions.append(Condition(field="value", operator=Operator.EQ, value=value))
filter_model = Group(logic="AND", children=conditions) if conditions else Group(logic="AND", children=[])
filter_model = _build_filter_group(conditions)
models = Artifact.list(filter_model, lazy_load=True)
result = []
for model in models[:limit]:
result.append(model.model_dump_json_for_ai())
return result
return _dump_models_for_ai(models, limit)
# Enrichment
def append_enrichment(
target_type: Annotated[str, "Target object type: CASE, ALERT, or ARTIFACT"],
target_id: Annotated[str, "Target object ID"],
@@ -229,7 +227,7 @@ def append_enrichment(
"""Create one enrichment and attach it to an existing case, alert, or artifact."""
normalized_target_type = target_type.strip().lower()
if normalized_target_type == "CASE":
if normalized_target_type == "case":
return Case.append_enrichment(
case_id=target_id,
name=name,
@@ -241,7 +239,7 @@ def append_enrichment(
data=data
)
if normalized_target_type == "ALERT":
if normalized_target_type == "alert":
return Alert.append_enrichment(
alert_id=target_id,
name=name,
@@ -253,7 +251,7 @@ def append_enrichment(
data=data
)
if normalized_target_type == "ARTIFACT":
if normalized_target_type == "artifact":
return Artifact.append_enrichment(
artifact_id=target_id,
name=name,
@@ -268,6 +266,7 @@ def append_enrichment(
raise ValueError("target_type must be one of: case, alert, artifact")
# Ticket
def create_ticket(
uid: Annotated[str, "External ticket ID to sync into SIRP"],
title: Annotated[str, "Ticket title"] = "",
@@ -303,13 +302,9 @@ def list_tickets(
if uid:
conditions.append(Condition(field="uid", operator=Operator.EQ, value=uid))
filter_model = Group(logic="AND", children=conditions) if conditions else Group(logic="AND", children=[])
filter_model = _build_filter_group(conditions)
models = Ticket.list(filter_model, lazy_load=True)
result = []
for model in models[:limit]:
result.append(model.model_dump_json_for_ai())
return result
return _dump_models_for_ai(models, limit)
def update_ticket(
@@ -331,6 +326,7 @@ def update_ticket(
)
# Playbook
def list_available_playbook_definitions(
) -> Annotated[str, "Runnable playbook definitions as JSON string, not playbook run records"]:
"""List all runnable built-in playbook definitions, not playbook run records."""
@@ -343,7 +339,7 @@ def list_playbook_runs(
job_status: Annotated[Optional[list[PlaybookJobStatus]], "Playbook job status filter"] = None,
type: Annotated[Optional[list[PlaybookType]], "Playbook type filter"] = None,
source_id: Annotated[Optional[str], "Playbook target record ID filter, e.g. case_000001, alert_000001, artifact_000001"] = None,
limit: Annotated[int, "Max playbooks to return"] = 10
limit: Annotated[int, "Max playbook runs to return"] = 10
) -> Annotated[list[str], "Matching playbook run records as AI-friendly JSON list"]:
"""List playbook run records with optional filters."""
conditions = []
@@ -357,13 +353,9 @@ def list_playbook_runs(
if type:
conditions.append(Condition(field="type", operator=Operator.IN, value=type))
filter_model = Group(logic="AND", children=conditions) if conditions else Group(logic="AND", children=[])
filter_model = _build_filter_group(conditions)
models = Playbook.list(filter_model, lazy_load=True)
result = []
for model in models[:limit]:
result.append(model.model_dump_json_for_ai())
return result
return _dump_models_for_ai(models, limit)
def execute_playbook(
@@ -407,13 +399,9 @@ def list_knowledge(
if tags:
conditions.append(Condition(field="tags", operator=Operator.CONTAINS, value=tags))
filter_model = Group(logic="AND", children=conditions) if conditions else Group(logic="AND", children=[])
filter_model = _build_filter_group(conditions)
models = Knowledge.list(filter_model, lazy_load=True)
result = []
for model in models[:limit]:
result.append(model.model_dump_json_for_ai())
return result
return _dump_models_for_ai(models, limit)
def update_knowledge(
TODO.md
+13 -14
View File
@@ -15,13 +15,6 @@
## 2. 高优先级
### 2.1 Playbook
1. 增加 Playbook 配置列表查询接口，直接使用 `PlaybookLoader.list_playbook_config()`，返回完整列表，不区分目标对象
2. 增加 Playbook 执行接口，直接使用 `Playbook.add_pending_playbook()` 创建 pending 记录，由其他逻辑继续处理
3. 为 Playbook 增加 `get_by_id()`，用于查询某次执行的状态与消息
4. 在 Playbook 执行接口中支持传入 `user_input`，作为用户对当前 Playbook 的附加自然语言要求，由 Playbook 内部决定是否加入 prompt
### 2.2 Artifact
1. 基于 Artifact 的内部资产查询
@@ -43,20 +36,25 @@
### 3.1 接口模型统一化
1. `list_available_playbooks(target_type, target_id)`
2. `run_playbook(target_type, target_id, playbook_name, user_input)`
3. `get_playbook_run(job_id)`
4. `list_playbook_runs(target_type, target_id, job_status)`
5. 继续统一 `target_type + target_id` 风格的 MCP 接口设计
1. 继续统一 MCP 列表接口的公共参数与返回风格
2. 收口 `list_*` 类接口中的重复过滤与序列化逻辑
3. 继续统一 `target_type + target_id` 风格的 MCP 接口设计
4. 明确对象 ID、运行记录 ID、定义名三类标识的命名边界
### 3.2 Knowledge
### 3.2 Playbook
1. 对 `execute_playbook` 增加 definition 名称校验或友好错误返回
2. 视需要增加按对象视角查询 playbook run 的更清晰别名接口
3. 评估是否需要为 playbook run 增加独立详情接口，而不是仅依赖 `list_playbook_runs`
### 3.3 Knowledge
1. 语义搜索接口
2. 按标签或场景批量检索
3. 将 Playbook/Agent 输出结果直接写入 Knowledge
4. 更清晰的知识生命周期，例如 `STORE`、`REMOVE`、`ACTIVE`、`EXPIRED`
### 3.3 结果沉淀方式
### 3.4 结果沉淀方式
1. 原始字段保留源事实
2. Enrichment 承载扩展结果
@@ -108,3 +106,4 @@
2. MCP 层尽量保持为参数 schema、对象路由、结果包装
3. 新能力优先围绕真实对象关系建模，不在 MCP 层虚构关系
4. 优先强化 Artifact、Enrichment、Knowledge、Playbook 之间的联动
5. 优先消除 MCP 接口命名歧义，尤其区分 definition、run、record、target object