Close GHSAs when the associated CVE has a patch (#54)

src/psrt_ghsa_bot/app.py

@@ -8,6 +8,7 @@
 import time
 import typing
 import urllib.parse
+from collections import abc
 
 from cvelib.cve_api import CveApi
 from dotenv import load_dotenv
@@ -206,6 +207,26 @@ def sort_security_advisory_credits(credits: list[dict[str, str]]) -> list[dict[s
     return sorted(credits, key=lambda c: (c["login"], c["type"]))
 
 
+def iter_cve_refs_with_tags(cve_api: CveApi, cve_id: str, tags: set[str]) -> abc.Iterator[str]:
+    """Yield reference URLs in the given CVE with any of the named tags"""
+    try:
+        cve = cve_api.show_cve_record(cve_id)
+        cve_state = cve["cveMetadata"]["state"]
+        cve_refs = cve["containers"]["cna"]["references"]
+    except KeyError, RequestFailed, TypeError:
+        # If we got a bad response, we can still proceed with other updates.
+        # Record the failure and continue.
+        # Note TypeError covers tests where cve_api is a mock without a
+        # return value for show_cve_record.
+        capture_exception()
+        return
+    if cve_state != "PUBLISHED":
+        return  # We don't want to act on CVEs that aren't published.
+    for ref in cve_refs:
+        if tags.intersection(ref.get("tags", ())):
+            yield ref["url"]
+
+
 def apply_to_repo(github: GitHub, owner: str, repo: str, cve_api: CveApi, *, reserve_cves: bool = True) -> None:
     """Applies the PSRT GitHub Security Advisory process to the repository."""
     security_advisories = get_repository_advisories(github, owner, repo)
@@ -258,12 +279,18 @@ def apply_to_repo(github: GitHub, owner: str, repo: str, cve_api: CveApi, *, res
                 capture_exception()
                 raise RuntimeError("Request to create a private fork failed") from None
 
-        # Advisories that are in the 'draft' state without a CVE ID
-        # should have one allocated by the PSF CVE Numbering Authority.
-        if reserve_cves and state == "draft" and security_advisory.get("cve_id") is None:
-            cve_id = reserve_one_cve(cve_api)
-            patch_data["cve_id"] = cve_id
-            print(f"       ✅ Will reserve CVE ID: {cve_id}")
+        # Keep draft GHSAs in sync with CVEs:
+        # * If there is no CVE, reserve and associate one.
+        # * If there is a CVE, and it has a patch, close the GHSA.
+        if state == "draft":
+            cve_id = security_advisory.get("cve_id")
+            if cve_id is None and reserve_cves:
+                cve_id = reserve_one_cve(cve_api)
+                patch_data["cve_id"] = cve_id
+                print(f"       ✅ Will reserve CVE ID: {cve_id}")
+            elif cve_id and any(iter_cve_refs_with_tags(cve_api, cve_id, {"patch"})):
+                print(f"       🎉 Will close {ghsa_id}: {cve_id} has a patch")
+                patch_data["state"] = "closed"
 
         collaborating_teams = {team["slug"] for team in security_advisory["collaborating_teams"]}
         if PSRT_GITHUB_TEAM_SLUG not in collaborating_teams:

tests/test_app.py

@@ -1,11 +1,16 @@
 import datetime
+import itertools
 import json
+from collections import abc
 from unittest import mock
 
 import pytest
 
 from psrt_ghsa_bot import app
 
+PSF_CNA_UUID = "28c92f92-d60d-412d-b760-e73465c3df22"
+PSF_CNA_SHORTNAME = "PSF"
+
 
 @pytest.fixture
 def year() -> str:
@@ -35,6 +40,72 @@ def cve_reserve_response(cve_id, year):
     }
 
 
+def cve_isoformat(date_time: datetime.datetime) -> str:
+    naive_utc = date_time.astimezone(datetime.UTC).replace(tzinfo=None)
+    return f"{naive_utc.isoformat(timespec='milliseconds')}Z"
+
+
+@pytest.fixture
+def show_cve_response(cve_id):
+    # See <https://github.com/CVEProject/cve-services/blob/dev/schemas/CVE_JSON_5.2.0_bundled.json>
+    # referenced from <https://github.com/CVEProject/cve-services/blob/dev/schemas/cve/get-cve-record-response.json>
+    # This is very incomplete for now but can be expanded as we introspect more
+    # fields.
+    date_updated = datetime.datetime.now(tz=datetime.UTC)
+    return {
+        "dataType": "CVE_RECORD",
+        "dataVersion": "5.2",
+        "cveMetadata": {
+            "cveId": cve_id,
+            "assignerOrgId": PSF_CNA_UUID,
+            "state": "PUBLISHED",
+            "assignerShortName": PSF_CNA_SHORTNAME,
+            "dateReserved": cve_isoformat(date_updated - datetime.timedelta(hours=60)),
+            "datePublished": cve_isoformat(date_updated - datetime.timedelta(hours=30)),
+            "dateUpdated": cve_isoformat(date_updated),
+        },
+        "containers": {
+            "cna": {
+                "credits": [
+                    {
+                        "lang": "en",
+                        "type": "reporter",
+                        "value": "User (https://example.com/~user)",
+                    },
+                    {
+                        "lang": "en",
+                        "type": "coordinator",
+                        "value": "Seth Larson (https://github.com/sethmlarson)",
+                    },
+                ],
+                "providerMetadata": {
+                    "orgId": PSF_CNA_UUID,
+                    "shortName": PSF_CNA_SHORTNAME,
+                    "dateUpdated": cve_isoformat(date_updated),
+                },
+                "references": [
+                    {
+                        "tags": ["issue-tracking"],
+                        "url": "https://github.com/python/cpython/issues/12345",
+                    }
+                ],
+                "source": {
+                    "discovery": "UNKNOWN",
+                },
+                "title": "Test Vulnerability",
+            },
+        },
+    }
+
+
+def _build_cve_references(tags: abc.Iterable[str]) -> abc.Iterator[dict]:
+    for index, tag in enumerate(tags, 1):
+        yield {
+            "tags": [tag],
+            "url": f"https://example.com/python/{tag}/{index:02d}",
+        }
+
+
 def _create_advisory_dict(state, cve_id, collaborating_teams, summary=""):
     """Helper to create a security advisory dictionary."""
     return {
@@ -425,3 +496,56 @@ def test_only_processes_installations_for_required_org(monkeypatch) -> None:
         app.run()
 
     github.auth.as_installation.assert_called_once_with(2)
+
+
+@pytest.mark.parametrize(
+    "state,tags",
+    list(
+        itertools.product(
+            ["PUBLISHED", "REJECTED"],
+            [
+                ["patch"],
+                ["issue-tracking", "patch"],
+                ["third-party-advisory", "patch", "exploit"],
+                [],
+                ["third-party-advisory"],
+                ["issue-tracking", "mailing-list"],
+            ],
+        )
+    ),
+)
+def test_ghsa_closed_when_cve_patch_published(show_cve_response, state, tags, cve_id) -> None:
+    show_cve_response["cveMetadata"]["state"] = state
+    show_cve_response["containers"]["cna"]["references"] = list(_build_cve_references(tags))
+    github = mock.Mock()
+    cve_api = mock.Mock()
+    cve_api.show_cve_record.return_value = show_cve_response
+    with mock.patch("psrt_ghsa_bot.app.get_repository_advisories") as get_repo_advs:
+        get_repo_advs.return_value = [_create_advisory_dict("draft", cve_id, [])]
+        app.apply_to_repo(github, "owner", "repo", cve_api)
+
+    update = github.rest.security_advisories.update_repository_advisory.call_args
+    try:
+        new_state = update.kwargs["data"]["state"]
+    except AttributeError, KeyError:
+        new_state = None
+    assert (new_state == "closed") == (state == "PUBLISHED" and "patch" in tags)
+
+
+@pytest.mark.parametrize("state", ["closed", "published", "triage"])
+def test_only_draft_ghsa_closed_on_cve(show_cve_response, state) -> None:
+    show_cve_response["containers"]["cna"]["references"] = list(_build_cve_references(["patch"]))
+    github = mock.Mock()
+    cve_api = mock.Mock()
+    cve_api.show_cve_record.return_value = show_cve_response
+    with mock.patch("psrt_ghsa_bot.app.get_repository_advisories") as get_repo_advs:
+        get_repo_advs.return_value = [_create_advisory_dict(state, cve_id, [])]
+        app.apply_to_repo(github, "owner", "repo", cve_api)
+
+    update = github.rest.security_advisories.update_repository_advisory.call_args
+    try:
+        new_state = update.kwargs["data"]["state"]
+    except AttributeError, KeyError:
+        pass  # We're testing the state didn't change, so no update is good.
+    else:
+        assert new_state == state