diff --git a/ASP/settings.py b/ASP/settings.py index e52b770..193f90b 100644 --- a/ASP/settings.py +++ b/ASP/settings.py @@ -102,7 +102,7 @@ LOGGING = { 'file', ], 'level': 'DEBUG', - 'propagate': True, + 'propagate': False, }, } diff --git a/MODULES/Cloud-01-AWS-IAM-Privilege-Escalation-via-AttachUserPolicy.py b/MODULES/Cloud-01-AWS-IAM-Privilege-Escalation-via-AttachUserPolicy.py index b322514..e40f0ad 100644 --- a/MODULES/Cloud-01-AWS-IAM-Privilege-Escalation-via-AttachUserPolicy.py +++ b/MODULES/Cloud-01-AWS-IAM-Privilege-Escalation-via-AttachUserPolicy.py @@ -7,7 +7,7 @@ from Lib.basemodule import BaseModule from PLUGINS.SIRP.correlation import Correlation from PLUGINS.SIRP.sirpapi import Alert, Case from PLUGINS.SIRP.sirpcoremodel import ArtifactType, ArtifactRole, Severity, Impact, Disposition, AlertAction, Confidence, AlertAnalyticType, ProductCategory, \ - AlertPolicyType, AlertRiskLevel, AlertStatus, CasePriority, ArtifactModel, AlertModel, CaseModel, EnrichmentModel + AlertPolicyType, AlertRiskLevel, AlertStatus, CasePriority, ArtifactModel, AlertModel, CaseModel, EnrichmentModel, CaseStatus class Module(BaseModule): @@ -178,7 +178,6 @@ class Module(BaseModule): # 保存告警 saved_alert_row_id = Alert.create(alert_model) - self.logger.info(f"Alert created: {saved_alert_row_id}") # 5. Case 处理 (Case Management) try: @@ -194,6 +193,7 @@ class Module(BaseModule): # 根据 Alert 计算 Case字段 new_case = CaseModel( title=f"Potential IAM Privilege Escalation in Account {account_id}", + status=CaseStatus.NEW, # 创建时显式设置为New severity=severity, impact=Impact.HIGH if outcome == "success" else Impact.MEDIUM, priority=CasePriority.HIGH if outcome == "success" else CasePriority.MEDIUM, diff --git a/PLUGINS/SIRP/nocolyapi.py b/PLUGINS/SIRP/nocolyapi.py index 85550fa..b07f1ce 100644 --- a/PLUGINS/SIRP/nocolyapi.py +++ b/PLUGINS/SIRP/nocolyapi.py @@ -1,5 +1,6 @@ from __future__ import annotations +import time from typing import Dict from typing import List, Any @@ -27,6 +28,24 @@ adapter = HTTPAdapter( HTTP_SESSION.mount('http://', adapter) HTTP_SESSION.mount('https://', adapter) + +def _request_with_timing(method: str, url: str, **kwargs): + started_at = time.perf_counter() + try: + response = HTTP_SESSION.request(method=method, url=url, **kwargs) + elapsed_ms = (time.perf_counter() - started_at) * 1000 + logger.debug( + f"[SIRP] request method={method} url={url} status={response.status_code} elapsed_ms={elapsed_ms:.2f}" + ) + return response + except Exception as exc: + elapsed_ms = (time.perf_counter() - started_at) * 1000 + logger.debug( + f"[SIRP] request failed method={method} url={url} elapsed_ms={elapsed_ms:.2f} error={exc}" + ) + raise + + SYSTEM_FIELDS = ['_initiatedBy', '_owner', '_updatedAt', '_createdAt', '_remainingTime', '_createdBy', '_updatedBy', '_processName', '_nodeAssignees', '_initiatedAt', '_nodeStartedAt', '_approvalCompletedAt', '_dueAt', '_processStatus'] @@ -43,7 +62,8 @@ class Worksheet(object): url = f"{SIRP_URL}/api/v3/app/worksheets/{worksheet_id}" - response = HTTP_SESSION.get( + response = _request_with_timing( + "GET", url ) response.raise_for_status() @@ -169,7 +189,8 @@ class WorksheetRow(object): def get(worksheet_id: str, row_id: str, include_system_fields=True) -> dict: url = f"{SIRP_URL}/api/v3/app/worksheets/{worksheet_id}/rows/{row_id}" fields = Worksheet.get_fields(worksheet_id) - response = HTTP_SESSION.get( + response = _request_with_timing( + "GET", url, timeout=SIRP_REQUEST_TIMEOUT, params={"includeSystemFields": include_system_fields} @@ -210,10 +231,11 @@ class WorksheetRow(object): "pageIndex": page_index } - response = HTTP_SESSION.post(url, - timeout=SIRP_REQUEST_TIMEOUT, - headers=HEADERS, - json=data) + response = _request_with_timing("POST", + url, + timeout=SIRP_REQUEST_TIMEOUT, + headers=HEADERS, + json=data) response.raise_for_status() response_data = response.json() @@ -250,9 +272,10 @@ class WorksheetRow(object): } try: - response = HTTP_SESSION.post(url, - timeout=SIRP_REQUEST_TIMEOUT, - json=data) + response = _request_with_timing("POST", + url, + timeout=SIRP_REQUEST_TIMEOUT, + json=data) response.raise_for_status() response_data = response.json() @@ -274,9 +297,10 @@ class WorksheetRow(object): "triggerWorkflow": trigger_workflow, "fields": fields } - response = HTTP_SESSION.patch(url, - timeout=SIRP_REQUEST_TIMEOUT, - json=data) + response = _request_with_timing("PATCH", + url, + timeout=SIRP_REQUEST_TIMEOUT, + json=data) response.raise_for_status() response_data = response.json() @@ -304,9 +328,10 @@ class WorksheetRow(object): } try: - response = HTTP_SESSION.post(url, - timeout=SIRP_REQUEST_TIMEOUT, - json=data) + response = _request_with_timing("POST", + url, + timeout=SIRP_REQUEST_TIMEOUT, + json=data) response.raise_for_status() response_data = response.json() @@ -335,9 +360,10 @@ class WorksheetRow(object): } try: - response = HTTP_SESSION.patch(url, - timeout=SIRP_REQUEST_TIMEOUT, - json=data) + response = _request_with_timing("PATCH", + url, + timeout=SIRP_REQUEST_TIMEOUT, + json=data) response.raise_for_status() response_data = response.json() @@ -357,9 +383,10 @@ class WorksheetRow(object): "triggerWorkflow": trigger_workflow, } - response = HTTP_SESSION.delete(url, - timeout=SIRP_REQUEST_TIMEOUT, - json=data) + response = _request_with_timing("DELETE", + url, + timeout=SIRP_REQUEST_TIMEOUT, + json=data) response.raise_for_status() response_data = response.json() @@ -378,9 +405,10 @@ class WorksheetRow(object): "triggerWorkflow": trigger_workflow, } - response = HTTP_SESSION.delete(url, - timeout=SIRP_REQUEST_TIMEOUT, - json=data) + response = _request_with_timing("DELETE", + url, + timeout=SIRP_REQUEST_TIMEOUT, + json=data) response.raise_for_status() response_data = response.json() @@ -395,8 +423,9 @@ class WorksheetRow(object): def get_discussions(worksheet_id: str, row_id: str) -> List[Dict]: url = f"{SIRP_URL}/api/v3/app/worksheets/{worksheet_id}/rows/{row_id}/discussions" - response = HTTP_SESSION.get(url, - timeout=SIRP_REQUEST_TIMEOUT) + response = _request_with_timing("GET", + url, + timeout=SIRP_REQUEST_TIMEOUT) response.raise_for_status() response_data = response.json() @@ -437,9 +466,10 @@ class WorksheetRow(object): if include_system_fields is not None: params["isReturnSystemFields"] = include_system_fields - response = HTTP_SESSION.get(url, - timeout=SIRP_REQUEST_TIMEOUT, - params=params) + response = _request_with_timing("GET", + url, + timeout=SIRP_REQUEST_TIMEOUT, + params=params) response.raise_for_status() response_data = response.json() @@ -465,8 +495,9 @@ class OptionSet(object): return cached_optionsets url = f"{SIRP_URL}/api/v3/app/optionsets" - response = HTTP_SESSION.get(url, - timeout=SIRP_REQUEST_TIMEOUT) + response = _request_with_timing("GET", + url, + timeout=SIRP_REQUEST_TIMEOUT) response.raise_for_status() response_data = response.json() diff --git a/PLUGINS/SIRP/sirpapi.py b/PLUGINS/SIRP/sirpapi.py index 31629ee..d7ad136 100644 --- a/PLUGINS/SIRP/sirpapi.py +++ b/PLUGINS/SIRP/sirpapi.py @@ -1,20 +1,338 @@ import json -from typing import List, Union, Annotated +from abc import ABC +from typing import List, Union, Annotated, Dict, Any, TypeVar, Generic, Type import requests from langchain_core.documents import Document +from pydantic import BaseModel from Lib.log import logger from PLUGINS.Embeddings.embeddings_qdrant import embedding_api_singleton_qdrant, SIRP_KNOWLEDGE_COLLECTION from PLUGINS.SIRP.CONFIG import SIRP_NOTICE_WEBHOOK from PLUGINS.SIRP.nocolyapi import WorksheetRow from PLUGINS.SIRP.nocolymodel import Condition, Group, Operator -from PLUGINS.SIRP.sirpbase import BaseWorksheetEntity -from PLUGINS.SIRP.sirpbasemodel import AutoAccount +from PLUGINS.SIRP.sirpbasemodel import AutoAccount, BaseSystemModel from PLUGINS.SIRP.sirpcoremodel import Severity, Confidence, EnrichmentModel, TicketModel, ArtifactModel, AlertModel, CaseModel from PLUGINS.SIRP.sirpextramodel import PlaybookType, PlaybookJobStatus, KnowledgeAction, MessageModel, PlaybookModel, KnowledgeModel +def model_to_fields(model_instance: BaseModel) -> List[Dict[str, Any]]: + fields = [] + model_data = model_instance.model_dump(mode='json', exclude_unset=True) + for key, value in model_data.items(): + field_info = model_instance.model_fields.get(key) + field_item = { + 'id': key, + 'value': value + } + if field_info and field_info.json_schema_extra: + field_item.update(field_info.json_schema_extra) + fields.append(field_item) + return fields + + +T = TypeVar('T', bound=BaseSystemModel) + + +class BaseSimpleEntity(ABC): + """简化的工作表实体基类（不使用模型）""" + + WORKSHEET_ID: str + + @classmethod + def list(cls, filter_dict: dict) -> List[Dict]: + """列表查询 + + Args: + filter_dict: 过滤条件字典 + + Returns: + 字典列表 + """ + return WorksheetRow.list(cls.WORKSHEET_ID, filter_dict, include_system_fields=False) + + @classmethod + def get(cls, row_id: str) -> Dict: + """获取单条记录 + + Args: + row_id: 记录ID + + Returns: + 字典 + """ + return WorksheetRow.get(cls.WORKSHEET_ID, row_id, include_system_fields=False) + + @classmethod + def create(cls, fields: List[Dict]) -> str: + """创建记录 + + Args: + fields: 字段列表 + + Returns: + 新创建的记录ID + """ + return WorksheetRow.create(cls.WORKSHEET_ID, fields) + + @classmethod + def update(cls, row_id: str, fields: List[Dict]) -> str: + """更新记录 + + Args: + row_id: 记录ID + fields: 字段列表 + + Returns: + 更新的记录ID + """ + return WorksheetRow.update(cls.WORKSHEET_ID, row_id, fields) + + +class BaseWorksheetEntity(ABC, Generic[T]): + """通用工作表实体基类 - 支持泛型和关联加载""" + + WORKSHEET_ID: str + MODEL_CLASS: Type[T] + + @classmethod + def get( + cls, + row_id: str, + include_system_fields: bool = True, + lazy_load: bool = False + ) -> T: + """获取单条记录 + + Args: + row_id: 记录ID + include_system_fields: 是否包含系统字段 + lazy_load: 是否延迟加载关联数据（True时不加载关联） + + Returns: + 模型实例 + """ + result = WorksheetRow.get( + cls.WORKSHEET_ID, + row_id, + include_system_fields=include_system_fields + ) + model = cls.MODEL_CLASS(**result) + + if not lazy_load: + model = cls._load_relations(model, include_system_fields) + + return model + + @classmethod + def list( + cls, + filter_model: Group, + include_system_fields: bool = True, + lazy_load: bool = False + ) -> List[T]: + """按过滤条件列表查询 + + Args: + filter_model: 过滤条件Group对象 + include_system_fields: 是否包含系统字段 + lazy_load: 是否延迟加载关联数据（True时不加载关联） + + Returns: + 模型实例列表 + """ + if filter_model.children: + filter_dict = filter_model.model_dump() + else: + filter_dict = {} + result = WorksheetRow.list( + cls.WORKSHEET_ID, + filter_dict, + include_system_fields=include_system_fields + ) + + model_list = [] + for item in result: + model_obj = cls.MODEL_CLASS(**item) + if not lazy_load: + model_obj = cls._load_relations(model_obj, include_system_fields) + model_list.append(model_obj) + + return model_list + + @classmethod + def update_by_filter(cls, + filter_model: Group, + model: T, + include_system_fields: bool = True) -> dict: + filter_dict = filter_model.model_dump() + result = WorksheetRow.list( + cls.WORKSHEET_ID, + filter_dict, + fields=["row_id"], + include_system_fields=include_system_fields + ) + row_ids = [] + for item in result: + row_ids.append(item["row_id"]) + + model = cls._prepare_for_save(model) + + fields = model_to_fields(model) + result = WorksheetRow.batch_update(cls.WORKSHEET_ID, row_ids, fields) + return result + + @classmethod + def list_by_row_ids( + cls, + row_ids: List[Any], + include_system_fields: bool = True, + lazy_load: bool = False + ) -> Union[List[T], List[str], None]: + """按ID列表查询 + + Args: + row_ids: 记录ID列表 + include_system_fields: 是否包含系统字段 + lazy_load: 是否延迟加载关联数据 + + Returns: + 模型实例列表或原始row_ids列表 + """ + + if row_ids is not None and row_ids != []: + if isinstance(row_ids[0], BaseSystemModel): + return row_ids + + filter_model = Group( + logic="AND", + children=[ + Condition( + field="row_id", + operator=Operator.IN, + value=row_ids + ) + ] + ) + return cls.list(filter_model, include_system_fields=include_system_fields, lazy_load=lazy_load) + return row_ids + + @classmethod + def create(cls, model: T) -> str: + """创建记录 + + Args: + model: 模型实例 + + Returns: + 新创建的记录ID + """ + model = cls._prepare_for_save(model) + + fields = model_to_fields(model) + row_id = WorksheetRow.create(cls.WORKSHEET_ID, fields) + return row_id + + @classmethod + def update(cls, model: T) -> str: + """更新记录 + + Args: + model: 模型实例（必须包含row_id） + + Returns: + 更新的记录ID + + Raises: + ValueError: 当row_id为None时 + """ + if model.row_id is None: + raise ValueError(f"{cls.__name__} row_id is None, cannot update.") + + model = cls._prepare_for_save(model) + + fields = model_to_fields(model) + row_id = WorksheetRow.update(cls.WORKSHEET_ID, model.row_id, fields) + return row_id + + @classmethod + def update_or_create(cls, model: T) -> str: + """更新或创建记录 + + Args: + model: 模型实例 + + Returns: + 记录ID + """ + model = cls._prepare_for_save(model) + + fields = model_to_fields(model) + + if model.row_id is None: + row_id = WorksheetRow.create(cls.WORKSHEET_ID, fields) + else: + row_id = WorksheetRow.update(cls.WORKSHEET_ID, model.row_id, fields) + + return row_id + + @classmethod + def batch_update_or_create(cls, model_list: List[Union[T, str]]) -> Union[List[str], None]: + """批量更新 + + Args: + model_list: 模型实例或ID字符串的列表 + + Returns: + 更新后的记录ID列表 + + Raises: + TypeError: 当列表中包含不支持的类型时 + """ + if model_list is None: + return model_list + + row_ids = [] + for model in model_list: + if isinstance(model, str): + row_ids.append(model) # just link + elif isinstance(model, cls.MODEL_CLASS): + row_id = cls.update_or_create(model) + row_ids.append(row_id) + else: + raise TypeError( + f"Unsupported {cls.__name__} data type: {type(model).__name__}. " + f"Expected str or {cls.MODEL_CLASS.__name__}" + ) + + return row_ids + + @classmethod + def _load_relations(cls, model: T, include_system_fields: bool = True) -> T: + """加载关联数据（子类可覆盖） + + Args: + model: 模型实例 + include_system_fields: 是否包含系统字段 + + Returns: + 加载了关联数据的模型实例 + """ + return model + + @classmethod + def _prepare_for_save(cls, model: T) -> T: + """保存前准备（子类可覆盖） + + Args: + model: 模型实例 + + Returns: + 准备好的模型实例 + """ + return model + + class Enrichment(BaseWorksheetEntity[EnrichmentModel]): """Enrichment 实体类""" WORKSHEET_ID = "enrichment" diff --git a/PLUGINS/SIRP/sirpbase.py b/PLUGINS/SIRP/sirpbase.py deleted file mode 100644 index 29305a0..0000000 --- a/PLUGINS/SIRP/sirpbase.py +++ /dev/null @@ -1,326 +0,0 @@ -from abc import ABC -from typing import TypeVar, Generic, Type, List, Dict, Union, Any - -from pydantic import BaseModel - -from PLUGINS.SIRP.nocolyapi import WorksheetRow -from PLUGINS.SIRP.nocolymodel import Condition, Group, Operator -from PLUGINS.SIRP.sirpbasemodel import BaseSystemModel - - -def model_to_fields(model_instance: BaseModel) -> List[Dict[str, Any]]: - fields = [] - model_data = model_instance.model_dump(mode='json', exclude_unset=True) - for key, value in model_data.items(): - field_info = model_instance.model_fields.get(key) - field_item = { - 'id': key, - 'value': value - } - if field_info and field_info.json_schema_extra: - field_item.update(field_info.json_schema_extra) - fields.append(field_item) - return fields - - -# 定义泛型类型 -T = TypeVar('T', bound=BaseSystemModel) - - -class BaseWorksheetEntity(ABC, Generic[T]): - """通用工作表实体基类 - 支持泛型和关联加载""" - - WORKSHEET_ID: str - MODEL_CLASS: Type[T] - - @classmethod - def get( - cls, - row_id: str, - include_system_fields: bool = True, - lazy_load: bool = False - ) -> T: - """获取单条记录 - - Args: - row_id: 记录ID - include_system_fields: 是否包含系统字段 - lazy_load: 是否延迟加载关联数据（True时不加载关联） - - Returns: - 模型实例 - """ - result = WorksheetRow.get( - cls.WORKSHEET_ID, - row_id, - include_system_fields=include_system_fields - ) - model = cls.MODEL_CLASS(**result) - - if not lazy_load: - model = cls._load_relations(model, include_system_fields) - - return model - - @classmethod - def list( - cls, - filter_model: Group, - include_system_fields: bool = True, - lazy_load: bool = False - ) -> List[T]: - """按过滤条件列表查询 - - Args: - filter_model: 过滤条件Group对象 - include_system_fields: 是否包含系统字段 - lazy_load: 是否延迟加载关联数据（True时不加载关联） - - Returns: - 模型实例列表 - """ - if filter_model.children: - filter_dict = filter_model.model_dump() - else: - filter_dict = {} - result = WorksheetRow.list( - cls.WORKSHEET_ID, - filter_dict, - include_system_fields=include_system_fields - ) - - model_list = [] - for item in result: - model_obj = cls.MODEL_CLASS(**item) - if not lazy_load: - model_obj = cls._load_relations(model_obj, include_system_fields) - model_list.append(model_obj) - - return model_list - - @classmethod - def update_by_filter(cls, - filter_model: Group, - model: T, - include_system_fields: bool = True) -> dict: - filter_dict = filter_model.model_dump() - result = WorksheetRow.list( - cls.WORKSHEET_ID, - filter_dict, - fields=["row_id"], - include_system_fields=include_system_fields - ) - row_ids = [] - for item in result: - row_ids.append(item["row_id"]) - - model = cls._prepare_for_save(model) - - fields = model_to_fields(model) - result = WorksheetRow.batch_update(cls.WORKSHEET_ID, row_ids, fields) - return result - - @classmethod - def list_by_row_ids( - cls, - row_ids: List[Any], - include_system_fields: bool = True, - lazy_load: bool = False - ) -> Union[List[T], List[str], None]: - """按ID列表查询 - - Args: - row_ids: 记录ID列表 - include_system_fields: 是否包含系统字段 - lazy_load: 是否延迟加载关联数据 - - Returns: - 模型实例列表或原始row_ids列表 - """ - - if row_ids is not None and row_ids != []: - if isinstance(row_ids[0], BaseSystemModel): - return row_ids - - filter_model = Group( - logic="AND", - children=[ - Condition( - field="row_id", - operator=Operator.IN, - value=row_ids - ) - ] - ) - return cls.list(filter_model, include_system_fields=include_system_fields, lazy_load=lazy_load) - return row_ids - - @classmethod - def create(cls, model: T) -> str: - """创建记录 - - Args: - model: 模型实例 - - Returns: - 新创建的记录ID - """ - model = cls._prepare_for_save(model) - - fields = model_to_fields(model) - row_id = WorksheetRow.create(cls.WORKSHEET_ID, fields) - return row_id - - @classmethod - def update(cls, model: T) -> str: - """更新记录 - - Args: - model: 模型实例（必须包含row_id） - - Returns: - 更新的记录ID - - Raises: - ValueError: 当row_id为None时 - """ - if model.row_id is None: - raise ValueError(f"{cls.__name__} row_id is None, cannot update.") - - model = cls._prepare_for_save(model) - - fields = model_to_fields(model) - row_id = WorksheetRow.update(cls.WORKSHEET_ID, model.row_id, fields) - return row_id - - @classmethod - def update_or_create(cls, model: T) -> str: - """更新或创建记录 - - Args: - model: 模型实例 - - Returns: - 记录ID - """ - model = cls._prepare_for_save(model) - - fields = model_to_fields(model) - - if model.row_id is None: - row_id = WorksheetRow.create(cls.WORKSHEET_ID, fields) - else: - row_id = WorksheetRow.update(cls.WORKSHEET_ID, model.row_id, fields) - - return row_id - - @classmethod - def batch_update_or_create(cls, model_list: List[Union[T, str]]) -> Union[List[str], None]: - """批量更新 - - Args: - model_list: 模型实例或ID字符串的列表 - - Returns: - 更新后的记录ID列表 - - Raises: - TypeError: 当列表中包含不支持的类型时 - """ - if model_list is None: - return model_list - - row_ids = [] - for model in model_list: - if isinstance(model, str): - row_ids.append(model) # just link - elif isinstance(model, cls.MODEL_CLASS): - row_id = cls.update_or_create(model) - row_ids.append(row_id) - else: - raise TypeError( - f"Unsupported {cls.__name__} data type: {type(model).__name__}. " - f"Expected str or {cls.MODEL_CLASS.__name__}" - ) - - return row_ids - - @classmethod - def _load_relations(cls, model: T, include_system_fields: bool = True) -> T: - """加载关联数据（子类可覆盖） - - Args: - model: 模型实例 - include_system_fields: 是否包含系统字段 - - Returns: - 加载了关联数据的模型实例 - """ - return model - - @classmethod - def _prepare_for_save(cls, model: T) -> T: - """保存前准备（子类可覆盖） - - Args: - model: 模型实例 - - Returns: - 准备好的模型实例 - """ - return model - - -class BaseSimpleEntity(ABC): - """简化的工作表实体基类（不使用模型）""" - - WORKSHEET_ID: str - - @classmethod - def list(cls, filter_dict: dict) -> List[Dict]: - """列表查询 - - Args: - filter_dict: 过滤条件字典 - - Returns: - 字典列表 - """ - return WorksheetRow.list(cls.WORKSHEET_ID, filter_dict, include_system_fields=False) - - @classmethod - def get(cls, row_id: str) -> Dict: - """获取单条记录 - - Args: - row_id: 记录ID - - Returns: - 字典 - """ - return WorksheetRow.get(cls.WORKSHEET_ID, row_id, include_system_fields=False) - - @classmethod - def create(cls, fields: List[Dict]) -> str: - """创建记录 - - Args: - fields: 字段列表 - - Returns: - 新创建的记录ID - """ - return WorksheetRow.create(cls.WORKSHEET_ID, fields) - - @classmethod - def update(cls, row_id: str, fields: List[Dict]) -> str: - """更新记录 - - Args: - row_id: 记录ID - fields: 字段列表 - - Returns: - 更新的记录ID - """ - return WorksheetRow.update(cls.WORKSHEET_ID, row_id, fields) diff --git a/PLUGINS/SIRP/sirpcoremodel.py b/PLUGINS/SIRP/sirpcoremodel.py index da4eb50..f3f3839 100644 --- a/PLUGINS/SIRP/sirpcoremodel.py +++ b/PLUGINS/SIRP/sirpcoremodel.py @@ -418,8 +418,7 @@ class CaseModel(BaseSystemModel): # 创建记录填写字段 title: Optional[str] = Field(default="", description="Case title (案例标题)") - severity: Optional[Severity] = Field(default=None, - description="Analyst-assessed severity (严重程度)") + severity: Optional[Severity] = Field(default=None, description="Analyst-assessed severity (严重程度)") impact: Optional[Impact] = Field(default=None, description="Analyst-assessed impact (影响)") priority: Optional[CasePriority] = Field(default=None, description="Response priority (响应优先级)")