diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 09bd395..60e65fb 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -2319,8 +2319,7 @@ def gitea_commit_files( return blocked blocked = _profile_permission_block( task_capability_map.required_permission("commit_files"), - commit="", - branch="", + commit="", branch="", ) if blocked: return blocked 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 d67499c..58dc7a9 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -1069,13 +1069,6 @@ 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", @@ -1083,8 +1076,6 @@ class TestCommitFiles(unittest.TestCase): @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) 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"} @@ -1092,7 +1083,10 @@ class TestCommitFiles(unittest.TestCase): files = [ {"operation": "create", "path": "test.txt", "content": "SGVsbG8="} ] - with patch.dict(os.environ, COMMIT_FILES_ENV, clear=False): + env = { + "GITEA_ALLOWED_OPERATIONS": "gitea.repo.commit", + } + with patch.dict(os.environ, env, clear=True): result = gitea_commit_files( files=files, message="Initial commit",