maq/LinkScope_Client
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update Resolutions

Browse Source
This commit is contained in:
AccentuSoft
2023-04-25 22:16:53 +03:00
parent 1abc1377ea
commit 758e2f7a03
31 changed files with 325 additions and 359 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
Core/Resolutions/Core/EmailExtractor.py
View File

@@ -96,9 +96,9 @@ class EmailExtractor:
for potentialEmail in potentialEmails:
with contextlib.suppress(EmailNotValidError):
valid = validate_email(potentialEmail, dns_resolver=resolver, check_deliverability=verifyDomain)
if valid.email not in allEmails:
allEmails.add(valid.email)
returnResults.append([{'Email Address': valid.email,
if valid.normalized not in allEmails:
allEmails.add(valid.normalized)
returnResults.append([{'Email Address': valid.normalized,
'Entity Type': 'Email Address'},
{currentUID: {'Resolution': 'Email Address Found',
'Notes': ''}}])
@@ -119,9 +119,7 @@ class EmailExtractor:
with sync_playwright() as p:
browser = p.chromium.launch()
context = browser.new_context(
viewport={'width': 1920, 'height': 1080},
user_agent='Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) '
'Chrome/101.0.4951.54 Safari/537.36'
viewport={'width': 1920, 'height': 1080}
)
for entity in entityJsonList:
uid = entity['uid']

5
Core/Resolutions/Core/FileExtractor.py
View File

@@ -45,7 +45,6 @@ class FileExtractor:
import tldextract
import requests
from hashlib import md5
from binascii import hexlify
from pathlib import Path
from bs4 import BeautifulSoup
from playwright.sync_api import sync_playwright, TimeoutError, Error
@@ -117,7 +116,7 @@ class FileExtractor:
'Notes': ''}}])
docProperName = link.split('/')[-1]
docFileName = f'{hexlify(md5(link.encode()).digest()).decode()} | {docProperName}'
docFileName = f'{md5(link.encode("UTF-8")).hexdigest()} | {docProperName}'
docFullPath = Path(parameters['Project Files Directory']) / docFileName
with contextlib.suppress(Exception):
@@ -156,7 +155,7 @@ class FileExtractor:
{uid: {'Resolution': 'File URL',
'Notes': ''}}])
docProperName = link.split('/')[-1]
docFileName = f'{hexlify(md5(link.encode()).digest()).decode()} | {docProperName}'
docFileName = f'{md5(link.encode()).hexdigest()} | {docProperName}'
docFullPath = Path(parameters['Project Files Directory']) / docFileName
with contextlib.suppress(Exception):

8
Modules/Aircraft_Inquiry/AircraftInquiryByDealer.py
View File

@@ -19,12 +19,11 @@ class AircraftInquiryByDealer:
uidList = []
return_result = []
submit_url = "https://registry.faa.gov/aircraftinquiry/Search/"
crafted_url = f"{submit_url}DealerResult"
submit_url = "https://registry.faa.gov/aircraftinquiry/Search/DealerResult"
with FuturesSession(max_workers=15) as session:
for entity in entityJsonList:
uidList.append(entity['uid'])
futures.append(session.post(crafted_url, data={"Dealertxt": entity['Company Name']}))
futures.append(session.post(submit_url, data={"Dealertxt": entity['Company Name']}))
for future in as_completed(futures):
uid = uidList[futures.index(future)]
try:
@@ -32,8 +31,7 @@ class AircraftInquiryByDealer:
except requests.exceptions.ConnectionError:
return "Please check your internet connection"
except ValueError:
return_result = []
return return_result
return "Error occurred when checking the data returned from the endpoint."
df = df_list[0]
for certificate_index in range(len(df["Certificate Number"])):
index_of_child = len(return_result)

7
Modules/Aircraft_Inquiry/AircraftInquiryByEngine.py
View File

@@ -23,13 +23,12 @@ class AircraftInquiryByEngine:
uidList = []
return_result = []
submit_url = "https://registry.faa.gov/aircraftinquiry/Search/"
crafted_url = f"{submit_url}EngineReferenceResult"
submit_url = "https://registry.faa.gov/aircraftinquiry/Search/EngineReferenceResult"
with FuturesSession(max_workers=15) as session:
for entity in entityJsonList:
uidList.append(entity['uid'])
futures.append(session.post(crafted_url, data={"Modeltxt": entity['Phrase'],
"MfrNametxt": Manufacturer}))
futures.append(session.post(submit_url, data={"Modeltxt": entity['Phrase'],
"MfrNametxt": Manufacturer}))
for future in as_completed(futures):
uid = uidList[futures.index(future)]
try:

5
Modules/Aircraft_Inquiry/AircraftInquiryByNNumber.py
View File

@@ -19,12 +19,11 @@ class AircraftInquiryByNNumber:
uidList = []
return_result = []
submit_url = "https://registry.faa.gov/aircraftinquiry/Search/"
crafted_url = f"{submit_url}NNumberResult"
submit_url = "https://registry.faa.gov/aircraftinquiry/Search/NNumberResult"
with FuturesSession(max_workers=15) as session:
for entity in entityJsonList:
uidList.append(entity['uid'])
futures.append(session.post(crafted_url, data={"NNumbertxt": entity['Phrase']}))
futures.append(session.post(submit_url, data={"NNumbertxt": entity['Phrase']}))
for future in as_completed(futures):
uid = uidList[futures.index(future)]
try:

15
Modules/Aircraft_Inquiry/AircraftInquiryByPersonName.py
View File

