Merge pull request 'Enforce issue-write tool gates to match task capability resolver (mcp-control-plane #69)' (#222) from feat/issue-69-issue-write-tool-gates into master

2026-07-06 00:14:40 -05:00
parent e727210aad 00394d43b7
commit 25c19b26d3
6 changed files with 413 additions and 100 deletions
@@ -162,7 +162,8 @@ class _AuditWiringBase(unittest.TestCase):
    def _env(self, **extra):
        env = {"GITEA_AUDIT_LOG": self.audit_path,
               "GITEA_PROFILE_NAME": "gitea-author",
-               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
+               "GITEA_ALLOWED_OPERATIONS": (
+                   "read,merge,gitea.issue.create,gitea.issue.close")}
        env.update(extra)
        return env

@@ -237,7 +238,10 @@ class TestSimpleToolAudit(_AuditWiringBase):
    def test_disabled_writes_nothing_and_no_extra_call(self, _auth, _get_all, mock_api, _role):
        # No GITEA_AUDIT_LOG -> audit is a no-op: one create POST, no file.
        mock_api.return_value = {"number": 1, "html_url": "http://x/1"}
-        with patch.dict(os.environ, {"GITEA_PROFILE_NAME": "gitea-author"}, clear=True):
+        with patch.dict(os.environ, {
+            "GITEA_PROFILE_NAME": "gitea-author",
+            "GITEA_ALLOWED_OPERATIONS": "gitea.issue.create",
+        }, clear=True):
            gitea_create_issue(title="x", remote="prgs")
        issue_posts = [
            c for c in mock_api.call_args_list if c.args[0] == "POST"
@@ -19,6 +19,9 @@ from mcp_server import gitea_create_issue  # noqa: E402
 from review_proofs import assess_duplicate_search_proof as proof_assess  # noqa: E402

 FAKE_AUTH = "Basic dGVzdDp0ZXN0"
+ISSUE_WRITE_ENV = {
+    "GITEA_ALLOWED_OPERATIONS": "gitea.issue.create",
+}
 CANONICAL_TITLE = (
    "Add hard queue-target resolution wall before PR inventory "
    "or empty-queue claims"
@@ -118,7 +121,8 @@ class TestCreateIssueMCPGate(unittest.TestCase):
        mock_get_all.return_value = [
            {"number": 201, "title": CANONICAL_TITLE, "state": "open"},
        ]
-        result = gitea_create_issue(title=CANONICAL_TITLE.upper(), remote="prgs")
+        with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True):
+            result = gitea_create_issue(title=CANONICAL_TITLE.upper(), remote="prgs")
        self.assertFalse(result["performed"])
        self.assertEqual(result["duplicate_gate"], VERDICT_DUPLICATE)
        mock_api.assert_not_called()
@@ -135,13 +139,14 @@ class TestCreateIssueMCPGate(unittest.TestCase):
            {"number": 201, "title": CANONICAL_TITLE, "state": "open"},
        ]
        mock_api.return_value = {"number": 210, "html_url": "https://gitea.prgs.cc/issues/210"}
-        result = gitea_create_issue(
-            title=CANONICAL_TITLE,
-            body="Follow-up scope",
-            remote="prgs",
-            allow_duplicate_override=True,
-            split_from_issue=201,
-        )
+        with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True):
+            result = gitea_create_issue(
+                title=CANONICAL_TITLE,
+                body="Follow-up scope",
+                remote="prgs",
+                allow_duplicate_override=True,
+                split_from_issue=201,
+            )
        self.assertEqual(result["number"], 210)
        payload = mock_api.call_args[0][3]
        self.assertIn("Operator-approved split from #201.", payload["body"])
@@ -156,7 +161,8 @@ class TestCreateIssueMCPGate(unittest.TestCase):
    ):
        mock_role_check.return_value = (True, [])
        mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
-        result = gitea_create_issue(title="Unique new issue", body="body text")
+        with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True):
+            result = gitea_create_issue(title="Unique new issue", body="body text")
        self.assertEqual(result["number"], 1)


