Merge pull request #2465 from NotoriousRebel/codex/upstream-dev-github-fragment-bounds-v2

fix: bound GitHub code-search fragments

CHANGELOG.md

@@ -30,6 +30,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Expanded offline regression coverage for discovery providers, configuration contracts, logging, output, documentation, workflow policy, and scope boundaries.
 
 ### Fixed
+- Fixed GitHub code-search fragment limits, boundary separation, and malformed-page termination with offline provider tests.
 - Fixed Baidu, Mojeek, and Yahoo page-response separation and Brave missing-credential reporting, with offline web-search provider contract coverage.
 - Fixed DNS candidate validation to omit names without usable A, AAAA, or CNAME evidence, normalize and deduplicate IPv4, IPv6, and canonical-name records, and preserve the existing `Checker.check()` and `DnsForce.run()` return shape.
 - Fixed REST `/query` requests with a filename so they no longer fail with an unbound local value and HTTP 500 response ([c358df80](https://github.com/laramies/theHarvester/commit/c358df80)).

tests/discovery/test_githubcode_contract.py

@@ -0,0 +1,201 @@
+from typing import Any
+
+import pytest
+
+from theHarvester.discovery import githubcode
+
+
+class FakeResponse:
+    status = 200
+
+    def __init__(self, payload: dict[str, Any], links: dict[str, Any]) -> None:
+        self.payload = payload
+        self.links = links
+
+    async def __aenter__(self) -> 'FakeResponse':
+        return self
+
+    async def __aexit__(self, *_args: Any) -> None:
+        return None
+
+    async def text(self) -> str:
+        return ''
+
+    async def json(self) -> dict[str, Any]:
+        return self.payload
+
+
+@pytest.fixture
+def install_github_responses(monkeypatch: pytest.MonkeyPatch):
+    requested_urls: list[str] = []
+
+    def install(*responses: FakeResponse) -> list[str]:
+        response_iterator = iter(responses)
+
+        class FakeSession:
+            def __init__(self, *, headers: dict[str, str]) -> None:
+                pass
+
+            async def __aenter__(self) -> 'FakeSession':
+                return self
+
+            async def __aexit__(self, *_args: Any) -> None:
+                return None
+
+            def get(self, url: str, *, proxy: str | None) -> FakeResponse:
+                requested_urls.append(url)
+                return next(response_iterator)
+
+        monkeypatch.setattr(githubcode.aiohttp, 'ClientSession', FakeSession)
+        return requested_urls
+
+    monkeypatch.setattr(githubcode.Core, 'github_key', staticmethod(lambda: 'test-token'))
+    monkeypatch.setattr(githubcode.Core, 'get_user_agent', staticmethod(lambda: 'test-agent'))
+    monkeypatch.setattr(githubcode, 'get_delay', lambda: 0)
+    return install
+
+
+@pytest.mark.asyncio
+async def test_github_code_retains_only_the_requested_fragments_across_pages(install_github_responses) -> None:
+    requested_urls = install_github_responses(
+        FakeResponse(
+            {
+                'items': [
+                    {'text_matches': [{'fragment': 'Contact Admin@Example.COM'}]},
+                    {'text_matches': [{'fragment': 'API host API.EXAMPLE.COM'}]},
+                ]
+            },
+            {
+                'next': {'url': 'https://api.github.com/search/code?q=example.com&page=2'},
+                'last': {'url': 'https://api.github.com/search/code?q=example.com&page=2'},
+            },
+        ),
+        FakeResponse(
+            {
+                'items': [
+                    {'text_matches': [{'fragment': 'Docs host Docs.Example.Com'}]},
+                    {'text_matches': [{'fragment': 'Ignored host ignored.example.com'}]},
+                ]
+            },
+            {},
+        ),
+    )
+    search = githubcode.SearchGithubCode('example.com', limit=3)
+
+    await search.process()
+
+    assert requested_urls == [
+        'https://api.github.com/search/code?q="example.com"&page=1',
+        'https://api.github.com/search/code?q="example.com"&page=2',
+    ]
+    assert search.counter == 3
+    assert await search.get_emails() == {'admin@example.com'}
+    assert await search.get_hostnames() == ['api.example.com', 'docs.example.com', 'example.com']
+
+
+@pytest.mark.asyncio
+async def test_github_code_exact_limit_makes_no_additional_request(install_github_responses) -> None:
+    requested_urls = install_github_responses(
+        FakeResponse(
+            {
+                'items': [
+                    {'text_matches': [{'fragment': 'api.example.com'}]},
+                    {'text_matches': [{'fragment': 'docs.example.com'}]},
+                ]
+            },
+            {
+                'next': {'url': 'https://api.github.com/search/code?q=example.com&page=2'},
+                'last': {'url': 'https://api.github.com/search/code?q=example.com&page=2'},
+            },
+        )
+    )
+    search = githubcode.SearchGithubCode('example.com', limit=2)
+
+    await search.process()
+
+    assert requested_urls == ['https://api.github.com/search/code?q="example.com"&page=1']
+    assert await search.get_hostnames() == ['api.example.com', 'docs.example.com']
+
+
+@pytest.mark.asyncio
+async def test_github_code_keeps_provider_fragments_separate(install_github_responses) -> None:
+    install_github_responses(
+        FakeResponse(
+            {
+                'items': [
+                    {
+                        'text_matches': [
+                            {'fragment': 'admin'},
+                            {'fragment': '@example.com'},
+                        ]
+                    }
+                ]
+            },
+            {},
+        )
+    )
+    search = githubcode.SearchGithubCode('example.com', limit=10)
+
+    await search.process()
+
+    assert await search.get_emails() == set()
+
+
+@pytest.mark.asyncio
+async def test_github_code_ignores_non_string_fragments_without_repeating_the_page(install_github_responses) -> None:
+    requested_urls = install_github_responses(
+        FakeResponse(
+            {
+                'items': [
+                    {
+                        'text_matches': [
+                            {'fragment': None},
+                            {'fragment': 42},
+                            {'fragment': ''},
+                            {'fragment': 'API host api.example.com'},
+                        ]
+                    }
+                ]
+            },
+            {},
+        ),
+        FakeResponse({'items': []}, {}),
+    )
+    search = githubcode.SearchGithubCode('example.com', limit=10)
+
+    await search.process()
+
+    assert requested_urls == ['https://api.github.com/search/code?q="example.com"&page=1']
+    assert await search.get_hostnames() == ['api.example.com']
+
+
+@pytest.mark.parametrize(
+    'payload',
+    [
+        {'items': 'not-a-list'},
+        {'items': [{'text_matches': 'not-a-list'}]},
+    ],
+    ids=['malformed-items', 'malformed-text-matches'],
+)
+@pytest.mark.asyncio
+async def test_github_code_malformed_page_terminates_without_following_pagination(
+    install_github_responses,
+    payload: dict[str, Any],
+) -> None:
+    requested_urls = install_github_responses(
+        FakeResponse(
+            payload,
+            {
+                'next': {'url': 'https://api.github.com/search/code?q=example.com&page=2'},
+                'last': {'url': 'https://api.github.com/search/code?q=example.com&page=2'},
+            },
+        ),
+        FakeResponse({'items': []}, {}),
+    )
+    search = githubcode.SearchGithubCode('example.com', limit=10)
+
+    await search.process()
+
+    assert requested_urls == ['https://api.github.com/search/code?q="example.com"&page=1']
+    assert await search.get_emails() == set()
+    assert await search.get_hostnames() == []

theHarvester/discovery/githubcode.py

@@ -57,17 +57,25 @@ class SearchGithubCode:
 
     @staticmethod
     async def fragments_from_response(json_data: dict) -> list[str]:
-        try:
-            return [
-                match['fragment']
-                for item in json_data.get('items', [])
-                for match in item.get('text_matches', [])
-                if match.get('fragment') is not None
-            ]
-        except Exception as e:
-            logger.info(f'Error extracting fragments: {e}')
+        items = json_data.get('items', [])
+        if not isinstance(items, list):
             return []
 
+        fragments: list[str] = []
+        for item in items:
+            if not isinstance(item, dict):
+                continue
+            text_matches = item.get('text_matches', [])
+            if not isinstance(text_matches, list):
+                continue
+            for match in text_matches:
+                if not isinstance(match, dict):
+                    continue
+                fragment = match.get('fragment')
+                if isinstance(fragment, str) and fragment:
+                    fragments.append(fragment)
+        return fragments
+
     @staticmethod
     async def page_from_response(page: str, links) -> int | None:
         try:
@@ -116,7 +124,7 @@ class SearchGithubCode:
     async def process(self, proxy: bool = False) -> None:
         try:
             self.proxy = proxy
-            while self.counter <= self.limit and self.page != 0:
+            while self.counter < self.limit and self.page != 0:
                 try:
                     api_response = await self.do_search(self.page)
                     result = await self.handle_response(api_response)
@@ -125,8 +133,16 @@ class SearchGithubCode:
                         # Reset retry counter on any successful response
                         self.retry_count = 0
                         logger.info(f'\tSearching {self.counter} results.')
-                        self.total_results += ''.join(result.fragments)
-                        self.counter += len(result.fragments)
+                        remaining = self.limit - self.counter
+                        fragments = result.fragments[:remaining]
+                        if not fragments:
+                            self.page = 0
+                            break
+                        self.total_results += f'{" ".join(fragments)} '
+                        self.counter += len(fragments)
+                        if self.counter >= self.limit:
+                            self.page = 0
+                            break
                         next_or_last = result.next_page or result.last_page
                         # Break if pagination does not advance to avoid infinite loop
                         if next_or_last == self.page: