From 2941ec529c1e9a27b46ea66723df4bd34d7f3c9f Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Thu, 9 Jul 2026 15:41:07 -0400 Subject: [PATCH] fix: enforce merger role boundaries (Closes #539) --- gitea_mcp_server.py | 69 ++++++++++++++++++++++++--- role_session_router.py | 30 +++++++++++- tests/test_resolve_task_capability.py | 43 +++++++++++++++++ tests/test_role_session_router.py | 42 +++++++++++++++- 4 files changed, 175 insertions(+), 9 deletions(-) diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index dd50942..ca17c32 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -211,6 +211,21 @@ def _effective_workspace_role() -> str: ) +def _profile_role_kind(profile: dict) -> str: + """Resolve a profile's declared role before inferring from permissions.""" + role = (profile.get("role") or profile.get("role_kind") or "").strip() + if role: + return role + profile_name = (profile.get("profile_name") or "").strip().lower() + for candidate in ("reconciler", "merger", "reviewer", "author"): + if candidate in profile_name: + return candidate + return _role_kind( + profile.get("allowed_operations") or [], + profile.get("forbidden_operations") or [], + ) + + def _actual_profile_role() -> str: """Resolve the workspace role from the ACTIVE PROFILE alone (#540). @@ -223,10 +238,7 @@ def _actual_profile_role() -> str: ``author`` here, so author blocking is preserved. """ profile = get_profile() - role = _role_kind( - profile.get("allowed_operations") or [], - profile.get("forbidden_operations") or [], - ) + role = _profile_role_kind(profile) return nwb.normalize_role_kind( role, profile_name=profile.get("profile_name"), @@ -9917,6 +9929,24 @@ def gitea_resolve_task_capability( required_permission = task_capability_map.required_permission(task) required_role = task_capability_map.required_role(task) + role_exclusive_tasks = { + "review_pr", + "approve_pr", + "request_changes_pr", + "blind_pr_queue_review", + "pr_queue_cleanup", + "pr-queue-cleanup", + "merge_pr", + "create_branch", + "push_branch", + "create_pr", + "commit_files", + "gitea_commit_files", + "address_pr_change_requests", + "delete_branch", + "work_issue", + "work-issue", + } infra_assessment = role_session_router.assess_infra_stop(PROJECT_ROOT) if required_role == "reviewer": @@ -10011,11 +10041,26 @@ def gitea_resolve_task_capability( # Load active permissions active_allowed = profile.get("allowed_operations") or [] active_forbidden = profile.get("forbidden_operations") or [] + active_role_kind = _profile_role_kind(profile) # Check if allowed in current session - allowed_in_current_session, _ = gitea_config.check_operation( + permission_allowed_in_current_session, _ = gitea_config.check_operation( required_permission, active_allowed, active_forbidden ) + role_matches_current_session = ( + task not in role_exclusive_tasks + or active_role_kind == required_role + ) + role_mismatch_reason = None + if not role_matches_current_session: + role_mismatch_reason = ( + f"Active profile role '{active_role_kind}' cannot perform " + f"{required_role} task '{task}' even if nearby permissions are " + "present (fail closed)." + ) + allowed_in_current_session = ( + permission_allowed_in_current_session and role_matches_current_session + ) switching = gitea_config.is_runtime_switching_enabled() available_in_session = allowed_in_current_session @@ -10042,7 +10087,13 @@ def gitea_resolve_task_capability( except Exception: pass ok, _ = gitea_config.check_operation(required_permission, p_allowed_n, p_forbidden_n) - if ok: + p_role = (p_data.get("role") or "").strip() + p_role_ok = ( + task not in role_exclusive_tasks + or not p_role + or p_role == required_role + ) + if ok and p_role_ok: matching_profiles.append(p_name) configured = len(matching_profiles) > 0 @@ -10070,6 +10121,8 @@ def gitea_resolve_task_capability( reason_msg = ( f"No profile configured with permission '{required_permission}'." ) + elif role_mismatch_reason: + reason_msg = role_mismatch_reason different_namespace_required = False next_safe_action = "None; ready for operations." @@ -10101,6 +10154,8 @@ def gitea_resolve_task_capability( # into author-side mutations. stop_required = not allowed_in_current_session task_role_guidance = [] + if role_mismatch_reason: + task_role_guidance.append(f"STOP: {role_mismatch_reason}") if required_role == "reviewer": if allowed_in_current_session: task_role_guidance.append( @@ -10146,7 +10201,9 @@ def gitea_resolve_task_capability( "required_role_kind": required_role, "active_profile": profile["profile_name"], "active_identity": username, + "active_role_kind": active_role_kind, "active_profile_allowed_operations": active_allowed, + "active_profile_permission_allowed": permission_allowed_in_current_session, "allowed_in_current_session": allowed_in_current_session, "available_in_session": available_in_session, "configured": configured, diff --git a/role_session_router.py b/role_session_router.py index 7fedcd7..1fdfd72 100644 --- a/role_session_router.py +++ b/role_session_router.py @@ -55,7 +55,6 @@ def skip_python_scan_walk_root(project_root: str, walk_root: str) -> bool: REVIEWER_TASKS = frozenset({ "review_pr", - "merge_pr", "blind_pr_queue_review", "pr_queue_cleanup", "pr-queue-cleanup", @@ -63,6 +62,10 @@ REVIEWER_TASKS = frozenset({ "approve_pr", }) +MERGER_TASKS = frozenset({ + "merge_pr", +}) + AUTHOR_TASKS = frozenset({ "create_issue", "comment_issue", @@ -98,7 +101,7 @@ TASK_REQUIRED_ROLE = { "address_pr_change_requests": "author", "delete_branch": "author", "review_pr": "reviewer", - "merge_pr": "reviewer", + "merge_pr": "merger", "blind_pr_queue_review": "reviewer", "pr_queue_cleanup": "reviewer", "pr-queue-cleanup": "reviewer", @@ -128,6 +131,10 @@ WRONG_ROLE_RECONCILER_MSG = ( "MCP namespace/profile with exact close capability." ) +WRONG_ROLE_MERGER_MSG = ( + "Wrong role/session for merger task. Launch merger MCP namespace." +) + _session_last_route: dict | None = None @@ -243,6 +250,25 @@ def route_task_session( _record_route(result) return result + if required_role == "merger": + result = { + "task_type": task_type, + "required_role": required_role, + "active_role": active_role_kind, + "active_profile": active_profile, + "route_result": ROUTE_WRONG_ROLE, + "downstream_allowed": False, + "reasons": [ + WRONG_ROLE_MERGER_MSG, + "Merger tasks cannot run in author or reviewer sessions.", + ], + "message": WRONG_ROLE_MERGER_MSG, + "runtime_switching_supported": runtime_switching_supported, + "profile_switch_blocked": not runtime_switching_supported, + } + _record_route(result) + return result + if required_role == "author": route = ROUTE_TO_AUTHOR message = ( diff --git a/tests/test_resolve_task_capability.py b/tests/test_resolve_task_capability.py index cdb2e9d..23f05bc 100644 --- a/tests/test_resolve_task_capability.py +++ b/tests/test_resolve_task_capability.py @@ -287,6 +287,49 @@ class TestResolveTaskCapability(unittest.TestCase): self.assertEqual(res["required_operation_permission"], "gitea.pr.merge") self.assertEqual(res["required_role_kind"], "merger") + @patch("mcp_server.api_request", return_value={"login": "reviewer-user"}) + @patch("mcp_server.get_auth_header", return_value="token reviewer-pass") + def test_reviewer_role_with_merge_permission_still_cannot_merge(self, _auth, _api): + # #539: exact permission alone is insufficient. The active profile + # role must also be merger for merge_pr. + config = json.loads(json.dumps(CONFIG_RESOLVER)) + config["profiles"]["reviewer-with-merge"] = { + "enabled": True, + "context": "ctx", + "role": "reviewer", + "username": "reviewer-user", + "auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"}, + "allowed_operations": [ + "gitea.read", "gitea.pr.review", "gitea.pr.approve", + "gitea.pr.merge", + ], + "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"], + "execution_profile": "reviewer-with-merge", + } + self._write_config(config) + with patch.dict(os.environ, self._env("reviewer-with-merge")): + res = mcp_server.gitea_resolve_task_capability( + task="merge_pr", remote="prgs" + ) + self.assertEqual(res["required_role_kind"], "merger") + self.assertEqual(res["active_role_kind"], "reviewer") + self.assertTrue(res["active_profile_permission_allowed"]) + self.assertFalse(res["allowed_in_current_session"]) + self.assertTrue(res["stop_required"]) + self.assertIn("cannot perform merger task", " ".join(res["task_role_guidance"])) + + @patch("mcp_server.api_request", return_value={"login": "merger-user"}) + @patch("mcp_server.get_auth_header", return_value="token merger-pass") + def test_merger_profile_can_resolve_merge_task(self, _auth, _api): + with patch.dict(os.environ, self._env("merger-profile")): + res = mcp_server.gitea_resolve_task_capability( + task="merge_pr", remote="prgs" + ) + self.assertEqual(res["required_role_kind"], "merger") + self.assertEqual(res["active_role_kind"], "merger") + self.assertTrue(res["allowed_in_current_session"]) + self.assertFalse(res["stop_required"]) + @patch("mcp_server.api_request") def test_self_review_blocked_structured(self, mock_api): # Test that self-approve (self-review gate) is blocked and returns structured reasons diff --git a/tests/test_role_session_router.py b/tests/test_role_session_router.py index f545963..3d5d0d0 100644 --- a/tests/test_role_session_router.py +++ b/tests/test_role_session_router.py @@ -52,6 +52,20 @@ CONFIG = { ], "execution_profile": "prgs-reviewer", }, + "prgs-merger": { + "enabled": True, + "context": "ctx", + "role": "merger", + "username": "sysadmin", + "auth": {"type": "env", "name": "GITEA_TOKEN_MERGER"}, + "allowed_operations": [ + "gitea.read", "gitea.pr.merge", "gitea.issue.comment", + ], + "forbidden_operations": [ + "gitea.pr.create", "gitea.branch.push", "gitea.pr.approve", + ], + "execution_profile": "prgs-merger", + }, "prgs-reconciler": { "enabled": True, "context": "ctx", @@ -104,6 +118,7 @@ class TestRoleSessionRouter(unittest.TestCase): "GITEA_MCP_PROFILE": profile, "GITEA_TOKEN_AUTHOR": "author-pass", "GITEA_TOKEN_REVIEWER": "reviewer-pass", + "GITEA_TOKEN_MERGER": "merger-pass", "GITEA_TOKEN_RECONCILER": "reconciler-pass", } @@ -227,6 +242,31 @@ class TestRoleSessionRouter(unittest.TestCase): self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED) self.assertTrue(route["downstream_allowed"]) + @patch("mcp_server.api_request", return_value={"login": "sysadmin"}) + @patch("mcp_server.get_auth_header", return_value="token reviewer-pass") + def test_merge_task_under_reviewer_profile_wrong_role_stop(self, _auth, _api): + # #539: a reviewer session must not pass the pre-task merge route + # even if its config accidentally includes gitea.pr.merge. + with patch.dict(os.environ, self._env("prgs-reviewer")): + route = mcp_server.gitea_route_task_session( + task_type="merge_pr", remote="prgs" + ) + self.assertEqual(route["required_role"], "merger") + self.assertEqual(route["route_result"], role_session_router.ROUTE_WRONG_ROLE) + self.assertFalse(route["downstream_allowed"]) + self.assertIn("merger", route["message"].lower()) + + @patch("mcp_server.api_request", return_value={"login": "sysadmin"}) + @patch("mcp_server.get_auth_header", return_value="token merger-pass") + def test_merge_task_under_merger_profile_allowed(self, _auth, _api): + with patch.dict(os.environ, self._env("prgs-merger")): + route = mcp_server.gitea_route_task_session( + task_type="merge_pr", remote="prgs" + ) + self.assertEqual(route["required_role"], "merger") + self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED) + self.assertTrue(route["downstream_allowed"]) + @patch("mcp_server.api_request", return_value={"login": "sysadmin"}) @patch("mcp_server.get_auth_header", return_value="token reviewer-pass") def test_author_task_under_reviewer_profile_routes_to_author(self, _auth, _api): @@ -334,4 +374,4 @@ class TestCheckMidMerge(unittest.TestCase): if __name__ == "__main__": - unittest.main() \ No newline at end of file + unittest.main() -- 2.43.7