diff --git a/.github/workflows/provider-smoke.yml b/.github/workflows/provider-smoke.yml new file mode 100644 index 00000000..7a88bf43 --- /dev/null +++ b/.github/workflows/provider-smoke.yml @@ -0,0 +1,80 @@ +name: Passive provider smoke + +on: + workflow_dispatch: + +permissions: + contents: read + +jobs: + Passive-provider-smoke: + runs-on: ubuntu-latest + timeout-minutes: 30 + env: + # IANA-reserved test data; keep this job passive. + SMOKE_TEST_DOMAIN: example.com + + steps: + - name: Harden the runner (Audit all outbound calls) + uses: step-security/harden-runner@bf7454d06d71f1098171f2acdf0cd4708d7b5920 # v2.20.0 + with: + egress-policy: audit + + - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 + with: + persist-credentials: false + + - name: Install uv + uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2 + with: + python-version: '3.14' + enable-cache: true + cache-dependency-glob: "uv.lock" + + - name: Install dependencies + run: | + sudo mkdir -p /usr/local/etc/theHarvester + sudo cp theHarvester/data/*.yaml /usr/local/etc/theHarvester/ + sudo chown -R runner:runner /usr/local/etc/theHarvester/ + uv sync --all-groups --frozen + echo "$GITHUB_WORKSPACE/.venv/bin" >> $GITHUB_PATH + + - name: Run opt-in live provider tests + timeout-minutes: 5 + run: pytest --run-live-network -m live_network + + - name: Run theHarvester module Baidu + timeout-minutes: 5 + run: theHarvester -d "$SMOKE_TEST_DOMAIN" -b baidu + + - name: Run theHarvester module CertSpotter + timeout-minutes: 5 + run: theHarvester -d "$SMOKE_TEST_DOMAIN" -b certspotter + + - name: Run theHarvester module Crtsh + timeout-minutes: 5 + run: theHarvester -d "$SMOKE_TEST_DOMAIN" -b crtsh + + - name: Run theHarvester module DuckDuckGo + timeout-minutes: 5 + run: theHarvester -d "$SMOKE_TEST_DOMAIN" -b duckduckgo + + - name: Run theHarvester module HackerTarget + timeout-minutes: 5 + run: theHarvester -d "$SMOKE_TEST_DOMAIN" -b hackertarget + + - name: Run theHarvester module Otx + timeout-minutes: 5 + run: theHarvester -d "$SMOKE_TEST_DOMAIN" -b otx + + - name: Run theHarvester module RapidDns + timeout-minutes: 5 + run: theHarvester -d "$SMOKE_TEST_DOMAIN" -b rapiddns + + - name: Run theHarvester module Urlscan + timeout-minutes: 5 + run: theHarvester -d "$SMOKE_TEST_DOMAIN" -b urlscan + + - name: Run theHarvester module Yahoo + timeout-minutes: 5 + run: theHarvester -d "$SMOKE_TEST_DOMAIN" -b yahoo diff --git a/.github/workflows/theHarvester.yml b/.github/workflows/theHarvester.yml index 797d4feb..72aef7d5 100644 --- a/.github/workflows/theHarvester.yml +++ b/.github/workflows/theHarvester.yml @@ -59,82 +59,4 @@ jobs: - name: Test with pytest run: | - pytest tests/** - - Passive-provider-smoke: - if: github.event_name == 'workflow_dispatch' - runs-on: ubuntu-latest - timeout-minutes: 30 - env: - # IANA-reserved test data; keep this job passive. - SMOKE_TEST_DOMAIN: example.com - - steps: - - name: Harden the runner (Audit all outbound calls) - uses: step-security/harden-runner@bf7454d06d71f1098171f2acdf0cd4708d7b5920 # v2.20.0 - with: - egress-policy: audit - - - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 - with: - persist-credentials: false - - - name: Install uv - uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2 - with: - python-version: '3.14' - enable-cache: true - cache-dependency-glob: "uv.lock" - - - name: Install dependencies - run: | - sudo mkdir -p /usr/local/etc/theHarvester - sudo cp theHarvester/data/*.yaml /usr/local/etc/theHarvester/ - sudo chown -R runner:runner /usr/local/etc/theHarvester/ - uv sync --all-groups --frozen - echo "$GITHUB_WORKSPACE/.venv/bin" >> $GITHUB_PATH - - - name: Run theHarvester module Baidu - timeout-minutes: 5 - run: | - theHarvester -d "$SMOKE_TEST_DOMAIN" -b baidu - - - name: Run theHarvester module CertSpotter - timeout-minutes: 5 - run: | - theHarvester -d "$SMOKE_TEST_DOMAIN" -b certspotter - - - name: Run theHarvester module Crtsh - timeout-minutes: 5 - run: | - theHarvester -d "$SMOKE_TEST_DOMAIN" -b crtsh - - - name: Run theHarvester module DuckDuckGo - timeout-minutes: 5 - run: | - theHarvester -d "$SMOKE_TEST_DOMAIN" -b duckduckgo - - - name: Run theHarvester module HackerTarget - timeout-minutes: 5 - run: | - theHarvester -d "$SMOKE_TEST_DOMAIN" -b hackertarget - - - name: Run theHarvester module Otx - timeout-minutes: 5 - run: | - theHarvester -d "$SMOKE_TEST_DOMAIN" -b otx - - - name: Run theHarvester module RapidDns - timeout-minutes: 5 - run: | - theHarvester -d "$SMOKE_TEST_DOMAIN" -b rapiddns - - - name: Run theHarvester module Urlscan - timeout-minutes: 5 - run: | - theHarvester -d "$SMOKE_TEST_DOMAIN" -b urlscan - - - name: Run theHarvester module Yahoo - timeout-minutes: 5 - run: | - theHarvester -d "$SMOKE_TEST_DOMAIN" -b yahoo + pytest diff --git a/pyproject.toml b/pyproject.toml index ae3576d6..7b97251c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -69,7 +69,10 @@ restfulHarvest = "theHarvester.restfulHarvest:main" minversion = "8.3.3" asyncio_mode = "auto" asyncio_default_fixture_loop_scope = "function" -addopts = "--no-header" +addopts = "--no-header --strict-markers" +markers = [ + "live_network: contacts an external service and runs only with --run-live-network", +] testpaths = ["tests"] [build-system] diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 00000000..d0f36ff5 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,146 @@ +from __future__ import annotations + +import ipaddress +import socket +from typing import Any + +import pytest + +NETWORK_GUARD = pytest.StashKey[pytest.MonkeyPatch]() + +_getaddrinfo = socket.getaddrinfo +_gethostbyaddr = socket.gethostbyaddr +_gethostbyname = socket.gethostbyname +_gethostbyname_ex = socket.gethostbyname_ex +_getnameinfo = socket.getnameinfo +_connect = socket.socket.connect +_connect_ex = socket.socket.connect_ex +_sendto = socket.socket.sendto + +_ERROR = ( + 'External networking through Python socket APIs is disabled in routine tests. ' + 'Mock the boundary or mark the test with @pytest.mark.live_network and pass ' + '--run-live-network -m live_network.' +) + + +def pytest_addoption(parser: pytest.Parser) -> None: + parser.addoption( + '--run-live-network', + action='store_true', + default=False, + help='run tests that contact external services', + ) + + +def pytest_sessionstart(session: pytest.Session) -> None: + guard = pytest.MonkeyPatch() + guard.setattr(socket, 'getaddrinfo', _guarded_getaddrinfo) + guard.setattr(socket, 'gethostbyaddr', _guarded_gethostbyaddr) + guard.setattr(socket, 'gethostbyname', _guarded_gethostbyname) + guard.setattr(socket, 'gethostbyname_ex', _guarded_gethostbyname_ex) + guard.setattr(socket, 'getnameinfo', _guarded_getnameinfo) + guard.setattr(socket.socket, 'connect', _guarded_connect) + guard.setattr(socket.socket, 'connect_ex', _guarded_connect_ex) + guard.setattr(socket.socket, 'sendto', _guarded_sendto) + session.config.stash[NETWORK_GUARD] = guard + + +def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item]) -> None: + if config.getoption('--run-live-network'): + if config.getoption('markexpr') != 'live_network': + raise pytest.UsageError('--run-live-network requires -m live_network') + return + + skip_live = pytest.mark.skip(reason='requires --run-live-network -m live_network') + for item in items: + if item.get_closest_marker('live_network') is not None: + item.add_marker(skip_live) + + +def pytest_collection_finish(session: pytest.Session) -> None: + if session.config.getoption('--run-live-network'): + _remove_network_guard(session.config) + + +def pytest_sessionfinish(session: pytest.Session) -> None: + _remove_network_guard(session.config) + + +def _remove_network_guard(config: pytest.Config) -> None: + guard = config.stash.get(NETWORK_GUARD, None) + if guard is not None: + guard.undo() + del config.stash[NETWORK_GUARD] + + +def _is_loopback_host(host: object) -> bool: + if host is None: + return True + if isinstance(host, bytes): + host = host.decode(errors='ignore') + if not isinstance(host, str): + return False + + normalized = host.strip('[]').split('%', 1)[0].lower() + if normalized in {'localhost', 'localhost.localdomain'}: + return True + try: + return ipaddress.ip_address(normalized).is_loopback + except ValueError: + return False + + +def _is_loopback_address(family: int, address: object) -> bool: + if family == socket.AF_UNIX: + return True + return isinstance(address, tuple) and bool(address) and _is_loopback_host(address[0]) + + +def _guarded_getaddrinfo(host: bytes | str | None, *args: Any, **kwargs: Any) -> list[tuple[Any, ...]]: + if not _is_loopback_host(host): + raise AssertionError(f'{_ERROR} Attempted host: {host!r}.') + return _getaddrinfo(host, *args, **kwargs) + + +def _guarded_gethostbyaddr(host: str) -> tuple[str, list[str], list[str]]: + if not _is_loopback_host(host): + raise AssertionError(f'{_ERROR} Attempted host: {host!r}.') + return _gethostbyaddr(host) + + +def _guarded_gethostbyname(host: str) -> str: + if not _is_loopback_host(host): + raise AssertionError(f'{_ERROR} Attempted host: {host!r}.') + return _gethostbyname(host) + + +def _guarded_gethostbyname_ex(host: str) -> tuple[str, list[str], list[str]]: + if not _is_loopback_host(host): + raise AssertionError(f'{_ERROR} Attempted host: {host!r}.') + return _gethostbyname_ex(host) + + +def _guarded_getnameinfo(address: tuple[Any, ...], flags: int) -> tuple[str, str]: + if not address or not _is_loopback_host(address[0]): + raise AssertionError(f'{_ERROR} Attempted address: {address!r}.') + return _getnameinfo(address, flags) + + +def _guarded_connect(sock: socket.socket, address: object) -> None: + if not _is_loopback_address(sock.family, address): + raise AssertionError(f'{_ERROR} Attempted address: {address!r}.') + _connect(sock, address) # type: ignore[arg-type] + + +def _guarded_connect_ex(sock: socket.socket, address: object) -> int: + if not _is_loopback_address(sock.family, address): + raise AssertionError(f'{_ERROR} Attempted address: {address!r}.') + return _connect_ex(sock, address) # type: ignore[arg-type] + + +def _guarded_sendto(sock: socket.socket, data: bytes, *args: Any) -> int: + address = args[-1] if args else None + if not _is_loopback_address(sock.family, address): + raise AssertionError(f'{_ERROR} Attempted address: {address!r}.') + return _sendto(sock, data, *args) diff --git a/tests/discovery/test_certspotter.py b/tests/discovery/test_certspotter.py index f3a1ab5d..422c4b75 100644 --- a/tests/discovery/test_certspotter.py +++ b/tests/discovery/test_certspotter.py @@ -1,39 +1,40 @@ #!/usr/bin/env python3 # coding=utf-8 -import os -from typing import Optional +from typing import Any -import pytest import httpx +import pytest from theHarvester.discovery import certspottersearch -from theHarvester.lib.core import * - -github_ci: Optional[str] = os.getenv( - "GITHUB_ACTIONS" -) # Github set this to be the following: true instead of True +from theHarvester.lib.core import Core class TestCertspotter(object): @staticmethod def domain() -> str: - return "metasploit.com" + return 'example.com' -@pytest.mark.skipif(github_ci == 'true', reason="Skipping this test for now") class TestCertspotterSearch(object): - @pytest.mark.asyncio - async def test_api(self) -> None: + @pytest.mark.live_network + 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) + request = httpx.get(base_url, headers=headers, timeout=30) assert request.status_code == 200 + payload = request.json() + assert isinstance(payload, list) + assert all(isinstance(item, dict) for item in payload) @pytest.mark.asyncio - async def test_search(self) -> None: + async def test_search(self, monkeypatch: pytest.MonkeyPatch) -> None: + async def fake_fetch_all(*_args: Any, **_kwargs: Any) -> list[list[dict[str, list[str]]]]: + return [[{'dns_names': ['api.example.com', 'www.example.com']}]] + + monkeypatch.setattr(certspottersearch.AsyncFetcher, 'fetch_all', fake_fetch_all) search = certspottersearch.SearchCertspoter(TestCertspotter.domain()) await search.process() - assert isinstance(await search.get_hostnames(), set) + assert await search.get_hostnames() == {'api.example.com', 'www.example.com'} if __name__ == "__main__": diff --git a/tests/discovery/test_otx.py b/tests/discovery/test_otx.py index e5b0a037..ec57d7c2 100644 --- a/tests/discovery/test_otx.py +++ b/tests/discovery/test_otx.py @@ -1,32 +1,44 @@ #!/usr/bin/env python3 # coding=utf-8 -import os -from typing import Optional +from typing import Any + import httpx import pytest from theHarvester.discovery import otxsearch -from theHarvester.lib.core import * - -github_ci: Optional[str] = os.getenv( - "GITHUB_ACTIONS" -) # Github set this to be the following: true instead of True +from theHarvester.lib.core import Core class TestOtx(object): @staticmethod def domain() -> str: - return "apple.com" + return 'example.com' + + @pytest.mark.live_network + def test_api(self) -> None: + url = f'https://otx.alienvault.com/api/v1/indicators/domain/{self.domain()}/passive_dns' + response = httpx.get(url, headers={'User-Agent': Core.get_user_agent()}, timeout=30) + + assert response.status_code == 200 + assert isinstance(response.json().get('passive_dns'), list) @pytest.mark.asyncio - async def test_search(self) -> None: + async def test_search(self, monkeypatch: pytest.MonkeyPatch) -> None: + async def fake_fetch_all(*_args: Any, **_kwargs: Any) -> list[dict[str, list[dict[str, str]]]]: + return [ + { + 'passive_dns': [ + {'hostname': 'api.example.com', 'address': '192.0.2.1'}, + {'hostname': 'www.example.com', 'address': 'NXDOMAIN'}, + ] + } + ] + + monkeypatch.setattr(otxsearch.AsyncFetcher, 'fetch_all', fake_fetch_all) search = otxsearch.SearchOtx(TestOtx.domain()) - try: - await search.process() - except (httpx.TimeoutException, httpx.RequestError): - pytest.skip("Skipping OTX search due to network error") - assert isinstance(await search.get_hostnames(), set) - assert isinstance(await search.get_ips(), set) + await search.process() + assert await search.get_hostnames() == {'api.example.com', 'www.example.com'} + assert await search.get_ips() == {'192.0.2.1'} if __name__ == "__main__": diff --git a/tests/discovery/test_thc.py b/tests/discovery/test_thc.py index 57114bbd..957700f4 100644 --- a/tests/discovery/test_thc.py +++ b/tests/discovery/test_thc.py @@ -10,8 +10,9 @@ THC provides multiple endpoints: API Documentation: https://ip.thc.org/docs/ """ -import os -from typing import Optional +from types import TracebackType +from typing import Any, Self +from urllib.parse import parse_qs, urlparse import httpx import pytest @@ -19,60 +20,90 @@ import pytest from theHarvester.discovery import thc from theHarvester.lib.core import Core -github_ci: Optional[str] = os.getenv('GITHUB_ACTIONS') + +class FakeResponse: + status = 200 + headers: dict[str, str] = {} + + def __init__(self, text: str) -> None: + self._text = text + + async def __aenter__(self) -> Self: + return self + + async def __aexit__( + self, + _exc_type: type[BaseException] | None, + _exc: BaseException | None, + _tb: TracebackType | None, + ) -> bool: + return False + + async def text(self) -> str: + return self._text + + +class FakeSession: + def __init__(self, **_kwargs: Any) -> None: + pass + + async def __aenter__(self) -> Self: + return self + + async def __aexit__( + self, + _exc_type: type[BaseException] | None, + _exc: BaseException | None, + _tb: TracebackType | None, + ) -> bool: + return False + + def get(self, url: str) -> FakeResponse: + domain = parse_qs(urlparse(url).query).get('domain', ['example.com'])[0] + return FakeResponse(f'WWW.{domain}\napi.{domain}\napi.{domain}\n') + + +@pytest.fixture(autouse=True) +def fake_thc_session(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(thc.aiohttp, 'ClientSession', FakeSession) # ============================================================================= # 1. Direct API Tests (Endpoint Validation) # ============================================================================= +@pytest.mark.live_network class TestThcApi: """Tests to validate that the THC API responds correctly.""" - @pytest.mark.asyncio - async def test_api_subdomains_download_endpoint_responds(self) -> None: + def test_api_subdomains_download_endpoint_responds(self) -> None: """Verify that the subdomain download endpoint responds.""" - url = 'https://ip.thc.org/api/v1/subdomains/download?domain=google.com&limit=10&hide_header=true' + url = 'https://ip.thc.org/api/v1/subdomains/download?domain=example.com&limit=10&hide_header=true' headers = {'User-Agent': Core.get_user_agent()} - try: - response = httpx.get(url, headers=headers, timeout=30) - assert response.status_code == 200 - except (httpx.TimeoutException, httpx.RequestError): - pytest.skip('Skipping due to network error') + response = httpx.get(url, headers=headers, timeout=30) + assert response.status_code == 200 - @pytest.mark.asyncio - async def test_api_subdomains_returns_text_format(self) -> None: + def test_api_subdomains_returns_text_format(self) -> None: """Verify that the response is plain text.""" - url = 'https://ip.thc.org/api/v1/subdomains/download?domain=google.com&limit=5&hide_header=true' + url = 'https://ip.thc.org/api/v1/subdomains/download?domain=example.com&limit=5&hide_header=true' headers = {'User-Agent': Core.get_user_agent()} - try: - response = httpx.get(url, headers=headers, timeout=30) - content_type = response.headers.get('content-type', '') - assert 'text' in content_type or 'octet-stream' in content_type or response.status_code == 200 - except (httpx.TimeoutException, httpx.RequestError): - pytest.skip('Skipping due to network error') + response = httpx.get(url, headers=headers, timeout=30) + content_type = response.headers.get('content-type', '') + assert 'text' in content_type or 'octet-stream' in content_type - @pytest.mark.asyncio - async def test_api_cli_subdomain_endpoint(self) -> None: + def test_api_cli_subdomain_endpoint(self) -> None: """Verify CLI endpoint /sb/{domain}.""" - url = 'https://ip.thc.org/sb/google.com?l=5&noheader' + url = 'https://ip.thc.org/sb/example.com?l=5&noheader' headers = {'User-Agent': Core.get_user_agent()} - try: - response = httpx.get(url, headers=headers, timeout=30) - assert response.status_code == 200 - except (httpx.TimeoutException, httpx.RequestError): - pytest.skip('Skipping due to network error') + response = httpx.get(url, headers=headers, timeout=30) + assert response.status_code == 200 - @pytest.mark.asyncio - async def test_api_returns_rate_limit_headers(self) -> None: + def test_api_returns_rate_limit_headers(self) -> None: """Verify that the API returns rate limit headers.""" url = 'https://ip.thc.org/api/v1/subdomains/download?domain=example.com&limit=1&hide_header=true' headers = {'User-Agent': Core.get_user_agent()} - try: - response = httpx.get(url, headers=headers, timeout=30) - assert 'x-ratelimit-limit' in response.headers - assert 'x-ratelimit-remaining' in response.headers - except (httpx.TimeoutException, httpx.RequestError): - pytest.skip('Skipping due to network error') + response = httpx.get(url, headers=headers, timeout=30) + assert 'x-ratelimit-limit' in response.headers + assert 'x-ratelimit-remaining' in response.headers # ============================================================================= @@ -83,20 +114,17 @@ class TestThcSubdomainSearch: @staticmethod def domain() -> str: - return 'tesla.com' + return 'example.com' @staticmethod def small_domain() -> str: - return 'thc.org' + return 'example.com' @pytest.mark.asyncio async def test_search_returns_set(self) -> None: """Verify that get_hostnames() returns a set.""" search = thc.SearchThc(self.domain()) - try: - await search.process() - except (httpx.TimeoutException, httpx.RequestError): - pytest.skip('Skipping due to network error') + await search.process() result = await search.get_hostnames() assert isinstance(result, set) @@ -104,21 +132,15 @@ class TestThcSubdomainSearch: async def test_search_finds_subdomains(self) -> None: """Verify that it finds subdomains for a known domain.""" search = thc.SearchThc(self.domain()) - try: - await search.process() - except (httpx.TimeoutException, httpx.RequestError): - pytest.skip('Skipping due to network error') + await search.process() result = await search.get_hostnames() - assert len(result) > 0, 'Should find at least one subdomain for tesla.com' + assert len(result) > 0, 'Should find at least one subdomain for example.com' @pytest.mark.asyncio async def test_search_results_contain_target_domain(self) -> None: """Verify that all results contain the target domain.""" search = thc.SearchThc(self.small_domain()) - try: - await search.process() - except (httpx.TimeoutException, httpx.RequestError): - pytest.skip('Skipping due to network error') + await search.process() result = await search.get_hostnames() for hostname in result: assert self.small_domain() in hostname, f'{hostname} should contain {self.small_domain()}' @@ -127,10 +149,7 @@ class TestThcSubdomainSearch: async def test_search_no_duplicates(self) -> None: """Verify that there are no duplicates in the results.""" search = thc.SearchThc(self.domain()) - try: - await search.process() - except (httpx.TimeoutException, httpx.RequestError): - pytest.skip('Skipping due to network error') + await search.process() result = await search.get_hostnames() result_list = list(result) assert len(result_list) == len(set(result_list)) @@ -146,12 +165,7 @@ class TestThcEdgeCases: async def test_search_nonexistent_domain(self) -> None: """Verify behavior with non-existent domain.""" search = thc.SearchThc('this-domain-definitely-does-not-exist-12345.com') - try: - await search.process() - except (httpx.TimeoutException, httpx.RequestError): - pytest.skip('Skipping due to network error') - except Exception: - pass + await search.process() result = await search.get_hostnames() assert isinstance(result, set) @@ -159,12 +173,7 @@ class TestThcEdgeCases: async def test_search_empty_domain(self) -> None: """Verify behavior with empty domain.""" search = thc.SearchThc('') - try: - await search.process() - except (httpx.TimeoutException, httpx.RequestError): - pytest.skip('Skipping due to network error') - except Exception: - pass + await search.process() result = await search.get_hostnames() assert isinstance(result, set) @@ -172,12 +181,7 @@ class TestThcEdgeCases: async def test_search_special_characters_domain(self) -> None: """Verify behavior with special characters.""" search = thc.SearchThc('example.com; DROP TABLE domains;--') - try: - await search.process() - except (httpx.TimeoutException, httpx.RequestError): - pytest.skip('Skipping due to network error') - except Exception: - pass + await search.process() result = await search.get_hostnames() assert isinstance(result, set) @@ -185,23 +189,15 @@ class TestThcEdgeCases: async def test_search_unicode_domain(self) -> None: """Verify behavior with IDN/unicode domain.""" search = thc.SearchThc('xn--mnchen-3ya.de') - try: - await search.process() - except (httpx.TimeoutException, httpx.RequestError): - pytest.skip('Skipping due to network error') - except Exception: - pass + await search.process() result = await search.get_hostnames() assert isinstance(result, set) @pytest.mark.asyncio async def test_search_subdomain_as_input(self) -> None: """Verify behavior when a subdomain is passed as input.""" - search = thc.SearchThc('www.google.com') - try: - await search.process() - except (httpx.TimeoutException, httpx.RequestError): - pytest.skip('Skipping due to network error') + search = thc.SearchThc('www.example.com') + await search.process() result = await search.get_hostnames() assert isinstance(result, set) @@ -220,10 +216,7 @@ class TestThcProxy: async def test_process_accepts_proxy_parameter(self) -> None: """Verify that process() accepts proxy parameter.""" search = thc.SearchThc(self.domain()) - try: - await search.process(proxy=False) - except (httpx.TimeoutException, httpx.RequestError): - pytest.skip('Skipping due to network error') + await search.process(proxy=False) result = await search.get_hostnames() assert isinstance(result, set) @@ -284,16 +277,13 @@ class TestThcResponseFormat: @staticmethod def domain() -> str: - return 'github.com' + return 'example.com' @pytest.mark.asyncio async def test_hostnames_are_strings(self) -> None: """Verify that all hostnames are strings.""" search = thc.SearchThc(self.domain()) - try: - await search.process() - except (httpx.TimeoutException, httpx.RequestError): - pytest.skip('Skipping due to network error') + await search.process() result = await search.get_hostnames() for hostname in result: assert isinstance(hostname, str) @@ -302,10 +292,7 @@ class TestThcResponseFormat: async def test_hostnames_are_valid_format(self) -> None: """Verify that hostnames have valid format.""" search = thc.SearchThc(self.domain()) - try: - await search.process() - except (httpx.TimeoutException, httpx.RequestError): - pytest.skip('Skipping due to network error') + await search.process() result = await search.get_hostnames() for hostname in result: assert ' ' not in hostname @@ -316,10 +303,7 @@ class TestThcResponseFormat: async def test_hostnames_are_lowercase(self) -> None: """Verify that hostnames are lowercase.""" search = thc.SearchThc(self.domain()) - try: - await search.process() - except (httpx.TimeoutException, httpx.RequestError): - pytest.skip('Skipping due to network error') + await search.process() result = await search.get_hostnames() for hostname in result: assert hostname == hostname.lower() @@ -328,7 +312,6 @@ class TestThcResponseFormat: # ============================================================================= # 7. Integration Tests with theHarvester # ============================================================================= -@pytest.mark.skipif(github_ci == 'true', reason='Skip integration tests in CI') class TestThcIntegration: """Integration tests with theHarvester framework.""" diff --git a/tests/test_harness.py b/tests/test_harness.py new file mode 100644 index 00000000..94fadd1f --- /dev/null +++ b/tests/test_harness.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +import socket +from contextlib import suppress +from typing import TYPE_CHECKING + +import httpx +import pytest + +if TYPE_CHECKING: + from pathlib import Path + + +def test_python_socket_network_is_blocked_by_default() -> None: + with pytest.raises(AssertionError, match='External networking through Python socket APIs is disabled'): + socket.getaddrinfo('example.com', 443) + with pytest.raises(AssertionError, match='External networking through Python socket APIs is disabled'): + socket.gethostbyname('example.com') + with ( + socket.socket() as tcp_socket, + pytest.raises(AssertionError, match='External networking through Python socket APIs is disabled'), + ): + tcp_socket.connect_ex(('192.0.2.1', 443)) + with ( + socket.socket(type=socket.SOCK_DGRAM) as udp_socket, + pytest.raises(AssertionError, match='External networking through Python socket APIs is disabled'), + ): + udp_socket.sendto(b'test', ('192.0.2.1', 53)) + + +def test_http_client_cannot_escape_network_guard() -> None: + with pytest.raises(AssertionError, match='External networking through Python socket APIs is disabled'): + httpx.get('https://example.com', trust_env=False) + + +def test_loopback_network_remains_available(tmp_path: Path) -> None: + addresses = socket.getaddrinfo('localhost', 0) + ipv6_addresses = socket.getaddrinfo('::1', 0) + assert addresses + assert ipv6_addresses + + with socket.socket() as client, suppress(OSError): + client.connect(('127.0.0.1', 0)) + + if hasattr(socket, 'AF_UNIX'): + with socket.socket(socket.AF_UNIX) as unix_client, suppress(OSError): + unix_client.connect(str(tmp_path / 'missing.sock')) diff --git a/tests/test_workflow_contract.py b/tests/test_workflow_contract.py new file mode 100644 index 00000000..c1953803 --- /dev/null +++ b/tests/test_workflow_contract.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import yaml + + +WORKFLOW_DIR = Path(__file__).parents[1] / '.github' / 'workflows' +CI_WORKFLOW_PATH = WORKFLOW_DIR / 'theHarvester.yml' +SMOKE_WORKFLOW_PATH = WORKFLOW_DIR / 'provider-smoke.yml' + + +def _workflow(path: Path) -> dict[str, Any]: + return yaml.load(path.read_text(encoding='utf-8'), Loader=yaml.BaseLoader) + + +def test_routine_ci_is_read_only_and_offline() -> None: + workflow = _workflow(CI_WORKFLOW_PATH) + assert workflow['permissions'] == {'contents': 'read'} + assert set(workflow['on']) == {'push', 'pull_request', 'workflow_dispatch'} + + routine_job = workflow['jobs']['Python'] + commands = '\n'.join(step.get('run', '') for step in routine_job['steps']) + assert 'git push' not in commands + assert 'theHarvester -d' not in commands + assert '\npytest\n' in f'\n{commands.strip()}\n' + + +def test_live_provider_smoke_requires_manual_dispatch() -> None: + workflow = _workflow(SMOKE_WORKFLOW_PATH) + smoke_job = workflow['jobs']['Passive-provider-smoke'] + commands = '\n'.join(step.get('run', '') for step in smoke_job['steps']) + + assert set(workflow['on']) == {'workflow_dispatch'} + assert workflow['permissions'] == {'contents': 'read'} + assert smoke_job['env']['SMOKE_TEST_DOMAIN'] == 'example.com' + assert 'pytest --run-live-network -m live_network' in commands