alerts last 24 hours (#56)
taylor_socfortress committed
Jul 19, 2023 at 14:28 UTC
a3f79df728d8dcb12ca14949b8f49db6b5b39d66
2 files changed
+75
-2
backend/app/routes/alerts.py
+18
@@ -43,6 +43,24 @@ def get_alerts_by_agent(agent_name: str) -> jsonify:
43
return jsonify(alerts)
44
45
46
+@bp.route("/alerts/last_24_hours", methods=["GET"])
47
+def get_alerts_last_24_hours() -> jsonify:
48
+ """
49
+ Retrieves all alerts from all wazuh indices with the last 24 hours.
50
+
51
+ This endpoint retrieves all available alerts from the AlertsService. It does this by creating an instance of
52
+ the AlertsService class and calling its `collect_alerts_by_agent` method. The result is a list of all alerts currently
53
+ available.
54
+
55
+ Returns:
56
+ jsonify: A JSON response containing a list of alerts. Each item in the list is a dictionary representing an alert,
57
+ containing all its associated data.
58
+ """
59
+ service = AlertsService()
60
+ alerts = service.collect_alerts_last_24_hours(size=1000)
61
+ return jsonify(alerts)
62
+
63
+
64
@bp.route("/alerts/top_10", methods=["GET"])
65
def get_top_10_alerts() -> jsonify:
66
"""
backend/app/services/WazuhIndexer/alerts.py
+57
-2
@@ -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
@@ -95,6 +97,31 @@ class AlertsService:
97
"alerts_summary": alerts_summary,
98
}
99
100
+ def collect_alerts_last_24_hours(self, size: int) -> Dict[str, object]:
101
+ """
102
+ Collects alerts from the last 24 hours from the Wazuh-Indexer.
103
+ """
104
+ indices_validation = self._collect_indices_and_validate()
105
+ if not indices_validation["success"]:
106
+ return indices_validation
107
+
108
+ alerts_summary = []
109
+ for index_name in indices_validation["indices"]:
110
+ alerts = self._collect_alerts(index_name, size=size, query=self._build_query_last_24_hours())
111
+ if alerts["success"] and len(alerts["alerts"]) > 0:
112
+ alerts_summary.append(
113
+ {
114
+ "index_name": index_name,
115
+ "total_alerts": len(alerts["alerts"]),
116
+ "alerts": alerts["alerts"],
117
+ },
118
+ )
119
+ return {
120
+ "message": f"Successfully collected top {size} alerts from the last 24 hours",
121
+ "success": True,
122
+ "alerts_summary": alerts_summary,
123
+ }
124
+
125
def collect_alerts_by_index(self, index_name: str, size: int) -> Dict[str, Any]:
126
"""
127
Collects alerts from the given index.
@@ -227,7 +254,7 @@ class AlertsService:
254
"""
255
return {"message": message, "success": False}
256
230
- def _collect_alerts(self, index_name: str, size: int = None) -> Dict[str, object]:
257
+ def _collect_alerts(self, index_name: str, size: int = None, query: Dict[str, object] = None) -> Dict[str, object]:
258
"""
259
Elasticsearch query to get the most recent alerts where the `rule_level` is 12 or higher or the
260
`syslog_level` field is `ALERT` and return the results in descending order by the `timestamp_utc` field.
@@ -236,12 +263,13 @@ class AlertsService:
263
Args:
264
index_name (str): The name of the index to query.
265
size (int, optional): The maximum number of alerts to return. If None, all alerts are returned.
266
+ query (Dict[str, object], optional): The Elasticsearch query to use. If None, the default query is used.
267
268
Returns:
269
Dict[str, object]: A dictionary containing success status and alerts or an error message.
270
"""
271
logger.info(f"Collecting alerts from {index_name}")
244
- query = self._build_query()
272
+ query = query or self._build_query() # Use the provided query or the default query
273
try:
274
alerts = self.es.search(index=index_name, body=query, size=size)
275
alerts_list = [alert for alert in alerts["hits"]["hits"]]
@@ -300,6 +328,33 @@ class AlertsService:
328
"sort": [{"timestamp_utc": {"order": "desc"}}],
329
}
330
331
+ def _build_query_last_24_hours(self) -> Dict[str, object]:
332
+ """
333
+ Builds the Elasticsearch query to get the most recent alerts where the `rule_level` is 12 or higher or
334
+ the `syslog_level` field is `ALERT`, and the `timestamp_utc` is within the last 24 hours.
335
+
336
+ Returns:
337
+ Dict[str, object]: A dictionary representing the Elasticsearch query.
338
+ """
339
+ # Calculate the time 24 hours ago
340
+ time_24_hours_ago = datetime.utcnow() - timedelta(hours=24)
341
+
342
+ # Convert the time to the format used in the Elasticsearch index
343
+ time_24_hours_ago = time_24_hours_ago.strftime("%Y-%m-%dT%H:%M:%S")
344
+
345
+ return {
346
+ "query": {
347
+ "bool": {
348
+ "must": [{"range": {"timestamp_utc": {"gte": time_24_hours_ago}}}],
349
+ "should": [
350
+ {"range": {"rule_level": {"gte": 12}}},
351
+ {"match": {"syslog_level": "ALERT"}},
352
+ ],
353
+ },
354
+ },
355
+ "sort": [{"timestamp_utc": {"order": "desc"}}],
356
+ }
357
+
358
def escalate_alert(self, alert_id: str, index: str) -> Dict[str, Any]:
359
"""
360
Escalates an alert by creating it in DFIR-IRIS