diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 037a844..46353fb 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -2295,6 +2295,106 @@ def gitea_get_file( } +def _prepare_commit_payload_files(files: list[dict]) -> tuple[list[dict], list[dict]]: + import base64 + import json + import os + + processed_files = [] + source_proofs = [] + + lock_data = {} + if os.path.exists(ISSUE_LOCK_FILE): + try: + with open(ISSUE_LOCK_FILE, "r", encoding="utf-8") as f: + lock_data = json.load(f) + except Exception: + pass + + locked_worktree = lock_data.get("worktree_path") + if locked_worktree: + locked_worktree = os.path.realpath(locked_worktree) + + for f in files: + f_copy = dict(f) + path = f_copy.get("path", "") + + content_keys = [k for k in ["content", "content_plain", "workspace_path", "local_path"] if k in f_copy] + if len(content_keys) > 1: + raise ValueError( + f"Multiple content sources specified for file '{path}'; provide exactly one of " + f"'content', 'content_plain', 'workspace_path', or 'local_path' (fail closed)" + ) + + source = "none" + if "content_plain" in f_copy: + content_plain = f_copy.pop("content_plain") + if content_plain is not None: + content_bytes = content_plain.encode("utf-8") + f_copy["content"] = base64.b64encode(content_bytes).decode("utf-8") + source = "inline_plain" + elif "workspace_path" in f_copy: + workspace_path = f_copy.pop("workspace_path") + if not locked_worktree: + raise RuntimeError( + f"Issue lock is missing or does not define a worktree path. " + f"Cannot resolve workspace_path '{workspace_path}' (fail closed)." + ) + target_path = os.path.realpath(os.path.abspath(os.path.join(locked_worktree, workspace_path))) + + is_inside = target_path == locked_worktree or target_path.startswith(os.path.join(locked_worktree, "")) + is_tmp = target_path.startswith(os.path.realpath("/tmp") + os.sep) or target_path.startswith("/tmp" + os.sep) + + if not (is_inside or is_tmp): + raise ValueError( + f"workspace_path '{workspace_path}' resolves to '{target_path}' which falls outside " + f"of locked worktree '{locked_worktree}' (fail closed)" + ) + + if not os.path.exists(target_path): + raise FileNotFoundError(f"File not found: {target_path} (fail closed)") + + with open(target_path, "rb") as fh: + file_bytes = fh.read() + f_copy["content"] = base64.b64encode(file_bytes).decode("utf-8") + source = "workspace_path" + elif "local_path" in f_copy: + local_path = f_copy.pop("local_path") + if not locked_worktree: + raise RuntimeError( + f"Issue lock is missing or does not define a worktree path. " + f"Cannot resolve local_path '{local_path}' (fail closed)." + ) + target_path = os.path.realpath(os.path.abspath(local_path)) + + is_inside = target_path == locked_worktree or target_path.startswith(os.path.join(locked_worktree, "")) + is_tmp = target_path.startswith(os.path.realpath("/tmp") + os.sep) or target_path.startswith("/tmp" + os.sep) + + if not (is_inside or is_tmp): + raise ValueError( + f"local_path '{local_path}' resolves to '{target_path}' which falls outside " + f"of locked worktree '{locked_worktree}' (fail closed)" + ) + + if not os.path.exists(target_path): + raise FileNotFoundError(f"File not found: {target_path} (fail closed)") + + with open(target_path, "rb") as fh: + file_bytes = fh.read() + f_copy["content"] = base64.b64encode(file_bytes).decode("utf-8") + source = "local_path" + elif "content" in f_copy: + source = "inline_base64" + + processed_files.append(f_copy) + source_proofs.append({ + "path": path, + "source": source + }) + + return processed_files, source_proofs + + @mcp.tool() def gitea_commit_files( files: list[dict], @@ -2309,7 +2409,7 @@ def gitea_commit_files( """Commit changes to multiple files in a Gitea repository in a single atomic commit. Args: - files: List of file operations. Each file dict must contain 'operation' ('create', 'update', 'delete', 'rename'), 'path', and 'content' (base64 encoded for create/update), and optionally 'sha' (required for update/delete) or 'from_path' (for rename). + files: List of file operations. Each file dict must contain 'operation' ('create', 'update', 'delete', 'rename'), 'path', and one of content payload sources: 'content' (base64), 'content_plain', 'workspace_path', or 'local_path'. message: The commit message. branch: Optional existing branch to start/commit from. new_branch: Optional new branch name to create for this commit. @@ -2321,13 +2421,38 @@ 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) + processed_files, source_proofs = _prepare_commit_payload_files(files) + h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) url = f"{repo_api_url(h, o, r)}/contents" payload = { - "files": files, + "files": processed_files, "message": message, } if branch is not None: @@ -2338,13 +2463,14 @@ def gitea_commit_files( with _audited("commit_files", host=h, remote=remote, org=o, repo=r, target_branch=(new_branch or branch), request_metadata={"message": message, - "paths": [f.get("path") for f in files], - "operations": [f.get("operation") for f in files]}): + "paths": [f.get("path") for f in processed_files], + "operations": [f.get("operation") for f in processed_files]}): data = api_request("POST", url, auth, payload) return { "success": True, "commit": data.get("commit", {}).get("sha", ""), "branch": data.get("branch", {}).get("name", ""), + "content_source_proof": source_proofs, } 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_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_commit_payloads.py b/tests/test_commit_payloads.py new file mode 100644 index 0000000..1b3a80d --- /dev/null +++ b/tests/test_commit_payloads.py @@ -0,0 +1,251 @@ +import json +import os +import tempfile +import unittest +from unittest.mock import MagicMock, patch + +import gitea_config +import mcp_server +import task_capability_map + +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.repo.commit", + "gitea.pr.create", + "gitea.branch.push", + ], + "forbidden_operations": [ + "gitea.pr.approve", + "gitea.pr.merge", + "gitea.pr.review", + ], + "execution_profile": "full-author", + }, + }, +} + + +class TestCommitPayloads(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() + + # Setup temp directories and lock 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)) + + self.locked_worktree_dir = tempfile.TemporaryDirectory() + self.locked_worktree_path = os.path.realpath(self.locked_worktree_dir.name) + + self.lock_file_path = "/tmp/gitea_issue_lock.json" + self.lock_data = { + "issue_number": 263, + "branch_name": "feat/issue-263-native-commit-payloads", + "remote": "prgs", + "org": "Example-Org", + "repo": "Example-Repo", + "worktree_path": self.locked_worktree_path, + } + with open(self.lock_file_path, "w", encoding="utf-8") as fh: + fh.write(json.dumps(self.lock_data)) + + # Reset preflight status to bypass/pass verification in tests + self.orig_whoami_called = mcp_server._preflight_whoami_called + self.orig_capability_called = mcp_server._preflight_capability_called + mcp_server._preflight_whoami_called = True + mcp_server._preflight_capability_called = True + + 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 + + self._dir.cleanup() + self.locked_worktree_dir.cleanup() + if os.path.exists(self.lock_file_path): + os.remove(self.lock_file_path) + + def _env(self, profile: str) -> dict: + return { + "GITEA_MCP_CONFIG": self.config_path, + "GITEA_MCP_PROFILE": profile, + "GITEA_TOKEN_AUTHOR": "author-pass", + "GITEA_TEST_PORCELAIN": "", + } + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_commit_files_content_plain(self, _auth, mock_api): + mock_api.return_value = { + "commit": {"sha": "sha-123"}, + "branch": {"name": "feat/issue-263-native-commit-payloads"} + } + with patch.dict(os.environ, self._env("full-author"), clear=True): + res = mcp_server.gitea_commit_files( + files=[ + { + "operation": "create", + "path": "hello.txt", + "content_plain": "Hello World!", + } + ], + message="Add hello.txt", + remote="prgs", + ) + print("DEBUG RES IS:", res) + self.assertTrue(res["success"]) + self.assertEqual(res["commit"], "sha-123") + self.assertEqual(res["content_source_proof"][0]["source"], "inline_plain") + + # Verify posted base64 content + post_args = mock_api.call_args[0] + payload = post_args[3] + self.assertEqual(payload["files"][0]["content"], "SGVsbG8gV29ybGQh") + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_commit_files_workspace_path(self, _auth, mock_api): + mock_api.return_value = { + "commit": {"sha": "sha-456"}, + "branch": {"name": "feat/issue-263-native-commit-payloads"} + } + + # Create a workspace file + file_relative = "src/code.py" + file_abs = os.path.join(self.locked_worktree_path, file_relative) + os.makedirs(os.path.dirname(file_abs), exist_ok=True) + with open(file_abs, "wb") as fh: + fh.write(b"print('hello')") + + with patch.dict(os.environ, self._env("full-author"), clear=True): + res = mcp_server.gitea_commit_files( + files=[ + { + "operation": "create", + "path": "src/code.py", + "workspace_path": file_relative, + } + ], + message="Add code.py", + remote="prgs", + ) + self.assertTrue(res["success"]) + self.assertEqual(res["content_source_proof"][0]["source"], "workspace_path") + + post_args = mock_api.call_args[0] + payload = post_args[3] + # "print('hello')" in base64 is cHJpbnQoJ2hlbGxvJyk= + self.assertEqual(payload["files"][0]["content"], "cHJpbnQoJ2hlbGxvJyk=") + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_commit_files_local_path(self, _auth, mock_api): + mock_api.return_value = { + "commit": {"sha": "sha-789"}, + "branch": {"name": "feat/issue-263-native-commit-payloads"} + } + + file_abs = os.path.join(self.locked_worktree_path, "local.bin") + with open(file_abs, "wb") as fh: + fh.write(b"\x00\x01\x02\x03\xff") + + with patch.dict(os.environ, self._env("full-author"), clear=True): + res = mcp_server.gitea_commit_files( + files=[ + { + "operation": "create", + "path": "local.bin", + "local_path": file_abs, + } + ], + message="Add local.bin", + remote="prgs", + ) + self.assertTrue(res["success"]) + self.assertEqual(res["content_source_proof"][0]["source"], "local_path") + + post_args = mock_api.call_args[0] + payload = post_args[3] + # b"\x00\x01\x02\x03\xff" in base64 is AAECA/8= + self.assertEqual(payload["files"][0]["content"], "AAECA/8=") + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_commit_files_traversal_blocked(self, _auth, mock_api): + # Remove active lock file to ensure it fails on traversal/invalid locks + os.remove(self.lock_file_path) + + with patch.dict(os.environ, self._env("full-author"), clear=True): + with self.assertRaises(RuntimeError) as ctx: + mcp_server.gitea_commit_files( + files=[ + { + "operation": "create", + "path": "hack.txt", + "workspace_path": "any.txt", + } + ], + message="Add hack", + remote="prgs", + ) + self.assertIn("Issue lock is missing", str(ctx.exception)) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_commit_files_outside_scope_blocked(self, _auth, mock_api): + with patch.dict(os.environ, self._env("full-author"), clear=True): + with self.assertRaises(ValueError) as ctx: + mcp_server.gitea_commit_files( + files=[ + { + "operation": "create", + "path": "hack.txt", + "workspace_path": "../outside.txt", + } + ], + message="Add hack", + remote="prgs", + ) + self.assertIn("falls outside of locked worktree", str(ctx.exception)) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_commit_files_multiple_sources_blocked(self, _auth, mock_api): + with patch.dict(os.environ, self._env("full-author"), clear=True): + with self.assertRaises(ValueError) as ctx: + mcp_server.gitea_commit_files( + files=[ + { + "operation": "create", + "path": "hello.txt", + "content_plain": "Hello!", + "content": "SGVsbG8=", + } + ], + message="Add hello", + remote="prgs", + ) + self.assertIn("Multiple content sources specified", str(ctx.exception)) 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")