mirror of
https://github.com/laramies/theHarvester.git
synced 2026-09-23 10:04:56 +02:00
317 lines
11 KiB
Python
317 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
import pytest
|
|
|
|
from theHarvester.discovery import githubcode
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import AsyncIterator
|
|
|
|
|
|
class FakeResponse:
|
|
status = 200
|
|
|
|
def __init__(self, payload: dict[str, Any], links: dict[str, Any]) -> None:
|
|
self.payload = payload
|
|
self.links = links
|
|
|
|
async def __aenter__(self) -> FakeResponse:
|
|
return self
|
|
|
|
async def __aexit__(self, *_args: Any) -> None:
|
|
return None
|
|
|
|
async def text(self) -> str:
|
|
return ''
|
|
|
|
async def json(self) -> dict[str, Any]:
|
|
return self.payload
|
|
|
|
|
|
@pytest.fixture
|
|
def install_github_responses(monkeypatch: pytest.MonkeyPatch):
|
|
requested_urls: list[str] = []
|
|
|
|
def install(*responses: FakeResponse) -> list[str]:
|
|
response_iterator = iter(responses)
|
|
|
|
class FakeSession:
|
|
def __init__(self, *, headers: dict[str, str]) -> None:
|
|
pass
|
|
|
|
async def __aenter__(self) -> FakeSession:
|
|
return self
|
|
|
|
async def __aexit__(self, *_args: Any) -> None:
|
|
return None
|
|
|
|
def get(self, url: str, *, proxy: str | None = None) -> FakeResponse:
|
|
requested_urls.append(url)
|
|
return next(response_iterator)
|
|
|
|
@contextlib.asynccontextmanager
|
|
async def fake_open_session(**_kwargs: object) -> AsyncIterator[FakeSession]:
|
|
async with FakeSession(headers={}) as session:
|
|
yield session
|
|
|
|
monkeypatch.setattr(githubcode.AsyncFetcher, 'open_session', fake_open_session)
|
|
return requested_urls
|
|
|
|
monkeypatch.setattr(githubcode.Core, 'github_key', staticmethod(lambda: 'test-token'))
|
|
monkeypatch.setattr(githubcode.Core, 'get_user_agent', staticmethod(lambda: 'test-agent'))
|
|
monkeypatch.setattr(githubcode, 'get_delay', lambda: 0)
|
|
return install
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_github_code_retains_only_the_requested_fragments_across_pages(install_github_responses) -> None:
|
|
requested_urls = install_github_responses(
|
|
FakeResponse(
|
|
{
|
|
'items': [
|
|
{'text_matches': [{'fragment': 'Contact Admin@Example.COM'}]},
|
|
{'text_matches': [{'fragment': 'API host API.EXAMPLE.COM'}]},
|
|
]
|
|
},
|
|
{
|
|
'next': {'url': 'https://api.github.com/search/code?q=example.com&page=2'},
|
|
'last': {'url': 'https://api.github.com/search/code?q=example.com&page=2'},
|
|
},
|
|
),
|
|
FakeResponse(
|
|
{
|
|
'items': [
|
|
{'text_matches': [{'fragment': 'Docs host Docs.Example.Com'}]},
|
|
{'text_matches': [{'fragment': 'Ignored host ignored.example.com'}]},
|
|
]
|
|
},
|
|
{},
|
|
),
|
|
)
|
|
search = githubcode.SearchGithubCode('example.com', limit=3)
|
|
|
|
await search.process()
|
|
|
|
assert requested_urls == [
|
|
'https://api.github.com/search/code?q="example.com"&page=1',
|
|
'https://api.github.com/search/code?q="example.com"&page=2',
|
|
]
|
|
assert search.counter == 3
|
|
assert await search.get_emails() == {'admin@example.com'}
|
|
assert await search.get_hostnames() == ['api.example.com', 'docs.example.com', 'example.com']
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_github_code_exact_limit_makes_no_additional_request(install_github_responses) -> None:
|
|
requested_urls = install_github_responses(
|
|
FakeResponse(
|
|
{
|
|
'items': [
|
|
{'text_matches': [{'fragment': 'api.example.com'}]},
|
|
{'text_matches': [{'fragment': 'docs.example.com'}]},
|
|
]
|
|
},
|
|
{
|
|
'next': {'url': 'https://api.github.com/search/code?q=example.com&page=2'},
|
|
'last': {'url': 'https://api.github.com/search/code?q=example.com&page=2'},
|
|
},
|
|
)
|
|
)
|
|
search = githubcode.SearchGithubCode('example.com', limit=2)
|
|
|
|
await search.process()
|
|
|
|
assert requested_urls == ['https://api.github.com/search/code?q="example.com"&page=1']
|
|
assert await search.get_hostnames() == ['api.example.com', 'docs.example.com']
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_github_code_keeps_provider_fragments_separate(install_github_responses) -> None:
|
|
install_github_responses(
|
|
FakeResponse(
|
|
{
|
|
'items': [
|
|
{
|
|
'text_matches': [
|
|
{'fragment': 'admin'},
|
|
{'fragment': '@example.com'},
|
|
]
|
|
}
|
|
]
|
|
},
|
|
{},
|
|
)
|
|
)
|
|
search = githubcode.SearchGithubCode('example.com', limit=10)
|
|
|
|
await search.process()
|
|
|
|
assert await search.get_emails() == set()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_github_code_ignores_non_string_fragments_without_repeating_the_page(install_github_responses) -> None:
|
|
requested_urls = install_github_responses(
|
|
FakeResponse(
|
|
{
|
|
'items': [
|
|
{
|
|
'text_matches': [
|
|
{'fragment': None},
|
|
{'fragment': 42},
|
|
{'fragment': ''},
|
|
{'fragment': 'API host api.example.com'},
|
|
]
|
|
}
|
|
]
|
|
},
|
|
{},
|
|
),
|
|
FakeResponse({'items': []}, {}),
|
|
)
|
|
search = githubcode.SearchGithubCode('example.com', limit=10)
|
|
|
|
await search.process()
|
|
|
|
assert requested_urls == ['https://api.github.com/search/code?q="example.com"&page=1']
|
|
assert await search.get_hostnames() == ['api.example.com']
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'payload',
|
|
[
|
|
{'items': 'not-a-list'},
|
|
{'items': [{'text_matches': 'not-a-list'}]},
|
|
],
|
|
ids=['malformed-items', 'malformed-text-matches'],
|
|
)
|
|
@pytest.mark.asyncio
|
|
async def test_github_code_malformed_page_terminates_without_following_pagination(
|
|
install_github_responses,
|
|
payload: dict[str, Any],
|
|
) -> None:
|
|
requested_urls = install_github_responses(
|
|
FakeResponse(
|
|
payload,
|
|
{
|
|
'next': {'url': 'https://api.github.com/search/code?q=example.com&page=2'},
|
|
'last': {'url': 'https://api.github.com/search/code?q=example.com&page=2'},
|
|
},
|
|
),
|
|
FakeResponse({'items': []}, {}),
|
|
)
|
|
search = githubcode.SearchGithubCode('example.com', limit=10)
|
|
|
|
await search.process()
|
|
|
|
assert requested_urls == ['https://api.github.com/search/code?q="example.com"&page=1']
|
|
assert await search.get_emails() == set()
|
|
assert await search.get_hostnames() == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_github_code_unlimited_pagination_cycle_preserves_partial_evidence(install_github_responses) -> None:
|
|
requested_urls = install_github_responses(
|
|
FakeResponse(
|
|
{'items': [{'text_matches': [{'fragment': 'api.example.com'}]}]},
|
|
{'next': {'url': 'https://api.github.com/search/code?q=example.com&page=2'}},
|
|
),
|
|
FakeResponse(
|
|
{'items': [{'text_matches': [{'fragment': 'docs.example.com'}]}]},
|
|
{'next': {'url': 'https://api.github.com/search/code?q=example.com&page=1'}},
|
|
),
|
|
)
|
|
search = githubcode.SearchGithubCode('example.com', limit=None)
|
|
|
|
report = await search.process()
|
|
|
|
assert requested_urls == [
|
|
'https://api.github.com/search/code?q="example.com"&page=1',
|
|
'https://api.github.com/search/code?q="example.com"&page=2',
|
|
]
|
|
assert await search.get_hostnames() == ['api.example.com', 'docs.example.com']
|
|
assert report == githubcode.SourceExecutionReport('partial', 'repeated-page')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_github_code_unlimited_repeated_content_stops_before_counting_duplicates(
|
|
install_github_responses,
|
|
) -> None:
|
|
requested_urls = install_github_responses(
|
|
FakeResponse(
|
|
{
|
|
'items': [
|
|
{'text_matches': [{'fragment': 'api.example.com'}, {'fragment': 'docs.example.com'}]},
|
|
]
|
|
},
|
|
{'next': {'url': 'https://api.github.com/search/code?q=example.com&page=2'}},
|
|
),
|
|
FakeResponse(
|
|
{
|
|
'items': [
|
|
{'text_matches': [{'fragment': 'docs.example.com'}, {'fragment': 'api.example.com'}]},
|
|
]
|
|
},
|
|
{'next': {'url': 'https://api.github.com/search/code?q=example.com&page=3'}},
|
|
),
|
|
)
|
|
search = githubcode.SearchGithubCode('example.com', limit=None)
|
|
|
|
report = await search.process()
|
|
|
|
assert requested_urls == [
|
|
'https://api.github.com/search/code?q="example.com"&page=1',
|
|
'https://api.github.com/search/code?q="example.com"&page=2',
|
|
]
|
|
assert search.counter == 2
|
|
assert await search.get_hostnames() == ['api.example.com', 'docs.example.com']
|
|
assert report == githubcode.SourceExecutionReport('partial', 'repeated-page')
|
|
|
|
|
|
@pytest.mark.parametrize('fragments', [[], ['api.example.com']], ids=['no-evidence', 'partial-evidence'])
|
|
@pytest.mark.asyncio
|
|
async def test_github_code_persistent_exceptions_stop_with_truthful_report(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
fragments: list[str],
|
|
) -> None:
|
|
monkeypatch.setattr(githubcode.Core, 'github_key', staticmethod(lambda: 'test-token'))
|
|
monkeypatch.setattr(githubcode, 'get_delay', lambda: 0)
|
|
search = githubcode.SearchGithubCode('example.com', limit=None)
|
|
search.total_results = ' '.join(fragments)
|
|
search.counter = len(fragments)
|
|
search.max_retries = 1
|
|
calls = 0
|
|
|
|
async def fail_search(*_args: Any, **_kwargs: Any) -> tuple[str, dict, int, Any]:
|
|
nonlocal calls
|
|
calls += 1
|
|
raise RuntimeError('persistent provider failure')
|
|
|
|
monkeypatch.setattr(search, 'do_search', fail_search)
|
|
|
|
report = await search.process()
|
|
|
|
assert calls == 2
|
|
assert report == githubcode.SourceExecutionReport('partial' if fragments else 'failed', 'transport-error')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_github_code_cancellation_propagates(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(githubcode.Core, 'github_key', staticmethod(lambda: 'test-token'))
|
|
search = githubcode.SearchGithubCode('example.com', limit=None)
|
|
|
|
async def cancel_search(*_args: Any, **_kwargs: Any) -> tuple[str, dict, int, Any]:
|
|
raise asyncio.CancelledError('operator-stop')
|
|
|
|
monkeypatch.setattr(search, 'do_search', cancel_search)
|
|
|
|
with pytest.raises(asyncio.CancelledError, match='operator-stop'):
|
|
await search.process()
|
|
|
|
|
|
pytestmark = pytest.mark.provider_contract('github-code')
|