Automatically credit remediation developers and reviewers

src/psrt_ghsa_bot/app.py

@@ -96,6 +96,77 @@ 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.
+        nonlocal 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:
+            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
+            pull_request_author = pull_request["user"]["login"]
+            credit_if_uncredited(
+                login=pull_request_author,
+                type="remediation_developer"
+            )
+            # fmt: on
+            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 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:
     """Sends a raw HTTP request using a GitHub API client"""
     headers = {"X-GitHub-Api-Version": client._REST_API_VERSION}
@@ -184,6 +255,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:

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,97 @@ 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 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"}]
+
+
 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