update typed

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" 或 ["<targetid>"]
+    NE = "ne"  # 不等于 "London" 或 ["<targetid>"]
+    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"  # 同时包含 ["<id1>", "<id2>"]
+    BELONGS_TO = "belongsto"  # 属于 ["<departmentid>"]
+    NOT_BELONGS_TO = "notbelongsto"  # 不属于 ["<departmentid>"]
+    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}"

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):

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 # 内部字段

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)