diff --git a/src/psrt_ghsa_bot/app.py b/src/psrt_ghsa_bot/app.py index 01febc4..9709302 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,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: diff --git a/tests/test_app.py b/tests/test_app.py index b6460d2..3e58a0d 100644 --- a/tests/test_app.py +++ b/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 + # 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,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