From 3d104fbff3562b710e5ede60cae9ed30bb8644ed Mon Sep 17 00:00:00 2001 From: Brett Smith Date: Mon, 2 Nov 2026 12:10:05 -0500 Subject: [PATCH 1/5] Close GHSAs when the associated CVE has a patch (#54) --- src/psrt_ghsa_bot/app.py | 37 ++++++++++++--- tests/test_app.py | 100 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 131 insertions(+), 6 deletions(-) diff --git a/src/psrt_ghsa_bot/app.py b/src/psrt_ghsa_bot/app.py index 01febc4..d56eef7 100644 --- a/src/psrt_ghsa_bot/app.py +++ b/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,24 @@ 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: abc.Set) -> 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_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 + for ref in cve_refs: + # Not sure why but abc.Set only defines isdisjoint. + if not tags.isdisjoint(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 +277,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: diff --git a/tests/test_app.py b/tests/test_app.py index b6460d2..e320669 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -1,11 +1,16 @@ import datetime import json +import operator +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 + # referenced from + # 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,32 @@ def test_only_processes_installations_for_required_org(monkeypatch) -> None: app.run() github.auth.as_installation.assert_called_once_with(2) + + +@pytest.mark.parametrize( + "tags", + [ + ["patch"], + ["issue-tracking", "patch"], + ["third-party-advisory", "patch", "exploit"], + [], + ["third-party-advisory"], + ["issue-tracking", "mailing-list"], + ], +) +def test_ghsa_closed_on_cve_patch(show_cve_response, tags, cve_id) -> None: + 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 + check_state = operator.eq if "patch" in tags else operator.ne + assert check_state(new_state, "closed") From cc5e93ad79e25a132e1091cbc7b2956506c01da9 Mon Sep 17 00:00:00 2001 From: Brett Smith Date: Mon, 2 Nov 2026 17:43:12 -0500 Subject: [PATCH 2/5] Add test that non-draft GHSAs are not updated --- tests/test_app.py | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/tests/test_app.py b/tests/test_app.py index e320669..e023889 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -525,3 +525,22 @@ def test_ghsa_closed_on_cve_patch(show_cve_response, tags, cve_id) -> None: new_state = None check_state = operator.eq if "patch" in tags else operator.ne assert check_state(new_state, "closed") + + +@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 From f3040c28c83f34e2b73d9600ac744ea9e2144537 Mon Sep 17 00:00:00 2001 From: Brett Smith Date: Mon, 2 Nov 2026 17:44:49 -0500 Subject: [PATCH 3/5] Simplify test per PR suggestion --- tests/test_app.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/tests/test_app.py b/tests/test_app.py index e023889..7031a16 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -1,6 +1,5 @@ import datetime import json -import operator from collections import abc from unittest import mock @@ -523,8 +522,7 @@ def test_ghsa_closed_on_cve_patch(show_cve_response, tags, cve_id) -> None: new_state = update.kwargs["data"]["state"] except AttributeError, KeyError: new_state = None - check_state = operator.eq if "patch" in tags else operator.ne - assert check_state(new_state, "closed") + assert (new_state == "closed") == ("patch" in tags) @pytest.mark.parametrize("state", ["closed", "published", "triage"]) From 20c89450abe6e8227c0ed2eadd4db4b6307ca934 Mon Sep 17 00:00:00 2001 From: Brett Smith Date: Tue, 3 Nov 2026 14:51:46 -0500 Subject: [PATCH 4/5] Simplify type annotation --- src/psrt_ghsa_bot/app.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/src/psrt_ghsa_bot/app.py b/src/psrt_ghsa_bot/app.py index d56eef7..cb27492 100644 --- a/src/psrt_ghsa_bot/app.py +++ b/src/psrt_ghsa_bot/app.py @@ -207,7 +207,7 @@ 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: abc.Set) -> abc.Iterator[str]: +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) @@ -220,8 +220,7 @@ def iter_cve_refs_with_tags(cve_api: CveApi, cve_id: str, tags: abc.Set) -> abc. capture_exception() return for ref in cve_refs: - # Not sure why but abc.Set only defines isdisjoint. - if not tags.isdisjoint(ref.get("tags", ())): + if tags.intersection(ref.get("tags", ())): yield ref["url"] From 15ebf3df098e084b1e3aef67a18925b17602350d Mon Sep 17 00:00:00 2001 From: Brett Smith Date: Tue, 3 Nov 2026 15:10:07 -0500 Subject: [PATCH 5/5] Limit CVE patch checks to published CVEs --- src/psrt_ghsa_bot/app.py | 3 +++ tests/test_app.py | 29 ++++++++++++++++++----------- 2 files changed, 21 insertions(+), 11 deletions(-) diff --git a/src/psrt_ghsa_bot/app.py b/src/psrt_ghsa_bot/app.py index cb27492..9709302 100644 --- a/src/psrt_ghsa_bot/app.py +++ b/src/psrt_ghsa_bot/app.py @@ -211,6 +211,7 @@ def iter_cve_refs_with_tags(cve_api: CveApi, cve_id: str, tags: set[str]) -> abc """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. @@ -219,6 +220,8 @@ def iter_cve_refs_with_tags(cve_api: CveApi, cve_id: str, tags: set[str]) -> abc # 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"] diff --git a/tests/test_app.py b/tests/test_app.py index 7031a16..3e58a0d 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -1,4 +1,5 @@ import datetime +import itertools import json from collections import abc from unittest import mock @@ -498,17 +499,23 @@ def test_only_processes_installations_for_required_org(monkeypatch) -> None: @pytest.mark.parametrize( - "tags", - [ - ["patch"], - ["issue-tracking", "patch"], - ["third-party-advisory", "patch", "exploit"], - [], - ["third-party-advisory"], - ["issue-tracking", "mailing-list"], - ], + "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_on_cve_patch(show_cve_response, tags, cve_id) -> None: +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() @@ -522,7 +529,7 @@ def test_ghsa_closed_on_cve_patch(show_cve_response, tags, cve_id) -> None: new_state = update.kwargs["data"]["state"] except AttributeError, KeyError: new_state = None - assert (new_state == "closed") == ("patch" in tags) + assert (new_state == "closed") == (state == "PUBLISHED" and "patch" in tags) @pytest.mark.parametrize("state", ["closed", "published", "triage"])