invoke asksocfortress for alerts (#46)
* invoke asksocfortress for alerts * precommit
taylor_socfortress committed
Jul 17, 2023 at 20:55 UTC
80d5902ddfd029a78a3c6da22eeaa188dbac47bf
2 files changed
+123
-2
backend/app/services/WazuhIndexer/alerts.py
+13
-2
@@ -4,6 +4,7 @@ from typing import Dict
4
from elasticsearch7 import Elasticsearch
5
from loguru import logger
6
7
+from app.services.ask_socfortress.univerval import AskSocfortressService
8
from app.services.WazuhIndexer.universal import UniversalService
9
10
@@ -35,6 +36,11 @@ class AlertsService:
36
max_retries=10,
37
retry_on_timeout=False,
38
)
39
+ self.asksocfortress_service = AskSocfortressService("AskSocfortress")
40
+ (
41
+ self.connector_url,
42
+ self.connector_api_key,
43
+ ) = self.asksocfortress_service.collect_asksocfortress_details("AskSocfortress")
44
45
def is_index_skipped(self, index_name: str) -> bool:
46
"""
@@ -82,7 +88,6 @@ class AlertsService:
88
"alerts": alerts["alerts"],
89
},
90
)
85
-
91
return {
92
"message": f"Successfully collected top {size} alerts",
93
"success": True,
@@ -209,10 +214,16 @@ class AlertsService:
214
try:
215
alerts = self.es.search(index=index_name, body=query, size=size)
216
alerts_list = [alert for alert in alerts["hits"]["hits"]]
217
+
218
+ # Iterate over each alert and invoke invoke_socfortress function
219
+ for alert in alerts_list:
220
+ ask_socfortress = self.asksocfortress_service.invoke_asksocfortress(alert["_source"]["rule_description"])
221
+ alert["ask_socfortress"] = ask_socfortress # Add the result to the alert
222
+
223
return {
224
"message": "Successfully collected alerts",
225
"success": True,
215
- "alerts": alerts_list,
226
+ "alerts": alerts_list, # Return the alerts list with the added results
227
}
228
except Exception as e:
229
logger.error(f"Failed to collect alerts: {e}")
backend/app/services/ask_socfortress/univerval.py
new
+110
@@ -0,0 +1,110 @@
1
+import json
2
+from typing import Any
3
+from typing import Dict
4
+from typing import Optional
5
+from typing import Tuple
6
+
7
+import requests
8
+from loguru import logger
9
+
10
+from app.models.connectors import Connector
11
+from app.models.connectors import connector_factory
12
+
13
+
14
+class AskSocfortressService:
15
+ """
16
+ A service class that encapsulates the logic for interfacing with ASK SOCFortress. This class handles tasks like retrieving connector
17
+ details, and invoking the ask_socfortress connector.
18
+ """
19
+
20
+ def __init__(self, connector_name: str) -> None:
21
+ """
22
+ Initializes the UniversalService by collecting AskSOCFortress details associated with the specified connector name.
23
+
24
+ Args:
25
+ connector_name (str): The name of the AskSOCFortress connector.
26
+ """
27
+ self.connector_url, self.connector_api_key = self.collect_asksocfortress_details(
28
+ connector_name,
29
+ )
30
+
31
+ def collect_asksocfortress_details(
32
+ self,
33
+ connector_name: str,
34
+ ) -> Tuple[Optional[str], Optional[str]]:
35
+ """
36
+ Collects the details of the DFIR-IRIS connector.
37
+
38
+ Args:
39
+ connector_name (str): The name of the DFIR-IRIS connector.
40
+
41
+ Returns:
42
+ tuple: A tuple containing the connection URL and API key. If the connection is not successful, both elements of the tuple are
43
+ None.
44
+ """
45
+ connector_instance = connector_factory.create(connector_name, connector_name)
46
+ connection_successful = connector_instance.verify_connection()
47
+ if connection_successful:
48
+ connection_details = Connector.get_connector_info_from_db(connector_name)
49
+ return (
50
+ connection_details.get("connector_url"),
51
+ connection_details.get("connector_api_key"),
52
+ )
53
+ else:
54
+ return None, None
55
+
56
+ def invoke_asksocfortress(self, data: str) -> Dict[str, Any]:
57
+ """
58
+ Invoke ASKSOCFortress API to enrich data via a POST request.
59
+
60
+ Attributes:
61
+ data (str): The data to be enriched.
62
+
63
+ Returns:
64
+ dict: A dictionary containing a success key indicating the success or failure of the connection
65
+ and a message key containing further information about the connection result.
66
+ """
67
+ headers = {
68
+ "Content-Type": "application/json",
69
+ "x-api-key": self.connector_api_key,
70
+ "module-version": "1.0",
71
+ }
72
+ logger.info(f"Invoking AskSOCFortress API with data: {data}")
73
+
74
+ payload = {"rule_description": data}
75
+
76
+ timeout = 120
77
+
78
+ try:
79
+ response = requests.post(
80
+ self.connector_url,
81
+ data=json.dumps(payload),
82
+ headers=headers,
83
+ timeout=timeout,
84
+ )
85
+ response.raise_for_status()
86
+ try:
87
+ response_data = response.json()
88
+ except ValueError:
89
+ logger.error(f"Unable to decode response from AskSOCFortress API: {response.text}")
90
+ raise
91
+ else:
92
+ return {
93
+ "success": True,
94
+ "response": response_data["message"],
95
+ "message": "Successfully invoked AskSOCFortress API",
96
+ }
97
+ except requests.exceptions.HTTPError as e:
98
+ logger.error(f"Unable to invoke AskSOCFortress API: {e}")
99
+ return {
100
+ "success": False,
101
+ "response": None,
102
+ "message": f"Unable to invoke AskSOCFortress API: {e}",
103
+ }
104
+ except Exception as e:
105
+ logger.error(f"Unable to invoke AskSOCFortress API: {e}")
106
+ return {
107
+ "success": False,
108
+ "response": None,
109
+ "message": f"Unable to invoke AskSOCFortress API: {e}",
110
+ }