diff --git a/src/psrt_ghsa_bot/app.py b/src/psrt_ghsa_bot/app.py index 971a5d6..c392db0 100644 --- a/src/psrt_ghsa_bot/app.py +++ b/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: diff --git a/tests/test_app.py b/tests/test_app.py index e130477..390ce6a 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,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