@cryptotaxi247 / CoPilot / commits / cea2786f

Suricata alert (#134)

* provision suricata alert within graylog * Add get_notification_id function to retrieve notification id * Update SURICATA_ALERT to SURICATA_ALERT_SEVERITY_1 * Add Suricata alerts schema * Refactor Suricata alerts schema and services * suricata rule progress * precommit fixes going to revist after refactor of frontend / backend ... left off with build asset payload for iris...marked with a TODO

taylor_socfortress committed Feb 7, 2024 at 08:30 UTC cea2786fa1ae17dc3ef27149c1c7c0084c860be1
8 files changed +925 -6
backend/app/integrations/monitoring_alert/routes/monitoring_alert.py
+44 -1
@@ -25,6 +25,7 @@ from app.integrations.monitoring_alert.schema.monitoring_alert import (
25 from app.integrations.monitoring_alert.schema.monitoring_alert import (
26 WazuhAnalysisResponse,
27 )
28 +from app.integrations.monitoring_alert.services.suricata import analyze_suricata_alerts
29 from app.integrations.monitoring_alert.services.wazuh import analyze_wazuh_alerts
30
31 monitoring_alerts_router = APIRouter()
@@ -146,7 +147,7 @@ async def run_wazuh_analysis(
147
148 monitoring_alerts = await session.execute(
149 select(MonitoringAlerts).where(
149 - MonitoringAlerts.customer_code == request.customer_code and MonitoringAlerts.alert_source == "WAZUH",
150 + (MonitoringAlerts.customer_code == request.customer_code) & (MonitoringAlerts.alert_source == "WAZUH"),
151 ),
152 )
153 monitoring_alerts = monitoring_alerts.scalars().all()
@@ -160,3 +161,45 @@ async def run_wazuh_analysis(
161 await analyze_wazuh_alerts(monitoring_alerts, customer_meta, session)
162
163 return WazuhAnalysisResponse(success=True, message="Analysis completed successfully")
164 +
165 +
166 +@monitoring_alerts_router.post("/run_analysis/suricata", response_model=WazuhAnalysisResponse)
167 +async def run_suricata_analysis(
168 + request: MonitoringWazuhAlertsRequestModel,
169 + session: AsyncSession = Depends(get_db),
170 +) -> WazuhAnalysisResponse:
171 + """
172 + This route is used to run analysis on the monitoring alerts.
173 +
174 + 1. Get all the monitoring alerts from the database where the customer_code matches the customer_code provided
175 + and the alert_source is SURICATA.
176 +
177 + 2. Call the anlayze_wazuh_alerts function to analyze the alerts.
178 +
179 + Args:
180 + request (MonitoringWazuhAlertsRequestModel): The customer code.
181 + session (AsyncSession, optional): The database session. Defaults to Depends(get_db).
182 +
183 + Returns:
184 + WazuhAnalysisResponse: The response containing the analysis results.
185 + """
186 + logger.info(f"Running analysis for customer_code: {request.customer_code}")
187 +
188 + customer_meta = await get_customer_meta(request.customer_code, session)
189 +
190 + monitoring_alerts = await session.execute(
191 + select(MonitoringAlerts).where(
192 + (MonitoringAlerts.customer_code == request.customer_code) & (MonitoringAlerts.alert_source == "SURICATA"),
193 + ),
194 + )
195 + monitoring_alerts = monitoring_alerts.scalars().all()
196 +
197 + logger.info(f"Found {len(monitoring_alerts)} monitoring alerts")
198 +
199 + if not monitoring_alerts:
200 + raise HTTPException(status_code=404, detail="No monitoring alerts found")
201 +
202 + # Call the analyze_wazuh_alerts function to analyze the alerts
203 + await analyze_suricata_alerts(monitoring_alerts, customer_meta, session)
204 +
205 + return WazuhAnalysisResponse(success=True, message="Analysis completed successfully")
backend/app/integrations/monitoring_alert/routes/provision.py
+38 -5
@@ -14,9 +14,14 @@ from app.integrations.monitoring_alert.schema.provision import (
14 from app.integrations.monitoring_alert.schema.provision import (
15 ProvisionWazuhMonitoringAlertResponse,
16 )
17 +from app.integrations.monitoring_alert.services.provision import (
18 + provision_suricata_monitoring_alert,
19 +)
20 from app.integrations.monitoring_alert.services.provision import (
21 provision_wazuh_monitoring_alert,
22 )
23 +from app.integrations.utils.event_shipper import event_shipper
24 +from app.integrations.utils.schema import EventShipperPayload
25 from app.schedulers.models.scheduler import CreateSchedulerRequest
26 from app.schedulers.scheduler import add_scheduler_jobs
27
@@ -36,15 +41,22 @@ async def invoke_provision_wazuh_monitoring_alert(request: ProvisionMonitoringAl
41 )
42
43
39 -# ! Comment out for now ! #
40 -# async def provision_other_alert(request):
41 -# # Provision the other alert
42 -# pass
44 +async def invoke_provision_suricata_monitoring_alert(request: ProvisionMonitoringAlertRequest):
45 + # Provision the Suricata monitoring alert
46 + await provision_suricata_monitoring_alert(request)
47 + await add_scheduler_jobs(
48 + CreateSchedulerRequest(
49 + function_name="invoke_suricata_monitoring_alert",
50 + time_interval=5,
51 + job_id="invoke_suricata_monitoring_alert",
52 + ),
53 + )
54 +
55
56 # Create a dictionary that maps alert names to provision functions
57 PROVISION_FUNCTIONS = {
58 "WAZUH_SYSLOG_LEVEL_ALERT": invoke_provision_wazuh_monitoring_alert,
47 - # "OTHER_ALERT": provision_other_alert,
59 + "SURICATA_ALERT_SEVERITY_1": invoke_provision_suricata_monitoring_alert,
60 # Add more alert names and functions as needed
61 }
62
@@ -102,3 +114,24 @@ async def provision_monitoring_alert_route(
114 await provision_function(request)
115
116 return ProvisionWazuhMonitoringAlertResponse(success=True, message="Wazuh monitoring alerts provisioned.")
117 +
118 +
119 +@monitoring_alerts_provision_router.post(
120 + "/provision/testing",
121 + response_model=ProvisionWazuhMonitoringAlertResponse,
122 + description="Used for testing purposes. To test, upload a JSON document.",
123 +)
124 +async def provision_monitoring_alert_testing_route(
125 + request: dict,
126 +) -> ProvisionWazuhMonitoringAlertResponse:
127 + """
128 + Used for testing purposes.
129 + """
130 + message = EventShipperPayload(
131 + customer_code="replace_me",
132 + integration="testing",
133 + version="1.0",
134 + **request,
135 + )
136 + await event_shipper(message)
137 + return ProvisionWazuhMonitoringAlertResponse(success=True, message="Event sent to log shipper successfully.")
backend/app/integrations/monitoring_alert/schema/monitoring_alert.py
+179
@@ -228,3 +228,182 @@ class WazuhIrisAlertPayload(BaseModel):
228
229 def to_dict(self):
230 return self.dict(exclude_none=True)
231 +
232 +
233 +########### ! SURICATA ALERTS SCHEMA ! ###########
234 +class SuricataSourceModel(BaseModel):
235 + alert_signature: str = Field(..., description="Signature of the alert")
236 + alert_severity: int = Field(..., description="Severity level of the alert")
237 + alert_signature_id: int = Field(..., description="Signature ID of the alert")
238 + src_ip: str = Field(..., description="Source IP address")
239 + dest_ip: str = Field(..., description="Destination IP address")
240 + app_proto: str = Field(..., description="Application protocol")
241 + agent_labels_customer: str = Field(..., description="Customer of the agent")
242 + timestamp: str = Field(..., description="The timestamp of the alert.")
243 + timestamp_utc: Optional[str] = Field(
244 + ...,
245 + description="The UTC timestamp of the alert.",
246 + )
247 + time_field: Optional[str] = Field(
248 + "timestamp",
249 + description="The timefield of the alert to be used when creating the IRIS alert.",
250 + )
251 + date: Optional[float] = Field(
252 + None,
253 + description="Date of the alert in Unix timestamp",
254 + )
255 + alert_metadata_tag: Optional[str] = Field(
256 + None,
257 + description="Metadata tag for the alert",
258 + )
259 + alert_gid: Optional[int] = Field(None, description="Alert group ID")
260 +
261 + class Config:
262 + allow_population_by_field_name = True
263 + extra = Extra.allow
264 +
265 + def to_dict(self):
266 + return self.dict(exclude_none=True)
267 +
268 +
269 +class SuricataAlertModel(BaseModel):
270 + _index: str
271 + _id: str
272 + _version: int
273 + _source: SuricataSourceModel
274 + asset_type_id: Optional[int] = Field(
275 + None,
276 + description="The asset type id of the alert which is needed for when we add the asset to IRIS.",
277 + )
278 + ioc_value: Optional[str] = Field(
279 + None,
280 + description="The IoC value of the alert which is needed for when we add the IoC to IRIS.",
281 + )
282 + ioc_type: Optional[str] = Field(
283 + None,
284 + description="The IoC type of the alert which is needed for when we add the IoC to IRIS.",
285 + )
286 +
287 + class Config:
288 + extra = Extra.allow
289 +
290 +
291 +########### ! Create Suricata Alerts In IRIS Schemas ! ###########
292 +class SuricataIrisAsset(BaseModel):
293 + asset_name: Optional[str] = Field(
294 + "Asset Does Not Apply to Suricata Alerts",
295 + description="Name of the asset",
296 + example="Server01",
297 + )
298 + asset_ip: Optional[str] = Field(
299 + "Asset Does Not Apply to Suricata Alerts",
300 + description="IP address of the asset",
301 + example="192.168.1.1",
302 + )
303 + asset_description: Optional[str] = Field(
304 + "Asset Does Not Apply to Suricata Alerts",
305 + description="Description of the asset",
306 + example="Windows Server",
307 + )
308 + asset_type_id: Optional[int] = Field(
309 + 9,
310 + description="Type ID of the asset",
311 + example=1,
312 + )
313 +
314 +
315 +class SuricataIrisIoc(BaseModel):
316 + ioc_value: str = Field(
317 + ...,
318 + description="Value of the IoC",
319 + example="www.google.com",
320 + )
321 + ioc_description: str = Field(
322 + ...,
323 + description="Description of the IoC",
324 + example="Google",
325 + )
326 + ioc_tlp_id: int = Field(1, description="TLP ID of the IoC", example=1)
327 + ioc_type_id: int = Field(20, description="Type ID of the IoC", example=20)
328 +
329 +
330 +class SuricataIrisAlertContext(BaseModel):
331 + _source: SuricataSourceModel
332 + alert_id: str = Field(..., description="ID of the alert", example="123")
333 + alert_name: str = Field(
334 + ...,
335 + description="Name of the alert",
336 + example="Intrusion Detected",
337 + )
338 + alert_level: int = Field(..., description="Severity level of the alert", example=3)
339 + rule_id: int = Field(
340 + ...,
341 + description="ID of the Suricata rule that triggered the alert",
342 + example="2001",
343 + )
344 + src_ip: str = Field(
345 + ...,
346 + description="Source IP address of the alert",
347 + example="1.1.1.1",
348 + )
349 + dest_ip: str = Field(
350 + ...,
351 + description="Destination IP address of the alert",
352 + example="8.8.8.8",
353 + )
354 + app_proto: str = Field(
355 + ...,
356 + description="Application protocol of the alert",
357 + example="TCP",
358 + )
359 +
360 +
361 +class SuricataIrisAlertPayload(BaseModel):
362 + alert_title: str = Field(
363 + ...,
364 + description="Title of the alert",
365 + example="Intrusion Detected",
366 + )
367 + alert_description: str = Field(
368 + ...,
369 + description="Description of the alert",
370 + example="Intrusion Detected by Firewall",
371 + )
372 + alert_source: str = Field(..., description="Source of the alert", example="Wazuh")
373 + assets: List[SuricataIrisAsset] = Field(..., description="List of affected assets")
374 + alert_source_link: str = Field(
375 + ...,
376 + description="Link to the alert within Grafana",
377 + example="https://grafana.com",
378 + )
379 + alert_status_id: int = Field(..., description="Status ID of the alert", example=3)
380 + alert_severity_id: int = Field(
381 + ...,
382 + description="Severity ID of the alert",
383 + example=5,
384 + )
385 + alert_customer_id: int = Field(
386 + ...,
387 + description="Customer ID related to the alert",
388 + example=1,
389 + )
390 + alert_source_content: Dict[str, Any] = Field(
391 + ...,
392 + description="Original content from the alert source",
393 + )
394 + alert_context: SuricataIrisAlertContext = Field(
395 + ...,
396 + description="Contextual information about the alert",
397 + )
398 + alert_iocs: Optional[List[IrisIoc]] = Field(
399 + None,
400 + description="List of IoCs related to the alert",
401 + )
402 + alert_source_event_time: str = Field(
403 + ...,
404 + description="Timestamp of the alert",
405 + example="2021-01-01T00:00:00.000Z",
406 + )
407 +
408 + def to_dict(self):
409 + return self.dict(exclude_none=True)
backend/app/integrations/monitoring_alert/schema/provision.py
+4
@@ -19,6 +19,10 @@ class AvailableMonitoringAlerts(str, Enum):
19 "it triggers an alert that is created within DFIR-IRIS. Ensure that you have a pipeline "
20 "rule that sets the SYSLOG_LEVEL field to ALERT when the Wazuh rule level is greater than 11."
21 )
22 + SURICATA_ALERT_SEVERITY_1 = (
23 + "This alert monitors the Suricata logs. When an the alert_severity field is 1, it triggers "
24 + "an alert that is created within DFIR-IRIS. Ensure that you have a pipeline rule that sets "
25 + )
26
27
28 class AvailableMonitoringAlertsResponse(BaseModel):
backend/app/integrations/monitoring_alert/services/provision.py
+122
@@ -93,6 +93,27 @@ async def check_if_url_whitelist_entry_exists(url: str) -> bool:
93 return False
94
95
96 +async def get_notification_id(notification_title: str) -> Optional[str]:
97 + """
98 + Get the notification id.
99 +
100 + Args:
101 + notification_title (str): The notification title.
102 +
103 + Returns:
104 + Optional[str]: The notification id if it exists, None otherwise.
105 + """
106 + event_notifications_response = await get_all_event_notifications()
107 + if not event_notifications_response.success:
108 + raise HTTPException(status_code=500, detail="Failed to collect event notifications")
109 + event_notifications_response = GraylogEventNotificationsResponse(**event_notifications_response.dict())
110 + logger.info(f"Event notifications collected: {event_notifications_response.event_notifications}")
111 + for event_notification in event_notifications_response.event_notifications.notifications:
112 + if event_notification.title == notification_title:
113 + return event_notification.id
114 + return None
115 +
116 +
117 async def build_url_whitelisted_entries(whitelist_url_model: GraylogUrlWhitelistEntryConfig) -> GraylogUrlWhitelistEntries:
118 """
119 Builds the URL Whitelisted Entries model.
@@ -284,3 +305,104 @@ async def provision_wazuh_monitoring_alert(request: ProvisionMonitoringAlertRequ
305 )
306
307 return ProvisionWazuhMonitoringAlertResponse(success=True, message="Wazuh monitoring alerts provisioned successfully")
308 +
309 +
310 +async def provision_suricata_monitoring_alert(request: ProvisionMonitoringAlertRequest) -> ProvisionWazuhMonitoringAlertResponse:
311 + """
312 + Provisions Suricata monitoring alerts.
313 +
314 + Returns:
315 + ProvisionWazuhMonitoringAlertResponse: The response indicating the success of provisioning the monitoring alerts.
316 + """
317 + #
318 + logger.info(f"Invoking provision_suricata_monitoring_alert with request: {request.dict()}")
319 + notification_exists = await check_if_event_notification_exists("SEND TO COPILOT")
320 + if not notification_exists:
321 + url_whitelisted = await check_if_url_whitelist_entry_exists(f"http://{os.getenv('SERVER_IP')}:5000/monitoring_alert/create")
322 + if not url_whitelisted:
323 + logger.info("Provisioning URL Whitelist")
324 + whitelisted_urls = await build_url_whitelisted_entries(
325 + whitelist_url_model=GraylogUrlWhitelistEntryConfig(
326 + id=await generate_random_id(),
327 + value=f"http://{os.getenv('SERVER_IP')}:5000/monitoring_alert/create",
328 + title="SEND TO COPILOT",
329 + type="literal",
330 + ),
331 + )
332 + await provision_webhook_url_whitelist(whitelisted_urls)
333 +
334 + logger.info("Provisioning SEND TO COPILOT Webhook")
335 + notification_id = await provision_webhook(
336 + GraylogAlertWebhookNotificationModel(
337 + title="SEND TO COPILOT",
338 + description="Send alert to Copilot",
339 + config={"url": f"http://{os.getenv('SERVER_IP')}:5000/monitoring_alert/create", "type": "http-notification-v1"},
340 + ),
341 + )
342 + logger.info(f"SEND TO COPILOT Webhook provisioned with id: {notification_id}")
343 + notification_id = await get_notification_id("SEND TO COPILOT")
344 + await provision_alert_definition(
345 + GraylogAlertProvisionModel(
346 + title="SURICATA ALERT SEVERITY 1",
347 + description="Alert on Suricata alerts",
348 + priority=2,
349 + config=GraylogAlertProvisionConfig(
350 + type="aggregation-v1",
351 + query="alert_severity:1 AND syslog_type:suricata",
352 + query_parameters=[],
353 + streams=[],
354 + group_by=[],
355 + series=[],
356 + conditions={
357 + "expression": None,
358 + },
359 + search_within_ms=await convert_seconds_to_milliseconds(request.search_within_last),
360 + execute_every_ms=await convert_seconds_to_milliseconds(request.execute_every),
361 + ),
362 + field_spec={
363 + "ALERT_ID": GraylogAlertProvisionFieldSpecItem(
364 + data_type="string",
365 + providers=[
366 + GraylogAlertProvisionProvider(
367 + type="template-v1",
368 + template="${source._id}",
369 + require_values=True,
370 + ),
371 + ],
372 + ),
373 + "CUSTOMER_CODE": GraylogAlertProvisionFieldSpecItem(
374 + data_type="string",
375 + providers=[
376 + GraylogAlertProvisionProvider(
377 + type="template-v1",
378 + template="${source.agent_labels_customer}",
379 + require_values=True,
380 + ),
381 + ],
382 + ),
383 + "ALERT_SOURCE": GraylogAlertProvisionFieldSpecItem(
384 + data_type="string",
385 + providers=[
386 + GraylogAlertProvisionProvider(
387 + type="template-v1",
388 + template="SURICATA",
389 + require_values=True,
390 + ),
391 + ],
392 + ),
393 + },
394 + key_spec=[],
395 + notification_settings=GraylogAlertProvisionNotificationSettings(
396 + grace_period_ms=0,
397 + backlog_size=None,
398 + ),
399 + notifications=[
400 + GraylogAlertProvisionNotification(
401 + notification_id=notification_id,
402 + ),
403 + ],
404 + alert=True,
405 + ),
406 + )
407 +
408 + return ProvisionWazuhMonitoringAlertResponse(success=True, message="Suricata monitoring alerts provisioned successfully")
backend/app/integrations/monitoring_alert/services/suricata.py new
+498
@@ -0,0 +1,498 @@
1 +import json
2 +from typing import Optional
3 +from typing import Set
4 +
5 +from fastapi import HTTPException
6 +from loguru import logger
7 +from sqlalchemy.ext.asyncio import AsyncSession
8 +
9 +from app.agents.routes.agents import get_agent
10 +from app.agents.schema.agents import AgentsResponse
11 +from app.connectors.dfir_iris.utils.universal import fetch_and_validate_data
12 +from app.connectors.dfir_iris.utils.universal import initialize_client_and_alert
13 +from app.connectors.wazuh_indexer.utils.universal import create_wazuh_indexer_client
14 +from app.db.universal_models import CustomersMeta
15 +from app.integrations.alert_creation.general.schema.alert import CreateAlertRequest
16 +from app.integrations.alert_creation.general.schema.alert import IrisAsset
17 +from app.integrations.alert_creation.general.schema.alert import IrisIoc
18 +from app.integrations.alert_creation.general.schema.alert import ValidIocFields
19 +from app.integrations.alert_creation.general.services.alert_multi_exclude import (
20 + AlertDetailsService,
21 +)
22 +from app.integrations.alert_escalation.schema.general_alert import (
23 + CreateAlertRequest as AddAlertRequest,
24 +)
25 +from app.integrations.alert_escalation.services.general_alert import (
26 + add_alert_to_document,
27 +)
28 +from app.integrations.monitoring_alert.models.monitoring_alert import MonitoringAlerts
29 +from app.integrations.monitoring_alert.schema.monitoring_alert import (
30 + FilterAlertsRequest,
31 +)
32 +from app.integrations.monitoring_alert.schema.monitoring_alert import SuricataAlertModel
33 +from app.integrations.monitoring_alert.schema.monitoring_alert import (
34 + SuricataIrisAlertContext,
35 +)
36 +from app.integrations.monitoring_alert.schema.monitoring_alert import SuricataIrisAsset
37 +from app.integrations.monitoring_alert.schema.monitoring_alert import (
38 + WazuhAnalysisResponse,
39 +)
40 +from app.integrations.monitoring_alert.schema.monitoring_alert import (
41 + WazuhIrisAlertContext,
42 +)
43 +from app.integrations.monitoring_alert.schema.monitoring_alert import (
44 + WazuhIrisAlertPayload,
45 +)
46 +from app.integrations.monitoring_alert.utils.db_operations import remove_alert_id
47 +from app.integrations.utils.alerts import get_asset_type_id
48 +from app.integrations.utils.alerts import validate_ioc_type
49 +from app.utils import get_customer_alert_settings
50 +
51 +
52 +def valid_ioc_fields() -> Set[str]:
53 + """
54 + Getter for the set of valid IoC fields.
55 + Returns
56 + -------
57 + Set[str]
58 + The set of valid IoC fields.
59 + """
60 + return {field.value for field in ValidIocFields}
61 +
62 +
63 +async def construct_alert_source_link(alert_details: CreateAlertRequest, session: AsyncSession) -> str:
64 + """
65 + Construct the alert source link for the alert details.
66 + Parameters
67 + ----------
68 + alert_details: CreateAlertRequest
69 + The alert details.
70 + Returns
71 + -------
72 + str
73 + The alert source link.
74 + """
75 + # Check if the alert has a process id and that it is not "No process ID found"
76 + if hasattr(alert_details, "process_id") and alert_details.process_id != "No process ID found":
77 + query_string = f"%22query%22:%22process_id:%5C%22{alert_details.process_id}%5C%22%20AND%20"
78 + else:
79 + query_string = f"%22query%22:%22_id:%5C%22{alert_details.id}%5C%22%20AND%20"
80 +
81 + grafana_url = (await get_customer_alert_settings(customer_code=alert_details.agent_labels_customer, session=session)).grafana_url
82 +
83 + return (
84 + f"{grafana_url}/explore?left=%5B%22now-6h%22,%22now%22,%22WAZUH%22,%7B%22refId%22:%22A%22,"
85 + f"{query_string}"
86 + f"agent_name:%5C%22{alert_details.agent_name}%5C%22%22,"
87 + "%22alias%22:%22%22,%22metrics%22:%5B%7B%22id%22:%221%22,%22type%22:%22logs%22,%22settings%22:%7B%22limit%22:%22500%22%7D%7D%5D,"
88 + "%22bucketAggs%22:%5B%5D,%22timeField%22:%22timestamp%22%7D%5D"
89 + )
90 +
91 +
92 +async def build_ioc_payload(alert_details: CreateAlertRequest) -> Optional[IrisIoc]:
93 + """
94 + Builds an IoC payload based on the provided alert details.
95 +
96 + Args:
97 + alert_details (CreateAlertRequest): The details of the alert.
98 +
99 + Returns:
100 + Optional[IrisIoc]: The constructed IoC payload, or None if no valid IoC fields are found.
101 + """
102 + for field in valid_ioc_fields():
103 + if hasattr(alert_details, field):
104 + ioc_value = getattr(alert_details, field)
105 + ioc_type = await validate_ioc_type(ioc_value=ioc_value)
106 + return IrisIoc(
107 + ioc_value=ioc_value,
108 + ioc_description="IoC found in alert",
109 + ioc_tlp_id=1,
110 + ioc_type_id=ioc_type,
111 + )
112 + return None
113 +
114 +
115 +async def build_asset_payload(agent_data: AgentsResponse, alert_details: CreateAlertRequest, session: AsyncSession) -> IrisAsset:
116 + """
117 + Build the payload for an IrisAsset object based on the agent data and alert details.
118 +
119 + Args:
120 + agent_data (AgentsResponse): The response containing agent data.
121 + alert_details: The details of the alert.
122 +
123 + Returns:
124 + IrisAsset: The constructed IrisAsset object.
125 + """
126 + # Get the agent_id based on the hostname from the Agents table
127 + if agent_data.success:
128 + return IrisAsset(
129 + asset_name=agent_data.agents[0].hostname,
130 + asset_ip=agent_data.agents[0].ip_address,
131 + asset_description=await construct_alert_source_link(alert_details, session=session),
132 + asset_type_id=await get_asset_type_id(agent_data.agents[0].os),
133 + asset_tags=f"agent_id:{agent_data.agents[0].agent_id}",
134 + )
135 + return IrisAsset()
136 +
137 +
138 +async def fetch_wazuh_indexer_details(alert_id: str, index: str) -> SuricataAlertModel:
139 + """
140 + Fetch the Wazuh alert details from the Wazuh-Indexer.
141 +
142 + Args:
143 + alert_id (str): The alert ID.
144 + index (str): The index.
145 +
146 + Returns:
147 + CollectAlertsResponse: The response from the Wazuh-Indexer.
148 + """
149 + logger.info(f"Fetching Wazuh alert details for alert_id: {alert_id} and index: {index}")
150 +
151 + es_client = await create_wazuh_indexer_client("Wazuh-Indexer")
152 + response = es_client.get(index=index, id=alert_id)
153 +
154 + return SuricataAlertModel(**response)
155 +
156 +
157 +async def fetch_alert_details(alert: MonitoringAlerts) -> SuricataAlertModel:
158 + logger.info(f"Analyzing Wazuh alert: {alert.alert_id}")
159 + alert_details = await fetch_wazuh_indexer_details(alert.alert_id, alert.alert_index)
160 + logger.info(f"Alert details: {alert_details}")
161 + return alert_details
162 +
163 +
164 +async def check_event_exclusion(alert_details: SuricataAlertModel, alert_detail_service: AlertDetailsService, session: AsyncSession):
165 + logger.info("Checking if alert is excluded due to multi exclusion.")
166 + logger.info(f"Alert details: {alert_details}")
167 + event_exclude_result = await alert_detail_service.collect_alert_timeline_process_id(
168 + agent_name=alert_details._source["agent_name"],
169 + process_id=alert_details._source.get("process_id", "n/a"),
170 + index=alert_details._index,
171 + session=session,
172 + )
173 + if event_exclude_result is True:
174 + raise HTTPException(
175 + status_code=400,
176 + detail="Alert excluded due to multi exclusion as set in the config.ini file.",
177 + )
178 + logger.info("Alert is not excluded due to multi exclusion.")
179 +
180 +
181 +async def check_if_open_alert_exists_in_iris(alert_details: SuricataAlertModel) -> list:
182 + """
183 + Check if the alert exists in IRIS.
184 +
185 + Args:
186 + alert_details (SuricataAlertModel): The alert details.
187 + session (AsyncSession): The database session.
188 +
189 + Returns:
190 + bool: True if the alert exists in IRIS, False otherwise.
191 + """
192 + client, alert_client = await initialize_client_and_alert("DFIR-IRIS")
193 + request = FilterAlertsRequest(alert_tags=alert_details._source["alert_signature_id"])
194 + params = construct_params(request)
195 + alert_exists = await fetch_and_validate_data(client, lambda: alert_client.filter_alerts(**params))
196 + logger.info(f"Alert exists: {alert_exists['data']['alerts']}")
197 + return alert_exists["data"]["alerts"][0]["alert_id"] if alert_exists["data"]["alerts"] else []
198 +
199 +
200 +def construct_params(request: FilterAlertsRequest) -> dict:
201 + """
202 + Constructs the parameters for the alert filtering request.
203 +
204 + Args:
205 + request (FilterAlertsRequest): The request object containing filtering criteria.
206 +
207 + Returns:
208 + dict: A dictionary of parameters for the alert filtering request.
209 + """
210 + params = {
211 + "page": request.page,
212 + "per_page": request.per_page,
213 + "sort": request.sort,
214 + "alert_tags": request.alert_tags,
215 + "alert_status_id": request.alert_status_id,
216 + # Add more parameters here as needed
217 + }
218 +
219 + # Remove parameters that have a value of None
220 + return {k: v for k, v in params.items() if v is not None}
221 +
222 +
223 +async def build_alert_context_payload(
224 + alert_details: CreateAlertRequest,
225 + agent_data: AgentsResponse,
226 + session: AsyncSession,
227 +) -> WazuhIrisAlertContext:
228 + """
229 + Builds the payload for the alert context.
230 +
231 + Args:
232 + alert_details (CreateAlertRequest): The details of the alert.
233 + agent_data (AgentsResponse): The agent data.
234 + session (AsyncSession): The async session.
235 +
236 + Returns:
237 + WazuhIrisAlertContext: The built alert context payload.
238 + """
239 + return WazuhIrisAlertContext(
240 + customer_iris_id=(
241 + await get_customer_alert_settings(customer_code=alert_details.agent_labels_customer, session=session)
242 + ).iris_customer_id,
243 + customer_name=(await get_customer_alert_settings(customer_code=alert_details.agent_labels_customer, session=session)).customer_name,
244 + customer_cases_index=(
245 + await get_customer_alert_settings(customer_code=alert_details.agent_labels_customer, session=session)
246 + ).iris_index,
247 + alert_name=alert_details.rule_description,
248 + alert_level=alert_details.rule_level,
249 + rule_id=alert_details.rule_id,
250 + rule_mitre_id=getattr(alert_details, "rule_mitre_id", "No rule mitre id found"),
251 + rule_mitre_tactic=getattr(
252 + alert_details,
253 + "rule_mitre_tactic",
254 + "No rule mitre tactic found",
255 + ),
256 + rule_mitre_technique=getattr(
257 + alert_details,
258 + "rule_mitre_technique",
259 + "No rule mitre technique found",
260 + ),
261 + )
262 +
263 +
264 +async def build_alert_payload(
265 + alert_details: CreateAlertRequest,
266 + agent_data,
267 + ioc_payload: Optional[IrisIoc],
268 + session: AsyncSession,
269 +) -> WazuhIrisAlertPayload:
270 + """
271 + Builds the payload for an alert based on the provided alert details, agent data, IoC payload, and session.
272 +
273 + Args:
274 + alert_details (CreateAlertRequest): The details of the alert.
275 + agent_data: The agent data associated with the alert.
276 + ioc_payload (Optional[IrisIoc]): The IoC payload associated with the alert.
277 + session (AsyncSession): The session used for database operations.
278 +
279 + Returns:
280 + WazuhIrisAlertPayload: The built alert payload.
281 + """
282 + asset_payload = await build_asset_payload(agent_data, alert_details=alert_details, session=session)
283 + context_payload = await build_alert_context_payload(alert_details=alert_details, agent_data=agent_data, session=session)
284 + timefield = (await get_customer_alert_settings(customer_code=alert_details.agent_labels_customer, session=session)).timefield
285 + # Get the timefield value from the alert_details
286 + if hasattr(alert_details, timefield):
287 + alert_details.time_field = getattr(alert_details, timefield)
288 + logger.info(f"Alert has context: {context_payload}")
289 + if ioc_payload:
290 + logger.info(f"Alert has IoC: {ioc_payload}")
291 + return WazuhIrisAlertPayload(
292 + alert_title=alert_details.rule_description,
293 + alert_description=alert_details.rule_description,
294 + alert_source="COPILOT WAZUH ANALYSIS",
295 + assets=[asset_payload],
296 + alert_status_id=3,
297 + alert_severity_id=5,
298 + alert_customer_id=(
299 + await get_customer_alert_settings(customer_code=alert_details.agent_labels_customer, session=session)
300 + ).iris_customer_id,
301 + alert_source_content=alert_details.to_dict(),
302 + alert_context=context_payload,
303 + alert_iocs=[ioc_payload],
304 + alert_source_event_time=alert_details.time_field,
305 + )
306 + else:
307 + logger.info("Alert does not have IoC")
308 + return WazuhIrisAlertPayload(
309 + alert_title=alert_details.rule_description,
310 + alert_description=alert_details.rule_description,
311 + alert_source="COPILOT WAZUH ANALYSIS",
312 + assets=[asset_payload],
313 + alert_status_id=3,
314 + alert_severity_id=5,
315 + alert_customer_id=(
316 + await get_customer_alert_settings(customer_code=alert_details.agent_labels_customer, session=session)
317 + ).iris_customer_id,
318 + alert_source_content=alert_details.to_dict(),
319 + alert_context=context_payload,
320 + alert_source_event_time=alert_details.time_field,
321 + )
322 +
323 +
324 +async def create_alert_details(alert_details: SuricataAlertModel) -> SuricataIrisAlertContext:
325 + """
326 + Create an alert details object from the Wazuh alert details.
327 +
328 + Args:
329 + alert_details (SuricataAlertModel): The Wazuh alert details.
330 +
331 + Returns:
332 + CreateAlertRequest: The alert details object.
333 + """
334 + logger.info(f"Creating alert details for alert: {alert_details}")
335 + return SuricataIrisAlertContext(
336 + index=alert_details._index,
337 + id=alert_details._id,
338 + alert_id=alert_details._source["alert_signature_id"],
339 + alert_name=alert_details._source["alert_signature"],
340 + alert_level=alert_details._source["alert_severity"],
341 + rule_id=alert_details._source["alert_signature_id"],
342 + src_ip=alert_details._source["src_ip"],
343 + dest_ip=alert_details._source["dest_ip"],
344 + app_proto=alert_details._source.get("app_proto", "No application protocol found"),
345 + )
346 +
347 +
348 +async def create_and_update_alert_in_iris(alert_details: SuricataAlertModel, session: AsyncSession) -> int:
349 + """
350 + Creates the alert, then updates the alert with the asset and IoC if available.
351 +
352 + Args:
353 + alert_details (SuricataAlertModel): The details of the alert.
354 + session (AsyncSession): The async session object.
355 +
356 + Returns:
357 + int: The ID of the created alert in IRIS.
358 + """
359 + logger.info("Alert does not exist in IRIS. Creating alert.")
360 + alert_details = await create_alert_details(alert_details)
361 + ioc_payload = await build_ioc_payload(alert_details)
362 + logger.info(f"Alert details: {alert_details}")
363 + # ! TODO: REVIST THIS TOMORROW
364 + iris_alert_payload = await build_alert_payload(
365 + alert_details=alert_details,
366 + # ! I DONT NEED TO BUILD THE AGENT DATA CAUSE I GET THIS FROM THE FUNCTION
367 + agent_data=SuricataIrisAsset(
368 + asset_name=alert_details.src_ip,
369 + asset_ip=alert_details.src_ip,
370 + asset_description="Source IP of the alert",
371 + asset_type_id=9,
372 + ),
373 + ioc_payload=ioc_payload,
374 + session=session,
375 + )
376 + client, alert_client = await initialize_client_and_alert("DFIR-IRIS")
377 + result = await fetch_and_validate_data(
378 + client,
379 + alert_client.add_alert,
380 + iris_alert_payload.to_dict(),
381 + )
382 + alert_id = result["data"]["alert_id"]
383 + logger.info(f"Successfully created alert {alert_id} in IRIS.")
384 + await fetch_and_validate_data(
385 + client,
386 + alert_client.update_alert,
387 + alert_id,
388 + {"alert_tags": f"{alert_details._source.alert_signature_id}"},
389 + )
390 + # Update the alert with the asset payload
391 + await fetch_and_validate_data(
392 + client,
393 + alert_client.update_alert,
394 + alert_id,
395 + {"assets": [dict(IrisAsset(**iris_alert_payload.assets[0].to_dict()))]},
396 + )
397 + if ioc_payload:
398 + await fetch_and_validate_data(
399 + client,
400 + alert_client.update_alert,
401 + alert_id,
402 + {"iocs": [dict(IrisIoc(**iris_alert_payload.alert_iocs[0].to_dict()))]},
403 + )
404 + return alert_id
405 +
406 +
407 +async def get_current_assets(client, alert_client, iris_alert_id):
408 + result = await fetch_and_validate_data(client, alert_client.get_alert, iris_alert_id)
409 + return result["data"]["assets"]
410 +
411 +
412 +async def update_alert_with_assets(client, alert_client, iris_alert_id, current_assets):
413 + await fetch_and_validate_data(
414 + client,
415 + alert_client.update_alert,
416 + iris_alert_id,
417 + {"assets": current_assets},
418 + )
419 +
420 +
421 +async def remove_duplicate_assets(current_assets):
422 + """
423 + Removes duplicate assets from the given list of current_assets.
424 +
425 + Args:
426 + current_assets (list): A list of dictionaries representing current assets.
427 +
428 + Returns:
429 + list: A list of dictionaries with duplicate assets removed.
430 + """
431 + current_assets = list({d["asset_name"]: d for d in current_assets}.values())
432 + current_assets_str = [json.dumps(d, sort_keys=True) for d in current_assets]
433 + current_assets_str = list(set(current_assets_str))
434 + current_assets = [json.loads(s) for s in current_assets_str]
435 + return current_assets
436 +
437 +
438 +async def analyze_suricata_alerts(
439 + monitoring_alerts: MonitoringAlerts,
440 + customer_meta: CustomersMeta,
441 + session: AsyncSession,
442 +) -> WazuhAnalysisResponse:
443 + """
444 + Analyze the given Wazuh alerts and create an alert if necessary. Otherwise update the existing alert with the asset.
445 +
446 + 1. For each alert, extract the metadata from the Wazuh-Indexer.
447 + 2. Check if the alert exists in IRIS. If it does, update the alert with the asset. If it does not, create the alert in IRIS.
448 + The alert will contain the asset and IoC if available.
449 + 3. Get the current list of assets from the alert to avoid overwriting them.
450 +
451 + Args:
452 + monitoring_alerts (MonitoringAlerts): The monitoring alert details.
453 + session (AsyncSession): The database session.
454 +
455 + Returns:
456 + WazuhAnalysisResponse: The analysis response.
457 + """
458 + logger.info(f"Analyzing Wazuh alerts with customer_meta: {customer_meta}")
459 + for alert in monitoring_alerts:
460 + alert_details = await fetch_alert_details(alert)
461 + iris_alert_id = await check_if_open_alert_exists_in_iris(alert_details)
462 + if iris_alert_id == []:
463 + logger.info(f"Alert {alert_details._id} does not exist in IRIS. Creating alert.")
464 + iris_alert_id = await create_and_update_alert_in_iris(alert_details, session)
465 + return None
466 + await remove_alert_id(alert.alert_id, session)
467 + es_client = await create_wazuh_indexer_client("Wazuh-Indexer")
468 + await add_alert_to_document(
469 + es_client=es_client,
470 + alert=AddAlertRequest(alert_id=alert_details._id, index_name=alert_details._index),
471 + soc_alert_id=iris_alert_id,
472 + session=session,
473 + )
474 +
475 + else:
476 + logger.info(f"Alert {iris_alert_id} exists in IRIS. Updating alert with the asset.")
477 + # Fetch the current list of assets from the alert to avoid overwriting them
478 + client, alert_client = await initialize_client_and_alert("DFIR-IRIS")
479 + current_assets = await get_current_assets(client, alert_client, iris_alert_id)
480 + alert_details = await create_alert_details(alert_details)
481 + agent_details = await get_agent(alert_details.agent_id, session)
482 + asset_payload = await build_asset_payload(agent_data=agent_details, alert_details=alert_details, session=session)
483 + current_assets.append(dict(IrisAsset(**asset_payload.to_dict())))
484 + current_assets = await remove_duplicate_assets(current_assets)
485 + await update_alert_with_assets(client, alert_client, iris_alert_id, current_assets)
486 + await remove_alert_id(alert.alert_id, session)
487 + es_client = await create_wazuh_indexer_client("Wazuh-Indexer")
488 + await add_alert_to_document(
489 + es_client=es_client,
490 + alert=AddAlertRequest(alert_id=alert_details.id, index_name=alert_details.index),
491 + soc_alert_id=iris_alert_id,
492 + session=session,
493 + )
494 +
495 + return WazuhAnalysisResponse(
496 + success=True,
497 + message="Wazuh alerts analyzed successfully",
498 + )
backend/app/schedulers/scheduler.py
+2
@@ -9,6 +9,7 @@ from app.schedulers.models.scheduler import JobMetadata
9 from app.schedulers.services.agent_sync import agent_sync
10 from app.schedulers.services.invoke_mimecast import invoke_mimecast_integration
11 from app.schedulers.services.invoke_mimecast import invoke_mimecast_integration_ttp
12 +from app.schedulers.services.monitoring_alert import invoke_suricata_monitoring_alert
13 from app.schedulers.services.monitoring_alert import invoke_wazuh_monitoring_alert
14
15
@@ -80,6 +81,7 @@ def get_function_by_name(function_name: str):
81 "invoke_mimecast_integration": invoke_mimecast_integration,
82 "invoke_mimecast_integration_ttp": invoke_mimecast_integration_ttp,
83 "invoke_wazuh_monitoring_alert": invoke_wazuh_monitoring_alert,
84 + "invoke_suricata_monitoring_alert": invoke_suricata_monitoring_alert,
85 # Add other function mappings here
86 }
87 return function_map.get(function_name, lambda: ValueError(f"Function {function_name} not found"))
backend/app/schedulers/services/monitoring_alert.py
+38
@@ -7,6 +7,9 @@ from sqlalchemy import select
7 from app.db.db_session import get_db_session
8 from app.db.db_session import get_sync_db_session
9 from app.db.universal_models import CustomersMeta
10 +from app.integrations.monitoring_alert.routes.monitoring_alert import (
11 + run_suricata_analysis,
12 +)
13 from app.integrations.monitoring_alert.routes.monitoring_alert import run_wazuh_analysis
14 from app.integrations.monitoring_alert.schema.monitoring_alert import (
15 MonitoringWazuhAlertsRequestModel,
@@ -52,3 +55,38 @@ async def invoke_wazuh_monitoring_alert() -> WazuhAnalysisResponse:
55 logger.error("JobMetadata for 'invoke_wazuh_monitoring_alert' not found.")
56
57 return WazuhAnalysisResponse(success=True, message="Wazuh monitoring alerts invoked.")
58 +
59 +
60 +async def invoke_suricata_monitoring_alert() -> WazuhAnalysisResponse:
61 + """
62 + Invokes the Suricata monitoring alerts scheduled job.
63 +
64 + Returns:
65 + WazuhAnalysisResponse: The response indicating the success of invoking the monitoring alerts.
66 + """
67 + logger.info("Invoking Suricata monitoring alerts scheduled job.")
68 + customer_codes = []
69 + async with get_db_session() as session:
70 + stmt = select(CustomersMeta)
71 + result = await session.execute(stmt)
72 + customer_codes = [row.customer_code for row in result.scalars()]
73 + logger.info(f"customer_codes: {customer_codes}")
74 + for customer_code in customer_codes:
75 + await run_suricata_analysis(
76 + MonitoringWazuhAlertsRequestModel(customer_code=customer_code),
77 + session,
78 + )
79 + # Close the session
80 + await session.close()
81 + with get_sync_db_session() as session:
82 + # Synchronous ORM operations
83 + job_metadata = session.query(JobMetadata).filter_by(job_id="invoke_suricata_monitoring_alert").one_or_none()
84 + if job_metadata:
85 + job_metadata.last_success = datetime.utcnow()
86 + session.add(job_metadata)
87 + session.commit()
88 + else:
89 + # Handle the case where job_metadata does not exist
90 + logger.error("JobMetadata for 'invoke_suricata_monitoring_alert' not found.")
91 +
92 + return WazuhAnalysisResponse(success=True, message="Suricata monitoring alerts invoked.")