@@ -0,0 +1,210 @@
+"""Regression tests: issue-write tool gates match gitea_resolve_task_capability (#69)."""
+import json
+import os
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import mcp_server
+import task_capability_map
+
+CONFIG = {
+    "version": 2,
+    "contexts": {
+        "ctx": {
+            "enabled": True,
+            "gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
+        }
+    },
+    "profiles": {
+        "comment-only": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "author",
+            "username": "commenter",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
+            "allowed_operations": ["gitea.read", "gitea.issue.comment"],
+            "forbidden_operations": [
+                "gitea.issue.create", "gitea.issue.close"],
+            "execution_profile": "comment-only",
+        },
+        "full-author": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "author",
+            "username": "author-user",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
+            "allowed_operations": [
+                "gitea.read", "gitea.issue.create", "gitea.issue.close",
+                "gitea.issue.comment"],
+            "forbidden_operations": [],
+            "execution_profile": "full-author",
+        },
+    },
+    "rules": {"allow_runtime_switching": False},
+}
+
+ISSUE_WRITE_TASKS = (
+    "create_issue", "close_issue", "comment_issue", "mark_issue",
+    "set_issue_labels",
+)
+
+TOOL_CALLS = {
+    "create_issue": lambda: mcp_server.gitea_create_issue(
+        title="New issue", remote="prgs"),
+    "close_issue": lambda: mcp_server.gitea_close_issue(
+        issue_number=9, remote="prgs"),
+    "comment_issue": lambda: mcp_server.gitea_create_issue_comment(
+        issue_number=9, body="hello", remote="prgs"),
+    "mark_issue": lambda: mcp_server.gitea_mark_issue(
+        issue_number=9, action="start", remote="prgs"),
+    "set_issue_labels": lambda: mcp_server.gitea_set_issue_labels(
+        issue_number=9, labels=["bug"], remote="prgs"),
+}
+
+
+class IssueWriteGateBase(unittest.TestCase):
+    def setUp(self):
+        self._remotes = patch.dict(mcp_server.REMOTES, {
+            "prgs": {"host": "gitea.example.com", "org": "Example-Org",
+                     "repo": "Example-Repo"},
+        })
+        self._remotes.start()
+        mcp_server._IDENTITY_CACHE.clear()
+        self._dir = tempfile.TemporaryDirectory()
+        self.config_path = os.path.join(self._dir.name, "profiles.json")
+        with open(self.config_path, "w", encoding="utf-8") as fh:
+            fh.write(json.dumps(CONFIG))
+
+    def tearDown(self):
+        self._remotes.stop()
+        mcp_server._IDENTITY_CACHE.clear()
+        self._dir.cleanup()
+
+    def _env(self, profile: str) -> dict:
+        return {
+            "GITEA_MCP_CONFIG": self.config_path,
+            "GITEA_MCP_PROFILE": profile,
+            "GITEA_TOKEN_AUTHOR": "author-pass",
+        }
+
+
+class TestResolverToolAlignment(unittest.TestCase):
+    def test_issue_mutation_tools_match_resolver_permissions(self):
+        for tool_name, task_key in (
+            task_capability_map.ISSUE_MUTATION_TOOL_TASKS.items()
+        ):
+            self.assertEqual(
+                task_capability_map.tool_required_permission(tool_name),
+                task_capability_map.required_permission(task_key),
+                msg=f"{tool_name} gate drift from resolver task {task_key}",
+            )
+
+
+class TestDeniedIssueWritesFailClosed(IssueWriteGateBase):
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.api_get_all", return_value=[])
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_resolver_denied_tasks_block_raw_tools(
+        self, _auth, mock_api, _get_all, _role
+    ):
+        mock_api.return_value = {"number": 1}
+        with patch.dict(os.environ, self._env("comment-only"), clear=True):
+            for task in ISSUE_WRITE_TASKS:
+                with self.subTest(task=task):
+                    resolve = mcp_server.gitea_resolve_task_capability(
+                        task=task, remote="prgs")
+                    if resolve["allowed_in_current_session"]:
+                        continue
+                    result = TOOL_CALLS[task]()
+                    self.assertFalse(result.get("success", True), task)
+                    self.assertFalse(result.get("performed", True), task)
+                    self.assertTrue(result.get("reasons"), task)
+                    self.assertIn("permission_report", result, task)
+                    self.assertEqual(
+                        result["permission_report"]["missing_permission"],
+                        resolve["required_operation_permission"],
+                    )
+
+
+class TestAllowedIssueWritesProceed(IssueWriteGateBase):
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.api_get_all", return_value=[])
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_resolver_allowed_create_issue_proceeds(
+        self, _auth, mock_api, _get_all, _role
+    ):
+        mock_api.return_value = {"number": 42, "html_url": "https://x/42"}
+        with patch.dict(os.environ, self._env("full-author"), clear=True):
+            resolve = mcp_server.gitea_resolve_task_capability(
+                task="create_issue", remote="prgs")
+            self.assertTrue(resolve["allowed_in_current_session"])
+            result = mcp_server.gitea_create_issue(title="Allowed", remote="prgs")
+        self.assertEqual(result.get("number"), 42)
+        post_calls = [
+            c for c in mock_api.call_args_list if c.args[0] == "POST"
+        ]
+        self.assertTrue(post_calls)
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_resolver_allowed_close_issue_proceeds(self, _auth, mock_api):
+        mock_api.return_value = {"state": "closed"}
+        with patch.dict(os.environ, self._env("full-author"), clear=True):
+            resolve = mcp_server.gitea_resolve_task_capability(
+                task="close_issue", remote="prgs")
+            self.assertTrue(resolve["allowed_in_current_session"])
+            result = mcp_server.gitea_close_issue(issue_number=9, remote="prgs")
+        self.assertTrue(result.get("success"))
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_resolver_allowed_mark_issue_proceeds(self, _auth, mock_api):
+        def api_side_effect(method, url, auth, payload=None):
+            if method == "GET" and url.endswith("/labels?limit=100"):
+                return [{"id": 10, "name": "status:in-progress"}]
+            if method == "POST" and "/issues/9/labels" in url:
+                return [{"name": "status:in-progress"}]
+            if method == "GET" and url.endswith("/user"):
+                return {"login": "author-user"}
+            return {}
+        mock_api.side_effect = api_side_effect
+        with patch.dict(os.environ, self._env("full-author"), clear=True):
+            resolve = mcp_server.gitea_resolve_task_capability(
+                task="mark_issue", remote="prgs")
+            self.assertTrue(resolve["allowed_in_current_session"])
+            result = mcp_server.gitea_mark_issue(
+                issue_number=9, action="start", remote="prgs")
+        self.assertTrue(result.get("success"))
+
+
+class TestCreateIssueMissingPermissionReport(IssueWriteGateBase):
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.api_get_all", return_value=[])
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_create_issue_without_create_permission_blocked(
+        self, _auth, mock_api, _get_all, _role
+    ):
+        with patch.dict(os.environ, self._env("comment-only"), clear=True):
+            result = mcp_server.gitea_create_issue(title="Blocked", remote="prgs")
+        self.assertFalse(result["success"])
+        self.assertFalse(result["performed"])
+        self.assertIsNone(result["number"])
+        self.assertEqual(
+            result["permission_report"]["missing_permission"],
+            "gitea.issue.create",
+        )
+        mock_api.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -47,6 +47,13 @@ import mcp_server

 FAKE_AUTH = "Basic dGVzdDp0ZXN0"

