search ioc value exists within the provided field name for wazuh index (#49)
* search ioc value exists within the provided field name for wazuh index * precommit
taylor_socfortress committed
Jul 18, 2023 at 10:30 UTC
030f308da1c48dc1f9fbc9c4d93255d65041b080
3 files changed
+305
backend/app/routes/threatintel.py
+26
@@ -1,10 +1,13 @@
1
from flask import Blueprint
2
from flask import jsonify
3
+from flask import request
4
from loguru import logger
5
6
+from app.models.connectors import SocfortressThreatIntelConnector
7
from app.services.threat_intel.socfortress.universal import (
8
SocfortressThreatIntelService,
9
)
10
+from app.services.WazuhIndexer.ioc_search import IocSearchService
11
12
bp = Blueprint("threatintel", __name__)
13
@@ -20,3 +23,26 @@ def get_socfortress_threatintel(ioc_value: str) -> jsonify:
23
logger.info("Received request to check IoC in Socfortress Threat Intel")
24
ioc_enriched = SocfortressThreatIntelService("SocfortressThreatIntel").invoke_socfortress_threat_intel(data=ioc_value)
25
return jsonify(ioc_enriched)
26
+
27
+
28
+@bp.route("/threatintel/socfortress/search/wazuh", methods=["POST"])
29
+def search_wazuh_threatintel() -> jsonify:
30
+ """
31
+ Endpoint to search IoC in Wazuh Threat Intel.
32
+
33
+ Returns:
34
+ jsonify: A JSON response containing the list of all alerts from Wazuh.
35
+ """
36
+ logger.info("Received request to search IoC in Wazuh Threat Intel")
37
+ service = IocSearchService()
38
+ field_name = request.json.get("field_name")
39
+ time_range = request.json.get("time_range")
40
+ # verify connection to SOCFortress Threat Intel
41
+ verified_connection = SocfortressThreatIntelConnector("SocfortressThreatIntel").verify_connection()
42
+ try:
43
+ if verified_connection["response"] is None:
44
+ message = {"message": "Connection to SOCFortress Threat Intel failed", "success": False}
45
+ return jsonify(message)
46
+ except Exception:
47
+ ioc_searched = service.search_ioc(field_name=field_name, time_range=time_range)
48
+ return jsonify(ioc_searched)
backend/app/services/WazuhIndexer/ioc_search.py
new
+210
@@ -0,0 +1,210 @@
1
+from typing import Any
2
+from typing import Dict
3
+
4
+from elasticsearch7 import Elasticsearch
5
+from loguru import logger
6
+
7
+from app.services.threat_intel.socfortress.universal import (
8
+ SocfortressThreatIntelService,
9
+)
10
+from app.services.WazuhIndexer.universal import UniversalService
11
+
12
+
13
+class IocSearchService:
14
+ """
15
+ A service class that encapsulates the logic for searching the Wazuh Indexer for IoCs.
16
+ """
17
+
18
+ SKIP_INDEX_NAMES: Dict[str, bool] = {
19
+ "wazuh-statistics": True,
20
+ "wazuh-monitoring": True,
21
+ }
22
+
23
+ INVALID_DOMAINS = [
24
+ ".internal",
25
+ ".home",
26
+ ".local",
27
+ ".lan",
28
+ ".corp",
29
+ ".localdomain",
30
+ ".intranet",
31
+ ".localnet",
32
+ ".priv",
33
+ ]
34
+
35
+ def __init__(self):
36
+ """
37
+ Initializes the service by collecting Wazuh-Indexer details and creating an Elasticsearch client.
38
+ """
39
+ self.universal_service = UniversalService()
40
+ (
41
+ self.connector_url,
42
+ self.connector_username,
43
+ self.connector_password,
44
+ ) = self.universal_service.collect_wazuhindexer_details("Wazuh-Indexer")
45
+ self.es = Elasticsearch(
46
+ [self.connector_url],
47
+ http_auth=(self.connector_username, self.connector_password),
48
+ verify_certs=False,
49
+ timeout=15,
50
+ max_retries=10,
51
+ retry_on_timeout=False,
52
+ )
53
+ self.socfortress_threat_intel_service = SocfortressThreatIntelService("SocfortressThreatIntel")
54
+
55
+ def is_index_skipped(self, index_name: str) -> bool:
56
+ """
57
+ Checks whether the given index name should be skipped.
58
+ """
59
+ return any(index_name.startswith(skipped) for skipped in self.SKIP_INDEX_NAMES)
60
+
61
+ def is_valid_index(self, index_name: str) -> bool:
62
+ """
63
+ Checks if the index name starts with "wazuh_" and is not in the SKIP_INDEX_NAMES list.
64
+ """
65
+ return index_name.startswith("wazuh_") and not self.is_index_skipped(index_name)
66
+
67
+ def _collect_indices_and_validate(self) -> Dict[str, Any]:
68
+ """
69
+ Collect indices and validate connector details.
70
+ """
71
+ if not all([self.connector_url, self.connector_username, self.connector_password]):
72
+ return self._error_response("Failed to collect Wazuh-Indexer details")
73
+
74
+ indices_list = self.universal_service.collect_indices()
75
+ if not indices_list["success"]:
76
+ return self._error_response("Failed to collect indices")
77
+
78
+ valid_indices = [index for index in indices_list["indices_list"] if self.is_valid_index(index)]
79
+
80
+ return {"success": True, "indices": valid_indices}
81
+
82
+ def search_ioc(self, field_name: str, time_range: str) -> Dict[str, object]:
83
+ """
84
+ Search for IoCs from the given field name.
85
+
86
+ Args:
87
+ field_name (str): The field name to search for IoCs.
88
+ time_range (str): The time range to search for IoCs. I.E: 24h
89
+
90
+ Returns:
91
+ Dict[str, object]: A dictionary containing the list of all alerts from Wazuh.
92
+ """
93
+ indices_validation = self._collect_indices_and_validate()
94
+ if not indices_validation["success"]:
95
+ return indices_validation
96
+
97
+ alerts_summary = []
98
+ for index_name in indices_validation["indices"]:
99
+ alerts = self._collect_iocs(index_name, field_name=field_name, time_range=time_range)
100
+ if alerts["success"] and len(alerts["alerts"]) > 0:
101
+ alerts_summary.append(
102
+ {
103
+ "index_name": index_name,
104
+ "total_alerts": len(alerts["alerts"]),
105
+ "alerts": alerts["alerts"],
106
+ },
107
+ )
108
+ return {
109
+ "message": f"Successfully collected alerts with discovered IoCs from {field_name}",
110
+ "success": True,
111
+ "alerts_summary": alerts_summary,
112
+ }
113
+
114
+ @staticmethod
115
+ def _error_response(message: str) -> Dict[str, bool]:
116
+ """
117
+ Standardizes the error response format.
118
+ """
119
+ return {"message": message, "success": False}
120
+
121
+ @staticmethod
122
+ def is_invalid_domain_name(field_value: str) -> bool:
123
+ """
124
+ Checks if the given field value is an internal domain name.
125
+
126
+ Args:
127
+ field_value (str): The field value to check.
128
+
129
+ Returns:
130
+ bool: True if the field value is an internal domain name, False otherwise.
131
+ """
132
+ return any(field_value.endswith(domain) for domain in IocSearchService.INVALID_DOMAINS)
133
+
134
+ def _collect_iocs(self, index_name: str, field_name: str, time_range: str) -> Dict[str, object]:
135
+ """
136
+ Elasticsearch query to retrieve all values of a given field and index to
137
+ invoke the IOC service to discover if the value is malicious or not.
138
+ I.E: `data_srcip` field in `wazuh-*` index.
139
+
140
+ Args:
141
+ index_name (str): The index name to query.
142
+ field_name (str): The field name to query.
143
+ time_range (str): The time range to query. I.E: 24h
144
+
145
+ Returns:
146
+ Dict[str, object]: The response from the IOC service.
147
+ """
148
+ logger.info(f"Collecting alerts from {index_name}")
149
+ query = self._build_query(field_name=field_name, time_range=time_range)
150
+ try:
151
+ alerts = self.es.search(index=index_name, body=query)
152
+ alerts_list = self._filter_and_enrich_alerts(alerts["hits"]["hits"], field_name)
153
+ return {
154
+ "message": "Successfully collected alerts",
155
+ "success": True,
156
+ "alerts": alerts_list,
157
+ }
158
+ except Exception as e:
159
+ logger.error(f"Failed to collect alerts: {e}")
160
+ return {"message": "Failed to collect alerts", "success": False}
161
+
162
+ def _filter_and_enrich_alerts(self, alerts_list, field_name):
163
+ """
164
+ Filters out invalid domain names and enriches the alerts with socfortress_threat_intel.
165
+ """
166
+ alerts_list_with_response = [] # Create a new list to store alerts with responses
167
+ for alert in alerts_list:
168
+ field_value = alert["_source"][field_name]
169
+ if self.is_invalid_domain_name(field_value):
170
+ continue
171
+ socfortress_threat_intel = self.socfortress_threat_intel_service.invoke_socfortress_threat_intel(data=field_value)
172
+ logger.info(f"Socfortress threat intel response: {socfortress_threat_intel}")
173
+ # if `response` is not empty, add it to the alert
174
+ if socfortress_threat_intel["response"]:
175
+ alert["_source"]["socfortress_threat_intel"] = socfortress_threat_intel["response"]
176
+ alerts_list_with_response.append(alert) # Add the alert to the new list
177
+ return alerts_list_with_response
178
+
179
+ @staticmethod
180
+ def _build_query(field_name: str, time_range: str) -> Dict[str, Any]:
181
+ """
182
+ Builds the Elasticsearch query to retrieve all values of a given field.
183
+
184
+ Args:
185
+ field_name (str): The field name to query.
186
+ time_range (str): The time range to query. I.E: 24h
187
+
188
+ Returns:
189
+ Dict[str, Any]: The Elasticsearch query.
190
+ """
191
+ return {
192
+ "query": {
193
+ "bool": {
194
+ "must": [
195
+ {
196
+ "exists": {
197
+ "field": field_name,
198
+ },
199
+ },
200
+ {
201
+ "range": {
202
+ "timestamp": {
203
+ "gte": f"now-{time_range}",
204
+ },
205
+ },
206
+ },
207
+ ],
208
+ },
209
+ },
210
+ }
backend/app/static/swagger.json
+69
@@ -3175,6 +3175,75 @@
3175
"operationId": "getSOCFortressThreatIntelByIOCValue",
3176
"tags": ["Threat Intel"]
3177
}
3178
+ },
3179
+ "/threatintel/socfortress/search/wazuh": {
3180
+ "post": {
3181
+ "tags": ["Threat Intel"],
3182
+ "summary": "Search field name for IoCs in Wazuh Indicies",
3183
+ "description": "Search field name for IoCs in Wazuh Indicies.",
3184
+ "requestBody": {
3185
+ "content": {
3186
+ "application/json": {
3187
+ "schema": {
3188
+ "type": "object",
3189
+ "properties": {
3190
+ "field_name": {
3191
+ "type": "string",
3192
+ "description": "Field name to search for IoCs"
3193
+ },
3194
+ "time_range": {
3195
+ "type": "string",
3196
+ "description": "Time range to search for IoCs"
3197
+ }
3198
+ },
3199
+ "required": ["field_name", "time_range"]
3200
+ }
3201
+ }
3202
+ }
3203
+ },
3204
+ "responses": {
3205
+ "200": {
3206
+ "description": "Successfully searched for IoCs.",
3207
+ "content": {
3208
+ "application/json": {
3209
+ "schema": {
3210
+ "type": "object",
3211
+ "properties": {
3212
+ "message": {
3213
+ "type": "string",
3214
+ "example": "Successfully searched for IoCs."
3215
+ },
3216
+ "success": {
3217
+ "type": "boolean",
3218
+ "example": true
3219
+ }
3220
+ }
3221
+ }
3222
+ }
3223
+ }
3224
+ },
3225
+ "400": {
3226
+ "description": "Invalid payload.",
3227
+ "content": {
3228
+ "application/json": {
3229
+ "schema": {
3230
+ "type": "object",
3231
+ "properties": {
3232
+ "message": {
3233
+ "type": "string",
3234
+ "example": "Invalid payload."
3235
+ },
3236
+ "success": {
3237
+ "type": "boolean",
3238
+ "example": false
3239
+ }
3240
+ }
3241
+ }
3242
+ }
3243
+ }
3244
+ }
3245
+ }
3246
+ }
3247
}
3248
},
3249
"components": {