WEBDIENSTE/theHarvester
1
0
Fork 0
mirror of https://github.com/laramies/theHarvester.git synced 2026-09-12 12:47:41 +02:00
Code Issues Packages Projects Releases Wiki Activity

Fixes #2229

Browse Source
This commit is contained in:
L1ghtn1ng
2026-02-20 13:35:40 +00:00
parent 419291a3e7
commit 06c2fbd955
2 changed files with 344 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/discovery/test_criminalip.py
+96
View File
@@ -0,0 +1,96 @@
#!/usr/bin/env python3
# coding=utf-8
import pytest
from theHarvester.discovery import criminalip
@pytest.mark.asyncio
async def test_parser_handles_missing_legacy_fields(monkeypatch) -> None:
monkeypatch.setattr(criminalip.Core, 'criminalip_key', lambda: 'test-key')
search = criminalip.SearchCriminalIP('example.com')
payload = {
'data': {
'certificates': [{'subject': 'www.example.com'}],
'connected_domain_subdomain': [{'main_domain': {'domain': 'example.com'}, 'subdomains': [{'domain': 'api.example.com'}]}],
'connected_ip': [{'ip': '93.184.216.34'}],
'connected_ip_info': [
{
'asn': '15133',
'ip': '93.184.216.34',
'domain_list': [{'domain': 'mail.example.com'}],
}
],
'cookies': [{'domain': '.portal.example.com'}],
'dns_record': {
'dns_record_type_a': {'ipv4': [{'ip': '93.184.216.34'}], 'ipv6': []},
'dns_record_type_ns': ['ns1.example.com.'],
},
'html_page_link_domains': [{'domain': 'www.iana.org', 'mapped_ips': [{'ip': '192.0.33.8'}]}],
'links': [{'url': 'https://docs.example.com/guide'}],
'mapped_ip': [{'ip': '203.0.113.10'}],
'network_logs': {
'data': [{'url': 'https://cdn.example.com/script.js', 'as_number': '64500', 'ip_port': '198.51.100.10:443'}]
},
'page_redirections': [[{'url': 'https://login.example.com'}]],
'subdomains': [{'subdomain_name': 'blog.example.com'}],
}
}
await search.parser(payload)
hostnames = await search.get_hostnames()
ips = await search.get_ips()
asns = await search.get_asns()
assert {'api.example.com', 'blog.example.com', 'cdn.example.com', 'docs.example.com', 'login.example.com'}.issubset(hostnames)
assert {'93.184.216.34', '198.51.100.10', '203.0.113.10'}.issubset(ips)
assert {'15133', '64500'}.issubset(asns)
@pytest.mark.asyncio
async def test_do_search_uses_v2_report_endpoint(monkeypatch) -> None:
monkeypatch.setattr(criminalip.Core, 'criminalip_key', lambda: 'test-key')
monkeypatch.setattr(criminalip.Core, 'get_user_agent', lambda: 'test-agent')
called_urls = []
async def fake_post_fetch(url, **kwargs):
assert url == 'https://api.criminalip.io/v1/domain/scan'
return {'status': 200, 'data': {'scan_id': 12345}}
async def fake_fetch_all(urls, **kwargs):
called_urls.append(urls[0])
if '/v1/domain/status/' in urls[0]:
return [{'status': 200, 'data': {'scan_percentage': 100}}]
if '/v2/domain/report/' in urls[0]:
return [
{
'status': 200,
'data': {
'certificates': [],
'connected_domain_subdomain': [],
'connected_ip': [],
'connected_ip_info': [],
'cookies': [],
'dns_record': {},
'html_page_link_domains': [],
'links': [],
'mapped_ip': [],
'network_logs': {'data': []},
'page_redirections': [],
'subdomains': [],
},
}
]
return [{'status': 500}]
monkeypatch.setattr(criminalip.AsyncFetcher, 'post_fetch', fake_post_fetch)
monkeypatch.setattr(criminalip.AsyncFetcher, 'fetch_all', fake_fetch_all)
search = criminalip.SearchCriminalIP('example.com')
await search.process()
assert any('/v2/domain/report/12345' in url for url in called_urls)
assert all('/v1/domain/report/' not in url for url in called_urls)
theHarvester/discovery/criminalip.py
+248 -133
View File
@@ -7,7 +7,7 @@ from theHarvester.lib.core import AsyncFetcher, Core
class SearchCriminalIP:
def __init__(self, word) -> None:
self.word = word
self.word = word.lower().strip()
self.totalhosts: set = set()
self.totalips: set = set()
self.asns: set = set()
@@ -16,10 +16,76 @@ class SearchCriminalIP:
raise MissingKey('criminalip')
self.proxy = False
def _normalize_host(self, hostname: str | None) -> str | None:
if not isinstance(hostname, str):
return None
cleaned = hostname.strip().lower().rstrip('.')
if not cleaned:
return None
if cleaned.startswith('*.'):
cleaned = cleaned[2:]
if ':' in cleaned and cleaned.count(':') == 1:
host, port = cleaned.rsplit(':', 1)
if port.isdigit():
cleaned = host
if cleaned == self.word or cleaned.endswith('.' + self.word):
return cleaned
return None
def _add_host(self, hostname: str | None, include_root: bool = True) -> bool:
normalized = self._normalize_host(hostname)
if normalized is None:
return False
if include_root is False and normalized == self.word:
return False
self.totalhosts.add(normalized)
return True
def _add_host_from_url(self, url: str | None) -> None:
if not isinstance(url, str) or not url:
return
parsed_url = urlparse(url)
netloc = parsed_url.netloc
if not netloc:
return
if '@' in netloc:
netloc = netloc.rsplit('@', 1)[1]
self._add_host(netloc)
def _add_ip(self, ip: str | None) -> None:
if isinstance(ip, str) and ip.strip():
self.totalips.add(ip.strip())
def _add_asn(self, asn: str | int | None) -> None:
if asn is None:
return
asn_value = str(asn).strip()
if asn_value:
self.asns.add(asn_value)
def _collect_hosts_from_value(self, value) -> None:
if isinstance(value, str):
self._add_host(value.split()[0].rstrip('.'))
return
if isinstance(value, list):
for item in value:
self._collect_hosts_from_value(item)
return
if isinstance(value, dict):
self._add_host(value.get('domain'))
self._add_host(value.get('hostname'))
self._add_host(value.get('subdomain_name'))
for nested_value in value.values():
self._collect_hosts_from_value(nested_value)
async def do_search(self) -> None:
# https://www.criminalip.io/developer/api/post-domain-scan
# https://www.criminalip.io/developer/api/get-domain-status-id
# https://www.criminalip.io/developer/api/get-domain-report-id
# https://www.criminalip.io/developer/api/get-v2-domain-report-id
url = 'https://api.criminalip.io/v1/domain/scan'
data = f'{{"query": "{self.word}"}}'
# print(f'Current key: {self.key}')
@@ -34,169 +100,218 @@ class SearchCriminalIP:
# print(f'My response: {response}')
# Expected response format:
# {'data': {'scan_id': scan_id}, 'message': 'api success', 'status': 200}
if 'status' in response.keys():
status = response['status']
if status != 200:
print(f'An error has occurred searching criminalip dumping response: {response}')
else:
scan_id = response['data']['scan_id']
scan_percentage = 0
counter = 0
while scan_percentage != 100:
status_url = f'https://api.criminalip.io/v1/domain/status/{scan_id}'
status_response = await AsyncFetcher.fetch_all(
[status_url],
json=True,
headers={'User-Agent': user_agent, 'x-api-key': f'{self.key}'},
proxy=self.proxy,
)
status = status_response[0]
# print(f'Status response: {status}')
# Expected format:
# {"data": {"scan_percentage": 100}, "message": "api success", "status": 200}
scan_percentage = status['data']['scan_percentage']
if scan_percentage == 100:
break
if scan_percentage == -2:
print(f'CriminalIP failed to scan: {self.word} does not exist, verify manually')
print(f'Dumping data: scan_response: {response} status_response: {status}')
return
if scan_percentage == -1:
print(f'CriminalIP scan failed dumping data: scan_response: {response} status_response: {status}')
return
# Wait for scan to finish
if counter >= 5:
await asyncio.sleep(20 * get_delay())
else:
await asyncio.sleep(10 * get_delay())
counter += 1
if counter == 10:
print(
'Ten iterations have occurred in CriminalIP waiting for scan to finish, returning to prevent infinite loop.'
)
print(
f'Verify results manually on CriminalIP dumping data: scan_response: {response} status_response: {status}'
)
return
if not isinstance(response, dict):
print(f'An error has occurred searching criminalip dumping response: {response}')
return
if response.get('status') != 200:
print(f'An error has occurred searching criminalip dumping response: {response}')
return
report_url = f'https://api.criminalip.io/v1/domain/report/{scan_id}'
scan_response = await AsyncFetcher.fetch_all(
[report_url],
json=True,
headers={'User-Agent': user_agent, 'x-api-key': f'{self.key}'},
proxy=self.proxy,
)
scan = scan_response[0]
# json_formatted_str = json.dumps(scan, indent=2)
# print(json_formatted_str)
try:
await self.parser(scan)
except Exception as e:
print(f'An exception occurred while parsing criminalip result: {e}')
print('Dumping json: ')
print(scan)
scan_id = response.get('data', {}).get('scan_id')
if scan_id is None:
print(f'CriminalIP did not return a scan_id, dumping response: {response}')
return
scan_percentage = 0
counter = 0
status = {}
while scan_percentage != 100:
status_url = f'https://api.criminalip.io/v1/domain/status/{scan_id}'
status_response = await AsyncFetcher.fetch_all(
[status_url],
json=True,
headers={'User-Agent': user_agent, 'x-api-key': f'{self.key}'},
proxy=self.proxy,
)
status = status_response[0] if isinstance(status_response, list) and len(status_response) > 0 else {}
if not isinstance(status, dict):
print(f'CriminalIP status response is malformed dumping data: {status_response}')
return
if status.get('status') != 200:
print(f'CriminalIP status check failed dumping data: status_response: {status}')
return
# Expected format:
# {"data": {"scan_percentage": 100}, "message": "api success", "status": 200}
scan_percentage = status.get('data', {}).get('scan_percentage')
if scan_percentage is None:
print(f'CriminalIP status did not include scan_percentage dumping data: {status}')
return
if scan_percentage == 100:
break
if scan_percentage == -2:
print(f'CriminalIP failed to scan: {self.word} does not exist, verify manually')
print(f'Dumping data: scan_response: {response} status_response: {status}')
return
if scan_percentage == -1:
print(f'CriminalIP scan failed dumping data: scan_response: {response} status_response: {status}')
return
# Wait for scan to finish
if counter >= 5:
await asyncio.sleep(20 * get_delay())
else:
await asyncio.sleep(10 * get_delay())
counter += 1
if counter == 10:
print('Ten iterations have occurred in CriminalIP waiting for scan to finish, returning to prevent infinite loop.')
print(f'Verify results manually on CriminalIP dumping data: scan_response: {response} status_response: {status}')
return
report_url = f'https://api.criminalip.io/v2/domain/report/{scan_id}'
scan_response = await AsyncFetcher.fetch_all(
[report_url],
json=True,
headers={'User-Agent': user_agent, 'x-api-key': f'{self.key}'},
proxy=self.proxy,
)
scan = scan_response[0] if isinstance(scan_response, list) and len(scan_response) > 0 else {}
if not isinstance(scan, dict):
print(f'CriminalIP report response is malformed dumping data: {scan_response}')
return
if scan.get('status') != 200:
print(f'CriminalIP report request failed dumping data: {scan}')
return
# json_formatted_str = json.dumps(scan, indent=2)
# print(json_formatted_str)
try:
await self.parser(scan)
except Exception as e:
print(f'An exception occurred while parsing criminalip result: {e}')
print('Dumping json: ')
print(scan)
async def parser(self, jlines):
# TODO when new scope field is added to parse lines for potential new scope!
# TODO map as_name to asn for asn data
# TODO determine if worth storing interesting urls
if 'data' not in jlines.keys():
if not isinstance(jlines, dict) or 'data' not in jlines.keys() or not isinstance(jlines['data'], dict):
print(f'Error with criminalip data, dumping: {jlines}')
return
data = jlines['data']
for cert in data['certificates']:
# print(f'Current cert: {cert}')
if cert['subject'].endswith('.' + self.word):
self.totalhosts.add(cert['subject'])
for connected_domain in data['connected_domain_subdomain']:
for cert in data.get('certificates', []):
# print(f'Current cert: {cert}')
if isinstance(cert, dict):
self._add_host(cert.get('subject'))
for connected_domain in data.get('connected_domain_subdomain', []):
try:
main_domain = connected_domain['main_domain']['domain']
subdomains = [sub['domain'] for sub in connected_domain['subdomains']]
if main_domain.endswith('.' + self.word):
self.totalhosts.add(main_domain)
if not isinstance(connected_domain, dict):
continue
main_domain = connected_domain.get('main_domain', {}).get('domain')
if main_domain is not None:
self._add_host(main_domain)
subdomains = [sub.get('domain') for sub in connected_domain.get('subdomains', []) if isinstance(sub, dict)]
for sub in subdomains:
# print(f'Current sub: {sub}')
if sub.endswith('.' + self.word):
self.totalhosts.add(sub)
self._add_host(sub)
except Exception as e:
print(f'An exception has occurred: {e}')
print(f'Main line: {connected_domain}')
for ip_info in data['connected_ip_info']:
self.asns.add(str(ip_info['asn']))
domains = [sub['domain'] for sub in ip_info['domain_list']]
for ip_info in data.get('connected_ip_info', []):
if not isinstance(ip_info, dict):
continue
self._add_asn(ip_info.get('asn'))
domains = [sub.get('domain') for sub in ip_info.get('domain_list', []) if isinstance(sub, dict)]
for sub in domains:
if sub.endswith('.' + self.word):
self.totalhosts.add(sub)
self.totalips.add(ip_info['ip'])
if self._add_host(sub):
self._add_ip(ip_info.get('ip'))
for cookie in data['cookies']:
if cookie['domain'] != '.' + self.word and cookie['domain'].endswith('.' + self.word):
self.totalhosts.add(cookie['domain'])
for subdomain in data.get('subdomains', []):
if isinstance(subdomain, dict):
self._add_host(subdomain.get('subdomain_name'))
self._add_host(subdomain.get('domain'))
for country in data['country']:
if country['domain'].endswith('.' + self.word):
self.totalhosts.add(country['domain'])
for ip in country['mapped_ips']:
self.totalips.add(ip['ip'])
for cookie in data.get('cookies', []):
if isinstance(cookie, dict):
cookie_domain = cookie.get('domain')
if isinstance(cookie_domain, str):
self._add_host(cookie_domain.lstrip('.'), include_root=False)
for k, v in data['dns_record'].items():
for connected_ip in data.get('connected_ip', []):
if isinstance(connected_ip, dict):
self._add_ip(connected_ip.get('ip'))
for mapped_ip in data.get('mapped_ip', []):
if isinstance(mapped_ip, dict):
self._add_ip(mapped_ip.get('ip'))
for country in data.get('country', []):
if not isinstance(country, dict):
continue
if self._add_host(country.get('domain')):
for ip in country.get('mapped_ips', []):
if isinstance(ip, dict):
self._add_ip(ip.get('ip'))
for k, v in data.get('dns_record', {}).items():
if k == 'dns_record_type_a':
for ip in data['dns_record'][k]['ipv4']:
self.totalips.add(ip['ip'])
dns_a_record = data.get('dns_record', {}).get(k, {})
if not isinstance(dns_a_record, dict):
continue
for ip in dns_a_record.get('ipv4', []):
if isinstance(ip, dict):
self._add_ip(ip.get('ip'))
elif isinstance(ip, str):
self._add_ip(ip)
for ip in dns_a_record.get('ipv6', []):
if isinstance(ip, dict):
self._add_ip(ip.get('ip'))
elif isinstance(ip, str):
self._add_ip(ip)
elif isinstance(v, list):
for item in v:
if isinstance(item, list):
for subitem in item:
if subitem.endswith('.' + self.word):
self.totalhosts.add(subitem)
elif item.endswith('.' + self.word):
self.totalhosts.add(item)
self._collect_hosts_from_value(v)
elif isinstance(v, dict):
self._collect_hosts_from_value(v)
for domain_list in data['domain_list']:
self.asns.add(str(domain_list['asn']))
domains = [sub['domain'] for sub in domain_list['domain_list']]
for domain_list in data.get('domain_list', []):
if not isinstance(domain_list, dict):
continue
self._add_asn(domain_list.get('asn'))
domains = [sub.get('domain') for sub in domain_list.get('domain_list', []) if isinstance(sub, dict)]
for sub in domains:
if sub.endswith('.' + self.word):
self.totalhosts.add(sub)
self.totalips.add(domain_list['ip'])
if self._add_host(sub):
self._add_ip(domain_list.get('ip'))
for html_page_links in data['html_page_link_domains']:
domain = html_page_links['domain']
if domain.endswith('.' + self.word):
self.totalhosts.add(domain)
for ip in html_page_links['mapped_ips']:
self.totalips.add(ip['ip'])
for html_page_links in data.get('html_page_link_domains', []):
if not isinstance(html_page_links, dict):
continue
if self._add_host(html_page_links.get('domain')):
for ip in html_page_links.get('mapped_ips', []):
if isinstance(ip, dict):
self._add_ip(ip.get('ip'))
# TODO combine data['links'] and data['network_logs'] urls into one list for one run through
for link in data['links']:
url = link['url']
parsed_url = urlparse(url)
netloc = parsed_url.netloc
if self.word in netloc:
if (':' in netloc and netloc.split(':')[0].endswith(self.word)) or netloc.endswith(self.word):
self.totalhosts.add(netloc)
for link in data.get('links', []):
if isinstance(link, dict):
self._add_host_from_url(link.get('url'))
for log in data['network_logs']:
url = log['url']
parsed_url = urlparse(url)
netloc = parsed_url.netloc
if self.word in netloc:
if (':' in netloc and netloc.split(':')[0].endswith(self.word)) or netloc.endswith(self.word):
self.totalhosts.add(netloc)
self.asns.add(str(log['as_number']))
network_logs = data.get('network_logs', [])
if isinstance(network_logs, dict):
network_logs = network_logs.get('data', [])
if not isinstance(network_logs, list):
network_logs = []
for log in network_logs:
if not isinstance(log, dict):
continue
self._add_host_from_url(log.get('url'))
self._add_asn(log.get('as_number'))
ip_port = log.get('ip_port')
if isinstance(ip_port, str) and ':' in ip_port:
self._add_ip(ip_port.rsplit(':', 1)[0])
for redirects in data['page_redirections']:
for redirects in data.get('page_redirections', []):
if isinstance(redirects, dict):
redirects = [redirects]
if not isinstance(redirects, list):
continue
for redirect in redirects:
url = redirect['url']
parsed_url = urlparse(url)
netloc = parsed_url.netloc
if self.word in netloc:
if (':' in netloc and netloc.split(':')[0].endswith(self.word)) or netloc.endswith(self.word):
self.totalhosts.add(netloc)
if isinstance(redirect, dict):
self._add_host_from_url(redirect.get('url'))
self.totalhosts = {host.replace('www.', '') for host in self.totalhosts if '*.' + self.word != host}
self.totalhosts = {host[4:] if host.startswith('www.') else host for host in self.totalhosts if '*.' + self.word != host}
# print(f'hostnames: {self.totalhosts}')
# print(f'asns: {self.asns}')