Merge pull request 'feat: gate gitea_delete_branch on gitea.branch.delete capability (Closes #408)' (#410) from feat/issue-408-delete-branch-capability-gate into master
This commit was merged in pull request #410.
This commit is contained in:
+18
-2
@@ -3464,16 +3464,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."}
|
||||
|
||||
|
||||
@@ -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()
|
||||
@@ -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"])
|
||||
|
||||
Reference in New Issue
Block a user