diff --git a/tests/discovery/test_api_endpoints.py b/tests/discovery/test_api_endpoints.py index 72504803..f752bd76 100644 --- a/tests/discovery/test_api_endpoints.py +++ b/tests/discovery/test_api_endpoints.py @@ -1107,7 +1107,7 @@ async def test_api_endpoint_scan_uses_a_configured_proxy(monkeypatch) -> None: await search.do_search() - assert requests == [(proxy, False), (proxy, False), (proxy, False)] + assert requests == [(False, False), (False, False), (False, False)] @pytest.mark.asyncio @@ -1155,10 +1155,24 @@ async def test_api_endpoint_scan_builds_one_socks_proxy_session(monkeypatch: pyt assert connector_calls == [(proxy, 'socks5')] assert len(owned_sessions) == 1 - assert requests == [(owned_sessions[0], proxy)] * 3 + assert requests == [(owned_sessions[0], False)] * 3 assert owned_sessions[0].closed is True +@pytest.mark.asyncio +async def test_api_endpoint_scan_normalizes_proxy_session_construction_failure( + monkeypatch: pytest.MonkeyPatch, +) -> None: + search = api_endpoints.SearchApiEndpoints('192.0.2.1', proxy='socks5://') + search.common_api_paths = ['/api'] + monkeypatch.setattr(search, '_load_wordlist', lambda: []) + + await search.do_search() + + assert search.scan_error_type == 'ResponseStreamError' + assert search.stop_reason == 'transport-error' + + @pytest.mark.parametrize('error', [aiohttp.ClientConnectionError(), TimeoutError()]) @pytest.mark.asyncio async def test_detect_schema_falls_back_only_when_https_cannot_connect(monkeypatch, error: Exception) -> None: diff --git a/tests/discovery/test_githubcode_contract.py b/tests/discovery/test_githubcode_contract.py index b03c71e8..1d09e498 100644 --- a/tests/discovery/test_githubcode_contract.py +++ b/tests/discovery/test_githubcode_contract.py @@ -2,6 +2,7 @@ from __future__ import annotations import asyncio import contextlib +import json from typing import TYPE_CHECKING, Any import pytest @@ -18,6 +19,8 @@ class FakeResponse: def __init__(self, payload: dict[str, Any], links: dict[str, Any]) -> None: self.payload = payload self.links = links + self.headers: dict[str, str] = {} + self.content = self async def __aenter__(self) -> FakeResponse: return self @@ -31,6 +34,9 @@ class FakeResponse: async def json(self) -> dict[str, Any]: return self.payload + async def iter_any(self): + yield json.dumps(self.payload).encode() + @pytest.fixture def install_github_responses(monkeypatch: pytest.MonkeyPatch): @@ -53,6 +59,10 @@ def install_github_responses(monkeypatch: pytest.MonkeyPatch): requested_urls.append(url) return next(response_iterator) + def request(self, method: str, url: str, **_kwargs: Any) -> FakeResponse: + assert method == 'GET' + return self.get(url) + @contextlib.asynccontextmanager async def fake_open_session(**_kwargs: object) -> AsyncIterator[FakeSession]: async with FakeSession(headers={}) as session: diff --git a/tests/discovery/test_intelxsearch.py b/tests/discovery/test_intelxsearch.py index 0742d6ea..2d70fc7c 100644 --- a/tests/discovery/test_intelxsearch.py +++ b/tests/discovery/test_intelxsearch.py @@ -1,5 +1,6 @@ from argparse import Namespace from contextlib import asynccontextmanager +import json from typing import Any import pytest @@ -14,6 +15,9 @@ class _Response: def __init__(self, payload: object, status: int = 200) -> None: self.payload = payload self.status = status + self.headers: dict[str, str] = {} + self.charset = 'utf-8' + self.content = self async def __aenter__(self): return self @@ -24,6 +28,9 @@ class _Response: async def json(self) -> object: return self.payload + async def iter_any(self): + yield json.dumps(self.payload).encode() + class _Session: def __init__(self, result: object, search_result: object | None = None) -> None: @@ -44,6 +51,11 @@ class _Session: def get(self, *_args, **_kwargs) -> _Response: return _Response(self.result.pop(0) if isinstance(self.result, list) else self.result) + def request(self, method: str, *args, **kwargs) -> _Response: + if method == 'POST': + return self.post(*args, **kwargs) + return self.get(*args, **kwargs) + @pytest.fixture(autouse=True) def proxy_aware_session(monkeypatch: pytest.MonkeyPatch) -> list[dict[str, Any]]: diff --git a/tests/discovery/test_takeover.py b/tests/discovery/test_takeover.py index e9afac6e..1bccecee 100644 --- a/tests/discovery/test_takeover.py +++ b/tests/discovery/test_takeover.py @@ -147,6 +147,17 @@ async def test_takeover_reuses_one_cookie_free_unlimited_http_session( assert shared_session.close_count == 1 +@pytest.mark.asyncio +async def test_takeover_normalizes_proxy_session_construction_failure(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(takeover.AsyncFetcher, '_proxy_list', {'http': [], 'socks5': ['socks5://']}) + scanner = takeover.TakeoverScanner([], target='example.test', nameservers=['1.1.1.1']) + + await scanner.process(proxy=True) + + assert scanner.scan_error_type == 'ResponseStreamError' + assert scanner.stop_reason == 'transport-error' + + @pytest.mark.asyncio async def test_takeover_requires_provider_dns_evidence_and_compound_http_predicates( monkeypatch: pytest.MonkeyPatch, diff --git a/tests/discovery/test_thc.py b/tests/discovery/test_thc.py index 301989e8..deb74b71 100644 --- a/tests/discovery/test_thc.py +++ b/tests/discovery/test_thc.py @@ -30,6 +30,8 @@ class FakeResponse: self._text = text self.status = status self.headers = headers or {} + self.charset = 'utf-8' + self.content = self async def __aenter__(self) -> Self: return self @@ -45,6 +47,9 @@ class FakeResponse: async def text(self) -> str: return self._text + async def iter_any(self): + yield self._text.encode() + class FakeSession: def __init__(self, **_kwargs: Any) -> None: @@ -65,6 +70,10 @@ class FakeSession: domain = parse_qs(urlparse(url).query).get('domain', ['example.com'])[0] return FakeResponse(f'WWW.{domain}\napi.{domain}\napi.{domain}\n') + def request(self, method: str, url: str, **_kwargs: Any) -> FakeResponse: + assert method == 'GET' + return self.get(url) + def session_for(*outcomes: FakeResponse | Exception) -> type[FakeSession]: remaining = iter(outcomes) diff --git a/tests/lib/test_api_v1.py b/tests/lib/test_api_v1.py index d80c7de0..c46a5cef 100644 --- a/tests/lib/test_api_v1.py +++ b/tests/lib/test_api_v1.py @@ -606,6 +606,15 @@ def test_api_rejects_direct_dns_work_in_proxy_mode(request_fields: dict[str, obj RunRequest(target='example.test', proxies=True, **request_fields) +def test_api_rejects_screenshot_capture_in_proxy_mode() -> None: + from pydantic import ValidationError + + from theHarvester.lib.api.run_models import RunRequest + + with pytest.raises(ValidationError, match='Screenshot capture supports direct transport only'): + RunRequest(target='example.test', sources=[], screenshot=True, proxies=True) + + @pytest.mark.parametrize( ('evidence_status', 'execution_status'), [('complete', 'failed'), ('partial', 'completed')], diff --git a/tests/lib/test_core.py b/tests/lib/test_core.py index 5166a7b3..f81996bd 100644 --- a/tests/lib/test_core.py +++ b/tests/lib/test_core.py @@ -489,6 +489,17 @@ async def test_open_session_fails_closed_when_proxy_mode_has_no_proxy(monkeypatc pytest.fail('a direct session must not open when proxy mode is required') +@pytest.mark.asyncio +async def test_open_session_normalizes_proxy_construction_failure(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(AsyncFetcher, '_proxy_list', {'http': [], 'socks5': ['socks5://']}) + + with AsyncFetcher.proxy_scope(True) as proxy: + with pytest.raises(ResponseStreamError, match='transport-error'): + async with AsyncFetcher.open_session(proxy=proxy): + pytest.fail('a malformed proxy must not open a session') + assert AsyncFetcher.proxy_transport_failed() is True + + @pytest.mark.asyncio async def test_open_session_preserves_project_ca_and_aiohttp_default_timeout( monkeypatch: pytest.MonkeyPatch, @@ -1027,6 +1038,23 @@ async def test_fetch_json_reuses_caller_owned_session_without_closing_it( assert len(DummySession.instances) == 1 +@pytest.mark.asyncio +async def test_fetch_json_preserves_pagination_links(monkeypatch: pytest.MonkeyPatch) -> None: + install_stream_response(monkeypatch, chunks=(b'{"items":[]}',)) + + def linked_response(self, method: str, url: str, **kwargs: Any) -> DummyResponse: + self.requests.append((method, url, kwargs)) + response = DummyResponse(chunks=(b'{"items":[]}',)) + response.links = {'next': {'url': 'https://provider.example/data?page=2'}} + return response + + monkeypatch.setattr(DummySession, 'request', linked_response) + + result = await AsyncFetcher.fetch_json('https://provider.example/data') + + assert result.links == {'next': {'url': 'https://provider.example/data?page=2'}} + + @pytest.mark.asyncio async def test_fetch_json_accepts_body_at_shared_limit(monkeypatch) -> None: install_stream_response(monkeypatch, chunks=(b'{"a":1}',)) @@ -1361,6 +1389,26 @@ async def test_post_fetch_can_include_response_metadata(monkeypatch) -> None: assert result == FetcherResponse(body='unavailable', status=503, headers={'retry-after': '30'}) +@pytest.mark.asyncio +async def test_post_fetch_enforces_explicit_response_limit(monkeypatch) -> None: + reset_dummy_sessions() + monkeypatch.setattr(core_module.aiohttp, 'ClientSession', DummySession) + + def oversized_response(self, method: str, url: str, **kwargs): + self.requests.append((method, url, kwargs)) + return DummyResponse(chunks=(b'12345',)) + + monkeypatch.setattr(DummySession, 'request', oversized_response) + + with pytest.raises(ResponseStreamError, match='response-limit'): + await AsyncFetcher.post_fetch( + 'https://example.com/api', + data='{}', + include_metadata=True, + response_byte_limit=4, + ) + + @pytest.mark.asyncio async def test_post_fetch_proxy_branch_posts_body_and_params_with_http_proxy(monkeypatch) -> None: reset_dummy_sessions() diff --git a/tests/lib/test_routeviews.py b/tests/lib/test_routeviews.py index 32aea841..ebf66718 100644 --- a/tests/lib/test_routeviews.py +++ b/tests/lib/test_routeviews.py @@ -8,7 +8,7 @@ from typing import Any import pytest import theHarvester.lib.routeviews as routeviews_module -from theHarvester.lib.core import FetcherResponse +from theHarvester.lib.core import FetcherResponse, ResponseStreamError from theHarvester.lib.network_evidence import ( BgpRouteObservation, PrefixOriginObservation, @@ -141,6 +141,22 @@ async def test_routeviews_reuses_one_session_for_every_request(monkeypatch) -> N assert sessions[0] is sessions[1] +@pytest.mark.asyncio +async def test_routeviews_reports_session_construction_failure(monkeypatch) -> None: + @asynccontextmanager + async def failed_open_session(**_kwargs: Any): + raise ResponseStreamError('transport-error') + yield + + monkeypatch.setattr(routeviews_module.AsyncFetcher, 'open_session', failed_open_session) + + result = await enrich_routeviews(['AS64500'], [], proxy=True) + + assert result.status == 'failed' + assert result.error_type == 'ResponseStreamError' + assert result.stop_reason == 'transport-error' + + @pytest.mark.asyncio async def test_routeviews_invalid_configured_key_fails_without_guest_downgrade(monkeypatch) -> None: calls, _elapsed = install_runtime(monkeypatch, [response(None, status=401)]) diff --git a/tests/lib/test_source_runner.py b/tests/lib/test_source_runner.py index 64ec6164..f9b1b915 100644 --- a/tests/lib/test_source_runner.py +++ b/tests/lib/test_source_runner.py @@ -506,6 +506,23 @@ async def test_source_runner_preserves_configured_proxy_transport_failure( assert outcome.execution.stop_reason == 'transport-error' +@pytest.mark.asyncio +async def test_source_runner_detects_proxy_failure_swallowed_by_adapter( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(AsyncFetcher, '_proxy_list', {'http': [], 'socks5': ['socks5://']}) + monkeypatch.setattr( + 'theHarvester.lib.source_runner.rocketreach.Core.rocketreach_key', + staticmethod(lambda: 'test-key'), + ) + + outcome = await run_source(SourceRequest('rocketreach', 'example.test', 25, 0, True, True)) + + assert outcome.execution.status == 'failed' + assert outcome.execution.stop_reason == 'transport-error' + assert outcome.observations == () + + @pytest.mark.asyncio async def test_source_runner_normalizes_structured_transport_failure( monkeypatch: pytest.MonkeyPatch, diff --git a/theHarvester/discovery/api_endpoints.py b/theHarvester/discovery/api_endpoints.py index 4fe9a3bd..98609916 100644 --- a/theHarvester/discovery/api_endpoints.py +++ b/theHarvester/discovery/api_endpoints.py @@ -466,29 +466,18 @@ class SearchApiEndpoints: return worker_count = min(self.concurrency, len(endpoints)) self._ssl_policy = self.verify_ssl - proxy_url, proxy_type = AsyncFetcher._resolve_proxy(self.proxy) - self.proxy = proxy_url - if proxy_type == 'socks5': - connector = await AsyncFetcher._create_connector( - proxy_url, - proxy_type, - AsyncFetcher._ssl_context(self.verify_ssl), - ) - else: + if not (self.proxy and self.proxy.startswith('socks5://')): connector = aiohttp.TCPConnector( limit=worker_count, limit_per_host=worker_count, ssl=self._ssl_policy, ) - session_options: dict[str, Any] = { - 'headers': self._get_headers(), - 'timeout': aiohttp.ClientTimeout(total=self.timeout), - 'connector': connector, - } - if proxy_type == 'http': - session_options['proxy'] = proxy_url - session = aiohttp.ClientSession( - **session_options, + session = await AsyncFetcher.create_session( + headers=self._get_headers(), + proxy=self.proxy, + request_timeout=self.timeout, + verify=self.verify_ssl, + connector=connector, ) self._session = session @@ -539,6 +528,10 @@ class SearchApiEndpoints: cancellation = error except TimeoutError: self.stop_reason = 'runtime-limit' + except ResponseStreamError as e: + self.scan_error_type = type(e).__name__ + self.stop_reason = e.reason + self.logger.error(f'Error in API endpoint scan: {e!s}', exc_info=True) except Exception as e: self.scan_error_type = type(e).__name__ self.stop_reason = 'scan-error' @@ -737,7 +730,7 @@ class SearchApiEndpoints: url=current_url, method=method, headers=headers, - proxy=self.proxy, + proxy=False, verify=self._ssl_policy, follow_redirects=False, request_timeout=self.timeout, diff --git a/theHarvester/discovery/githubcode.py b/theHarvester/discovery/githubcode.py index 17e75fe5..77581a55 100644 --- a/theHarvester/discovery/githubcode.py +++ b/theHarvester/discovery/githubcode.py @@ -4,7 +4,7 @@ import urllib.parse as urlparse from typing import TYPE_CHECKING, Any, NamedTuple from theHarvester.discovery.constants import MissingKey, get_delay -from theHarvester.lib.core import AsyncFetcher, Core +from theHarvester.lib.core import AsyncFetcher, Core, ResponseStreamError from theHarvester.lib.source_execution import SourceExecutionReport, SourceReportStatus from theHarvester.parsers import myparser @@ -122,8 +122,11 @@ class SearchGithubCode: async with AsyncFetcher.open_session(headers=self.headers, proxy=self.proxy) as owned_session: return await self.do_search(page, owned_session) url = f'{self.base_url}&page={page}' if page else self.base_url - async with session.get(url) as resp: - return await resp.text(), await resp.json(), resp.status, resp.links + response = await AsyncFetcher.fetch_json(url, session=session) + body = response.body if isinstance(response.body, dict) else {} + return '', body, response.status, response.links + except ResponseStreamError: + raise except Exception as e: logger.info(f'Error performing search: {e}') return '', {}, 500, {} @@ -187,6 +190,10 @@ class SearchGithubCode: self.page = 0 reason = 'access-denied' if result.status_code in {401, 403} else 'provider-error' return self._failure_report(reason) + except ResponseStreamError as e: + logger.info(f'Error processing page: {e}') + self.page = 0 + return self._failure_report(e.reason) except Exception as e: logger.info(f'Error processing page: {e}') self.retry_count += 1 diff --git a/theHarvester/discovery/intelxsearch.py b/theHarvester/discovery/intelxsearch.py index c090e3c5..ee977bac 100644 --- a/theHarvester/discovery/intelxsearch.py +++ b/theHarvester/discovery/intelxsearch.py @@ -6,7 +6,7 @@ from urllib.parse import urlparse import aiohttp from theHarvester.discovery.constants import MissingKey -from theHarvester.lib.core import AsyncFetcher, Core +from theHarvester.lib.core import MAX_PROVIDER_JSON_BYTES, AsyncFetcher, Core, ResponseStreamError from theHarvester.lib.hostnames import normalize_scoped_hostname from theHarvester.lib.source_execution import SourceExecutionReport from theHarvester.parsers import intelxparser @@ -56,14 +56,24 @@ class SearchIntelx: try: async with asyncio.timeout(self.MAX_RUNTIME_SECONDS): async with AsyncFetcher.open_session(headers=headers, proxy=self.proxy) as session: - async with session.post(f'{self.database}/phonebook/search', headers=headers, json=data) as response: - if response.status in {401, 403}: - return SourceExecutionReport('failed', 'access-denied') - if response.status == 429: - return SourceExecutionReport('rate-limited', 'http-429') - if not 200 <= response.status < 300: - return SourceExecutionReport('failed', f'http-{response.status}') - search_data = await response.json() + search_response = await AsyncFetcher.post_fetch( + f'{self.database}/phonebook/search', + session=session, + headers=headers, + json=True, + json_body=data, + include_metadata=True, + response_byte_limit=MAX_PROVIDER_JSON_BYTES, + ) + if search_response is None: + return SourceExecutionReport('failed', 'transport-error') + if search_response.status in {401, 403}: + return SourceExecutionReport('failed', 'access-denied') + if search_response.status == 429: + return SourceExecutionReport('rate-limited', 'http-429') + if not 200 <= search_response.status < 300: + return SourceExecutionReport('failed', f'http-{search_response.status}') + search_data = search_response.body if ( not isinstance(search_data, dict) or search_data.get('success') is False @@ -74,18 +84,19 @@ class SearchIntelx: phonebook_id = search_data['id'] while self.limit is None or collected < self.limit: page_size = min(self.PAGE_SIZE, self.limit - collected) if self.limit is not None else self.PAGE_SIZE - async with session.get( + page_response = await AsyncFetcher.fetch_json( f'{self.database}/phonebook/search/result', + session=session, headers=headers, params={'id': phonebook_id, 'limit': page_size}, - ) as response: - if response.status in {401, 403}: - return SourceExecutionReport('failed', 'access-denied') - if response.status == 429: - return SourceExecutionReport('rate-limited', 'http-429') - if not 200 <= response.status < 300: - return SourceExecutionReport('failed', f'http-{response.status}') - page = await response.json() + ) + if page_response.status in {401, 403}: + return SourceExecutionReport('failed', 'access-denied') + if page_response.status == 429: + return SourceExecutionReport('rate-limited', 'http-429') + if not 200 <= page_response.status < 300: + return SourceExecutionReport('failed', f'http-{page_response.status}') + page = page_response.body if ( not isinstance(page, dict) or isinstance(page.get('status'), bool) @@ -120,6 +131,8 @@ class SearchIntelx: return SourceExecutionReport('partial', 'runtime-limit') except asyncio.CancelledError: raise + except ResponseStreamError as error: + return SourceExecutionReport('failed', error.reason) except aiohttp.ClientError, OSError: return SourceExecutionReport('failed', 'transport-error') return None diff --git a/theHarvester/discovery/takeover.py b/theHarvester/discovery/takeover.py index 74c4e3fa..b37e6e5e 100644 --- a/theHarvester/discovery/takeover.py +++ b/theHarvester/discovery/takeover.py @@ -551,15 +551,11 @@ class TakeoverScanner: self.completed_count += 1 try: - ssl_context = AsyncFetcher._ssl_context() - proxy_url, proxy_type = AsyncFetcher._resolve_proxy(proxy) - session = await AsyncFetcher._build_session( - {'User-Agent': Core.get_browser_user_agent()}, - aiohttp.ClientTimeout(total=None), - proxy_url, - proxy_type, - ssl_context, + session = await AsyncFetcher.create_session( + headers={'User-Agent': Core.get_browser_user_agent()}, + proxy=proxy, cookie_jar=aiohttp.DummyCookieJar(), + unlimited_timeout=True, ) resolvers = tuple(TakeoverDNSResolver(nameserver) for nameserver in self.nameservers) async with asyncio.TaskGroup() as group: @@ -569,6 +565,10 @@ class TakeoverScanner: self.scan_error_type = 'CancelledError' self.stop_reason = 'cancelled' cancellation = error + except ResponseStreamError as error: + phase_error = error + self.scan_error_type = type(error).__name__ + self.stop_reason = error.reason except Exception as error: phase_error = error self.scan_error_type = type(error).__name__ diff --git a/theHarvester/discovery/thc.py b/theHarvester/discovery/thc.py index adbdc830..a010830b 100644 --- a/theHarvester/discovery/thc.py +++ b/theHarvester/discovery/thc.py @@ -4,7 +4,7 @@ from urllib.parse import urlencode import aiohttp -from theHarvester.lib.core import AsyncFetcher, Core +from theHarvester.lib.core import AsyncFetcher, Core, ResponseStreamError from theHarvester.lib.hostnames import normalize_scoped_hostname from theHarvester.lib.source_execution import SourceExecutionReport @@ -36,32 +36,32 @@ class SearchThc: async with AsyncFetcher.open_session(headers=headers, proxy=self.proxy, request_timeout=60) as session: for attempt in range(self.max_retries): try: - async with session.get(url) as response: - if response.status == 429: - rate_remaining = response.headers.get('x-ratelimit-remaining', '0') - if attempt == self.max_retries - 1: - logger.info(f'THC returned status 429 after {self.max_retries} attempts') - return SourceExecutionReport('rate-limited', 'http-429') - wait_time = self.base_delay * (attempt + 1) - logger.info( - f'THC rate limit hit (remaining: {rate_remaining}). Waiting {wait_time}s before retry...' - ) - await asyncio.sleep(wait_time) - continue + response = await AsyncFetcher.fetch_text(url, session=session, request_timeout=None) + if response.status == 429: + rate_remaining = response.headers.get('x-ratelimit-remaining', '0') + if attempt == self.max_retries - 1: + logger.info(f'THC returned status 429 after {self.max_retries} attempts') + return SourceExecutionReport('rate-limited', 'http-429') + wait_time = self.base_delay * (attempt + 1) + logger.info(f'THC rate limit hit (remaining: {rate_remaining}). Waiting {wait_time}s before retry...') + await asyncio.sleep(wait_time) + continue - if response.status != 200: - logger.info(f'THC returned status {response.status}') - return SourceExecutionReport('failed', f'http-{response.status}') + if response.status != 200: + logger.info(f'THC returned status {response.status}') + return SourceExecutionReport('failed', f'http-{response.status}') - text = await response.text() - lines = text.splitlines() - for line in lines: - if hostname := normalize_scoped_hostname(line, self.word): - self.results.add(hostname) - if len(lines) >= requested and (self.limit is None or self.limit > self.PROVIDER_MAX_RESULTS): - return SourceExecutionReport('partial', 'provider-limit') - return None + lines = response.body.splitlines() + for line in lines: + if hostname := normalize_scoped_hostname(line, self.word): + self.results.add(hostname) + if len(lines) >= requested and (self.limit is None or self.limit > self.PROVIDER_MAX_RESULTS): + return SourceExecutionReport('partial', 'provider-limit') + return None + except ResponseStreamError as e: + logger.info(f'An exception has occurred in THC: {e}') + return SourceExecutionReport('failed', e.reason) except Exception as e: error_msg = str(e).lower() if '429' in error_msg or 'rate' in error_msg: @@ -74,6 +74,9 @@ class SearchThc: continue logger.info(f'An exception has occurred in THC: {e}') return SourceExecutionReport('failed', 'transport-error') + except ResponseStreamError as e: + logger.info(f'An exception has occurred in THC: {e}') + return SourceExecutionReport('failed', e.reason) except (aiohttp.ClientError, OSError, ValueError) as e: logger.info(f'An exception has occurred in THC: {e}') return SourceExecutionReport('failed', 'transport-error') diff --git a/theHarvester/lib/api/run_models.py b/theHarvester/lib/api/run_models.py index 306625c8..073dafdd 100644 --- a/theHarvester/lib/api/run_models.py +++ b/theHarvester/lib/api/run_models.py @@ -273,6 +273,8 @@ class RunRequest(BaseModel): def validate_proxy_transport(self) -> Self: if not self.proxies: return self + if self.screenshot: + raise ValueError('Screenshot capture supports direct transport only; proxies must be disabled') direct_sources = [ source for source in resolve_sources(self.sources) if source in SOURCE_SPECS and source_requires_direct_dns(source) ] diff --git a/theHarvester/lib/core.py b/theHarvester/lib/core.py index 93e495e7..856b32f1 100644 --- a/theHarvester/lib/core.py +++ b/theHarvester/lib/core.py @@ -10,7 +10,7 @@ import re import ssl import tempfile from contextvars import ContextVar -from dataclasses import dataclass +from dataclasses import dataclass, field from pathlib import Path from typing import TYPE_CHECKING, Any, ClassVar, Literal @@ -83,6 +83,16 @@ class FetcherResponse: body: Any status: int headers: dict[str, str] + links: dict[str, dict[str, str]] = field(default_factory=dict) + + +def _response_links(response: aiohttp.ClientResponse) -> dict[str, dict[str, str]]: + try: + return { + str(relation): {str(name): str(value) for name, value in link.items()} for relation, link in response.links.items() + } + except AttributeError, TypeError, ValueError: + return {} def _reject_json_constant(value: str) -> None: @@ -560,10 +570,11 @@ class AsyncFetcher: proxy_type: str | None = None, ssl_context: ssl.SSLContext | bool | None = None, cookie_jar: AbstractCookieJar | None = None, + connector: aiohttp.BaseConnector | None = None, ) -> aiohttp.ClientSession: - connector = None + owns_connector = connector is None if proxy_url is not None or proxy_type is not None or ssl_context is not None: - connector = await cls._create_connector(proxy_url, proxy_type, ssl_context) + connector = connector or await cls._create_connector(proxy_url, proxy_type, ssl_context) session_kwargs: dict[str, Any] = { 'headers': headers, 'timeout': client_timeout, @@ -573,7 +584,41 @@ class AsyncFetcher: session_kwargs['proxy'] = proxy_url if cookie_jar is not None: session_kwargs['cookie_jar'] = cookie_jar - return aiohttp.ClientSession(**session_kwargs) + try: + return aiohttp.ClientSession(**session_kwargs) + except BaseException: + if owns_connector and connector is not None: + await connector.close() + raise + + @classmethod + async def create_session( + cls, + *, + headers: dict[str, str] | None = None, + proxy: str | bool | None = '', + request_timeout: int | None = None, + cookie_jar: AbstractCookieJar | None = None, + verify: bool | None = True, + unlimited_timeout: bool = False, + connector: aiohttp.BaseConnector | None = None, + ) -> aiohttp.ClientSession: + """Create a caller-owned provider session with normalized transport failures.""" + proxy_url, proxy_type = cls._resolve_proxy(proxy) + try: + build_options = {'connector': connector} if connector is not None else {} + return await cls._build_session( + cls._default_headers(headers), + aiohttp.ClientTimeout(total=None) if unlimited_timeout else cls._request_timeout(request_timeout), + proxy_url, + proxy_type, + cls._ssl_context() if verify is True else cls._ssl_context(verify), + cookie_jar, + **build_options, + ) + except (aiohttp.ClientError, TimeoutError, OSError, ssl.SSLError, ValueError) as error: + _mark_proxy_transport_failed() + raise ResponseStreamError('transport-error') from error @classmethod @contextlib.asynccontextmanager @@ -584,16 +629,15 @@ class AsyncFetcher: proxy: str | bool | None = '', request_timeout: int | None = None, cookie_jar: AbstractCookieJar | None = None, + verify: bool | None = True, ) -> AsyncIterator[aiohttp.ClientSession]: """Own one connection pool, proxy identity, and cookie jar for a provider conversation.""" - proxy_url, proxy_type = cls._resolve_proxy(proxy) - session = await cls._build_session( - cls._default_headers(headers), - cls._request_timeout(request_timeout), - proxy_url, - proxy_type, - cls._ssl_context(), - cookie_jar, + session = await cls.create_session( + headers=headers, + proxy=proxy, + request_timeout=request_timeout, + cookie_jar=cookie_jar, + verify=verify, ) body_error: BaseException | None = None try: @@ -745,6 +789,7 @@ class AsyncFetcher: json_body: dict[str, Any] | None = None, *, session: aiohttp.ClientSession | None = None, + response_byte_limit: int | None = None, ) -> Any: headers = cls._default_headers(headers) # By default, timeout is 5 minutes, changed to 12-minutes @@ -760,6 +805,7 @@ class AsyncFetcher: json=json, include_metadata=include_metadata, json_body=json_body, + response_byte_limit=response_byte_limit, ) request_kwargs: dict[str, Any] = { 'data': cls._normalize_data(data) if json_body is None else None, @@ -773,6 +819,7 @@ class AsyncFetcher: json=json, json_body=json_body, include_metadata=include_metadata, + response_byte_limit=response_byte_limit, **request_kwargs, ) except aiohttp.ClientError, TimeoutError, OSError, ssl.SSLError, UnicodeDecodeError, ValueError: @@ -919,7 +966,12 @@ class AsyncFetcher: ) as response: response_headers = {name.lower(): value for name, value in response.headers.items()} if not 200 <= response.status < 300 or response.status == 204: - return FetcherResponse(body=None, status=response.status, headers=response_headers) + return FetcherResponse( + body=None, + status=response.status, + headers=response_headers, + links=_response_links(response), + ) try: if int(response_headers.get('content-length', '0')) > MAX_PROVIDER_JSON_BYTES: raise ResponseStreamError('response-limit') @@ -935,7 +987,12 @@ class AsyncFetcher: parsed = json_loader.loads(text, parse_constant=_reject_json_constant) except (UnicodeDecodeError, ValueError, RecursionError) as error: raise ResponseStreamError('invalid-response') from error - return FetcherResponse(body=parsed, status=response.status, headers=response_headers) + return FetcherResponse( + body=parsed, + status=response.status, + headers=response_headers, + links=_response_links(response), + ) @classmethod async def fetch_text( diff --git a/theHarvester/lib/routeviews.py b/theHarvester/lib/routeviews.py index fe3c64b9..ed430e8b 100644 --- a/theHarvester/lib/routeviews.py +++ b/theHarvester/lib/routeviews.py @@ -502,4 +502,9 @@ async def enrich_routeviews( attribution. Bare ASN findings are not expanded into complete prefix inventories; that requires an explicit ASN target. """ - return await _RouteViewsRuntime(api_key, proxy=proxy).run(asns, network_seeds) + runtime = _RouteViewsRuntime(api_key, proxy=proxy) + try: + return await runtime.run(asns, network_seeds) + except ResponseStreamError as error: + runtime._record_error(type(error).__name__, error.reason, override=True) + return runtime._result()