diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 6fb55cd..f4eae9f 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -517,6 +517,19 @@ def _clear_preflight_capability_state() -> None: _preflight_reviewer_violation_files = [] +def _clear_resolved_capability_stamp() -> None: + """#723 AC3: drop only the resolved role/task stamp (denied resolve). + + ``record_preflight_check("capability")`` without a role intentionally + preserves a prior stamp, so a DENIED resolve must clear it explicitly — + otherwise the stale role poisons :func:`_effective_workspace_role` and a + later mutation preflight raises instead of failing closed with reasons. + """ + global _preflight_resolved_role, _preflight_resolved_task + _preflight_resolved_role = None + _preflight_resolved_task = None + + def record_preflight_check( type_name: str, resolved_role: str | None = None, @@ -4096,9 +4109,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 = { @@ -4116,6 +4126,20 @@ def _evaluate_pr_review_submission( "reasons": [], } reasons = result["reasons"] + # #723 AC4: workspace/role-binding failures fail closed with structured + # reasons — never escape as a generic internal_error (PR #721: a stamped + # 'merger' role raised RuntimeError here and the client saw an + # unactionable internal_error after mark_final had already succeeded). + try: + _verify_role_mutation_workspace( + remote, worktree_path=worktree_path, task="review_pr" + ) + except RuntimeError as exc: + result["blocker_kind"] = "workspace_role_binding" + reasons.append( + "workspace/role binding failed (fail closed, #723): " f"{exc}" + ) + return result if workflow_blockers: reasons.extend(workflow_blockers) reasons.extend(review_workflow_load.recovery_handoff_without_replay()) @@ -11891,8 +11915,16 @@ _RUNTIME_CAPABILITY_TASKS = ( def _matching_configured_profiles( config: dict | None, required_permission: str, + required_role_kind: str | None = None, ) -> list[str]: - """Profile names that allow *required_permission* (redacted metadata only).""" + """Profile names that allow *required_permission* (redacted metadata only). + + #723 AC5: when *required_role_kind* is supplied (role-exclusive tasks), + a profile with a declared role must also match it — mirroring the + resolver's matching rule so the two lists cannot drift. Profiles without + a declared role still match on permission alone, exactly like the + resolver. + """ if not config or "profiles" not in config: return [] matches: list[str] = [] @@ -11916,7 +11948,13 @@ def _matching_configured_profiles( ok, _ = gitea_config.check_operation( required_permission, p_allowed_n, p_forbidden_n ) - if ok: + p_role = (p_data.get("role") or "").strip() + role_ok = ( + required_role_kind is None + or not p_role + or p_role == required_role_kind + ) + if ok and role_ok: matches.append(p_name) return sorted(matches) @@ -11925,8 +11963,16 @@ def _build_runtime_task_capabilities( allowed: list[str], forbidden: list[str], config: dict | None, + active_role_kind: str | None = None, ) -> dict: - """Per-task capability summary for role-aware runtime context (#139).""" + """Per-task capability summary for role-aware runtime context (#139). + + #723 AC5: applies the same role-exclusive filter as + ``gitea_resolve_task_capability`` when *active_role_kind* is supplied, so + runtime context can never report a role-exclusive task as allowed while + the resolver fail-closes it in the same session (incident #722). Without + an *active_role_kind* the view is permission-only and each entry says so. + """ task_entries = [] flags: dict[str, bool] = {} flag_keys = { @@ -11939,18 +11985,35 @@ def _build_runtime_task_capabilities( "close_issue": "can_close_issues", "reconcile_already_landed_pr": "can_reconcile_already_landed_prs", } + role_filter_applied = active_role_kind is not None for task in _RUNTIME_CAPABILITY_TASKS: permission = task_capability_map.required_permission(task) - allowed_here, _ = gitea_config.check_operation( + required_role_kind = task_capability_map.required_role(task) + role_exclusive = task in task_capability_map.ROLE_EXCLUSIVE_TASKS + permission_allowed, _ = gitea_config.check_operation( permission, allowed, forbidden ) + role_ok = ( + not role_exclusive + or not role_filter_applied + or active_role_kind == required_role_kind + ) + allowed_here = permission_allowed and role_ok entry = { "task": task, "required_permission": permission, - "required_role_kind": task_capability_map.required_role(task), + "required_role_kind": required_role_kind, + "role_exclusive": role_exclusive, + "capability_view": ( + "role_filtered" if role_filter_applied else "permission_only" + ), "allowed_in_current_session": allowed_here, "matching_configured_profiles": _matching_configured_profiles( - config, permission + config, + permission, + required_role_kind=( + required_role_kind if role_exclusive else None + ), ), } task_entries.append(entry) @@ -12403,7 +12466,10 @@ def gitea_get_runtime_context( ) session_capabilities = _build_runtime_task_capabilities( - allowed, forbidden, config + allowed, + forbidden, + config, + active_role_kind=_profile_role_kind(profile), ) preflight = assess_preflight_status(worktree_path) @@ -13821,26 +13887,9 @@ 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", - "cleanup_merged_pr_branch", - "reconciliation_cleanup", - "work_issue", - "work-issue", - } + # #723 AC5: single shared definition; the runtime-context capability + # report applies the same set so the two views cannot drift. + role_exclusive_tasks = task_capability_map.ROLE_EXCLUSIVE_TASKS infra_assessment = role_session_router.assess_infra_stop(PROJECT_ROOT) if required_role == "reviewer": @@ -13921,7 +13970,12 @@ def gitea_resolve_task_capability( "exact_safe_next_action": next_safe_action, } - record_preflight_check("capability", required_role, resolved_task=task) + # #723 AC3: record the capability purity baseline now, but DEFER the + # resolved-role/task stamp until after the allow/deny decision below. A + # denied resolve previously stamped ``_preflight_resolved_role`` anyway + # (PR #721: a denied review_pr resolve stamped 'merger', and the later + # submit escaped as a generic internal_error instead of failing closed). + record_preflight_check("capability") # Try automatic dispatch switching _ensure_matching_profile(required_permission, required_role, remote, host) @@ -13956,6 +14010,14 @@ def gitea_resolve_task_capability( permission_allowed_in_current_session and role_matches_current_session ) + # #723 AC3: stamp the resolved role/task only for an ALLOWED resolve; a + # denied resolve clears any stale stamp so later mutation preflights key + # off the real profile role and fail closed with structured reasons. + if allowed_in_current_session: + record_preflight_check("capability", required_role, resolved_task=task) + else: + _clear_resolved_capability_stamp() + switching = gitea_config.is_runtime_switching_enabled() available_in_session = allowed_in_current_session configured = False diff --git a/task_capability_map.py b/task_capability_map.py index 02826af..54f84a4 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -330,6 +330,32 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { }, } +# #723 AC1/AC5: tasks where permission alone is NOT enough — the profile's +# role kind must also match. Shared by ``gitea_resolve_task_capability`` and +# the runtime-context capability report so the two can never disagree about +# which tasks are role-exclusive (incident #722: runtime context said +# review_pr was allowed while the resolver fail-closed the same session). +ROLE_EXCLUSIVE_TASKS: frozenset[str] = frozenset({ + "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", + "cleanup_merged_pr_branch", + "reconciliation_cleanup", + "work_issue", + "work-issue", +}) + # Issue-mutating MCP tools and their resolver task keys. ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = { "gitea_create_issue": "create_issue", diff --git a/tests/test_issue_723_role_stamp_and_submission.py b/tests/test_issue_723_role_stamp_and_submission.py new file mode 100644 index 0000000..98f360c --- /dev/null +++ b/tests/test_issue_723_role_stamp_and_submission.py @@ -0,0 +1,337 @@ +"""#723 AC3/AC4/AC5: denied resolves must not poison the session role stamp, +review-submission workspace failures must fail closed with structured +reasons, and runtime-context capabilities must apply the resolver's +role-exclusive filter. + +Reproduces the PR #721 sequence: a reviewer session resolved a task whose +capability-map role had drifted to ``merger``; the denied resolve stamped +``_preflight_resolved_role = "merger"`` anyway, ``mark_final`` succeeded, and +``gitea_submit_pr_review`` raised RuntimeError out of the tool as a generic +``internal_error``. +""" + +from __future__ import annotations + +import os +import sys +import unittest +from pathlib import Path +from unittest.mock import patch + +ROOT = str(Path(__file__).resolve().parent.parent) +if ROOT not in sys.path: + sys.path.insert(0, ROOT) + +import gitea_mcp_server as mcp_server +import task_capability_map + + +REVIEWER_PROFILE = { + "profile_name": "prgs-reviewer", + "role": "reviewer", + "allowed_operations": [ + "gitea.read", + "gitea.pr.review", + "gitea.pr.approve", + "gitea.pr.request_changes", + "gitea.pr.comment", + "gitea.issue.comment", + ], + "forbidden_operations": [ + "gitea.branch.create", + "gitea.branch.push", + "gitea.repo.commit", + "gitea.pr.create", + "gitea.pr.merge", + ], +} + +CONFIG = { + "profiles": { + "prgs-reviewer": { + "role": "reviewer", + "allowed_operations": REVIEWER_PROFILE["allowed_operations"], + "forbidden_operations": REVIEWER_PROFILE["forbidden_operations"], + }, + "prgs-merger": { + "role": "merger", + "allowed_operations": [ + "gitea.read", + "gitea.pr.merge", + "gitea.pr.comment", + "gitea.issue.comment", + ], + "forbidden_operations": [ + "gitea.pr.approve", + "gitea.pr.review", + "gitea.pr.request_changes", + ], + }, + } +} + + +def _reset_preflight(): + mcp_server._clear_preflight_capability_state() + mcp_server._preflight_whoami_called = False + mcp_server._preflight_whoami_violation = False + + +class _ResolveHarness(unittest.TestCase): + """Shared patched-resolver harness.""" + + def setUp(self): + _reset_preflight() + + def tearDown(self): + _reset_preflight() + + def _resolve(self, task, profile=REVIEWER_PROFILE, required_role=None): + """Run gitea_resolve_task_capability with a fixed profile/config.""" + patches = [ + patch.object(mcp_server, "get_profile", return_value=profile), + patch.object( + mcp_server.gitea_config, "load_config", return_value=CONFIG + ), + patch.object( + mcp_server, "_authenticated_username", return_value="tester" + ), + patch.object( + mcp_server, "_ensure_matching_profile", return_value=None + ), + patch.object( + mcp_server, "init_review_decision_lock", return_value=None + ), + ] + if required_role is not None: + patches.append( + patch.object( + mcp_server.task_capability_map, + "required_role", + side_effect=lambda t: ( + required_role + if t == task + else task_capability_map.TASK_CAPABILITY_MAP[t]["role"] + ), + ) + ) + for p in patches: + p.__enter__() + try: + return mcp_server.gitea_resolve_task_capability( + task=task, remote="prgs" + ) + finally: + for p in reversed(patches): + p.__exit__(None, None, None) + + +class TestAC3DeniedResolveDoesNotStampRole(_ResolveHarness): + def test_allowed_resolve_stamps_role(self): + result = self._resolve("review_pr") + self.assertTrue(result["allowed_in_current_session"], result) + self.assertEqual(mcp_server._preflight_resolved_role, "reviewer") + self.assertEqual(mcp_server._preflight_resolved_task, "review_pr") + + def test_denied_resolve_does_not_stamp_required_role(self): + """The PR #721 poison: review_pr requiring merger under a reviewer.""" + result = self._resolve("review_pr", required_role="merger") + self.assertFalse(result["allowed_in_current_session"], result) + self.assertIsNone( + mcp_server._preflight_resolved_role, + "denied resolve must not stamp the required role (#723 AC3)", + ) + self.assertIsNone(mcp_server._preflight_resolved_task) + + def test_denied_resolve_clears_prior_stale_stamp(self): + allowed = self._resolve("review_pr") + self.assertTrue(allowed["allowed_in_current_session"]) + self.assertEqual(mcp_server._preflight_resolved_role, "reviewer") + + denied = self._resolve("merge_pr") # reviewer profile cannot merge + self.assertFalse(denied["allowed_in_current_session"], denied) + self.assertIsNone( + mcp_server._preflight_resolved_role, + "a denied resolve must clear the previous stamp, not keep it", + ) + + def test_denied_resolve_keeps_effective_role_on_actual_profile(self): + with patch.object( + mcp_server, "get_profile", return_value=REVIEWER_PROFILE + ): + self._resolve("review_pr", required_role="merger") + self.assertEqual( + mcp_server._effective_workspace_role(), "reviewer" + ) + + +class TestAC4SubmissionFailsClosedStructured(unittest.TestCase): + def setUp(self): + _reset_preflight() + + def tearDown(self): + _reset_preflight() + + def test_workspace_binding_failure_returns_reasons_not_raise(self): + boom = RuntimeError( + "namespace workspace binding blocked: role 'merger' cannot " + "mutate from reviewer session workspace" + ) + with patch.object( + mcp_server, + "_verify_role_mutation_workspace", + side_effect=boom, + ), patch.object( + mcp_server, "get_profile", return_value=REVIEWER_PROFILE + ): + result = mcp_server._evaluate_pr_review_submission( + pr_number=721, + action="approve", + expected_head_sha="80f59b334e6671b08006725292c08a8e8b6c823f", + remote="prgs", + live=True, + final_review_decision_ready=True, + ) + self.assertFalse(result["performed"]) + self.assertEqual(result["blocker_kind"], "workspace_role_binding") + self.assertTrue( + any("workspace/role binding failed" in r for r in result["reasons"]), + result["reasons"], + ) + self.assertTrue( + any("cannot mutate from reviewer session" in r for r in result["reasons"]), + "the underlying binding error text must be preserved", + ) + + def test_stale_runtime_failure_also_structured(self): + boom = RuntimeError( + "stale-runtime: The active Gitea MCP server process is stale" + ) + with patch.object( + mcp_server, + "_verify_role_mutation_workspace", + side_effect=boom, + ), patch.object( + mcp_server, "get_profile", return_value=REVIEWER_PROFILE + ): + result = mcp_server._evaluate_pr_review_submission( + pr_number=721, + action="approve", + remote="prgs", + live=True, + ) + self.assertFalse(result["performed"]) + self.assertEqual(result["blocker_kind"], "workspace_role_binding") + self.assertTrue( + any("stale-runtime" in r for r in result["reasons"]), + result["reasons"], + ) + + def test_dry_run_also_fails_closed_structured(self): + boom = RuntimeError("binding blocked") + with patch.object( + mcp_server, + "_verify_role_mutation_workspace", + side_effect=boom, + ), patch.object( + mcp_server, "get_profile", return_value=REVIEWER_PROFILE + ): + result = mcp_server._evaluate_pr_review_submission( + pr_number=721, + action="approve", + remote="prgs", + live=False, + ) + self.assertFalse(result["would_perform"]) + self.assertEqual(result["blocker_kind"], "workspace_role_binding") + + +class TestAC5RuntimeContextRoleFilter(unittest.TestCase): + def test_role_filtered_view_matches_resolver_denial(self): + """Reviewer session: merge_pr stays denied under the role filter even + when the permission is (pathologically) present.""" + allowed_ops = [ + "gitea.read", + "gitea.pr.review", + "gitea.pr.approve", + "gitea.pr.request_changes", + "gitea.pr.comment", + "gitea.issue.comment", + "gitea.pr.merge", + ] + caps = mcp_server._build_runtime_task_capabilities( + allowed_ops, [], CONFIG, active_role_kind="reviewer" + ) + merge_entry = next( + t for t in caps["task_capabilities"] if t["task"] == "merge_pr" + ) + self.assertTrue(merge_entry["role_exclusive"]) + self.assertEqual(merge_entry["capability_view"], "role_filtered") + self.assertFalse( + merge_entry["allowed_in_current_session"], + "role-exclusive merge_pr must stay denied for a reviewer role " + "even when the permission is present (#723 AC5)", + ) + self.assertFalse(caps["can_merge_prs"]) + + def test_role_filter_restricts_matching_profiles(self): + caps = mcp_server._build_runtime_task_capabilities( + ["gitea.read"], [], CONFIG, active_role_kind="author" + ) + review_entry = next( + t for t in caps["task_capabilities"] if t["task"] == "review_pr" + ) + self.assertEqual( + review_entry["matching_configured_profiles"], + ["prgs-reviewer"], + "role-exclusive review_pr must not list the merger profile", + ) + merge_entry = next( + t for t in caps["task_capabilities"] if t["task"] == "merge_pr" + ) + self.assertEqual( + merge_entry["matching_configured_profiles"], ["prgs-merger"] + ) + + def test_permission_only_view_is_labeled(self): + caps = mcp_server._build_runtime_task_capabilities( + ["gitea.pr.merge", "gitea.read"], [], CONFIG + ) + merge_entry = next( + t for t in caps["task_capabilities"] if t["task"] == "merge_pr" + ) + self.assertEqual(merge_entry["capability_view"], "permission_only") + # Legacy permission-only semantics preserved when no role supplied. + self.assertTrue(merge_entry["allowed_in_current_session"]) + + def test_incident_722_shape_runtime_context_agrees_with_resolver(self): + """Post-970e68b shape: review_pr requires merger; a reviewer session + must see review_pr denied in runtime context, matching the resolver.""" + with patch.object( + mcp_server.task_capability_map, + "required_role", + side_effect=lambda t: ( + "merger" + if t == "review_pr" + else task_capability_map.TASK_CAPABILITY_MAP[t]["role"] + ), + ): + caps = mcp_server._build_runtime_task_capabilities( + REVIEWER_PROFILE["allowed_operations"], + REVIEWER_PROFILE["forbidden_operations"], + CONFIG, + active_role_kind="reviewer", + ) + review_entry = next( + t for t in caps["task_capabilities"] if t["task"] == "review_pr" + ) + self.assertFalse( + review_entry["allowed_in_current_session"], + "runtime context must not report review_pr allowed when the " + "resolver would fail-close it (incident #722)", + ) + self.assertFalse(caps["can_review_prs"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_task_capability_role_invariants.py b/tests/test_task_capability_role_invariants.py index a77f570..5ffce7b 100644 --- a/tests/test_task_capability_role_invariants.py +++ b/tests/test_task_capability_role_invariants.py @@ -19,7 +19,12 @@ import unittest import gitea_config from role_session_router import MERGER_TASKS, REVIEWER_TASKS -from task_capability_map import required_permission, required_role +from task_capability_map import ( + ROLE_EXCLUSIVE_TASKS, + TASK_CAPABILITY_MAP, + required_permission, + required_role, +) # Canonical role-profile permission shape. Mirrors the configured # author/reviewer/merger/reconciler profiles (profiles.json v2 role split): @@ -201,5 +206,44 @@ class TestMergerBoundary(unittest.TestCase): ) +class TestRoleExclusiveSetIntegrity(unittest.TestCase): + """#723 AC1/AC5: the shared role-exclusive set stays coherent.""" + + def test_every_role_exclusive_task_exists_in_capability_map(self): + for task in sorted(ROLE_EXCLUSIVE_TASKS): + with self.subTest(task=task): + self.assertIn( + task, + TASK_CAPABILITY_MAP, + f"role-exclusive task {task!r} missing from the " + f"capability map — required_role() would raise and the " + f"resolver would 500 instead of failing closed", + ) + + def test_formal_review_tasks_are_role_exclusive(self): + for task in FORMAL_REVIEW_TASKS: + with self.subTest(task=task): + self.assertIn(task, ROLE_EXCLUSIVE_TASKS) + + def test_merge_pr_is_role_exclusive_and_merger_satisfiable(self): + """AC1 merger equivalent: merging must stay possible for a canonical + merger profile (permission AND role together).""" + self.assertIn("merge_pr", ROLE_EXCLUSIVE_TASKS) + self.assertTrue( + _profile_satisfies("merger", "merge_pr"), + "no canonical merger profile satisfies merge_pr — merging would " + "be impossible for every configured profile", + ) + + def test_reconciler_cleanup_tasks_stay_reconciler_satisfiable(self): + for task in ("cleanup_merged_pr_branch", "reconciliation_cleanup"): + with self.subTest(task=task): + self.assertIn(task, ROLE_EXCLUSIVE_TASKS) + self.assertTrue( + _profile_satisfies("reconciler", task), + f"no canonical reconciler profile satisfies {task!r}", + ) + + if __name__ == "__main__": unittest.main()