From eeff3816c3882deb62f33aafdb38fcb3d0dcc8b3 Mon Sep 17 00:00:00 2001 From: NotoriousRebel <36310667+NotoriousRebel@users.noreply.github.com> Date: Wed, 5 Aug 2026 13:53:08 -0400 Subject: [PATCH] fix: normalize GitLab public discovery (#243) --- CHANGELOG.md | 1 + tests/discovery/test_gitlabsearch.py | 120 +++++++++++++++++++++++++ theHarvester/discovery/gitlabsearch.py | 51 +++++------ 3 files changed, 145 insertions(+), 27 deletions(-) create mode 100644 tests/discovery/test_gitlabsearch.py diff --git a/CHANGELOG.md b/CHANGELOG.md index a82a38e4..dee12219 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -31,6 +31,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - Fixed GitHub code-search fragment limits, boundary separation, and malformed-page termination with offline provider tests. +- Fixed GitLab public discovery scope normalization, request bounds, and default-branch README requests. - 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)). diff --git a/tests/discovery/test_gitlabsearch.py b/tests/discovery/test_gitlabsearch.py new file mode 100644 index 00000000..593227a5 --- /dev/null +++ b/tests/discovery/test_gitlabsearch.py @@ -0,0 +1,120 @@ +import json +from typing import Any + +import pytest + +from theHarvester.discovery import gitlabsearch + + +@pytest.mark.asyncio +async def test_public_discovery_normalizes_evidence_and_uses_bounded_requests( + monkeypatch: pytest.MonkeyPatch, +) -> None: + requests: list[dict[str, object]] = [] + projects = [ + { + 'id': 'group/project', + 'default_branch': 'feature/readme', + 'description': 'API.Example.TEST. and false.example.test.evil', + 'name': 'Example project', + 'path_with_namespace': 'group/project', + 'web_url': 'https://gitlab.com/group/project', + } + ] + users = [ + { + 'name': 'Example user', + 'username': 'example', + 'bio': 'Status.Example.TEST.', + 'web_url': 'https://gitlab.com/example', + 'website_url': 'https://Portal.Example.TEST./profile', + 'public_email': 'SECURITY@Example.TEST', + }, + { + 'name': 'Outsider', + 'username': 'outsider', + 'bio': 'api.notexample.test and example.test.evil', + 'web_url': 'https://gitlab.com/outsider', + 'website_url': 'https://api.notexample.test', + 'public_email': 'outsider@notexample.test', + }, + ] + + async def fake_fetch_all( + urls: list[str] | set[str], + headers: dict[str, str] | None = None, + proxy: bool = False, + **_kwargs: Any, + ) -> list[str]: + url = next(iter(urls)) + requests.append({'url': url, 'headers': headers, 'proxy': proxy}) + responses = { + 'https://gitlab.com/api/v4/projects?search=example.test&per_page=20': json.dumps(projects), + 'https://gitlab.com/api/v4/projects/group%2Fproject/repository/files/README.md/raw?ref=feature%2Freadme': ( + 'Contact Admin@Example.TEST. at docs.example.test; ignore admin@notexample.test' + ), + 'https://gitlab.com/api/v4/projects?search=*.example.test&per_page=20': '[]', + 'https://gitlab.com/api/v4/users?search=example.test&per_page=10': json.dumps(users), + } + if url not in responses: + raise AssertionError(f'unexpected GitLab request: {url}') + return [responses[url]] + + monkeypatch.setattr(gitlabsearch.Core, 'get_user_agent', staticmethod(lambda: 'UA')) + monkeypatch.setattr(gitlabsearch.AsyncFetcher, 'fetch_all', fake_fetch_all) + search = gitlabsearch.SearchGitlab('example.test') + + await search.process(proxy=True) + + assert requests == [ + {'url': url, 'headers': {'User-agent': 'UA'}, 'proxy': True} + for url in ( + 'https://gitlab.com/api/v4/projects?search=example.test&per_page=20', + 'https://gitlab.com/api/v4/projects/group%2Fproject/repository/files/README.md/raw?ref=feature%2Freadme', + 'https://gitlab.com/api/v4/projects?search=*.example.test&per_page=20', + 'https://gitlab.com/api/v4/users?search=example.test&per_page=10', + ) + ] + assert await search.get_hostnames() == { + 'api.example.test', + 'docs.example.test', + 'example.test', + 'portal.example.test', + 'status.example.test', + } + assert await search.get_emails() == {'admin@example.test', 'security@example.test'} + + +@pytest.mark.asyncio +async def test_decoded_pages_are_accepted_without_silent_slicing(monkeypatch: pytest.MonkeyPatch) -> None: + projects = [ + { + 'id': index, + 'default_branch': None, + 'description': f'project-{index}.example.test', + } + for index in range(1, 22) + ] + users = [{'public_email': f'user-{index}@example.test'} for index in range(1, 12)] + + async def fake_fetch_all(urls: list[str] | set[str], **_kwargs: Any) -> list[object]: + url = next(iter(urls)) + if 'projects?search=example.test&' in url: + return [projects] + if 'projects?search=*.example.test&' in url: + return [[]] + if '/users?' in url: + return [users] + raise AssertionError(f'unexpected GitLab request: {url}') + + monkeypatch.setattr(gitlabsearch.AsyncFetcher, 'fetch_all', fake_fetch_all) + search = gitlabsearch.SearchGitlab('example.test') + + await search.process() + + hostnames = await search.get_hostnames() + emails = await search.get_emails() + assert len(hostnames) == 21 + assert 'project-21.example.test' in hostnames + assert len(emails) == 11 + assert 'user-11@example.test' in emails diff --git a/theHarvester/discovery/gitlabsearch.py b/theHarvester/discovery/gitlabsearch.py index 263e1a0f..f0f141e7 100644 --- a/theHarvester/discovery/gitlabsearch.py +++ b/theHarvester/discovery/gitlabsearch.py @@ -2,8 +2,10 @@ import json as _stdlib_json import logging import re from types import ModuleType +from urllib.parse import quote from theHarvester.lib.core import AsyncFetcher, Core +from theHarvester.lib.hostnames import normalize_scoped_hostname logger = logging.getLogger(__name__) @@ -19,7 +21,7 @@ except Exception: class SearchGitlab: - """Class uses GitLab API to search for domain references in projects and code""" + """Search public GitLab project metadata, README files, and user profiles.""" def __init__(self, word) -> None: self.word = word @@ -30,9 +32,9 @@ class SearchGitlab: self.hostname = 'https://gitlab.com' @staticmethod - def _safe_parse_json(payload: object) -> dict: - # If already a dict, return it; if string, try parse; else return {} - if isinstance(payload, dict): + def _safe_parse_json(payload: object) -> dict | list: + # If already decoded, return it; if string, try parse; else return {} + if isinstance(payload, (dict, list)): return payload if isinstance(payload, str): try: @@ -47,14 +49,8 @@ class SearchGitlab: if not text: return domains - # Look for subdomains of our target domain - pattern = rf'[a-zA-Z0-9.-]*\.{re.escape(self.word)}' - matches = re.findall(pattern, text, re.IGNORECASE) - - for match in matches: - # Clean up the match - domain = match.lower().strip('.') - if domain.endswith(self.word) and domain != self.word: + for candidate in re.findall(r'[a-zA-Z0-9.-]+', text): + if domain := normalize_scoped_hostname(candidate.strip('.'), self.word): domains.add(domain) return domains @@ -65,13 +61,10 @@ class SearchGitlab: if not text: return emails - email_pattern = rf'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]*\.?{re.escape(self.word)}' - matches = re.findall(email_pattern, text, re.IGNORECASE) - - for match in matches: - email = match.lower() - if self.word in email: - emails.add(email) + for candidate in re.findall(r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+', text): + local_part, domain = candidate.lower().split('@', maxsplit=1) + if normalized_domain := normalize_scoped_hostname(domain, self.word): + emails.add(f'{local_part}@{normalized_domain}') return emails @@ -85,7 +78,7 @@ class SearchGitlab: for term in search_terms: # Search projects - projects_url = f'{self.hostname}/api/v4/projects?search={term}&per_page=100' + projects_url = f'{self.hostname}/api/v4/projects?search={term}&per_page=20' response = await AsyncFetcher.fetch_all([projects_url], headers=headers, proxy=self.proxy) if not response or not isinstance(response, list) or not response[0]: @@ -96,7 +89,7 @@ class SearchGitlab: if not isinstance(projects, list): continue - for project in projects[:20]: # Limit to first 20 projects + for project in projects: if not isinstance(project, dict): continue @@ -117,8 +110,12 @@ class SearchGitlab: # Try to get README content for more detailed search project_id = project.get('id') - if project_id: - readme_url = f'{self.hostname}/api/v4/projects/{project_id}/repository/files/README.md/raw?ref=main' + default_branch = project.get('default_branch') + if project_id and isinstance(default_branch, str) and default_branch: + readme_url = ( + f'{self.hostname}/api/v4/projects/{quote(str(project_id), safe="")}' + f'/repository/files/README.md/raw?ref={quote(default_branch, safe="")}' + ) try: readme_response = await AsyncFetcher.fetch_all([readme_url], headers=headers, proxy=self.proxy) if readme_response and readme_response[0]: @@ -142,7 +139,7 @@ class SearchGitlab: headers = {'User-agent': Core.get_user_agent()} # Search for users mentioning our domain - users_url = f'{self.hostname}/api/v4/users?search={self.word}&per_page=50' + users_url = f'{self.hostname}/api/v4/users?search={self.word}&per_page=10' response = await AsyncFetcher.fetch_all([users_url], headers=headers, proxy=self.proxy) if not response or not isinstance(response, list) or not response[0]: @@ -153,7 +150,7 @@ class SearchGitlab: if not isinstance(users, list): return - for user in users[:10]: # Limit to first 10 users + for user in users: if not isinstance(user, dict): continue @@ -170,8 +167,8 @@ class SearchGitlab: self.totalhosts.update(self._extract_domains_from_text(all_text)) # Check email - if public_email and self.word in public_email: - self.totalemails.add(public_email) + if public_email: + self.totalemails.update(self._extract_emails_from_text(public_email)) # Check website URL if website_url and self.word in website_url: