Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 24 additions & 20 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,12 @@
PSRT GHSA Bot is a GitHub App that automates the [Python Security Response Team
(PSRT)](https://devguide.python.org/security/psrt/)'s
handling of GitHub Security Advisories. It runs hourly (or by manual dispatch)
and, for every advisory it closes ones marked as completed, promotes accepted ones
from triage to draft, reserves CVE IDs, creates private forks, and adds the
PSRT team as collaborators.
and does housekeeping on every advisory.

* For newly accepted advisories, it reserves a CVE ID, creates a private
fork, and adds the PSRT team as collaborators.
* For accepted advisories, if the associated CVE has been published with
a patch, it closes the advisory.

It only processes installations owned by the account named in the `GH_REQUIRED_ORG`
environment variable; installations under any other user are skipped.
Expand All @@ -20,29 +23,30 @@ flowchart TD

S -- Closed --> Skip([Skip advisory]):::terminal
S -- Triage --> Collab
S -- Draft --> Collab
S -- Draft --> Fork

Collab{"PSRT missing as collaborators?"}:::decision
Collab -- yes --> AddCollab[Add PSRT as collaborators]:::write --> Tag
Collab -- no --> Tag
Fork{"No private fork?"}:::decision
Fork -- yes --> MkFork[Create private fork]:::write --> CveId
Fork -- no --> CveId

Tag{"Summary has a completion tag?"}:::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, Draft" --> Fork
CveId{"CVE ID assigned?"}:::decision
CveId -- yes --> CveDone
CveId -- no --> Reserve[Reserve a CVE ID]:::write --> Collab

Accept -- yes --> ToDraft[Move to the draft state]:::write --> Fork
Accept -- no --> Update
CveDone{"CVE published with patch?"}:::decision
CveDone -- yes --> Close[Close advisory]:::write --> Collab
CveDone -- no --> Collab

Fork{"No private fork?"}:::decision
Fork -- yes --> MkFork[Create private fork]:::write --> Cve
Fork -- no --> Cve
Collab{"PSRT missing as collaborators?"}:::decision
Collab -- yes --> AddCollab[Add PSRT as collaborators]:::write --> Credits
Collab -- no --> Credits

Cve{"No CVE ID assigned?"}:::decision
Cve -- yes --> Reserve[Reserve a CVE ID]:::write --> Update
Cve -- no --> Update
Credits[For each user associated with a pull request]:::loop --> CheckCredit
CheckCredit{"User credited on advisory?"}:::decision
CheckCredit -- yes --> Update
CheckCredit -- no --> AddCredit[Add user to advisory credits]:::write --> Update

Update[Update advisory]:::write --> Done
Update[Update advisory]:::terminal

classDef entry stroke:#0C0,stroke-width:2px;
classDef loop stroke:#00C;
Expand Down
27 changes: 0 additions & 27 deletions src/psrt_ghsa_bot/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,15 +29,6 @@
load_dotenv()

PSRT_GITHUB_TEAM_SLUG = "psrt"
COMPLETION_TAGS = (
"CLOSE",
"CLOSED",
"COMPLETE",
"COMPLETED",
"NOTPLANNED",
"INVALID",
"DUPLICATE",
)


def get_repository_advisories(
Expand Down Expand Up @@ -243,27 +234,9 @@ def apply_to_repo(github: GitHub, owner: str, repo: str, cve_api: CveApi, *, res
continue

print(f" 📋 Processing {ghsa_id} (state: {state})")

# If the summary contains a completion tag then we can close the ticket.
summary = security_advisory.get("summary", "")
if re.search(rf"\[(?:{'|'.join(COMPLETION_TAGS)})\]", summary.upper()) is not None:
github.rest.security_advisories.update_repository_advisory(
owner=owner,
repo=repo,
ghsa_id=ghsa_id,
data={"state": "closed"},
)
print(f" 🧹 Closed {ghsa_id}")
continue

# 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:
patch_data["state"] = state = "draft"
print(f" ✅ Will accept {ghsa_id}")

# Advisories that are in the 'draft' state without a private
# fork active will have a fork requested.
if state == "draft" and security_advisory.get("private_fork") is None:
Expand Down
61 changes: 0 additions & 61 deletions tests/test_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -263,67 +263,6 @@ def test_create_private_fork() -> None:
)


@pytest.mark.parametrize(
"summary",
[
"[CLOSE] perl is better than Python",
"[CLOSED] 0.1 + 0.2 is broken?!?!?!?!?!",
"[COMPLETE] some boring security thing",
"fix soemthing in datetime module [COMPLETED]",
"blah blah [closed] lowercase blah",
"[NOTPLANNED] no fix planned",
"[INVALID] some annoying spam",
"[DUPLICATE] we've seen this one before",
],
)
def test_closes_advisory_with_completion_tag(summary) -> None:
security_advisory = _create_advisory_dict("triage", None, [], summary=summary)

github = mock.Mock()
cve_api = mock.Mock()

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": "closed"},
)


@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},
)


def test_get_security_advisory_credits_no_private_fork():
github = mock.Mock()
credits = app.get_security_advisory_credits(
Expand Down
Loading