diff --git a/README.md b/README.md index 49c73ff..06c7fa3 100644 --- a/README.md +++ b/README.md @@ -26,9 +26,9 @@ flowchart TD Collab -- yes --> AddCollab[Add PSRT as collaborators]:::write --> Tag Collab -- no --> Tag - Tag{"Summary has a completion tag?"}:::decision + Tag{"Summary has a completion tag or latest command is !close?"}:::decision Tag -- yes --> Close[Close advisory]:::write --> Done([Continue to next advisory]):::terminal - Tag -- "No, Triage" --> Accept{"Summary has an accept tag?"}:::decision + Tag -- "No, Triage" --> Accept{"Latest command is !accept?"}:::decision Tag -- "No, Draft" --> Fork Accept -- yes --> ToDraft[Move to the draft state]:::write --> Fork diff --git a/src/psrt_ghsa_bot/app.py b/src/psrt_ghsa_bot/app.py index 01febc4..c5a19ef 100644 --- a/src/psrt_ghsa_bot/app.py +++ b/src/psrt_ghsa_bot/app.py @@ -96,6 +96,55 @@ def get_repository_advisories( raise RuntimeError("Request to paginate advisories failed.") +def get_advisory_comments( + github: GitHub, + owner: str, + repo: str, + ghsa_id: str, +) -> list[dict[str, typing.Any]]: + """Lists comments on a repository security advisory using the REST API.""" + comments = [] + page = 1 + try: + while True: + response = github_client_request( + client=github.rest.security_advisories, + method="GET", + url=f"/repos/{owner}/{repo}/security-advisories/{ghsa_id}/comments", + params={"per_page": 100, "page": page}, + ) + page_comments = json.loads(response.content) + comments.extend(page_comments) + if len(page_comments) < 100: + break + page += 1 + except RequestFailed as e: + if e.response.status_code == 404: + print(f" ⚠️ Comments not found for {ghsa_id}!") + return [] + capture_exception() + raise RuntimeError("Request to list advisory comments failed") from None + return comments + + +def get_advisory_command(comments: typing.Iterable[dict[str, typing.Any]]) -> str | None: + """Finds the most recent command issued by an organization member in the + comments of a security advisory. + + There is no tracking of which commands have already been executed, so + every command must be idempotent. + """ + author_associations = ("OWNER", "MEMBER") + command_re = re.compile(r"^!(close|accept)\s*$", re.IGNORECASE | re.MULTILINE) + command = None + for comment in comments: + if comment.get("author_association") not in author_associations: + continue + for name in command_re.findall(comment.get("body") or ""): + command = name.lower() + return command + + def get_security_advisory_credits( github: GitHub, security_advisory: dict[str, typing.Any], @@ -222,9 +271,14 @@ def apply_to_repo(github: GitHub, owner: str, repo: str, cve_api: CveApi, *, res print(f" 📋 Processing {ghsa_id} (state: {state})") - # If the summary contains a completion tag then we can close the ticket. + command = None + if security_advisory.get("comments"): + command = get_advisory_command(get_advisory_comments(github, owner, repo, ghsa_id)) + + # If the summary contains a completion tag or the latest + # command is '!close' then we can close the ticket. summary = security_advisory.get("summary", "") - if re.search(rf"\[(?:{'|'.join(COMPLETION_TAGS)})\]", summary.upper()) is not None: + if command == "close" or re.search(rf"\[(?:{'|'.join(COMPLETION_TAGS)})\]", summary.upper()) is not None: github.rest.security_advisories.update_repository_advisory( owner=owner, repo=repo, @@ -237,8 +291,8 @@ def apply_to_repo(github: GitHub, owner: str, repo: str, cve_api: CveApi, *, res # Maintain a dictionary of updates to make and then submit them all at once. patch_data = {} - # If the summary contains '[ACCEPT{ED}]' we can move the ticket to draft - if state == "triage" and re.search(r"\[ACCEPT(?:ED)?\]", summary.upper()) is not None: + # If the latest command is '!accept' we can move the ticket to draft. + if state == "triage" and command == "accept": patch_data["state"] = state = "draft" print(f" ✅ Will accept {ghsa_id}") diff --git a/tests/test_app.py b/tests/test_app.py index b6460d2..f10d2c6 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -3,6 +3,7 @@ from unittest import mock import pytest +from githubkit.exception import RequestFailed from psrt_ghsa_bot import app @@ -47,6 +48,16 @@ def _create_advisory_dict(state, cve_id, collaborating_teams, summary=""): } +def _create_comment_dict(body, author_association="MEMBER"): + """Helper to create a GHSA comment dictionary.""" + return { + "id": 1, + "body": body, + "user": {"login": "octocat"}, + "author_association": author_association, + } + + @pytest.mark.parametrize("state", ["draft", "triage"]) def test_adds_psrt_github_team_to_security_advisories(state) -> None: security_advisory = _create_advisory_dict(state, "CVE-0000-0000", []) @@ -224,23 +235,63 @@ def test_closes_advisory_with_completion_tag(summary) -> None: @pytest.mark.parametrize( - "summary", + "body", + [ + "!close", + "!CLOSE", + "!close ", + "This is a duplicate.\r\n\r\n!close\r\n", + "!accept\n!close", + ], +) +def test_closes_advisory_with_close_command(body) -> None: + security_advisory = _create_advisory_dict("triage", None, []) + security_advisory["comments"] = 1 + + github = mock.Mock() + cve_api = mock.Mock() + + with ( + mock.patch("psrt_ghsa_bot.app.get_repository_advisories") as get_repo_advs, + mock.patch("psrt_ghsa_bot.app.get_advisory_comments") as get_adv_comments, + ): + get_repo_advs.return_value = [security_advisory] + get_adv_comments.return_value = [_create_comment_dict(body)] + + app.apply_to_repo(github, "owner", "repo", cve_api) + + get_adv_comments.assert_called_once_with(github, "owner", "repo", "GHSA-xxxx-xxxx-xxxx") + github.rest.security_advisories.update_repository_advisory.assert_called_once_with( + owner="owner", + repo="repo", + ghsa_id="GHSA-xxxx-xxxx-xxxx", + data={"state": "closed"}, + ) + + +@pytest.mark.parametrize( + "body", [ - "[ACCEPT] Python is better than C", - "[ACCEPTED] 0.1 + 0.2 is broken?!?!?!?!?!", - "fix soemthing in datetime module [ACCEPTED]", - "blah blah [accepted] lowercase blah", + "!accept", + "!Accept", + "Looks valid to me.\n\n!accept", + "!close\n!accept", ], ) -def test_accepts_advisory_with_accept_tag(summary, cve_id, cve_reserve_response) -> None: - security_advisory = _create_advisory_dict("triage", None, ["psrt"], summary=summary) +def test_accepts_advisory_with_accept_command(body, cve_id, cve_reserve_response) -> None: + security_advisory = _create_advisory_dict("triage", None, ["psrt"]) + security_advisory["comments"] = 1 github = mock.Mock() cve_api = mock.Mock() cve_api.reserve.return_value = cve_reserve_response - with mock.patch("psrt_ghsa_bot.app.get_repository_advisories") as get_repo_advs: + with ( + mock.patch("psrt_ghsa_bot.app.get_repository_advisories") as get_repo_advs, + mock.patch("psrt_ghsa_bot.app.get_advisory_comments") as get_adv_comments, + ): get_repo_advs.return_value = [security_advisory] + get_adv_comments.return_value = [_create_comment_dict(body)] app.apply_to_repo(github, "owner", "repo", cve_api) @@ -252,6 +303,45 @@ def test_accepts_advisory_with_accept_tag(summary, cve_id, cve_reserve_response) ) +@pytest.mark.parametrize( + ("comments", "command"), + [ + ([], None), + ([_create_comment_dict("Thanks for the report!")], None), + ([_create_comment_dict("!close")], "close"), + ([_create_comment_dict("!accept")], "accept"), + ([_create_comment_dict("!accept"), _create_comment_dict("!close")], "close"), + ([_create_comment_dict("!close"), _create_comment_dict("!accept")], "accept"), + ([_create_comment_dict("!close"), _create_comment_dict("unrelated")], "close"), + ([_create_comment_dict("please !close this")], None), + ([_create_comment_dict("> !close")], None), + ([_create_comment_dict("!closed")], None), + ([_create_comment_dict(None)], None), + ([_create_comment_dict("!close", author_association="NONE")], None), + ([_create_comment_dict("!accept", author_association="CONTRIBUTOR")], None), + ([_create_comment_dict("!accept", author_association="COLLABORATOR")], None), + ([_create_comment_dict("!accept", author_association="OWNER")], "accept"), + ( + [_create_comment_dict("!close"), _create_comment_dict("!accept", author_association="NONE")], + "close", + ), + ], +) +def test_get_advisory_command(comments, command) -> None: + assert app.get_advisory_command(comments) == command + + +def test_get_advisory_comments_not_found() -> None: + github = mock.Mock() + + with mock.patch("psrt_ghsa_bot.app.github_client_request") as client_request: + client_request.side_effect = RequestFailed(mock.Mock(status_code=404)) + + comments = app.get_advisory_comments(github, "owner", "repo", "GHSA-xxxx-xxxx-xxxx") + + assert comments == [] + + def test_get_security_advisory_credits_no_private_fork(): github = mock.Mock() credits = app.get_security_advisory_credits(