Add comment commands for closing/accepting advisories

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

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}")
 

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(