feat(python): implemented pull request vouching and gated review system

- Introduced a vouching mechanism to manage PR authorization via a tracked user list and discussion-based management workflows.
- Added automated PR gatekeeping workflows to close contributions from unvouched users and require specific labels for review.
- Refactored PR event handling to support label-based review deferral and enforce authorization checks for labelers.
- Added comprehensive test coverage for vouch-gate logic, including label activation and unauthorized access scenarios.
This commit is contained in:
can1357
2026-06-19 03:14:43 +02:00
parent 71144825ec
commit aca5d5f48a
9 changed files with 491 additions and 18 deletions
+12
View File
@@ -38,6 +38,18 @@ class Settings(BaseSettings):
git_author_email: str = Field(..., alias="ROBOMP_GIT_AUTHOR_EMAIL")
repo_allowlist_raw: str = Field("", alias="ROBOMP_REPO_ALLOWLIST")
pr_review_enabled: bool = Field(True, alias="ROBOMP_PR_REVIEW_ENABLED")
# PR review trigger. "open" (default) reviews incoming PRs on
# opened/reopened/ready_for_review. "vouched_label" DEFERS review until the
# vouch GitHub Action labels the PR `vouch_review_label`, so robomp reviews
# only PRs that survive the vouch gate. `pr_review_enabled` remains the
# master switch (False disables review under either trigger).
pr_review_trigger: Literal["open", "vouched_label"] = Field("open", alias="ROBOMP_PR_REVIEW_TRIGGER")
vouch_review_label: str = Field("vouched", alias="ROBOMP_VOUCH_REVIEW_LABEL")
# In vouched_label mode, only `labeled` events from this actor trigger a
# review, so a manual label by a triage/maintainer cannot bypass the gate.
# Default is the actor for the stock GITHUB_TOKEN; set to your App's bot
# login (e.g. "vouch-bot[bot]") if the vouch workflow labels via an App.
vouch_review_labeler: str = Field("github-actions[bot]", alias="ROBOMP_VOUCH_REVIEW_LABELER")
# gh-proxy. Set BOTH to route GitHub through the proxy; leave both empty
# to keep PAT-on-orchestrator behavior. Mixing the two (PAT + proxy) is
+41 -18
View File
@@ -140,6 +140,23 @@ def is_implementation_authorizer(
return False
def _pr_review_pr(pr: Mapping[str, Any], repo: str, action: str, bot_login: str) -> RouteDecision:
"""Build a `review_pr` decision for an incoming PR, or the matching skip."""
if str(pr.get("state") or "open") != "open":
return RouteDecision("skip", None, repo, None, "PR not open")
if bool(pr.get("draft")):
return RouteDecision("skip", None, repo, None, "draft PR")
if _is_bot_account(pr.get("user") or {}, bot_login):
return RouteDecision("skip", None, repo, None, "bot-authored PR")
number = pr.get("number")
if not isinstance(number, int):
return RouteDecision("skip", None, repo, None, "PR missing number")
login, assoc = _submitter_info(pr)
return RouteDecision(
"queue", "review_pr", repo, issue_key(repo, number), f"pull_request.{action}", submitter=login, association=assoc
)
def route(
event_type: str,
payload: Mapping[str, Any],
@@ -150,6 +167,9 @@ def route(
reviewer_bots: frozenset[str] = frozenset(),
resolve_issue_from_pr: PrIssueResolver = None,
pr_review_enabled: bool = True,
pr_review_trigger: str = "open",
vouch_review_label: str = "vouched",
vouch_review_labeler: str = "github-actions[bot]",
) -> RouteDecision:
"""Decide whether and how to handle a webhook event.
@@ -281,24 +301,27 @@ def route(
if not pr_review_enabled:
return RouteDecision("skip", None, repo, None, "PR review disabled")
pr = payload.get("pull_request") or {}
if bool(pr.get("draft")):
return RouteDecision("skip", None, repo, None, "draft PR")
pr_user = pr.get("user") or {}
if _is_bot_account(pr_user, bot_login):
return RouteDecision("skip", None, repo, None, "bot-authored PR")
number = pr.get("number")
if not isinstance(number, int):
return RouteDecision("skip", None, repo, None, "PR missing number")
login, assoc = _submitter_info(pr)
return RouteDecision(
"queue",
"review_pr",
repo,
issue_key(repo, number),
f"pull_request.{action}",
submitter=login,
association=assoc,
)
if pr_review_trigger == "vouched_label":
# Defer to the vouch gate. robomp reviews ONLY on the `labeled`
# event the workflow emits AFTER a fresh check (it re-applies the
# vouch label on every opened/reopened/ready_for_review). Never
# trust a persisted label here: a since-denounced author must not
# slip through on reopen.
return RouteDecision("skip", None, repo, None, "deferred to vouch label")
return _pr_review_pr(pr, repo, action, bot_login)
if event_type == "pull_request" and action == "labeled" and pr_review_trigger == "vouched_label":
if not pr_review_enabled:
return RouteDecision("skip", None, repo, None, "PR review disabled")
label = payload.get("label")
label_name = str(label.get("name") or "") if isinstance(label, Mapping) else ""
if label_name.lower() != vouch_review_label.lower():
return RouteDecision("skip", None, repo, None, f"label {label_name!r} not vouch label")
sender = payload.get("sender")
labeler = str(sender.get("login") or "") if isinstance(sender, Mapping) else ""
if labeler.lower() != vouch_review_labeler.lower():
return RouteDecision("skip", None, repo, None, f"vouch label not from trusted labeler ({labeler!r})")
return _pr_review_pr(payload.get("pull_request") or {}, repo, action, bot_login)
if event_type == "pull_request_review_comment" and action == "created":
comment = payload.get("comment") or {}
+3
View File
@@ -341,6 +341,9 @@ def create_app(settings: Settings | None = None) -> FastAPI:
maintainers=cfg.maintainer_logins,
reviewer_bots=cfg.reviewer_bots,
pr_review_enabled=cfg.pr_review_enabled,
pr_review_trigger=cfg.pr_review_trigger,
vouch_review_label=cfg.vouch_review_label,
vouch_review_labeler=cfg.vouch_review_labeler,
resolve_issue_from_pr=_resolve,
)