fix(mcp): fail closed when merger lease cannot resolve live PR head

gitea_acquire_merger_pr_lease previously continued when GET /pulls returned
an empty head SHA. Fail closed so exact-head scoping never proceeds with an
unresolvable live head. Add regression coverage for empty head payloads.

Fixes #718
Refs #723
This commit is contained in:
2026-07-17 13:25:34 -04:00
parent 4a160e0e0c
commit 29c2cf2ef6
2 changed files with 60 additions and 1 deletions
+14 -1
View File
@@ -10821,7 +10821,20 @@ def gitea_acquire_merger_pr_lease(
or "" or ""
) )
live_head = str(live_head).strip() live_head = str(live_head).strip()
if live_head and live_head != head_pin: # Fail closed when live PR head cannot be resolved — never proceed with
# an unpinned/unknown head even if candidate_head was supplied (#718).
if not live_head:
return {
"success": False,
"acquired": False,
"reasons": [
"live PR head could not be resolved from GET /pulls response "
"(exact-head scoping; fail closed)"
],
"live_head": None,
"candidate_head": head_pin,
}
if live_head != head_pin:
return { return {
"success": False, "success": False,
"acquired": False, "acquired": False,
+46
View File
@@ -176,6 +176,52 @@ class TestMergerLeaseAcquireTool(unittest.TestCase):
self.assertTrue(any("does not match live PR head" in r for r in res["reasons"])) self.assertTrue(any("does not match live PR head" in r for r in res["reasons"]))
self.assertIsNone(leases._SESSION_LEASE) self.assertIsNone(leases._SESSION_LEASE)
@patch("gitea_mcp_server.api_request")
@patch("gitea_mcp_server._auth", return_value="token pass")
@patch("gitea_mcp_server._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"))
@patch("gitea_mcp_server.get_profile")
@patch("gitea_mcp_server._fetch_pr_comments", return_value=[])
@patch("gitea_mcp_server._verify_role_mutation_workspace")
def test_acquire_merger_fails_closed_when_live_head_unresolvable(
self, _verify, _comments, mock_profile, _resolve, _auth, mock_api
):
"""Empty/missing live head must not silently proceed past exact-head gate."""
mock_profile.return_value = self._merger_profile()
# PR payload present but head.sha missing / empty / non-dict.
mock_api.return_value = {"state": "open", "head": {}}
res = mcp_server.gitea_acquire_merger_pr_lease(
pr_number=718,
worktree="branches/issue-718",
session_id="merger-session",
candidate_head=HEAD,
)
self.assertFalse(res["success"], res)
self.assertFalse(res.get("acquired"), res)
self.assertTrue(
any("live PR head could not be resolved" in r for r in res["reasons"]),
res,
)
self.assertIsNone(res.get("live_head"))
self.assertEqual(res.get("candidate_head"), HEAD)
self.assertEqual(
[c for c in mock_api.call_args_list if c[0][0] == "POST"], []
)
self.assertIsNone(leases._SESSION_LEASE)
# Completely empty GET /pulls body also fails closed.
mock_api.return_value = {}
res2 = mcp_server.gitea_acquire_merger_pr_lease(
pr_number=718,
worktree="branches/issue-718",
candidate_head=HEAD,
)
self.assertFalse(res2["success"], res2)
self.assertTrue(
any("live PR head could not be resolved" in r for r in res2["reasons"]),
res2,
)
self.assertIsNone(leases._SESSION_LEASE)
@patch("gitea_mcp_server._profile_operation_gate") @patch("gitea_mcp_server._profile_operation_gate")
def test_author_denied_merge_permission(self, mock_gate): def test_author_denied_merge_permission(self, mock_gate):
"""Author profile lacks gitea.pr.merge → fail closed before mutation.""" """Author profile lacks gitea.pr.merge → fail closed before mutation."""