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/3] 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/3] 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/3] 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