From 06c2fbd95512fbc34528ddc8a8ad36a9db2e763d Mon Sep 17 00:00:00 2001 From: L1ghtn1ng Date: Fri, 20 Feb 2026 13:35:40 +0000 Subject: [PATCH] Fixes #2229 --- tests/discovery/test_criminalip.py | 96 +++++++ theHarvester/discovery/criminalip.py | 381 +++++++++++++++++---------- 2 files changed, 344 insertions(+), 133 deletions(-) create mode 100644 tests/discovery/test_criminalip.py diff --git a/tests/discovery/test_criminalip.py b/tests/discovery/test_criminalip.py new file mode 100644 index 00000000..6799e44f --- /dev/null +++ b/tests/discovery/test_criminalip.py @@ -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) diff --git a/theHarvester/discovery/criminalip.py b/theHarvester/discovery/criminalip.py index 1f3ed6f1..77ec0cd3 100644 --- a/theHarvester/discovery/criminalip.py +++ b/theHarvester/discovery/criminalip.py @@ -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}')