Automatically credit remediation developers and reviewers

src/psrt_ghsa_bot/app.py

@@ -96,6 +96,75 @@ 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 = (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.
+        if any(c["login"].lower() == login.lower() for c in 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"]
+
+        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:
+            capture_exception()
+            raise RuntimeError("Request to list pull requests failed") from None
+
+        for pull_request in pull_requests:
+            pull_request_author = pull_request["user"]["login"]
+            credit_if_uncredited(login=pull_request_author, type="remediation_developer")
+            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:
+                review_login = review["user"]["login"]
+                if review_login == pull_request_author:
+                    continue  # Developers can't be reviewers too.
+                credit_if_uncredited(
+                    login=review_login,
+                    type="remediation_reviewer",
+                )
+
+    return sort_security_advisory_credits(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}
@@ -119,6 +188,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)
@@ -184,6 +258,13 @@ 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.
+        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
+            print("       📋 Will add credits for developing and reviewing remediation")
+
         # Apply updates, if any, to the security advisory.
         if patch_data:
             try:

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,148 @@ 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",
+        per_page=100,
+    )
+
+
+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 kept as 'coordinator', not 'remediation_reviewer'.
+    assert credits == [
+        {"login": "author", "type": "remediation_developer"},
+        {"login": "reviewer1", "type": "coordinator"},
+        {"login": "reviewer2", "type": "remediation_reviewer"},
+    ]
+
+
+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"}]
+
+
+@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