From 13bf06debfc22854acb52124abc21ef7a35aafe0 Mon Sep 17 00:00:00 2001 From: rootkit Date: Mon, 19 Jan 2026 18:30:50 +0800 Subject: [PATCH] update typed --- PLUGINS/{ => Mock}/SIRP/fields_debug.py | 0 PLUGINS/{ => Mock}/SIRP/list_data.py | 0 PLUGINS/SIRP/nocolyapi.py | 69 ++- PLUGINS/SIRP/sirpapi.py | 144 +++++- PLUGINS/SIRP/sirptype.py | 51 +- PLUGINS/SIRP/test_sirptype.py | 651 ++++++++++++++++++++++++ 6 files changed, 853 insertions(+), 62 deletions(-) rename PLUGINS/{ => Mock}/SIRP/fields_debug.py (100%) rename PLUGINS/{ => Mock}/SIRP/list_data.py (100%) create mode 100644 PLUGINS/SIRP/test_sirptype.py diff --git a/PLUGINS/SIRP/fields_debug.py b/PLUGINS/Mock/SIRP/fields_debug.py similarity index 100% rename from PLUGINS/SIRP/fields_debug.py rename to PLUGINS/Mock/SIRP/fields_debug.py diff --git a/PLUGINS/SIRP/list_data.py b/PLUGINS/Mock/SIRP/list_data.py similarity index 100% rename from PLUGINS/SIRP/list_data.py rename to PLUGINS/Mock/SIRP/list_data.py diff --git a/PLUGINS/SIRP/nocolyapi.py b/PLUGINS/SIRP/nocolyapi.py index 19ab773..df967fe 100644 --- a/PLUGINS/SIRP/nocolyapi.py +++ b/PLUGINS/SIRP/nocolyapi.py @@ -1,4 +1,6 @@ -from typing import TypedDict, Literal, List, Union, Any, Dict +from __future__ import annotations + +from typing import TypedDict, Dict import requests from requests.adapters import HTTPAdapter @@ -57,18 +59,47 @@ class OptionType(TypedDict): score: float -# Define the recursive types first -class ConditionType(TypedDict): - type: Literal["condition"] +from enum import Enum +from typing import List, Union, Any, Literal, Optional +from pydantic import BaseModel, Field + + +class Operator(str, Enum): + """查询运算符枚举""" + EQ = "eq" # 等于 "Beijing" 或 [""] + NE = "ne" # 不等于 "London" 或 [""] + GT = "gt" # 大于 20 或 "2025-02-06 00:00:00" + GE = "ge" # 大于等于 10 + LT = "lt" # 小于 20 + LE = "le" # 小于等于 100 + IN = "in" # 是其中一个 ["value1", "value2"] + NOT_IN = "notin" # 不是任意一个 ["value1", "value2"] + CONTAINS = "contains" # 包含 "Ch" 或 ["销售部", "市场部"] + NOT_CONTAINS = "notcontains" # 不包含 "Ch" 或 ["销售部", "市场部"] + CONCURRENT = "concurrent" # 同时包含 ["", ""] + BELONGS_TO = "belongsto" # 属于 [""] + NOT_BELONGS_TO = "notbelongsto" # 不属于 [""] + STARTS_WITH = "startswith" # 开头是 "张" + NOT_STARTS_WITH = "notstartswith" # 开头不是 "李" + ENDS_WITH = "endswith" # 结尾是 "公司" + NOT_ENDS_WITH = "notendswith" # 结尾不是 "有限公司" + BETWEEN = "between" # 在范围内 ["2025-01-01", "2025-01-31"] + NOT_BETWEEN = "notbetween" # 不在范围内 ["10", "20"] + IS_EMPTY = "isempty" # 为空 (不需要 value) + IS_NOT_EMPTY = "isnotempty" # 不为空 (不需要 value) + + +class Condition(BaseModel): + type: Literal["condition"] = "condition" field: str - operator: str - value: Any + operator: Operator = Field(..., description="运算符列表") + value: Optional[Any] = None -class GroupType(TypedDict): - type: Literal["group"] - logic: Literal["AND", "OR"] - children: List[Union["GroupType", ConditionType]] +class Group(BaseModel): + type: Literal["group"] = "group" + logic: Literal["AND", "OR"] = "AND" + children: List[Union[Group, Condition]] class Worksheet(object): @@ -108,7 +139,7 @@ class WorksheetRow(object): pass @staticmethod - def get(worksheet_id: str, row_id: str, include_system_fields=True): + 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( @@ -127,7 +158,7 @@ class WorksheetRow(object): raise Exception(f"error_code: {response_data.get('error_code')} error_msg: {response_data.get('error_msg')}") @staticmethod - def _format_row(row, fields, include_system_fields=True): + def _format_row(row, fields, include_system_fields=True) -> dict: data_new = {} for alias in row: if alias in SYSTEM_FIELDS: @@ -192,7 +223,7 @@ class WorksheetRow(object): filter_data["value"] = [value_to_key.get(v, v) for v in filter_data["value"]] @staticmethod - def list(worksheet_id: str, filter: dict, include_system_fields=True): + def list(worksheet_id: str, filter: dict, include_system_fields=True) -> List: url = f"{SIRP_URL}/api/v3/app/worksheets/{worksheet_id}/rows/list" all_rows = [] page_index = 1 @@ -242,7 +273,7 @@ class WorksheetRow(object): return all_rows @staticmethod - def create(worksheet_id: str, fields: list, trigger_workflow: bool = False): + def create(worksheet_id: str, fields: List, trigger_workflow: bool = True): fields = [ field for field in fields if field.get("value") is not None and field.get("value") != "" ] @@ -269,7 +300,7 @@ class WorksheetRow(object): raise e @staticmethod - def update(worksheet_id: str, row_id: str, fields: list): + def update(worksheet_id: str, row_id: str, fields: List, trigger_workflow: bool = True): fields = [ field for field in fields if field.get("value") is not None @@ -277,7 +308,7 @@ class WorksheetRow(object): url = f"{SIRP_URL}/api/v3/app/worksheets/{worksheet_id}/rows/{row_id}" data = { - "triggerWorkflow": True, + "triggerWorkflow": trigger_workflow, "fields": fields } response = HTTP_SESSION.patch(url, @@ -292,13 +323,13 @@ class WorksheetRow(object): raise Exception(f"error_code: {response_data.get('error_code')} error_msg: {response_data.get('error_msg')}") @staticmethod - def delete(worksheet_id: str, row_ids: list): + def delete(worksheet_id: str, row_ids: list, trigger_workflow: bool = True): url = f"{SIRP_URL}/api/v3/app/worksheets/{worksheet_id}/rows/batch" data = { "rowids": row_ids, - "triggerWorkflow": True, + "triggerWorkflow": trigger_workflow, } response = HTTP_SESSION.delete(url, @@ -314,7 +345,7 @@ class WorksheetRow(object): @staticmethod def relations(worksheet_id: str, row_id: str, field: str, relation_worksheet_id: str, include_system_fields: bool = True, page_size: int = 1000, - page_index: int = None): + page_index: int = 1): fields = Worksheet.get_fields(relation_worksheet_id) url = f"{SIRP_URL}/api/v3/app/worksheets/{worksheet_id}/rows/{row_id}/relations/{field}" diff --git a/PLUGINS/SIRP/sirpapi.py b/PLUGINS/SIRP/sirpapi.py index d4fdfd2..62af989 100644 --- a/PLUGINS/SIRP/sirpapi.py +++ b/PLUGINS/SIRP/sirpapi.py @@ -3,12 +3,14 @@ from enum import StrEnum from typing import TypedDict, List, Optional, Union, Dict, Any, NotRequired, Literal 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 +from PLUGINS.SIRP.nocolyapi import WorksheetRow, OptionSet, Group, Condition, Operator +from PLUGINS.SIRP.sirptype import EnrichmentModel, ArtifactModel class InputCase(TypedDict): @@ -58,6 +60,80 @@ class InputArtifact(TypedDict): 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 + + +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: + # 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 + + +class Enrichment(object): + WORKSHEET_ID = "enrichment" + + def __init__(self): + pass + + @staticmethod + def get(rowid, include_system_fields=True) -> EnrichmentModel: + result = WorksheetRow.get(Enrichment.WORKSHEET_ID, rowid, include_system_fields=include_system_fields) + model = EnrichmentModel(**result) + return model + + @staticmethod + def list(model: Group, include_system_fields=True) -> List[EnrichmentModel]: + filter = model.model_dump() + result = WorksheetRow.list(Enrichment.WORKSHEET_ID, filter, include_system_fields=include_system_fields) + model_list = [] + for one in result: + model_list.append(EnrichmentModel(**one)) + return model_list + + @staticmethod + def update(model: EnrichmentModel) -> str: + if model.rowid is not None: + fields = model_to_fields(model) + rowid = WorksheetRow.update(Enrichment.WORKSHEET_ID, model.rowid, fields) + else: + raise Exception("Enrichment rowid is None, cannot update.") + return rowid + + @staticmethod + def create(model: EnrichmentModel) -> str: + fields = model_to_fields(model) + rowid = WorksheetRow.create(Enrichment.WORKSHEET_ID, fields) + return rowid + + @staticmethod + def update_or_create(model: EnrichmentModel) -> str: + fields = model_to_fields(model) + if model.rowid is None: + rowid = WorksheetRow.create(Enrichment.WORKSHEET_ID, fields) + else: + rowid = WorksheetRow.update(Enrichment.WORKSHEET_ID, model.rowid, fields) + return rowid + + class Artifact(object): WORKSHEET_ID = "artifact" @@ -65,38 +141,56 @@ class Artifact(object): pass @staticmethod - def get(rowid, include_system_fields=False): - artifact = WorksheetRow.get(Artifact.WORKSHEET_ID, rowid, include_system_fields=include_system_fields) - return artifact + 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 + + return model @staticmethod - def list(filter: dict): - result = WorksheetRow.list(Artifact.WORKSHEET_ID, filter) + def list(model: Group, include_system_fields=True) -> List[ArtifactModel]: + filter = model.model_dump() + result = WorksheetRow.list(Artifact.WORKSHEET_ID, filter, include_system_fields=include_system_fields) return result @staticmethod - def update(rowid, fields: List): - row_id = WorksheetRow.update(Artifact.WORKSHEET_ID, rowid, fields) - return row_id + def update_or_create(model: ArtifactModel) -> str: - @staticmethod - def create(fields: List): - row_id = WorksheetRow.create(Artifact.WORKSHEET_ID, fields) - return row_id + # 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") - @staticmethod - def update_or_create(fields: List, filters: dict) -> List: - rows = Artifact.list(filters) - if rows: - row_id_list = [] - for row in rows: - rowid = row['rowid'] - rowid_updated = Artifact.update(rowid, fields) - row_id_list.append(rowid_updated) - return row_id_list + model.enrichments = enrichments_rowid_list + + fields = model_to_fields(model) + if model.rowid is None: + rowid = WorksheetRow.create(Artifact.WORKSHEET_ID, fields) else: - rowid_created = Artifact.create(fields) - return [rowid_created] + rowid = WorksheetRow.update(Artifact.WORKSHEET_ID, model.rowid, fields) + return rowid class Alert(object): diff --git a/PLUGINS/SIRP/sirptype.py b/PLUGINS/SIRP/sirptype.py index 3744d71..c49c3ab 100644 --- a/PLUGINS/SIRP/sirptype.py +++ b/PLUGINS/SIRP/sirptype.py @@ -3,7 +3,7 @@ from __future__ import annotations from datetime import datetime from typing import List, Optional, Literal, Any, Union -from pydantic import BaseModel, Field, field_validator, ConfigDict +from pydantic import BaseModel, Field, field_validator, ConfigDict, field_serializer class AccountModel(BaseModel): @@ -76,17 +76,29 @@ class BaseSystemModel(BaseModel): def parse_datetime(cls, v: Any) -> Any: if isinstance(v, str) and v.strip(): try: - return datetime.strptime(v, "%Y-%m-%d %H:%M:%S") + return datetime.strptime(v, "%Y-%m-%dT%H:%M:%SZ") except ValueError: return v return v + @field_serializer( + "ctime", "utime", "wfctime", "wfrtime", "wfcotime", "wfdtime", + "created_time", "modified_time", "first_seen_time", "last_seen_time", "acknowledged_time", "closed_time", + check_fields=False, + when_used="json" + ) + def serialize_datetime(self, v: Any) -> Any: + if isinstance(v, datetime): + # 强制输出为你需要的格式 + return v.strftime("%Y-%m-%dT%H:%M:%SZ") + return v + class MessageModel(BaseSystemModel): playbook_rowid: str = Field(..., description="所属Playbook的唯一行ID") node: Optional[str] = Field(default="", description="消息来源的节点名称或ID") content: Optional[str] = Field(default="", description="消息的文本内容") - json: Optional[str] = Field(default="", description="消息的JSON格式内容，通常用于工具调用和返回") + data: Optional[str] = Field(default="", description="消息的JSON格式内容，通常用于工具调用和返回") type: Optional[Literal["SystemMessage", "HumanMessage", "ToolMessage", "AIMessage", None]] = Field(default=None, description="消息类型，用于区分不同角色的发言") @@ -99,12 +111,15 @@ class PlaybookModel(BaseSystemModel): remark: Optional[str] = Field(default="", description="关于Playbook执行的备注信息") type: Optional[Literal["CASE", "ALERT", "ARTIFACT", None]] = Field(default=None, description="Playbook关联的对象类型") name: Optional[str] = Field(default="", description="执行的Playbook的名称") - messages: Optional[List[MessageModel]] = Field(default=None, description="Playbook执行过程中的所有消息记录，构成对话历史") + user_input: Optional[str] = Field(default="", description="用户对Playbook的初始输入或后续指令") user: Optional[AccountModel] = Field(default=None, description="发起或与Playbook交互的用户") + # 关联表 + messages: Optional[List[Union[MessageModel, str]]] = Field(default=None, description="Playbook执行过程中的所有消息记录，构成对话历史") -class Knowledge(BaseSystemModel): + +class KnowledgeModel(BaseSystemModel): title: str = Field(..., description="知识库条目的标题") body: Optional[str] = Field(default="", description="知识库条目的正文内容") using: Optional[bool] = Field(default=False, description="当前是否正在使用该知识") @@ -114,8 +129,8 @@ class Knowledge(BaseSystemModel): class EnrichmentModel(BaseSystemModel): name: str = Field(..., description="富化信息的名称或标题") - type: Literal["Other"] = Field(..., description="富化信息的类型") - provider: Optional[Literal["Other", None]] = Field(default=None, description="富化信息的提供方，例如威胁情报厂商") + type: str = Field(..., description="富化信息的类型", json_schema_extra={"type": 2}) + provider: str = Field(..., description="富化信息的提供方，例如威胁情报厂商", json_schema_extra={"type": 2}) created_time: Optional[Union[datetime, str]] = Field(default=None, description="富化信息创建的时间") value: str = Field(..., description="富化信息的具体值") src_url: Optional[str] = Field(default="", description="富化信息的来源URL，方便溯源") @@ -141,13 +156,13 @@ class ArtifactModel(BaseSystemModel): description="实体在事件中扮演的角色, 如攻击者(Actor)、受害者(Target)等") owner: Optional[str] = Field(default="", description="实体归属的系统或用户") value: Optional[str] = Field(default="", description="实体的具体值, 如 '192.168.1.1'") - reputation_provider: Optional[str] = Field(default="", description="提供信誉评分的威胁情报厂商名称") + reputation_provider: Optional[str] = Field(default="", description="提供信誉评分的威胁情报厂商名称", json_schema_extra={"type": 2}) reputation_score: Optional[Literal[ 'Unknown', 'Very Safe', 'Safe', 'Probably Safe', 'Leans Safe', 'May not be Safe', 'Exercise Caution', 'Suspicious/Risky', 'Possibly Malicious', 'Probably Malicious', 'Malicious', 'Other', None]] = Field( default=None, description="实体的信誉评分") # 关联表 - enrichments: Optional[List[EnrichmentModel]] = Field(default=[], description="针对此实体的一系列富化结果") + enrichments: Optional[List[Union[EnrichmentModel, str]]] = Field(default=None, description="针对此实体的一系列富化结果") # None 时表示无需处理,[] 时表示要将 link 清空 # playbooks: Optional[Any] = "" # 内部字段 # playbook: Optional[Literal['TI Enrichment By AlienVaultOTX', 'TI Enrichment By Mock', None]] = None # 内部字段 @@ -182,7 +197,7 @@ class AlertModel(BaseSystemModel): src_url: Optional[str] = Field(default="", description="在源安全产品中查看此告警的URL") source_uid: Optional[str] = Field(default="", description="告警在源产品中的唯一ID") - data_sources: Optional[List[str]] = Field(default=[], description="告警的数据来源，如'EDR', 'Firewall'等") + data_sources: Optional[List[str]] = Field(default=[], description="告警的数据来源，如 'EDR', 'Firewall' 等") analytic: Optional[Any] = Field(default="", description="分析引擎的详细信息") analytic_name: Optional[str] = Field(default="", description="分析引擎的名称") @@ -224,10 +239,10 @@ class AlertModel(BaseSystemModel): summary_ai: Optional[str] = Field(default="", description="AI提供的汇总摘要") # 关联表 - case: Optional[CaseModel] = Field(default=None, description="此告警关联到的安全事件（Case）") - attachments: Optional[AttachmentModel] = Field(default=[], description="告警的附件") - artifacts: Optional[List[ArtifactModel]] = Field(default=[], description="从告警中提取出的实体（Artifact）列表") - enrichments: Optional[List[EnrichmentModel]] = Field(default=[], description="对整个告警进行的富化结果") + 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="对整个告警进行的富化结果") # playbooks: Optional[Any] = "" # 内部字段 # playbook: Optional[Literal["Alert Analysis Agent", None]] = None # 内部字段 @@ -285,10 +300,10 @@ class CaseModel(BaseSystemModel): respond_time: Optional[Any] = Field(default=None, description="事件处置完成时间(closed_time), 用于计算MTTR") # 关联表 - attachments: Optional[List[AttachmentModel]] = Field(default=[], description="与事件相关的附件列表") - tickets: Optional[List[TicketModel]] = Field(default=[], description="与此事件关联的外部工单列表") - enrichments: Optional[List[EnrichmentModel]] = Field(default=[], description="对整个事件进行的富化结果") - alerts: Optional[List[AlertModel]] = Field(default=[], description="合并到此事件中的告警列表") + 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="合并到此事件中的告警列表") # playbooks: Optional[Any] = "" # 内部字段 # playbook: Optional[Literal["Threat Hunting Agent", "L3 SOC Analyst Agent", "L3 SOC Analyst Agent With Tools", None]] = None # 内部字段 diff --git a/PLUGINS/SIRP/test_sirptype.py b/PLUGINS/SIRP/test_sirptype.py new file mode 100644 index 0000000..bc1e510 --- /dev/null +++ b/PLUGINS/SIRP/test_sirptype.py @@ -0,0 +1,651 @@ +import json +from datetime import datetime, timedelta, timezone + +from PLUGINS.SIRP.nocolyapi import Group, Condition, Operator +from PLUGINS.SIRP.sirpapi import Enrichment, Artifact +from PLUGINS.SIRP.sirptype import CaseModel, AlertModel, ArtifactModel, EnrichmentModel, TicketModel + +now = datetime.now(timezone.utc) +past_10m = now - timedelta(minutes=10) +past_5m = now - timedelta(minutes=5) + + +def generate_test_cases(): + """ + Generates three distinct and meticulously detailed test cases for security incidents, + ensuring 100% field coverage for all specified models as per user's strict requirements. + """ + + # --- Reusable Enrichment Snippets --- + enrichment_otx = EnrichmentModel( + name="OTX Pulse for evil-domain.com", + type="Other", + provider="Other", + created_time=now, + value="evil-domain.com", + src_url="https://otx.alienvault.com/indicator/domain/evil-domain.com", + desc="This domain is associated with the 'Gootkit' malware family.", + data=json.dumps({"pulse_count": 42, "tags": ["malware", "c2", "gootkit"]}) + ) + + enrichment_virustotal = EnrichmentModel( + name="VirusTotal Report for Hash 'a1b2c3d4...'", + type="Other", + provider="Other", + created_time=now, + value="a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2", + src_url="https://www.virustotal.com/gui/file/a1b2c3d4e5f6.../detection", + desc="72/75 vendors flagged this as malicious 'Trojan.Generic'.", + data=json.dumps({"scan_id": "a1b2c3d4e5f6-1678886400", "positives": 72, "total": 75}) + ) + + # === Case 1: Phishing Email Attack (100% Coverage) === + case1_phishing = CaseModel( + title="Phishing Campaign Detected - 'Urgent Payroll Update'", + severity="High", + impact="Medium", + priority="High", + src_url="https://sirp.example.com/cases/1", + confidence="High", + description="A targeted phishing campaign was identified. The email lured users to a fake login page to harvest credentials and deployed malware via an attachment.", + category="Email", + tags=["phishing", "credential-harvesting", "malware-delivery", "FIN-department"], + created_time=now, + status="In Progress", + acknowledged_time=now, + comment="L1 Analyst: Confirmed phishing. Escalating to L2 for impact analysis and remediation tracking.", + closed_time=None, + verdict=None, + summary="", + correlation_uid="CORR-PHISH-XYZ-123", + workbook="### Phishing Investigation Playbook\n1. Analyze headers (`done`)\n2. Detonate URL/Attachment (`done`)\n3. Identify recipients (`in-progress`)\n4. Purge emails from mailboxes\n5. Reset compromised user passwords\n", + analysis_rationale_ai="The email originates from an external, un-reputable domain and uses urgent language, a common phishing tactic. The URL leads to a non-standard login page with a self-signed certificate. The attachment hash matches known malware.", + recommended_actions_ai="- Block sender domain 'evil-domain.com'\n- Reset passwords for all users who clicked the link\n- Scan all endpoints for the malware hash 'a1b2c3d4e5f6...'", + attack_stage_ai="Initial Access, Execution", + severity_ai="High", + confidence_ai="High", + threat_hunting_report_ai="Threat hunting query initiated to find other emails from the same sender IP or with similar subject lines across the organization.", + # Time-based fields for metrics + start_time=past_10m.isoformat(), + end_time=None, + detect_time=past_5m.isoformat(), + acknowledge_time=now.isoformat(), + respond_time=None, + tickets=[ + TicketModel( + status='In Progress', + type='Jira', + title='[Security] Investigate Phishing Campaign SEC-1234', + uid='SEC-1234', + src_url='https://jira.example.com/browse/SEC-1234' + ) + ], + enrichments=[ + EnrichmentModel( + name="Affected Business Unit", type="Other", provider="Other", + value="Finance Department", desc="Internal CMDB Information: High-value target." + ) + ], + alerts=[ + AlertModel( + title="User Reported Phishing Email via Outlook Plugin", + severity="Medium", + impact="Low", + disposition="Notified", + action="Observed", + confidence="High", + uid="ALERT-USER-001", + labels=["user-reported", "phishing"], + desc="User 'john.doe' reported a suspicious email with subject 'Urgent Payroll Update'.", + created_time=past_5m, + modified_time=now, + first_seen_time=past_10m, + last_seen_time=past_10m, + rule_id="USER-REPORT-01", + rule_name="User Reported Phishing", + correlation_uid="CORR-PHISH-XYZ-123", + count=1, + src_url="https://exchange.example.com/messages/msg-id-12345", + source_uid="MSG-ID-12345", + data_sources=["MS Exchange", "Outlook Plugin"], + analytic=json.dumps({"plugin_version": "1.2.3"}), + analytic_name="Phishing Report Plugin", + analytic_type="Tagging", + analytic_state="Active", + analytic_desc="Identifies emails reported by users.", + tactic="Reconnaissance", + technique="T1598.003", + sub_technique="", + mitigation="User Training, Email Filtering", + product_category="Email", + product_vender="Microsoft", + product_name="Outlook", + product_feature="Phishing Report Add-in", + policy_name="", + policy_type=None, + policy_desc="", + risk_level="Medium", + risk_details="Potential for credential theft.", + status="New", + status_detail="Awaiting analyst review.", + remediation="", + comment="Initial report from user.", + unmapped=json.dumps({"x-original-ip": "123.123.123.123"}), + raw_data=json.dumps({"subject": "Urgent Payroll Update", "from": "no-reply@evil-domain.com", "to": "john.doe@example.com"}), + summary_ai="A user reported a suspicious email with urgent language regarding payroll.", + case=None, + enrichments=[], + artifacts=[ + ArtifactModel( + name="no-reply@evil-domain.com", + type="Email Address", + role="Actor", + value="no-reply@evil-domain.com", + reputation_provider="Internal Blocklist", + reputation_score="Malicious", + enrichments=[enrichment_otx] + ), + ArtifactModel( + name="http://fake-payroll-login.com", + type="URL String", + role="Related", + value="http://fake-payroll-login.com", + reputation_score="Suspicious/Risky" + ) + ] + ), + AlertModel( + title="Malicious Attachment Blocked by Email Gateway", + severity="High", + impact="Medium", + disposition="Blocked", + action="Denied", + confidence="High", + uid="ALERT-GW-002", + labels=["malware", "email-gateway", "trojan"], + desc="Email Gateway blocked an attachment 'payroll_update.zip' containing known malware 'Trojan.Generic'.", + created_time=past_5m, + modified_time=now, + first_seen_time=past_10m, + last_seen_time=past_10m, + rule_id="MAL-ATTACH-101", + rule_name="BlockKnownMalwareAttachment.VirusTotal", + correlation_uid="CORR-PHISH-XYZ-123", + count=5, + src_url="https://gateway.example.com/logs/log-id-abcdef", + source_uid="log-id-abcdef", + data_sources=["Email Gateway", "VirusTotal API"], + analytic=json.dumps({"engine": "sig-matcher-v3"}), + analytic_name="Gateway Malware Scanner", + analytic_type="Rule", + analytic_state="Active", + analytic_desc="Blocks attachments with hashes matching high-confidence threat feeds.", + tactic="Execution", + technique="T1204.002", + sub_technique="", + mitigation="Email Attachment Sandboxing, Threat Intelligence Feed Integration", + product_category="Email", + product_vender="SecureMail Inc.", + product_name="SecureMail Gateway", + product_feature="AV-Scan-Module", + policy_name="Inbound-Malware-Policy", + policy_type=None, + policy_desc="Blocks all inbound attachments with a VT score > 50.", + risk_level="High", + risk_details="Malware could lead to endpoint compromise.", + status="Resolved", + status_detail="File was quarantined successfully.", + remediation="File quarantined. No user impact.", + comment="Blocked 5 attempts to deliver this file to different users.", + unmapped="", + raw_data=json.dumps({"attachment_hash": "a1b2c3d4e5f6...", "recipient_count": 5}), + summary_ai="The email gateway blocked a malicious attachment identified by its hash.", + case=None, + enrichments=[enrichment_virustotal], + artifacts=[ + ArtifactModel( + name="payroll_update.zip", + type="File Name", + role="Related", + value="payroll_update.zip" + ), + ArtifactModel( + name="a1b2c3d4e5f6...", + type="Hash", + role="Related", + value="a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2", + reputation_provider="VirusTotal", + reputation_score="Malicious", + enrichments=[enrichment_virustotal] + ) + ] + ) + ] + ) + + # === Case 2: Endpoint Lateral Movement (100% Coverage) === + case2_lateral_movement = CaseModel( + title="Lateral Movement Detected via PsExec from DC01 to WS-FINANCE-05", + severity="Critical", + impact="High", + priority="Critical", + description="An attacker, having compromised the Domain Controller 'DC01', is attempting to move laterally to a high-value workstation 'WS-FINANCE-05' in the Finance department using PsExec.", + category="EDR", + tags=["lateral-movement", "psexec", "golden-ticket", "domain-compromise"], + created_time=now, + status="Resolved", + acknowledged_time=past_5m, + comment="Incident Response Complete. IOCs have been added to blocklists. Awaiting final report.", + closed_time=now, + verdict="True Positive", + summary="Attacker compromised DC01 and moved to WS-FINANCE-05. Both hosts have been isolated and are pending reimaging. All domain admin credentials have been rotated.", + correlation_uid="CORR-LAT-MOV-456", + workbook="### Lateral Movement Playbook\n1. Isolate source and destination (`done`)\n2. Dump memory from hosts (`done`)\n3. Analyze for persistence (`done`)\n4. Rotate credentials (`done`)", + analysis_rationale_ai="PsExec execution from a domain controller to a workstation is highly anomalous. The initial compromise vector on DC01 appears to be related to a credential dumping alert moments before the lateral movement.", + recommended_actions_ai="- Isolate both DC01 and WS-FINANCE-05 immediately.\n- Investigate DC01 for initial compromise.\n- Rotate all privileged credentials.", + attack_stage_ai="Lateral Movement", + severity_ai="Critical", + confidence_ai="High", + threat_hunting_report_ai="", + start_time=past_10m.isoformat(), + end_time=now.isoformat(), + detect_time=past_5m.isoformat(), + acknowledge_time=past_5m.isoformat(), + respond_time=now.isoformat(), + tickets=[ + TicketModel( + status='Resolved', + type='ServiceNow', + title='CRITICAL: Active Lateral Movement Detected', + uid='INC001002', + src_url='https://servicenow.example.com/nav_to.do?uri=incident.do?sys_id=INC001002' + ) + ], + enrichments=[], + alerts=[ + AlertModel( + title="Suspicious Service Installation (PSEXESVC) on WS-FINANCE-05", + severity="High", + impact="High", + disposition="Detected", + action="Observed", + confidence="High", + uid="ALERT-EDR-101", + labels=["psexec", "lateral-movement"], + desc="PsExec service (PSEXESVC.exe) was created and started on WS-FINANCE-05, originating from DC01.", + created_time=past_5m, + modified_time=now, + first_seen_time=past_5m, + last_seen_time=past_5m, + rule_id="EDR-RULE-LM-001", + rule_name="PsExec Service Execution", + correlation_uid="CORR-LAT-MOV-456", + count=1, + src_url="https://edr.example.com/alerts/ALERT-EDR-101", + source_uid="be7a2f3a-8b1d-4a8a-9b1a-5d1e3e0f1e1a", + data_sources=["EDR", "Windows Security Events"], + analytic=json.dumps({"SysmonEventID": 7}), + analytic_name="Sysmon Behavioral Detection", + analytic_type="Behavioral", + analytic_state="Active", + analytic_desc="Detects the creation of the PsExec service executable.", + tactic="Lateral Movement", + technique="T1569.002", + sub_technique="", + mitigation="Restrict Service Creation, Network Segmentation", + product_category="EDR", + product_vender="CrowdStrike", + product_name="Falcon", + product_feature="Behavioral-Detection-Engine", + policy_name="Default Workstation Policy", + policy_type="Identity Policy", + policy_desc="Monitors for suspicious service installations.", + risk_level="High", + risk_details="Indicates an attacker is moving through the network.", + status="Archived", + status_detail="Alert has been correlated into Case-2.", + remediation="Host was isolated by SOAR playbook.", + comment="Clear indicator of lateral movement.", + unmapped="", + raw_data=json.dumps({"event_id": 4697, "service_name": "PSEXESVC", "source_host": "DC01"}), + summary_ai="PsExec was used to move from DC01 to a finance workstation.", + case=None, + enrichments=[], + artifacts=[ + ArtifactModel( + name="PSEXESVC.exe", + type="Process Name", + role="Related", + value="PSEXESVC.exe", + owner="System" + ), + ArtifactModel( + name="DC01", + type="Hostname", + role="Actor", + value="DC01", + description="Source of lateral movement." + ) + ] + ), + AlertModel( + title="Credential Dumping via LSASS Memory Access on DC01", + severity="Critical", + impact="Critical", + disposition="Alert", + action="Observed", + confidence="High", + uid="ALERT-EDR-100", + labels=["credential-dumping", "mimikatz", "lsass"], + desc="An untrusted process 'mimikatz.exe' accessed the memory of lsass.exe, indicating credential dumping.", + created_time=past_10m, + modified_time=now, + first_seen_time=past_10m, + last_seen_time=past_10m, + rule_id="EDR-RULE-CD-005", + rule_name="LSASS Memory Access by Untrusted Process", + correlation_uid="CORR-LAT-MOV-456", + count=1, + src_url="https://edr.example.com/alerts/ALERT-EDR-100", + source_uid="aa1b2c3d-4e5f-6a7b-8c9d-0e1f2a3b4c5d", + data_sources=["EDR"], + analytic=json.dumps({"target_process": "lsass.exe"}), + analytic_name="Credential Access Detection", + analytic_type="Behavioral", + analytic_state="Active", + analytic_desc="Monitors for processes reading memory from LSASS.", + tactic="Credential Access", + technique="T1003.001", + sub_technique="", + mitigation="Credential Guard, LSA Protection", + product_category="EDR", + product_vender="CrowdStrike", + product_name="Falcon", + product_feature="Credential-Theft-Protection", + policy_name="Domain Controller Policy", + policy_type=None, + policy_desc="", + risk_level="Critical", + risk_details="Domain credentials may be compromised.", + status="Archived", + status_detail="Alert has been correlated into Case-2.", + remediation="", + comment="This was likely the initial point of credential theft enabling lateral movement.", + unmapped="", + raw_data=json.dumps({"source_process": "mimikatz.exe", "target_process": "lsass.exe"}), + summary_ai="Credential dumping tool Mimikatz was detected on the domain controller.", + case=None, + enrichments=[], + artifacts=[ + ArtifactModel( + name="lsass.exe", + type="Process Name", + role="Target", + value="lsass.exe", + owner="System" + ), + ArtifactModel( + name="mimikatz.exe", + type="Process Name", + role="Actor", + value="mimikatz.exe", + description="Anomalous process accessing LSASS." + ) + ] + ) + ] + ) + + # === Case 3: DNS Tunneling C2 (100% Coverage) === + case3_dns_tunnel = CaseModel( + title="Suspected DNS Tunneling for C2 Communication from WS-MARKETING-12", + severity="Medium", + impact="Low", + priority="Medium", + description="An endpoint 'WS-MARKETING-12' is exhibiting DNS query patterns indicative of DNS tunneling, likely for command-and-control (C2) communication. This is a low-and-slow exfiltration or C2 method.", + category="NDR", + tags=["dns-tunneling", "c2", "ndr", "exfiltration"], + created_time=now, + status="On Hold", + acknowledged_time=now, + comment="Awaiting more data. Placed host in a monitoring group. No immediate action taken to avoid tipping off the attacker.", + closed_time=None, + verdict="Suspicious", + summary="", + correlation_uid="CORR-DNS-TUN-789", + workbook="### DNS Tunneling Playbook\n1. Analyze query patterns (TXT/NULL record types, query length)\n2. Check domain reputation\n3. Perform packet capture on host\n4. Compare against baseline DNS traffic", + analysis_rationale_ai="The high volume of TXT queries to a single, non-business related domain is a strong indicator of DNS tunneling. The query payloads appear to be encoded.", + recommended_actions_ai="- Place the host in a sinkhole network to observe C2 traffic safely.\n- Do not block immediately to gather more intelligence on the attacker's infrastructure.", + attack_stage_ai="Command and Control", + severity_ai="Medium", + confidence_ai="Medium", + threat_hunting_report_ai="", + start_time=past_10m.isoformat(), + end_time=None, + detect_time=now.isoformat(), + acknowledge_time=now.isoformat(), + respond_time=None, + tickets=[], + enrichments=[], + alerts=[ + AlertModel( + title="Anomalous DNS Query Volume (TXT Records)", + severity="Medium", + impact="Low", + action="Observed", + disposition="Logged", + confidence="Medium", + uid="ALERT-NDR-301", + labels=["dns-tunneling", "ndr"], + desc="Endpoint 10.1.1.5 (WS-MARKETING-12) made an unusually high number of DNS TXT queries to a single domain, c2.bad-actor-infra.net.", + created_time=now, + modified_time=now, + first_seen_time=past_10m, + last_seen_time=now, + rule_id="NDR-DNS-007", + rule_name="High Volume of DNS TXT Queries to Single Domain", + correlation_uid="CORR-DNS-TUN-789", + count=245, + src_url="https://ndr.example.com/alerts/ALERT-NDR-301", + source_uid="ndr-flow-98765", + data_sources=["NDR", "DNS Logs"], + analytic=json.dumps({"query_type": "TXT", "threshold": 50, "time_window": "5m"}), + analytic_name="DNS Exfiltration Detector", + analytic_type="Behavioral", + analytic_state="Active", + analytic_desc="Flags high-frequency TXT/NULL queries.", + tactic="Command and Control", + technique="T1071.004", + sub_technique="", + mitigation="DNS Sinkholing, Egress Traffic Filtering", + product_category="NDR", + product_vender="Vectra", + product_name="Cognito", + product_feature="DNS-Analytics", + policy_name="", + policy_type=None, + policy_desc="", + risk_level="Medium", + risk_details="Potential for covert C2 channel or data exfiltration.", + status="New", + status_detail="", + remediation="", + comment="", + unmapped="", + raw_data=json.dumps({"query_count": 245, "domain": "c2.bad-actor-infra.net"}), + summary_ai="High volume of DNS TXT queries suggests a DNS tunnel.", + case=None, + enrichments=[enrichment_otx], + artifacts=[ + ArtifactModel( + name="10.1.1.5", + type="IP Address", + role="Actor", + value="10.1.1.5", + owner="Workstation-Pool-DHCP" + ), + ArtifactModel( + name="c2.bad-actor-infra.net", + type="Hostname", + role="Related", + value="c2.bad-actor-infra.net", + reputation_score="Suspicious/Risky" + ) + ] + ), + AlertModel( + title="Firewall Detected Unusually Long DNS Query", + severity="Low", + impact="Low", + action="Logged", + disposition="Allowed", + confidence="Low", + uid="ALERT-FW-905", + labels=["dns", "firewall"], + desc="A DNS query with an unusually long label (>63 chars) was observed, which can be an indicator of tunneling.", + created_time=past_5m, + modified_time=now, + first_seen_time=past_5m, + last_seen_time=past_5m, + rule_id="FW-DNS-002", + rule_name="Long DNS Label Detected", + correlation_uid="CORR-DNS-TUN-789", + count=1, + src_url="https://fw.example.com/logs/log-id-54321", + source_uid="log-id-54321", + data_sources=["Firewall"], + analytic=json.dumps({"label_length": 85}), + analytic_name="Firewall DNS Protocol Anomaly", + analytic_type="Rule", + analytic_state="Experimental", + analytic_desc="Flags DNS queries that violate standard label length.", + tactic="Command and Control", + technique="T1071.004", + sub_technique="", + mitigation="Egress DNS Filtering", + product_category="Cloud", + product_vender="Palo Alto", + product_name="PA-Series Firewall", + product_feature="DNS-Security", + policy_name="Default-DNS-Allow", + policy_type="Service Control Policy", + policy_desc="Default policy allowing outbound DNS traffic.", + risk_level="Low", + risk_details="Suspicious but could be a false positive from non-standard software.", + status="New", + status_detail="", + remediation="", + comment="Correlates with the NDR alert, increasing confidence.", + unmapped=json.dumps({"dns_flags": "RD"}), + raw_data=json.dumps({"qname": "verylonglabelthatmightbeencodeddata.c2.bad-actor-infra.net"}), + summary_ai="An unusually long DNS query was detected by the firewall.", + case=None, + enrichments=[], + artifacts=[ + ArtifactModel( + name="UDP-53", + type="Port", + role="Related", + value="53", + ), + ArtifactModel( + name="8.8.8.8", + type="IP Address", + role="Related", + value="8.8.8.8", + description="Public DNS Resolver" + ) + ] + ) + ] + ) + + return case1_phishing, case2_lateral_movement, case3_dns_tunnel + + +def test_generate_cases(): + """ + Prints the generated test cases to the console in JSON format. + """ + test_cases = generate_test_cases() + for i, case in enumerate(test_cases, 1): + print(f"--- Test Case {i}: {case.title} ---") + print(case.model_dump_json(indent=2)) + print("\n\n") + + +def test_enrichment(): + enrichment_to_convert = EnrichmentModel( + name="OTX Pulse for evil-domain.com", + type="Other", + provider="Other", + created_time=now, + value="evil-domain.com", + src_url="https://otx.alienvault.com/indicator/domain/evil-domain.com", + desc="This domain is associated with the 'Gootkit' malware family.", + data=json.dumps({"pulse_count": 42, "tags": ["malware", "c2", "gootkit"]}) + ) + # + # rowid = Enrichment.create(enrichment_to_convert) + # enrichment_to_convert.rowid = rowid + # Enrichment.get(rowid="761bf560-15d9-4137-8a18-62e243cb1ee9") + + filter_model = Group( + logic="AND", + children=[ + Condition( + field="type", + operator=Operator.IN, + value=["Other"] + ) + ] + ) + + Enrichment.list(filter_model) + + +if __name__ == "__main__": + import os + import django + + os.environ.setdefault("DJANGO_SETTINGS_MODULE", "ASP.settings") + django.setup() + + rowid = Artifact.get("0e4527f9-a0b9-4d71-a805-95a7d8d3267e") + + artifact_model = ArtifactModel( + rowid="0e4527f9-a0b9-4d71-a805-95a7d8d3267e", + name="http://fake-payroll-login.com1", + type="URL String", + role="Related", + owner="admin", + value="http://fake-payroll-login.com", + reputation_provider="OTX", + reputation_score="Suspicious/Risky", + enrichments=[ + EnrichmentModel( + rowid="761bf560-15d9-4137-8a18-62e243cb1ee9", + name="OTX Pulse for evil-domain.com update", + type="TI", + provider="OTX", + created_time=now, + value="evil-domain.com", + src_url="https://otx.alienvault.com/indicator/domain/evil-domain.com", + desc="This domain is associated with the 'Gootkit' malware family.", + data=json.dumps({"pulse_count": 42, "tags": ["malware", "c2", "gootkit"]}) + ), + EnrichmentModel( + name="OTX Pulse for fake-payroll-login.com", + type="Other", + provider="Other", + created_time=now, + value="fake-payroll-login.com", + src_url="https://otx.alienvault.com/indicator/domain/fake-payroll-login.com", + desc="This domain is associated with the 'Gootkit' malware family.", + data=json.dumps({"pulse_count": 42, "tags": ["malware", "c2", "gootkit"]}) + ) + + ] + ) + rowid = Artifact.update_or_create(artifact_model) + print(rowid)