+# Issue-write tools are profile-gated (#69).
+ISSUE_WRITE_ENV = {
+    "GITEA_ALLOWED_OPERATIONS": (
+        "gitea.issue.create,gitea.issue.close,gitea.issue.comment"
+    ),
+}
+

 # ---------------------------------------------------------------------------
 # Create Issue
@@ -60,7 +67,7 @@ class TestCreateIssue(unittest.TestCase):
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_creates_issue(self, _auth, _get_all, mock_api, _role):
        mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
-        with patch.dict(os.environ, {}, clear=True):
+        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
            result = gitea_create_issue(title="Test issue", body="body text")
        self.assertEqual(result["number"], 1)
        self.assertNotIn("url", result)
@@ -77,7 +84,8 @@ class TestCreateIssue(unittest.TestCase):
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_issue_reveal_opt_in_includes_url(self, _auth, _get_all, mock_api, _role):
        mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
-        with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
+        env = {**ISSUE_WRITE_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
+        with patch.dict(os.environ, env, clear=True):
            result = gitea_create_issue(title="Test issue", body="body text")
        self.assertIn("issues/1", result["url"])

@@ -88,7 +96,8 @@ class TestCreateIssue(unittest.TestCase):
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_creates_on_prgs(self, _auth, _get_all, mock_api, _role):
        mock_api.return_value = {"number": 5, "html_url": "https://gitea.prgs.cc/issues/5"}
-        result = gitea_create_issue(title="Test", remote="prgs")
+        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
+            result = gitea_create_issue(title="Test", remote="prgs")
        self.assertEqual(result["number"], 5)
        url = mock_api.call_args[0][1]
        self.assertIn("gitea.prgs.cc", url)
@@ -136,7 +145,8 @@ class TestCloseIssue(unittest.TestCase):
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_closes_issue(self, _auth, mock_api):
        mock_api.return_value = {"state": "closed"}
-        result = gitea_close_issue(issue_number=42)
+        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
+            result = gitea_close_issue(issue_number=42)
        self.assertTrue(result["success"])
        self.assertIn("42", result["message"])
        patch_call = next(call for call in mock_api.call_args_list if call[0][0] == "PATCH")
@@ -235,7 +245,8 @@ class TestMarkIssue(unittest.TestCase):
            [{"id": 10, "name": "status:in-progress"}],
            [{"name": "status:in-progress"}],
        ]
