From dfb5ebd0afd97fafc8c120176f1bcf8226e6fa96 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Fri, 17 Jul 2026 01:03:26 -0400 Subject: [PATCH] Fix #723: Prevent poisoned role stamp during task capability denial This fixes #723 Defect B where attempting to acquire a reviewer lease from a merger session caused the session role to be incorrectly stamped as 'reviewer', leading to downstream workspace binding exceptions. This commit moves the preflight capability recording to only apply when the task is allowed, and adds a try/except downstream to correctly fail-closed instead of throwing RuntimeError. --- gitea_mcp_server.py | 32 ++++++++++++++++++++------- tests/test_resolve_task_capability.py | 18 +++++++++++++++ 2 files changed, 42 insertions(+), 8 deletions(-) diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 6fb55cd..7097ac0 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -4096,9 +4096,6 @@ def _evaluate_pr_review_submission( worktree_path: str | None = None, ) -> dict: """Shared gate chain for live submit and dry-run review tools.""" - _verify_role_mutation_workspace( - remote, worktree_path=worktree_path, task="review_pr" - ) action = (action or "").strip().lower() workflow_blockers = _review_workflow_load_gate_reasons() if live else [] result = { @@ -4115,6 +4112,13 @@ def _evaluate_pr_review_submission( "remote": remote if remote in REMOTES else None, "reasons": [], } + try: + _verify_role_mutation_workspace( + remote, worktree_path=worktree_path, task="review_pr" + ) + except RuntimeError as e: + result["reasons"].append(str(e)) + return result reasons = result["reasons"] if workflow_blockers: reasons.extend(workflow_blockers) @@ -10018,9 +10022,20 @@ def gitea_adopt_merger_pr_lease( "permission_report": _permission_block_report("gitea.pr.merge"), } - _verify_role_mutation_workspace( - remote, worktree=worktree, task="adopt_merger_pr_lease" - ) + try: + _verify_role_mutation_workspace( + remote, worktree=worktree, task="adopt_merger_pr_lease" + ) + except RuntimeError as e: + return { + "success": False, + "adopted": False, + "pr_number": pr_number, + "reasons": [str(e)], + "active_lease": None, + "expected_head_sha": expected_head_sha, + "live_head_sha": None, + } h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) profile = get_profile() @@ -13921,8 +13936,6 @@ def gitea_resolve_task_capability( "exact_safe_next_action": next_safe_action, } - record_preflight_check("capability", required_role, resolved_task=task) - # Try automatic dispatch switching _ensure_matching_profile(required_permission, required_role, remote, host) @@ -13956,6 +13969,9 @@ def gitea_resolve_task_capability( permission_allowed_in_current_session and role_matches_current_session ) + if allowed_in_current_session: + record_preflight_check("capability", required_role, resolved_task=task) + switching = gitea_config.is_runtime_switching_enabled() available_in_session = allowed_in_current_session configured = False diff --git a/tests/test_resolve_task_capability.py b/tests/test_resolve_task_capability.py index 23f05bc..9a0b0aa 100644 --- a/tests/test_resolve_task_capability.py +++ b/tests/test_resolve_task_capability.py @@ -442,5 +442,23 @@ class TestResolveTaskCapability(unittest.TestCase): with self.assertRaises(ValueError): mcp_server.gitea_resolve_task_capability(task=unknown, remote="prgs") + @patch("mcp_server.api_request", return_value={"login": "author-user"}) + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_resolve_task_capability_does_not_poison_role_stamp_on_denial(self, _auth, _api): + # Issue #723: attempting to acquire a reviewer lease from a merger profile + # should fail without poisoning the session role stamp as 'reviewer'. + with patch.dict(os.environ, self._env("merger-profile")): + # Initially no preflight check is recorded + self.assertEqual(mcp_server._preflight_resolved_role, None) + + # Request reviewer lease task which should be denied for merger + res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs") + + # The task is denied + self.assertFalse(res["allowed_in_current_session"]) + + # The session role stamp should NOT be poisoned + self.assertNotEqual(mcp_server._preflight_resolved_role, "reviewer") + if __name__ == "__main__": unittest.main()