Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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


# ---------------------------------------------------------------------------
Expand Down
Original file line number Diff line number Diff line change
@@ -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
24 changes: 22 additions & 2 deletions scripts/credential_audit.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


# ---------------------------------------------------------------------------
Expand Down
58 changes: 58 additions & 0 deletions scripts/tests/test_credential_audit.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@

sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

import credential_audit
from credential_audit import (
MergeRecord,
SprayCitation,
Expand All @@ -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
# ---------------------------------------------------------------------------
Expand Down
Loading