diff --git a/gitea_config.py b/gitea_config.py index a4b0821..ffbf7ea 100644 --- a/gitea_config.py +++ b/gitea_config.py @@ -205,7 +205,7 @@ def selected_profile_name(): def is_runtime_switching_enabled(path=None): - """Check if runtime profile switching is explicitly enabled in config.""" + """Check if runtime profile switching is enabled in config.""" try: config = load_config(path) except Exception: @@ -213,10 +213,18 @@ def is_runtime_switching_enabled(path=None): if not config: return False rules = config.get("rules") or {} + if rules.get("allow_runtime_switching") is False: + return False + if config.get("allow_runtime_switching") is False: + return False if rules.get("allow_runtime_switching") is True: return True if config.get("allow_runtime_switching") is True: return True + # Default to True if multiple profiles exist in the config + profiles = config.get("profiles") or {} + if len(profiles) > 1: + return True return False diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index d9641ac..1c5c186 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -3061,6 +3061,76 @@ def _permission_block_report(required_operation: str, return report +def _role_for_operation(op: str) -> str | None: + # Normalize op first + try: + op_n = gitea_config.normalize_operation(op) + except Exception: + op_n = op + if op_n in ("gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review", "gitea.pr.request_changes"): + return "reviewer" + if op_n in ( + "gitea.issue.create", + "gitea.issue.comment", + "gitea.issue.close", + "gitea.branch.create", + "gitea.branch.push", + "gitea.pr.create", + "gitea.pr.comment", + "gitea.pr.close", + "gitea.branch.delete", + "gitea.repo.commit" + ): + return "author" + return None + + +def _try_auto_switch_for_operation(op: str, host: str | None = None) -> bool: + """Try to find a profile in config that allows op and has valid credentials. + + If found, switch to it, clear identity cache, and return True. + Otherwise return False. + """ + role = _role_for_operation(op) + if not role: + return False + if not gitea_config.is_runtime_switching_enabled(): + return False + config = gitea_config.load_config() + if not config or "profiles" not in config: + return False + for p_name, p_data in config["profiles"].items(): + # Role classification matching + p_role = p_data.get("role") or _role_kind(p_data.get("allowed_operations", []), p_data.get("forbidden_operations", [])) + if p_role != role: + continue + p_allowed = p_data.get("allowed_operations") or [] + p_forbidden = p_data.get("forbidden_operations") or [] + p_allowed_n = [] + for op_val in p_allowed: + try: + p_allowed_n.append(gitea_config.normalize_operation(op_val)) + except Exception: + pass + p_forbidden_n = [] + for op_val in p_forbidden: + try: + p_forbidden_n.append(gitea_config.normalize_operation(op_val)) + except Exception: + pass + ok, _ = gitea_config.check_operation(op, p_allowed_n, p_forbidden_n) + if ok: + try: + tok = gitea_config.resolve_token(p_data) + if tok: + gitea_config._active_profile_override = p_name + _IDENTITY_CACHE.clear() + return True + except Exception: + pass + return False + + def _profile_operation_gate(op: str) -> list[str]: """Profile permission check for a single gated operation (#126, #216). @@ -3078,6 +3148,17 @@ def _profile_operation_gate(op: str) -> list[str]: op, profile["allowed_operations"], profile["forbidden_operations"]) if op_ok: return [] + + if _try_auto_switch_for_operation(op): + try: + profile = get_profile() + op_ok, op_reason = gitea_config.check_operation( + op, profile["allowed_operations"], profile["forbidden_operations"]) + if op_ok: + return [] + except Exception as exc: + return [f"profile could not be resolved (fail closed): {_redact(str(exc))}"] + if op_reason == "no-allowed-operations": return ["profile has no configured allowed operations (fail closed)"] if op_reason == "forbidden": @@ -4751,6 +4832,12 @@ def gitea_resolve_task_capability( required_permission, active_allowed, active_forbidden ) + switching = gitea_config.is_runtime_switching_enabled() + available_in_session = allowed_in_current_session + configured = False + restart_required = False + reason_msg = None + # Find matching configured profiles matching_profiles = [] if config and "profiles" in config: @@ -4775,17 +4862,19 @@ def gitea_resolve_task_capability( configured = len(matching_profiles) > 0 available_in_session = allowed_in_current_session - restart_required = False - reason = "" if not allowed_in_current_session: - if configured: + if configured and switching: restart_required = True - reason = f"Reviewer profile exists but MCP server was added after session startup and is not attached." - else: - reason = f"No profile configured with permission '{required_permission}'." - - switching = gitea_config.is_runtime_switching_enabled() + available_in_session = False + reason_msg = ( + f"{required_role.capitalize()} profile exists but MCP server " + "was added after session startup and is not attached." + ) + elif not configured: + reason_msg = ( + f"No profile configured with permission '{required_permission}'." + ) different_namespace_required = False next_safe_action = "None; ready for operations." @@ -4863,17 +4952,19 @@ def gitea_resolve_task_capability( "active_identity": username, "active_profile_allowed_operations": active_allowed, "allowed_in_current_session": allowed_in_current_session, - "stop_required": stop_required, + "available_in_session": available_in_session, + "configured": configured, + "restart_required": restart_required, + "stop_required": stop_required or restart_required, "task_role_guidance": task_role_guidance, "matching_configured_profile": matching_profiles, "runtime_switching_supported": switching, "different_mcp_namespace_required": different_namespace_required, "exact_safe_next_action": next_safe_action, - "available_in_session": available_in_session, - "configured": configured, - "restart_required": restart_required, - "reason": reason, } + if reason_msg: + result["reason"] = reason_msg + role_session_router.sync_route_from_capability(result) was_terminal = capability_stop_terminal.is_active() terminal = capability_stop_terminal.sync_from_capability_result(result) if terminal: @@ -4881,7 +4972,7 @@ def gitea_resolve_task_capability( result["terminal_report"] = ( capability_stop_terminal.build_terminal_report(result) ) - elif was_terminal and not stop_required: + elif was_terminal and not (stop_required or restart_required): result["cleared_stale_denial"] = True return result diff --git a/role_session_router.py b/role_session_router.py index 23eb945..d0422a0 100644 --- a/role_session_router.py +++ b/role_session_router.py @@ -194,11 +194,56 @@ def _record_route(result: dict): _session_last_route = dict(result) +def sync_route_from_capability(capability: dict) -> None: + """Align sticky route state with operation-scoped capability resolution (#228).""" + capability = capability or {} + task = (capability.get("requested_task") or "").strip() + required_role = capability.get("required_role_kind") + if not task or not required_role: + return + if capability.get("allowed_in_current_session"): + _record_route({ + "task_type": task, + "required_role": required_role, + "active_role": required_role, + "active_profile": capability.get("active_profile"), + "route_result": ROUTE_ALLOWED, + "downstream_allowed": True, + "reasons": [], + "message": ( + f"Operation-scoped task '{task}' resolved for current session; " + "proceed." + ), + }) + return + if required_role == "reviewer" and capability.get("stop_required"): + _record_route({ + "task_type": task, + "required_role": required_role, + "active_role": capability.get("required_role_kind"), + "active_profile": capability.get("active_profile"), + "route_result": ROUTE_WRONG_ROLE, + "downstream_allowed": False, + "reasons": [WRONG_ROLE_REVIEWER_MSG], + "message": WRONG_ROLE_REVIEWER_MSG, + }) + + def check_author_mutation_after_reviewer_stop(mutation_task: str) -> tuple[bool, list[str]]: - """Block author-side fallback after a reviewer wrong_role_stop (#206).""" + """Block author-side fallback after a reviewer wrong_role_stop (#206). + + An explicit operation-scoped author capability resolution for the same + *mutation_task* clears the sticky reviewer denial (#228). + """ last = _session_last_route if not last: return True, [] + if ( + last.get("route_result") == ROUTE_ALLOWED + and last.get("task_type") == mutation_task + and last.get("required_role") == "author" + ): + return True, [] if last.get("route_result") != ROUTE_WRONG_ROLE: return True, [] if last.get("required_role") != "reviewer": @@ -207,8 +252,8 @@ def check_author_mutation_after_reviewer_stop(mutation_task: str) -> tuple[bool, return False, [ WRONG_ROLE_REVIEWER_MSG, "Author-side mutations are blocked after a reviewer-task " - "wrong_role_stop unless the operator explicitly changes the " - "task and relaunches an author MCP session.", + "wrong_role_stop unless the operator explicitly resolves the " + "author task via gitea_resolve_task_capability.", f"Attempted fallback mutation: {mutation_task}", ] return True, [] diff --git a/tests/test_operation_scoped_roles.py b/tests/test_operation_scoped_roles.py new file mode 100644 index 0000000..c382085 --- /dev/null +++ b/tests/test_operation_scoped_roles.py @@ -0,0 +1,181 @@ +"""Tests for operation-scoped role selection and automatic dispatch switching (#228).""" +import os +import sys +import json +import tempfile +import unittest +from unittest.mock import patch, MagicMock + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +import gitea_config +import gitea_auth +import mcp_server +from reviewer_worktree import assess_author_worktree_continuity + +CONFIG_TEST = { + "version": 2, + "contexts": { + "ctx": { + "enabled": True, + "gitea": { + "enabled": True, + "base_url": "https://gitea.example.com" + } + } + }, + "profiles": { + "author-profile": { + "enabled": True, + "context": "ctx", + "role": "author", + "username": "author-user", + "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"}, + "allowed_operations": ["gitea.read", "gitea.issue.create", "gitea.pr.create", "gitea.branch.push", "gitea.issue.comment"], + "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"], + "execution_profile": "author-profile" + }, + "reviewer-profile": { + "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", "gitea.issue.comment"], + "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"], + "execution_profile": "reviewer-profile" + } + } +} + + +class TestOperationScopedRoles(unittest.TestCase): + def setUp(self): + self._remotes_patch = patch.dict(mcp_server.REMOTES, { + "dadeschools": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"}, + "prgs": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"} + }) + self._remotes_patch.start() + mcp_server._IDENTITY_CACHE.clear() + gitea_config._active_profile_override = None + mcp_server._MUTATION_AUTHORITY = None + self._dir = tempfile.TemporaryDirectory() + self.config_path = os.path.join(self._dir.name, "profiles.json") + self._write_config(CONFIG_TEST) + + def tearDown(self): + self._remotes_patch.stop() + mcp_server._IDENTITY_CACHE.clear() + gitea_config._active_profile_override = None + mcp_server._MUTATION_AUTHORITY = None + self._dir.cleanup() + + def _write_config(self, obj): + with open(self.config_path, "w", encoding="utf-8") as fh: + fh.write(json.dumps(obj)) + + def _env(self, profile="author-profile", with_reviewer_token=True): + env = { + "GITEA_MCP_CONFIG": self.config_path, + "GITEA_MCP_PROFILE": profile, + "GITEA_TOKEN_AUTHOR": "author-pass", + } + if with_reviewer_token: + env["GITEA_TOKEN_REVIEWER"] = "reviewer-pass" + return env + + @patch("mcp_server.api_request") + def test_auto_switch_to_reviewer(self, mock_api): + # mock identity resolution to return username matching profile + mock_api.side_effect = lambda method, url, header: ( + {"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"} + ) + with patch.dict(os.environ, self._env("author-profile")): + # initially we are author-profile + self.assertEqual(gitea_config.selected_profile_name(), "author-profile") + self.assertEqual(mcp_server.get_profile()["profile_name"], "author-profile") + + # resolve a reviewer task (review_pr) + res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs") + + # verify it automatically switched to reviewer-profile + self.assertTrue(res["allowed_in_current_session"]) + self.assertTrue(res["available_in_session"]) + self.assertEqual(res["active_profile"], "reviewer-profile") + self.assertEqual(res["active_identity"], "reviewer-user") + self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile") + + @patch("mcp_server.api_request") + def test_auto_switch_to_author(self, mock_api): + mock_api.side_effect = lambda method, url, header: ( + {"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"} + ) + with patch.dict(os.environ, self._env("reviewer-profile")): + # initially we are reviewer-profile + self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile") + + # resolve an author task (create_issue) + res = mcp_server.gitea_resolve_task_capability(task="create_issue", remote="prgs") + + # verify it automatically switched to author-profile + self.assertTrue(res["allowed_in_current_session"]) + self.assertTrue(res["available_in_session"]) + self.assertEqual(res["active_profile"], "author-profile") + self.assertEqual(res["active_identity"], "author-user") + self.assertEqual(gitea_config.selected_profile_name(), "author-profile") + + @patch("mcp_server.api_request") + def test_restart_required_when_unattached(self, mock_api): + mock_api.side_effect = lambda method, url, header: {"login": "author-user"} + # launch without reviewer token in env + with patch.dict(os.environ, self._env("author-profile", with_reviewer_token=False)): + self.assertEqual(gitea_config.selected_profile_name(), "author-profile") + + # resolve a reviewer task (review_pr) + res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs") + + # verify it did NOT switch and reports restart_required + self.assertFalse(res["allowed_in_current_session"]) + self.assertFalse(res["available_in_session"]) + self.assertTrue(res["configured"]) + self.assertTrue(res["restart_required"]) + self.assertTrue(res["stop_required"]) + self.assertIn("Reviewer profile exists but MCP server was added after session startup and is not attached", res["reason"]) + + def test_author_continuity_dirty_worktree(self): + # author is allowed to keep dirty worktree + res = assess_author_worktree_continuity({ + "task_role": "author", + "dirty_files": ["review_proofs.py"], + }) + self.assertTrue(res["allowed"]) + + # reviewer is blocked by reviewer worktree proof + res2 = assess_author_worktree_continuity({ + "task_role": "reviewer", + "worktree_path": "/repo", + "dirty_files": ["review_proofs.py"], + "pr_scope_files": ["docs/wiki/Repositories.md"], + "scratch_used": False, + }) + self.assertFalse(res2["proven"]) + + @patch("mcp_server.api_request") + def test_mutating_actions_auto_switch(self, mock_api): + mock_api.side_effect = lambda method, url, header: ( + {"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"} + ) + with patch.dict(os.environ, self._env("author-profile")): + # verify we are author-profile + self.assertEqual(gitea_config.selected_profile_name(), "author-profile") + + # call a reviewer mutation check helper (like _profile_permission_block with reviewer permission) + blocked = mcp_server._profile_permission_block("gitea.pr.merge", remote="prgs") + self.assertIsNone(blocked) # should switch to reviewer-profile and allow it (no permission block) + + # verify we dynamically switched to reviewer-profile + self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_role_session_router.py b/tests/test_role_session_router.py index 7aedd53..a2cdffb 100644 --- a/tests/test_role_session_router.py +++ b/tests/test_role_session_router.py @@ -119,6 +119,25 @@ class TestRoleSessionRouter(unittest.TestCase): ] self.assertEqual(issue_posts, []) + @patch("mcp_server.api_get_all", return_value=[]) + @patch("mcp_server.api_request", return_value={"number": 888}) + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_author_task_allowed_after_capability_resolve( + self, _auth, _api, _get_all + ): + """#228: explicit create_issue capability clears reviewer wrong_role_stop.""" + with patch.dict(os.environ, self._env("prgs-author")): + mcp_server.gitea_route_task_session(task_type="review_pr", remote="prgs") + cap = mcp_server.gitea_resolve_task_capability( + task="create_issue", remote="prgs" + ) + self.assertTrue(cap["allowed_in_current_session"]) + result = mcp_server.gitea_create_issue( + title="operation-scoped author recovery", + remote="prgs", + ) + self.assertEqual(result.get("number"), 888) + @patch("mcp_server.api_get_all", return_value=[]) @patch("mcp_server.api_request", return_value={"number": 999}) @patch("mcp_server.get_auth_header", return_value="token author-pass")