Create index.py
taylor_socfortress committed
Jul 10, 2023 at 16:43 UTC
39cf99008c6457f3e7b936408a008512c38e09a0
1 file changed
+148
backend/app/services/Graylog/index.py
new
+148
@@ -0,0 +1,148 @@
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 app.models.connectors import connector_factory, Connector, GraylogConnector
12
+from app.services.Graylog.universal import UniversalService
13
+
14
+
15
+class IndexService:
16
+ """
17
+ A service class that encapsulates the logic for pulling index data from Graylog
18
+ """
19
+
20
+ HEADERS: Dict[str, str] = {"X-Requested-By": "CoPilot"}
21
+
22
+ def __init__(self):
23
+ (
24
+ self.connector_url,
25
+ self.connector_username,
26
+ self.connector_password,
27
+ ) = UniversalService().collect_graylog_details("Graylog")
28
+
29
+ def collect_indices(self):
30
+ """
31
+ Collects the indices that are managed by Graylog.
32
+
33
+ Returns:
34
+ list: A list containing the indices.
35
+ """
36
+ if (
37
+ self.connector_url is None
38
+ or self.connector_username is None
39
+ or self.connector_password is None
40
+ ):
41
+ return {"message": "Failed to collect Graylog details", "success": False}
42
+
43
+ managed_indices = self._collect_managed_indices()
44
+
45
+ if managed_indices["success"]:
46
+ index_names = self._extract_index_names(managed_indices)
47
+ managed_indices["index_names"] = index_names
48
+
49
+ return managed_indices
50
+
51
+ def _collect_managed_indices(self) -> Dict[str, object]:
52
+ """
53
+ Collects the indices that are managed by Graylog.
54
+
55
+ Returns:
56
+ dict: A dictionary containing the success status, a message and potentially the indices.
57
+ """
58
+ try:
59
+ managed_indices = requests.get(
60
+ f"{self.connector_url}/api/system/indexer/indices",
61
+ headers=self.HEADERS,
62
+ auth=(self.connector_username, self.connector_password),
63
+ verify=False,
64
+ )
65
+ return {
66
+ "message": "Successfully collected managed indices",
67
+ "success": True,
68
+ "indices": managed_indices.json()["all"]["indices"],
69
+ }
70
+ except Exception as e:
71
+ logger.error(f"Failed to collect managed indices: {e}")
72
+ return {"message": "Failed to collect managed indices", "success": False}
73
+
74
+ def _extract_index_names(self, response: Dict[str, object]) -> List[str]:
75
+ """
76
+ Extracts index names from the provided response.
77
+
78
+ Args:
79
+ response (dict): The dictionary containing the response.
80
+
81
+ Returns:
82
+ list: A list containing the index names.
83
+ """
84
+ index_names = list(response.get("indices", {}).keys())
85
+ return index_names
86
+
87
+ def delete_index(self, index_name: str) -> Dict[str, object]:
88
+ """
89
+ Deletes the specified index from Graylog.
90
+
91
+ Args:
92
+ index_name (str): The name of the index to delete.
93
+
94
+ Returns:
95
+ dict: A dictionary containing the response.
96
+ """
97
+ logger.info(f"Deleting index {index_name} from Graylog")
98
+ if (
99
+ self.connector_url is None
100
+ or self.connector_username is None
101
+ or self.connector_password is None
102
+ ):
103
+ return {"message": "Failed to collect Graylog details", "success": False}
104
+
105
+ # Check if the index exists in Graylog
106
+ managed_indices = self._collect_managed_indices()
107
+ if managed_indices["success"]:
108
+ index_names = self._extract_index_names(managed_indices)
109
+ if index_name not in index_names:
110
+ return {
111
+ "message": f"Index {index_name} is not managed by Graylog",
112
+ "success": False,
113
+ }
114
+ # Invoke _delete_index
115
+ return self._delete_index(index_name)
116
+
117
+ return {
118
+ "message": f"Failed to delete index {index_name} from Graylog",
119
+ "success": False,
120
+ }
121
+
122
+ def _delete_index(self, index_name: str) -> Dict[str, object]:
123
+ """
124
+ Deletes the specified index from Graylog.
125
+
126
+ Args:
127
+ index_name (str): The name of the index to delete.
128
+
129
+ Returns:
130
+ dict: A dictionary containing the response.
131
+ """
132
+ try:
133
+ delete_index_response = requests.delete(
134
+ f"{self.connector_url}/api/system/indexer/indices/{index_name}",
135
+ headers=self.HEADERS,
136
+ auth=(self.connector_username, self.connector_password),
137
+ verify=False,
138
+ )
139
+ return {
140
+ "message": f"Successfully deleted index {index_name} from Graylog",
141
+ "success": True,
142
+ }
143
+ except Exception as e:
144
+ logger.error(f"Failed to delete index {index_name} from Graylog: {e}")
145
+ return {
146
+ "message": f"Failed to delete index {index_name} from Graylog. If this is the current index, it cannot be deleted.",
147
+ "success": False,
148
+ }