diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index dd1480c..60e65fb 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -2301,6 +2301,29 @@ 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_gate.py b/tests/test_commit_files_gate.py new file mode 100644 index 0000000..4c72d46 --- /dev/null +++ b/tests/test_commit_files_gate.py @@ -0,0 +1,251 @@ +"""Regression tests: gitea_commit_files tool gates match resolver and whoami verification. + +Covers Issue #262 requirements. +""" +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 +import gitea_config + +CONFIG = { + "version": 2, + "contexts": { + "ctx": { + "enabled": True, + "gitea": {"enabled": True, "base_url": "https://gitea.example.com"}, + } + }, + "profiles": { + "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.repo.commit" + ], + "forbidden_operations": [], + "execution_profile": "full-author", + }, + "reviewer-no-commit": { + "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" + ], + "forbidden_operations": [ + "gitea.repo.commit", "gitea.pr.create", "gitea.branch.push" + ], + "execution_profile": "reviewer-no-commit", + }, + }, + "rules": {"allow_runtime_switching": False}, +} + + +class TestCommitFilesGate(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() + gitea_config._active_profile_override = None + 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() + gitea_config._active_profile_override = None + 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", + } + + @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": "abc123commit"}, + "branch": {"name": "some-branch"}, + } + with patch.dict(os.environ, self._env("full-author"), clear=True): + resolve = mcp_server.gitea_resolve_task_capability( + task="commit_files", remote="prgs" + ) + self.assertTrue(resolve["allowed_in_current_session"]) + + res = mcp_server.gitea_commit_files( + files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}], + message="Add x", + remote="prgs", + ) + self.assertTrue(res["success"]) + self.assertEqual(res["commit"], "abc123commit") + + @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 reviewer-pass") + def test_denied_reviewer_blocked(self, _auth, mock_api, _role): + mock_api.return_value = {"login": "reviewer-user"} + with patch.dict(os.environ, self._env("reviewer-no-commit"), clear=True): + resolve = mcp_server.gitea_resolve_task_capability( + task="commit_files", remote="prgs" + ) + self.assertFalse(resolve["allowed_in_current_session"]) + + res = mcp_server.gitea_commit_files( + files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}], + message="Add x", + 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" + ) + post_calls = [c for c in mock_api.call_args_list if len(c.args) > 0 and c.args[0] == "POST"] + self.assertFalse(post_calls) + + @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 reviewer-pass") + def test_unknown_profile_fails_closed(self, _auth, mock_api, _role): + mock_api.return_value = {"login": "reviewer-user"} + with patch.dict(os.environ, self._env("non-existent"), clear=True): + res = mcp_server.gitea_commit_files( + files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}], + message="Add x", + remote="prgs", + ) + self.assertFalse(res["success"]) + self.assertFalse(res.get("performed", True)) + post_calls = [c for c in mock_api.call_args_list if len(c.args) > 0 and c.args[0] == "POST"] + self.assertFalse(post_calls) + + +class TestPreflightCommitFilesGate(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.orig_whoami_called = mcp_server._preflight_whoami_called + self.orig_capability_called = mcp_server._preflight_capability_called + self.orig_whoami_violation = mcp_server._preflight_whoami_violation + self.orig_capability_violation = mcp_server._preflight_capability_violation + self.orig_resolved_role = mcp_server._preflight_resolved_role + self.orig_process_start = mcp_server._process_start_porcelain + self.orig_whoami_baseline = mcp_server._preflight_whoami_baseline_porcelain + self.orig_capability_baseline = mcp_server._preflight_capability_baseline_porcelain + self.orig_whoami_files = mcp_server._preflight_whoami_violation_files + self.orig_capability_files = mcp_server._preflight_capability_violation_files + self.orig_reviewer_files = mcp_server._preflight_reviewer_violation_files + + mcp_server._preflight_whoami_called = False + mcp_server._preflight_capability_called = False + mcp_server._preflight_whoami_violation = False + mcp_server._preflight_capability_violation = False + mcp_server._preflight_resolved_role = None + mcp_server._process_start_porcelain = "" + mcp_server._preflight_whoami_baseline_porcelain = None + mcp_server._preflight_capability_baseline_porcelain = None + mcp_server._preflight_whoami_violation_files = [] + mcp_server._preflight_capability_violation_files = [] + mcp_server._preflight_reviewer_violation_files = [] + + 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 = self.orig_whoami_called + mcp_server._preflight_capability_called = self.orig_capability_called + mcp_server._preflight_whoami_violation = self.orig_whoami_violation + mcp_server._preflight_capability_violation = self.orig_capability_violation + mcp_server._preflight_resolved_role = self.orig_resolved_role + mcp_server._process_start_porcelain = self.orig_process_start + mcp_server._preflight_whoami_baseline_porcelain = self.orig_whoami_baseline + mcp_server._preflight_capability_baseline_porcelain = self.orig_capability_baseline + mcp_server._preflight_whoami_violation_files = self.orig_whoami_files + mcp_server._preflight_capability_violation_files = self.orig_capability_files + mcp_server._preflight_reviewer_violation_files = self.orig_reviewer_files + + self._dir.cleanup() + os.environ.pop("GITEA_TEST_FORCE_DIRTY", None) + os.environ.pop("GITEA_TEST_PORCELAIN", None) + + def _env(self, profile: str) -> dict: + return { + "GITEA_MCP_CONFIG": self.config_path, + "GITEA_MCP_PROFILE": profile, + "GITEA_TOKEN_AUTHOR": "author-pass", + } + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_preflight_not_called_blocks_commit(self, _auth, mock_api): + mock_api.return_value = {"login": "author-user"} + with patch.dict(os.environ, self._env("full-author"), clear=True): + os.environ["GITEA_TEST_PORCELAIN"] = "" + with self.assertRaises(RuntimeError) as ctx: + mcp_server.gitea_commit_files( + files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}], + message="Add x", + remote="prgs", + ) + self.assertIn("Identity (gitea_whoami) has not been verified", str(ctx.exception)) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_dirty_workspace_before_whoami_blocks_commit(self, _auth, mock_api): + mock_api.return_value = {"login": "author-user"} + with patch.dict(os.environ, self._env("full-author"), clear=True): + os.environ["GITEA_TEST_FORCE_DIRTY"] = "1" + mcp_server.record_preflight_check("whoami") + mcp_server.record_preflight_check("capability", resolved_role="author") + + with self.assertRaises(RuntimeError) as ctx: + mcp_server.gitea_commit_files( + files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}], + message="Add x", + remote="prgs", + ) + self.assertIn("Workspace file edits occurred before gitea_whoami verification", str(ctx.exception)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 133e515..58dc7a9 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -1071,9 +1071,11 @@ class TestGetFile(unittest.TestCase): # --------------------------------------------------------------------------- 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): mock_api.return_value = { "commit": {"sha": "commit-sha-123"}, "branch": {"name": "test-branch"} @@ -1081,11 +1083,15 @@ 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" - ) + env = { + "GITEA_ALLOWED_OPERATIONS": "gitea.repo.commit", + } + with patch.dict(os.environ, env, clear=True): + 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")