mirror of
https://github.com/laramies/theHarvester.git
synced 2026-09-06 17:57:40 +02:00
292 lines
9.9 KiB
Python
292 lines
9.9 KiB
Python
import asyncio
|
|
import logging
|
|
|
|
import pytest
|
|
|
|
from theHarvester.discovery import pentesttools
|
|
from theHarvester.lib.source_execution import SourceExecutionReport
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_successful_scan_returns_normalized_hostnames_and_ips(monkeypatch) -> None:
|
|
monkeypatch.setattr(pentesttools.Core, 'pentest_tools_key', lambda: 'test-key')
|
|
start_requests = []
|
|
get_requests = []
|
|
responses = iter(
|
|
[
|
|
{'data': {'status_name': 'finished'}},
|
|
{
|
|
'data': {
|
|
'output_type': 'subdomain_list',
|
|
'output_data': {
|
|
'subdomains': [
|
|
{'hostname': 'Api.Example.TEST.', 'ip_address': '203.0.113.10'},
|
|
{'hostname': 'Example.TEST.', 'ip_address': '203.0.113.11'},
|
|
{'hostname': 'NoIp.Example.TEST.', 'ip_address': ''},
|
|
{'hostname': 'www.notexample.test', 'ip_address': '198.51.100.1'},
|
|
{'hostname': 'broken'},
|
|
]
|
|
},
|
|
}
|
|
},
|
|
]
|
|
)
|
|
|
|
async def fake_post_fetch(**kwargs):
|
|
start_requests.append(kwargs)
|
|
return {'data': {'created_id': 420323, 'target_id': 5426912}}
|
|
|
|
async def fake_fetch(**kwargs):
|
|
get_requests.append(kwargs)
|
|
return next(responses)
|
|
|
|
async def no_sleep(*_args, **_kwargs):
|
|
return None
|
|
|
|
monkeypatch.setattr(pentesttools.AsyncFetcher, 'post_fetch', fake_post_fetch)
|
|
monkeypatch.setattr(pentesttools.AsyncFetcher, 'fetch', fake_fetch)
|
|
monkeypatch.setattr(pentesttools.asyncio, 'sleep', no_sleep)
|
|
|
|
search = pentesttools.SearchPentestTools(' Example.TEST. ')
|
|
await search.process()
|
|
|
|
assert await search.get_hostnames() == {'api.example.test', 'noip.example.test'}
|
|
assert await search.get_ips() == {'203.0.113.10', '203.0.113.11'}
|
|
headers = {
|
|
'Accept': 'application/json',
|
|
'Authorization': 'Bearer test-key',
|
|
'Content-Type': 'application/json',
|
|
}
|
|
assert start_requests == [
|
|
{
|
|
'url': 'https://app.pentest-tools.com/api/v2/scans',
|
|
'headers': headers,
|
|
'json_body': {
|
|
'tool_id': 20,
|
|
'target_name': 'example.test',
|
|
'tool_params': {'scan_type': 'light', 'web_details': False, 'unresolved_results': True},
|
|
},
|
|
'json': True,
|
|
'proxy': False,
|
|
}
|
|
]
|
|
assert get_requests == [
|
|
{
|
|
'url': 'https://app.pentest-tools.com/api/v2/scans/420323',
|
|
'headers': headers,
|
|
'json': True,
|
|
'proxy': False,
|
|
},
|
|
{
|
|
'url': 'https://app.pentest-tools.com/api/v2/scans/420323/output',
|
|
'headers': headers,
|
|
'json': True,
|
|
'proxy': False,
|
|
},
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_error_payload_is_not_logged(monkeypatch, caplog) -> None:
|
|
monkeypatch.setattr(pentesttools.Core, 'pentest_tools_key', lambda: 'test-key')
|
|
|
|
async def fake_post_fetch(**_kwargs):
|
|
return {'data': {'created_id': 420323}}
|
|
|
|
async def fake_fetch(**_kwargs):
|
|
return {
|
|
'data': {
|
|
'status_name': 'failed to start',
|
|
'status_message': 'provider-secret-payload private target data',
|
|
}
|
|
}
|
|
|
|
async def no_sleep(*_args, **_kwargs):
|
|
return None
|
|
|
|
monkeypatch.setattr(pentesttools.AsyncFetcher, 'post_fetch', fake_post_fetch)
|
|
monkeypatch.setattr(pentesttools.AsyncFetcher, 'fetch', fake_fetch)
|
|
monkeypatch.setattr(pentesttools.asyncio, 'sleep', no_sleep)
|
|
caplog.set_level(logging.INFO, logger=pentesttools.__name__)
|
|
|
|
report = await pentesttools.SearchPentestTools('example.com').process()
|
|
|
|
assert report == SourceExecutionReport('failed', 'provider-error')
|
|
assert 'provider-secret-payload' not in caplog.text
|
|
assert 'private target data' not in caplog.text
|
|
assert 'did not finish successfully' in caplog.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
'response',
|
|
['not-json', {}, {'data': {}}, {'data': {'created_id': 'scan-1'}}, {'data': {'created_id': True}}],
|
|
)
|
|
async def test_malformed_start_response_completes_without_evidence(monkeypatch, caplog, response) -> None:
|
|
monkeypatch.setattr(pentesttools.Core, 'pentest_tools_key', lambda: 'test-key')
|
|
|
|
async def fake_post_fetch(*_args, **_kwargs):
|
|
return response
|
|
|
|
monkeypatch.setattr(pentesttools.AsyncFetcher, 'post_fetch', fake_post_fetch)
|
|
caplog.set_level(logging.INFO, logger=pentesttools.__name__)
|
|
|
|
search = pentesttools.SearchPentestTools('example.test')
|
|
await search.process()
|
|
|
|
assert not await search.get_hostnames()
|
|
assert not await search.get_ips()
|
|
|
|
assert 'malformed' in caplog.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_waiting_scan_stops_after_ten_status_checks(monkeypatch, caplog) -> None:
|
|
monkeypatch.setattr(pentesttools.Core, 'pentest_tools_key', lambda: 'test-key')
|
|
status_calls = 0
|
|
|
|
async def fake_post_fetch(**_kwargs):
|
|
return {'data': {'created_id': 420323}}
|
|
|
|
async def fake_fetch(**_kwargs):
|
|
nonlocal status_calls
|
|
status_calls += 1
|
|
return {'data': {'status_name': 'waiting'}}
|
|
|
|
async def no_sleep(*_args, **_kwargs):
|
|
return None
|
|
|
|
monkeypatch.setattr(pentesttools.AsyncFetcher, 'post_fetch', fake_post_fetch)
|
|
monkeypatch.setattr(pentesttools.AsyncFetcher, 'fetch', fake_fetch)
|
|
monkeypatch.setattr(pentesttools.asyncio, 'sleep', no_sleep)
|
|
caplog.set_level(logging.INFO, logger=pentesttools.__name__)
|
|
|
|
report = await pentesttools.SearchPentestTools('example.test').process()
|
|
|
|
assert status_calls == 10
|
|
assert report == SourceExecutionReport('partial', 'runtime-limit')
|
|
assert 'still waiting after 10 status checks' in caplog.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_polling_cancellation_propagates(monkeypatch) -> None:
|
|
monkeypatch.setattr(pentesttools.Core, 'pentest_tools_key', lambda: 'test-key')
|
|
|
|
async def fake_post_fetch(**_kwargs):
|
|
return {'data': {'created_id': 420323}}
|
|
|
|
async def cancel(*_args, **_kwargs):
|
|
raise asyncio.CancelledError
|
|
|
|
monkeypatch.setattr(pentesttools.AsyncFetcher, 'post_fetch', fake_post_fetch)
|
|
monkeypatch.setattr(pentesttools.asyncio, 'sleep', cancel)
|
|
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await pentesttools.SearchPentestTools('example.test').process()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_provider_timeout_returns_explicit_transport_error(monkeypatch) -> None:
|
|
monkeypatch.setattr(pentesttools.Core, 'pentest_tools_key', lambda: 'test-key')
|
|
|
|
async def timeout(**_kwargs):
|
|
raise TimeoutError
|
|
|
|
monkeypatch.setattr(pentesttools.AsyncFetcher, 'post_fetch', timeout)
|
|
|
|
assert await pentesttools.SearchPentestTools('example.test').process() == SourceExecutionReport('failed', 'transport-error')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_malformed_status_response_completes_without_evidence(monkeypatch, caplog) -> None:
|
|
monkeypatch.setattr(pentesttools.Core, 'pentest_tools_key', lambda: 'test-key')
|
|
|
|
async def fake_post_fetch(**_kwargs):
|
|
return {'data': {'created_id': 420323}}
|
|
|
|
async def fake_fetch(**_kwargs):
|
|
return {'data': {}}
|
|
|
|
async def no_sleep(*_args, **_kwargs):
|
|
return None
|
|
|
|
monkeypatch.setattr(pentesttools.AsyncFetcher, 'post_fetch', fake_post_fetch)
|
|
monkeypatch.setattr(pentesttools.AsyncFetcher, 'fetch', fake_fetch)
|
|
monkeypatch.setattr(pentesttools.asyncio, 'sleep', no_sleep)
|
|
caplog.set_level(logging.INFO, logger=pentesttools.__name__)
|
|
|
|
search = pentesttools.SearchPentestTools('example.test')
|
|
await search.process()
|
|
|
|
assert not await search.get_hostnames()
|
|
assert 'malformed status response' in caplog.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_null_output_data_completes_without_evidence(monkeypatch) -> None:
|
|
monkeypatch.setattr(pentesttools.Core, 'pentest_tools_key', lambda: 'test-key')
|
|
responses = iter(
|
|
[
|
|
{'data': {'status_name': 'finished'}},
|
|
{'data': {'output_type': 'subdomain_list', 'output_data': None}},
|
|
]
|
|
)
|
|
|
|
async def fake_post_fetch(**_kwargs):
|
|
return {'data': {'created_id': 420323}}
|
|
|
|
async def fake_fetch(**_kwargs):
|
|
return next(responses)
|
|
|
|
async def no_sleep(*_args, **_kwargs):
|
|
return None
|
|
|
|
monkeypatch.setattr(pentesttools.AsyncFetcher, 'post_fetch', fake_post_fetch)
|
|
monkeypatch.setattr(pentesttools.AsyncFetcher, 'fetch', fake_fetch)
|
|
monkeypatch.setattr(pentesttools.asyncio, 'sleep', no_sleep)
|
|
|
|
search = pentesttools.SearchPentestTools('example.test')
|
|
await search.process()
|
|
|
|
assert not await search.get_hostnames()
|
|
assert not await search.get_ips()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_subdomain_output_is_not_collected(monkeypatch) -> None:
|
|
monkeypatch.setattr(pentesttools.Core, 'pentest_tools_key', lambda: 'test-key')
|
|
|
|
async def fake_post_fetch(**_kwargs):
|
|
return {'data': {'created_id': 420323}}
|
|
|
|
responses = iter(
|
|
[
|
|
{'data': {'status_name': 'finished'}},
|
|
{
|
|
'data': {
|
|
'output_type': 'finding_list',
|
|
'output_data': {'subdomains': [{'hostname': 'injected.example.test', 'ip_address': '203.0.113.10'}]},
|
|
}
|
|
},
|
|
]
|
|
)
|
|
|
|
async def fake_fetch(**_kwargs):
|
|
return next(responses)
|
|
|
|
async def no_sleep(*_args, **_kwargs):
|
|
return None
|
|
|
|
monkeypatch.setattr(pentesttools.AsyncFetcher, 'post_fetch', fake_post_fetch)
|
|
monkeypatch.setattr(pentesttools.AsyncFetcher, 'fetch', fake_fetch)
|
|
monkeypatch.setattr(pentesttools.asyncio, 'sleep', no_sleep)
|
|
|
|
search = pentesttools.SearchPentestTools('example.test')
|
|
await search.process()
|
|
|
|
assert not await search.get_hostnames()
|
|
assert not await search.get_ips()
|
|
|
|
|
|
pytestmark = pytest.mark.provider_contract('pentesttools')
|