Create universal.py
taylor_socfortress committed
Jul 10, 2023 at 16:45 UTC
dfc03fb83e425248a436f903e6c719b2f22358f0
1 file changed
+99
backend/app/services/WazuhIndexer/universal.py
new
+99
@@ -0,0 +1,99 @@
1
+from app.models.agents import (
2
+ AgentMetadata,
3
+ agent_metadata_schema,
4
+ agent_metadatas_schema,
5
+)
6
+from typing import Dict, List
7
+from app import db
8
+from datetime import datetime
9
+import requests
10
+from loguru import logger
11
+from elasticsearch7 import Elasticsearch
12
+from app.models.connectors import connector_factory, Connector
13
+
14
+
15
+class UniversalService:
16
+ """
17
+ A service class that encapsulates the logic for polling messages from the Wazuh-Indexer.
18
+ """
19
+
20
+ def __init__(self) -> None:
21
+ self.collect_wazuhindexer_details("Wazuh-Indexer")
22
+ (
23
+ self.connector_url,
24
+ self.connector_username,
25
+ self.connector_password,
26
+ ) = self.collect_wazuhindexer_details("Wazuh-Indexer")
27
+ self.es = Elasticsearch(
28
+ [self.connector_url],
29
+ http_auth=(self.connector_username, self.connector_password),
30
+ verify_certs=False,
31
+ timeout=15,
32
+ max_retries=10,
33
+ retry_on_timeout=False,
34
+ )
35
+
36
+ def collect_wazuhindexer_details(self, connector_name: str):
37
+ """
38
+ Collects the details of the Wazuh-Indexer connector.
39
+
40
+ Args:
41
+ connector_name (str): The name of the Wazuh-Indexer connector.
42
+
43
+ Returns:
44
+ tuple: A tuple containing the connection URL, username, and password.
45
+ """
46
+ connector_instance = connector_factory.create(connector_name, connector_name)
47
+ connection_successful = connector_instance.verify_connection()
48
+ if connection_successful:
49
+ connection_details = Connector.get_connector_info_from_db(connector_name)
50
+ return (
51
+ connection_details.get("connector_url"),
52
+ connection_details.get("connector_username"),
53
+ connection_details.get("connector_password"),
54
+ )
55
+ else:
56
+ return None, None, None
57
+
58
+ def collect_indices(self):
59
+ """
60
+ Collects the indices from the Wazuh-Indexer.
61
+
62
+ Returns:
63
+ list: A list containing the indices.
64
+ """
65
+ if (
66
+ self.connector_url is None
67
+ or self.connector_username is None
68
+ or self.connector_password is None
69
+ ):
70
+ return {
71
+ "message": "Failed to collect Wazuh-Indexer details",
72
+ "success": False,
73
+ }
74
+
75
+ indices = self._collect_indices()
76
+
77
+ if indices["success"]:
78
+ return indices
79
+
80
+ return {"message": "Failed to collect indices", "success": False}
81
+
82
+ def _collect_indices(self) -> Dict[str, object]:
83
+ """
84
+ Wazuh-Indexer query to retrievce all indices.
85
+
86
+ Returns:
87
+ Dict[str, object]: _description_
88
+ """
89
+ try:
90
+ indices_dict = self.es.indices.get_alias("*")
91
+ indices_list = list(indices_dict.keys())
92
+ return {
93
+ "message": "Successfully collected indices",
94
+ "success": True,
95
+ "indices_list": indices_list,
96
+ }
97
+ except Exception as e:
98
+ logger.error(f"Failed to collect indices: {e}")
99
+ return {"message": "Failed to collect indices", "success": False}