diff --git a/mcp_server.py b/mcp_server.py index 7551e47..3364e26 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -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 diff --git a/tests/test_llm_agent_sha.py b/tests/test_llm_agent_sha.py index edbe518..4c827b1 100644 --- a/tests/test_llm_agent_sha.py +++ b/tests/test_llm_agent_sha.py @@ -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( diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 2e48a1c..abff0d0 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -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( diff --git a/tests/test_permission_reports.py b/tests/test_permission_reports.py index b2959ef..b1a4d2d 100644 --- a/tests/test_permission_reports.py +++ b/tests/test_permission_reports.py @@ -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"]) diff --git a/tests/test_pr_queue_inventory.py b/tests/test_pr_queue_inventory.py new file mode 100644 index 0000000..facc68d --- /dev/null +++ b/tests/test_pr_queue_inventory.py @@ -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)