logic to check if valid hash, domain, or ipv4 (#50)
taylor_socfortress committed
Jul 18, 2023 at 11:00 UTC
b6de0516c5501e0be8e9841e5a83edffd98fa6b4
2 files changed
+88
-2
backend/app/routes/threatintel.py
+1
-1
@@ -26,7 +26,7 @@ def get_socfortress_threatintel(ioc_value: str) -> jsonify:
26
27
28
@bp.route("/threatintel/socfortress/search/wazuh", methods=["POST"])
29
-def search_wazuh_threatintel() -> jsonify:
29
+def search_socfortress_wazuh_threatintel() -> jsonify:
30
"""
31
Endpoint to search IoC in Wazuh Threat Intel.
32
backend/app/services/WazuhIndexer/ioc_search.py
+87
-1
@@ -1,3 +1,5 @@
1
+import ipaddress
2
+import re
3
from typing import Any
4
from typing import Dict
5
@@ -131,6 +133,88 @@ class IocSearchService:
133
"""
134
return any(field_value.endswith(domain) for domain in IocSearchService.INVALID_DOMAINS)
135
136
+ @staticmethod
137
+ def is_valid_domain_name(domain: str) -> bool:
138
+ """
139
+ Checks if the given domain name is valid.
140
+
141
+ Args:
142
+ domain (str): The domain name to check.
143
+
144
+ Returns:
145
+ bool: True if the domain name is valid, False otherwise.
146
+ """
147
+ pattern = r"(?:[a-z0-9](?:[a-z0-9\-]{0,61}[a-z0-9])?\.)+[a-z0-9][a-z0-9\-]{0,61}[a-z0-9]"
148
+ return re.fullmatch(pattern, domain) is not None
149
+
150
+ @staticmethod
151
+ def is_valid_ipv4(field_value: str) -> bool:
152
+ """
153
+ Checks if the given field value is a valid IPv4 address.
154
+
155
+ Args:
156
+ field_value (str): The field value to check.
157
+
158
+ Returns:
159
+ bool: True if the field value is a valid IPv4 address, False otherwise.
160
+ """
161
+ try:
162
+ ipaddress.IPv4Address(field_value)
163
+ return True
164
+ except ValueError:
165
+ return False
166
+
167
+ @staticmethod
168
+ def is_valid_md5(field_value: str) -> bool:
169
+ """
170
+ Checks if the given field value is a valid MD5 hash.
171
+
172
+ Args:
173
+ field_value (str): The field value to check.
174
+
175
+ Returns:
176
+ bool: True if the field value is a valid MD5 hash, False otherwise.
177
+ """
178
+ return bool(re.fullmatch(r"[a-fA-F0-9]{32}", field_value))
179
+
180
+ @staticmethod
181
+ def is_valid_sha256(field_value: str) -> bool:
182
+ """
183
+ Checks if the given field value is a valid SHA256 hash.
184
+
185
+ Args:
186
+ field_value (str): The field value to check.
187
+
188
+ Returns:
189
+ bool: True if the field value is a valid SHA256 hash, False otherwise.
190
+ """
191
+ return bool(re.fullmatch(r"[a-fA-F0-9]{64}", field_value))
192
+
193
+ def is_valid_field_value(self, field_value: str) -> bool:
194
+ """
195
+ Validates the field value as per different conditions.
196
+
197
+ Args:
198
+ field_value (str): The field value to check.
199
+
200
+ Returns:
201
+ bool: True if the field value is valid, False otherwise.
202
+ """
203
+ if self.is_invalid_domain_name(field_value):
204
+ logger.info(f"Skipping invalid domain name: {field_value}")
205
+ return False
206
+
207
+ if not (
208
+ self.is_valid_ipv4(field_value)
209
+ or self.is_valid_md5(field_value)
210
+ or self.is_valid_sha256(field_value)
211
+ or self.is_valid_domain_name(field_value)
212
+ ):
213
+ logger.info(f"Skipping invalid field value: {field_value}")
214
+ return False
215
+
216
+ return True
217
+
218
def _collect_iocs(self, index_name: str, field_name: str, time_range: str) -> Dict[str, object]:
219
"""
220
Elasticsearch query to retrieve all values of a given field and index to
@@ -166,8 +250,10 @@ class IocSearchService:
250
alerts_list_with_response = [] # Create a new list to store alerts with responses
251
for alert in alerts_list:
252
field_value = alert["_source"][field_name]
169
- if self.is_invalid_domain_name(field_value):
253
+ # Skip if field value is invalid
254
+ if not self.is_valid_field_value(field_value):
255
continue
256
+
257
socfortress_threat_intel = self.socfortress_threat_intel_service.invoke_socfortress_threat_intel(data=field_value)
258
logger.info(f"Socfortress threat intel response: {socfortress_threat_intel}")
259
# if `response` is not empty, add it to the alert