diff --git a/.github/getTrending.py b/.github/getTrending.py index c8ede4c7d26..b1940c73657 100644 --- a/.github/getTrending.py +++ b/.github/getTrending.py @@ -51,7 +51,7 @@ ROOT = os.path.join(os.path.dirname(os.path.abspath(__file__)), os.pardir) sys.path.insert(0, os.path.join(ROOT, "scripts")) from brand import BRAND, DESCRIPTION, SEARCH_GUIDE, SITE, SLUG -from update_cves import load_blacklist, qualifying_repo_cves, attach_source_artifacts, ensure_cve_entries +from update_cves import load_blacklist, qualifying_repo_cves, attach_source_artifacts, ensure_cve_entries, repository_is_nonpublic import releases from source_artifacts import reviewed_artifacts @@ -72,6 +72,8 @@ KEV_MARK = f'KEV tuple[int, list[dict]]: """How many repositories a search matches, and every one it exposes, most recently updated first.""" + if "is:public" not in query.split(): + query += " is:public" headers = {"Accept": "application/vnd.github+json", "User-Agent": USER_AGENT} # The token authenticates the API only; it is never placed in the URL, # written to disk, or printed, so it cannot leak through logs or the commit. @@ -115,7 +117,7 @@ def search(query: str) -> tuple[int, list[dict]]: # Results shift between pages while they are read, so a repository # can come back twice; GitHub names are case-insensitive. name = str(repo.get("full_name") or "").lower() - if name and name not in seen: + if name and name not in seen and not repository_is_nonpublic(repo): seen.add(name) found.append(repo) if not items or len(found) >= min(total, SEARCH_LIMIT): @@ -402,7 +404,7 @@ def qualifying_repositories( code: dict[str, list[str]] = {} blacklist = load_blacklist() fields = """ - databaseId createdAt pushedAt defaultBranchRef { target { oid } } + databaseId createdAt pushedAt isPrivate visibility defaultBranchRef { target { oid } } readmeMd: object(expression: \"HEAD:README.md\") { ... on Blob { text } } readmeUpper: object(expression: \"HEAD:README.MD\") { ... on Blob { text } } readmeRst: object(expression: \"HEAD:README.rst\") { ... on Blob { text } } @@ -435,6 +437,8 @@ def qualifying_repositories( "nameWithOwner": full_name, "description": repo.get("description") or "", "isFork": bool(repo.get("fork")), + "private": repo.get("private"), + "visibility": content.get("visibility") or repo.get("visibility"), "repositoryTopics": {"nodes": topics}, } cve = cve_of(repo) diff --git a/scripts/sync_pocingithub.py b/scripts/sync_pocingithub.py index 6983dec1013..c40b77f7e06 100644 --- a/scripts/sync_pocingithub.py +++ b/scripts/sync_pocingithub.py @@ -40,6 +40,7 @@ from update_cves import ( load_blacklist, ensure_cve_entries, qualifying_repo_cves, + repository_is_nonpublic, attach_source_artifacts, reconcile_inventory, replace_section, @@ -55,7 +56,7 @@ USER_AGENT = "0xMarcio-cve-poc-in-github" FIELDS = """ defaultBranchRef { target { oid } } - nameWithOwner url description isFork isArchived + nameWithOwner url description isFork isArchived isPrivate visibility repositoryTopics(first: 20) { nodes { topic { name } } } readmeMd: object(expression: "HEAD:README.md") { ... on Blob { text } } readmeUpper: object(expression: "HEAD:README.MD") { ... on Blob { text } } @@ -135,7 +136,7 @@ def describe(client: GitHubClient, names: list[str]) -> dict[str, dict]: described: dict[str, dict] = {} for index, name in enumerate(names): repo = data.get(f"r{index}") - if repo: + if repo and not repository_is_nonpublic(repo): described[name] = repo return described diff --git a/scripts/update_cves.py b/scripts/update_cves.py index 9ceba386dbb..f14064dcb8d 100755 --- a/scripts/update_cves.py +++ b/scripts/update_cves.py @@ -91,6 +91,8 @@ query($query: String!, $cursor: String) { description isArchived isFork + isPrivate + visibility pushedAt defaultBranchRef { target { oid } } repositoryTopics(first: 20) { nodes { topic { name } } } @@ -367,13 +369,18 @@ def readme_has_poc_context(readme: str, cve_id: str, full_name: str = "") -> boo return False +def repository_is_nonpublic(repo: dict[str, Any]) -> bool: + return (repo.get("isPrivate") is True or repo.get("private") is True + or str(repo.get("visibility") or "").lower() in {"private", "internal"}) + + def qualifying_repo_cves( repo: dict[str, Any], year: int, blacklist: Collection[str], ) -> set[str]: full_name = str(repo.get("nameWithOwner") or "") - if not full_name or repo.get("isFork"): + if not full_name or repo.get("isFork") or repository_is_nonpublic(repo): return set() owner, repo_name = full_name.lower().split("/", 1) if owner == repo_name or repo_name.endswith(".github.io"): @@ -553,6 +560,7 @@ class GitHubClient: full_names: list[str], ) -> tuple[set[str], dict[str, str], dict[str, list[dict[str, str]]]]: fields = """ + isPrivate visibility readmeMd: object(expression: \"HEAD:README.md\") { ... on Blob { text } } readmeUpper: object(expression: \"HEAD:README.MD\") { ... on Blob { text } } readmeRst: object(expression: \"HEAD:README.rst\") { ... on Blob { text } } @@ -587,7 +595,7 @@ class GitHubClient: existing: set[str] = set() for index, full_name in enumerate(names): repo = data.get(f"repo{index}") - if repo is None: + if repo is None or repository_is_nonpublic(repo): continue existing.add(full_name) text = readme_text(repo) @@ -609,7 +617,7 @@ def split_date_range(start: date, end: date) -> tuple[tuple[date, date], tuple[d def build_search_query(search_terms: str, qualifier: str, start: date, end: date) -> str: return ( - f"{search_terms} fork:false archived:false " + f"{search_terms} is:public fork:false archived:false " f"{qualifier}:{start.isoformat()}..{end.isoformat()}" ) @@ -655,7 +663,7 @@ def search_range( def search_without_range(client: GitHubClient, search_terms: str) -> Iterator[dict[str, Any]]: - query_text = f"{search_terms} fork:false archived:false" + query_text = f"{search_terms} is:public fork:false archived:false" cursor: str | None = None while True: page = client.search_page(query_text, cursor) @@ -691,7 +699,7 @@ def discover_github_pocs( def remember(repo: dict[str, Any]) -> None: full_name = str(repo.get("nameWithOwner") or "") - if full_name and not is_blacklisted_repo(full_name, blacklist): + if full_name and not repository_is_nonpublic(repo) and not is_blacklisted_repo(full_name, blacklist): repositories[full_name] = repo def collect(repo: dict[str, Any], year: int) -> None: @@ -750,7 +758,7 @@ def discover_github_pocs( for index, full_name in enumerate(missing): owner, name = full_name.split("/", 1) aliases.append(f"repo{index}: repository(owner: {json.dumps(owner)}, name: {json.dumps(name)}) " - "{ nameWithOwner url description isFork defaultBranchRef { target { oid } } }") + "{ nameWithOwner url description isFork isPrivate visibility defaultBranchRef { target { oid } } }") payload = http_json(GITHUB_GRAPHQL_URL, headers=client.headers, data={"query": "query { " + " ".join(aliases) + " rateLimit { remaining resetAt } }"}) replay = github_repository_data(payload, {f"repo{index}" for index in range(len(missing))}) @@ -807,7 +815,7 @@ def attach_source_artifacts( ) -> None: """Add locally verified nested evidence without weakening the ordinary gate.""" name = str(repo.get("nameWithOwner") or "") - if not name or repo.get("isFork") or is_blacklisted_repo(name, blacklist): + if not name or repo.get("isFork") or repository_is_nonpublic(repo) or is_blacklisted_repo(name, blacklist): return if "defaultBranchRef" in repo and repo["defaultBranchRef"] is None: return diff --git a/tests/test_public_discovery.py b/tests/test_public_discovery.py new file mode 100644 index 00000000000..4ae9ccd9dea --- /dev/null +++ b/tests/test_public_discovery.py @@ -0,0 +1,116 @@ +from __future__ import annotations + +import importlib.util +import io +import json +import sys +import unittest +from datetime import date +from pathlib import Path +from unittest.mock import Mock, patch +from urllib.parse import parse_qs, urlsplit + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / 'scripts')) +import source_artifacts +import sync_pocingithub +import update_cves + +SPEC = importlib.util.spec_from_file_location('public_trending', ROOT / '.github/getTrending.py') +trending = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(trending) +CVE = 'CVE-2024-1234' +NAME = 'researcher/' + CVE + + +def repo(**fields): + return {'nameWithOwner': NAME, 'url': 'https://github.com/' + NAME, 'description': 'Proof of concept', + 'isFork': False, 'defaultBranchRef': {'target': {'oid': 'a' * 40}}, + 'root': {'entries': [{'name': 'exploit.py', 'type': 'blob'}]}, + 'readmeMd': {'text': 'Proof of concept for ' + CVE}, **fields} + + +class PublicDiscoveryTests(unittest.TestCase): + def test_explicit_nonpublic_repositories_never_qualify(self): + for fields in ({'isPrivate': True}, {'private': True}, {'visibility': 'PRIVATE'}, {'visibility': 'INTERNAL'}): + with self.subTest(fields=fields): + self.assertEqual(update_cves.qualifying_repo_cves(repo(**fields), 2024, []), set()) + self.assertEqual(update_cves.qualifying_repo_cves(repo(isPrivate=False, visibility='PUBLIC'), 2024, []), {CVE}) + + def test_reviewed_private_payload_is_not_inspected_or_published(self): + row = source_artifacts.load_reviews()[0] + artifact = source_artifacts.ArtifactEvidence(row['cve'], row['repository'], row['path'], + row['revision'], row['sha256'], row['id']) + candidate = repo(nameWithOwner=row['repository'], isPrivate=True, _source_artifacts=(artifact,)) + with patch.object(source_artifacts, 'inspect_repository') as inspect: + update_cves.attach_source_artifacts(candidate, {}, []) + inspect.assert_not_called() + self.assertEqual(update_cves.qualifying_repo_cves(candidate, int(row['cve'].split('-')[1]), []), set()) + + def test_rest_search_is_public_only_and_discards_explicit_nonpublic_rows(self): + rows = [{'full_name': 'public/' + CVE, 'private': False, 'visibility': 'public'}, + {'full_name': 'private/' + CVE, 'private': True}, + {'full_name': 'internal/' + CVE, 'visibility': 'internal'}] + urls = [] + def response(req, **kwargs): + urls.append(req.full_url) + return io.BytesIO(json.dumps({'total_count': 3, 'items': rows}).encode()) + with patch.object(trending, 'github_token', return_value='test'), \ + patch.object(trending.request, 'urlopen', side_effect=response): + _, found = trending.search(CVE) + self.assertEqual(found, [rows[0]]) + self.assertTrue(all('is:public' in parse_qs(urlsplit(url).query)['q'][0].split() for url in urls)) + + def test_graphql_searches_request_public_visibility(self): + query = update_cves.build_search_query(CVE, 'pushed', date(2026, 10, 1), date(2026, 10, 7)) + self.assertIn('is:public', query.split()) + client = Mock() + client.search_page.return_value = {'repositoryCount': 0, 'nodes': [], 'pageInfo': {'hasNextPage': False}} + list(update_cves.search_without_range(client, CVE)) + self.assertIn('is:public', client.search_page.call_args.args[0].split()) + self.assertIn('isPrivate', update_cves.GITHUB_SEARCH_QUERY) + self.assertIn('visibility', update_cves.GITHUB_SEARCH_QUERY) + + def test_direct_description_and_readme_refresh_discard_nonpublic_nodes(self): + for visibility in ('PRIVATE', 'INTERNAL'): + candidate = repo(visibility=visibility) + for module, alias, read in ( + (sync_pocingithub, 'r0', lambda client: sync_pocingithub.describe(client, [NAME])), + (update_cves, 'repo0', lambda client: client.fetch_readmes([NAME])), + ): + payload = {'data': {alias: candidate, 'rateLimit': {'remaining': 100}}} + with self.subTest(visibility=visibility, module=module.__name__), \ + patch.object(module, 'http_json', return_value=payload) as fetch: + result = read(update_cves.GitHubClient('test')) + self.assertFalse(result[0] if isinstance(result, tuple) else result) + query = fetch.call_args.kwargs['data']['query'] + self.assertIn('isPrivate', query) + self.assertIn('visibility', query) + + def test_pending_replay_does_not_publish_newly_private_reviewed_repo(self): + row = source_artifacts.load_reviews()[0] + name, cve = row['repository'], row['cve'] + payload = {'data': {'repo0': repo(nameWithOwner=name, isPrivate=True), 'rateLimit': {'remaining': 100}}} + with patch.object(source_artifacts, 'pending_reviewed_names', return_value=[name.lower()]), \ + patch.object(update_cves, 'search_range', return_value=[]), \ + patch.object(update_cves, 'http_json', return_value=payload) as fetch, \ + patch.object(update_cves.GitHubClient, 'fetch_readmes') as readmes: + found = update_cves.discover_github_pocs('test', years=[int(cve.split('-')[1])], + lookback_days=3, backfill=False, cve_filter=set(), artifact_cache_write=False) + self.assertEqual(found, {}) + readmes.assert_not_called() + self.assertIn('isPrivate', fetch.call_args.kwargs['data']['query']) + self.assertIn('visibility', fetch.call_args.kwargs['data']['query']) + + def test_trending_refresh_rejects_private_or_internal_repository(self): + row = {'full_name': NAME, 'name': CVE, 'html_url': 'https://github.com/' + NAME} + for fields in ({'isPrivate': True}, {'visibility': 'INTERNAL'}): + with self.subTest(fields=fields), patch.object(trending, 'graphql', return_value={'r0': repo(**fields)}) as fetch: + accepted, paths = trending.qualifying_repositories([dict(row)], 'test') + self.assertEqual((accepted, paths), ([], {})) + self.assertIn('isPrivate', fetch.call_args.args[0]) + self.assertIn('visibility', fetch.call_args.args[0]) + + +if __name__ == '__main__': + unittest.main()