Create alerts.py
taylor_socfortress committed
Jul 10, 2023 at 16:45 UTC
f80d2df1a6f26e77e4c3e7f9e7447ad6adc53188
1 file changed
+120
backend/app/services/WazuhIndexer/alerts.py
new
+120
@@ -0,0 +1,120 @@
1
+from elasticsearch7 import Elasticsearch
2
+from loguru import logger
3
+from typing import Dict, List
4
+
5
+from app.services.WazuhIndexer.universal import UniversalService
6
+from app.services.WazuhIndexer.index import IndexService
7
+
8
+
9
+class AlertsService:
10
+ """
11
+ A service class that encapsulates the logic for pulling alerts from the Wazuh-Indexer.
12
+ """
13
+
14
+ SKIP_INDEX_NAMES: Dict[str, bool] = {
15
+ "wazuh-statistics": True,
16
+ "wazuh-monitoring": True,
17
+ }
18
+
19
+ def __init__(self):
20
+ (
21
+ self.connector_url,
22
+ self.connector_username,
23
+ self.connector_password,
24
+ ) = UniversalService().collect_wazuhindexer_details("Wazuh-Indexer")
25
+ self.es = Elasticsearch(
26
+ [self.connector_url],
27
+ http_auth=(self.connector_username, self.connector_password),
28
+ verify_certs=False,
29
+ timeout=15,
30
+ max_retries=10,
31
+ retry_on_timeout=False,
32
+ )
33
+
34
+ def is_index_skipped(self, index_name: str) -> bool:
35
+ """Check if the index should be skipped."""
36
+ for skipped in self.SKIP_INDEX_NAMES:
37
+ if index_name.startswith(skipped):
38
+ return True
39
+ return False
40
+
41
+ def collect_alerts(self) -> Dict[str, object]:
42
+ """
43
+ Collects the alerts from the Wazuh-Indexer where the index name starts with "wazuh_" and is not in the SKIP_INDEX_NAMES list.
44
+ Returns the 10 previous alerts based on the `timestamp_utc` field.
45
+
46
+ Returns:
47
+ Dict[str, object]: A dictionary containing success status and alerts or an error message.
48
+ """
49
+ if not all(
50
+ [self.connector_url, self.connector_username, self.connector_password]
51
+ ):
52
+ return {
53
+ "message": "Failed to collect Wazuh-Indexer details",
54
+ "success": False,
55
+ }
56
+
57
+ indices_list = UniversalService().collect_indices()
58
+ if not indices_list["success"]:
59
+ return {"message": "Failed to collect indices", "success": False}
60
+
61
+ alerts_summary = []
62
+ for index_name in indices_list["indices_list"]:
63
+ if not index_name.startswith("wazuh_") or self.is_index_skipped(index_name):
64
+ continue
65
+
66
+ alerts = self._collect_alerts(index_name)
67
+ if alerts["success"] and len(alerts["alerts"]) > 0:
68
+ alerts_summary.append(
69
+ {
70
+ "index_name": index_name,
71
+ "total_alerts": len(alerts["alerts"]),
72
+ "last_10_alerts": alerts["alerts"],
73
+ }
74
+ )
75
+
76
+ return {
77
+ "message": "Successfully collected alerts",
78
+ "success": True,
79
+ "alerts_summary": alerts_summary,
80
+ }
81
+
82
+ def _collect_alerts(self, index_name: str) -> Dict[str, object]:
83
+ """
84
+ Elasticsearch query to get the 10 most recent alerts where the `rule_level` is 12 or higher or the
85
+ `syslog_level` field is `ALERT` and return the results in descending order by the `timestamp_utc` field.
86
+
87
+ Args:
88
+ index_name (str): The name of the index to query.
89
+
90
+ Returns:
91
+ Dict[str, object]: A dictionary containing success status and alerts or an error message.
92
+ """
93
+ logger.info(f"Collecting alerts from {index_name}")
94
+ query = self._build_query()
95
+ try:
96
+ alerts = self.es.search(index=index_name, body=query, size=10)
97
+ alerts_list = [alert for alert in alerts["hits"]["hits"]]
98
+ return {
99
+ "message": "Successfully collected alerts",
100
+ "success": True,
101
+ "alerts": alerts_list,
102
+ }
103
+ except Exception as e:
104
+ logger.error(f"Failed to collect alerts: {e}")
105
+ return {"message": "Failed to collect alerts", "success": False}
106
+
107
+ @staticmethod
108
+ def _build_query() -> Dict[str, object]:
109
+ """Builds and returns the query."""
110
+ return {
111
+ "query": {
112
+ "bool": {
113
+ "should": [
114
+ {"range": {"rule_level": {"gte": 12}}},
115
+ {"match": {"syslog_level": "ALERT"}},
116
+ ]
117
+ }
118
+ },
119
+ "sort": [{"timestamp_utc": {"order": "desc"}}],
120
+ }