From 8ef9129e44156c2fc0380f272841c569e1d9728e Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Sun, 5 Jul 2026 03:21:51 -0400 Subject: [PATCH] feat: add read-only Gitea task capability resolver tool (#141) --- mcp_server.py | 146 ++++++++++++++++++++++++++ tests/test_resolve_task_capability.py | 138 ++++++++++++++++++++++++ 2 files changed, 284 insertions(+) create mode 100644 tests/test_resolve_task_capability.py diff --git a/mcp_server.py b/mcp_server.py index 3964f76..88b275d 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -2742,6 +2742,152 @@ def gitea_mirror_refs( "return_code": result.returncode, } +@mcp.tool() +def gitea_resolve_task_capability( + task: str, + remote: str = "dadeschools", + host: str | None = None, +) -> dict: + """Read-only: Resolve which capability, profile, and namespace is required for a Gitea task. + + Helps the client or LLM determine the correct namespace or profile before acting, + and returns exact next action instructions if the current session is not authorized. + + Args: + task: The task/action to check (e.g. review_pr, create_issue). + remote: Known remote instance name. + host: Optional override for the Gitea host. + """ + TASK_MAP = { + "create_issue": { + "permission": "gitea.issue.create", + "role": "author", + }, + "comment_issue": { + "permission": "gitea.issue.comment", + "role": "author", + }, + "close_issue": { + "permission": "gitea.issue.close", + "role": "author", + }, + "claim_issue": { + "permission": "gitea.issue.write", + "role": "author", + }, + "create_branch": { + "permission": "gitea.branch.create", + "role": "author", + }, + "push_branch": { + "permission": "gitea.branch.push", + "role": "author", + }, + "create_pr": { + "permission": "gitea.pr.create", + "role": "author", + }, + "comment_pr": { + "permission": "gitea.pr.comment", + "role": "author", + }, + "review_pr": { + "permission": "gitea.pr.review", + "role": "reviewer", + }, + "merge_pr": { + "permission": "gitea.pr.merge", + "role": "reviewer", + }, + "delete_branch": { + "permission": "gitea.branch.delete", + "role": "author", + }, + } + + if task not in TASK_MAP: + raise ValueError(f"Unknown task/action: '{task}' (fail closed)") + + required_permission = TASK_MAP[task]["permission"] + required_role = TASK_MAP[task]["role"] + + profile = get_profile() + config = gitea_config.load_config() + + h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None) + username = _authenticated_username(h) if h else None + + # Load active permissions + active_allowed = profile.get("allowed_operations") or [] + active_forbidden = profile.get("forbidden_operations") or [] + + # Check if allowed in current session + allowed_in_current_session, _ = gitea_config.check_operation( + required_permission, active_allowed, active_forbidden + ) + + # Find matching configured profiles + matching_profiles = [] + if config and "profiles" in config: + for p_name, p_data in config["profiles"].items(): + p_allowed = p_data.get("allowed_operations") or [] + p_forbidden = p_data.get("forbidden_operations") or [] + p_allowed_n = [] + for op in p_allowed: + try: + p_allowed_n.append(gitea_config.normalize_operation(op)) + except Exception: + pass + p_forbidden_n = [] + for op in p_forbidden: + try: + p_forbidden_n.append(gitea_config.normalize_operation(op)) + except Exception: + pass + ok, _ = gitea_config.check_operation(required_permission, p_allowed_n, p_forbidden_n) + if ok: + matching_profiles.append(p_name) + + switching = gitea_config.is_runtime_switching_enabled() + different_namespace_required = False + next_safe_action = "None; ready for operations." + + if not allowed_in_current_session: + if switching: + different_namespace_required = False + next_safe_action = f"Switch to a profile that has the required permission by calling gitea_activate_profile (matching configured profiles: {matching_profiles})." + else: + different_namespace_required = True + if required_role == "reviewer": + next_safe_action = ( + "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." + ) + elif required_role == "author": + next_safe_action = ( + "Switch to the author MCP session (e.g. gitea-author) which has author permissions configured, " + "or ask the operator to update GITEA_MCP_PROFILE to an author profile." + ) + else: + next_safe_action = ( + f"Relaunch the server with GITEA_MCP_PROFILE set to a profile that has the required permission (e.g. one of: {matching_profiles}) " + "or use the corresponding MCP namespace." + ) + + return { + "requested_task": task, + "required_operation_permission": required_permission, + "required_role_kind": required_role, + "active_profile": profile["profile_name"], + "active_identity": username, + "active_profile_allowed_operations": active_allowed, + "allowed_in_current_session": allowed_in_current_session, + "matching_configured_profile": matching_profiles, + "runtime_switching_supported": switching, + "different_mcp_namespace_required": different_namespace_required, + "exact_safe_next_action": next_safe_action, + } + # ── Entry point ─────────────────────────────────────────────────────────────── diff --git a/tests/test_resolve_task_capability.py b/tests/test_resolve_task_capability.py new file mode 100644 index 0000000..1111c10 --- /dev/null +++ b/tests/test_resolve_task_capability.py @@ -0,0 +1,138 @@ +"""Tests for gitea_resolve_task_capability tool. + +Covers Issue #141 requirements. +""" +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 + +CONFIG_RESOLVER = { + "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" + } + }, + "rules": { + "allow_runtime_switching": False + } +} + +class TestResolveTaskCapability(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 + self._dir = tempfile.TemporaryDirectory() + self.config_path = os.path.join(self._dir.name, "profiles.json") + self._write_config(CONFIG_RESOLVER) + + def tearDown(self): + self._remotes_patch.stop() + mcp_server._IDENTITY_CACHE.clear() + gitea_config._active_profile_override = 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"): + return { + "GITEA_MCP_CONFIG": self.config_path, + "GITEA_MCP_PROFILE": profile, + "GITEA_TOKEN_AUTHOR": "author-pass", + "GITEA_TOKEN_REVIEWER": "reviewer-pass", + } + + @patch("mcp_server.api_request", return_value={"login": "author-user"}) + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_resolve_author_capable_task(self, _auth, _api): + with patch.dict(os.environ, self._env("author-profile")): + res = mcp_server.gitea_resolve_task_capability(task="create_issue", remote="prgs") + self.assertEqual(res["requested_task"], "create_issue") + self.assertEqual(res["required_operation_permission"], "gitea.issue.create") + self.assertEqual(res["required_role_kind"], "author") + self.assertEqual(res["active_profile"], "author-profile") + self.assertTrue(res["allowed_in_current_session"]) + self.assertIn("author-profile", res["matching_configured_profile"]) + self.assertFalse(res["different_mcp_namespace_required"]) + self.assertIn("ready for operations", res["exact_safe_next_action"]) + + @patch("mcp_server.api_request", return_value={"login": "author-user"}) + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_resolve_reviewer_capable_task_blocked(self, _auth, _api): + with patch.dict(os.environ, self._env("author-profile")): + res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs") + self.assertEqual(res["requested_task"], "review_pr") + self.assertEqual(res["required_operation_permission"], "gitea.pr.review") + self.assertEqual(res["required_role_kind"], "reviewer") + self.assertEqual(res["active_profile"], "author-profile") + self.assertFalse(res["allowed_in_current_session"]) + self.assertIn("reviewer-profile", res["matching_configured_profile"]) + self.assertTrue(res["different_mcp_namespace_required"]) + self.assertIn("gitea-reviewer", res["exact_safe_next_action"]) + + @patch("mcp_server.api_request", return_value={"login": "reviewer-user"}) + @patch("mcp_server.get_auth_header", return_value="token reviewer-pass") + def test_resolve_reviewer_capable_task_allowed(self, _auth, _api): + with patch.dict(os.environ, self._env("reviewer-profile")): + res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs") + self.assertTrue(res["allowed_in_current_session"]) + self.assertFalse(res["different_mcp_namespace_required"]) + + @patch("mcp_server.api_request", return_value={"login": "author-user"}) + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_resolve_issue_comment_task(self, _auth, _api): + with patch.dict(os.environ, self._env("author-profile")): + res = mcp_server.gitea_resolve_task_capability(task="comment_issue", remote="prgs") + self.assertTrue(res["allowed_in_current_session"]) + self.assertIn("author-profile", res["matching_configured_profile"]) + self.assertIn("reviewer-profile", res["matching_configured_profile"]) + + def test_resolve_unknown_task_fails_closed(self): + with patch.dict(os.environ, self._env("author-profile")): + with self.assertRaises(ValueError): + mcp_server.gitea_resolve_task_capability(task="invalid_task_xyz", remote="prgs") + +if __name__ == "__main__": + unittest.main()