mirror of
https://github.com/laramies/theHarvester.git
synced 2026-09-27 12:04:51 +02:00
Merge pull request #2408 from NotoriousRebel/codex/upstream-dev-configuration-boundary
Introduce lazy and injectable configuration boundaries
This commit is contained in:
Matt
@@ -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
|
||||
@@ -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)
|
||||
@@ -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'
|
||||
|
||||
@@ -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
|
||||
@@ -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]:
|
||||
|
||||
Reference in New Issue
Block a user