diff --git a/PLUGINS/MCP/llmfunc.py b/PLUGINS/MCP/llmfunc.py index 4629165..4b0d99e 100644 --- a/PLUGINS/MCP/llmfunc.py +++ b/PLUGINS/MCP/llmfunc.py @@ -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( diff --git a/TODO.md b/TODO.md index 0892bde..0bced35 100644 --- a/TODO.md +++ b/TODO.md @@ -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