diff --git a/tests/lib/test_configuration_contract.py b/tests/lib/test_configuration_contract.py new file mode 100644 index 00000000..5eb5c0dd --- /dev/null +++ b/tests/lib/test_configuration_contract.py @@ -0,0 +1,209 @@ +from __future__ import annotations + +import os +import subprocess +import sys +import textwrap +from pathlib import Path +from typing import TYPE_CHECKING, NamedTuple + +import pytest +import yaml + +from theHarvester.lib.configuration import FileSystemCredentialAdapter + +if TYPE_CHECKING: + from collections.abc import Callable + from types import ModuleType + + from theHarvester.lib.core import Core + + +class ConfigurationEnvironment(NamedTuple): + core: type[Core] + module: ModuleType + directories: list[Path] + + +def test_importing_core_does_not_access_configuration_files(tmp_path: Path) -> None: + script = textwrap.dedent( + """ + import sys + + configuration_accesses: list[str] = [] + + def record_configuration_access(event: str, arguments: tuple[object, ...]) -> None: + if event != 'open': + return + path = arguments[0] + if isinstance(path, str) and path.endswith(('api-keys.yaml', 'proxies.yaml')): + configuration_accesses.append(path) + + sys.addaudithook(record_configuration_access) + + import theHarvester.lib.core + + assert configuration_accesses == [], configuration_accesses + + first_proxy_list = theHarvester.lib.core.AsyncFetcher().proxy_list + access_count = len(configuration_accesses) + second_proxy_list = theHarvester.lib.core.AsyncFetcher().proxy_list + + assert second_proxy_list is first_proxy_list + assert len(configuration_accesses) == access_count + """ + ) + environment = os.environ.copy() + environment['HOME'] = str(tmp_path) + + result = subprocess.run( + [sys.executable, '-c', script], + capture_output=True, + check=False, + cwd=Path(__file__).parents[2], + env=environment, + text=True, + ) + + assert result.returncode == 0, result.stderr + + +@pytest.fixture +def configuration_environment( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> ConfigurationEnvironment: + monkeypatch.setenv('HOME', str(tmp_path)) + import theHarvester.lib.core as core_module + + directories = [tmp_path / name for name in ('user', 'system', 'local')] + for directory in directories: + directory.mkdir() + monkeypatch.setattr(core_module, 'CONFIG_DIRS', directories) + return ConfigurationEnvironment(core_module.Core, core_module, directories) + + +@pytest.mark.parametrize( + ('present_indexes', 'expected_key'), + [ + ((0, 1, 2), 'user-key'), + ((1, 2), 'system-key'), + ((2,), 'local-key'), + ], +) +def test_api_keys_and_filesystem_credentials_use_first_available_configuration( + configuration_environment: ConfigurationEnvironment, + present_indexes: tuple[int, ...], + expected_key: str, +) -> None: + core = configuration_environment.core + configuration_dirs = configuration_environment.directories + keys = ('user-key', 'system-key', 'local-key') + for index in present_indexes: + (configuration_dirs[index] / 'api-keys.yaml').write_text( + f'apikeys:\n brave:\n key: {keys[index]}\n', + encoding='utf-8', + ) + + assert (core.api_keys(), FileSystemCredentialAdapter().get('brave')) == ( + {'brave': {'key': expected_key}}, + expected_key, + ) + + +def test_missing_api_keys_uses_and_creates_bundled_default( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + configuration_environment: ConfigurationEnvironment, +) -> None: + core = configuration_environment.core + core_module = configuration_environment.module + configuration_dirs = configuration_environment.directories + bundled_directory = tmp_path / 'bundled' + bundled_directory.mkdir() + bundled_content = 'apikeys:\n brave:\n key: bundled-key\n' + (bundled_directory / 'api-keys.yaml').write_text(bundled_content, encoding='utf-8') + monkeypatch.setattr(core_module, 'DATA_DIR', bundled_directory) + + assert ( + core.api_keys(), + (configuration_dirs[0] / 'api-keys.yaml').read_text(encoding='utf-8'), + ) == ( + {'brave': {'key': 'bundled-key'}}, + bundled_content, + ) + + +@pytest.mark.parametrize( + ('filename', 'configuration_reader'), + [ + ('api-keys.yaml', 'api_keys'), + ('proxies.yaml', 'proxy_list'), + ], +) +def test_malformed_configuration_raises_yaml_error( + configuration_environment: ConfigurationEnvironment, + filename: str, + configuration_reader: str, +) -> None: + core = configuration_environment.core + configuration_dirs = configuration_environment.directories + (configuration_dirs[0] / filename).write_text('key: [unterminated\n', encoding='utf-8') + reader: Callable[[], dict[str, object]] = getattr(core, configuration_reader) + + with pytest.raises(yaml.YAMLError): + reader() + + +def test_provider_accessors_return_single_and_multi_field_credentials( + configuration_environment: ConfigurationEnvironment, +) -> None: + core = configuration_environment.core + configuration_dirs = configuration_environment.directories + (configuration_dirs[0] / 'api-keys.yaml').write_text( + 'apikeys:\n bevigil:\n key: bevigil-key\n censys:\n id: censys-id\n secret: censys-secret\n', + encoding='utf-8', + ) + + assert (core.bevigil_key(), core.censys_key()) == ( + 'bevigil-key', + ('censys-id', 'censys-secret'), + ) + + +def test_brave_key_returns_configured_value( + configuration_environment: ConfigurationEnvironment, +) -> None: + core = configuration_environment.core + configuration_dirs = configuration_environment.directories + (configuration_dirs[0] / 'api-keys.yaml').write_text( + 'apikeys:\n brave:\n key: brave-key\n', + encoding='utf-8', + ) + + assert core.brave_key() == 'brave-key' + + +@pytest.mark.parametrize( + ('contents', 'expected'), + [ + ( + 'http: [proxy.local:8080]\nsocks5: [socks.local:1080]\n', + { + 'http': ['http://proxy.local:8080'], + 'socks5': ['socks5://socks.local:1080'], + }, + ), + ('http:\n', {'http': [], 'socks5': []}), + ], +) +def test_proxy_lookup_normalizes_configured_addresses( + configuration_environment: ConfigurationEnvironment, + contents: str, + expected: dict[str, list[str]], +) -> None: + core = configuration_environment.core + configuration_dirs = configuration_environment.directories + (configuration_dirs[0] / 'proxies.yaml').write_text(contents, encoding='utf-8') + + assert core.proxy_list() == expected diff --git a/tests/test_bravesearch.py b/tests/test_bravesearch.py new file mode 100644 index 00000000..35a8495c --- /dev/null +++ b/tests/test_bravesearch.py @@ -0,0 +1,49 @@ +from __future__ import annotations + +from typing import Any + +import pytest + +from theHarvester.discovery.bravesearch import SearchBrave +from theHarvester.discovery.constants import MissingKey +from theHarvester.lib.configuration import InMemoryCredentialAdapter +from theHarvester.lib.core import AsyncFetcher + + +@pytest.mark.asyncio +async def test_brave_collects_with_in_memory_credentials(monkeypatch: pytest.MonkeyPatch) -> None: + request_headers: list[dict[str, str]] = [] + + async def fetch(*, headers: dict[str, str], **_kwargs: Any) -> dict[str, Any]: + request_headers.append(headers) + return { + 'web': { + 'results': [ + { + 'title': 'Documentation', + 'description': 'Example documentation', + 'url': 'https://docs.example.com', + } + ] + } + } + + monkeypatch.setattr(AsyncFetcher, 'fetch', fetch) + search = SearchBrave( + 'example.com', + 1, + credential_adapter=InMemoryCredentialAdapter({'brave': {'key': 'memory-key'}}), + ) + + await search.process() + + assert set(await search.get_hostnames()) == {'docs.example.com'} + assert request_headers + assert {headers['X-Subscription-Token'] for headers in request_headers} == {'memory-key'} + + +def test_brave_rejects_empty_in_memory_credentials() -> None: + credentials = InMemoryCredentialAdapter({'brave': {'key': ''}}) + + with pytest.raises(MissingKey, match='Brave Search'): + SearchBrave('example.com', 1, credential_adapter=credentials) diff --git a/theHarvester/discovery/bravesearch.py b/theHarvester/discovery/bravesearch.py index 82afc7bc..cfdc42a5 100644 --- a/theHarvester/discovery/bravesearch.py +++ b/theHarvester/discovery/bravesearch.py @@ -1,20 +1,32 @@ import asyncio import logging +from typing import Any from urllib.parse import quote from theHarvester.discovery.constants import MissingKey, get_delay -from theHarvester.lib.core import AsyncFetcher, Core +from theHarvester.lib.configuration import CredentialAdapter, FileSystemCredentialAdapter +from theHarvester.lib.core import AsyncFetcher from theHarvester.parsers import myparser logger = logging.getLogger(__name__) class SearchBrave: - def __init__(self, word, limit): + """Search Brave while allowing credentials to be supplied without file access. + + Provider API: + https://api-dashboard.search.brave.com/app/documentation/web-search/query + + Filesystem credentials remain the production default; injection keeps tests and + embedded use independent of operator configuration files. + """ + + def __init__(self, word: str, limit: int, credential_adapter: CredentialAdapter | None = None) -> None: self.word = word - self.results = [] + self.results: list[dict[str, Any]] = [] self.totalresults = '' - self.api_key = Core.brave_key() + credentials = credential_adapter if credential_adapter is not None else FileSystemCredentialAdapter() + self.api_key = credentials.get('brave') if self.api_key is None or self.api_key == '': raise MissingKey('Brave Search') self.server = 'https://api.search.brave.com/res/v1/web/search' diff --git a/theHarvester/lib/configuration.py b/theHarvester/lib/configuration.py new file mode 100644 index 00000000..d90c5f8a --- /dev/null +++ b/theHarvester/lib/configuration.py @@ -0,0 +1,31 @@ +"""Credential boundaries that keep provider code independent of configuration files. + +Production access retains Core's existing file precedence. In-memory access lets +tests and embedded callers provide credentials without filesystem or global state. +""" + +from collections.abc import Mapping +from dataclasses import dataclass + +from theHarvester.lib.core import Core + + +class FileSystemCredentialAdapter: + """Read production credentials through Core's existing file precedence.""" + + @staticmethod + def get(provider: str, field: str = 'key') -> str: + return Core.api_keys()[provider][field] + + +@dataclass(frozen=True) +class InMemoryCredentialAdapter: + """Provide credentials directly for isolated tests and programmatic callers.""" + + credentials: Mapping[str, Mapping[str, str]] + + def get(self, provider: str, field: str = 'key') -> str: + return self.credentials[provider][field] + + +type CredentialAdapter = FileSystemCredentialAdapter | InMemoryCredentialAdapter diff --git a/theHarvester/lib/core.py b/theHarvester/lib/core.py index f9e9e7d4..0e9a00a0 100644 --- a/theHarvester/lib/core.py +++ b/theHarvester/lib/core.py @@ -413,7 +413,17 @@ class Core: class AsyncFetcher: - proxy_list = Core.proxy_list() + _proxy_list: ClassVar[dict | None] = None + + @property + def proxy_list(self) -> dict: + """Load and cache proxies on first use instead of during module import.""" + + proxy_list = self.__class__._proxy_list + if proxy_list is None: + proxy_list = Core.proxy_list() + self.__class__._proxy_list = proxy_list + return proxy_list @staticmethod def _default_headers(headers: dict[str, str] | None = None) -> dict[str, str]: