Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 73 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

Allow read-only PR queue inventory under author profiles while preserving hard review/merge gates

Browse Source
This commit is contained in:
sysadmin
2026-07-05 12:14:26 -04:00
parent 82d6bf286a
commit 485ba22c8f
5 changed files with 367 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mcp_server.py
+171 -2
View File
@@ -555,6 +555,41 @@ def gitea_check_pr_eligibility(
else:
reasons.append(f"profile is not allowed to {action}")
# Stop here immediately if the action implies reviewer role and the
# profile doesn't permit it. No identity/PR API calls on this path;
# the structured #142 report is built config-only (no network).
if action in ("review", "approve", "request_changes", "merge"):
result["eligible"] = False
missing_perm = (
f"gitea.pr.{action}"
if action in ("approve", "request_changes", "merge", "review")
else f"gitea.{action}")
result["missing_permission"] = missing_perm
result["active_profile"] = profile["profile_name"]
result["active_identity"] = None
result["permission_report"] = _permission_block_report(
missing_perm)
req_profile = None
if action in ("approve", "request_changes", "review"):
req_profile = "A profile with reviewer role permissions (allowing approve/merge/review, and forbidding author operations)"
elif action == "merge":
req_profile = "A profile with reviewer role permissions and explicit merge permission"
result["required_profile"] = req_profile
switching_supported = gitea_config.is_runtime_switching_enabled()
if switching_supported:
result["fixable_by_profile_switch"] = True
result["requires_different_namespace"] = False
safe_step = f"Switch to a reviewer profile by calling gitea_activate_profile with a profile that allows {action}."
else:
result["fixable_by_profile_switch"] = False
result["requires_different_namespace"] = True
safe_step = "Switch to the reviewer MCP session (e.g. gitea-reviewer) which has reviewer permissions configured, or ask the operator to update GITEA_MCP_PROFILE to a reviewer profile."
result["safe_next_step"] = safe_step
return result
h, o, r = _resolve(remote, host, org, repo)
# Authenticated identity (read-only). Fail soft; never leak error/secret.
@@ -1304,6 +1339,131 @@ def gitea_review_pr(
}
action = event_map[event]
# --- Phase 1: Read-only PR queue inventory ---
profile = get_profile()
allowed = profile.get("allowed_operations") or []
forbidden = profile.get("forbidden_operations") or []
switching_supported = gitea_config.is_runtime_switching_enabled()
can_review, _ = gitea_config.check_operation("gitea.pr.review", allowed, forbidden)
can_approve, _ = gitea_config.check_operation("gitea.pr.approve", allowed, forbidden)
can_merge, _ = gitea_config.check_operation("gitea.pr.merge", allowed, forbidden)
can_read, _ = gitea_config.check_operation("gitea.read", allowed, forbidden)
is_reviewer = can_review or can_approve or can_merge
inventory_attempted = False
prs_found_count = 0
pr_details_list = []
inventory_msg = ""
h, o, r = _resolve(remote, host, org, repo)
auth = None
try:
auth = _auth(h)
except Exception:
pass
auth_user = None
if auth:
try:
who = api_request("GET", gitea_url(h, "/api/v1/user"), auth)
auth_user = (who or {}).get("login")
except Exception:
pass
if can_read and auth:
inventory_attempted = True
try:
url = f"{repo_api_url(h, o, r)}/pulls?state=open"
prs = api_get_all(url, auth)
prs_found_count = len(prs)
for pr in prs:
labels = [l["name"] for l in pr.get("labels", [])]
body_text = pr.get("body") or ""
linked = []
# Simple extraction of linked issues from body
for match in re.findall(r'#(\d+)', body_text):
linked.append(f"#{match}")
pr_head_sha = (pr.get("head") or {}).get("sha")
pr_author = (pr.get("user") or {}).get("login")
# Determine review block status
req_non_author = False
if pr_author and auth_user and pr_author == auth_user:
req_non_author = True
pr_details_list.append({
"number": pr["number"],
"title": pr["title"],
"author": pr_author,
"base": pr["base"]["ref"],
"head": pr["head"]["ref"],
"mergeable": pr.get("mergeable"),
"linked_issues": list(set(linked)),
"labels": labels,
"head_sha": pr_head_sha,
"requires_non_author_reviewer": req_non_author
})
except Exception as e:
inventory_msg = f"PR inventory failed: {str(e)}"
else:
inventory_msg = "PR inventory not attempted"
# Build inventory report string
report_lines = []
if inventory_attempted:
if pr_details_list:
report_lines.append(f"Open PRs found: {prs_found_count}")
for item in pr_details_list:
status_str = "Review/merge blocked (Current profile is Author)"
if item["requires_non_author_reviewer"]:
status_str = "Review/merge blocked (Self-author). Requires a genuine non-author reviewer"
elif is_reviewer:
status_str = "Review/merge eligible under current profile"
report_lines.append(
f"- PR #{item['number']}: {item['title']}\n"
f" Author: {item['author']}\n"
f" Base/Head: {item['base']} / {item['head']}\n"
f" Head SHA: {item['head_sha']}\n"
f" Mergeability: {item['mergeable']}\n"
f" Linked Issues: {item['linked_issues']}\n"
f" Labels: {item['labels']}\n"
f" Status: {status_str}"
)
else:
if inventory_msg:
report_lines.append(inventory_msg)
else:
report_lines.append("Open PRs found: 0")
else:
report_lines.append(inventory_msg)
inventory_report = "\n".join(report_lines)
# If the current profile is not a reviewer, or lacks necessary permissions for review action
if not is_reviewer:
# Determine the safe next step based on static vs dynamic profile switching support
if switching_supported:
safe_step = f"Switch to a reviewer profile by calling gitea_activate_profile with a profile that allows {action}."
else:
safe_step = "Switch to the reviewer MCP session (e.g. gitea-reviewer) which has reviewer permissions configured, or ask the operator to update GITEA_MCP_PROFILE to a reviewer profile."
blocked_message = (
f"=== PR Queue Inventory ===\n{inventory_report}\n\n"
f"=== Review/Merge Blocked ===\n"
f"The active profile '{profile['profile_name']}' is an Author profile and does not have review or merge permissions.\n"
f"Safe next step: {safe_step}"
)
return {
"success": False,
"message": blocked_message
}
# --- Phase 2: Reviewer-only review/merge actions ---
# Since it is a reviewer profile, we proceed to submit the review
result = gitea_submit_pr_review(
pr_number=pr_number,
action=action,
@@ -1315,11 +1475,20 @@ def gitea_review_pr(
repo=repo
)
# Include the inventory report in the response message
if result.get("performed"):
return {"success": True, "message": f"Successfully submitted review for PR #{pr_number} with event '{event}'."}
msg = (
f"=== PR Queue Inventory ===\n{inventory_report}\n\n"
f"Successfully submitted review for PR #{pr_number} with event '{event}'."
)
return {"success": True, "message": msg}
else:
reasons = result.get("reasons", [])
out = {"success": False, "message": f"Review submission failed eligibility gates: {reasons}"}
msg = (
f"=== PR Queue Inventory ===\n{inventory_report}\n\n"
f"Review submission failed eligibility gates: {reasons}"
)
out = {"success": False, "message": msg}
if result.get("permission_report"):
out["permission_report"] = result["permission_report"]
return out
tests/test_llm_agent_sha.py
+8 -2
View File
@@ -118,10 +118,16 @@ class TestShaCannotBypassSelfReview(unittest.TestCase):
f"self-merge mutation reached the API: {method} {url}",
)
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_review_tool_refuses_self_approval_despite_sha(self, _auth, mock_api):
mock_api.side_effect = [{"login": "jcwalker3"}, _pr("jcwalker3")]
def test_review_tool_refuses_self_approval_despite_sha(self, _auth, mock_api, mock_get_all):
mock_get_all.return_value = [{"number": 9, "title": "PR 9", "state": "open", "head": {"ref": "branch9", "sha": "abc1234"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "jcwalker3"}}]
mock_api.side_effect = [
{"login": "jcwalker3"}, # /user (inventory)
{"login": "jcwalker3"}, # /user (eligibility)
{"state": "open", "head": {"sha": "abc1234"}, "mergeable": True, "user": {"login": "jcwalker3"}} # /pulls/9 (eligibility)
]
env = self._env(SHA_WOULD_BE_REVIEWER, "reviewer")
with patch.dict(os.environ, env, clear=True):
r = gitea_review_pr(
tests/test_mcp_server.py
+12 -11
View File
@@ -750,10 +750,11 @@ class TestReviewPR(unittest.TestCase):
self.assertIn("no longer supported", result["message"])
self.assertEqual(mock_api.call_count, 0)
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
def test_legacy_review_pr_uses_gates(self, mock_get_profile, _auth, mock_api):
def test_legacy_review_pr_uses_gates(self, mock_get_profile, _auth, mock_api, mock_get_all):
# Mock profile to lack approve capability (fails gate)
mock_get_profile.return_value = {
"profile_name": "gitea-readonly",
@@ -761,11 +762,8 @@ class TestReviewPR(unittest.TestCase):
"forbidden_operations": [],
"base_url": None,
}
# mock_api responses for auth_user and pr_author
mock_api.side_effect = [
{"login": "reviewer1"}, # /api/v1/user
{"state": "open", "head": {"sha": "abc1234"}, "mergeable": True, "user": {"login": "author1"}}, # /pulls/1
]
mock_get_all.return_value = [{"number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1", "sha": "abc1234"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "author1"}}]
mock_api.return_value = {"login": "reviewer1"}
result = gitea_review_pr(
pr_number=1,
event="APPROVE",
@@ -773,22 +771,25 @@ class TestReviewPR(unittest.TestCase):
merge=False
)
self.assertFalse(result["success"])
self.assertIn("Review submission failed eligibility gates", result["message"])
self.assertIn("not allowed to approve", result["message"])
self.assertIn("Review/Merge Blocked", result["message"])
self.assertIn("Author profile", result["message"])
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
def test_legacy_review_pr_self_approval_blocked(self, mock_get_profile, _auth, mock_api):
def test_legacy_review_pr_self_approval_blocked(self, mock_get_profile, _auth, mock_api, mock_get_all):
mock_get_profile.return_value = {
"profile_name": "gitea-reviewer",
"allowed_operations": ["read", "approve"],
"forbidden_operations": [],
"base_url": None,
}
# mock_api responses for auth_user and pr_author
mock_get_all.return_value = [{"number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1", "sha": "abc1234"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "jcwalker3"}}]
# mock_api responses: 1) /user (inventory), 2) /user (eligibility), 3) /pulls/1 (eligibility)
mock_api.side_effect = [
{"login": "jcwalker3"}, # /api/v1/user
{"login": "jcwalker3"}, # /api/v1/user (inventory)
{"login": "jcwalker3"}, # /api/v1/user (eligibility)
{"state": "open", "head": {"sha": "abc1234"}, "mergeable": True, "user": {"login": "jcwalker3"}}, # /pulls/1
]
result = gitea_review_pr(
tests/test_permission_reports.py
+3 -1
View File
@@ -184,7 +184,9 @@ class TestEligibilityDenialReport(PermissionReportBase):
self._assert_report_safe(report)
self.assertEqual(report["missing_permission"], "gitea.pr.approve")
self.assertEqual(report["active_profile"], "author-profile")
self.assertEqual(report["active_identity"], "author-user")
# "active identity if resolved": the #164 fail-fast path performs no
# identity lookup on a pure permission block, so None is acceptable.
self.assertIn(report["active_identity"], (None, "author-user"))
self.assertEqual(
report["matching_configured_profiles"], ["reviewer-profile"])
self.assertTrue(report["different_mcp_namespace_required"])
tests/test_pr_queue_inventory.py
+173
View File
@@ -0,0 +1,173 @@
import os
import sys
import unittest
import re
from unittest.mock import patch, MagicMock
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
from mcp_server import (
gitea_list_prs,
gitea_review_pr,
gitea_check_pr_eligibility,
)
import gitea_config
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
class TestPRQueueInventory(unittest.TestCase):
@patch("mcp_server.api_get_all")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
def test_author_profile_can_list_open_prs(self, mock_get_profile, _auth, mock_get_all):
mock_get_profile.return_value = {
"profile_name": "gitea-author",
"allowed_operations": ["read", "gitea.pr.comment"],
"forbidden_operations": [],
"base_url": None,
}
mock_get_all.return_value = [
{"number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1", "sha": "abc1"}, "base": {"ref": "master"}, "mergeable": True}
]
result = gitea_list_prs(state="open", remote="prgs")
self.assertEqual(len(result), 1)
self.assertEqual(result[0]["number"], 1)
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
@patch("gitea_config.is_runtime_switching_enabled", return_value=True)
def test_author_profile_can_produce_pending_reviewer_queue_report(self, mock_switch, mock_get_profile, _auth, mock_api, mock_get_all):
mock_get_profile.return_value = {
"profile_name": "gitea-author",
"allowed_operations": ["read"],
"forbidden_operations": [],
"base_url": None,
}
mock_get_all.return_value = [
{
"number": 1,
"title": "PR 1",
"state": "open",
"head": {"ref": "branch1", "sha": "abc1"},
"base": {"ref": "master"},
"mergeable": True,
"user": {"login": "jcwalker3"},
"labels": [{"name": "bug"}],
"body": "Fixes #10"
},
{
"number": 2,
"title": "PR 2",
"state": "open",
"head": {"ref": "branch2", "sha": "abc2"},
"base": {"ref": "master"},
"mergeable": True,
"user": {"login": "other_user"},
"labels": [],
"body": ""
}
]
mock_api.return_value = {"login": "jcwalker3"} # whoami
result = gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs")
self.assertFalse(result["success"])
self.assertIn("=== PR Queue Inventory ===", result["message"])
self.assertIn("Open PRs found: 2", result["message"])
self.assertIn("Review/merge blocked (Self-author). Requires a genuine non-author reviewer", result["message"])
self.assertIn("Review/merge blocked (Current profile is Author)", result["message"])
self.assertIn("gitea_activate_profile", result["message"])
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
def test_author_profile_never_calls_mutation_tools_during_inventory(self, mock_get_profile, _auth, mock_api, mock_get_all):
mock_get_profile.return_value = {
"profile_name": "gitea-author",
"allowed_operations": ["read"],
"forbidden_operations": [],
"base_url": None,
}
mock_get_all.return_value = []
mock_api.return_value = {"login": "jcwalker3"}
gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs")
for call in mock_api.call_args_list:
method = call.args[0]
self.assertEqual(method, "GET")
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
def test_reviewer_profile_can_proceed_from_inventory_to_review_gates(self, mock_get_profile, _auth, mock_api, mock_get_all):
mock_get_profile.return_value = {
"profile_name": "gitea-reviewer",
"allowed_operations": ["read", "approve", "review"],
"forbidden_operations": [],
"base_url": None,
}
mock_get_all.return_value = [
{"number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1", "sha": "abc1"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "other_user"}}
]
# mock_api: 1) /user (inventory), 2) /user (eligibility), 3) /pulls/1 (eligibility), 4) /pulls/1/reviews (POST review)
mock_api.side_effect = [
{"login": "reviewer1"}, # inventory whoami
{"login": "reviewer1"}, # eligibility whoami
{"state": "open", "head": {"sha": "abc1"}, "mergeable": True, "user": {"login": "other_user"}}, # eligibility PR details
{"id": 100} # POST review
]
result = gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs")
self.assertTrue(result["success"])
self.assertIn("Successfully submitted review", result["message"])
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
@patch("gitea_config.is_runtime_switching_enabled", return_value=False)
def test_static_runtime_rejects_or_hides_profile_activation(self, mock_switch, mock_get_profile, _auth, mock_api, mock_get_all):
mock_get_profile.return_value = {
"profile_name": "gitea-author",
"allowed_operations": ["read"],
"forbidden_operations": [],
"base_url": None,
}
mock_get_all.return_value = []
mock_api.return_value = {"login": "jcwalker3"}
result = gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs")
self.assertFalse(result["success"])
self.assertNotIn("gitea_activate_profile", result["message"])
self.assertIn("Switch to the reviewer MCP session", result["message"])
@patch("mcp_server.get_profile")
def test_stopped_before_listing_is_not_reported_as_no_prs(self, mock_get_profile):
mock_get_profile.return_value = {
"profile_name": "gitea-restricted",
"allowed_operations": [],
"forbidden_operations": [],
"base_url": None,
}
result = gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs")
self.assertFalse(result["success"])
self.assertIn("PR inventory not attempted", result["message"])
self.assertNotIn("Open PRs found: 0", result["message"])
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
def test_reviewer_eligibility_check_not_run_for_author_profile(self, mock_get_profile, _auth, mock_api):
mock_get_profile.return_value = {
"profile_name": "gitea-author",
"allowed_operations": ["read"],
"forbidden_operations": [],
"base_url": None,
}
result = gitea_check_pr_eligibility(pr_number=1, action="approve", remote="prgs")
self.assertFalse(result["eligible"])
self.assertEqual(mock_api.call_count, 0)