mirror of
https://github.com/laramies/theHarvester.git
synced 2026-09-13 21:27:42 +02:00
NotoriousRebel
ea5db87915
Preserve provider outcome release notes alongside the HTTP proxy wording. Keep the dev domain glossary and carry the master HTTP-only proxy policy into its relocated architecture guide. Retain RocketReach cancellation coverage while removing the obsolete global dependency stub. Align provider lifecycle regressions with the master session-construction boundary: construction ValueError is normalized to transport-error, while cancellation, lifecycle RuntimeError, and teardown ValueError still propagate.
878 lines
36 KiB
Python
878 lines
36 KiB
Python
import asyncio
|
|
import logging
|
|
import socket
|
|
import sys
|
|
from collections import OrderedDict
|
|
from datetime import UTC
|
|
from types import SimpleNamespace
|
|
from uuid import UUID
|
|
|
|
import pytest
|
|
|
|
from theHarvester.lib.core import FetcherResponse
|
|
from theHarvester.lib.source_execution import SourceExecutionReport
|
|
|
|
|
|
def patch_resolution(monkeypatch, module, addresses=('203.0.113.10',)):
|
|
requested_targets = []
|
|
|
|
async def resolve_ip_addresses(target, *, family=socket.AF_UNSPEC):
|
|
requested_targets.append((target, family))
|
|
return tuple(addresses)
|
|
|
|
monkeypatch.setattr(module, 'resolve_ip_addresses', resolve_ip_addresses, raising=True)
|
|
return requested_targets
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def disable_shodan_pacing(monkeypatch):
|
|
from theHarvester.discovery import shodansearch
|
|
|
|
monkeypatch.setattr(shodansearch.SearchShodan, 'REQUEST_INTERVAL_SECONDS', 0.0)
|
|
|
|
|
|
class TestShodanEngine:
|
|
@pytest.mark.asyncio
|
|
async def test_shared_ipv4_resolution_returns_every_unique_address_and_closes(self, monkeypatch):
|
|
from theHarvester.lib import hostchecker
|
|
|
|
closed = False
|
|
|
|
class Resolver:
|
|
async def getaddrinfo(self, target, *, family):
|
|
assert target == 'example.test'
|
|
assert family == socket.AF_INET
|
|
return SimpleNamespace(
|
|
nodes=[
|
|
SimpleNamespace(addr=(b'203.0.113.11', 0)),
|
|
SimpleNamespace(addr=('invalid', 0)),
|
|
SimpleNamespace(addr=('203.0.113.10', 0)),
|
|
SimpleNamespace(addr=('203.0.113.11', 0)),
|
|
]
|
|
)
|
|
|
|
async def close(self):
|
|
nonlocal closed
|
|
closed = True
|
|
|
|
monkeypatch.setattr(hostchecker.aiodns, 'DNSResolver', Resolver)
|
|
|
|
assert await hostchecker.resolve_ip_addresses('example.test', family=socket.AF_INET) == (
|
|
'203.0.113.10',
|
|
'203.0.113.11',
|
|
)
|
|
assert closed is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shodan_calls_direct_host_api_through_proxy_and_retains_attribution(self, monkeypatch):
|
|
from theHarvester.discovery import shodansearch
|
|
from theHarvester.lib.core import FetcherResponse
|
|
|
|
async def fetch_json(url, *, session, params, proxy, request_timeout):
|
|
assert url == 'https://api.shodan.io/shodan/host/192.0.2.10'
|
|
assert 'test-key' not in url
|
|
assert session is None
|
|
assert params == {'key': 'test-key'}
|
|
assert proxy is True
|
|
assert request_timeout is None
|
|
return FetcherResponse(
|
|
body={
|
|
'asn': 'AS64496',
|
|
'domains': ['Example.TEST.'],
|
|
'hostnames': ['API.Example.TEST.'],
|
|
'ip_str': '192.0.2.10',
|
|
'isp': 'Example ISP Label',
|
|
'org': 'Example Transit',
|
|
'data': [
|
|
{
|
|
'port': 53,
|
|
'transport': 'udp',
|
|
'product': 'dnsmasq',
|
|
'version': '2.90',
|
|
'timestamp': '2026-08-14T11:58:00Z',
|
|
'cpe': ['cpe:/a:thekelleys:dnsmasq:2.90'],
|
|
'data': 'raw provider banner must not be retained',
|
|
},
|
|
{
|
|
'port': 443,
|
|
'transport': 'tcp',
|
|
'product': 'nginx',
|
|
'version': '1.24.0',
|
|
'timestamp': '2026-08-14T12:01:00Z',
|
|
'http': {
|
|
'components': {'nginx': {}, 'Python': {}},
|
|
'server': 'nginx',
|
|
'title': 'Example',
|
|
'html': 'raw response body must not be retained',
|
|
},
|
|
'ssl': {
|
|
'jarm': 'example-jarm',
|
|
'cert': {
|
|
'subject': {'CN': 'api.example.test'},
|
|
'issuer': {'CN': 'Example CA'},
|
|
'expires': '2027-08-14T00:00:00Z',
|
|
'fingerprint': {'sha256': '0123456789abcdef'},
|
|
'extensions': [
|
|
{
|
|
'name': 'subjectAltName',
|
|
'data': r'0\x82\x10api.example.test\x82\x10www.example.test',
|
|
}
|
|
],
|
|
},
|
|
},
|
|
},
|
|
],
|
|
},
|
|
status=200,
|
|
headers={},
|
|
)
|
|
|
|
monkeypatch.setattr(shodansearch.Core, 'shodan_key', lambda: 'test-key')
|
|
monkeypatch.setattr(shodansearch.AsyncFetcher, 'fetch_json', fetch_json)
|
|
search = shodansearch.SearchShodan()
|
|
|
|
result = await search.search_ip('192.0.2.10', proxy=True)
|
|
|
|
assert result['192.0.2.10'] == {
|
|
'asn': 'AS64496',
|
|
'domains': ['example.test'],
|
|
'hostnames': ['api.example.test'],
|
|
'isp': 'Example ISP Label',
|
|
'organization': 'Example Transit',
|
|
'services': [
|
|
{
|
|
'cpes': ['cpe:/a:thekelleys:dnsmasq:2.90'],
|
|
'observed_at': '2026-08-14T11:58:00Z',
|
|
'port': 53,
|
|
'product': 'dnsmasq',
|
|
'transport': 'udp',
|
|
'version': '2.90',
|
|
},
|
|
{
|
|
'http': {
|
|
'components': ['Python', 'nginx'],
|
|
'server': 'nginx',
|
|
'title': 'Example',
|
|
},
|
|
'observed_at': '2026-08-14T12:01:00Z',
|
|
'port': 443,
|
|
'product': 'nginx',
|
|
'tls': {
|
|
'expires_at': '2027-08-14T00:00:00Z',
|
|
'issuer_cn': 'Example CA',
|
|
'jarm': 'example-jarm',
|
|
'sha256': '0123456789abcdef',
|
|
'subject_alt_names': ['www.example.test'],
|
|
'subject_cn': 'api.example.test',
|
|
},
|
|
'transport': 'tcp',
|
|
'version': '1.24.0',
|
|
},
|
|
],
|
|
}
|
|
attribution = next(iter(await search.get_asn_attributions()))
|
|
assert attribution.producer_kind == 'action'
|
|
assert attribution.producer == 'shodan'
|
|
assert attribution.asn == 'AS64496'
|
|
assert attribution.organization_label == 'Example Transit'
|
|
assert attribution.subject_kind == 'ip'
|
|
assert attribution.subject_value == '192.0.2.10'
|
|
assert attribution.collected_at.tzinfo is UTC
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shodan_maps_404_to_a_successful_zero_yield(self, monkeypatch, caplog):
|
|
from theHarvester.discovery import shodansearch
|
|
from theHarvester.lib.core import FetcherResponse
|
|
|
|
async def fetch_json(*_args, **_kwargs):
|
|
return FetcherResponse(body=None, status=404, headers={})
|
|
|
|
monkeypatch.setattr(shodansearch.Core, 'shodan_key', lambda: 'test-key')
|
|
monkeypatch.setattr(shodansearch.AsyncFetcher, 'fetch_json', fetch_json)
|
|
caplog.set_level(logging.INFO, logger=shodansearch.__name__)
|
|
|
|
search = shodansearch.SearchShodan()
|
|
result = await search.search_ip('203.0.113.1')
|
|
|
|
assert result == OrderedDict({'203.0.113.1': 'Not in Shodan'})
|
|
assert search.error_type is None
|
|
assert '203.0.113.1: Not in Shodan' in caplog.text
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize('status', [401, 403, 429, 500])
|
|
async def test_shodan_classifies_http_failures_without_provider_payload(self, monkeypatch, caplog, status):
|
|
from theHarvester.discovery import shodansearch
|
|
from theHarvester.lib.core import FetcherResponse
|
|
|
|
async def fetch_json(*_args, **_kwargs):
|
|
return FetcherResponse(body=None, status=status, headers={'x-provider-secret': 'secret'})
|
|
|
|
monkeypatch.setattr(shodansearch.Core, 'shodan_key', lambda: 'test-key')
|
|
monkeypatch.setattr(shodansearch.AsyncFetcher, 'fetch_json', fetch_json)
|
|
caplog.set_level(logging.INFO, logger=shodansearch.__name__)
|
|
|
|
search = shodansearch.SearchShodan()
|
|
result = await search.search_ip('203.0.113.1')
|
|
|
|
assert result == OrderedDict({'203.0.113.1': 'Shodan request failed'})
|
|
assert search.error_type == f'HTTP{status}Error'
|
|
assert 'secret' not in caplog.text
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shodan_classifies_bounded_transport_failure(self, monkeypatch):
|
|
from theHarvester.discovery import shodansearch
|
|
from theHarvester.lib.core import ResponseStreamError
|
|
|
|
async def fetch_json(*_args, **_kwargs):
|
|
raise ResponseStreamError('response-limit')
|
|
|
|
monkeypatch.setattr(shodansearch.Core, 'shodan_key', lambda: 'test-key')
|
|
monkeypatch.setattr(shodansearch.AsyncFetcher, 'fetch_json', fetch_json)
|
|
|
|
search = shodansearch.SearchShodan()
|
|
result = await search.search_ip('203.0.113.1')
|
|
|
|
assert result == OrderedDict({'203.0.113.1': 'Shodan request failed'})
|
|
assert search.error_type == 'ResponseLimitError'
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shodan_discovery_queries_every_resolved_ipv4_and_keeps_partial_results(self, monkeypatch):
|
|
from theHarvester.discovery import shodansearch
|
|
from theHarvester.lib.core import FetcherResponse
|
|
|
|
queried_urls = []
|
|
|
|
async def fetch_json(url, **_kwargs):
|
|
if url.endswith('/search'):
|
|
return FetcherResponse(body={'matches': [], 'total': 0}, status=200, headers={})
|
|
queried_urls.append(url)
|
|
ip = url.rsplit('/', 1)[-1]
|
|
if ip == '203.0.113.10':
|
|
return FetcherResponse(body=None, status=500, headers={})
|
|
return FetcherResponse(
|
|
body={
|
|
'data': [{'ip_str': ip, 'port': 443, 'transport': 'tcp'}],
|
|
'hostnames': ['CDN.WWW.Example.TEST.', 'outside.test'],
|
|
},
|
|
status=200,
|
|
headers={},
|
|
)
|
|
|
|
monkeypatch.setattr(shodansearch.Core, 'shodan_key', lambda: 'test-key')
|
|
monkeypatch.setattr(shodansearch.AsyncFetcher, 'fetch_json', fetch_json)
|
|
monkeypatch.setattr(
|
|
shodansearch.AsyncFetcher,
|
|
'_proxy_list',
|
|
{'http': ['http://proxy.example:8080'], 'socks5': []},
|
|
)
|
|
targets = patch_resolution(monkeypatch, shodansearch, ('203.0.113.10', '203.0.113.11'))
|
|
|
|
search = shodansearch.SearchShodan('WWW.Example.TEST.')
|
|
report = await search.process(proxy=True)
|
|
|
|
assert targets == [('www.example.test', socket.AF_INET)]
|
|
assert queried_urls == [
|
|
'https://api.shodan.io/shodan/host/203.0.113.10',
|
|
'https://api.shodan.io/shodan/host/203.0.113.11',
|
|
]
|
|
assert await search.get_hostnames() == {'cdn.www.example.test'}
|
|
assert report.status == 'failed'
|
|
assert report.stop_reason == 'provider-errors'
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shodan_discovery_paginates_hostname_and_tls_searches_with_scoped_certificate_names(self, monkeypatch):
|
|
from contextlib import asynccontextmanager
|
|
|
|
from theHarvester.discovery import shodansearch
|
|
from theHarvester.lib.core import FetcherResponse
|
|
|
|
search_calls = []
|
|
session = object()
|
|
session_calls = []
|
|
|
|
@asynccontextmanager
|
|
async def open_session(**kwargs):
|
|
session_calls.append(kwargs)
|
|
yield session
|
|
|
|
def banner(
|
|
ip,
|
|
port,
|
|
*,
|
|
hostnames=(),
|
|
subject_cn=None,
|
|
subject_alt_name='',
|
|
):
|
|
cert = {
|
|
'extensions': [
|
|
{'name': 'subjectAltName', 'data': subject_alt_name},
|
|
{'name': 'keyUsage', 'data': 'Digital Signature'},
|
|
]
|
|
}
|
|
if subject_cn is not None:
|
|
cert['subject'] = {'CN': subject_cn}
|
|
return {
|
|
'ip_str': ip,
|
|
'hostnames': list(hostnames),
|
|
'port': port,
|
|
'transport': 'tcp',
|
|
'ssl': {'cert': cert},
|
|
}
|
|
|
|
hostname_pages = {
|
|
1: {
|
|
'total': 2,
|
|
'matches': [
|
|
banner(
|
|
'198.51.100.20',
|
|
443,
|
|
hostnames=('API.Example.TEST.',),
|
|
subject_cn='*.example.test',
|
|
subject_alt_name=(r'0\x82\x10api.example.test\x82\x12outside.invalid\x82\x13vpn.example.test'),
|
|
)
|
|
],
|
|
},
|
|
2: {
|
|
'total': 2,
|
|
'matches': [banner('198.51.100.20', 8443, hostnames=('api.example.test',))],
|
|
},
|
|
}
|
|
ssl_page = {
|
|
'total': 3,
|
|
'matches': [
|
|
banner(
|
|
'198.51.100.20',
|
|
443,
|
|
hostnames=('api.example.test',),
|
|
subject_cn='*.example.test',
|
|
subject_alt_name=r'0\x82\x10api.example.test\x82\x13vpn.example.test',
|
|
),
|
|
banner(
|
|
'198.51.100.21',
|
|
443,
|
|
subject_cn='cert.example.test',
|
|
subject_alt_name=r'0\x82\x13cert.example.test\x82\x15outside.invalid',
|
|
),
|
|
banner(
|
|
'198.51.100.22',
|
|
443,
|
|
hostnames=('outside.invalid',),
|
|
subject_cn='outside.invalid',
|
|
subject_alt_name=r'0\x82\x15outside.invalid',
|
|
),
|
|
],
|
|
}
|
|
|
|
async def fetch_json(url, *, params, session: object | None = None, proxy='', request_timeout):
|
|
assert session is not None
|
|
assert proxy == ''
|
|
assert request_timeout is None
|
|
assert params['key'] == 'test-key'
|
|
if url.endswith('/search'):
|
|
search_calls.append(dict(params))
|
|
query = params['query']
|
|
if query == 'hostname:example.test':
|
|
return FetcherResponse(body=hostname_pages[params['page']], status=200, headers={})
|
|
assert query == 'ssl:example.test'
|
|
assert params['page'] == 1
|
|
return FetcherResponse(body=ssl_page, status=200, headers={})
|
|
assert url == 'https://api.shodan.io/shodan/host/203.0.113.10'
|
|
return FetcherResponse(
|
|
body={'data': [{'port': 80, 'transport': 'tcp'}], 'hostnames': ['www.example.test']},
|
|
status=200,
|
|
headers={},
|
|
)
|
|
|
|
monkeypatch.setattr(shodansearch.Core, 'shodan_key', lambda: 'test-key')
|
|
monkeypatch.setattr(shodansearch.AsyncFetcher, 'fetch_json', fetch_json)
|
|
monkeypatch.setattr(shodansearch.AsyncFetcher, 'open_session', open_session)
|
|
monkeypatch.setattr(
|
|
shodansearch.AsyncFetcher,
|
|
'_proxy_list',
|
|
{'http': ['http://proxy.example:8080'], 'socks5': []},
|
|
)
|
|
patch_resolution(monkeypatch, shodansearch)
|
|
|
|
search = shodansearch.SearchShodan('example.test')
|
|
report = await search.process(proxy=True)
|
|
|
|
assert session_calls == [{'proxy': True}]
|
|
assert [(call['query'], call['page']) for call in search_calls] == [
|
|
('hostname:example.test', 1),
|
|
('hostname:example.test', 2),
|
|
('ssl:example.test', 1),
|
|
]
|
|
assert all(call['minify'] == 'false' and call['fields'] == search.SEARCH_FIELDS for call in search_calls)
|
|
assert await search.get_hostnames() == {
|
|
'api.example.test',
|
|
'cert.example.test',
|
|
'vpn.example.test',
|
|
'www.example.test',
|
|
}
|
|
hosts = {host.ip: host.to_details() for host in await search.get_shodan_hosts()}
|
|
assert set(hosts) == {'198.51.100.20', '198.51.100.21', '203.0.113.10'}
|
|
assert [service['port'] for service in hosts['198.51.100.20']['services']] == [443, 8443]
|
|
assert hosts['198.51.100.20']['services'][0]['tls'] == {
|
|
'subject_alt_names': ['api.example.test', 'vpn.example.test'],
|
|
'subject_cn': '*.example.test',
|
|
}
|
|
assert hosts['198.51.100.21']['services'][0]['tls'] == {'subject_cn': 'cert.example.test'}
|
|
assert report is None
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
('hostnames', 'expected_status'),
|
|
[(('api.example.test',), 'partial'), (('outside.invalid',), 'failed')],
|
|
)
|
|
async def test_shodan_repeated_search_page_is_not_counted_toward_provider_total(
|
|
self,
|
|
monkeypatch,
|
|
hostnames,
|
|
expected_status,
|
|
):
|
|
from theHarvester.discovery import shodansearch
|
|
from theHarvester.lib.core import FetcherResponse
|
|
|
|
search_calls = []
|
|
repeated_match = {
|
|
'ip_str': '198.51.100.20',
|
|
'hostnames': list(hostnames),
|
|
'port': 443,
|
|
'transport': 'tcp',
|
|
}
|
|
|
|
async def fetch_json(url, *, params, **_kwargs):
|
|
if not url.endswith('/search'):
|
|
return FetcherResponse(body=None, status=404, headers={})
|
|
search_calls.append((params['query'], params['page']))
|
|
if params['query'] == 'hostname:example.test':
|
|
return FetcherResponse(body={'matches': [repeated_match], 'total': 3}, status=200, headers={})
|
|
return FetcherResponse(body={'matches': [], 'total': 0}, status=200, headers={})
|
|
|
|
monkeypatch.setattr(shodansearch.Core, 'shodan_key', lambda: 'test-key')
|
|
monkeypatch.setattr(shodansearch.AsyncFetcher, 'fetch_json', fetch_json)
|
|
patch_resolution(monkeypatch, shodansearch)
|
|
search = shodansearch.SearchShodan('example.test')
|
|
|
|
report = await search.process()
|
|
|
|
assert search_calls == [
|
|
('hostname:example.test', 1),
|
|
('hostname:example.test', 2),
|
|
('ssl:example.test', 1),
|
|
]
|
|
assert await search.get_hostnames() == ({'api.example.test'} if expected_status == 'partial' else set())
|
|
assert report.status == expected_status
|
|
assert report.stop_reason == 'repeated-page'
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shodan_discovery_counts_service_only_evidence_as_a_result(self, monkeypatch):
|
|
from theHarvester.discovery import shodansearch
|
|
from theHarvester.lib.core import FetcherResponse
|
|
|
|
async def fetch_json(url, **_kwargs):
|
|
if url.endswith('/search'):
|
|
return FetcherResponse(body={'matches': [], 'total': 0}, status=200, headers={})
|
|
return FetcherResponse(
|
|
body={'data': [{'port': 53, 'transport': 'udp'}], 'hostnames': []},
|
|
status=200,
|
|
headers={},
|
|
)
|
|
|
|
monkeypatch.setattr(shodansearch.Core, 'shodan_key', lambda: 'test-key')
|
|
monkeypatch.setattr(shodansearch.AsyncFetcher, 'fetch_json', fetch_json)
|
|
patch_resolution(monkeypatch, shodansearch)
|
|
|
|
search = shodansearch.SearchShodan('example.test')
|
|
report = await search.process()
|
|
|
|
assert report is None
|
|
assert [host.ip for host in await search.get_shodan_hosts()] == ['203.0.113.10']
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shodan_discovery_retains_valid_services_and_reports_malformed_provider_data(self, monkeypatch):
|
|
from theHarvester.discovery import shodansearch
|
|
from theHarvester.lib.core import FetcherResponse
|
|
|
|
async def fetch_json(*_args, **_kwargs):
|
|
return FetcherResponse(
|
|
body={
|
|
'data': [
|
|
{'port': 53, 'transport': 'udp', 'product': {'raw': 'provider object'}},
|
|
{'transport': 'tcp'},
|
|
],
|
|
'hostnames': [],
|
|
},
|
|
status=200,
|
|
headers={},
|
|
)
|
|
|
|
monkeypatch.setattr(shodansearch.Core, 'shodan_key', lambda: 'test-key')
|
|
monkeypatch.setattr(shodansearch.AsyncFetcher, 'fetch_json', fetch_json)
|
|
patch_resolution(monkeypatch, shodansearch)
|
|
|
|
search = shodansearch.SearchShodan('example.test')
|
|
report = await search.process()
|
|
|
|
assert report.status == 'failed'
|
|
assert report.stop_reason == 'provider-error'
|
|
assert search.error_type == 'InvalidResponseError'
|
|
assert (await search.get_shodan_hosts())[0].to_details() == {'services': [{'port': 53, 'transport': 'udp'}]}
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
('status', 'expected_reason'),
|
|
[(401, 'access-denied'), (403, 'access-denied'), (429, 'rate-limited')],
|
|
)
|
|
async def test_shodan_discovery_reports_specific_provider_stop_reason(self, monkeypatch, status, expected_reason):
|
|
from theHarvester.discovery import shodansearch
|
|
from theHarvester.lib.core import FetcherResponse
|
|
|
|
async def fetch_json(*_args, **_kwargs):
|
|
return FetcherResponse(body=None, status=status, headers={})
|
|
|
|
monkeypatch.setattr(shodansearch.Core, 'shodan_key', lambda: 'test-key')
|
|
monkeypatch.setattr(shodansearch.AsyncFetcher, 'fetch_json', fetch_json)
|
|
patch_resolution(monkeypatch, shodansearch)
|
|
|
|
search = shodansearch.SearchShodan('example.test')
|
|
report = await search.process()
|
|
|
|
assert report.status == ('rate-limited' if status == 429 else 'failed')
|
|
assert report.stop_reason == expected_reason
|
|
|
|
def test_shodan_discovery_requires_a_configured_key(self, monkeypatch):
|
|
from theHarvester.discovery import shodansearch
|
|
from theHarvester.discovery.constants import MissingKey
|
|
|
|
monkeypatch.setattr(shodansearch.Core, 'shodan_key', lambda: None)
|
|
|
|
with pytest.raises(MissingKey):
|
|
shodansearch.SearchShodan('example.test')
|
|
|
|
def test_shodan_discovery_rejects_query_filter_injection(self, monkeypatch):
|
|
from theHarvester.discovery import shodansearch
|
|
|
|
monkeypatch.setattr(shodansearch.Core, 'shodan_key', lambda: 'test-key')
|
|
|
|
with pytest.raises(ValueError, match='must be a hostname'):
|
|
shodansearch.SearchShodan('example.test ssl:true')
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shodan_discovery_reports_dns_failure_after_searching_provider(self, monkeypatch):
|
|
from theHarvester.discovery import shodansearch
|
|
from theHarvester.lib.core import FetcherResponse
|
|
|
|
provider_queries = []
|
|
|
|
async def fail_resolution(_target, *, family):
|
|
assert family == socket.AF_INET
|
|
raise shodansearch.aiodns.error.DNSError(shodansearch.aiodns.error.ARES_ENOTFOUND, 'not found')
|
|
|
|
async def fetch_json(url, *, params, **_kwargs):
|
|
assert url.endswith('/search')
|
|
provider_queries.append(params['query'])
|
|
return FetcherResponse(body={'matches': [], 'total': 0}, status=200, headers={})
|
|
|
|
monkeypatch.setattr(shodansearch.Core, 'shodan_key', lambda: 'test-key')
|
|
monkeypatch.setattr(shodansearch, 'resolve_ip_addresses', fail_resolution)
|
|
monkeypatch.setattr(shodansearch.AsyncFetcher, 'fetch_json', fetch_json)
|
|
|
|
search = shodansearch.SearchShodan('example.test')
|
|
report = await search.process()
|
|
|
|
assert not await search.get_hostnames()
|
|
assert provider_queries == ['hostname:example.test', 'ssl:example.test']
|
|
assert report.status == 'failed'
|
|
assert report.stop_reason == 'dns-resolution-failed'
|
|
|
|
def test_shodan_preserves_an_explicit_www_search_scope(self, monkeypatch):
|
|
from theHarvester.discovery import shodansearch
|
|
|
|
monkeypatch.setattr(shodansearch.Core, 'shodan_key', lambda: 'test-key')
|
|
|
|
search = shodansearch.SearchShodan('WWW.Example.TEST.')
|
|
|
|
assert search.word == 'www.example.test'
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shodan_direct_request_cancellation_propagates(self, monkeypatch):
|
|
from theHarvester.discovery import shodansearch
|
|
|
|
entered = asyncio.Event()
|
|
release = asyncio.Event()
|
|
|
|
async def fetch_json(*_args, **_kwargs):
|
|
entered.set()
|
|
await release.wait()
|
|
raise AssertionError('cancelled request resumed')
|
|
|
|
monkeypatch.setattr(shodansearch.Core, 'shodan_key', lambda: 'test-key')
|
|
monkeypatch.setattr(shodansearch.AsyncFetcher, 'fetch_json', fetch_json)
|
|
patch_resolution(monkeypatch, shodansearch)
|
|
|
|
task = asyncio.create_task(shodansearch.SearchShodan('example.test').process())
|
|
await entered.wait()
|
|
task.cancel('operator-stop')
|
|
with pytest.raises(asyncio.CancelledError, match='operator-stop'):
|
|
await task
|
|
release.set()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shodan_engine_processes_and_persists_sourced_hostnames(self, monkeypatch, capsys, tmp_path):
|
|
import theHarvester.__main__ as main_module
|
|
from theHarvester.lib.completed_result import ResultObservation, SourceExecution
|
|
from theHarvester.lib.database import ResultStore
|
|
from theHarvester.lib.shodan_evidence import ShodanHostObservation
|
|
|
|
database = tmp_path / 'stash.sqlite'
|
|
monkeypatch.setattr(main_module, 'ResultStore', lambda: ResultStore(database), raising=True)
|
|
|
|
class DummySearchShodan:
|
|
def __init__(self, domain):
|
|
assert domain == 'example.com'
|
|
self.shodan_host = ShodanHostObservation.from_record(
|
|
'192.0.2.10',
|
|
{
|
|
'organization': 'Example Transit',
|
|
'hostnames': ['a.example.com', 'b.example.com'],
|
|
'services': [
|
|
{'port': 53, 'transport': 'udp'},
|
|
{'port': 443, 'transport': 'tcp', 'product': 'nginx'},
|
|
],
|
|
},
|
|
)
|
|
|
|
async def process(self, proxy=False):
|
|
return None
|
|
|
|
async def get_hostnames(self):
|
|
return {'a.example.com', 'b.example.com'}
|
|
|
|
async def get_shodan_hosts(self):
|
|
return (self.shodan_host,)
|
|
|
|
monkeypatch.setattr(main_module.shodansearch, 'SearchShodan', DummySearchShodan, raising=True)
|
|
monkeypatch.setattr(sys, 'argv', ['theHarvester', '-d', 'example.com', '-b', 'shodan'], raising=True)
|
|
|
|
with pytest.raises(SystemExit) as excinfo:
|
|
await main_module.start()
|
|
assert excinfo.value.code == 0
|
|
|
|
out = capsys.readouterr().out
|
|
assert 'An error occurred while processing a "work item"' not in out
|
|
assert {'a.example.com', 'b.example.com'} <= set(out.split())
|
|
assert '"type": "shodan-host"' in out
|
|
assert '"value": "192.0.2.10"' in out
|
|
assert '"transport": "udp"' in out
|
|
assert '"transport": "tcp"' in out
|
|
|
|
store = ResultStore(database)
|
|
try:
|
|
runs = await store.list_runs()
|
|
completed = await store.load_run(UUID(str(runs[0]['run_id'])))
|
|
assert len(completed.source_executions) == 1
|
|
execution = completed.source_executions[0]
|
|
assert execution == SourceExecution('shodan', 'completed', execution.duration_ms, 3)
|
|
assert completed.observations == (
|
|
ResultObservation('shodan', 'hostname', 'a.example.com'),
|
|
ResultObservation('shodan', 'hostname', 'b.example.com'),
|
|
ResultObservation('shodan', 'shodan-host', '192.0.2.10'),
|
|
)
|
|
assert completed.shodan_hosts == (DummySearchShodan('example.com').shodan_host,)
|
|
finally:
|
|
await store.dispose()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shodan_internetdb_uses_shared_resolution_and_normalizes_scope(self, monkeypatch):
|
|
from theHarvester.discovery import shodan_internetdb
|
|
|
|
targets = patch_resolution(monkeypatch, shodan_internetdb, ('203.0.113.1',))
|
|
|
|
async def fake_fetch_all(urls, json=False, proxy=False, include_metadata=False):
|
|
assert urls == ['https://internetdb.shodan.io/203.0.113.1']
|
|
assert include_metadata is True
|
|
assert proxy is True
|
|
return [
|
|
FetcherResponse(
|
|
{
|
|
'ip': '203.0.113.1',
|
|
'hostnames': ['API.Example.TEST.', 'www.notexample.test', None],
|
|
'ports': [443],
|
|
},
|
|
200,
|
|
{},
|
|
)
|
|
]
|
|
|
|
monkeypatch.setattr(shodan_internetdb.AsyncFetcher, 'fetch_all', fake_fetch_all, raising=True)
|
|
|
|
search = shodan_internetdb.SearchShodanInternetDB(' Example.TEST. ')
|
|
report = await search.process(proxy=True)
|
|
|
|
assert targets == [('example.test', socket.AF_UNSPEC)]
|
|
assert await search.get_hostnames() == {'api.example.test'}
|
|
assert await search.get_ips() == {'203.0.113.1'}
|
|
assert report is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shodan_internetdb_rejects_evidence_for_an_unrequested_ip(self, monkeypatch):
|
|
from theHarvester.discovery import shodan_internetdb
|
|
|
|
patch_resolution(monkeypatch, shodan_internetdb, ('203.0.113.1',))
|
|
|
|
async def fake_fetch_all(_urls, json=False, proxy=False, include_metadata=False):
|
|
assert include_metadata is True
|
|
return [
|
|
FetcherResponse(
|
|
{
|
|
'ip': '198.51.100.2',
|
|
'hostnames': ['injected.example.test'],
|
|
'ports': [443],
|
|
'vulns': ['CVE-2026-0001'],
|
|
'tags': ['injected'],
|
|
'cpes': ['cpe:/a:injected'],
|
|
},
|
|
200,
|
|
{},
|
|
)
|
|
]
|
|
|
|
monkeypatch.setattr(shodan_internetdb.AsyncFetcher, 'fetch_all', fake_fetch_all, raising=True)
|
|
|
|
search = shodan_internetdb.SearchShodanInternetDB('example.test')
|
|
report = await search.process()
|
|
|
|
assert not await search.get_hostnames()
|
|
assert not await search.get_ips()
|
|
assert not await search.get_ports()
|
|
assert not await search.get_vulns()
|
|
assert not await search.get_tags()
|
|
assert not await search.get_cpes()
|
|
assert report == SourceExecutionReport('failed', 'invalid-response')
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shodan_internetdb_resolution_failure_skips_provider_request(self, monkeypatch):
|
|
from theHarvester.discovery import shodan_internetdb
|
|
|
|
async def fail_resolution(_target):
|
|
raise shodan_internetdb.aiodns.error.DNSError(shodan_internetdb.aiodns.error.ARES_ENOTFOUND, 'not found')
|
|
|
|
async def fail_fetch(*_args, **_kwargs):
|
|
raise AssertionError('provider request must not run')
|
|
|
|
monkeypatch.setattr(shodan_internetdb, 'resolve_ip_addresses', fail_resolution)
|
|
monkeypatch.setattr(shodan_internetdb.AsyncFetcher, 'fetch_all', fail_fetch, raising=True)
|
|
|
|
search = shodan_internetdb.SearchShodanInternetDB('example.test')
|
|
report = await search.process()
|
|
|
|
assert not await search.get_hostnames()
|
|
assert not await search.get_ips()
|
|
assert report == SourceExecutionReport('failed', 'dns-resolution-failed')
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shodan_internetdb_provider_failure_reports_transport_error(self, monkeypatch, caplog):
|
|
from theHarvester.discovery import shodan_internetdb
|
|
|
|
patch_resolution(monkeypatch, shodan_internetdb, ('203.0.113.1',))
|
|
|
|
async def fail_fetch(*_args, **_kwargs):
|
|
raise RuntimeError('provider-secret-payload')
|
|
|
|
monkeypatch.setattr(shodan_internetdb.AsyncFetcher, 'fetch_all', fail_fetch, raising=True)
|
|
caplog.set_level(logging.INFO, logger=shodan_internetdb.__name__)
|
|
|
|
search = shodan_internetdb.SearchShodanInternetDB('example.test')
|
|
report = await search.process()
|
|
|
|
assert not await search.get_hostnames()
|
|
assert not await search.get_ips()
|
|
assert 'Shodan InternetDB request failed' in caplog.text
|
|
assert 'provider-secret-payload' not in caplog.text
|
|
assert report.status == 'failed'
|
|
assert report.stop_reason == 'transport-error'
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shodan_internetdb_no_resolved_ips_is_valid_empty(self, monkeypatch):
|
|
from theHarvester.discovery import shodan_internetdb
|
|
|
|
patch_resolution(monkeypatch, shodan_internetdb, ())
|
|
|
|
async def fail_fetch(*_args, **_kwargs):
|
|
raise AssertionError('provider request must not run')
|
|
|
|
monkeypatch.setattr(shodan_internetdb.AsyncFetcher, 'fetch_all', fail_fetch)
|
|
|
|
search = shodan_internetdb.SearchShodanInternetDB('example.test')
|
|
|
|
assert await search.process() is None
|
|
assert not await search.get_hostnames()
|
|
assert not await search.get_ips()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shodan_internetdb_404_is_valid_empty(self, monkeypatch):
|
|
from theHarvester.discovery import shodan_internetdb
|
|
|
|
patch_resolution(monkeypatch, shodan_internetdb, ('203.0.113.1',))
|
|
|
|
async def fake_fetch_all(*_args, **kwargs):
|
|
assert kwargs['include_metadata'] is True
|
|
return [FetcherResponse({'detail': 'provider detail'}, 404, {})]
|
|
|
|
monkeypatch.setattr(shodan_internetdb.AsyncFetcher, 'fetch_all', fake_fetch_all)
|
|
search = shodan_internetdb.SearchShodanInternetDB('example.test')
|
|
|
|
assert await search.process() is None
|
|
assert not await search.get_hostnames()
|
|
assert not await search.get_ips()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shodan_internetdb_retains_one_ip_when_another_request_fails(self, monkeypatch):
|
|
from theHarvester.discovery import shodan_internetdb
|
|
|
|
patch_resolution(monkeypatch, shodan_internetdb, ('203.0.113.1', '203.0.113.2'))
|
|
|
|
async def fake_fetch_all(*_args, **kwargs):
|
|
assert kwargs['include_metadata'] is True
|
|
return [
|
|
FetcherResponse({'ip': '203.0.113.1', 'hostnames': ['API.Example.TEST.']}, 200, {}),
|
|
FetcherResponse({}, 503, {}),
|
|
]
|
|
|
|
monkeypatch.setattr(shodan_internetdb.AsyncFetcher, 'fetch_all', fake_fetch_all)
|
|
search = shodan_internetdb.SearchShodanInternetDB('example.test')
|
|
|
|
report = await search.process()
|
|
|
|
assert await search.get_hostnames() == {'api.example.test'}
|
|
assert await search.get_ips() == {'203.0.113.1'}
|
|
assert report == SourceExecutionReport('partial', 'http-503')
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shodan_internetdb_malformed_response_and_cancellation_are_distinct(self, monkeypatch):
|
|
from theHarvester.discovery import shodan_internetdb
|
|
|
|
patch_resolution(monkeypatch, shodan_internetdb, ('203.0.113.1',))
|
|
|
|
async def malformed(*_args, **_kwargs):
|
|
return [FetcherResponse('not-json', 200, {})]
|
|
|
|
monkeypatch.setattr(shodan_internetdb.AsyncFetcher, 'fetch_all', malformed)
|
|
assert await shodan_internetdb.SearchShodanInternetDB('example.test').process() == SourceExecutionReport(
|
|
'failed', 'invalid-response'
|
|
)
|
|
|
|
async def cancelled(*_args, **_kwargs):
|
|
raise asyncio.CancelledError('operator-stop')
|
|
|
|
monkeypatch.setattr(shodan_internetdb.AsyncFetcher, 'fetch_all', cancelled)
|
|
with pytest.raises(asyncio.CancelledError, match='operator-stop'):
|
|
await shodan_internetdb.SearchShodanInternetDB('example.test').process()
|
|
|
|
|
|
pytestmark = [
|
|
pytest.mark.provider_contract('shodan'),
|
|
pytest.mark.provider_contract('shodanInternetDB'),
|
|
]
|