-        result = gitea_mark_issue(issue_number=5, action="start")
+        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
+            result = gitea_mark_issue(issue_number=5, action="start")
        self.assertTrue(result["success"])
        self.assertIn("claimed", result["message"])

@@ -246,7 +257,8 @@ class TestMarkIssue(unittest.TestCase):
            [{"id": 10, "name": "status:in-progress"}],
            None,
        ]
-        result = gitea_mark_issue(issue_number=5, action="done")
+        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
+            result = gitea_mark_issue(issue_number=5, action="done")
        self.assertTrue(result["success"])
        self.assertIn("released", result["message"])

@@ -258,8 +270,9 @@ class TestMarkIssue(unittest.TestCase):
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_missing_label_raises(self, _auth, mock_api):
        mock_api.return_value = []  # no labels exist
-        with self.assertRaises(RuntimeError):
-            gitea_mark_issue(issue_number=5, action="start")
+        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
+            with self.assertRaises(RuntimeError):
+                gitea_mark_issue(issue_number=5, action="start")


 # ---------------------------------------------------------------------------
@@ -269,12 +282,14 @@ class TestAuthErrors(unittest.TestCase):

    @patch("mcp_server.get_auth_header", return_value=None)
    def test_no_credentials_raises(self, _auth):
-        with self.assertRaises(RuntimeError):
-            gitea_create_issue(title="test")
+        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
+            with self.assertRaises(RuntimeError):
+                gitea_create_issue(title="test")

    def test_unknown_remote_raises(self):
-        with self.assertRaises(ValueError):
-            gitea_create_issue(title="test", remote="nonexistent")
+        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
+            with self.assertRaises(ValueError):
+                gitea_create_issue(title="test", remote="nonexistent")


 # ---------------------------------------------------------------------------
@@ -1939,7 +1954,7 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
        patch("gitea_audit.audit_enabled", return_value=True).start()
        self.mock_audit = patch("gitea_audit.write_event").start()
        # gitea.pr.close: closing a PR via gitea_edit_pr is capability-gated (#216).
-        patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["merge", "edit", "close", "gitea.pr.close"], "audit_label": "test", "forbidden_operations": []}).start()
+        patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["merge", "edit", "close", "gitea.pr.close", "gitea.issue.close"], "audit_label": "test", "forbidden_operations": []}).start()

    def tearDown(self):
        patch.stopall()