diff --git a/PLUGINS/SIRP/old_nouse_code.py b/PLUGINS/SIRP/old_nouse_code.py new file mode 100644 index 0000000..e590097 --- /dev/null +++ b/PLUGINS/SIRP/old_nouse_code.py @@ -0,0 +1,72 @@ + +def create_alert_with_group_rule(alert: InputAlert, rule_def: GroupRule) -> str: + """ + Create alerts and cases using alert aggregation rules. + The function will automatically generate a deduplication fingerprint based on the definition of rule_def, and decide whether to create a new case or update an existing case. + """ + + # alert + row_id_alert = Alert.create(alert) + + artifacts = alert.get("artifact", []) + + # case + timestamp = string_to_timestamp(alert["alert_date"], "%Y-%m-%dT%H:%M:%SZ") + deduplication_key = rule_def.generate_deduplication_key(artifacts=artifacts, timestamp=timestamp) + + row = Case.get_by_deduplication_key(deduplication_key) + if row is None: + if rule_def.workbook is None: + workbook = "# There is no workbook for this source." + else: + workbook = rule_def.workbook + + case_status_new = OptionSet.get_option_key_by_name_and_value("case_status", "New") + + case: InputCase = { + "title": rule_def.generate_case_title(artifacts=artifacts), + "deduplication_key": deduplication_key, + "alert": [row_id_alert], + "case_status": case_status_new, + "created_date": get_current_time_str(), + "tags": alert["tags"], + "severity": alert["severity"], + "type": rule_def.source, + "description": alert["description"], + "workbook": workbook, + } + row_id_create = Case.create(case) + return row_id_create + else: + row_id_case = row.get("rowid") + existing_alerts = row.get("alert", []) + if row_id_alert not in existing_alerts: + existing_alerts.append(row_id_alert) + + case_field = [ + {"id": "alert", "value": existing_alerts}, + ] + + # change case severity if new alert severity is higher + if rule_def.follow_alert_severity: + option_new_score = OptionSet.get_option_by_name_and_value("alert_case_severity", alert["severity"]).get("score", 0) + + severity_value_exist = row.get("severity") + option_exist_score = OptionSet.get_option_by_name_and_value("alert_case_severity", severity_value_exist).get("score", 0) + + if option_new_score > option_exist_score: + severity = alert["severity"] + else: + severity = severity_value_exist + case_field.append({"id": "severity", "value": severity}) + + # append alert tags to case tags + if rule_def.append_alert_tags: + tags_exist = row.get("tags", []) + for tag in alert["tags"]: + if tag not in tags_exist: + tags_exist.append(tag) + case_field.append({"id": "tags", "value": tags_exist, "type": 2}) + + row_id_updated = Case.update(row_id_case, case_field) + return row_id_updated diff --git a/PLUGINS/SIRP/sirpapi.py b/PLUGINS/SIRP/sirpapi.py index a36cfdf..8fbcec9 100644 --- a/PLUGINS/SIRP/sirpapi.py +++ b/PLUGINS/SIRP/sirpapi.py @@ -1,73 +1,13 @@ -import os from enum import StrEnum -from typing import TypedDict, List, Optional, Union, Dict, Any, NotRequired, Literal +from typing import List, Dict, Any, Literal, Union import requests from pydantic import BaseModel -from Lib.api import string_to_timestamp, get_current_time_str from Lib.log import logger from PLUGINS.SIRP.CONFIG import SIRP_NOTICE_WEBHOOK -from PLUGINS.SIRP.grouprule import GroupRule from PLUGINS.SIRP.nocolyapi import WorksheetRow, OptionSet, Group, Condition, Operator -from PLUGINS.SIRP.sirptype import EnrichmentModel, ArtifactModel - - -class InputCase(TypedDict): - """ - Need to be consistent with the SIRP Case table structure - If there are new fields in SIRP, they need to be added here - """ - title: str - deduplication_key: str - case_status: str - created_date: str - tags: List[str] - severity: str - type: str - description: str - workbook: str - - # AI fields - confidence_ai: NotRequired[str] - analysis_rationale_ai: NotRequired[str] - recommended_actions_ai: NotRequired[str] - recommended_actions_ai: NotRequired[List[str]] - - alert: List[str] - - -class InputAlert(TypedDict): - source: str - rule_id: str - rule_name: str - name: str - alert_date: str - created_date: str - tags: List[str] - severity: str - reference: NotRequired[str] - source_data_identifier: NotRequired[str] - description: str - summary_ai: NotRequired[Optional[Union[str, Dict[str, Any]]]] - artifact: List[Dict] - raw_log: NotRequired[Optional[Union[str, Dict[str, Any]]]] - - -class InputArtifact(TypedDict): - type: str - value: str - enrichment: NotRequired[Dict[str, Any]] - - -# def model_to_fields(model_instance: BaseModel) -> List[Dict[str, Any]]: -# fields = [] -# model_dict = model_instance.model_dump(mode='json', exclude_unset=True) -# -# for key, value in model_dict.items(): -# fields.append({'id': key, 'value': value}) -# -# return fields +from PLUGINS.SIRP.sirptype import EnrichmentModel, ArtifactModel, AlertModel, CaseModel, TicketModel def model_to_fields(model_instance: BaseModel) -> List[Dict[str, Any]]: @@ -80,9 +20,6 @@ def model_to_fields(model_instance: BaseModel) -> List[Dict[str, Any]]: 'value': value } if field_info and field_info.json_schema_extra: - # custom_type = field_info.json_schema_extra.get('type') - # if custom_type is not None: - # field_item['type'] = custom_type field_item.update(field_info.json_schema_extra) fields.append(field_item) return fields @@ -109,6 +46,24 @@ class Enrichment(object): model_list.append(EnrichmentModel(**one)) return model_list + @staticmethod + def list_by_rowids(rowids: Union[List[str], None], include_system_fields=True) -> Union[List[EnrichmentModel], List[str], None]: + if rowids is not None and rowids != []: + filter_model = Group( + logic="AND", + children=[ + Condition( + field="rowid", + operator=Operator.IN, + value=rowids + ) + ] + ) + enrichment_list = Enrichment.list(filter_model, include_system_fields=include_system_fields) + return enrichment_list + else: + return rowids + @staticmethod def update(model: EnrichmentModel) -> str: if model.rowid is not None: @@ -118,6 +73,24 @@ class Enrichment(object): raise Exception("Enrichment rowid is None, cannot update.") return rowid + @staticmethod + def batch_update(model_list: List[Union[EnrichmentModel, str]]) -> Union[List[str], None]: + if model_list is not None: + rowids = [] + for model in model_list: + if isinstance(model, str): + rowids.append(model) # just link + continue + elif isinstance(model, EnrichmentModel): + rowid = Enrichment.update_or_create(model) # update or create record + rowids.append(rowid) + else: + raise Exception("Unsupported enrichment data type") + + return rowids + else: + return model_list + @staticmethod def create(model: EnrichmentModel) -> str: fields = model_to_fields(model) @@ -134,6 +107,88 @@ class Enrichment(object): return rowid +class Ticket(object): + WORKSHEET_ID = "ticket" + + def __init__(self): + pass + + @staticmethod + def get(rowid, include_system_fields=True) -> TicketModel: + result = WorksheetRow.get(Ticket.WORKSHEET_ID, rowid, include_system_fields=include_system_fields) + model = TicketModel(**result) + return model + + @staticmethod + def list(model: Group, include_system_fields=True) -> List[TicketModel]: + filter = model.model_dump() + result = WorksheetRow.list(Ticket.WORKSHEET_ID, filter, include_system_fields=include_system_fields) + model_list = [] + for one in result: + model_list.append(TicketModel(**one)) + return model_list + + @staticmethod + def list_by_rowids(rowids: Union[List[str], None], include_system_fields=True) -> Union[List[TicketModel], List[str], None]: + if rowids is not None and rowids != []: + filter_model = Group( + logic="AND", + children=[ + Condition( + field="rowid", + operator=Operator.IN, + value=rowids + ) + ] + ) + ticket_list = Ticket.list(filter_model, include_system_fields=include_system_fields) + return ticket_list + else: + return rowids + + @staticmethod + def update(model: TicketModel) -> str: + if model.rowid is not None: + fields = model_to_fields(model) + rowid = WorksheetRow.update(Ticket.WORKSHEET_ID, model.rowid, fields) + else: + raise Exception("Ticket rowid is None, cannot update.") + return rowid + + @staticmethod + def batch_update(model_list: List[Union[TicketModel, str]]) -> Union[List[str], None]: + if model_list is not None: + rowids = [] + for model in model_list: + if isinstance(model, str): + rowids.append(model) # just link + continue + elif isinstance(model, TicketModel): + rowid = Ticket.update_or_create(model) # update or create record + rowids.append(rowid) + else: + raise Exception("Unsupported ticket data type") + + return rowids + else: + return model_list + + @staticmethod + def create(model: TicketModel) -> str: + fields = model_to_fields(model) + rowid = WorksheetRow.create(Ticket.WORKSHEET_ID, fields) + return rowid + + @staticmethod + def update_or_create(model: TicketModel) -> str: + fields = model_to_fields(model) + if model.rowid is None: + rowid = WorksheetRow.create(Ticket.WORKSHEET_ID, fields) + else: + rowid = WorksheetRow.update(Ticket.WORKSHEET_ID, model.rowid, fields) + return rowid + + class Artifact(object): WORKSHEET_ID = "artifact" @@ -144,20 +199,9 @@ class Artifact(object): def get(rowid, include_system_fields=True) -> ArtifactModel: result = WorksheetRow.get(Artifact.WORKSHEET_ID, rowid, include_system_fields=include_system_fields) model = ArtifactModel(**result) - if model.enrichments is not None and model.enrichments != []: - # enrichments - filter_model = Group( - logic="AND", - children=[ - Condition( - field="rowid", - operator=Operator.IN, - value=model.enrichments - ) - ] - ) - enrichment_list = Enrichment.list(filter_model) - model.enrichments = enrichment_list + + # enrichments + model.enrichments = Enrichment.list_by_rowids(model.enrichments) return model @@ -168,40 +212,35 @@ class Artifact(object): artifact_list = [] for artifact_data in result: artifact_model = ArtifactModel(**artifact_data) - if artifact_model.enrichments is not None and artifact_model.enrichments != []: - # enrichments - filter_model = Group( - logic="AND", - children=[ - Condition( - field="rowid", - operator=Operator.IN, - value=artifact_model.enrichments - ) - ] - ) - enrichment_list = Enrichment.list(filter_model) - artifact_model.enrichments = enrichment_list + + # enrichments + artifact_model.enrichments = Enrichment.list_by_rowids(artifact_model.enrichments) + artifact_list.append(artifact_model) return artifact_list + @staticmethod + def list_by_rowids(rowids: Union[List[str], None], include_system_fields=True) -> Union[List[ArtifactModel], List[str], None]: + if rowids is not None and rowids != []: + filter_model = Group( + logic="AND", + children=[ + Condition( + field="rowid", + operator=Operator.IN, + value=rowids + ) + ] + ) + artifact_list = Artifact.list(filter_model, include_system_fields=include_system_fields) + return artifact_list + else: + return rowids + @staticmethod def update_or_create(model: ArtifactModel) -> str: - # enrichments - if model.enrichments is not None: - enrichments_rowid_list = [] - for enrichment in model.enrichments: - if isinstance(enrichment, str): - enrichments_rowid_list.append(enrichment) # just link - continue - elif isinstance(enrichment, EnrichmentModel): - rowid = Enrichment.update_or_create(enrichment) # update or create record - enrichments_rowid_list.append(rowid) - else: - raise Exception("Unsupported enrichment data type") - - model.enrichments = enrichments_rowid_list + model.enrichments = Enrichment.batch_update(model.enrichments) fields = model_to_fields(model) if model.rowid is None: @@ -210,104 +249,155 @@ class Artifact(object): rowid = WorksheetRow.update(Artifact.WORKSHEET_ID, model.rowid, fields) return rowid + @staticmethod + def batch_update(model_list: List[Union[ArtifactModel, str]]) -> Union[List[str], None]: + if model_list is not None: + rowids = [] + for model in model_list: + if isinstance(model, str): + rowids.append(model) # just link + continue + elif isinstance(model, ArtifactModel): + rowid = Artifact.update_or_create(model) # update or create record + rowids.append(rowid) + else: + raise Exception("Unsupported enrichment data type") + + return rowids + else: + return model_list + class Alert(object): WORKSHEET_ID = "alert" - ARTIFACT_FIELD_ID = "artifacts" def __init__(self): pass @staticmethod - def get(rowid, include_system_fields=False): - alert = WorksheetRow.get(Alert.WORKSHEET_ID, rowid, include_system_fields=include_system_fields) - artifacts = WorksheetRow.relations(Alert.WORKSHEET_ID, rowid, Alert.ARTIFACT_FIELD_ID, relation_worksheet_id=Artifact.WORKSHEET_ID, - include_system_fields=False) - alert[Alert.ARTIFACT_FIELD_ID] = artifacts - return alert + def get(rowid, include_system_fields=True) -> AlertModel: + result = WorksheetRow.get(Alert.WORKSHEET_ID, rowid, include_system_fields=include_system_fields) + model = AlertModel(**result) + + # artifacts + model.artifacts = Artifact.list_by_rowids(model.artifacts) + + # enrichments + model.enrichments = Enrichment.list_by_rowids(model.enrichments) + + return model @staticmethod - def update(rowid, fields: list): - row_id = WorksheetRow.update(Alert.WORKSHEET_ID, rowid, fields) - return row_id + def list(model: Group, include_system_fields=True) -> List[AlertModel]: + filter = model.model_dump() + result = WorksheetRow.list(Alert.WORKSHEET_ID, filter, include_system_fields=include_system_fields) + alert_list = [] + for alert_data in result: + alert_model = AlertModel(**alert_data) + + # artifacts + alert_model.artifacts = Artifact.list_by_rowids(alert_model.artifacts) + + # enrichments + alert_model.enrichments = Enrichment.list_by_rowids(alert_model.enrichments) + + alert_list.append(alert_model) + return alert_list @staticmethod - def create(alert: InputAlert): - artifact_rowid_list = [] - artifacts: list[dict] = alert.get("artifact", []) - for artifact in artifacts: - artifact_fields = [ - {"id": "type", "value": artifact.get("type")}, - {"id": "value", "value": artifact.get("value")}, - {"id": "enrichment", "value": artifact.get("enrichment")}, - ] - - artifact_filter = { - "type": "group", - "logic": "AND", - "children": [ - { - "type": "condition", - "field": "type", - "operator": "eq", - "value": artifact.get("type") - }, - { - "type": "condition", - "field": "value", - "operator": "eq", - "value": artifact.get("value") - } + def list_by_rowids(rowids: Union[List[str], None], include_system_fields=True) -> Union[List[AlertModel], List[str], None]: + if rowids is not None and rowids != []: + filter_model = Group( + logic="AND", + children=[ + Condition( + field="rowid", + operator=Operator.IN, + value=rowids + ) ] - } + ) + model_list = Alert.list(filter_model, include_system_fields=include_system_fields) + return model_list + else: + return rowids - row_id_list = Artifact.update_or_create(artifact_fields, artifact_filter) - artifact_rowid_list.extend(row_id_list) + @staticmethod + def update_or_create(model: AlertModel) -> str: - if alert.get("created_date") is None: - alert["created_date"] = get_current_time_str() + # artifacts + model.artifacts = Artifact.batch_update(model.artifacts) - alert_fields = [ - {"id": "tags", "value": alert.get("tags"), "type": 2}, - {"id": "severity", "value": alert.get("severity")}, - {"id": "source", "value": alert.get("source")}, - {"id": "alert_date", "value": alert.get("alert_date")}, - {"id": "created_date", "value": alert.get("created_date")}, - {"id": "reference", "value": alert.get("reference")}, - {"id": "description", "value": alert.get("description")}, - {"id": "raw_log", "value": alert.get("raw_log")}, - {"id": "rule_id", "value": alert.get("rule_id")}, - {"id": "rule_name", "value": alert.get("rule_name")}, - {"id": "name", "value": alert.get("name")}, - {"id": "summary_ai", "value": alert.get("summary_ai")}, - {"id": "artifact", "value": artifact_rowid_list}, - ] + # enrichments + model.enrichments = Enrichment.batch_update(model.enrichments) - # alert - row_id = WorksheetRow.create(Alert.WORKSHEET_ID, alert_fields) + fields = model_to_fields(model) + if model.rowid is None: + rowid = WorksheetRow.create(Alert.WORKSHEET_ID, fields) + else: + rowid = WorksheetRow.update(Alert.WORKSHEET_ID, model.rowid, fields) + return rowid - return row_id + @staticmethod + def batch_update(model_list: List[Union[AlertModel, str]]) -> Union[List[str], None]: + if model_list is not None: + rowids = [] + for model in model_list: + if isinstance(model, str): + rowids.append(model) # just link + continue + elif isinstance(model, AlertModel): + rowid = Alert.update_or_create(model) # update or create record + rowids.append(rowid) + else: + raise Exception("Unsupported enrichment data type") + + return rowids + else: + return model_list class Case(object): WORKSHEET_ID = "case" - ALERT_FIELD_ID = "alert" def __init__(self): pass @staticmethod - def get(rowid, include_system_fields=False) -> InputCase: - case = WorksheetRow.get(Case.WORKSHEET_ID, rowid, include_system_fields=include_system_fields) - # alert id - alerts = WorksheetRow.relations(Case.WORKSHEET_ID, rowid, Case.ALERT_FIELD_ID, relation_worksheet_id=Alert.WORKSHEET_ID, - include_system_fields=include_system_fields) - for alert in alerts: - artifacts = WorksheetRow.relations(Alert.WORKSHEET_ID, alert.get("rowid"), Alert.ARTIFACT_FIELD_ID, relation_worksheet_id=Artifact.WORKSHEET_ID, - include_system_fields=include_system_fields) - alert[Alert.ARTIFACT_FIELD_ID] = artifacts - case[Case.ALERT_FIELD_ID] = alerts - return case + def get(rowid, include_system_fields=True) -> CaseModel: + result = WorksheetRow.get(Case.WORKSHEET_ID, rowid, include_system_fields=include_system_fields) + model = CaseModel(**result) + + # alerts + model.alerts = Alert.list_by_rowids(model.alerts) + + # enrichments + model.enrichments = Enrichment.list_by_rowids(model.enrichments) + + # tickets + model.tickets = Ticket.list_by_rowids(model.tickets) + + return model + + @staticmethod + def list(model: Group, include_system_fields=True) -> List[CaseModel]: + filter = model.model_dump() + result = WorksheetRow.list(Case.WORKSHEET_ID, filter, include_system_fields=include_system_fields) + case_list = [] + for case_data in result: + case_model = CaseModel(**case_data) + + # alerts + case_model.alerts = Alert.list_by_rowids(case_model.alerts) + + # enrichments + case_model.enrichments = Enrichment.list_by_rowids(case_model.enrichments) + + # tickets + case_model.tickets = Ticket.list_by_rowids(case_model.tickets) + + case_list.append(case_model) + return case_list @staticmethod def get_raw_data(rowid, include_system_fields=False) -> Dict: @@ -321,14 +411,14 @@ class Case(object): case_clean = {key: case[key] for key in useful_case_fields if key in case} # alert id - alerts = WorksheetRow.relations(Case.WORKSHEET_ID, rowid, Case.ALERT_FIELD_ID, relation_worksheet_id=Alert.WORKSHEET_ID, + alerts = WorksheetRow.relations(Case.WORKSHEET_ID, rowid, "alerts", relation_worksheet_id=Alert.WORKSHEET_ID, include_system_fields=include_system_fields) alerts_clean = [] for alert in alerts: useful_alert_fields = ["rowid", 'severity', 'rule_id', 'rule_name', 'id'] alert_clean = {key: alert[key] for key in useful_alert_fields if key in alert} - artifacts = WorksheetRow.relations(Alert.WORKSHEET_ID, alert.get("rowid"), Alert.ARTIFACT_FIELD_ID, relation_worksheet_id=Artifact.WORKSHEET_ID, + artifacts = WorksheetRow.relations(Alert.WORKSHEET_ID, alert.get("rowid"), "artifacts", relation_worksheet_id=Artifact.WORKSHEET_ID, include_system_fields=include_system_fields) artifacts_clean = [] for artifact in artifacts: @@ -336,55 +426,30 @@ class Case(object): artifact_clean = {key: artifact[key] for key in useful_artifact_fields if key in artifact} artifacts_clean.append(artifact_clean) - alert_clean[Alert.ARTIFACT_FIELD_ID] = artifacts_clean + alert_clean["artifacts"] = artifacts_clean alerts_clean.append(alert_clean) - case_clean[Case.ALERT_FIELD_ID] = alerts_clean + case_clean["alerts"] = alerts_clean return case_clean @staticmethod - def create(case: InputCase): - case_fields = [ - {"id": "title", "value": case["title"]}, - {"id": "deduplication_key", "value": case["deduplication_key"]}, - {"id": "alert", "value": case["alert"]}, - {"id": "case_status", "value": case["case_status"]}, - {"id": "created_date", "value": case["created_date"]}, - {"id": "tags", "value": case["tags"], "type": 2}, - {"id": "severity", "value": case["severity"]}, - {"id": "type", "value": case["type"]}, - {"id": "description", "value": case["description"]}, - {"id": "workbook", "value": case["workbook"]}, - ] - row_id = WorksheetRow.create(Case.WORKSHEET_ID, case_fields) - return row_id + def update_or_create(model: CaseModel) -> str: - @staticmethod - def update(row_id, fields: list): - row_id = WorksheetRow.update(Case.WORKSHEET_ID, row_id, fields) - return row_id + # alerts + model.alerts = Alert.batch_update(model.alerts) - @staticmethod - def get_by_deduplication_key(deduplication_key: str): - filter = { - "type": "group", - "logic": "AND", - "children": [ - { - "type": "condition", - "field": "deduplication_key", - "operator": "eq", - "value": deduplication_key - }, - ] - } - rows = WorksheetRow.list(Case.WORKSHEET_ID, filter) - if rows: - if len(rows) > 1: - logger.warning(f"found multiple rows with deduplication_key {deduplication_key}") - return rows[0] + # enrichments + model.enrichments = Enrichment.batch_update(model.enrichments) + + # tickets + model.tickets = Ticket.batch_update(model.tickets) + + fields = model_to_fields(model) + if model.rowid is None: + rowid = WorksheetRow.create(Case.WORKSHEET_ID, fields) else: - return None + rowid = WorksheetRow.update(Case.WORKSHEET_ID, model.rowid, fields) + return rowid @staticmethod def get_by_case_id(case_id: str): @@ -408,21 +473,6 @@ class Case(object): else: return None - @staticmethod - def load_workbook_md(workbook_name: str) -> str: - ## TODO remove this function - - """ - Read the content of DATA/WORKBOOK/{workbook_name}.md according to the workbook name and return a string. - The path is relative to the project root (two levels up to the asf folder). - """ - base_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..')) - md_path = os.path.join(base_dir, 'DATA', 'WORKBOOK', f"{workbook_name}.md") - if not os.path.exists(md_path): - raise FileNotFoundError(f"workbook md not found: {md_path}") - with open(md_path, 'r', encoding='utf-8') as f: - return f.read() - class PlaybookStatus(StrEnum): SUCCESS = 'Success' @@ -574,76 +624,3 @@ class Notice(object): def send(user, title, body=None): result = requests.post(SIRP_NOTICE_WEBHOOK, json={"title": title, "body": body, "user": user}) return result - - -def create_alert_with_group_rule(alert: InputAlert, rule_def: GroupRule) -> str: - """ - Create alerts and cases using alert aggregation rules. - The function will automatically generate a deduplication fingerprint based on the definition of rule_def, and decide whether to create a new case or update an existing case. - """ - - # alert - row_id_alert = Alert.create(alert) - - artifacts = alert.get("artifact", []) - - # case - timestamp = string_to_timestamp(alert["alert_date"], "%Y-%m-%dT%H:%M:%SZ") - deduplication_key = rule_def.generate_deduplication_key(artifacts=artifacts, timestamp=timestamp) - - row = Case.get_by_deduplication_key(deduplication_key) - if row is None: - if rule_def.workbook is None: - workbook = "# There is no workbook for this source." - else: - workbook = rule_def.workbook - - case_status_new = OptionSet.get_option_key_by_name_and_value("case_status", "New") - - case: InputCase = { - "title": rule_def.generate_case_title(artifacts=artifacts), - "deduplication_key": deduplication_key, - "alert": [row_id_alert], - "case_status": case_status_new, - "created_date": get_current_time_str(), - "tags": alert["tags"], - "severity": alert["severity"], - "type": rule_def.source, - "description": alert["description"], - "workbook": workbook, - } - row_id_create = Case.create(case) - return row_id_create - else: - row_id_case = row.get("rowid") - existing_alerts = row.get("alert", []) - if row_id_alert not in existing_alerts: - existing_alerts.append(row_id_alert) - - case_field = [ - {"id": "alert", "value": existing_alerts}, - ] - - # change case severity if new alert severity is higher - if rule_def.follow_alert_severity: - option_new_score = OptionSet.get_option_by_name_and_value("alert_case_severity", alert["severity"]).get("score", 0) - - severity_value_exist = row.get("severity") - option_exist_score = OptionSet.get_option_by_name_and_value("alert_case_severity", severity_value_exist).get("score", 0) - - if option_new_score > option_exist_score: - severity = alert["severity"] - else: - severity = severity_value_exist - case_field.append({"id": "severity", "value": severity}) - - # append alert tags to case tags - if rule_def.append_alert_tags: - tags_exist = row.get("tags", []) - for tag in alert["tags"]: - if tag not in tags_exist: - tags_exist.append(tag) - case_field.append({"id": "tags", "value": tags_exist, "type": 2}) - - row_id_updated = Case.update(row_id_case, case_field) - return row_id_updated diff --git a/PLUGINS/SIRP/sirptype.py b/PLUGINS/SIRP/sirptype.py index c49c3ab..79c0c9a 100644 --- a/PLUGINS/SIRP/sirptype.py +++ b/PLUGINS/SIRP/sirptype.py @@ -236,11 +236,15 @@ class AlertModel(BaseSystemModel): raw_data: Optional[str] = Field(default="", description="原始告警日志，通常为JSON格式的字符串") + attachments: Optional[List[Union[AttachmentModel, str]]] = Field(default=[], description="告警的附件") + + # AI字段 summary_ai: Optional[str] = Field(default="", description="AI提供的汇总摘要") + # 反向关联 + case: Optional[str] = Field(default=None, description="此告警关联到的安全事件（Case）(只保留rowid,避免循环引用)") + # 关联表 - case: Optional[Union[CaseModel, str]] = Field(default=None, description="此告警关联到的安全事件（Case）") - attachments: Optional[List[Union[AttachmentModel, str]]] = Field(default=[], description="告警的附件") artifacts: Optional[List[Union[ArtifactModel, str]]] = Field(default=[], description="从告警中提取出的实体（Artifact）列表") enrichments: Optional[List[Union[EnrichmentModel, str]]] = Field(default=[], description="对整个告警进行的富化结果") @@ -269,6 +273,7 @@ class CaseModel(BaseSystemModel): assignee_l1: Optional[AccountModel] = Field(default=None, description="分配给L1一线分析师") acknowledged_time: Optional[Union[datetime, str]] = Field(default=None, description="L1分析师首次确认接收事件的时间") comment: Optional[str] = Field(default="", description="分析师对整个事件的评论或处置记录") + attachments: Optional[List[Union[AttachmentModel, str]]] = Field(default=[], description="与事件相关的附件列表") assignee_l2: Optional[AccountModel] = Field(default=None, description="分配或升级给L2二线分析师") assignee_l3: Optional[AccountModel] = Field(default=None, description="分配或升级给L3专家分析师") @@ -300,7 +305,6 @@ class CaseModel(BaseSystemModel): respond_time: Optional[Any] = Field(default=None, description="事件处置完成时间(closed_time), 用于计算MTTR") # 关联表 - attachments: Optional[List[Union[AttachmentModel, str]]] = Field(default=[], description="与事件相关的附件列表") tickets: Optional[List[Union[TicketModel, str]]] = Field(default=[], description="与此事件关联的外部工单列表") enrichments: Optional[List[Union[EnrichmentModel, str]]] = Field(default=[], description="对整个事件进行的富化结果") alerts: Optional[List[Union[AlertModel, str]]] = Field(default=[], description="合并到此事件中的告警列表")