diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index dd1480c..09bd395 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -2301,6 +2301,30 @@ def gitea_commit_files( Returns: dict with success status and commit/branch information. """ + ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop( + "commit_files" + ) + if not ok: + return { + "success": False, + "performed": False, + "commit": "", + "branch": "", + "reasons": block_reasons, + } + blocked = _namespace_mutation_block( + "commit_files", commit="", branch="", remote=remote + ) + if blocked: + return blocked + blocked = _profile_permission_block( + task_capability_map.required_permission("commit_files"), + commit="", + branch="", + ) + if blocked: + return blocked + verify_preflight_purity(remote) h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) diff --git a/task_capability_map.py b/task_capability_map.py index fd3c344..a485491 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -80,6 +80,14 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.branch.delete", "role": "author", }, + "commit_files": { + "permission": "gitea.repo.commit", + "role": "author", + }, + "gitea_commit_files": { + "permission": "gitea.repo.commit", + "role": "author", + }, } # Issue-mutating MCP tools and their resolver task keys. @@ -89,6 +97,7 @@ ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = { "gitea_create_issue_comment": "comment_issue", "gitea_mark_issue": "mark_issue", "gitea_set_issue_labels": "set_issue_labels", + "gitea_commit_files": "commit_files", } diff --git a/tests/test_commit_files_capability.py b/tests/test_commit_files_capability.py new file mode 100644 index 0000000..d1203bd --- /dev/null +++ b/tests/test_commit_files_capability.py @@ -0,0 +1,199 @@ +"""Tests for commit_files task capability resolver mapping (#262).""" +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": { + "commit-author": { + "enabled": True, + "context": "ctx", + "role": "author", + "username": "author-user", + "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"}, + "allowed_operations": ["gitea.read", "gitea.repo.commit"], + "forbidden_operations": ["gitea.pr.create"], + "execution_profile": "commit-author", + }, + "pr-only-author": { + "enabled": True, + "context": "ctx", + "role": "author", + "username": "pr-author", + "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"}, + "allowed_operations": ["gitea.read", "gitea.pr.create"], + "forbidden_operations": ["gitea.repo.commit"], + "execution_profile": "pr-only-author", + }, + "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.merge"], + "forbidden_operations": [ + "gitea.repo.commit", "gitea.pr.create", "gitea.branch.push", + ], + "execution_profile": "reviewer-profile", + }, + }, + "rules": {"allow_runtime_switching": False}, +} + + +class CommitFilesCapabilityBase(unittest.TestCase): + def setUp(self): + self._preflight_snapshot = ( + mcp_server._preflight_whoami_called, + mcp_server._preflight_capability_called, + ) + mcp_server._preflight_whoami_called = False + mcp_server._preflight_capability_called = False + 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() + mcp_server._preflight_whoami_called, mcp_server._preflight_capability_called = ( + self._preflight_snapshot + ) + 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", + "GITEA_TOKEN_REVIEWER": "reviewer-pass", + } + + +class TestCommitFilesResolver(CommitFilesCapabilityBase): + @patch("mcp_server.api_request", return_value={"login": "author-user"}) + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_commit_files_resolves_to_repo_commit(self, _auth, _api): + with patch.dict(os.environ, self._env("commit-author"), clear=True): + for task in ("commit_files", "gitea_commit_files"): + with self.subTest(task=task): + res = mcp_server.gitea_resolve_task_capability( + task=task, remote="prgs") + self.assertEqual(res["required_operation_permission"], + "gitea.repo.commit") + self.assertEqual(res["required_role_kind"], "author") + self.assertTrue(res["allowed_in_current_session"]) + + @patch("mcp_server.api_request", return_value={"login": "pr-author"}) + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_create_pr_does_not_satisfy_commit_files(self, _auth, _api): + with patch.dict(os.environ, self._env("pr-only-author"), clear=True): + commit_res = mcp_server.gitea_resolve_task_capability( + task="commit_files", remote="prgs") + pr_res = mcp_server.gitea_resolve_task_capability( + task="create_pr", remote="prgs") + self.assertFalse(commit_res["allowed_in_current_session"]) + self.assertTrue(pr_res["allowed_in_current_session"]) + self.assertEqual(commit_res["required_operation_permission"], + "gitea.repo.commit") + self.assertEqual(pr_res["required_operation_permission"], + "gitea.pr.create") + + @patch("mcp_server.api_request", return_value={"login": "reviewer-user"}) + @patch("mcp_server.get_auth_header", return_value="token reviewer-pass") + def test_reviewer_denied_commit_files(self, _auth, _api): + with patch.dict(os.environ, self._env("reviewer-profile"), clear=True): + res = mcp_server.gitea_resolve_task_capability( + task="commit_files", remote="prgs") + self.assertFalse(res["allowed_in_current_session"]) + self.assertEqual(res["required_operation_permission"], + "gitea.repo.commit") + self.assertTrue(res["different_mcp_namespace_required"]) + + +class TestCommitFilesToolGate(CommitFilesCapabilityBase): + @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", + return_value=(True, [])) + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_allowed_author_proceeds(self, _auth, mock_api, _role): + mock_api.return_value = { + "commit": {"sha": "abc"}, + "branch": {"name": "feat/x"}, + } + mcp_server.record_preflight_check("whoami") + mcp_server.record_preflight_check("capability", resolved_role="author") + with patch.dict(os.environ, self._env("commit-author"), clear=True): + res = mcp_server.gitea_commit_files( + files=[{"operation": "create", "path": "a.txt", "content": "YQ=="}], + message="test", + new_branch="feat/x", + remote="prgs", + ) + self.assertTrue(res["success"]) + + @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", + return_value=(True, [])) + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_reviewer_blocked_with_permission_report(self, _auth, _role): + mcp_server.record_preflight_check("whoami") + mcp_server.record_preflight_check("capability", resolved_role="author") + with patch.dict(os.environ, self._env("reviewer-profile"), clear=True): + res = mcp_server.gitea_commit_files( + files=[{"operation": "create", "path": "a.txt", "content": "YQ=="}], + message="test", + remote="prgs", + ) + self.assertFalse(res["success"]) + self.assertFalse(res.get("performed", True)) + self.assertIn("permission_report", res) + self.assertEqual(res["permission_report"]["missing_permission"], + "gitea.repo.commit") + + @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", + return_value=(True, [])) + def test_preflight_blocks_before_api(self, _role): + env = {**self._env("commit-author"), "GITEA_TEST_FORCE_DIRTY": "1"} + with patch.dict(os.environ, env, clear=True): + with self.assertRaises(RuntimeError) as ctx: + mcp_server.gitea_commit_files( + files=[{"operation": "create", "path": "a.txt", "content": "YQ=="}], + message="test", + remote="prgs", + ) + self.assertIn("gitea_whoami", str(ctx.exception)) + + +class TestCommitFilesMapAlignment(unittest.TestCase): + def test_tool_gate_matches_resolver(self): + self.assertEqual( + task_capability_map.tool_required_permission("gitea_commit_files"), + task_capability_map.required_permission("commit_files"), + ) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 133e515..d67499c 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -1069,11 +1069,22 @@ class TestGetFile(unittest.TestCase): # --------------------------------------------------------------------------- # Commit Files # --------------------------------------------------------------------------- +COMMIT_FILES_ENV = { + "GITEA_PROFILE_NAME": "commit-author", + "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.repo.commit", + "GITEA_FORBIDDEN_OPERATIONS": "gitea.pr.approve,gitea.pr.merge", +} + + class TestCommitFiles(unittest.TestCase): + @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", + return_value=(True, [])) @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) - def test_commit_files_success(self, _auth, mock_api): + def test_commit_files_success(self, _auth, mock_api, _role): + mcp_server.record_preflight_check("whoami") + mcp_server.record_preflight_check("capability", resolved_role="author") mock_api.return_value = { "commit": {"sha": "commit-sha-123"}, "branch": {"name": "test-branch"} @@ -1081,11 +1092,12 @@ class TestCommitFiles(unittest.TestCase): files = [ {"operation": "create", "path": "test.txt", "content": "SGVsbG8="} ] - result = gitea_commit_files( - files=files, - message="Initial commit", - new_branch="test-branch" - ) + with patch.dict(os.environ, COMMIT_FILES_ENV, clear=False): + result = gitea_commit_files( + files=files, + message="Initial commit", + new_branch="test-branch" + ) self.assertTrue(result["success"]) self.assertEqual(result["commit"], "commit-sha-123") self.assertEqual(result["branch"], "test-branch")