diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 248b7af..4894f4d 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -3463,16 +3463,32 @@ def gitea_delete_branch( repo: Override the repository name. Returns: - dict with 'success' and 'message'. + dict with 'success' and 'message'; on a permission block, + 'success'/'performed' False, 'reasons', and a structured + 'permission_report' (#142) with no API call made. """ + gate_reasons = _profile_operation_gate("gitea.branch.delete") + if gate_reasons: + return { + "success": False, + "performed": False, + "required_permission": "gitea.branch.delete", + "reasons": gate_reasons, + "permission_report": _permission_block_report("gitea.branch.delete"), + } + verify_preflight_purity(remote) h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) import urllib.parse encoded_branch = urllib.parse.quote(branch, safe="") url = f"{repo_api_url(h, o, r)}/branches/{encoded_branch}" + request_metadata = { + "branch": branch, + "required_permission": "gitea.branch.delete", + } with _audited("delete_branch", host=h, remote=remote, org=o, repo=r, - target_branch=branch, request_metadata={"branch": branch}): + target_branch=branch, request_metadata=request_metadata): api_request("DELETE", url, auth) return {"success": True, "message": f"Remote branch '{branch}' deleted."} diff --git a/tests/test_delete_branch_capability.py b/tests/test_delete_branch_capability.py new file mode 100644 index 0000000..e67d45a --- /dev/null +++ b/tests/test_delete_branch_capability.py @@ -0,0 +1,217 @@ +"""Tests for gitea_delete_branch capability gate (Issue #408). + +``gitea_delete_branch`` requires the exact ``gitea.branch.delete`` operation: +without it the delete fails closed (no preflight, no auth lookup, no API call, +structured permission report). With it, deletion proceeds through existing +preflight and audit unchanged. +""" +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 +from mcp_server import gitea_delete_branch + +FAKE_AUTH = "token fake" + +AUTHOR_NO_DELETE = { + "profile_name": "prgs-author", + "allowed_operations": [ + "gitea.read", "gitea.pr.create", "gitea.pr.comment", + "gitea.branch.push", "gitea.issue.comment", + ], + "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"], + "audit_label": "prgs-author", +} + +AUTHOR_WITH_DELETE = { + "profile_name": "prgs-author-deleter", + "allowed_operations": AUTHOR_NO_DELETE["allowed_operations"] + [ + "gitea.branch.delete", + ], + "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"], + "audit_label": "prgs-author-deleter", +} + +CONFIG = { + "version": 2, + "contexts": { + "ctx": { + "enabled": True, + "gitea": {"enabled": True, "base_url": "https://gitea.example.com"}, + } + }, + "profiles": { + "author-no-delete": { + "enabled": True, + "context": "ctx", + "role": "author", + "username": "author-user", + "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"}, + "allowed_operations": AUTHOR_NO_DELETE["allowed_operations"], + "forbidden_operations": [], + "execution_profile": "author-no-delete", + }, + "author-with-delete": { + "enabled": True, + "context": "ctx", + "role": "author", + "username": "deleter-user", + "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"}, + "allowed_operations": AUTHOR_WITH_DELETE["allowed_operations"], + "forbidden_operations": [], + "execution_profile": "author-with-delete", + }, + "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.branch.delete", "gitea.branch.push", "gitea.pr.create", + ], + "execution_profile": "reviewer-profile", + }, + }, + "rules": {"allow_runtime_switching": False}, +} + + +class TestDeleteBranchToolGate(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() + patch("gitea_config.load_config", return_value={}).start() + patch( + "gitea_config.is_runtime_switching_enabled", return_value=False + ).start() + patch("gitea_audit.audit_enabled", return_value=False).start() + self.mock_api = patch("mcp_server.api_request").start() + self.mock_auth = patch( + "mcp_server.get_auth_header", return_value=FAKE_AUTH + ).start() + + def tearDown(self): + patch.stopall() + mcp_server._IDENTITY_CACHE.clear() + mcp_server._preflight_whoami_called, mcp_server._preflight_capability_called = ( + self._preflight_snapshot + ) + + def _set_profile(self, profile): + patch("mcp_server.get_profile", return_value=profile).start() + + def test_blocked_without_delete_capability(self): + self._set_profile(AUTHOR_NO_DELETE) + res = gitea_delete_branch(branch="feat/branch", remote="prgs") + self.assertFalse(res["success"]) + self.assertFalse(res["performed"]) + self.assertEqual(res["required_permission"], "gitea.branch.delete") + self.assertTrue(res["reasons"]) + self.assertEqual( + res["permission_report"]["missing_permission"], + "gitea.branch.delete", + ) + self.mock_api.assert_not_called() + self.mock_auth.assert_not_called() + + def test_allowed_delete_proceeds(self): + self._set_profile(AUTHOR_WITH_DELETE) + mcp_server.record_preflight_check("whoami") + mcp_server.record_preflight_check("capability", resolved_role="author") + res = gitea_delete_branch(branch="feat/branch", remote="prgs") + self.assertTrue(res["success"]) + self.assertIn("deleted", res["message"]) + delete_calls = [ + c for c in self.mock_api.call_args_list if c.args[0] == "DELETE" + ] + self.assertTrue(delete_calls) + + def test_allowed_delete_audited_with_capability_proof(self): + self._set_profile(AUTHOR_WITH_DELETE) + patch("gitea_audit.audit_enabled", return_value=True).start() + mock_write = patch("gitea_audit.write_event").start() + mcp_server.record_preflight_check("whoami") + mcp_server.record_preflight_check("capability", resolved_role="author") + self.mock_api.return_value = {} + gitea_delete_branch(branch="feat/branch", remote="prgs") + mock_write.assert_called() + event = mock_write.call_args[0][0] + self.assertEqual(event["action"], "delete_branch") + self.assertEqual( + event["request_metadata"]["required_permission"], + "gitea.branch.delete", + ) + + +class TestDeleteBranchResolverParity(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._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)) + patch( + "gitea_config.is_runtime_switching_enabled", return_value=False + ).start() + patch("gitea_audit.audit_enabled", return_value=False).start() + self.mock_api = patch("mcp_server.api_request").start() + patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start() + + def tearDown(self): + patch.stopall() + mcp_server._IDENTITY_CACHE.clear() + 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", + } + + def test_reviewer_resolver_denial_blocks_raw_tool(self): + with patch.dict(os.environ, self._env("reviewer-profile"), clear=True): + resolve = mcp_server.gitea_resolve_task_capability( + task="delete_branch", remote="prgs") + self.assertFalse(resolve["allowed_in_current_session"]) + res = gitea_delete_branch(branch="feat/branch", remote="prgs") + self.assertFalse(res["success"]) + self.assertEqual( + res["permission_report"]["missing_permission"], + resolve["required_operation_permission"], + ) + delete_calls = [ + c for c in self.mock_api.call_args_list if c.args[0] == "DELETE" + ] + self.assertFalse(delete_calls) + + +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 fcc3b64..4002b5e 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -1005,9 +1005,19 @@ class TestReviewPR(unittest.TestCase): # --------------------------------------------------------------------------- class TestDeleteBranch(unittest.TestCase): + DELETE_PROFILE = { + "profile_name": "test-deleter", + "allowed_operations": ["gitea.read", "gitea.branch.delete"], + "forbidden_operations": [], + "audit_label": "test-deleter", + } + + @patch("mcp_server.get_profile", return_value=DELETE_PROFILE) @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) - def test_delete_branch(self, _auth, mock_api): + def test_delete_branch(self, _auth, mock_api, _profile): + mcp_server.record_preflight_check("whoami") + mcp_server.record_preflight_check("capability", resolved_role="author") mock_api.return_value = {} result = gitea_delete_branch(branch="feat/branch") self.assertTrue(result["success"])