From 1d89775612827c2cd227737e1e1a4739d1d0f4e1 Mon Sep 17 00:00:00 2001 From: Seth Michael Larson Date: Tue, 28 Jul 2026 14:23:46 -0500 Subject: [PATCH 1/6] Automatically credit remediation developers and reviewers --- src/psrt_ghsa_bot/app.py | 62 +++++++++++++++++++++++++++++++++++++ tests/test_app.py | 66 +++++++++++++++++++++++++++++++++++++++- 2 files changed, 127 insertions(+), 1 deletion(-) diff --git a/src/psrt_ghsa_bot/app.py b/src/psrt_ghsa_bot/app.py index 971a5d6..2edc7d4 100644 --- a/src/psrt_ghsa_bot/app.py +++ b/src/psrt_ghsa_bot/app.py @@ -96,6 +96,64 @@ def get_repository_advisories( raise RuntimeError("Request to paginate advisories failed.") +def get_security_advisory_credits( + github: GitHub, + security_advisory: dict[str, typing.Any], +) -> list[dict[str, str]]: + """Generates a list of credits to apply to a security + advisory, such as developing or reviewing a remediation. + Respects credits that already exist on an advisory. + """ + + credits = [] + + def credit_if_uncredited(login: str, type: str) -> None: + # GHSA only allows one credit type per user, + # so we don't want to overwrite existing credits. + nonlocal credits + if any(c["login"] == login for c in (security_advisory["credits"] + credits)): + return + credits.append( + { + "login": login, + "type": type, + } + ) + + if (private_fork := security_advisory.get("private_fork")) is not None: + private_fork_owner = private_fork["owner"]["login"] + private_fork_repo = private_fork["name"] + + pull_requests = json.loads( + github.rest.pulls.list( + owner=private_fork_owner, + repo=private_fork_repo, + state="open", + ).content + ) + for pull_request in pull_requests: + # fmt: off + credit_if_uncredited( + login=pull_request["user"]["login"], + type="remediation_developer" + ) + # fmt: on + reviews = json.loads( + github.rest.pulls.list_reviews( + owner=private_fork_owner, + repo=private_fork_repo, + pull_number=pull_request["number"], + ).content + ) + for review in reviews: + credit_if_uncredited( + login=review["user"]["login"], + type="remediation_reviewer", + ) + + return credits + + def github_client_request(client: typing.Any, method: str, url: str, params: dict[str, str | int]) -> typing.Any: """Sends a raw HTTP request using a GitHub API client""" headers = {"X-GitHub-Api-Version": client._REST_API_VERSION} @@ -184,6 +242,10 @@ def apply_to_repo(github: GitHub, owner: str, repo: str, cve_api: CveApi, *, res patch_data["collaborating_teams"] = sorted(collaborating_teams) print(f" ➕ Will ensure team present: {PSRT_GITHUB_TEAM_SLUG}") + # Find new credits for the security advisory. + if credits := get_security_advisory_credits(github, security_advisory): + patch_data["credits"] = credits + # Apply updates, if any, to the security advisory. if patch_data: try: diff --git a/tests/test_app.py b/tests/test_app.py index e130477..e64316f 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -1,4 +1,5 @@ import datetime +import json from unittest import mock import pytest @@ -43,7 +44,6 @@ def _create_advisory_dict(state, cve_id, collaborating_teams, summary=""): "cve_id": cve_id, "collaborating_teams": [{"slug": team} for team in collaborating_teams], "collaborating_users": [{"login": "octocat", "id": 1, "type": "User"}], - "private_fork": {"name": "repo-ghsa-xxxx-xxxx-xxxx", "owner": {"login": "owner"}}, } @@ -252,6 +252,70 @@ def test_accepts_advisory_with_accept_tag(summary, cve_id, cve_reserve_response) ) +def test_get_security_advisory_credits_no_private_fork(): + github = mock.Mock() + credits = app.get_security_advisory_credits( + github=github, + security_advisory={ + "private_fork": None, + }, + ) + assert credits == [] + + +def test_get_security_advisory_credits_no_prs(): + github = mock.Mock() + pulls_list = mock.Mock() + pulls_list.content = "[]" + github.rest.pulls.list.return_value = pulls_list + credits = app.get_security_advisory_credits( + github=github, + security_advisory={ + "private_fork": { + "owner": {"login": "fork-owner"}, + "name": "fork-name", + }, + }, + ) + assert credits == [] + github.rest.pulls.list.assert_called_with( + owner="fork-owner", + repo="fork-name", + state="open", + ) + + +def test_get_security_advisory_credits(): + github = mock.Mock() + + pulls_list = mock.Mock() + pulls_list.content = json.dumps([{"number": 1, "user": {"login": "author"}}]) + github.rest.pulls.list.return_value = pulls_list + + reviews_list = mock.Mock() + reviews_list.content = json.dumps([{"user": {"login": "reviewer1"}}, {"user": {"login": "reviewer2"}}]) + github.rest.pulls.list_reviews.return_value = reviews_list + + credits = app.get_security_advisory_credits( + github=github, + security_advisory={ + "private_fork": { + "owner": {"login": "fork-owner"}, + "name": "fork-name", + }, + "credits": [ + {"type": "coordinator", "login": "reviewer1"}, + ], + }, + ) + + # reviewer1 is skipped because they are already coordinator. + assert credits == [ + {"login": "author", "type": "remediation_developer"}, + {"login": "reviewer2", "type": "remediation_reviewer"}, + ] + + def test_reserve_one_cve_id(cve_reserve_response, cve_id, year) -> None: cve_api = mock.Mock() cve_api.reserve.return_value = cve_reserve_response From 27dd5d5ae493c930ce40a7d83c14a4cc67aae01c Mon Sep 17 00:00:00 2001 From: Seth Michael Larson Date: Wed, 12 Aug 2026 13:20:18 -0500 Subject: [PATCH 2/6] Handle RequestFailed gracefully, don't clobber existing --- src/psrt_ghsa_bot/app.py | 45 ++++++++++++++++++++++++---------------- tests/test_app.py | 3 ++- 2 files changed, 29 insertions(+), 19 deletions(-) diff --git a/src/psrt_ghsa_bot/app.py b/src/psrt_ghsa_bot/app.py index 2edc7d4..4388dca 100644 --- a/src/psrt_ghsa_bot/app.py +++ b/src/psrt_ghsa_bot/app.py @@ -104,14 +104,13 @@ def get_security_advisory_credits( advisory, such as developing or reviewing a remediation. Respects credits that already exist on an advisory. """ - - credits = [] + credits = (security_advisory.get("credits", None) or [])[:] def credit_if_uncredited(login: str, type: str) -> None: # GHSA only allows one credit type per user, # so we don't want to overwrite existing credits. nonlocal credits - if any(c["login"] == login for c in (security_advisory["credits"] + credits)): + if any(c["login"].lower() == login.lower() for c in credits): return credits.append( { @@ -124,13 +123,18 @@ def credit_if_uncredited(login: str, type: str) -> None: private_fork_owner = private_fork["owner"]["login"] private_fork_repo = private_fork["name"] - pull_requests = json.loads( - github.rest.pulls.list( - owner=private_fork_owner, - repo=private_fork_repo, - state="open", - ).content - ) + try: + pull_requests = json.loads( + github.rest.pulls.list( + owner=private_fork_owner, + repo=private_fork_repo, + state="open", + ).content + ) + except RequestFailed: + capture_exception() + raise RuntimeError("Request to list pull requests failed") from None + for pull_request in pull_requests: # fmt: off credit_if_uncredited( @@ -138,20 +142,25 @@ def credit_if_uncredited(login: str, type: str) -> None: type="remediation_developer" ) # fmt: on - reviews = json.loads( - github.rest.pulls.list_reviews( - owner=private_fork_owner, - repo=private_fork_repo, - pull_number=pull_request["number"], - ).content - ) + try: + reviews = json.loads( + github.rest.pulls.list_reviews( + owner=private_fork_owner, + repo=private_fork_repo, + pull_number=pull_request["number"], + ).content + ) + except RequestFailed: + capture_exception() + raise RuntimeError("Request to list pull requests reviews failed") from None + for review in reviews: credit_if_uncredited( login=review["user"]["login"], type="remediation_reviewer", ) - return credits + return sorted(credits, key=lambda c: (c["login"], c["type"])) def github_client_request(client: typing.Any, method: str, url: str, params: dict[str, str | int]) -> typing.Any: diff --git a/tests/test_app.py b/tests/test_app.py index e64316f..ad4c48e 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -309,9 +309,10 @@ def test_get_security_advisory_credits(): }, ) - # reviewer1 is skipped because they are already coordinator. + # reviewer1 is kept as 'coordinator', not 'remediation_reviewer'. assert credits == [ {"login": "author", "type": "remediation_developer"}, + {"login": "reviewer1", "type": "coordinator"}, {"login": "reviewer2", "type": "remediation_reviewer"}, ] From 3593cbfe091d87db42821804ef2fff4a443c992b Mon Sep 17 00:00:00 2001 From: Seth Michael Larson Date: Wed, 12 Aug 2026 13:23:24 -0500 Subject: [PATCH 3/6] Don't credit as reviewer for self-review --- src/psrt_ghsa_bot/app.py | 8 ++++++-- tests/test_app.py | 26 ++++++++++++++++++++++++++ 2 files changed, 32 insertions(+), 2 deletions(-) diff --git a/src/psrt_ghsa_bot/app.py b/src/psrt_ghsa_bot/app.py index 4388dca..c392db0 100644 --- a/src/psrt_ghsa_bot/app.py +++ b/src/psrt_ghsa_bot/app.py @@ -137,8 +137,9 @@ def credit_if_uncredited(login: str, type: str) -> None: for pull_request in pull_requests: # fmt: off + pull_request_author = pull_request["user"]["login"] credit_if_uncredited( - login=pull_request["user"]["login"], + login=pull_request_author, type="remediation_developer" ) # fmt: on @@ -155,8 +156,11 @@ def credit_if_uncredited(login: str, type: str) -> None: raise RuntimeError("Request to list pull requests reviews failed") from None for review in reviews: + review_login = review["user"]["login"] + if review_login == pull_request_author: + continue # Developers can't be reviewers too. credit_if_uncredited( - login=review["user"]["login"], + login=review_login, type="remediation_reviewer", ) diff --git a/tests/test_app.py b/tests/test_app.py index ad4c48e..390ce6a 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -317,6 +317,32 @@ def test_get_security_advisory_credits(): ] +def test_get_security_advisory_credits_self_review(): + github = mock.Mock() + + pulls_list = mock.Mock() + pulls_list.content = json.dumps([{"number": 1, "user": {"login": "author"}}]) + github.rest.pulls.list.return_value = pulls_list + + reviews_list = mock.Mock() + reviews_list.content = json.dumps([{"user": {"login": "author"}}]) + github.rest.pulls.list_reviews.return_value = reviews_list + + credits = app.get_security_advisory_credits( + github=github, + security_advisory={ + "private_fork": { + "owner": {"login": "fork-owner"}, + "name": "fork-name", + }, + "credits": [], + }, + ) + + # Developer is favored over reviewer. + assert credits == [{"login": "author", "type": "remediation_developer"}] + + def test_reserve_one_cve_id(cve_reserve_response, cve_id, year) -> None: cve_api = mock.Mock() cve_api.reserve.return_value = cve_reserve_response From 4ca76f68a45ff7a0453be3745e85ad3f493fc91d Mon Sep 17 00:00:00 2001 From: Seth Michael Larson Date: Fri, 21 Aug 2026 12:23:01 -0500 Subject: [PATCH 4/6] Address review comments --- src/psrt_ghsa_bot/app.py | 24 ++++++++++++------- tests/test_app.py | 51 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 66 insertions(+), 9 deletions(-) diff --git a/src/psrt_ghsa_bot/app.py b/src/psrt_ghsa_bot/app.py index c392db0..1d47a20 100644 --- a/src/psrt_ghsa_bot/app.py +++ b/src/psrt_ghsa_bot/app.py @@ -124,11 +124,15 @@ def credit_if_uncredited(login: str, type: str) -> None: private_fork_repo = private_fork["name"] try: + # Pagination shouldn't be necessary here, there isn't likely + # to be more than 100 pull requests on a single GHSA private repo. + # Usually there'll be 2 at most. pull_requests = json.loads( github.rest.pulls.list( owner=private_fork_owner, repo=private_fork_repo, state="open", + per_page=100, ).content ) except RequestFailed: @@ -136,13 +140,8 @@ def credit_if_uncredited(login: str, type: str) -> None: raise RuntimeError("Request to list pull requests failed") from None for pull_request in pull_requests: - # fmt: off pull_request_author = pull_request["user"]["login"] - credit_if_uncredited( - login=pull_request_author, - type="remediation_developer" - ) - # fmt: on + credit_if_uncredited(login=pull_request_author, type="remediation_developer") try: reviews = json.loads( github.rest.pulls.list_reviews( @@ -164,7 +163,7 @@ def credit_if_uncredited(login: str, type: str) -> None: type="remediation_reviewer", ) - return sorted(credits, key=lambda c: (c["login"], c["type"])) + return sort_security_advisory_credits(credits) def github_client_request(client: typing.Any, method: str, url: str, params: dict[str, str | int]) -> typing.Any: @@ -190,6 +189,11 @@ def reserve_one_cve(cve_api: CveApi) -> str: return cve_ids[0] +def sort_security_advisory_credits(credits: list[dict[str, str]]) -> list[dict[str, str]]: + """Sorts the 'credits' field in a GitHub Security Advisory for comparison""" + return sorted(credits, key=lambda c: (c["login"], c["type"])) + + def apply_to_repo(github: GitHub, owner: str, repo: str, cve_api: CveApi, *, reserve_cves: bool = True) -> None: """Applies the PSRT GitHub Security Advisory process to the repository.""" security_advisories = get_repository_advisories(github, owner, repo) @@ -256,8 +260,10 @@ def apply_to_repo(github: GitHub, owner: str, repo: str, cve_api: CveApi, *, res print(f" ➕ Will ensure team present: {PSRT_GITHUB_TEAM_SLUG}") # Find new credits for the security advisory. - if credits := get_security_advisory_credits(github, security_advisory): - patch_data["credits"] = credits + existing_credits = sort_security_advisory_credits(security_advisory.get("credits", None) or []) + new_credits = get_security_advisory_credits(github, security_advisory) + if new_credits and existing_credits != new_credits: + patch_data["credits"] = new_credits # Apply updates, if any, to the security advisory. if patch_data: diff --git a/tests/test_app.py b/tests/test_app.py index 390ce6a..b6460d2 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -282,6 +282,7 @@ def test_get_security_advisory_credits_no_prs(): owner="fork-owner", repo="fork-name", state="open", + per_page=100, ) @@ -343,6 +344,56 @@ def test_get_security_advisory_credits_self_review(): assert credits == [{"login": "author", "type": "remediation_developer"}] +@pytest.mark.parametrize("remove_credits", [(), ("reviewer1",), ("reviewer1", "reviewer2")]) +def test_get_security_advisory_credits_no_change(remove_credits: tuple[str, ...]) -> None: + github = mock.Mock() + cve_api = mock.Mock() + + pulls_list = mock.Mock() + pulls_list.content = json.dumps([{"number": 1, "user": {"login": "author"}}]) + github.rest.pulls.list.return_value = pulls_list + + reviews_list = mock.Mock() + reviews_list.content = json.dumps([{"user": {"login": "reviewer1"}}, {"user": {"login": "reviewer2"}}]) + github.rest.pulls.list_reviews.return_value = reviews_list + + security_advisory = _create_advisory_dict("draft", "CVE-2026-1234", ["psrt"]) + security_advisory["private_fork"] = { + "owner": {"login": "fork-owner"}, + "name": "fork-name", + } + security_advisory["credits"] = [ + # Deliberately out of sorting order. + {"login": "reviewer2", "type": "remediation_reviewer"}, + {"login": "author", "type": "remediation_developer"}, + {"login": "reviewer1", "type": "coordinator"}, + ] + + if remove_credits: + security_advisory["credits"] = [c for c in security_advisory["credits"] if c["login"] not in remove_credits] + + with mock.patch("psrt_ghsa_bot.app.get_repository_advisories") as get_repo_advs: + get_repo_advs.return_value = [security_advisory] + + app.apply_to_repo(github, "owner", "repo", cve_api) + + if remove_credits: + github.rest.security_advisories.update_repository_advisory.assert_called_once_with( + owner="owner", + repo="repo", + ghsa_id="GHSA-xxxx-xxxx-xxxx", + data={ + "credits": [ + {"login": "author", "type": "remediation_developer"}, + {"login": "reviewer1", "type": "remediation_reviewer"}, + {"login": "reviewer2", "type": "remediation_reviewer"}, + ] + }, + ) + else: # Nothing to update. + github.rest.security_advisories.update_repository_advisory.assert_not_called() + + def test_reserve_one_cve_id(cve_reserve_response, cve_id, year) -> None: cve_api = mock.Mock() cve_api.reserve.return_value = cve_reserve_response From f5f5de9cc3bc520f1019a759135ee99fd1057c0d Mon Sep 17 00:00:00 2001 From: Seth Larson Date: Fri, 21 Aug 2026 15:15:17 -0500 Subject: [PATCH 5/6] Apply suggestions from code review Co-authored-by: Stan Ulbrych --- src/psrt_ghsa_bot/app.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/psrt_ghsa_bot/app.py b/src/psrt_ghsa_bot/app.py index 1d47a20..c1a940e 100644 --- a/src/psrt_ghsa_bot/app.py +++ b/src/psrt_ghsa_bot/app.py @@ -109,7 +109,6 @@ def get_security_advisory_credits( def credit_if_uncredited(login: str, type: str) -> None: # GHSA only allows one credit type per user, # so we don't want to overwrite existing credits. - nonlocal credits if any(c["login"].lower() == login.lower() for c in credits): return credits.append( @@ -264,6 +263,7 @@ def apply_to_repo(github: GitHub, owner: str, repo: str, cve_api: CveApi, *, res new_credits = get_security_advisory_credits(github, security_advisory) if new_credits and existing_credits != new_credits: patch_data["credits"] = new_credits + print(f" 📋 Will add credits for developing and reviewing remediation") # Apply updates, if any, to the security advisory. if patch_data: From 68b965d8fd4dc40cbad9664d725db7f82edcb6b5 Mon Sep 17 00:00:00 2001 From: Seth Larson Date: Fri, 21 Aug 2026 15:16:40 -0500 Subject: [PATCH 6/6] Remove f-string --- src/psrt_ghsa_bot/app.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/psrt_ghsa_bot/app.py b/src/psrt_ghsa_bot/app.py index c1a940e..b3bd002 100644 --- a/src/psrt_ghsa_bot/app.py +++ b/src/psrt_ghsa_bot/app.py @@ -263,7 +263,7 @@ def apply_to_repo(github: GitHub, owner: str, repo: str, cve_api: CveApi, *, res new_credits = get_security_advisory_credits(github, security_advisory) if new_credits and existing_credits != new_credits: patch_data["credits"] = new_credits - print(f" 📋 Will add credits for developing and reviewing remediation") + print(" 📋 Will add credits for developing and reviewing remediation") # Apply updates, if any, to the security advisory. if patch_data: