added timerange and size to /alerts route (#57)
taylor_socfortress committed
Jul 19, 2023 at 15:49 UTC
0ba46b41c758aff2582bef696430e21b64bbd7de
2 files changed
+57
-10
backend/app/routes/alerts.py
+3
-1
@@ -20,8 +20,10 @@ def get_alerts() -> jsonify:
20
jsonify: A JSON response containing a list of alerts. Each item in the list is a dictionary representing an alert,
21
containing all its associated data.
22
"""
23
+ size = request.args.get("size", default=10, type=int)
24
+ timerange = request.args.get("timerange", default="24h", type=str)
25
service = AlertsService()
24
- alerts = service.collect_alerts(size=10) # replace `collect_all_alerts` with `collect_alerts(size=1000)`
26
+ alerts = service.collect_alerts(size=size, timerange=timerange) # replace `collect_all_alerts` with `collect_alerts(size=1000)`
27
return jsonify(alerts)
28
29
backend/app/services/WazuhIndexer/alerts.py
+54
-9
@@ -1,3 +1,5 @@
1
+from datetime import datetime
2
+from datetime import timedelta
3
from typing import Any
4
from typing import Dict
5
@@ -70,9 +72,16 @@ class AlertsService:
72
73
return {"success": True, "indices": valid_indices}
74
73
- def collect_alerts(self, size: int) -> Dict[str, object]:
75
+ def collect_alerts(self, size: int, timerange: str) -> Dict[str, object]:
76
"""
77
Collects alerts from the Wazuh-Indexer.
78
+
79
+ Args:
80
+ size (int): The maximum number of alerts to return.
81
+ timerange (str): The time range to collect alerts from. This is a string like "24h", "1w", etc.
82
+
83
+ Returns:
84
+ Dict[str, object]: A dictionary containing success status and alerts or an error message.
85
"""
86
indices_validation = self._collect_indices_and_validate()
87
if not indices_validation["success"]:
@@ -80,7 +89,7 @@ class AlertsService:
89
90
alerts_summary = []
91
for index_name in indices_validation["indices"]:
83
- alerts = self._collect_alerts(index_name, size=size)
92
+ alerts = self._collect_alerts(index_name, size=size, timerange=timerange)
93
if alerts["success"] and len(alerts["alerts"]) > 0:
94
alerts_summary.append(
95
{
@@ -90,7 +99,7 @@ class AlertsService:
99
},
100
)
101
return {
93
- "message": f"Successfully collected top {size} alerts",
102
+ "message": f"Successfully collected top {size} alerts from the last {timerange}",
103
"success": True,
104
"alerts_summary": alerts_summary,
105
}
@@ -252,7 +261,7 @@ class AlertsService:
261
"""
262
return {"message": message, "success": False}
263
255
- def _collect_alerts(self, index_name: str, size: int = None) -> Dict[str, object]:
264
+ def _collect_alerts(self, index_name: str, size: int = None, timerange: str = "24h") -> Dict[str, object]:
265
"""
266
Elasticsearch query to get the most recent alerts where the `rule_level` is 12 or higher or the
267
`syslog_level` field is `ALERT` and return the results in descending order by the `timestamp_utc` field.
@@ -261,12 +270,13 @@ class AlertsService:
270
Args:
271
index_name (str): The name of the index to query.
272
size (int, optional): The maximum number of alerts to return. If None, all alerts are returned.
273
+ timerange (str, optional): The time range to collect alerts from. This is a string like "24h", "1w", etc.
274
275
Returns:
276
Dict[str, object]: A dictionary containing success status and alerts or an error message.
277
"""
278
logger.info(f"Collecting alerts from {index_name}")
269
- query = self._build_query() # Use the provided query
279
+ query = self._build_query(timerange=timerange) # Use the provided query
280
try:
281
alerts = self.es.search(index=index_name, body=query, size=size)
282
alerts_list = [alert for alert in alerts["hits"]["hits"]]
@@ -305,20 +315,55 @@ class AlertsService:
315
return {"message": "Failed to collect alert", "success": False}
316
317
@staticmethod
308
- def _build_query() -> Dict[str, object]:
318
+ def _get_time_range_start(timerange: str) -> str:
319
+ """
320
+ Determines the start time of the time range based on the current time and the provided timerange.
321
+
322
+ Args:
323
+ timerange (str): The time range to collect alerts from. This is a string like "24h", "1w", etc.
324
+
325
+ Returns:
326
+ str: A string representing the start time of the time range in ISO format.
327
+ """
328
+ if timerange.endswith("h"):
329
+ delta = timedelta(hours=int(timerange[:-1]))
330
+ elif timerange.endswith("d"):
331
+ delta = timedelta(days=int(timerange[:-1]))
332
+ elif timerange.endswith("w"):
333
+ delta = timedelta(weeks=int(timerange[:-1]))
334
+ else:
335
+ raise ValueError("Invalid timerange format. Expected a string like '24h', '1d', '1w', etc.")
336
+
337
+ start = datetime.utcnow() - delta
338
+ return start.isoformat() + "Z" # Elasticsearch expects the time in ISO format with a Z at the end
339
+
340
+ @staticmethod
341
+ def _build_query(timerange: str) -> Dict[str, object]:
342
"""
343
Builds the Elasticsearch query to get the most recent alerts where the `rule_level` is 12 or higher or
344
the `syslog_level` field is `ALERT`.
345
346
+ Args:
347
+ timerange (str): The time range to collect alerts from. This is a string like "24h", "1w", etc.
348
+
349
Returns:
350
Dict[str, object]: A dictionary representing the Elasticsearch query.
351
"""
352
+ start = AlertsService._get_time_range_start(timerange)
353
+
354
return {
355
"query": {
356
"bool": {
319
- "should": [
320
- {"range": {"rule_level": {"gte": 12}}},
321
- {"match": {"syslog_level": "ALERT"}},
357
+ "must": [
358
+ {"range": {"timestamp_utc": {"gte": start, "lte": "now"}}},
359
+ {
360
+ "bool": {
361
+ "should": [
362
+ {"range": {"rule_level": {"gte": 12}}},
363
+ {"match": {"syslog_level": "ALERT"}},
364
+ ],
365
+ },
366
+ },
367
],
368
},
369
},