Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 74 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

Enforce issue-write tool gates to match task capability resolver (Issue #69)

Browse Source 
Extract TASK_CAPABILITY_MAP into task_capability_map.py as the single source
of truth shared by gitea_resolve_task_capability and issue-mutating tools.
Gate gitea_create_issue, gitea_close_issue, gitea_mark_issue, and
gitea_set_issue_labels with structured #142 permission reports. Add regression
tests proving resolver-denied tasks fail closed at the raw tool layer.

Implements mcp-control-plane #69.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
sysadmin
2026-07-05 21:09:34 -04:00
co-authored by Claude Opus 4.8
parent c89ae2eb5c
commit 00394d43b7
6 changed files with 413 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_mcp_server.py
+28 -13
View File
@@ -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()