mirror of
https://github.com/rennf93/roboco.git
synced 2026-08-03 07:23:24 +02:00
feat(orchestrator): author allowlist for inbound external-PR review
At ingest, a non-empty external_pr_author_allowlist restricts which external PRs are reviewed to those GitHub logins (case-insensitive). An empty allowlist (default) reviews every external PR — safe because the review is read-only; the confirmed_by_human gate still guards any later supersede that runs fork code. Unit-tested (_pr_author_allowed).
This commit is contained in:
@@ -4569,12 +4569,15 @@ Start by:
|
||||
task_service = get_task_service(db)
|
||||
projects = await get_project_service(db).list_all(active_only=True)
|
||||
system_id = _foundation.AGENTS["system"].uuid
|
||||
allowlist = {a.lower() for a in settings.external_pr_author_allowlist}
|
||||
ingested = 0
|
||||
for project in projects:
|
||||
for pr in await git.list_open_prs(project.slug):
|
||||
number = pr.get("number")
|
||||
if number is None or not self._is_external_pr(pr):
|
||||
continue
|
||||
if not self._pr_author_allowed(pr, allowlist):
|
||||
continue
|
||||
created = await task_service.ingest_external_pr(
|
||||
project_id=cast("UUID", project.id),
|
||||
pr=pr,
|
||||
@@ -4586,6 +4589,18 @@ Start by:
|
||||
await db.commit()
|
||||
return ingested
|
||||
|
||||
@staticmethod
|
||||
def _pr_author_allowed(pr: dict[str, Any], allowlist: set[str]) -> bool:
|
||||
"""With a non-empty allowlist, only those GitHub authors are reviewed.
|
||||
|
||||
An empty allowlist (the default) reviews every external PR — the review
|
||||
is read-only, so it is safe; the ``confirmed_by_human`` gate still
|
||||
protects any later supersede that would run the contributor's code.
|
||||
"""
|
||||
if not allowlist:
|
||||
return True
|
||||
return (pr.get("user_login") or "").lower() in allowlist
|
||||
|
||||
@staticmethod
|
||||
def _is_external_pr(pr: dict[str, Any]) -> bool:
|
||||
"""A PR the org did not author: a fork head or a non-member author."""
|
||||
|
||||
@@ -32,3 +32,22 @@ from roboco.runtime.orchestrator import AgentOrchestrator
|
||||
)
|
||||
def test_is_external_pr(pr: dict[str, object], *, expected: bool) -> None:
|
||||
assert AgentOrchestrator._is_external_pr(pr) is expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("pr", "allowlist", "expected"),
|
||||
[
|
||||
# Empty allowlist -> every external PR is reviewed (read-only, safe).
|
||||
({"user_login": "corey"}, set(), True),
|
||||
# Non-empty allowlist gates by GitHub login (case-insensitive).
|
||||
({"user_login": "corey"}, {"corey"}, True),
|
||||
({"user_login": "Corey"}, {"corey"}, True),
|
||||
({"user_login": "mallory"}, {"corey"}, False),
|
||||
({"user_login": None}, {"corey"}, False),
|
||||
({}, {"corey"}, False),
|
||||
],
|
||||
)
|
||||
def test_pr_author_allowed(
|
||||
pr: dict[str, object], allowlist: set[str], *, expected: bool
|
||||
) -> None:
|
||||
assert AgentOrchestrator._pr_author_allowed(pr, allowlist) is expected
|
||||
|
||||
Reference in New Issue
Block a user