From ab8f8571dd44702af56229964eb8cfbf7d6869da Mon Sep 17 00:00:00 2001 From: Karth Date: Mon, 24 Aug 2026 21:32:39 -0400 Subject: [PATCH] Fix credential_audit.py silently dropping results past the first search page _search() only ever fetched one page from GitHub's Search API, so a subject with more issues (find_spray_citations, per_page=100) or merged PRs (find_merges, per_page=50) than fit on the first page had every result past that page silently dropped. GitHub's Search API returns up to 1000 results per query across multiple pages; a single unpaginated request only sees the first one. This is exactly backwards for what this tool is for: a contributor prolific enough to have filed more than 100 issues total is also the contributor most likely to have real spray citations sitting on page 2+, so the audit could under-report risk (or report NONE/LOW) precisely for the highest-volume accounts it exists to catch. Fixes it in both copies of the tool (the packaged agent_compliance.cli.credential_audit module and the standalone scripts/credential_audit.py), since they implement the same _search() helper independently. Pages through up to GitHub's 1000-result window, stopping early on a short (last) page. Added pagination tests to both test suites: verifies multi-page aggregation, that a short page stops iteration, that the 1000-result window caps iteration rather than looping unboundedly, and that an empty first page returns no items. Confirmed no regressions: existing tests in both suites still pass (13/13 in scripts/tests, 7/7 in agent-governance-python/agent-compliance/tests, run locally with pytest 9.0.3 / Python 3.12.3). Note: agent_compliance/cli/contributor_check.py's _search_issues() has the identical single-page pattern at several call sites. Left untouched to keep this PR scoped to credential_audit.py - flagging it here in case it's worth a follow-up. Signed-off-by: Karth --- .../agent_compliance/cli/credential_audit.py | 24 +++++++- .../tests/test_credential_audit.py | 59 +++++++++++++++++++ scripts/credential_audit.py | 24 +++++++- scripts/tests/test_credential_audit.py | 58 ++++++++++++++++++ 4 files changed, 161 insertions(+), 4 deletions(-) create mode 100644 agent-governance-python/agent-compliance/tests/test_credential_audit.py diff --git a/agent-governance-python/agent-compliance/src/agent_compliance/cli/credential_audit.py b/agent-governance-python/agent-compliance/src/agent_compliance/cli/credential_audit.py index 8a4a0aca9..8e95be7b4 100644 --- a/agent-governance-python/agent-compliance/src/agent_compliance/cli/credential_audit.py +++ b/agent-governance-python/agent-compliance/src/agent_compliance/cli/credential_audit.py @@ -145,9 +145,29 @@ def _api(path: str, params: dict[str, str] | None = None) -> Any: return None +# GitHub's search endpoints cap results at 1000 total (the "Search API" +# result-window limit, documented at +# https://docs.github.com/en/rest/search#about-search). A single request +# only ever returns one page of up to `per_page` items, so a subject with +# more issues/PRs than fit on the first page silently loses coverage on +# exactly the pages most likely to contain the later, farther-out citations +# a real credential spray would produce. Page through the full result +# window instead of trusting the first response alone. +_SEARCH_RESULT_WINDOW = 1000 + + def _search(endpoint: str, query: str, per_page: int = 100) -> list[dict]: - data = _api(f"/search/{endpoint}", {"q": query, "per_page": str(per_page)}) - return data.get("items", []) if data else [] + items: list[dict] = [] + max_pages = max(1, _SEARCH_RESULT_WINDOW // per_page) + for page in range(1, max_pages + 1): + data = _api(f"/search/{endpoint}", {"q": query, "per_page": str(per_page), "page": str(page)}) + page_items = data.get("items", []) if data else [] + if not page_items: + break + items.extend(page_items) + if len(page_items) < per_page: + break + return items # --------------------------------------------------------------------------- diff --git a/agent-governance-python/agent-compliance/tests/test_credential_audit.py b/agent-governance-python/agent-compliance/tests/test_credential_audit.py new file mode 100644 index 000000000..c3e9c8489 --- /dev/null +++ b/agent-governance-python/agent-compliance/tests/test_credential_audit.py @@ -0,0 +1,59 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. +"""Tests for the packaged credential_audit CLI module.""" + +from unittest.mock import patch + +from agent_compliance.cli import credential_audit + + +class TestSearchPagination: + def test_paginates_across_multiple_pages(self): + """A full first page must still trigger a request for the next page, + not be treated as the end of the results.""" + pages = { + "1": {"items": [{"number": i} for i in range(100)]}, + "2": {"items": [{"number": i} for i in range(100, 150)]}, + } + + def fake_api(path, params=None): + return pages.get(params["page"]) + + with patch.object(credential_audit, "_api", side_effect=fake_api) as mock_api: + items = credential_audit._search("issues", "author:x is:issue", per_page=100) + + assert len(items) == 150 + assert mock_api.call_count == 2 + + def test_stops_when_a_short_page_is_returned(self): + pages = {"1": {"items": [{"number": 1}, {"number": 2}]}} + + def fake_api(path, params=None): + return pages.get(params["page"]) + + with patch.object(credential_audit, "_api", side_effect=fake_api) as mock_api: + items = credential_audit._search("issues", "author:x is:issue", per_page=100) + + assert len(items) == 2 + assert mock_api.call_count == 1 + + def test_stops_at_github_search_result_window(self): + """GitHub's Search API never returns more than 1000 results for a + query; a subject with more than that must not cause unbounded + pagination.""" + + def fake_api(path, params=None): + return {"items": [{"number": i} for i in range(int(params["per_page"]))]} + + with patch.object(credential_audit, "_api", side_effect=fake_api) as mock_api: + items = credential_audit._search("issues", "author:x is:issue", per_page=100) + + assert len(items) == 1000 + assert mock_api.call_count == 10 + + def test_empty_first_page_returns_no_items(self): + with patch.object(credential_audit, "_api", return_value=None) as mock_api: + items = credential_audit._search("issues", "author:x is:issue", per_page=100) + + assert items == [] + assert mock_api.call_count == 1 diff --git a/scripts/credential_audit.py b/scripts/credential_audit.py index 680b8045e..cbd379653 100644 --- a/scripts/credential_audit.py +++ b/scripts/credential_audit.py @@ -86,9 +86,29 @@ def _api(path: str, params: dict[str, str] | None = None) -> Any: raise +# GitHub's search endpoints cap results at 1000 total (the "Search API" +# result-window limit, documented at +# https://docs.github.com/en/rest/search#about-search). A single request +# only ever returns one page of up to `per_page` items, so a subject with +# more issues/PRs than fit on the first page silently loses coverage on +# exactly the pages most likely to contain the later, farther-out citations +# a real credential spray would produce. Page through the full result +# window instead of trusting the first response alone. +_SEARCH_RESULT_WINDOW = 1000 + + def _search(endpoint: str, query: str, per_page: int = 100) -> list[dict]: - data = _api(f"/search/{endpoint}", {"q": query, "per_page": str(per_page)}) - return data.get("items", []) if data else [] + items: list[dict] = [] + max_pages = max(1, _SEARCH_RESULT_WINDOW // per_page) + for page in range(1, max_pages + 1): + data = _api(f"/search/{endpoint}", {"q": query, "per_page": str(per_page), "page": str(page)}) + page_items = data.get("items", []) if data else [] + if not page_items: + break + items.extend(page_items) + if len(page_items) < per_page: + break + return items # --------------------------------------------------------------------------- diff --git a/scripts/tests/test_credential_audit.py b/scripts/tests/test_credential_audit.py index f1218937e..4d84b43e3 100644 --- a/scripts/tests/test_credential_audit.py +++ b/scripts/tests/test_credential_audit.py @@ -15,6 +15,7 @@ sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) +import credential_audit from credential_audit import ( MergeRecord, SprayCitation, @@ -23,6 +24,63 @@ ) +# --------------------------------------------------------------------------- +# _search pagination tests +# --------------------------------------------------------------------------- + +class TestSearchPagination: + def test_paginates_across_multiple_pages(self): + """A short first page must still trigger a request for the next page, + not be treated as the end of the results.""" + pages = { + "1": {"items": [{"number": i} for i in range(100)]}, + "2": {"items": [{"number": i} for i in range(100, 150)]}, + } + + def fake_api(path, params=None): + return pages.get(params["page"]) + + with patch.object(credential_audit, "_api", side_effect=fake_api) as mock_api: + items = credential_audit._search("issues", "author:x is:issue", per_page=100) + + assert len(items) == 150 + assert mock_api.call_count == 2 + + def test_stops_when_a_short_page_is_returned(self): + """A page with fewer than per_page items is the last page; no further + request should be made.""" + pages = {"1": {"items": [{"number": 1}, {"number": 2}]}} + + def fake_api(path, params=None): + return pages.get(params["page"]) + + with patch.object(credential_audit, "_api", side_effect=fake_api) as mock_api: + items = credential_audit._search("issues", "author:x is:issue", per_page=100) + + assert len(items) == 2 + assert mock_api.call_count == 1 + + def test_stops_at_github_search_result_window(self): + """GitHub's Search API never returns more than 1000 results for a + query; a subject with more than that must not cause unbounded + pagination.""" + def fake_api(path, params=None): + return {"items": [{"number": i} for i in range(int(params["per_page"]))]} + + with patch.object(credential_audit, "_api", side_effect=fake_api) as mock_api: + items = credential_audit._search("issues", "author:x is:issue", per_page=100) + + assert len(items) == 1000 + assert mock_api.call_count == 10 + + def test_empty_first_page_returns_no_items(self): + with patch.object(credential_audit, "_api", return_value=None) as mock_api: + items = credential_audit._search("issues", "author:x is:issue", per_page=100) + + assert items == [] + assert mock_api.call_count == 1 + + # --------------------------------------------------------------------------- # CredentialAuditReport tests # ---------------------------------------------------------------------------