Add URL scanning to IOC type determination and VirusTotal API fetching - still WIP (#344)
taylor_socfortress committed
Nov 30, 2024 at 11:09 UTC
66762441c6d8285bc80f112846337ae7d8674b31
1 file changed
+23
-4
backend/app/threat_intel/services/socfortress.py
+23
-4
@@ -168,6 +168,7 @@ def determine_ioc_type(ioc_value: str) -> str:
168
ip_pattern = re.compile(r"^(?:[0-9]{1,3}\.){3}[0-9]{1,3}$")
169
domain_pattern = re.compile(r"^(?:[a-zA-Z0-9-]+\.)+[a-zA-Z]{2,}$")
170
hash_pattern = re.compile(r"^[a-fA-F0-9]{32}$|^[a-fA-F0-9]{40}$|^[a-fA-F0-9]{64}$")
171
+ url_pattern = re.compile(r"^(https?|ftp):\/\/[^\s/$.?#].[^\s]*$")
172
173
if ip_pattern.match(ioc_value):
174
return f"/ip_addresses/{ioc_value}"
@@ -175,20 +176,28 @@ def determine_ioc_type(ioc_value: str) -> str:
176
return f"/domains/{ioc_value}"
177
elif hash_pattern.match(ioc_value):
178
return f"/files/{ioc_value}"
179
+ elif url_pattern.match(ioc_value):
180
+ raise HTTPException(
181
+ status_code=400,
182
+ detail="URL scanning is currently not supported.",
183
+ )
184
+ return "/urls"
185
else:
186
raise HTTPException(
187
status_code=400,
181
- detail="Invalid IOC value provided. Only IP addresses, domains, and hashes are supported.",
188
+ detail="Invalid IOC value provided. Only IP addresses, domains, URLs, and hashes are supported.",
189
)
190
191
185
-async def fetch_virustotal_data(api_key: str, full_url: str) -> dict:
192
+async def fetch_virustotal_data(api_key: str, full_url: str, ioc_value: str, is_url: bool) -> dict:
193
"""
194
Fetch data from the VirusTotal API.
195
196
Args:
197
api_key (str): The API key for authentication.
198
full_url (str): The full URL of the VirusTotal API endpoint.
199
+ ioc_value (str): The IOC value.
200
+ is_url (bool): Flag indicating if the IOC value is a URL.
201
202
Returns:
203
dict: The JSON response from the VirusTotal API.
@@ -198,7 +207,16 @@ async def fetch_virustotal_data(api_key: str, full_url: str) -> dict:
207
"""
208
headers = {"x-apikey": api_key}
209
async with httpx.AsyncClient() as client:
201
- response = await client.get(full_url, headers=headers)
210
+ if is_url:
211
+ headers["Content-Type"] = "application/x-www-form-urlencoded"
212
+ data = {"url": ioc_value}
213
+ response = await client.post(full_url, headers=headers, data=data)
214
+ response.raise_for_status()
215
+ analysis_id = response.json()["data"]["id"]
216
+ url_report_url = f"https://www.virustotal.com/api/v3/urls/{analysis_id}"
217
+ response = await client.get(url_report_url, headers=headers)
218
+ else:
219
+ response = await client.get(full_url, headers=headers)
220
response.raise_for_status()
221
return VirusTotalResponse.parse_obj(response.json())
222
@@ -225,7 +243,8 @@ async def invoke_virustotal_api(
243
ioc_value = request.ioc_value
244
endpoint = determine_ioc_type(ioc_value)
245
full_url = f"{url}{endpoint}"
228
- return await fetch_virustotal_data(api_key, full_url)
246
+ is_url = endpoint == "/urls"
247
+ return await fetch_virustotal_data(api_key, full_url, ioc_value, is_url)
248
249
250
async def invoke_socfortress_process_name_api(