Merge pull request 'feat: add read-only Gitea task capability resolver tool (#141)' (#157) from feat/issue-141-task-capability-resolver into master

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 ───────────────────────────────────────────────────────────────
 

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()