From 81071812587a0b444606f8f22252a2043f2e098f Mon Sep 17 00:00:00 2001 From: Stan Ulbrych Date: Sun, 4 Oct 2026 20:16:55 +0100 Subject: [PATCH 1/2] Add comment commands for closing/accepting advisories --- README.md | 4 +- src/psrt_ghsa_bot/app.py | 60 ++++++++++++++++++-- tests/test_app.py | 119 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 177 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 49c73ff..1fe1cd2 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{"Summary has an accept tag or 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..38ac0d3 100644 --- a/src/psrt_ghsa_bot/app.py +++ b/src/psrt_ghsa_bot/app.py @@ -36,6 +36,8 @@ "INVALID", "DUPLICATE", ) +COMMAND_AUTHOR_ASSOCIATIONS = ("OWNER", "MEMBER") +COMMAND_RE = re.compile(r"^!(close|accept)\s*$", re.IGNORECASE | re.MULTILINE) def get_repository_advisories( @@ -96,6 +98,50 @@ 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. + """ + command = None + for comment in comments: + if comment.get("author_association") not in COMMAND_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 +268,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 +288,9 @@ 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 summary contains '[ACCEPT{ED}]' or the latest + # command is '!accept' we can move the ticket to draft. + if state == "triage" and (command == "accept" or re.search(r"\[ACCEPT(?:ED)?\]", summary.upper()) is not None): 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..2407ee6 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", []) @@ -252,6 +263,114 @@ def test_accepts_advisory_with_accept_tag(summary, cve_id, cve_reserve_response) ) +@pytest.mark.parametrize( + "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", + "!Accept", + "Looks valid to me.\n\n!accept", + "!close\n!accept", + ], +) +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, + 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) + + github.rest.security_advisories.update_repository_advisory.assert_called_once_with( + owner="owner", + repo="repo", + ghsa_id="GHSA-xxxx-xxxx-xxxx", + data={"state": "draft", "cve_id": cve_id}, + ) + + +@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( From dbcdeca605d10268bd0caf87e8d9a9b3de05d05a Mon Sep 17 00:00:00 2001 From: Stan Ulbrych Date: Wed, 14 Oct 2026 22:20:21 +0100 Subject: [PATCH 2/2] Apply Seth's suggestions --- README.md | 2 +- src/psrt_ghsa_bot/app.py | 16 +++++++++------- tests/test_app.py | 29 ----------------------------- 3 files changed, 10 insertions(+), 37 deletions(-) diff --git a/README.md b/README.md index 1fe1cd2..06c7fa3 100644 --- a/README.md +++ b/README.md @@ -28,7 +28,7 @@ flowchart TD 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 or latest command is !accept?"}:::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 38ac0d3..c5a19ef 100644 --- a/src/psrt_ghsa_bot/app.py +++ b/src/psrt_ghsa_bot/app.py @@ -36,8 +36,6 @@ "INVALID", "DUPLICATE", ) -COMMAND_AUTHOR_ASSOCIATIONS = ("OWNER", "MEMBER") -COMMAND_RE = re.compile(r"^!(close|accept)\s*$", re.IGNORECASE | re.MULTILINE) def get_repository_advisories( @@ -132,12 +130,17 @@ def get_advisory_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 COMMAND_AUTHOR_ASSOCIATIONS: + if comment.get("author_association") not in author_associations: continue - for name in COMMAND_RE.findall(comment.get("body") or ""): + for name in command_re.findall(comment.get("body") or ""): command = name.lower() return command @@ -288,9 +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}]' or the latest - # command is '!accept' we can move the ticket to draft. - if state == "triage" and (command == "accept" or 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 2407ee6..f10d2c6 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -234,35 +234,6 @@ def test_closes_advisory_with_completion_tag(summary) -> None: ) -@pytest.mark.parametrize( - "summary", - [ - "[ACCEPT] Python is better than C", - "[ACCEPTED] 0.1 + 0.2 is broken?!?!?!?!?!", - "fix soemthing in datetime module [ACCEPTED]", - "blah blah [accepted] lowercase blah", - ], -) -def test_accepts_advisory_with_accept_tag(summary, cve_id, cve_reserve_response) -> None: - security_advisory = _create_advisory_dict("triage", None, ["psrt"], summary=summary) - - 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: - get_repo_advs.return_value = [security_advisory] - - app.apply_to_repo(github, "owner", "repo", cve_api) - - github.rest.security_advisories.update_repository_advisory.assert_called_once_with( - owner="owner", - repo="repo", - ghsa_id="GHSA-xxxx-xxxx-xxxx", - data={"state": "draft", "cve_id": cve_id}, - ) - - @pytest.mark.parametrize( "body", [