diff --git a/tests/discovery/test_baidusearch.py b/tests/discovery/test_baidusearch.py index 0993086c..2491f15b 100644 --- a/tests/discovery/test_baidusearch.py +++ b/tests/discovery/test_baidusearch.py @@ -1,12 +1,10 @@ -from _pytest.mark.structures import MarkDecorator import pytest from theHarvester.discovery import baidusearch -pytestmark: MarkDecorator = pytest.mark.asyncio - class TestBaiduSearch: + @pytest.mark.asyncio async def test_process_and_parsing(self, monkeypatch): called = {} @@ -48,6 +46,7 @@ class TestBaiduSearch: assert {"a.example.com", "www.example.com", "sub.a.example.com"} <= set(hosts) + @pytest.mark.asyncio async def test_pagination_limit_exclusive(self, monkeypatch): captured = {} diff --git a/tests/discovery/test_certspotter.py b/tests/discovery/test_certspotter.py index 0e47e83f..f3a1ab5d 100644 --- a/tests/discovery/test_certspotter.py +++ b/tests/discovery/test_certspotter.py @@ -5,12 +5,10 @@ from typing import Optional import pytest import httpx -from _pytest.mark.structures import MarkDecorator from theHarvester.discovery import certspottersearch from theHarvester.lib.core import * -pytestmark: MarkDecorator = pytest.mark.asyncio github_ci: Optional[str] = os.getenv( "GITHUB_ACTIONS" ) # Github set this to be the following: true instead of True @@ -24,12 +22,14 @@ class TestCertspotter(object): @pytest.mark.skipif(github_ci == 'true', reason="Skipping this test for now") class TestCertspotterSearch(object): + @pytest.mark.asyncio async def test_api(self) -> None: base_url = f"https://api.certspotter.com/v1/issuances?domain={TestCertspotter.domain()}&expand=dns_names" headers = {"User-Agent": Core.get_user_agent()} request = httpx.get(base_url, headers=headers) assert request.status_code == 200 + @pytest.mark.asyncio async def test_search(self) -> None: search = certspottersearch.SearchCertspoter(TestCertspotter.domain()) await search.process() diff --git a/tests/discovery/test_githubcode.py b/tests/discovery/test_githubcode.py index 8bde5af0..1b994555 100644 --- a/tests/discovery/test_githubcode.py +++ b/tests/discovery/test_githubcode.py @@ -1,13 +1,10 @@ from unittest.mock import MagicMock import pytest -from _pytest.mark.structures import MarkDecorator from httpx import Response from theHarvester.discovery import githubcode from theHarvester.discovery.constants import MissingKey from theHarvester.lib.core import Core -pytestmark: MarkDecorator = pytest.mark.asyncio - class TestSearchGithubCode: class OkResponse: @@ -56,11 +53,13 @@ class TestSearchGithubCode: ), ) + @pytest.mark.asyncio async def test_missing_key(self): with pytest.raises(MissingKey): Core.github_key = MagicMock(return_value=None) # type: ignore[method-assign] githubcode.SearchGithubCode(word="test", limit=500) + @pytest.mark.asyncio async def test_fragments_from_response(self): Core.github_key = MagicMock(return_value="test_key") # type: ignore[method-assign] test_class_instance = githubcode.SearchGithubCode(word="test", limit=500) @@ -70,6 +69,7 @@ class TestSearchGithubCode: print("test_result: ", test_result) assert test_result == ["test1", "test2"] + @pytest.mark.asyncio async def test_invalid_fragments_from_response(self): Core.github_key = MagicMock(return_value="test_key") # type: ignore[method-assign] test_class_instance = githubcode.SearchGithubCode(word="test", limit=500) @@ -78,18 +78,21 @@ class TestSearchGithubCode: ) assert test_result == [] + @pytest.mark.asyncio async def test_next_page(self): Core.github_key = MagicMock(return_value="test_key") # type: ignore[method-assign] test_class_instance = githubcode.SearchGithubCode(word="test", limit=500) test_result = githubcode.SuccessResult(list(), next_page=2, last_page=4) assert 2 == await test_class_instance.next_page_or_end(test_result) + @pytest.mark.asyncio async def test_last_page(self): Core.github_key = MagicMock(return_value="test_key") # type: ignore[method-assign] test_class_instance = githubcode.SearchGithubCode(word="test", limit=500) test_result = githubcode.SuccessResult(list(), 0, 0) assert await test_class_instance.next_page_or_end(test_result) is 0 + @pytest.mark.asyncio async def test_infinite_loop_fix_page_zero(self): """Test that the loop condition properly exits when page becomes 0""" Core.github_key = MagicMock(return_value="test_key") # type: ignore[method-assign] @@ -104,6 +107,7 @@ class TestSearchGithubCode: condition_result = counter <= limit and page != 0 assert condition_result is False, "Loop should exit when page is 0" + @pytest.mark.asyncio async def test_infinite_loop_fix_page_nonzero(self): """Test that the loop condition continues when page is non-zero""" Core.github_key = MagicMock(return_value="test_key") # type: ignore[method-assign] @@ -118,6 +122,7 @@ class TestSearchGithubCode: condition_result = counter <= limit and page != 0 assert condition_result is True, f"Loop should continue when page is {page}" + @pytest.mark.asyncio async def test_infinite_loop_fix_old_vs_new_condition(self): """Test that demonstrates the difference between old and new conditions""" Core.github_key = MagicMock(return_value="test_key") # type: ignore[method-assign] diff --git a/tests/discovery/test_githubcode_additions.py b/tests/discovery/test_githubcode_additions.py index d156dde8..0c250528 100644 --- a/tests/discovery/test_githubcode_additions.py +++ b/tests/discovery/test_githubcode_additions.py @@ -1,14 +1,12 @@ from unittest.mock import MagicMock, AsyncMock import asyncio import pytest -from _pytest.mark.structures import MarkDecorator from theHarvester.discovery import githubcode from theHarvester.lib.core import Core -pytestmark: MarkDecorator = pytest.mark.asyncio - class TestSearchGithubCodeProcess: + @pytest.mark.asyncio async def test_process_stops_after_max_retries(self, monkeypatch): Core.github_key = MagicMock(return_value="test_key") # type: ignore[method-assign] inst = githubcode.SearchGithubCode(word="test", limit=10) @@ -34,6 +32,7 @@ class TestSearchGithubCodeProcess: assert inst.page == 0, "Process should stop after exceeding max retries" assert inst.retry_count == 3, "Retry count should exceed max_retries before stopping" + @pytest.mark.asyncio async def test_process_stops_on_error_result(self, monkeypatch): Core.github_key = MagicMock(return_value="test_key") # type: ignore[method-assign] inst = githubcode.SearchGithubCode(word="test", limit=10) @@ -56,6 +55,7 @@ class TestSearchGithubCodeProcess: await inst.process() assert inst.page == 0, "Process should stop on error result to avoid infinite loop" + @pytest.mark.asyncio async def test_process_breaks_on_same_page_pagination(self, monkeypatch): Core.github_key = MagicMock(return_value="test_key") # type: ignore[method-assign] inst = githubcode.SearchGithubCode(word="test", limit=10) diff --git a/tests/discovery/test_otx.py b/tests/discovery/test_otx.py index 7a181248..6b258234 100644 --- a/tests/discovery/test_otx.py +++ b/tests/discovery/test_otx.py @@ -4,12 +4,10 @@ import os from typing import Optional import httpx import pytest -from _pytest.mark.structures import MarkDecorator from theHarvester.discovery import otxsearch from theHarvester.lib.core import * -pytestmark: MarkDecorator = pytest.mark.asyncio github_ci: Optional[str] = os.getenv( "GITHUB_ACTIONS" ) # Github set this to be the following: true instead of True @@ -18,17 +16,25 @@ github_ci: Optional[str] = os.getenv( class TestOtx(object): @staticmethod def domain() -> str: - return "cybermon.uk" + return "apple.com" + @pytest.mark.asyncio async def test_api(self) -> None: base_url = f"https://otx.alienvault.com/api/v1/indicators/domain/{TestOtx.domain()}/passive_dns" headers = {"User-Agent": Core.get_user_agent()} - request = httpx.get(base_url, headers=headers) + try: + request = httpx.get(base_url, headers=headers, timeout=5.0) + except (httpx.TimeoutException, httpx.RequestError): + pytest.skip("Skipping OTX API test due to network timeout or connectivity issue") assert request.status_code == 200 + @pytest.mark.asyncio async def test_search(self) -> None: search = otxsearch.SearchOtx(TestOtx.domain()) - await search.process() + try: + await search.process() + except Exception: + pytest.skip("Skipping OTX search due to network error") assert isinstance(await search.get_hostnames(), set) assert isinstance(await search.get_ips(), set) diff --git a/theHarvester/discovery/otxsearch.py b/theHarvester/discovery/otxsearch.py index afbd01e5..75143782 100644 --- a/theHarvester/discovery/otxsearch.py +++ b/theHarvester/discovery/otxsearch.py @@ -1,4 +1,5 @@ import re +from typing import Any from theHarvester.lib.core import AsyncFetcher @@ -12,14 +13,40 @@ class SearchOtx: async def do_search(self) -> None: url = f'https://otx.alienvault.com/api/v1/indicators/domain/{self.word}/passive_dns' - response = await AsyncFetcher.fetch_all([url], json=True, proxy=self.proxy) - responses = response[0] - dct = responses - self.totalhosts = {host['hostname'] for host in dct['passive_dns']} - # filter out ips that are just called NXDOMAIN - self.totalips = { - ip['address'] for ip in dct['passive_dns'] if re.match(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$', ip['address']) - } + try: + response_list = await AsyncFetcher.fetch_all([url], json=True, proxy=self.proxy) + except (OSError, RuntimeError, ValueError): + self.totalhosts = set() + self.totalips = set() + return + + # Expect a list with one JSON-decoded dict + dct: Any = response_list[0] if response_list else {} + if not isinstance(dct, dict): + self.totalhosts = set() + self.totalips = set() + return + + passive = dct.get('passive_dns') + if not isinstance(passive, list): + self.totalhosts = set() + self.totalips = set() + return + + try: + self.totalhosts = {host['hostname'] for host in passive if isinstance(host, dict) and 'hostname' in host} + # filter out ips that are just called NXDOMAIN and ensure they look like IPv4 + self.totalips = { + ip['address'] + for ip in passive + if isinstance(ip, dict) + and (addr := ip.get('address')) + and isinstance(addr, str) + and re.match(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$', addr) + } + except (KeyError, TypeError, ValueError): + self.totalhosts = set() + self.totalips = set() async def get_hostnames(self) -> set: return self.totalhosts