escalate alert to create alert in dfir-iris (#51)
* escalate alert to create alert in dfir-iris * precommit fixes
taylor_socfortress committed
Jul 18, 2023 at 14:24 UTC
304d8ac13dd20758f2b4bb2cc6e9157e5281a8a9
7 files changed
+649
-26
backend/app/routes/alerts.py
+18
@@ -1,5 +1,6 @@
1
from flask import Blueprint
2
from flask import jsonify
3
+from flask import request
4
5
from app.services.WazuhIndexer.alerts import AlertsService
6
@@ -112,3 +113,20 @@ def get_rules_by_host() -> jsonify:
113
service = AlertsService()
114
rules = service.collect_alerts_by_rule_per_host()
115
return jsonify(rules)
116
+
117
+
118
+@bp.route("/alerts/escalate", methods=["POST"])
119
+def escalate_alert() -> jsonify:
120
+ """
121
+ Accepts a POST request with an alert_uid and esacalates the alert by creating an alert in DFIR-IRIS
122
+
123
+ This endpoint accepts a POST request with an alert_uid and esacalates the alert by creating an alert in DFIR-IRIS.
124
+
125
+ Returns:
126
+ jsonify: A JSON response containing the status of the escalation.
127
+ """
128
+ service = AlertsService()
129
+ alert_id = request.json["alert_id"]
130
+ index = request.json["index"]
131
+ status = service.escalate_alert(alert_id=alert_id, index=index)
132
+ return jsonify(status)
backend/app/routes/dfir_iris.py
+2
-2
@@ -1,7 +1,7 @@
1
from flask import Blueprint
2
from flask import request
3
4
-from app.services.DFIR_IRIS.alerts import AlertsService
4
+from app.services.DFIR_IRIS.alerts import IRISAlertsService
5
from app.services.DFIR_IRIS.assets import AssetsService
6
from app.services.DFIR_IRIS.cases import CasesService
7
from app.services.DFIR_IRIS.notes import NotesService
@@ -100,6 +100,6 @@ def get_alerts():
100
Returns:
101
Response: A Flask Response object carrying a JSON representation of the list of alerts.
102
"""
103
- service = AlertsService()
103
+ service = IRISAlertsService()
104
alerts = service.list_alerts()
105
return alerts
backend/app/services/DFIR_IRIS/alerts.py
+252
-24
@@ -1,46 +1,52 @@
1
+# Standard library imports
2
+from typing import Any
3
from typing import Dict
4
+from typing import Set
5
3
-# import requests
6
+# Local application imports
7
from dfir_iris_client.alert import Alert
8
6
-# from dfir_iris_client.helper.utils import assert_api_resp
7
-# from dfir_iris_client.helper.utils import get_data_from_resp
9
+# Third-party library imports
10
from loguru import logger
11
12
+from app.models.agents import agent_metadata_schema
13
+from app.services.agents.agents import AgentService
14
+from app.services.DFIR_IRIS.host_enrichment import AssetTypeResolver
15
+from app.services.DFIR_IRIS.ioc_enrichment import DomainValidator
16
+from app.services.DFIR_IRIS.ioc_enrichment import HashValidator
17
+from app.services.DFIR_IRIS.ioc_enrichment import IPv4AddressValidator
18
from app.services.DFIR_IRIS.universal import UniversalService
19
20
13
-class AlertsService:
21
+class IRISAlertsService:
22
"""
15
- A service class that encapsulates the logic for pulling alerts from DFIR-IRIS. It creates a DFIR-IRIS session upon
16
- initialization and uses it to fetch alerts.
23
+ A service class that encapsulates the logic for pulling alerts from DFIR-IRIS.
24
"""
25
26
def __init__(self):
27
"""
21
- Initializes the AlertsService by creating a UniversalService object for "DFIR-IRIS" and establishing a session.
22
- If the session creation is unsuccessful, an error is logged and the iris_session attribute is set to None.
28
+ Initializes the AlertsService by creating a UniversalService object for "DFIR-IRIS".
29
"""
30
self.universal_service = UniversalService("DFIR-IRIS")
25
- session_result = self.universal_service.create_session()
31
+ self.iris_session = self._create_iris_session()
32
33
+ def _create_iris_session(self):
34
+ """
35
+ Create a session with the universal service. If the session creation fails,
36
+ log an error and return None.
37
+ """
38
+ session_result = self.universal_service.create_session()
39
if not session_result["success"]:
40
logger.error(session_result["message"])
29
- self.iris_session = None
30
- else:
31
- self.iris_session = session_result["session"]
41
+ return None
42
+ return session_result["session"]
43
44
def list_alerts(self) -> Dict[str, object]:
45
"""
35
- List all alerts from DFIR-IRIS. If the iris_session attribute is None, this indicates that the session creation
36
- was unsuccessful, and a dictionary with "success" set to False is returned. Otherwise, it attempts to fetch and
37
- parse the alerts data.
38
-
39
- Returns:
40
- dict: A dictionary containing the success status, a message, and potentially the fetched alerts. The
41
- "success" key is a boolean indicating whether the operation was successful. The "message" key is a string
42
- providing details about the operation. The "results" key, included when "success" is True, contains the
43
- fetched alerts data.
46
+ List all alerts from DFIR-IRIS.
47
+ If the iris_session attribute is None, this indicates that the session creation
48
+ was unsuccessful, and a dictionary with "success" set to False is returned.
49
+ Otherwise, it attempts to fetch and parse the alerts data.
50
"""
51
if self.iris_session is None:
52
return {
@@ -48,7 +54,6 @@ class AlertsService:
54
"message": "DFIR-IRIS session was not successfully created.",
55
}
56
51
- logger.info("Collecting cases from DFIR-IRIS")
57
alert = Alert(session=self.iris_session)
58
result = self.universal_service.fetch_and_parse_data(
59
self.iris_session,
@@ -58,11 +63,234 @@ class AlertsService:
63
if not result["success"]:
64
return {
65
"success": False,
61
- "message": "Failed to collect cases from DFIR-IRIS",
66
+ "message": "Failed to collect alerts from DFIR-IRIS",
67
}
68
69
return {
70
"success": True,
66
- "message": "Successfully collected cases from DFIR-IRIS",
71
+ "message": "Successfully collected alerts from DFIR-IRIS",
72
"results": result["data"],
73
}
74
+
75
+ def create_alert_general(self, alert_data: Dict[str, Any], alert_id: str, index: str) -> Dict[str, Any]:
76
+ """
77
+ Create an alert with the provided data.
78
+ """
79
+ # Get agent data
80
+ agent_id = alert_data["agent_id"]
81
+ service = AgentService()
82
+ agent = service.get_agent(agent_id)
83
+ agent_data = agent_metadata_schema.dump(agent)
84
+
85
+ # Use AssetTypeResolver to determine the asset type ID
86
+ asset_resolver = AssetTypeResolver(agent_data["os"])
87
+ agent_asset_type = asset_resolver.get_asset_type_id()
88
+
89
+ # Append the asset type ID to the alert data
90
+ alert_data["asset_type_id"] = agent_asset_type
91
+
92
+ # Check if IoC field exists
93
+ ioc_field_present = self.field_exists_ioc(alert_data)
94
+ if ioc_field_present["success"]:
95
+ logger.info(f"Found IoC field: {ioc_field_present}")
96
+ alert_data["ioc_value"] = ioc_field_present["field_value"]
97
+
98
+ # Define the validator classes to be used
99
+ validators = [IPv4AddressValidator, HashValidator, DomainValidator]
100
+ ioc_type = None
101
+
102
+ # Iterate over each validator class
103
+ for Validator in validators:
104
+ validator = Validator(ioc_field_present["field_value"])
105
+ result = validator.validate()
106
+
107
+ # If the validation is successful, store the ioc_type and break the loop
108
+ if result["success"]:
109
+ ioc_type = result["ioc_type"]
110
+ break
111
+
112
+ if ioc_type is not None:
113
+ alert_data["ioc_type"] = ioc_type
114
+ else:
115
+ logger.error("Failed to validate IoC value.")
116
+
117
+ alert_payload = self.create_general_ioc_payload(alert_data=alert_data, agent_data=agent_data, alert_id=alert_id, index=index)
118
+
119
+ # Create an alert
120
+ alert = Alert(session=self.iris_session)
121
+ result = self.universal_service.fetch_and_parse_data(
122
+ self.iris_session,
123
+ alert.add_alert,
124
+ alert_payload,
125
+ )
126
+
127
+ if not result["success"]:
128
+ return {
129
+ "success": False,
130
+ "message": "Failed to create alert in DFIR-IRIS",
131
+ }
132
+
133
+ return {
134
+ "success": True,
135
+ "message": "Successfully created alert in DFIR-IRIS",
136
+ "results": result["data"],
137
+ }
138
+
139
+ # Create alert payload
140
+ alert_payload = self.create_general_payload(alert_data=alert_data, agent_data=agent_data, alert_id=alert_id, index=index)
141
+
142
+ # Create an alert
143
+ alert = Alert(session=self.iris_session)
144
+ result = self.universal_service.fetch_and_parse_data(
145
+ self.iris_session,
146
+ alert.add_alert,
147
+ alert_payload,
148
+ )
149
+
150
+ if not result["success"]:
151
+ return {
152
+ "success": False,
153
+ "message": "Failed to create alert in DFIR-IRIS",
154
+ }
155
+
156
+ return {
157
+ "success": True,
158
+ "message": "Successfully created alert in DFIR-IRIS",
159
+ "results": result["data"],
160
+ }
161
+
162
+ def field_exists_ioc(self, alert_data: Dict[str, Any]) -> Dict[str, Any]:
163
+ """
164
+ Check if an IoC field exists in the alert data.
165
+ """
166
+ for field_name in self.valid_ioc_fields:
167
+ if field_name in alert_data:
168
+ return {
169
+ "success": True,
170
+ "field_name": field_name,
171
+ "field_value": alert_data[field_name],
172
+ }
173
+ return {"success": False, "field_name": None}
174
+
175
+ @property
176
+ def valid_ioc_fields(self) -> Set[str]:
177
+ """
178
+ Get the set of valid IoC fields.
179
+ """
180
+ return {"misp_value", "opencti_value", "threat_intel_value"}
181
+
182
+ def create_general_payload(self, alert_data: Dict[str, Any], agent_data: Dict[str, Any], alert_id: str, index: str) -> Dict[str, Any]:
183
+ """
184
+ Craft the general alert payload when it does not contain an IoC.
185
+ """
186
+ try:
187
+ payload = {
188
+ "alert_title": alert_data["rule_description"],
189
+ "alert_description": alert_data["rule_description"],
190
+ "alert_source": "Wazuh",
191
+ "assets": [
192
+ {
193
+ "asset_name": agent_data["hostname"],
194
+ "asset_ip": agent_data["ip_address"],
195
+ "asset_description": agent_data["os"],
196
+ "asset_type_id": alert_data["asset_type_id"],
197
+ },
198
+ ],
199
+ # "alert_source_link": f"{self.grafana_url}/explore?left=%5B%22now-6h%22,%22now%22,%22WAZUH%22,%7B%22refId%22"
200
+ # ":%22A%22,%22query%22:%22process_id:%5C%22"
201
+ # f"{alert_data['process_id']}%5C%22%20AND%20"
202
+ # f"agent_name:%5C%22{alert_data['agent_name']}%5C%22%22,%22alias%22"
203
+ # ":%22%22,%22metrics%22:%5B%7B%22id%22:%221%22,%22type%22:%22logs%22,%22settings%22:%7B%22limit%22:%22500%22"
204
+ # "%7D%7D%5D,%22bucketAggs%22:%5B%5D,%22timeField%22:%22timestamp%22%7D%5D",
205
+ "alert_status_id": 3,
206
+ "alert_severity_id": 5,
207
+ # "alert_customer_id": customer_code_details["customer_code_iris_id"],
208
+ "alert_customer_id": 1,
209
+ "alert_source_content": alert_data,
210
+ "alert_context": {
211
+ # "customer_id": f"{alert_data['alert_payload']['alert_details']['_source']['agent_labels_customer']},"
212
+ # f"{customer_code_details['customer_code_iris_index']}",
213
+ "alert_id": alert_id,
214
+ "alert_name": alert_data["rule_description"],
215
+ "alert_level": alert_data["rule_level"],
216
+ "rule_id": alert_data["rule_id"],
217
+ "asset_name": agent_data["hostname"],
218
+ "asset_ip": agent_data["ip_address"],
219
+ "asset_type": alert_data["asset_type_id"],
220
+ "process_id": alert_data["process_id"],
221
+ # If the `rule_mitre_id` field exists in the alert_details, add it to the payload
222
+ "rule_mitre_id": alert_data.get("rule_mitre_id", "n/a"),
223
+ "rule_mitre_tactic": alert_data.get("rule_mitre_tactic", "n/a"),
224
+ "rule_mitre_technique": alert_data.get("rule_mitre_technique", "n/a"),
225
+ },
226
+ "alert_note": alert_data.get("ask_socfortress", "Ask SOCFortress not enabled"),
227
+ }
228
+ return payload
229
+ except Exception as e:
230
+ logger.error(f"Error creating general alert payload: {e}")
231
+ return {"success": False, "message": f"Error creating general alert payload: {e}"}
232
+
233
+ def create_general_ioc_payload(
234
+ self,
235
+ alert_data: Dict[str, Any],
236
+ agent_data: Dict[str, Any],
237
+ alert_id: str,
238
+ index: str,
239
+ ) -> Dict[str, Any]:
240
+ """
241
+ Craft the general alert payload when it does contain an IoC.
242
+ """
243
+ try:
244
+ payload = {
245
+ "alert_title": alert_data["rule_description"],
246
+ "alert_description": alert_data["rule_description"],
247
+ "alert_source": "Wazuh",
248
+ "assets": [
249
+ {
250
+ "asset_name": agent_data["hostname"],
251
+ "asset_ip": agent_data["ip_address"],
252
+ "asset_description": agent_data["os"],
253
+ "asset_type_id": alert_data["asset_type_id"],
254
+ },
255
+ ],
256
+ # "alert_source_link": f"{self.grafana_url}/explore?left=%5B%22now-6h%22,%22now%22,%22WAZUH%22,%7B%22refId%22"
257
+ # ":%22A%22,%22query%22:%22process_id:%5C%22"
258
+ # f"{alert_data['process_id']}%5C%22%20AND%20"
259
+ # f"agent_name:%5C%22{alert_data['agent_name']}%5C%22%22,%22alias%22"
260
+ # ":%22%22,%22metrics%22:%5B%7B%22id%22:%221%22,%22type%22:%22logs%22,%22settings%22:%7B%22limit%22:%22500%22"
261
+ # "%7D%7D%5D,%22bucketAggs%22:%5B%5D,%22timeField%22:%22timestamp%22%7D%5D",
262
+ "alert_status_id": 3,
263
+ "alert_severity_id": 5,
264
+ # "alert_customer_id": customer_code_details["customer_code_iris_id"],
265
+ "alert_customer_id": 1,
266
+ "alert_source_content": alert_data,
267
+ "alert_context": {
268
+ # "customer_id": f"{alert_data['alert_payload']['alert_details']['_source']['agent_labels_customer']},"
269
+ # f"{customer_code_details['customer_code_iris_index']}",
270
+ "alert_id": alert_id,
271
+ "alert_name": alert_data["rule_description"],
272
+ "alert_level": alert_data["rule_level"],
273
+ "rule_id": alert_data["rule_id"],
274
+ "asset_name": agent_data["hostname"],
275
+ "asset_ip": agent_data["ip_address"],
276
+ "asset_type": alert_data["asset_type_id"],
277
+ "process_id": alert_data["process_id"],
278
+ # If the `rule_mitre_id` field exists in the alert_details, add it to the payload
279
+ "rule_mitre_id": alert_data.get("rule_mitre_id", "n/a"),
280
+ "rule_mitre_tactic": alert_data.get("rule_mitre_tactic", "n/a"),
281
+ "rule_mitre_technique": alert_data.get("rule_mitre_technique", "n/a"),
282
+ },
283
+ "alert_note": alert_data.get("ask_socfortress", "Ask SOCFortress not enabled"),
284
+ "alert_iocs": [
285
+ {
286
+ "ioc_value": alert_data["ioc_value"],
287
+ "ioc_description": "IoC found in the alert",
288
+ "ioc_tlp_id": 1,
289
+ "ioc_type_id": alert_data["ioc_type"],
290
+ },
291
+ ],
292
+ }
293
+ return payload
294
+ except Exception as e:
295
+ logger.error(f"Error creating general alert payload with IoC: {e}")
296
+ return {"success": False, "message": f"Error creating general alert payload with IoC: {e}"}
backend/app/services/DFIR_IRIS/host_enrichment.py
new
+160
@@ -0,0 +1,160 @@
1
+from abc import ABC
2
+from typing import Dict
3
+from typing import Union
4
+
5
+
6
+class AssetValidator(ABC):
7
+ """
8
+ Base class for asset validators.
9
+
10
+ Attributes:
11
+ os (str): The OS to be validated.
12
+ """
13
+
14
+ ASSET_TYPE_ID: int = 1
15
+
16
+ def __init__(self, os: str) -> None:
17
+ """
18
+ Initialize a Validator.
19
+
20
+ Args:
21
+ os (str): The OS to be validated.
22
+ """
23
+ self.os = os.lower()
24
+
25
+ def validate(self) -> Dict[str, Union[bool, str, int]]:
26
+ """
27
+ Validate the OS.
28
+
29
+ If the OS matches the type of this validator,
30
+ the method returns a dictionary indicating success, the matching message, and the asset type id.
31
+
32
+ Returns:
33
+ Dict[str, Union[bool, str, int]]: The validation result.
34
+ """
35
+ raise NotImplementedError
36
+
37
+
38
+class WindowsAssetValidator(AssetValidator):
39
+ """
40
+ Class to check if an OS is Windows.
41
+ """
42
+
43
+ ASSET_TYPE_ID = 9
44
+
45
+ def validate(self) -> Dict[str, Union[bool, str, int]]:
46
+ if "windows" in self.os:
47
+ return {
48
+ "success": True,
49
+ "message": f"{self.os} is a valid Windows OS.",
50
+ "asset_type_id": self.ASSET_TYPE_ID,
51
+ }
52
+ else:
53
+ return {
54
+ "success": False,
55
+ "message": f"{self.os} is not a Windows OS.",
56
+ "asset_type_id": self.ASSET_TYPE_ID,
57
+ }
58
+
59
+
60
+class LinuxAssetValidator(AssetValidator):
61
+ """
62
+ Class to check if an OS is Linux.
63
+ """
64
+
65
+ ASSET_TYPE_ID = 4
66
+
67
+ def validate(self) -> Dict[str, Union[bool, str, int]]:
68
+ if "linux" in self.os:
69
+ return {
70
+ "success": True,
71
+ "message": f"{self.os} is a valid Linux OS.",
72
+ "asset_type_id": self.ASSET_TYPE_ID,
73
+ }
74
+ else:
75
+ return {
76
+ "success": False,
77
+ "message": f"{self.os} is not a Linux OS.",
78
+ "asset_type_id": self.ASSET_TYPE_ID,
79
+ }
80
+
81
+
82
+class FirewallAssetValidator(AssetValidator):
83
+ """
84
+ Class to check if an OS is Firewall.
85
+ """
86
+
87
+ ASSET_TYPE_ID = 2
88
+
89
+ def validate(self) -> Dict[str, Union[bool, str, int]]:
90
+ if "firewall" in self.os:
91
+ return {
92
+ "success": True,
93
+ "message": f"{self.os} is a valid Firewall OS.",
94
+ "asset_type_id": self.ASSET_TYPE_ID,
95
+ }
96
+ else:
97
+ return {
98
+ "success": False,
99
+ "message": f"{self.os} is not a Firewall OS.",
100
+ "asset_type_id": self.ASSET_TYPE_ID,
101
+ }
102
+
103
+
104
+class UbuntuAssetValidator(AssetValidator):
105
+ """
106
+ Class to check if an OS is Ubuntu.
107
+ """
108
+
109
+ ASSET_TYPE_ID = 4
110
+
111
+ def validate(self) -> Dict[str, Union[bool, str, int]]:
112
+ if "ubuntu" in self.os:
113
+ return {
114
+ "success": True,
115
+ "message": f"{self.os} is a valid Ubuntu OS.",
116
+ "asset_type_id": self.ASSET_TYPE_ID,
117
+ }
118
+ else:
119
+ return {
120
+ "success": False,
121
+ "message": f"{self.os} is not an Ubuntu OS.",
122
+ "asset_type_id": self.ASSET_TYPE_ID,
123
+ }
124
+
125
+
126
+class AssetTypeResolver:
127
+ """
128
+ Class to iterate over asset validators and return the successful validator's asset type id.
129
+ """
130
+
131
+ def __init__(self, os: str):
132
+ """
133
+ Initialize AssetTypeResolver.
134
+
135
+ Args:
136
+ os (str): The OS to be validated.
137
+ """
138
+ self.os = os
139
+ self.validators = [
140
+ WindowsAssetValidator,
141
+ LinuxAssetValidator,
142
+ FirewallAssetValidator,
143
+ UbuntuAssetValidator,
144
+ ]
145
+
146
+ def get_asset_type_id(self) -> int:
147
+ """
148
+ Iterate over validators and return the successful validator's asset type id.
149
+
150
+ Returns:
151
+ int: The asset type id.
152
+ """
153
+ for Validator in self.validators:
154
+ validator = Validator(self.os)
155
+ result = validator.validate()
156
+ if result["success"] is True:
157
+ return result["asset_type_id"]
158
+
159
+ # Return default asset type id (1) if no validators succeed
160
+ return 1
backend/app/services/DFIR_IRIS/ioc_enrichment.py
new
+105
@@ -0,0 +1,105 @@
1
+import ipaddress
2
+import re
3
+from abc import ABC
4
+from typing import Dict
5
+from typing import Optional
6
+from typing import Union
7
+
8
+import regex
9
+from loguru import logger
10
+
11
+
12
+class IoCValidator(ABC):
13
+ """
14
+ Base class for validators.
15
+
16
+ Attributes:
17
+ value (str): The value to be validated.
18
+ """
19
+
20
+ PATTERN: Optional[str] = None # type: ignore
21
+ IOC_TYPE: Optional[int] = None # type: ignore
22
+
23
+ def __init__(self, value: str) -> None:
24
+ """
25
+ Initialize a Validator.
26
+
27
+ Args:
28
+ value (str): The value to be validated.
29
+ """
30
+ self.value = value
31
+
32
+ def validate(self) -> Dict[str, Union[bool, str, int]]:
33
+ """
34
+ Validate the value.
35
+
36
+ If the value matches the pattern,
37
+ the method returns a dictionary indicating success, the matching message, and the IOC type.
38
+
39
+ Returns:
40
+ Dict[str, Union[bool, str, int]]: The validation result.
41
+ """
42
+ logger.info(f"Validating {self.value} against {self.PATTERN}.")
43
+ if self.PATTERN and regex.match(self.PATTERN, self.value, re.IGNORECASE):
44
+ return {
45
+ "success": True,
46
+ "message": f"{self.value} matches the pattern.",
47
+ "ioc_type": self.IOC_TYPE,
48
+ }
49
+ else:
50
+ return {
51
+ "success": False,
52
+ "message": f"{self.value} does not match the pattern.",
53
+ "ioc_type": self.IOC_TYPE,
54
+ }
55
+
56
+
57
+class IPv4AddressValidator(IoCValidator):
58
+ """
59
+ Class to check if a string is a valid IPv4 address.
60
+ """
61
+
62
+ IOC_TYPE = 76
63
+
64
+ def validate(self) -> Dict[str, Union[bool, str, int]]:
65
+ """
66
+ Validate if the given value is a valid IPv4 address.
67
+
68
+ Returns:
69
+ dict: A dictionary containing success status, message, and the associated IoC type.
70
+ """
71
+ try:
72
+ # if the value is like this `162.159.133.233|443` strip the port
73
+ if "|" in self.value:
74
+ self.value = self.value.split("|")[0]
75
+ logger.info(f"Validating {self.value} as an IPv4 address.")
76
+ ipaddress.IPv4Address(self.value)
77
+ return {
78
+ "success": True,
79
+ "message": f"{self.value} is a valid IPv4 address.",
80
+ "ioc_type": self.IOC_TYPE,
81
+ }
82
+ except ValueError:
83
+ return {
84
+ "success": False,
85
+ "message": f"{self.value} is not a valid IPv4 address.",
86
+ "ioc_type": self.IOC_TYPE,
87
+ }
88
+
89
+
90
+class HashValidator(IoCValidator):
91
+ """
92
+ Class to check if a string is a valid SHA256 hash.
93
+ """
94
+
95
+ PATTERN = r"^[a-fA-F\d]{64}$"
96
+ IOC_TYPE = 113
97
+
98
+
99
+class DomainValidator(IoCValidator):
100
+ """
101
+ Class to check if a string is a valid domain name.
102
+ """
103
+
104
+ PATTERN = r"^([a-z0-9]+(-[a-z0-9]+)*\.)+[a-z]{2,}$"
105
+ IOC_TYPE = 20
backend/app/services/WazuhIndexer/alerts.py
+43
@@ -5,6 +5,7 @@ from elasticsearch7 import Elasticsearch
5
from loguru import logger
6
7
from app.services.ask_socfortress.universal import AskSocfortressService
8
+from app.services.DFIR_IRIS.alerts import IRISAlertsService
9
from app.services.WazuhIndexer.universal import UniversalService
10
11
@@ -229,6 +230,25 @@ class AlertsService:
230
logger.error(f"Failed to collect alerts: {e}")
231
return {"message": "Failed to collect alerts", "success": False}
232
233
+ def _collect_alert(self, alert_id: str, index: str) -> Dict[str, Any]:
234
+ """
235
+ Collects an alert from Elasticsearch.
236
+
237
+ Args:
238
+ alert_uid (str): The alert UID.
239
+ index (str): The name of the index to query.
240
+
241
+ Returns:
242
+ Dict[str, Any]: A dictionary containing success status and the alert or an error message.
243
+ """
244
+ logger.info(f"Collecting alert {alert_id} from {index}")
245
+ try:
246
+ alert = self.es.get(index=index, id=alert_id)
247
+ return {"message": "Successfully collected alert", "success": True, "alert": alert["_source"]}
248
+ except Exception as e:
249
+ logger.error(f"Failed to collect alert: {e}")
250
+ return {"message": "Failed to collect alert", "success": False}
251
+
252
@staticmethod
253
def _build_query() -> Dict[str, object]:
254
"""
@@ -249,3 +269,26 @@ class AlertsService:
269
},
270
"sort": [{"timestamp_utc": {"order": "desc"}}],
271
}
272
+
273
+ def escalate_alert(self, alert_id: str, index: str) -> Dict[str, Any]:
274
+ """
275
+ Escalates an alert by creating it in DFIR-IRIS
276
+
277
+ Args:
278
+ alert_id (str): The alert UID.
279
+ index (str): The index name.
280
+
281
+ Returns:
282
+ Dict[str, Any]: A dictionary containing success status and escalation message or an error message.
283
+ """
284
+ try:
285
+ alert_details = self._collect_alert(alert_id=alert_id, index=index)
286
+ service = IRISAlertsService()
287
+ logger.info(f"Escalating alert {alert_details} to DFIR-IRIS")
288
+ ask_socfortress = self.asksocfortress_service.invoke_asksocfortress(alert_details["alert"]["rule_description"])
289
+ alert_details["alert"]["ask_socfortress"] = ask_socfortress["message"]
290
+ escalation = service.create_alert_general(alert_data=alert_details["alert"], alert_id=alert_id, index=index)
291
+ return {"message": escalation["message"], "success": True}
292
+ except Exception as e:
293
+ logger.error(f"Failed to escalate alert: {e}")
294
+ return {"message": "Failed to escalate alert", "success": False}
backend/app/static/swagger.json
+69
@@ -1261,6 +1261,75 @@
1261
"tags": ["Wazuh-Indexer"]
1262
}
1263
},
1264
+ "/alerts/escalate": {
1265
+ "post": {
1266
+ "tags": ["Wazuh-Indexer"],
1267
+ "summary": "Escalate an alert.",
1268
+ "description": "Collects the alert from the Wazuh-Indexer then creates the alert in DFIR-IRIS",
1269
+ "requestBody": {
1270
+ "content": {
1271
+ "application/json": {
1272
+ "schema": {
1273
+ "type": "object",
1274
+ "properties": {
1275
+ "alert_id": {
1276
+ "type": "string",
1277
+ "description": "Alert ID"
1278
+ },
1279
+ "index": {
1280
+ "type": "string",
1281
+ "description": "Index Name"
1282
+ }
1283
+ },
1284
+ "required": ["alert_id", "index"]
1285
+ }
1286
+ }
1287
+ }
1288
+ },
1289
+ "responses": {
1290
+ "200": {
1291
+ "description": "Successfully searched for IoCs.",
1292
+ "content": {
1293
+ "application/json": {
1294
+ "schema": {
1295
+ "type": "object",
1296
+ "properties": {
1297
+ "message": {
1298
+ "type": "string",
1299
+ "example": "Successfully searched for IoCs."
1300
+ },
1301
+ "success": {
1302
+ "type": "boolean",
1303
+ "example": true
1304
+ }
1305
+ }
1306
+ }
1307
+ }
1308
+ }
1309
+ },
1310
+ "400": {
1311
+ "description": "Invalid payload.",
1312
+ "content": {
1313
+ "application/json": {
1314
+ "schema": {
1315
+ "type": "object",
1316
+ "properties": {
1317
+ "message": {
1318
+ "type": "string",
1319
+ "example": "Invalid payload."
1320
+ },
1321
+ "success": {
1322
+ "type": "boolean",
1323
+ "example": false
1324
+ }
1325
+ }
1326
+ }
1327
+ }
1328
+ }
1329
+ }
1330
+ }
1331
+ }
1332
+ },
1333
"/wazuh_indexer/allocation": {
1334
"get": {
1335
"summary": "Get node allocation of the Wazuh-Indexer nodes",