diff --git a/tests/discovery/test_githubcode.py b/tests/discovery/test_githubcode.py index 853771a3..4f6c02cd 100644 --- a/tests/discovery/test_githubcode.py +++ b/tests/discovery/test_githubcode.py @@ -93,7 +93,7 @@ class TestSearchGithubCode: @pytest.mark.asyncio async def test_infinite_loop_fix_page_zero(self): - """Test that the loop condition properly exits when page becomes 0""" + """Stop pagination when the next page is zero.""" Core.github_key = MagicMock(return_value="test_key") # type: ignore[method-assign] test_class_instance = githubcode.SearchGithubCode(word="test", limit=500) @@ -108,7 +108,7 @@ class TestSearchGithubCode: @pytest.mark.asyncio async def test_infinite_loop_fix_page_nonzero(self): - """Test that the loop condition continues when page is non-zero""" + """Continue pagination while the next page is nonzero.""" Core.github_key = MagicMock(return_value="test_key") # type: ignore[method-assign] test_class_instance = githubcode.SearchGithubCode(word="test", limit=500) @@ -123,7 +123,7 @@ class TestSearchGithubCode: @pytest.mark.asyncio async def test_infinite_loop_fix_old_vs_new_condition(self): - """Test that demonstrates the difference between old and new conditions""" + """Treat zero as the end of pagination, not as another page.""" Core.github_key = MagicMock(return_value="test_key") # type: ignore[method-assign] test_class_instance = githubcode.SearchGithubCode(word="test", limit=500) diff --git a/tests/discovery/test_thc.py b/tests/discovery/test_thc.py index 05999439..0cdb74f7 100644 --- a/tests/discovery/test_thc.py +++ b/tests/discovery/test_thc.py @@ -1,14 +1,13 @@ #!/usr/bin/env python3 # coding=utf-8 -""" -Tests for THC (ip.thc.org) discovery module. +"""Tests for the THC (ip.thc.org) discovery source. THC provides multiple endpoints: - Subdomain enumeration - CNAME lookup - Reverse DNS lookup -API Documentation: https://ip.thc.org/docs/ +API documentation: https://ip.thc.org/docs/ """ from types import TracebackType from typing import Any, Self @@ -96,17 +95,17 @@ def recorded_sleeps(monkeypatch: pytest.MonkeyPatch) -> list[int]: # ============================================================================= @pytest.mark.live_network class TestThcApi: - """Tests to validate that the THC API responds correctly.""" + """Check the THC API contract.""" def test_api_subdomains_download_endpoint_responds(self, live_test_domain: str) -> None: - """Verify that the subdomain download endpoint responds.""" + """The subdomain download endpoint responds.""" url = f'https://ip.thc.org/api/v1/subdomains/download?domain={live_test_domain}&limit=10&hide_header=true' headers = {'User-Agent': Core.get_user_agent()} response = httpx.get(url, headers=headers, timeout=30) assert response.status_code == 200 def test_api_subdomains_returns_text_format(self, live_test_domain: str) -> None: - """Verify that the response is plain text.""" + """The subdomain response is plain text.""" url = f'https://ip.thc.org/api/v1/subdomains/download?domain={live_test_domain}&limit=5&hide_header=true' headers = {'User-Agent': Core.get_user_agent()} response = httpx.get(url, headers=headers, timeout=30) @@ -114,14 +113,14 @@ class TestThcApi: assert 'text' in content_type or 'octet-stream' in content_type def test_api_cli_subdomain_endpoint(self, live_test_domain: str) -> None: - """Verify CLI endpoint /sb/{domain}.""" + """The CLI endpoint accepts ``/sb/{domain}``.""" url = f'https://ip.thc.org/sb/{live_test_domain}?l=5&noheader' headers = {'User-Agent': Core.get_user_agent()} response = httpx.get(url, headers=headers, timeout=30) assert response.status_code == 200 def test_api_returns_rate_limit_headers(self, live_test_domain: str) -> None: - """Verify that the API returns rate limit headers.""" + """The API returns rate-limit headers.""" url = f'https://ip.thc.org/api/v1/subdomains/download?domain={live_test_domain}&limit=1&hide_header=true' headers = {'User-Agent': Core.get_user_agent()} response = httpx.get(url, headers=headers, timeout=30) @@ -133,7 +132,7 @@ class TestThcApi: # 2. Subdomain Search Tests (Main Functionality) # ============================================================================= class TestThcSubdomainSearch: - """Tests for subdomain search functionality.""" + """Check THC subdomain searches.""" @staticmethod def domain() -> str: @@ -145,7 +144,7 @@ class TestThcSubdomainSearch: @pytest.mark.asyncio async def test_search_returns_set(self) -> None: - """Verify that get_hostnames() returns a set.""" + """Return hostnames as a set.""" search = thc.SearchThc(self.domain()) await search.process() result = await search.get_hostnames() @@ -153,7 +152,7 @@ class TestThcSubdomainSearch: @pytest.mark.asyncio async def test_search_finds_subdomains(self) -> None: - """Verify that it finds subdomains for a known domain.""" + """Find subdomains for a known domain.""" search = thc.SearchThc(self.domain()) await search.process() result = await search.get_hostnames() @@ -161,7 +160,7 @@ class TestThcSubdomainSearch: @pytest.mark.asyncio async def test_search_results_contain_target_domain(self) -> None: - """Verify that all results contain the target domain.""" + """Keep every result within the target domain.""" search = thc.SearchThc(self.small_domain()) await search.process() result = await search.get_hostnames() @@ -170,7 +169,7 @@ class TestThcSubdomainSearch: @pytest.mark.asyncio async def test_search_no_duplicates(self) -> None: - """Verify that there are no duplicates in the results.""" + """Deduplicate the results.""" search = thc.SearchThc(self.domain()) await search.process() result = await search.get_hostnames() @@ -266,11 +265,11 @@ class TestThcSubdomainSearch: # 3. Edge Case Tests # ============================================================================= class TestThcEdgeCases: - """Tests for edge cases and error handling.""" + """Check unusual and invalid targets.""" @pytest.mark.asyncio async def test_search_nonexistent_domain(self) -> None: - """Verify behavior with non-existent domain.""" + """Handle a nonexistent domain.""" search = thc.SearchThc('this-domain-definitely-does-not-exist-12345.com') await search.process() result = await search.get_hostnames() @@ -278,7 +277,7 @@ class TestThcEdgeCases: @pytest.mark.asyncio async def test_search_empty_domain(self) -> None: - """Verify behavior with empty domain.""" + """Handle an empty domain.""" search = thc.SearchThc('') await search.process() result = await search.get_hostnames() @@ -286,7 +285,7 @@ class TestThcEdgeCases: @pytest.mark.asyncio async def test_search_special_characters_domain(self) -> None: - """Verify behavior with special characters.""" + """Handle special characters in a domain.""" search = thc.SearchThc('example.com; DROP TABLE domains;--') await search.process() result = await search.get_hostnames() @@ -294,7 +293,7 @@ class TestThcEdgeCases: @pytest.mark.asyncio async def test_search_unicode_domain(self) -> None: - """Verify behavior with IDN/unicode domain.""" + """Handle an internationalized domain name.""" search = thc.SearchThc('xn--mnchen-3ya.de') await search.process() result = await search.get_hostnames() @@ -302,7 +301,7 @@ class TestThcEdgeCases: @pytest.mark.asyncio async def test_search_subdomain_as_input(self) -> None: - """Verify behavior when a subdomain is passed as input.""" + """Accept a subdomain as the target.""" search = thc.SearchThc('www.example.com') await search.process() result = await search.get_hostnames() @@ -313,7 +312,7 @@ class TestThcEdgeCases: # 4. Proxy Tests # ============================================================================= class TestThcProxy: - """Tests for proxy functionality.""" + """Check proxy configuration.""" @staticmethod def domain() -> str: @@ -321,7 +320,7 @@ class TestThcProxy: @pytest.mark.asyncio async def test_process_accepts_proxy_parameter(self) -> None: - """Verify that process() accepts proxy parameter.""" + """Accept the proxy argument in ``process()``.""" search = thc.SearchThc(self.domain()) await search.process(proxy=False) result = await search.get_hostnames() @@ -329,7 +328,7 @@ class TestThcProxy: @pytest.mark.asyncio async def test_proxy_attribute_is_set(self) -> None: - """Verify that the proxy attribute is set correctly.""" + """Store the configured proxy value.""" search = thc.SearchThc(self.domain()) assert search.proxy is False @@ -338,27 +337,27 @@ class TestThcProxy: # 5. Initialization and Attributes Tests # ============================================================================= class TestThcInitialization: - """Tests for class initialization and structure.""" + """Check the initial search state.""" def test_init_sets_word(self) -> None: - """Verify that __init__ sets the domain.""" + """Store the target domain.""" domain = 'test.com' search = thc.SearchThc(domain) assert search.word == domain def test_init_creates_empty_results(self) -> None: - """Verify that results is initialized empty.""" + """Start with no results.""" search = thc.SearchThc('test.com') assert hasattr(search, 'results') assert len(search.results) == 0 def test_init_proxy_default_false(self) -> None: - """Verify that proxy is False by default.""" + """Disable the proxy by default.""" search = thc.SearchThc('test.com') assert search.proxy is False def test_init_has_rate_limit_settings(self) -> None: - """Verify that rate limit settings are initialized.""" + """Initialize the rate-limit settings.""" search = thc.SearchThc('test.com') assert hasattr(search, 'max_retries') assert hasattr(search, 'base_delay') @@ -366,7 +365,7 @@ class TestThcInitialization: assert search.base_delay == 2 def test_class_has_required_methods(self) -> None: - """Verify that the class has the required methods.""" + """Expose the methods required by the source runner.""" search = thc.SearchThc('test.com') assert hasattr(search, 'do_search') assert hasattr(search, 'get_hostnames') @@ -380,7 +379,7 @@ class TestThcInitialization: # 6. Response Format Tests # ============================================================================= class TestThcResponseFormat: - """Tests to verify response format.""" + """Check normalized hostname results.""" @staticmethod def domain() -> str: @@ -388,7 +387,7 @@ class TestThcResponseFormat: @pytest.mark.asyncio async def test_hostnames_are_strings(self) -> None: - """Verify that all hostnames are strings.""" + """Return every hostname as a string.""" search = thc.SearchThc(self.domain()) await search.process() result = await search.get_hostnames() @@ -397,7 +396,7 @@ class TestThcResponseFormat: @pytest.mark.asyncio async def test_hostnames_are_valid_format(self) -> None: - """Verify that hostnames have valid format.""" + """Return valid hostname syntax.""" search = thc.SearchThc(self.domain()) await search.process() result = await search.get_hostnames() @@ -408,7 +407,7 @@ class TestThcResponseFormat: @pytest.mark.asyncio async def test_hostnames_are_lowercase(self) -> None: - """Verify that hostnames are lowercase.""" + """Return lowercase hostnames.""" search = thc.SearchThc(self.domain()) await search.process() result = await search.get_hostnames() @@ -420,23 +419,23 @@ class TestThcResponseFormat: # 7. Integration Tests with theHarvester # ============================================================================= class TestThcIntegration: - """Integration tests with theHarvester framework.""" + """Check the source-runner interface.""" @pytest.mark.asyncio async def test_module_can_be_imported(self) -> None: - """Verify that the module can be imported.""" + """Import the THC discovery module.""" from theHarvester.discovery import thc as thc_module assert thc_module is not None @pytest.mark.asyncio async def test_search_class_exists(self) -> None: - """Verify that SearchThc class exists.""" + """Expose the ``SearchThc`` adapter.""" from theHarvester.discovery import thc as thc_module assert hasattr(thc_module, 'SearchThc') @pytest.mark.asyncio async def test_compatible_with_store_function(self) -> None: - """Verify compatibility with store function from __main__.py.""" + """Return results accepted by the main result store.""" search = thc.SearchThc('example.com') assert hasattr(search, 'process') assert hasattr(search, 'get_hostnames') diff --git a/tests/test_security.py b/tests/test_security.py index 0f214e21..9f778146 100644 --- a/tests/test_security.py +++ b/tests/test_security.py @@ -10,7 +10,7 @@ from theHarvester.__main__ import sanitize_filename, sanitize_for_xml class TestCORSConfiguration: - """Test CORS security configuration.""" + """Check CORS configuration.""" def test_api_does_not_enable_cross_origin_requests(self): from theHarvester.lib.api.api import app @@ -19,14 +19,10 @@ class TestCORSConfiguration: class TestXMLInjectionPrevention: - """Test XML injection prevention.""" + """Check XML escaping.""" def test_sanitize_for_xml_escapes_special_characters(self): - """ - Security Test: Verify XML special characters are properly escaped. - - Prevents XML injection attacks. - """ + """Escape XML special characters.""" # Test all XML special characters test_cases = [ ('&', '&'), @@ -44,9 +40,7 @@ class TestXMLInjectionPrevention: assert result == expected_output, f'Failed to properly escape: {input_text}' def test_sanitize_for_xml_prevents_xml_entity_injection(self): - """ - Security Test: Prevent XML entity injection attempts. - """ + """Escape XML entity declarations and references.""" malicious_inputs = [ ']>', '', @@ -61,12 +55,7 @@ class TestXMLInjectionPrevention: assert '<' not in result or result == malicious_input.replace('<', '<'), f'XML tags not escaped: {malicious_input}' def test_command_line_args_are_sanitized_in_xml_output(self): - """ - Security Test: Command line arguments must be sanitized before XML output. - - This test is a conceptual check - in real usage, ensure the XML writing - code uses sanitize_for_xml() on all user-controlled data. - """ + """Escape command-line arguments before writing them to XML.""" # Simulate dangerous command line arguments dangerous_args = [ '--domain=test.com', @@ -83,21 +72,17 @@ class TestXMLInjectionPrevention: class TestInformationDisclosure: - """Test information disclosure prevention.""" + """Check that API errors do not disclose internal details.""" @pytest.fixture def client(self): - """Create a test client for API testing.""" + """Create an API test client.""" from theHarvester.lib.api.api import app return TestClient(app) def test_api_does_not_expose_traceback_in_error_responses(self, client): - """ - Security Test: API should never expose stack traces to clients. - - Stack traces can reveal sensitive information about the system. - """ + """Keep stack traces out of API error responses.""" response = client.get('/api/v1/sources') # Even if there's an error, traceback should not be in response @@ -108,9 +93,7 @@ class TestInformationDisclosure: assert 'File "' not in str(response_data), 'File paths exposed in response' def test_error_responses_do_not_leak_internal_paths(self, client, tmp_path, monkeypatch): - """ - Security Test: Error messages should not reveal internal file paths. - """ + """Keep internal paths out of API error responses.""" fetch_all = AsyncMock(side_effect=AssertionError('API security test attempted a provider request')) monkeypatch.setenv('THEHARVESTER_API_KEY', 'operator-secret') monkeypatch.setenv('THEHARVESTER_RUN_DB', str(tmp_path / 'runs.sqlite')) @@ -139,9 +122,7 @@ class TestInformationDisclosure: fetch_all.assert_not_awaited() def test_debug_mode_does_not_expose_sensitive_info(self, client, monkeypatch): - """ - Security Test: Even with DEBUG=1, sensitive info should not be exposed to clients. - """ + """Keep sensitive details hidden when ``DEBUG=1``.""" # Set DEBUG environment variable monkeypatch.setenv('DEBUG', '1') @@ -155,11 +136,11 @@ class TestInformationDisclosure: class TestAPIAuthentication: - """Test authentication and error handling for the versioned API.""" + """Check authentication and errors for the versioned API.""" @pytest.fixture def client(self): - """Create a test client for API testing.""" + """Create an API test client.""" from theHarvester.lib.api.api import app return TestClient(app) @@ -211,12 +192,10 @@ class TestAPIAuthentication: class TestPathTraversalPrevention: - """Test path traversal prevention.""" + """Check filename sanitization and path containment.""" def test_sanitize_filename_removes_path_components(self): - """ - Security Test: Filenames should not contain path traversal sequences. - """ + """Remove path components from filenames.""" dangerous_filenames = [ '../../../etc/passwd', '..\\..\\..\\windows\\system32\\config\\sam', @@ -241,9 +220,7 @@ class TestPathTraversalPrevention: assert os.path.dirname(result) == '', f'Path component remains: {result}' def test_sanitize_filename_removes_dangerous_characters(self): - """ - Security Test: Filenames should only contain safe characters. - """ + """Remove shell metacharacters from filenames.""" test_cases = [ 'file; rm -rf /', 'file`whoami`.txt', @@ -268,9 +245,7 @@ class TestPathTraversalPrevention: assert re.match(r'^[a-zA-Z0-9._-]+$', result), f'Invalid characters in sanitized filename: {result}' def test_sanitize_filename_prevents_hidden_files(self): - """ - Security Test: Prevent creation of hidden files. - """ + """Prevent sanitized filenames from naming hidden files.""" hidden_files = ['.bashrc', '.ssh_config', '.env', '..hidden', '.'] for hidden_file in hidden_files: @@ -281,9 +256,7 @@ class TestPathTraversalPrevention: assert not result.startswith('.'), f'Hidden file not prevented: {result}' def test_filename_sanitization_preserves_safe_filenames(self): - """ - Security Test: Safe filenames should remain mostly unchanged. - """ + """Preserve safe filenames and their extensions.""" safe_filenames = [ 'report.json', 'results_2024-01-17.xml', @@ -299,9 +272,7 @@ class TestPathTraversalPrevention: assert '.' in result if '.' in safe_filename else True, 'File extension removed incorrectly' def test_path_traversal_in_file_operations(self): - """ - Integration Test: Verify file operations don't allow path traversal. - """ + """Keep a sanitized output path inside its destination directory.""" # This tests the actual usage in the code from theHarvester.__main__ import sanitize_filename @@ -321,12 +292,10 @@ class TestPathTraversalPrevention: class TestSecurityBestPractices: - """Additional security best practices tests.""" + """Check repository and API security invariants.""" def test_no_hardcoded_secrets_in_code(self): - """ - Security Test: Ensure no hardcoded secrets in main code files. - """ + """Reject common hard-coded secret patterns in application files.""" # Check main application files for common secret patterns files_to_check = [ 'theHarvester/__main__.py', @@ -358,9 +327,7 @@ class TestSecurityBestPractices: assert not real_matches, f'Potential hardcoded secret in {file_path}: {real_matches}' def test_sensitive_endpoints_require_validation(self, monkeypatch): - """ - Security Test: Ensure sensitive endpoints validate input. - """ + """Reject invalid requests to authenticated endpoints.""" from fastapi.testclient import TestClient from theHarvester.lib.api.api import app diff --git a/theHarvester/discovery/api_endpoints.py b/theHarvester/discovery/api_endpoints.py index 2eac0605..71ec047d 100644 --- a/theHarvester/discovery/api_endpoints.py +++ b/theHarvester/discovery/api_endpoints.py @@ -32,7 +32,7 @@ _DIAGNOSTIC_RESPONSE_HEADERS = { @dataclass class EndpointResult: - """Data class for storing endpoint scan results.""" + """One endpoint scan result.""" url: str status_code: int = 0 @@ -53,7 +53,7 @@ class EndpointResult: parameters: list[str] = field(default_factory=list) def to_dict(self) -> dict[str, Any]: - """Convert to dictionary.""" + """Return the result as a dictionary.""" return asdict(self) @@ -590,7 +590,7 @@ class SearchApiEndpoints: return min(delay + jitter, cls.MAX_RETRY_DELAY_SECONDS) async def _detect_schema(self, path: str = '') -> str: - """Detect if the domain supports HTTPS or fall back to HTTP.""" + """Use HTTPS when available, otherwise fall back to HTTP.""" https_url = f'https://{self.word}{path}' if self._session is None: raise RuntimeError('API endpoint session is not initialized') @@ -607,7 +607,7 @@ class SearchApiEndpoints: return 'http' def _load_wordlist(self) -> list[str]: - """Load endpoints from wordlist file with advanced filtering.""" + """Load and filter endpoint paths from the wordlist.""" try: with open(self.wordlist) as f: lines = [line.strip() for line in f if line.strip() and not line.strip().startswith('#')] @@ -634,13 +634,13 @@ class SearchApiEndpoints: return [] async def _check_endpoint(self, url: str) -> EndpointResult | None: - """Check if an endpoint exists and analyze its properties. + """Request one endpoint and describe it when it exists. Args: url: The URL to check. Returns: - Optional[EndpointResult]: Result object or None if not found + The endpoint result, or ``None`` if the endpoint was not found. """ # Other standard HTTP methods can change or delete data on the target. @@ -782,7 +782,7 @@ class SearchApiEndpoints: return False def _get_headers(self) -> dict[str, str]: - """Get request headers with optional custom additions.""" + """Build request headers, including operator-supplied additions.""" headers = { 'User-Agent': self.user_agent, 'Accept': 'application/json, text/plain, */*', @@ -804,10 +804,10 @@ class SearchApiEndpoints: *, body_truncated: bool = False, ) -> EndpointResult | None: - """Process and categorize API endpoint response with detailed analysis. + """Classify one API endpoint response. Returns: - Optional[EndpointResult]: Result object or None if not relevant + The endpoint result, or ``None`` if the response is not relevant. """ status = getattr(response, 'status', 0) @@ -979,7 +979,7 @@ class SearchApiEndpoints: return result async def _post_scan_analysis(self) -> None: - """Perform additional analysis after completing the initial scan.""" + """Log path patterns found among interesting endpoints.""" # Analyze patterns in successful endpoints if self.interesting_endpoints: self.logger.info(f'Performing post-scan analysis on {len(self.interesting_endpoints)} interesting endpoints') @@ -997,10 +997,10 @@ class SearchApiEndpoints: self.logger.info(f'Identified {len(path_patterns)} API path patterns for potential further scanning') def get_results_summary(self) -> dict[str, Any]: - """Get a comprehensive summary of scan results. + """Summarize the scan results. Returns: - Dict[str, Any]: Summary of scan results + Counts and selected scan metadata. """ return { @@ -1026,10 +1026,10 @@ class SearchApiEndpoints: return summary def get_detailed_results(self) -> list[dict[str, Any]]: - """Get detailed results for all endpoints. + """Return the result record for every endpoint. Returns: - List[Dict[str, Any]]: List of endpoint result dictionaries + Endpoint result dictionaries. """ return [result.to_dict() for result in self.found_endpoints.values()] @@ -1043,15 +1043,15 @@ class SearchApiEndpoints: return self.endpoints def get_found_endpoints(self) -> dict[str, EndpointResult]: - """Get dictionary of found and accessible endpoints with detailed results.""" + """Return endpoints that were found and accessible.""" return self.found_endpoints def get_interesting_endpoints(self) -> dict[str, EndpointResult]: - """Get dictionary of interesting endpoints with detailed results.""" + """Return endpoints classified as interesting.""" return self.interesting_endpoints def get_auth_required(self) -> dict[str, EndpointResult]: - """Get dictionary of endpoints requiring authentication with detailed results.""" + """Return endpoints that require authentication.""" return self.auth_required def get_api_versions(self) -> set[str]: @@ -1083,14 +1083,14 @@ class SearchApiEndpoints: return self.schema_detected def export_results(self, output_file: str | None = None, format: str = 'json') -> str | dict | None: - """Export scan results to a file or return as string/dict. + """Write scan results to a file or return them to the caller. Args: - output_file: Optional file path to save results - format: Export format ('json', 'dict') + output_file: Optional destination path. + format: Either ``json`` or ``dict``. Returns: - Union[str, Dict, None]: Results in requested format or None if saved to file + The requested representation, or ``None`` when saved to a file. """ results = {'summary': self.get_results_summary(), 'endpoints': self.get_detailed_results()} diff --git a/theHarvester/discovery/commoncrawl.py b/theHarvester/discovery/commoncrawl.py index 14c7009b..2813e456 100644 --- a/theHarvester/discovery/commoncrawl.py +++ b/theHarvester/discovery/commoncrawl.py @@ -34,7 +34,7 @@ class SearchCommoncrawl: @staticmethod def _safe_parse_json_lines(payload: str) -> list: - """Parse JSON lines format""" + """Parse JSON Lines records, skipping malformed lines.""" results: list = [] malformed = False if not payload: @@ -55,7 +55,7 @@ class SearchCommoncrawl: return results def _extract_domain_from_url(self, url: object) -> str: - """Extract domain from URL""" + """Return the hostname from a URL.""" if not isinstance(url, str) or not url: return '' diff --git a/theHarvester/discovery/constants.py b/theHarvester/discovery/constants.py index d4e04a7b..670a3787 100644 --- a/theHarvester/discovery/constants.py +++ b/theHarvester/discovery/constants.py @@ -4,11 +4,10 @@ from theHarvester.lib.core import AsyncFetcher, Core async def splitter(links): - """Method that tries to remove duplicates - LinkedinLists pulls a lot of profiles with the same name. - This method tries to remove duplicates from the list. - :param links: list of links to remove duplicates from - :return: a unique-ish list + """Deduplicate profile URLs using name-like path segments. + + :param links: Profile URLs to deduplicate. + :return: URLs with repeated name segments removed. """ unique_list = [] name_check = [] @@ -28,9 +27,10 @@ async def splitter(links): def filter(lst): - """Method that filters list - :param lst: list to be filtered - :return: new filtered list + """Normalize a collection into unique, filtered lowercase strings. + + :param lst: Values to filter. + :return: The filtered values. """ if lst is None: return [] @@ -46,14 +46,14 @@ def filter(lst): def get_delay() -> float: - """Method that is used to generate a random delay""" + """Return a random delay between 0.5 and 2.5 seconds.""" return random.randint(1, 3) - 0.5 async def search(text: str) -> bool: - """Helper function to check if Google has blocked traffic. - :param text: See if specific text is returned, which means Google is blocking us - :return bool: + """Return whether text contains Google's automated-traffic block page. + + :param text: Response text to inspect. """ for line in text.strip().splitlines(): if ( @@ -66,9 +66,10 @@ async def search(text: str) -> bool: async def google_workaround(visit_url: str) -> bool | str: - """Function that makes a request on our behalf if Google starts to block us - :param visit_url: Url to scrape - :return: Correct html that can be parsed by BS4 + """Fetch a Google result page through the websniffer fallback. + + :param visit_url: Google URL to fetch. + :return: Decoded HTML, or ``True`` when no usable page is returned. """ url = 'https://websniffer.cc/' data = { @@ -105,7 +106,7 @@ async def google_workaround(visit_url: str) -> bool | str: class MissingKeyError(Exception): - """:raise: When there is a module that has not been provided its API key""" + """Raised when a discovery source is missing required credentials.""" def __init__(self, source: str | None) -> None: if source: diff --git a/theHarvester/discovery/dnssearch.py b/theHarvester/discovery/dnssearch.py index dde82af9..4643bb4f 100644 --- a/theHarvester/discovery/dnssearch.py +++ b/theHarvester/discovery/dnssearch.py @@ -1,9 +1,4 @@ -"""============ -DNS Browsing -============ - -Explore the space around known hosts & ips for extra catches. -""" +"""DNS brute-force and reverse-lookup helpers.""" import asyncio import logging @@ -80,21 +75,19 @@ NETWORK_REGEX: str = rf'\b({IP_REGEX})(?:\:({PORT_REGEX}))?(?:\/({NETMASK_REGEX} def serialize_ip_range(ip: str, netmask: str = '24') -> str: - """Serialize a network range in a constant format, 'x.x.x.x/y'. + """Normalize the first IPv4 address in a string as a CIDR range. Parameters ---------- - ip: str. - A serialized ip in the format 'x.x.x.x'. - Extra information like port (':z') or subnet ('/n') - will be ignored. - netmask: str. - The subnet subdivision, represented by a 2 digit netmask. + ip: str + Text containing an IPv4 address. Ports and embedded netmasks are ignored. + netmask: str + Netmask to apply. The default is ``24``. Returns ------- - out: str. - The network OSI address, like '192.168.0.0/24'. + str + A range such as ``192.168.0.0/24``, or an empty string for invalid input. """ __ip_matches = re.search(NETWORK_REGEX, ip, re.IGNORECASE) @@ -121,34 +114,36 @@ def iter_ips_in_network_range(iprange: str) -> Iterator[str]: def list_ips_in_network_range(iprange: str) -> list[str]: - """List all the IPs in the range. + """Return every usable address in an IPv4 range. Parameters ---------- - iprange: str. - A serialized ip range, like '1.2.3.0/24'. - The last digit can be set to anything, it will be ignored. + iprange: str + A range such as ``1.2.3.0/24``. Host bits are ignored. Returns ------- - out: list. - The list of IPs in the range. + list[str] + Usable addresses in the range. """ return list(iter_ips_in_network_range(iprange)) async def reverse_single_ip(ip: str, resolver: DNSResolver, error_types: set[str] | None = None) -> str: - """Reverse a single IP and output the linked CNAME, if it exists. + """Return the PTR hostname for an IP address, or an empty string. Parameters ---------- - :param ip: IP address to reverse - :param resolver: DNS server to use + ip: str + IP address to resolve. + resolver: DNSResolver + Resolver to query. Returns ------- - :return str: with the corresponding CNAME or None + str + The resolved hostname, or an empty string when resolution fails. """ try: @@ -299,23 +294,18 @@ async def reverse_all_ips_in_range( nameservers: list[str] | None = None, error_types: set[str] | None = None, ) -> None: - """Reverse one range through the bounded global reverse-DNS implementation. + """Resolve usable addresses from one range with the shared bounds. Parameters ---------- - iprange: str. - An IPv4 range formatted as 'x.x.x.x/y'. - The last 2 digits of the ip can be set to anything, - they will be ignored. - callback: Callable. - Arbitrary postprocessing function. - nameservers: List[str]. - Optional list of DNS servers. - error_types: set[str]. - Optional sink for unexpected resolver or transport error names. - Returns - ------- - out: None. + iprange: str + An IPv4 range formatted as ``x.x.x.x/y``. Host bits are ignored. + callback: Callable + Function called for each resolved hostname. + nameservers: list[str] | None + DNS servers to query. + error_types: set[str] | None + Sink for unexpected resolver or transport error names. """ await reverse_ip_ranges((iprange,), callback, nameservers, error_types) @@ -327,58 +317,20 @@ async def reverse_all_ips_in_range( def log_query(ip: str) -> None: - """Display the current query in the console. - - Parameters - ---------- - ip: str. - Queried ip. - - Results - ------- - out: None. - - """ + """Display the IP address currently being queried.""" sys.stdout.write(chr(27) + '[2K' + chr(27) + '[G') sys.stdout.write('\r' + ip + ' - ') sys.stdout.flush() def log_result(host: str) -> None: - """Display the query result in the console. - - Parameters - ---------- - host: str. - Host name returned by the DNS query. - - Results - ------- - out: None. - - """ + """Log a hostname returned by reverse DNS.""" if host: logger.info(host) def generate_postprocessing_callback(target: str, **allhosts: list[str]) -> Callable: - """Postprocess the query results asynchronously too, instead of waiting for - the querying stage to be completely finished. - - Parameters - ---------- - target: str. - The domain wanted as TLD. - allhosts: List. - A collection of all the subdomains -of target- found so far. - - Returns - ------- - out: Callable. - A function that will update the collection of target subdomains - when the query result is satisfying. - - """ + """Return a callback that appends matching PTR hostnames to each collection.""" def append_matching_hosts(host: str) -> None: if host and target in host: diff --git a/theHarvester/discovery/fullhuntsearch.py b/theHarvester/discovery/fullhuntsearch.py index 2f9f5676..7f7dbc96 100644 --- a/theHarvester/discovery/fullhuntsearch.py +++ b/theHarvester/discovery/fullhuntsearch.py @@ -13,54 +13,51 @@ logger = logging.getLogger(__name__) class SearchFullHunt: - """Class to search FullHunt API for domain information + """Search the FullHunt API for domain and host data. - FullHunt provides various endpoints for attack surface discovery: - - Domain Details: Full domain information including hosts, DNS records, ports, etc. - - Subdomains: Just the list of subdomains for a domain - - Host Details: Detailed information about specific hosts - - Data Intelligence: Access to FullHunt's attack surface database + FullHunt provides endpoints for domain details, subdomains, host details, + and its data-intelligence search. Supported search filters (with examples): - General Filters: - - domain: Domain (required for all filters) - domain:kaspersky.com - - ip: IP address associated with the asset - ip:8.8.8.8 - - tech: Identified Technologies - tech:drupal - - host: Specific host - host:ecommerce.kaspersky.com - - subdomain: Subdomain - subdomain:video.kaspersky.com - - tld: Top-Level Domain - tld:com - - tag: Tags - tag:cdn - - is_dos_defense: DDoS prevention solution - is_dos_defense:true + General filters: + - domain: required domain, for example ``domain:kaspersky.com`` + - ip: asset IP address, for example ``ip:8.8.8.8`` + - tech: detected technology, for example ``tech:drupal`` + - host: specific host, for example ``host:ecommerce.kaspersky.com`` + - subdomain: subdomain, for example ``subdomain:video.kaspersky.com`` + - tld: top-level domain, for example ``tld:com`` + - tag: tag, for example ``tag:cdn`` + - is_dos_defense: DDoS protection flag, for example ``is_dos_defense:true`` - Network Filters: - - port: Network Port - port:80 - - has_private_ip: Has Private IP - has_private_ip:true - - has_ipv6: Has IPv6 - has_ipv6:true - - is_live: Is the asset live - is_live:true - - is_resolvable: Is resolvable - is_resolvable:true + Network filters: + - port: network port, for example ``port:80`` + - has_private_ip: private-IP flag, for example ``has_private_ip:true`` + - has_ipv6: IPv6 flag, for example ``has_ipv6:true`` + - is_live: live-asset flag, for example ``is_live:true`` + - is_resolvable: resolvable-asset flag, for example ``is_resolvable:true`` - dns_a, dns_aaaa, dns_cname, dns_mx, dns_txt, dns_ptr, dns_ns: DNS records - HTTP Filters: - - http_title: HTTP Title - http_title:Nginx - - http_status_code: HTTP Status code - http_status_code:302 - - http_favicon_hash: HTTP Favicon Hash - http_favicon_hash:9888t96t6ctsdgc9gc + HTTP filters: + - http_title: page title, for example ``http_title:Nginx`` + - http_status_code: status code, for example ``http_status_code:302`` + - http_favicon_hash: favicon hash, for example ``http_favicon_hash:9888t96t6ctsdgc9gc`` - Geographic Filters: - - country_code/country: Country Code - country_code:us - - city: City - city:ashburn - - asn: Autonomous System Number - asn:123124 + Geographic filters: + - country_code/country: country code, for example ``country_code:us`` + - city: city, for example ``city:ashburn`` + - asn: autonomous system number, for example ``asn:123124`` - Cloud Filters: - - is_cloud: Is on cloud - is_cloud:true - - cloud_provider: Cloud Provider - cloud_provider:Linode - - cloud_region: Cloud Region - cloud_region:us-ca + Cloud filters: + - is_cloud: cloud-hosted flag, for example ``is_cloud:true`` + - cloud_provider: provider, for example ``cloud_provider:Linode`` + - cloud_region: region, for example ``cloud_region:us-ca`` - Technology Filters: - - product: Identified Product - product:wordpress - - service: Identified Service - service:http + Technology filters: + - product: detected product, for example ``product:wordpress`` + - service: detected service, for example ``service:http`` - Certificate Filters: + Certificate filters: - cert_issuer_common_name, cert_issuer_organization, cert_issuer_country - cert_issuer_serial_number, cert_signature_algorithm - cert_subject_common_name, cert_subject_country, cert_subject_province @@ -148,11 +145,11 @@ class SearchFullHunt: self._report = SourceExecutionReport(status, reason) def _get_headers(self) -> dict[str, str]: - """Returns the headers needed for API requests""" + """Return headers for FullHunt API requests.""" return {'User-Agent': Core.get_user_agent(), 'X-API-KEY': self.key} async def _fetch_data(self, endpoint: str, session: Any | None = None) -> dict[str, Any]: - """Generic method to fetch data from a specific endpoint""" + """Fetch JSON data from one FullHunt endpoint.""" url = f'{self.BASE_URL}/{endpoint}' response = await AsyncFetcher.fetch_all( [url], @@ -173,14 +170,14 @@ class SearchFullHunt: return metadata.body def add_filter(self, filter_name: str, filter_value: str) -> None: - """Add a search filter to be used in advanced searches + """Add a filter for data-intelligence searches. Args: - filter_name: Name of the filter from the supported filter list - filter_value: Value for the filter + filter_name: Name from the supported filter list. + filter_value: Filter value. Raises: - ValueError: If the filter name is not supported + ValueError: If the filter name is unsupported. """ if filter_name not in self.ALL_FILTERS: @@ -190,24 +187,24 @@ class SearchFullHunt: self.filters[filter_name] = filter_value def add_filters(self, filters: dict[str, str]) -> None: - """Add multiple search filters at once + """Add several data-intelligence search filters. Args: - filters: Dictionary of filter name to filter value + filters: Mapping of filter names to values. Raises: - ValueError: If any filter name is not supported + ValueError: If any filter name is unsupported. """ for name, value in filters.items(): self.add_filter(name, value) def clear_filters(self) -> None: - """Clear all filters""" + """Clear all search filters.""" self.filters = {} def _build_query_string(self) -> str: - """Build a query string from the current filters""" + """Build a query string from the current filters.""" # Start with the domain filter which is required query_parts = [f'domain:{self.word}'] @@ -222,13 +219,12 @@ class SearchFullHunt: return ' '.join(query_parts) async def advanced_search(self, session: Any | None = None) -> dict[str, Any]: - """Perform an advanced search using the configured filters + """Search the data-intelligence endpoint with the configured filters. - This method uses the search endpoint with the filters configured via add_filter - or add_filters methods. + Configure filters with ``add_filter`` or ``add_filters`` first. Returns: - Dict containing the search results + The search response. """ query = self._build_query_string() @@ -237,52 +233,52 @@ class SearchFullHunt: return await self._fetch_data(endpoint, session) async def get_domain_details(self, session: Any | None = None) -> dict[str, Any]: - """Get comprehensive details about a domain""" + """Return FullHunt details for the target domain.""" endpoint = f'domain/{self.word}/details' return await self._fetch_data(endpoint, session) async def get_subdomains(self, session: Any | None = None) -> dict[str, Any]: - """Get subdomains for a domain""" + """Return subdomains for the target domain.""" endpoint = f'domain/{self.word}/subdomains' return await self._fetch_data(endpoint, session) async def get_host_details(self, host: str) -> dict[str, Any]: - """Get detailed information about a specific host""" + """Return FullHunt details for one host.""" endpoint = f'host?host={host}' return await self._fetch_data(endpoint) async def search_tech(self, tech_name: str) -> dict[str, Any]: - """Search for hosts using a specific technology""" + """Search for hosts using a technology.""" self.add_filter('tech', tech_name) return await self.advanced_search() async def search_service(self, service_name: str) -> dict[str, Any]: - """Search for hosts running a specific service""" + """Search for hosts running a service.""" self.add_filter('service', service_name) return await self.advanced_search() async def search_port(self, port: int) -> dict[str, Any]: - """Search for hosts with a specific open port""" + """Search for hosts with an open port.""" self.add_filter('port', str(port)) return await self.advanced_search() async def search_country(self, country_code: str) -> dict[str, Any]: - """Search for hosts in a specific country""" + """Search for hosts in a country.""" self.add_filter('country_code', country_code) return await self.advanced_search() async def search_cloud_provider(self, provider: str) -> dict[str, Any]: - """Search for hosts on a specific cloud provider""" + """Search for hosts on a cloud provider.""" self.add_filter('cloud_provider', provider) return await self.advanced_search() async def search_http_status(self, status_code: int) -> dict[str, Any]: - """Search for hosts with a specific HTTP status code""" + """Search for hosts with an HTTP status code.""" self.add_filter('http_status_code', str(status_code)) return await self.advanced_search() async def search_certificate(self, filter_name: str, value: str) -> dict[str, Any]: - """Search for hosts with specific certificate properties""" + """Search for hosts with matching certificate properties.""" if filter_name not in self.CERT_FILTERS: valid_filters = ', '.join(self.CERT_FILTERS) raise ValueError(f'Invalid certificate filter: {filter_name}. Valid filters: {valid_filters}') @@ -291,7 +287,7 @@ class SearchFullHunt: return await self.advanced_search() async def search_with_dns(self, dns_type: str, value: str) -> dict[str, Any]: - """Search for hosts with specific DNS records""" + """Search for hosts with a matching DNS record.""" dns_filter = f'dns_{dns_type.lower()}' if dns_filter not in self.NETWORK_FILTERS: valid_filters = [f for f in self.NETWORK_FILTERS if f.startswith('dns_')] @@ -302,7 +298,7 @@ class SearchFullHunt: return await self.advanced_search() async def extract_data_from_domain_details(self, details: dict[str, Any]) -> None: - """Extract useful information from domain details response""" + """Collect normalized results from a domain-details response.""" if 'hosts' not in details: return @@ -415,7 +411,7 @@ class SearchFullHunt: self.total_results['tags'] = list(set(self.total_results['tags'])) async def extract_data_from_search_results(self, results: dict[str, Any]) -> None: - """Extract useful information from search results""" + """Collect normalized results from a search response.""" if 'hosts' not in results: return @@ -426,7 +422,7 @@ class SearchFullHunt: await self.extract_data_from_domain_details(results) async def do_search(self) -> None: - """Main search method that calls the various endpoints""" + """Query the FullHunt endpoints used by this source.""" try: async with AsyncFetcher.open_session( headers=self._get_headers(), @@ -466,47 +462,47 @@ class SearchFullHunt: return async def get_hostnames(self) -> list[str]: - """Return list of discovered subdomains""" + """Return discovered subdomains.""" return self.total_results['hosts'] async def get_ips(self) -> list[str]: - """Return list of discovered IP addresses""" + """Return discovered IP addresses.""" return self.total_results['ips'] async def get_ports(self) -> list[int]: - """Return list of open ports""" + """Return open ports.""" return list(self.total_results['ports']) async def get_technologies(self) -> list[str]: - """Return list of technologies found""" + """Return detected technologies.""" return self.total_results['technologies'] async def get_tags(self) -> list[str]: - """Return list of tags""" + """Return FullHunt tags.""" return self.total_results['tags'] async def get_dns_records(self) -> dict[str, dict[str, list[str]]]: - """Return DNS records for hosts""" + """Return DNS records for hosts.""" return self.total_results['dns_records'] async def get_http_info(self) -> dict[str, dict[str, Any]]: - """Return HTTP information for hosts""" + """Return HTTP information for hosts.""" return self.total_results['http_info'] async def get_geo_info(self) -> dict[str, dict[str, Any]]: - """Return geographic information for hosts""" + """Return geographic information for hosts.""" return self.total_results['geo_info'] async def get_cloud_info(self) -> dict[str, dict[str, Any]]: - """Return cloud provider information for hosts""" + """Return cloud-provider information for hosts.""" return self.total_results['cloud_info'] async def get_certificate_info(self) -> list[dict[str, Any]]: - """Return certificate information for hosts""" + """Return certificate information for hosts.""" return self.total_results['cert_info'] async def get_all_results(self) -> dict[str, Any]: - """Return all collected results""" + """Return all collected results.""" return self.total_results async def process( @@ -514,11 +510,11 @@ class SearchFullHunt: proxy: bool = False, filters: dict[str, str] | None = None, ) -> SourceExecutionReport | None: - """Main processing method + """Run the FullHunt search. Args: - proxy: Whether to use a proxy for requests - filters: Optional dictionary of filters to apply to the search + proxy: Whether to use a proxy for requests. + filters: Optional search filters. """ self.proxy = proxy diff --git a/theHarvester/discovery/gitlabsearch.py b/theHarvester/discovery/gitlabsearch.py index 8c606486..0d63ef2f 100644 --- a/theHarvester/discovery/gitlabsearch.py +++ b/theHarvester/discovery/gitlabsearch.py @@ -33,7 +33,7 @@ class SearchGitlab: return {} def _extract_domains_from_text(self, text: str) -> set: - """Extract domain names from text that match our target domain""" + """Extract domain names that match the target domain.""" domains: set[str] = set() if not text: return domains @@ -45,7 +45,7 @@ class SearchGitlab: return domains def _extract_emails_from_text(self, text: str) -> set: - """Extract email addresses that match our target domain""" + """Extract email addresses that match the target domain.""" emails: set[str] = set() if not text: return emails @@ -65,7 +65,7 @@ class SearchGitlab: return bool(hosts or emails) async def search_projects(self) -> None: - """Search GitLab projects for domain references""" + """Search GitLab projects for references to the target domain.""" try: headers = {'User-agent': Core.get_user_agent()} @@ -127,7 +127,7 @@ class SearchGitlab: logger.info(f'GitLab API projects search error: {e}') async def search_users(self) -> None: - """Search GitLab users for domain references""" + """Search GitLab users for references to the target domain.""" try: headers = {'User-agent': Core.get_user_agent()} diff --git a/theHarvester/discovery/hackertarget.py b/theHarvester/discovery/hackertarget.py index d6730c0a..b4b0a70b 100644 --- a/theHarvester/discovery/hackertarget.py +++ b/theHarvester/discovery/hackertarget.py @@ -11,10 +11,10 @@ if TYPE_CHECKING: class SearchHackerTarget: - """Class uses the HackerTarget API to gather subdomains and IPs. + """Use the HackerTarget API to gather subdomains and IP addresses. - This version supports reading a Hackertarget API key (if present) and - appending it to the hackertarget request URLs as `apikey=`. + When configured, the adapter appends the HackerTarget API key to request + URLs as ``apikey=``. """ def __init__(self, word) -> None: diff --git a/theHarvester/discovery/hudsonrocksearch.py b/theHarvester/discovery/hudsonrocksearch.py index 29d293e9..84e9a5b7 100644 --- a/theHarvester/discovery/hudsonrocksearch.py +++ b/theHarvester/discovery/hudsonrocksearch.py @@ -6,17 +6,17 @@ from theHarvester.lib.core import AsyncFetcher, FetcherResponse class SearchHudsonRock: - """Hudson Rock API integration for discovering compromised credentials and stealer logs. + """Search Hudson Rock for compromised credentials and infostealer data. - This class provides comprehensive search capabilities using Hudson Rock's Cavalier API - to discover leaked credentials, compromised hosts, and infostealer intelligence. + The adapter queries the Cavalier API for leaked credentials, compromised + hosts, and infostealer records. """ def __init__(self, word: str) -> None: - """Initialize Hudson Rock search. + """Configure a Hudson Rock search. Args: - word: Domain or email to search for + word: Domain or email address to search. """ self.word = word.strip().lower() @@ -34,10 +34,9 @@ class SearchHudsonRock: self.max_retries = 3 async def do_search(self) -> None: - """Search Hudson Rock for infostealer intelligence data. + """Query by domain and, for email targets, by email address. - Performs comprehensive search using both domain and email endpoints - with proper error handling and rate limiting. + Requests are retried when the provider rate limits them. """ self.logger.info(f'Starting Hudson Rock search for: {self.word}') @@ -70,10 +69,10 @@ class SearchHudsonRock: """Validate email format. Args: - email: Email address to validate + email: Email address to validate. Returns: - True if email format is valid + Whether the email address matches the supported format. """ import re @@ -85,7 +84,7 @@ class SearchHudsonRock: """Search Hudson Rock by domain with retry logic. Args: - domain: Domain to search for + domain: Domain to search. """ url = f'{self.base_url}/search-by-domain?domain={domain}' @@ -97,7 +96,7 @@ class SearchHudsonRock: """Search Hudson Rock by email with retry logic. Args: - email: Email address to search for + email: Email address to search. """ url = f'{self.base_url}/search-by-email?email={email}' @@ -145,7 +144,7 @@ class SearchHudsonRock: """Process domain search response from Hudson Rock API. Args: - response: JSON response from Hudson Rock domain search API + response: Hudson Rock domain-search response. """ try: @@ -189,8 +188,8 @@ class SearchHudsonRock: """Extract hostnames from URL data. Args: - urls_data: List of URL data dictionaries - source_type: Type of source (employee, user, third_party) + urls_data: URL records. + source_type: Source category: employee, user, or third party. """ extracted_count = 0 @@ -219,7 +218,7 @@ class SearchHudsonRock: """Extract email addresses from response data. Args: - data: Response data dictionary + data: Hudson Rock response data. """ # Look for emails in various data fields @@ -243,7 +242,7 @@ class SearchHudsonRock: """Process email search response from Hudson Rock API. Args: - response: JSON response from Hudson Rock email search API + response: Hudson Rock email-search response. """ try: @@ -294,10 +293,10 @@ class SearchHudsonRock: """Validate IP address format. Args: - ip: IP address to validate + ip: IP address to validate. Returns: - True if IP format is valid + Whether the value is a supported IPv4 address. """ if not ip or '*' in ip or '•' in ip: @@ -315,7 +314,7 @@ class SearchHudsonRock: """Extract hostnames from service data. Args: - services: List of service dictionaries + services: Service records. """ for service in services: @@ -342,7 +341,7 @@ class SearchHudsonRock: """Return discovered hostnames. Returns: - Set of unique hostnames discovered from Hudson Rock data + Unique hostnames found in Hudson Rock data. """ return self.totalhosts @@ -351,7 +350,7 @@ class SearchHudsonRock: """Return discovered IP addresses. Returns: - Set of unique IP addresses discovered from Hudson Rock data + Unique IP addresses found in Hudson Rock data. """ return self.totalips @@ -360,7 +359,7 @@ class SearchHudsonRock: """Return discovered email addresses. Returns: - Set of unique email addresses discovered from Hudson Rock data + Unique email addresses found in Hudson Rock data. """ return self.emails @@ -369,7 +368,7 @@ class SearchHudsonRock: """Return infostealer intelligence data. Returns: - List of dictionaries containing detailed stealer information + Infostealer records. """ return self.infostealers @@ -378,7 +377,7 @@ class SearchHudsonRock: """Return compromised data statistics. Returns: - Dictionary containing statistics about compromised data + Compromised-data counts. """ return self.compromised_data @@ -387,7 +386,7 @@ class SearchHudsonRock: """Get a summary of all discovered data. Returns: - Dictionary containing summary statistics + Counts for the collected result types. """ return { @@ -403,10 +402,10 @@ class SearchHudsonRock: } async def process(self, proxy: bool = False) -> None: - """Main processing method. + """Run the Hudson Rock search. Args: - proxy: Whether to use proxy for requests + proxy: Whether to use a proxy for requests. """ self.proxy = proxy diff --git a/theHarvester/discovery/robtex.py b/theHarvester/discovery/robtex.py index 77c315b1..a517e6c2 100644 --- a/theHarvester/discovery/robtex.py +++ b/theHarvester/discovery/robtex.py @@ -21,7 +21,7 @@ class SearchRobtex: @staticmethod def _safe_parse_json_lines(payload: str) -> list: - """Parse JSONL (JSON Lines) format""" + """Parse JSON Lines records, skipping malformed lines.""" results: list = [] if not payload: return results diff --git a/theHarvester/discovery/thc.py b/theHarvester/discovery/thc.py index f061fc02..6ddbb5ef 100644 --- a/theHarvester/discovery/thc.py +++ b/theHarvester/discovery/thc.py @@ -9,7 +9,7 @@ logger = logging.getLogger(__name__) class SearchThc: - """Class to search for subdomains using THC (ip.thc.org).""" + """Search THC (ip.thc.org) for subdomains.""" def __init__(self, word: str) -> None: self.word = word diff --git a/theHarvester/discovery/waybackarchive.py b/theHarvester/discovery/waybackarchive.py index 1f3e43b3..6d77e352 100644 --- a/theHarvester/discovery/waybackarchive.py +++ b/theHarvester/discovery/waybackarchive.py @@ -27,7 +27,7 @@ class SearchWaybackarchive: self.hostname = 'https://web.archive.org' def _extract_domain_from_url(self, url: str) -> str: - """Extract domain from URL""" + """Return the hostname from a URL.""" if not url: return '' try: diff --git a/theHarvester/discovery/windvane.py b/theHarvester/discovery/windvane.py index 107027be..f9e024a6 100644 --- a/theHarvester/discovery/windvane.py +++ b/theHarvester/discovery/windvane.py @@ -8,8 +8,9 @@ logger = logging.getLogger(__name__) class SearchWindvane: - """Class uses the Windvane API to gather subdomains and domain intelligence - API Documentation: https://windvane.lichoin.com + """Use the Windvane API to gather subdomains and domain data. + + API documentation: https://windvane.lichoin.com The API provides several endpoints: - /ListSubDomain - Subdomain enumeration @@ -17,13 +18,10 @@ class SearchWindvane: - /ListDomainWhois - Historical whois lookup - /ListEmail - Domain name email query - Note: This API requires authentication for full access. - - With API key: Full access to all endpoints with pagination - - Without API key: Limited unauthenticated API access + The provider grants full endpoint access and pagination with an API key. + Unauthenticated requests have limited access. - Set API key via: - - Environment variable: export WINDVANE_API_KEY="your-key" - - Or call search.set_api_key("your-key") + Set the key with ``WINDVANE_API_KEY`` or ``search.set_api_key("your-key")``. """ def __init__(self, word) -> None: @@ -68,7 +66,7 @@ class SearchWindvane: return {} async def do_search(self) -> None: - """Main search function that queries multiple Windvane API endpoints""" + """Query the Windvane endpoints used by this source.""" try: headers = {'User-agent': Core.get_user_agent(), 'Content-Type': 'application/json', 'Accept': 'application/json'} @@ -89,7 +87,7 @@ class SearchWindvane: logger.info(f'Windvane API error: {e}') async def _search_subdomains(self, headers: dict) -> None: - """Search for subdomains using /ListSubDomain endpoint""" + """Search for subdomains with ``/ListSubDomain``.""" try: url = f'{self.hostname}/ListSubDomain' @@ -132,7 +130,7 @@ class SearchWindvane: logger.info(f'Windvane subdomain search error: {e}') async def _search_dns_history(self, headers: dict) -> None: - """Search DNS history using /ListDNS endpoint for additional subdomains and IPs""" + """Collect subdomains and IP addresses from ``/ListDNS`` history.""" try: url = f'{self.hostname}/ListDNS' @@ -178,7 +176,7 @@ class SearchWindvane: logger.info(f'Windvane DNS history search error: {e}') async def _search_emails(self, headers: dict) -> None: - """Search for emails using /ListEmail endpoint""" + """Search for email addresses with ``/ListEmail``.""" try: url = f'{self.hostname}/ListEmail' @@ -210,7 +208,7 @@ class SearchWindvane: logger.info(f'Windvane email search error: {e}') async def _search_subdomains_limited(self, headers: dict) -> None: - """Limited subdomain search without API key - tries simpler approaches""" + """Search the unauthenticated subdomain endpoints.""" try: # Try basic subdomain endpoint with minimal parameters url = f'{self.hostname}/ListSubDomain' @@ -253,16 +251,16 @@ class SearchWindvane: logger.info(f'Windvane limited search error: {e}') def set_api_key(self, api_key: str) -> None: - """Set the API key for authenticated requests + """Set the API key for authenticated requests. Args: - api_key: Windvane API key for authenticated access + api_key: Windvane API key. """ self.api_key = api_key def _is_valid_ip(self, ip: str) -> bool: - """Validate if string is a valid IP address""" + """Return whether a string is a valid IP address.""" try: parts = ip.split('.') return len(parts) == 4 and all(0 <= int(part) <= 255 for part in parts) @@ -279,10 +277,10 @@ class SearchWindvane: return self.totalemails async def process(self, proxy: bool = False) -> None: - """Process the search with optional proxy and API key configuration + """Run the Windvane search. Args: - proxy: Whether to use proxy for requests + proxy: Whether to use a proxy for requests. """ self.proxy = proxy diff --git a/theHarvester/lib/core.py b/theHarvester/lib/core.py index ee548569..72d2dbaa 100644 --- a/theHarvester/lib/core.py +++ b/theHarvester/lib/core.py @@ -631,9 +631,7 @@ class AsyncFetcher: @staticmethod def _get_random_proxy(proxy_dict: dict) -> tuple[str | None, str | None]: - """Get a random proxy from the proxy dictionary. - Returns (proxy_url, proxy_type) where proxy_type is 'http' or 'socks5' - """ + """Return a random proxy URL and its ``http`` or ``socks5`` type.""" all_proxies = [] for proxy_type, proxies in proxy_dict.items(): if proxies: @@ -649,9 +647,7 @@ class AsyncFetcher: async def _create_connector( proxy_url: str | None, proxy_type: str | None, ssl_context: ssl.SSLContext | bool | None = None ) -> aiohttp.BaseConnector: - """Create an appropriate connector for the given proxy type. - Returns a connector that can be used with aiohttp.ClientSession. - """ + """Create an aiohttp connector for the selected proxy type.""" if proxy_url and proxy_type == 'socks5': # Create SOCKS5 proxy connector using aiohttp-socks # ProxyConnector.from_url can handle socks5://host:port URLs @@ -723,11 +719,12 @@ class AsyncFetcher: include_metadata: bool = False, response_byte_limit: int | None = None, ) -> Any: - """Generic HTTP request helper. - - If a session is not provided, one will be created and closed automatically. - - Supports optional headers, method selection, proxy, ssl verification, redirects and timeout. - - An explicit response byte limit raises ``ResponseStreamError`` instead of buffering beyond it. - - Returns response text or json depending on `json` flag. + """Send an HTTP request and return its text or JSON body. + + When no session is supplied, this method creates and closes one. It + supports custom headers, methods, proxies, TLS verification, redirects, + and timeouts. A response that exceeds an explicit byte limit raises + ``ResponseStreamError`` instead of buffering the remaining body. """ try: owns_session = session is None diff --git a/theHarvester/parsers/intelxparser.py b/theHarvester/parsers/intelxparser.py index a9141717..a5f2a4b5 100644 --- a/theHarvester/parsers/intelxparser.py +++ b/theHarvester/parsers/intelxparser.py @@ -4,9 +4,10 @@ class Parser: self.selectors: set[str] = set() async def parse_dictionaries(self, results: object) -> tuple[set[str], set[str]]: - """Parse method to parse json results - :param results: Dictionary containing a list of dictionaries known as selectors - :return: tuple of emails and non-email selectors + """Split Intelligence X selectors into emails and other values. + + :param results: Mapping containing selector records. + :return: A tuple of emails and non-email selectors. """ if not isinstance(results, dict): return self.emails, self.selectors diff --git a/theHarvester/parsers/securitytrailsparser.py b/theHarvester/parsers/securitytrailsparser.py index c50b48ad..d55bac73 100644 --- a/theHarvester/parsers/securitytrailsparser.py +++ b/theHarvester/parsers/securitytrailsparser.py @@ -9,10 +9,10 @@ class Parser: self.ips: set = set() async def parse_text(self) -> tuple[set, set]: - """Parse SecurityTrails data and extract IPs and hostnames. - - Supports structured dict with keys {"domain": {...}, "subdomains": {...}} - - Also supports raw dict from either endpoint. - - Falls back to legacy string parsing when input is a string. + """Extract IP addresses and hostnames from SecurityTrails data. + + Accept structured ``domain`` and ``subdomains`` mappings, a raw mapping + from either endpoint, or the older string representation. """ # sanitize base domain base_domain = self.word.replace('www.', '') if 'www' in self.word else self.word diff --git a/theHarvester/screenshot/screenshot.py b/theHarvester/screenshot/screenshot.py index a5ec16dc..fd1743ea 100644 --- a/theHarvester/screenshot/screenshot.py +++ b/theHarvester/screenshot/screenshot.py @@ -1,6 +1,4 @@ -"""Screenshot module that utilizes playwright to asynchronously -take screenshots -""" +"""Take screenshots asynchronously with Playwright.""" import asyncio import logging