@@ -6,7 +6,7 @@ class AircraftInquiryByPersonName:
category = "Aircraft"
description = "Find information about aircraft identifications from https://registry.faa.gov/aircraftinquiry/"
originTypes = {"Person"}
resultTypes = {'Phrase', 'Person', 'Identification Number', 'Company'}
resultTypes = {'Phrase', 'Identification Number', 'Company'}
parameters = {}
def resolution(self, entityJsonList, parameters):
@@ -19,12 +19,11 @@ class AircraftInquiryByPersonName:
uidList = []
return_result = []
submit_url = "https://registry.faa.gov/aircraftinquiry/Search/"
crafted_url = f"{submit_url}NameResult"
submit_url = "https://registry.faa.gov/aircraftinquiry/Search/NameResult"
with FuturesSession(max_workers=15) as session:
for entity in entityJsonList:
uidList.append(entity['uid'])
futures.append(session.post(crafted_url, data={"nametxt": entity['Full Name'], "sort_option": "1"}))
futures.append(session.post(submit_url, data={"nametxt": entity['Full Name'], "sort_option": "1"}))
for future in as_completed(futures):
uid = uidList[futures.index(future)]
try:
@@ -34,14 +33,14 @@ class AircraftInquiryByPersonName:
except ValueError:
return "No results retrieved"
df = df_list[0]
for i in range(len(df["N-Number"])):
return_result.append([{'Phrase': df["N-Number"][0],
for index in range(len(df["N-Number"])):
return_result.append([{'Phrase': df["N-Number"][index],
'Entity Type': 'Phrase'},
{uid: {'Resolution': 'Aircraft N-Number', 'Notes': ''}}])
return_result.append([{'ID Number': str(df['Serial Number'][0]),
return_result.append([{'ID Number': str(df['Serial Number'][index]),
'Entity Type': 'Identification Number'},
{uid: {'Resolution': 'Aircraft Identification Number', 'Notes': ''}}])
return_result.append([{'Company Name': df['Manufacturer Name Model'][0],
return_result.append([{'Company Name': df['Manufacturer Name Model'][index],
'Entity Type': 'Company'},
{uid: {'Resolution': 'Aircraft Manufacturer Name', 'Notes': ''}}])
return return_result

15
Modules/Aircraft_Inquiry/AircraftInquiryBySerialNumber.py
View File

@@ -6,7 +6,7 @@ class AircraftInquiryBySerialNumber:
category = "Aircraft"
description = "Find information about aircraft identifications from https://registry.faa.gov/aircraftinquiry/"
originTypes = {"Identification Number"}
resultTypes = {'Phrase', 'Person', 'Identification Number', 'Company'}
resultTypes = {'Phrase', 'Company'}
parameters = {}
def resolution(self, entityJsonList, parameters):
@@ -19,12 +19,11 @@ class AircraftInquiryBySerialNumber:
uidList = []
return_result = []
submit_url = "https://registry.faa.gov/aircraftinquiry/Search/"
crafted_url = f"{submit_url}SerialResult"
submit_url = "https://registry.faa.gov/aircraftinquiry/Search/SerialResult"
with FuturesSession(max_workers=15) as session:
for entity in entityJsonList:
uidList.append(entity['uid'])
futures.append(session.post(crafted_url, data={"Serialtxt": entity['ID Number'], "sort_option": "1"}))
futures.append(session.post(submit_url, data={"Serialtxt": entity['ID Number'], "sort_option": "1"}))
for future in as_completed(futures):
uid = uidList[futures.index(future)]
try:
@@ -34,14 +33,14 @@ class AircraftInquiryBySerialNumber:
except ValueError:
return "No results retrieved"
df = df_list[0]
for i in range(len(df["N-Number"])):
return_result.append([{'Company Name': df["Manufacturer Name"][i],
for index in range(len(df["N-Number"])):
return_result.append([{'Company Name': df["Manufacturer Name"][index],
'Entity Type': 'Company'},
{uid: {'Resolution': 'Aircraft Manufacturer', 'Notes': ''}}])
return_result.append([{'Phrase': df["N-Number"][i],
return_result.append([{'Phrase': df["N-Number"][index],
'Entity Type': 'Phrase'},
{uid: {'Resolution': 'Aircraft N-Number', 'Notes': ''}}])
return_result.append([{'Phrase': df["Model"][i],
return_result.append([{'Phrase': df["Model"][index],
'Entity Type': 'Phrase'},
{uid: {'Resolution': 'Aircraft Model', 'Notes': ''}}])
return return_result

27
Modules/Aleph/Aleph_Entity_Search.py
View File

@@ -6,16 +6,18 @@ class Aleph_Entity_Search:
category = "Aleph OCCRP"
description = "Find information about a given search parameter"
originTypes = {'Phrase', 'Person', 'Politically Exposed Person'}
resultTypes = {'Phrase'}
resultTypes = {'Phrase', 'Person', 'Company', 'Address', 'Country', 'Aleph ID', 'Aleph Collection ID',
'Phone Number', 'Organization'}
parameters = {'Max Results': {'description': 'The maximum number of results to return.',
'type': 'String',
'value': 'Enter the number of results you want returned',
'default': '1'},
'Aleph Disclaimer': {'description': 'The content on Aleph is provided for general information only.\n'
'It is not intended to amount to advice on which you should place'
'sole and entire reliance.\n'
'We recommend that you conduct your own independent fact checking'
'against the data and materials that you access on Aleph.\n'
'It is not intended to amount to advice on which you should '
'place sole and entire reliance.\n'
'We recommend that you conduct your own independent fact '
'checking against the data and materials that you '
'access on Aleph.\n'
'Aleph API is not a replacement for traditional due diligence '
'checks and know-your-customer background checks.',
'type': 'String',
@@ -53,7 +55,7 @@ class Aleph_Entity_Search:
for entity in entityJsonList:
uidList.append(entity['uid'])
primary_field = entity[list(entity)[1]].strip()
crafted_url = url + f"?q={primary_field}&filter:schemata=Thing&limit={max_results}"
crafted_url = f"{url}?q={primary_field}&filter:schemata=Thing&limit={max_results}"
time.sleep(1)
futures.append(session.get(crafted_url, headers=headers))
for future in as_completed(futures):
@@ -62,17 +64,15 @@ class Aleph_Entity_Search:
response = future.result().json()
except requests.exceptions.ConnectionError:
return "Please check your internet connection"
# print(response)
for schema in response['results']:
index_of_child = len(return_result)
try:
if schema['schema'] == "Person":
if schema['properties'].get('gender') is not None \
and schema['properties'].get('gender')[0] == "F":
gender = "Female"
elif schema['properties'].get('gender') is not None \
and schema['properties'].get('gender')[0] == "M":
gender = "Male"
if schema['properties'].get('gender') is not None:
if schema['properties'].get('gender')[0] == "F":
gender = "Female"
elif schema['properties'].get('gender')[0] == "M":
gender = "Male"
if schema['properties'].get('legalForm') is not None:
return_result.append(
[{'Full Name': schema['properties']['name'][0],
@@ -268,6 +268,5 @@ class Aleph_Entity_Search:
'Entity Type': 'Aleph Collection ID'},
{index_of_child_of_child: {'Resolution': 'Aleph Entity Search', 'Notes': ''}}])
except (TypeError, KeyError):
# print(repr(e))
continue
return return_result

15
Modules/Aleph/GetCollectionByID.py
View File

@@ -2,16 +2,17 @@
class GetCollectionByID:
name = "Get Collections By ID"
name = "Get Collection for Phrase"
category = "Aleph OCCRP"
description = "Find information about Collections and their IDs"
description = "Get the Collections that contain info related to the given Phrases."
originTypes = {'Phrase'}
resultTypes = {'Phrase. Person, Address, Phone Number, Email Address, Country, Bank Account'}
resultTypes = {'Phrase', 'Person', 'Address', 'Phone Number', 'Email Address', 'Country', 'Bank Account'}
parameters = {'Aleph Disclaimer': {'description': 'The content on Aleph is provided for general information only.\n'
'It is not intended to amount to advice on which you should place'
'sole and entire reliance.\n'
'We recommend that you conduct your own independent fact checking'
'against the data and materials that you access on Aleph.\n'
'It is not intended to amount to advice on which you should '
'place sole and entire reliance.\n'
'We recommend that you conduct your own independent fact '
'checking against the data and materials that you '
'access on Aleph.\n'
'Aleph API is not a replacement for traditional due diligence '
'checks and know-your-customer background checks.',
'type': 'String',

4
Modules/Aleph/GetCollectionsInfo.py
View File

@@ -9,7 +9,7 @@ class GetCollectionsInfo:
resultTypes = {'Phrase, Aleph ID'}
parameters = {'Max Results': {'description': 'Please enter the maximum number of results to return.',
'type': 'String',
'default': '1'},
'default': ''},
'Aleph Disclaimer': {'description': 'The content on Aleph is provided for general information only.\n'
'It is not intended to amount to advice on which you should place'
'sole and entire reliance.\n'
@@ -46,7 +46,7 @@ class GetCollectionsInfo:
with FuturesSession(max_workers=15) as session:
for entity in entityJsonList:
uidList.append(entity['uid'])
url = f"https://aleph.occrp.org/api/2/collections?offset=0&limit=300&page"
url = f"https://aleph.occrp.org/api/2/collections?offset=0&limit=300&q={entity['Phrase']}"
time.sleep(1)
futures.append(session.get(url, headers=headers))
for future in as_completed(futures):

11
Modules/Aleph/GetSimilarEntities.py
View File

@@ -6,12 +6,13 @@ class GetSimilarEntities:
category = "Aleph OCCRP"
description = "Find information about similar entities"
originTypes = {'Phrase', 'Person', 'Politically Exposed Person'}
resultTypes = {'Phrase', 'Person', 'Address', 'Aleph ID'}
resultTypes = {'Phrase', 'Person', 'Address', 'Aleph ID', 'Company'}
parameters = {'Aleph Disclaimer': {'description': 'The content on Aleph is provided for general information only.\n'
'It is not intended to amount to advice on which you should place'
'sole and entire reliance.\n'
'We recommend that you conduct your own independent fact checking'
'against the data and materials that you access on Aleph.\n'
'It is not intended to amount to advice on which you should '
'place sole and entire reliance.\n'
'We recommend that you conduct your own independent fact '
'checking against the data and materials that you '
'access on Aleph.\n'
'Aleph API is not a replacement for traditional due diligence '
'checks and know-your-customer background checks.',
'type': 'String',

7
Modules/BigMatch/BigMatch.py
View File

@@ -52,10 +52,9 @@ class BigMatch:
return []
for link in soup.find_all('a'):
potentialLink = link.get('href', None)
if potentialLink is not None:
if 'github' in potentialLink:
return_result.append([{'URL': potentialLink, 'Entity Type': 'Website'},
{uid: {'Resolution': 'BigMatch Github Link', 'Notes': ''}}])
if potentialLink is not None and 'github' in potentialLink:
return_result.append([{'URL': potentialLink, 'Entity Type': 'Website'},
{uid: {'Resolution': 'BigMatch Github Link', 'Notes': ''}}])
break
except TimeoutError:

8
Modules/BinaryEdge/BinaryEdgeHost.py
View File

@@ -43,8 +43,6 @@ class BinaryEdgeHost:
for event in requestContent['events']:
for result in event['results']:
originDetails = result['origin']
targetDetails = result['target']
resultDetails = result['result']
if 'state' not in resultDetails['data']:
# Discard return result if it doesn't actually give us useful info about the state of the port.
@@ -52,8 +50,10 @@ class BinaryEdgeHost:
# There seems to always be a result with the simple port info, so we will use that one.
continue
returnResults.append([{'Port': targetDetails['ip'] + ':' + str(targetDetails['port']) + ':' +
targetDetails['protocol'],
originDetails = result['origin']
targetDetails = result['target']
returnResults.append([{'Port': f"{targetDetails['ip']}:{targetDetails['port']}:"
f"{targetDetails['protocol']}",
'State': resultDetails['data']['state']['state'],
'Banner': resultDetails['data']['service'].get('banner', 'N/A'),
'Product': resultDetails['data']['service'].get('product', 'Unknown'),

11
Modules/BreachData/HIBPBreachToDomain.py
View File

@@ -16,11 +16,10 @@ class HIBPBreachToDomain:
for entity in entityJsonList:
domainMaybe = entity.get('Breach Domain')
if isinstance(domainMaybe, str):
if domainMaybe.strip() != '':
returnResults.append([{'Domain Name': domainMaybe,
'Entity Type': 'Domain'},
{entity['uid']: {'Resolution': 'Data Breach to Domain',
'Notes': ''}}])
if isinstance(domainMaybe, str) and domainMaybe.strip() != '':
returnResults.append([{'Domain Name': domainMaybe,
'Entity Type': 'Domain'},
{entity['uid']: {'Resolution': 'Data Breach to Domain',
'Notes': ''}}])
return returnResults

10
Modules/GitHubRSAFinder/OrgSearch_GitAllSecrets.py
View File

@@ -8,11 +8,12 @@ class OrgSearch_GitAllSecrets:
category = "Secrets & Leaks"
# A string that describes this resolution.
description = "Returns Nodes of Relationship Info. Requires Docker to be installed."
description = "Searches Github Organization repositories for exposed secrets. Requires Docker to be installed."
originTypes = {'GitHub Organisation'}
resultTypes = {}
resultTypes = {'GitHub Organisation', 'GitHub Secret', 'GitHub Repository', 'GitHub FilePath', 'Hash',
'GitHub Branch'}
parameters = {'Github Token': {'description': 'Github personal access token.\n'
'We need this because unauthenticated requests to the Github API '
@@ -39,7 +40,6 @@ class OrgSearch_GitAllSecrets:
client = docker.from_env()
with tempfile.TemporaryDirectory() as tempDir:
tempPath = Path(tempDir).absolute()
# print(tempPath, tempPath.exists())
client.containers.run('abhartiya/tools_gitallsecrets:latest',
f'-token={parameters["Token"]} '
f'-org={entity[list(entity)[1]]} -output=/home/out.txt',
@@ -60,9 +60,6 @@ class OrgSearch_GitAllSecrets:
if line.startswith('OrgorUser'):
orgOrUser.append(line.split(' '))
# print(hogSecret)
# print(jsonContents)
data = pattern.findall(repoSupervisor)
for value in data:
data.append(json.loads(value))
@@ -95,7 +92,6 @@ class OrgSearch_GitAllSecrets:
'Entity Type': 'Phrase'},
{index_of_child: {'Resolution': 'GitHub Secret',
'Notes': ''}}])
# print(ansi_escape.sub('', hogSecret[hogIndex]))
hogIndex += 1
elif 'Hash' in line:
returnResults.append([{'Hash Value': ansi_escape.sub('', line),

2
Modules/InternetDB/InternetDB.py
View File

@@ -29,7 +29,7 @@ class InternetDB:
for entity in entityJsonList:
primaryField = entity['IP Address']
entityUID = entity['uid']
requestResult = requests.get("https://internetdb.shodan.io/" + primaryField).json()
requestResult = requests.get(f"https://internetdb.shodan.io/{primaryField}").json()
if "detail" in requestResult:
returnResults.append([{'Phrase': requestResult['detail'],

30
Modules/Offshore_Leaks/Offshore_Leaks_Entities.py
View File

@@ -39,7 +39,7 @@ class Offshore_Leaks_Entities:
primary_field = entity[list(entity)[1]].strip()
df_list = []
for batch in range(0, nextHundred, 100):
crafted_url = url + f"search?cat=0&from={batch}&q={primary_field}&utf8=✓"
crafted_url = f"{url}search?cat=0&from={batch}&q={primary_field}&utf8=✓"
try:
r = requests.get(crafted_url)
df_list.append(pd.read_html(r.text)[0])
@@ -62,30 +62,28 @@ class Offshore_Leaks_Entities:
'Entity Type': 'Company'},
{uid: {'Resolution': 'Offshore Leaks Entity', 'Notes': ''}}])
if isinstance(df_part['Data from'][entry], str) and \
"not identified" not in df_part['Data from'][entry].lower():
"not identified" not in df_part['Data from'][entry].lower():
return_result.append([{'Phrase': df_part['Data from'][entry],
'Entity Type': 'Phrase'},
{index_of_child: {'Resolution': 'Offshore Leaks Leak', 'Notes': ''}}])
try:
countries = df_part['Linked to'][entry]
if isinstance(countries, str):
if "not identified" not in countries.lower():
countries_list = countries.split(",")
for country in countries_list:
return_result.append([{'Country Name': country,
'Entity Type': 'Country'},
{index_of_child: {'Resolution': 'Linked to', 'Notes': ''}}])
if isinstance(countries, str) and "not identified" not in countries.lower():
countries_list = countries.split(",")
for country in countries_list:
return_result.append([{'Country Name': country,
'Entity Type': 'Country'},
{index_of_child: {'Resolution': 'Linked to', 'Notes': ''}}])
except AttributeError:
continue
try:
countries = df_part['Jurisdiction'][entry]
if isinstance(countries, str):
if "not identified" not in countries.lower():
countries_list = countries.split(",")
for country in countries_list:
return_result.append([{'Country Name': country,
'Entity Type': 'Country'},
{index_of_child: {'Resolution': 'Jurisdiction', 'Notes': ''}}])
if isinstance(countries, str) and "not identified" not in countries.lower():
countries_list = countries.split(",")
for country in countries_list:
return_result.append([{'Country Name': country,
'Entity Type': 'Country'},
{index_of_child: {'Resolution': 'Jurisdiction', 'Notes': ''}}])
except AttributeError:
continue
return return_result

15
Modules/Offshore_Leaks/Offshore_Leaks_Intermediaries.py
View File

@@ -39,7 +39,7 @@ class Offshore_Leaks_Intermediaries:
primary_field = entity[list(entity)[1]].strip()
df_list = []
for batch in range(0, nextHundred, 100):
crafted_url = url + f"search?cat=2&from={batch}&q={primary_field}&utf8=✓"
crafted_url = f"{url}search?cat=2&from={batch}&q={primary_field}&utf8=✓"
try:
r = requests.get(crafted_url)
df_list.append(pd.read_html(r.text)[0])
@@ -68,13 +68,12 @@ class Offshore_Leaks_Intermediaries:
{index_of_child: {'Resolution': 'Offshore Leaks Leak', 'Notes': ''}}])
try:
countries = df_part['Linked to'][entry]
if isinstance(countries, str):
if "not identified" not in countries.lower():
countries_list = countries.split(",")
for country in countries_list:
return_result.append([{'Country Name': country,
'Entity Type': 'Country'},
{index_of_child: {'Resolution': 'Linked to', 'Notes': ''}}])
if isinstance(countries, str) and "not identified" not in countries.lower():
countries_list = countries.split(",")
for country in countries_list:
return_result.append([{'Country Name': country,
'Entity Type': 'Country'},
{index_of_child: {'Resolution': 'Linked to', 'Notes': ''}}])
except AttributeError:
continue
return return_result

3
Modules/Offshore_Leaks/requirements.txt
View File

@@ -1,2 +1,3 @@
requests
pandas
pandas
html5lib

6
Modules/Reddit/RedditSearch.py
View File

@@ -22,7 +22,6 @@ class RedditSearch:
def resolution(self, entityJsonList, parameters):
import requests
import hashlib
from binascii import hexlify
from requests_futures.sessions import FuturesSession
from concurrent.futures import as_completed
@@ -67,7 +66,7 @@ class RedditSearch:
if resolution_name is None:
resolution_name = 'Link ID: ' + value.get('link_id', 'N/A')
else:
resolution_name = 'https://reddit.com' + resolution_name
resolution_name = f'https://reddit.com{resolution_name}'
if submission_endpoint in future.result().url:
return_result.append([{'User Name': value['author'],
@@ -81,8 +80,7 @@ class RedditSearch:
'Notes': ''}}])
comment_resolution = 'Reddit Comment Hash'
comment = hashlib.md5(value.get('body', 'N/A').encode()) # nosec
comment = hexlify(comment.digest()).decode()
comment = hashlib.md5(value.get('body', 'N/A').encode()).hexdigest() # nosec
return_result.append([{'Comment': comment,
'Notes': value.get('body', 'N/A'),
'Entity Type': 'Reddit Comment'},

4
Modules/Shodan/ShodanDomainScan.py
View File

@@ -17,7 +17,6 @@ class ShodanDomainScan:
def resolution(self, entityJsonList, parameters):
import shodan
import hashlib
from binascii import hexlify
return_result = []
api_key = parameters['Shodan API Key'].strip()
@@ -67,8 +66,7 @@ class ShodanDomainScan:
elif dns_type == "TXT":
# Text records could be massive - do not want them breaking the UI
textPrimaryField = hashlib.md5(value.encode()) # nosec
return_result.append([{'Phrase': primary_field + ' TXT Record: ' +
hexlify(textPrimaryField.digest()).decode(),
return_result.append([{'Phrase': f"{primary_field} TXT Record: {textPrimaryField.hexdigest()}",
'Entity Type': 'Phrase',
'Notes': value},
{uid: {'Resolution': 'Shodan Domain TXT records', 'Notes': ''}}])

10
Modules/SocialMedia/SteamGroupMembersChecker.py
View File

@@ -20,6 +20,7 @@ class SteamGroupMembersChecker:
from PySide6.QtGui import QImage
from json import loads
import requests
import contextlib
gamesMembersBaseURL = 'https://steamcommunity.com/games/'
@@ -75,7 +76,7 @@ class SteamGroupMembersChecker:
# Ignore non-steam group URLs.
continue
if groupURL.startswith('https://steamcommunity.com/groups/'):
membersURL = groupURL + '/members'
membersURL = f'{groupURL}/members'
else:
for _ in range(3):
try:
@@ -92,15 +93,12 @@ class SteamGroupMembersChecker:
page.goto(groupURL, wait_until="load", timeout=10000)
except Error:
continue
try:
with contextlib.suppress(Error):
# If we get a warning about age restriction, click the checkbox and view the actual page.
# Click text=Don't warn me again for
page.locator("text=Don't warn me again for").click()
# Click text=View Page
page.locator("text=View Page").click()
except Error:
pass
appPage = BeautifulSoup(page.content(), 'lxml')
appPageDataConfig = appPage.findChild('div', {'id': 'application_config'}).get('data-community')
appPageVanityID = loads(appPageDataConfig)["VANITY_ID"]
@@ -109,7 +107,7 @@ class SteamGroupMembersChecker:
pageCount = 1
urlsFound = []
while len(urlsFound) < maxResults:
pageURL = membersURL + '/?p=' + str(pageCount)
pageURL = f'{membersURL}/?p={str(pageCount)}'
pageResolved = False
for _ in range(3):

2
Modules/SocialMedia/SteamUsernameAliasChecker.py
View File

@@ -36,7 +36,7 @@ class SteamUsernameAliasChecker:
else:
continue
aliasURL = entityURL + '/ajaxaliases'
aliasURL = f'{entityURL}/ajaxaliases'
aliases = requests.get(aliasURL).json()
newHandleIndex = len(returnResults)

16
Modules/SocialMedia/SteamUsernameChecker.py
View File

@@ -41,14 +41,12 @@ class SteamUsernameChecker:
for entity in entityJsonList:
uid = entity['uid']
entityType = entity['Entity Type']
if entityType == 'Phrase':
if entityType in ['Person', 'Politically Exposed Person']:
groupName = entity['Full Name']
elif entityType == 'Phrase':
groupName = entity['Phrase']
elif entityType == 'Social Media Handle':
groupName = entity['User Name']
elif entityType == 'Person':
groupName = entity['Full Name']
elif entityType == 'Politically Exposed Person':
groupName = entity['Full Name']
else:
continue
@@ -87,10 +85,10 @@ class SteamUsernameChecker:
else:
userID = userURL.split('https://steamcommunity.com/profiles/', 1)[1]
if userURL == 'https://steamcommunity.com/profiles/76561198067124199':
# Placeholder user by Steam.
continue
elif userURL in urlsFound:
if (
userURL == 'https://steamcommunity.com/profiles/76561198067124199' # Steam Placeholder user
or userURL in urlsFound
):
continue
else:
urlsFound.append(userURL)

39
Modules/SocialMedia/Twayback.py
View File

@@ -17,6 +17,7 @@ class Twayback:
def resolution(self, entityJsonList, parameters):
import requests
import contextlib
import bs4
import re
from requests_futures.sessions import FuturesSession
@@ -45,8 +46,15 @@ class Twayback:
cdx_page_text.splitlines()}
with FuturesSession(max_workers=10) as session:
for twitter_url in tweet_id_and_url_dict:
futures.append(session.get(twitter_url, headers=headers, timeout=30, allow_redirects=False))
futures.extend(
session.get(
twitter_url,
headers=headers,
timeout=30,
allow_redirects=False,
)
for twitter_url in tweet_id_and_url_dict
)
missing_tweets = {}
for future in as_completed(futures):
@@ -57,19 +65,20 @@ class Twayback:
split_fin = re.split(r'\D', split_once)[0]
missing_tweets[page_response.url] = split_fin
wayback_url_list = {}
for url, number in missing_tweets.items():
wayback_url_list[number] = f"https://web.archive.org/web/{number}/{url}"
wayback_url_list = {
number: f"https://web.archive.org/web/{number}/{url}"
for url, number in missing_tweets.items()
}
deleted_tweets_futures_retry = []
futures_list = []
regex = re.compile('.*TweetTextSize TweetTextSize--jumbo.*')
with FuturesSession(max_workers=10) as session:
for number, url in wayback_url_list.items():
futures_list.append(session.get(url, headers=headers, timeout=30))
futures_list.extend(
session.get(url, headers=headers, timeout=30)
for url in wayback_url_list.values()
)
for future in as_completed(futures_list):
result = None
try:
@@ -86,23 +95,17 @@ class Twayback:
deleted_tweets_futures_retry.append(result.url)
# Second try, if things go wrong.
if len(deleted_tweets_futures_retry) > 0:
if deleted_tweets_futures_retry:
sleep(10)
futures_list = []
with FuturesSession(max_workers=10) as session:
for url in deleted_tweets_futures_retry:
futures_list.append(session.get(url))
futures_list.extend(session.get(url) for url in deleted_tweets_futures_retry)
for future in as_completed(futures_list):
try:
with contextlib.suppress(AttributeError, ConnectionError):
result = future.result()
tweet = bs4.BeautifulSoup(result.content, "lxml").find("p", {"class": regex}).getText()
returnResults.append([{'URL': result.url,
'Entity Type': 'Website',
'Notes': tweet},
{uid: {'Resolution': 'Deleted Tweet'}}])
except AttributeError:
pass
except ConnectionError:
pass
return returnResults

99
Modules/SocialMedia/TwitterUser.py
View File

@@ -130,7 +130,7 @@ class TwitterUser:
'Entity Type': 'Website'},
{selfItemIndex: {'Resolution': 'Picture in Tweet'}}])
if tweetItem.coordinates:
placeName = 'Tweet Location ' + str(tweetItem.id)
placeName = f'Tweet Location: {str(tweetItem.id)}'
if tweetItem.place:
if tweetItem.place.fullName:
placeName = str(tweetItem.place.fullName)
@@ -166,25 +166,42 @@ class TwitterUser:
childImageBuffer.close()
except Exception:
childIconByteArrayFin = None
returnResults.append([{'Twitter Handle': '@' + tweetItem.inReplyToUser.username,
'User ID': str(tweetItem.inReplyToUser.id),
'User URL': tweetItem.inReplyToUser.url,
'Verified': str(tweetItem.inReplyToUser.verified),
'Display Name': str(tweetItem.inReplyToUser.displayname),
'Location': str(tweetItem.inReplyToUser.location),
'Description': tweetItem.inReplyToUser.description,
'Protected': str(tweetItem.inReplyToUser.protected),
'Followers': str(tweetItem.inReplyToUser.followersCount),
'Following': str(tweetItem.inReplyToUser.friendsCount),
'Statuses': str(tweetItem.inReplyToUser.statusesCount),
'Favourites': str(tweetItem.inReplyToUser.favouritesCount),
'Listed': str(tweetItem.inReplyToUser.listedCount),
'Media': str(tweetItem.inReplyToUser.mediaCount),
'Entity Type': 'Twitter User',
'Icon': childIconByteArrayFin, # If None -> default twitter user pic
'Date Created': None if tweetItem.inReplyToUser.created is None else
tweetItem.inReplyToUser.created.isoformat()},
{selfItemIndex: {'Resolution': 'Replying to Twitter User'}}])
returnResults.append(
[
{
'Twitter Handle': f'@{tweetItem.inReplyToUser.username}',
'User ID': str(tweetItem.inReplyToUser.id),
'User URL': tweetItem.inReplyToUser.url,
'Verified': str(tweetItem.inReplyToUser.verified),
'Display Name': str(
tweetItem.inReplyToUser.displayname
),
'Location': str(tweetItem.inReplyToUser.location),
'Description': tweetItem.inReplyToUser.description,
'Protected': str(tweetItem.inReplyToUser.protected),
'Followers': str(
tweetItem.inReplyToUser.followersCount
),
'Following': str(tweetItem.inReplyToUser.friendsCount),
'Statuses': str(tweetItem.inReplyToUser.statusesCount),
'Favourites': str(
tweetItem.inReplyToUser.favouritesCount
),
'Listed': str(tweetItem.inReplyToUser.listedCount),
'Media': str(tweetItem.inReplyToUser.mediaCount),
'Entity Type': 'Twitter User',
'Icon': childIconByteArrayFin, # If None -> default twitter user pic
'Date Created': None
if tweetItem.inReplyToUser.created is None
else tweetItem.inReplyToUser.created.isoformat(),
},
{
selfItemIndex: {
'Resolution': 'Replying to Twitter User'
}
},
]
)
if tweetItem.quotedTweet:
parseTweet(tweetItem.quotedTweet, selfItemIndex)
@@ -225,24 +242,30 @@ class TwitterUser:
imageBuffer.close()
except Exception:
iconByteArrayFin = None
returnResults.append([{'Twitter Handle': '@' + item.user.username,
'User ID': str(item.user.id),
'User URL': item.user.url,
'Verified': str(item.user.verified),
'Display Name': str(item.user.displayname),
'Location': str(item.user.location),
'Description': item.user.renderedDescription,
'Protected': str(item.user.protected),
'Followers': str(item.user.followersCount),
'Following': str(item.user.friendsCount),
'Statuses': str(item.user.statusesCount),
'Favourites': str(item.user.favouritesCount),
'Listed': str(item.user.listedCount),
'Media': str(item.user.mediaCount),
'Entity Type': 'Twitter User',
'Icon': iconByteArrayFin, # If None -> default twitter user pic
'Date Created': item.user.created.isoformat()},
{uid: {'Resolution': 'Twitter User'}}])
returnResults.append(
[
{
'Twitter Handle': f'@{item.user.username}',
'User ID': str(item.user.id),
'User URL': item.user.url,
'Verified': str(item.user.verified),
'Display Name': str(item.user.displayname),
'Location': str(item.user.location),
'Description': item.user.renderedDescription,
'Protected': str(item.user.protected),
'Followers': str(item.user.followersCount),
'Following': str(item.user.friendsCount),
'Statuses': str(item.user.statusesCount),
'Favourites': str(item.user.favouritesCount),
'Listed': str(item.user.listedCount),
'Media': str(item.user.mediaCount),
'Entity Type': 'Twitter User',
'Icon': iconByteArrayFin, # If None -> default twitter user pic
'Date Created': item.user.created.isoformat(),
},
{uid: {'Resolution': 'Twitter User'}},
]
)
parseTweet(item, childIndex)

138
Modules/URLScan/URLScan.py
View File

@@ -13,19 +13,8 @@ class URLScan:
'status code 429.',
'type': 'String',
'global': True,
'value': ''},
'ip results': {'description': 'Enter the number of IP Addresses you want to be returned',
'type': 'String',
'value': '0'},
'domain results': {'description': 'Enter the number of Domains you want to be returned',
'type': 'String',
'value': '0'},
'hash results': {'description': 'Enter the number of Hashes you want to be returned',
'type': 'String',
'value': '0'},
'url results': {'description': 'Enter the number of Urls you want to be returned',
'type': 'String',
'value': '0'}}
'value': ''}
}
def resolution(self, entityJsonList, parameters):
import requests
@@ -33,18 +22,7 @@ class URLScan:
import time
return_result = []
api_key = parameters['api key']
hash_results = parameters['hash results']
domain_results = parameters['domain results']
ip_results = parameters['ip results']
url_results = parameters['url results']
try:
hash_results = int(hash_results)
domain_results = int(domain_results)
ip_results = int(ip_results)
url_results = int(url_results)
except ValueError:
"The value for at least 1 of the parameter fields is not a valid integer."
api_key = parameters['URLScan API Key']
for entity in entityJsonList:
uid = entity['uid']
primary_field = entity[list(entity)[1]].strip()
@@ -52,71 +30,49 @@ class URLScan:
data = {"url": primary_field, "visibility": "public"}
response = requests.post('https://urlscan.io/api/v1/scan/', headers=headers, data=json.dumps(data))
if response.status_code == 429:
return_result = []
return "The API Key provided is Invalid or the you are sending requests above the rate limit"
else:
while response.status_code == 404:
time.sleep(0.5)
try:
response = requests.post('https://urlscan.io/api/v1/scan/', headers=headers,
data=json.dumps(data))
except requests.exceptions.ConnectionError:
return "Please check your internet connection"
response = response.json()
response_uuid = response['uuid']
result_response = requests.get(f"https://urlscan.io/api/v1/result/{response_uuid}")
while result_response.status_code == 404 or "message" in result_response:
time.sleep(0.5)
try:
result_response = requests.get(f"https://urlscan.io/api/v1/result/{response_uuid}")
except requests.exceptions.ConnectionError:
return "Please check your internet connection"
result_response = result_response.json()
return_result.append([{'URL': f"https://urlscan.io/api/v1/result/{response_uuid}",
'Entity Type': 'Website',
'Notes': "Request ID: " + str(
result_response['data']['requests'][0]['request']['requestId'])},
{uid: {'Resolution': 'URLScan ID', 'Notes': ''}}])
for hash in result_response['lists']['hashes']:
if hash_results == 0:
return_result.append([{'Hash Value': hash,
'Hash Algorithm': "SHA256",
'Entity Type': 'Hash'},
{uid: {'Resolution': 'URLScan SHA25 Hash256', 'Notes': ''}}])
else:
for i in range(hash_results):
return_result.append([{'Hash Value': hash,
'Hash Algorithm': "SHA256",
'Entity Type': 'Hash'},
{uid: {'Resolution': 'URLScan SHA256 Hash', 'Notes': ''}}])
for ip in result_response['lists']['ips']:
if ip_results == 0:
return_result.append([{'IP Address': ip,
'Entity Type': 'IP Address'},
{uid: {'Resolution': 'URLScan IP Address', 'Notes': ''}}])
else:
for i in range(ip_results):
return_result.append([{'IP Address': ip,
'Entity Type': 'IP Address'},
{uid: {'Resolution': 'URLScan IP Address', 'Notes': ''}}])
for domain in result_response['lists']['domains']:
if domain_results == 0:
return_result.append([{'Domain Name': domain,
'Entity Type': 'Domain'},
{uid: {'Resolution': 'URLScan Domains', 'Notes': ''}}])
else:
for i in range(domain_results):
return_result.append([{'Domain Name': domain,
'Entity Type': 'Domain'},
{uid: {'Resolution': 'URLScan Domains', 'Notes': ''}}])
for url in result_response['lists']['urls']:
if url_results == 0:
return_result.append([{'URL': url,
'Entity Type': 'Website'},
{uid: {'Resolution': 'URLScan URLs', 'Notes': ''}}])
else:
for i in range(url_results):
return_result.append([{'URL': url,
'Entity Type': 'Website'},
{uid: {'Resolution': 'URLScan URLs', 'Notes': ''}}])
while response.status_code == 404:
time.sleep(0.5)
try:
response = requests.post('https://urlscan.io/api/v1/scan/', headers=headers,
data=json.dumps(data))
except requests.exceptions.ConnectionError:
return "Please check your internet connection"
response = response.json()
response_uuid = response['uuid']
result_response = requests.get(f"https://urlscan.io/api/v1/result/{response_uuid}")
while result_response.status_code == 404 or "message" in result_response:
time.sleep(0.5)
try:
result_response = requests.get(f"https://urlscan.io/api/v1/result/{response_uuid}")
except requests.exceptions.ConnectionError:
return "Please check your internet connection"
result_response = result_response.json()
return_result.append([{'URL': f"https://urlscan.io/api/v1/result/{response_uuid}",
'Entity Type': 'Website',
'Notes': "Request ID: " + str(
result_response['data']['requests'][0]['request']['requestId'])},
{uid: {'Resolution': 'URLScan ID', 'Notes': ''}}])
return_result.extend(
[{'Hash Value': hash_result,
'Hash Algorithm': "SHA256",
'Entity Type': 'Hash'},
{uid: {'Resolution': 'URLScan SHA25 Hash256', 'Notes': ''}}]
for hash_result in result_response['lists']['hashes']
)
return_result.extend(
[{'IP Address': ip, 'Entity Type': 'IP Address'},
{uid: {'Resolution': 'URLScan IP Address', 'Notes': ''}}]
for ip in result_response['lists']['ips']
)
return_result.extend(
[{'Domain Name': domain, 'Entity Type': 'Domain'},
{uid: {'Resolution': 'URLScan Domains', 'Notes': ''}}]
for domain in result_response['lists']['domains']
)
return_result.extend(
[{'URL': url, 'Entity Type': 'Website'},
{uid: {'Resolution': 'URLScan URLs', 'Notes': ''}}]
for url in result_response['lists']['urls']
)
return return_result

10
Modules/Username_Crosscheck/PinterestUsersSearch.py
View File

@@ -11,13 +11,14 @@ class PinterestUsersSearch:
parameters = {}
def resolution(self, entityJsonList, parameters):
import requests
import contextlib
from playwright.sync_api import sync_playwright, Error
from time import sleep
from bs4 import BeautifulSoup
from PySide6.QtCore import QByteArray, QBuffer, QIODevice, QSize
from PySide6.QtGui import QImage
from random import random
import requests
baseURL = "https://www.pinterest.com/search/users/?q="
@@ -38,7 +39,7 @@ class PinterestUsersSearch:
uid = entity['uid']
primaryField = entity[list(entity)[1]]
try:
with contextlib.suppress(Error):
page.goto(baseURL + primaryField, wait_until="networkidle", timeout=60000)
# Wait until everything is 100% loaded, just in case.
sleep(10)
@@ -48,7 +49,7 @@ class PinterestUsersSearch:
for listElement in listElements:
username = listElement.find('a')['href'][1:-1]
href = "https://www.pinterest.com/" + username + "/"
href = f"https://www.pinterest.com/{username}/"
imgElement = listElement.find('img')
textStrings = listElement.findAll(text=True)
prettyName = textStrings[0]
@@ -77,9 +78,6 @@ class PinterestUsersSearch:
'Icon': childIconByteArrayFin},
{uid: {'Resolution': 'Account Found',
'Notes': ''}}])
except Error:
pass
# Wait a bit before moving on to the next query.
sleep(2 + random())

52
Modules/Username_Crosscheck/Social_Analyzer_Detector.py
View File

@@ -49,33 +49,33 @@ class Social_Analyzer_Detector:
headers=headers)
if firstResponse.status_code >= 300:
return False
else:
modifiedUsername = "".join(random.choices( # nosec
string.ascii_uppercase + string.digits, k=32))
usernameRegex = re.compile(social_field, re.IGNORECASE)
r = requests.get(original_url, timeout=30, verify=False, headers=headers) # nosec
originalContent = r.text
if len(originalUsernameRegex.findall(originalContent)) == 0:
# False Positive
return False
modified_url = original_url.replace(social_field, modifiedUsername)
r = requests.get(modified_url, timeout=30, verify=False, headers=headers) # nosec
modifiedUsernameContent = r.text
for regexMatch in commentsRegex.findall(originalContent):
originalContent = originalContent.replace(regexMatch, '')
for regexMatch in commentsRegex.findall(modifiedUsernameContent):
modifiedUsernameContent = modifiedUsernameContent.replace(regexMatch, '')
for regexMatch in usernameRegex.findall(originalContent):
originalContent = originalContent.replace(regexMatch, modifiedUsername)
if modifiedUsernameContent == originalContent:
# False positive
return False
else:
return [{'URL': original_url,
'Entity Type': 'Website'},
{uid: {'Resolution': 'Social Analyzer Report', 'Notes': ''}}]
modifiedUsername = "".join(random.choices( # nosec
string.ascii_uppercase + string.digits, k=32))
usernameRegex = re.compile(social_field, re.IGNORECASE)
r = requests.get(original_url, timeout=30, verify=False, headers=headers) # nosec
originalContent = r.text
if len(originalUsernameRegex.findall(originalContent)) == 0:
# False Positive
return False
modified_url = original_url.replace(social_field, modifiedUsername)
r = requests.get(modified_url, timeout=30, verify=False, headers=headers) # nosec
modifiedUsernameContent = r.text
for regexMatch in commentsRegex.findall(originalContent):
originalContent = originalContent.replace(regexMatch, '')
for regexMatch in commentsRegex.findall(modifiedUsernameContent):
modifiedUsernameContent = modifiedUsernameContent.replace(regexMatch, '')
for regexMatch in usernameRegex.findall(originalContent):
originalContent = originalContent.replace(regexMatch, modifiedUsername)
return (
False
if modifiedUsernameContent == originalContent # False positive
else [{'URL': original_url,
'Entity Type': 'Website'},
{uid: {'Resolution': 'Social Analyzer Report',
'Notes': ''}}]
)
except (ConnectionError, RequestException) as error:
return "Connection error: " + str(error)
return f"Connection error: {str(error)}"
return_result = []
for entity in entityJsonList:

20
Modules/Username_Crosscheck/Whats_My_Name.py
View File

@@ -83,18 +83,16 @@ class Whats_My_Name:
{uid: {'Resolution': 'Whats My Name Account Match',
'Notes': ''}}])
break
elif post_body := site.get('post_body'):
futures[session.post(original_uri, data=post_body, headers=headers,
timeout=10, allow_redirects=False)] = \
(uid, account_existence_code, account_existence_string,
account_missing_string, account_missing_code)
else:
post_body = site.get('post_body')
if post_body:
futures[session.post(original_uri, data=post_body, headers=headers,
timeout=10, allow_redirects=False)] = \
(uid, account_existence_code, account_existence_string,
account_missing_string, account_missing_code)
else:
futures[session.get(original_uri, headers=headers,
timeout=10, allow_redirects=False)] = \
(uid, account_existence_code, account_existence_string,
account_missing_string, account_missing_code)
futures[session.get(original_uri, headers=headers,
timeout=10, allow_redirects=False)] = \
(uid, account_existence_code, account_existence_string,
account_missing_string, account_missing_code)
for future in as_completed(futures):
parent_uid = futures[future][0]
account_existence_code = futures[future][1]

80
Modules/VirusTotal/VirusTotal_Domain.py
View File

@@ -7,8 +7,8 @@ class VirusTotal_Domain:
description = "Find information about a domain using VirusTotal.com"
originTypes = {"Domain"}
resultTypes = {'Country', 'Domain', 'Company', 'Email Address', 'Phrase'}
parameters = {'VirusTotal API Key': {'description': 'Enter your api key under your profile after'
' signing up on https://virustotal.com. '
parameters = {'VirusTotal API Key': {'description': 'Enter your api key under your profile after '
'signing up on https://virustotal.com. '
'Free usage of the API is limited to 500 requests per day '
'with a rate of 4 per minute.',
'type': 'String',
@@ -18,7 +18,6 @@ class VirusTotal_Domain:
def resolution(self, entityJsonList, parameters):
import json
import hashlib
from binascii import hexlify
from vtapi3 import VirusTotalAPIDomains, VirusTotalAPIError
return_result = []
@@ -34,9 +33,9 @@ class VirusTotal_Domain:
else:
if vt_api_domains.get_last_http_error() != vt_api_domains.HTTP_OK:
return f'HTTP Error [{vt_api_domains.get_last_http_error()}]'
results = json.loads(results)
analysis_stats = results['data']['attributes']['last_analysis_stats']
return_result.append([{'Phrase': f"VirusTotal Scan Results for {primary_field}",
'VT Malicious Votes': analysis_stats['malicious'],
'VT Suspicious Votes': analysis_stats['suspicious'],
@@ -46,15 +45,12 @@ class VirusTotal_Domain:
'Entity Type': 'Phrase'
},
{uid: {'Resolution': 'VirusTotal Domain Scan', 'Notes': ''}}])
for result in results['data']['attributes']['last_dns_records']:
index_of_child = len(return_result)
dns_type = result['type']
value = result['value']
if dns_type == "MX":
return_result.append([{'Domain Name': value,
'Entity Type': 'Domain'},
{uid: {'Resolution': 'VirusTotal Domain MX records', 'Notes': ''}}])
elif dns_type == "A":
if dns_type == "A":
return_result.append([{'IP Address': value,
'Entity Type': 'IP Address'},
{index_of_child: {'Resolution': 'VirusTotal Domain A records',
@@ -63,26 +59,37 @@ class VirusTotal_Domain:
return_result.append([{'IPv6 Address': value,
'Entity Type': 'IPv6 Address'},
{uid: {'Resolution': 'VirusTotal Domain AAAA records', 'Notes': ''}}])
elif dns_type == "TXT":
# Text records could be massive - do not want them breaking the UI
textPrimaryField = hashlib.md5(value.encode()) # nosec
return_result.append([{'Phrase': primary_field + ' TXT Record: ' +
hexlify(textPrimaryField.digest()).decode(),
'Entity Type': 'Phrase',
'Notes': value},
{uid: {'Resolution': 'VirusTotal Domain TXT records', 'Notes': ''}}])
elif dns_type == "SOA":
return_result.append([{'Domain Name': result['rname'],
'Entity Type': 'Domain'},
{uid: {'Resolution': 'VirusTotal Domain SOA records', 'Notes': ''}}])
elif dns_type == "MX":
return_result.append([{'Domain Name': value,
'Entity Type': 'Domain'},
{index_of_child: {'Resolution': 'Start Of Authority DNS',
'Notes': ''}}])
{uid: {'Resolution': 'VirusTotal Domain MX records', 'Notes': ''}}])
elif dns_type == "NS":
return_result.append([{'Domain Name': value,
'Entity Type': 'Domain'},
{uid: {'Resolution': 'VirusTotal NS records', 'Notes': ''}}])
elif dns_type == "SOA":
return_result.extend(
(
[{'Domain Name': result['rname'],
'Entity Type': 'Domain'},
{uid: {'Resolution': 'VirusTotal Domain SOA records',
'Notes': ''}}],
[{'Domain Name': value,
'Entity Type': 'Domain'},
{index_of_child: {'Resolution': 'Start Of Authority DNS',
'Notes': ''}}]
)
)
elif dns_type == "TXT":
# Text records could be massive - do not want them breaking the UI
textPrimaryField = hashlib.md5(value.encode()).hexdigest() # nosec
return_result.append(
[{'Phrase': f'{primary_field} TXT Record: {textPrimaryField}',
'Entity Type': 'Phrase',
'Notes': value},
{uid: {'Resolution': 'VirusTotal Domain TXT records',
'Notes': ''}}]
)
fields = results['data']['attributes']['whois'].split("\n")
for field in fields:
field = field.split(":")
@@ -94,28 +101,33 @@ class VirusTotal_Domain:
return_result.append([{'Country Name': field[1],
'Entity Type': 'Country'},
{uid: {'Resolution': 'VirusTotal Domain Company Country',
'notes': ''}}])
'Notes': ''}}])
elif field[0] == "Registrar":
return_result.append([{'Company Name': field[1],
'Entity Type': 'Company'},
{uid: {'Resolution': 'VirusTotal Domain Registrar', 'notes': ''}}])
{uid: {'Resolution': 'VirusTotal Domain Registrar', 'Notes': ''}}])
elif field[0] == "Registrar Country":
return_result.append([{'Country Name': field[1],
'Entity Type': 'Country'},
{uid: {'Resolution': 'VirusTotal Domain Registrar Country',
'notes': ''}}])
'Notes': ''}}])
elif field[0] == "Registry Domain ID":
return_result.append([{'Phrase': field[0] + ":" + field[1],
'Entity Type': 'Phrase'},
{uid: {'Resolution': 'VirusTotal Domain Registry ID', 'notes': ''}}])
return_result.append(
[{'Phrase': f"{field[0]}:{field[1]}",
'Entity Type': 'Phrase'},
{uid: {'Resolution': 'VirusTotal Domain Registry ID',
'Notes': ''}}]
)
elif field[0] == "Registrar IANA ID":
return_result.append([{'Phrase': field[0] + ":" + field[1],
'Entity Type': 'Phrase'},
{uid: {'Resolution': 'VirusTotal Domain Registrar IANA ID',
'notes': ''}}])
return_result.append(
[{'Phrase': f"{field[0]}:{field[1]}",
'Entity Type': 'Phrase'},
{uid: {'Resolution': 'VirusTotal Domain Registrar IANA ID',
'Notes': ''}}]
)
elif field[0] == "Registrar Abuse Contact Email":
return_result.append([{'Email Address': field[1],
'Entity Type': 'Email Address'},
{uid: {'Resolution': 'VirusTotal Domain Email Address',
'notes': ''}}])
'Notes': ''}}])
return return_result