feat: gate gitea_merge_pr behind identity/profile/eligibility + confirmation (#16)

Replace the ungated gitea_merge_pr with a gated merge workflow. This is now the only merge path the MCP server exposes; the merge API is called only after every safety gate passes. Gates (fail-closed at each step): 1. Merge method is merge | squash | rebase. 2. Explicit confirmation: confirmation must equal "MERGE PR <n>" (without it, zero API calls are made). 3. Reuse gitea_check_pr_eligibility (#14) with action 'merge': proves the authenticated identity, the active profile (and that it allows merge), the PR author, blocks self-merge, requires the PR open, and fails closed when the PR is not mergeable or mergeability is unknown. 4. Optional expected_head_sha: refuse if the PR head moved. 5. Optional expected_changed_files: refuse if the PR's changed file set differs. 6. Redundant self-merge block (auth user == PR author). The force/ignore-checks option was removed — Gitea's own mergeable signal (which reflects branch-protection required reviews/checks) must be positive, so required approval/check state is honoured, never bypassed. Output reports performed, authenticated user, profile name, PR author, PR number, head SHA checked, merge method, gates passed/blocked, and merge result / merge commit — never a token, auth header, or credential. Error text is scrubbed via _redact. Surface audit: no ungated merge path remains. The /merge endpoint appears only inside gitea_merge_pr; gitea_review_pr fails closed on merge=True before any API call; gitea_submit_pr_review has no merge parameter and 'merge' is not a reviewable action. Tests assert all three. Tests cover: merge succeeds only when all gates pass; self-author blocked; unknown identity/profile blocked; profile without merge permission blocked; missing/wrong confirmation blocked (no API call); head-SHA mismatch blocked; changed-files mismatch blocked; closed PR blocked; non-mergeable blocked; unknown mergeability fail-closed; no merge call when gates fail; invalid merge method rejected; output and error redaction; and the no-ungated-merge-path audit. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-01 15:03:49 -04:00
parent cb926e25d3
commit f04cf44975
3 changed files with 462 additions and 27 deletions
@@ -298,20 +298,309 @@ class TestViewPR(unittest.TestCase):
 # Merge PR
 # ---------------------------------------------------------------------------
 class TestMergePR(unittest.TestCase):
+    """Gated merge workflow (#16). gitea_merge_pr is the only merge path."""
+
+    def _pr(self, author, state="open", sha="abc123", mergeable=True):
+        return {
+            "user": {"login": author},
+            "state": state,
+            "head": {"sha": sha},
+            "mergeable": mergeable,
+        }
+
+    def _confirm(self, n):
+        return f"MERGE PR {n}"
+
+    def _assert_no_merge_call(self, mock_api):
+        for c in mock_api.call_args_list:
+            method, url = c.args[0], c.args[1]
+            self.assertFalse(
+                method == "POST" and url.endswith("/merge"),
+                f"unexpected merge mutation: {method} {url}",
+            )
+
+    # -- success --------------------------------------------------------------

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_merge_pr(self, _auth, mock_api):
-        mock_api.return_value = {}
-        result = gitea_merge_pr(pr_number=1, do="squash", title="T", message="M", force=True)
-        self.assertTrue(result["success"])
-        self.assertIn("merged", result["message"])
-        # Check payload
-        payload = mock_api.call_args[0][3]
+    def test_merge_succeeds_when_all_gates_pass(self, _auth, mock_api):
+        mock_api.side_effect = [
+            {"login": "merger-bot"}, self._pr("author-bot"),
+            {},  # merge POST
+            {"merged_commit_sha": "mergecommit99"},  # read-back
+        ]
+        env = {"GITEA_PROFILE_NAME": "gitea-merger",
+               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
+        with patch.dict(os.environ, env, clear=True):
+            r = gitea_merge_pr(
+                pr_number=8, confirmation=self._confirm(8),
+                expected_head_sha="abc123", do="squash",
+                title="T", message="M", remote="prgs")
+        self.assertTrue(r["performed"])
+        self.assertEqual(r["authenticated_user"], "merger-bot")
+        self.assertEqual(r["pr_author"], "author-bot")
+        self.assertEqual(r["head_sha"], "abc123")
+        self.assertEqual(r["merge_method"], "squash")
+        self.assertEqual(r["merge_commit"], "mergecommit99")
+        # 3rd call is the merge POST with the requested method/title/message.
+        merge_call = mock_api.call_args_list[2]
+        self.assertEqual(merge_call.args[0], "POST")
+        self.assertTrue(merge_call.args[1].endswith("/pulls/8/merge"))
+        payload = merge_call.args[3]
        self.assertEqual(payload["Do"], "squash")
        self.assertEqual(payload["MergeTitleField"], "T")
        self.assertEqual(payload["MergeMessageField"], "M")
-        self.assertEqual(payload["force_merge"], True)
+        self.assertNotIn("force_merge", payload)
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_expected_changed_files_match_allows(self, _auth, mock_api):
+        mock_api.side_effect = [
+            {"login": "merger-bot"}, self._pr("author-bot"),
+            [{"filename": "a.py"}, {"filename": "b.py"}],  # files
+            {},  # merge POST
+            {"merged_commit_sha": "c1"},  # read-back
+        ]
+        env = {"GITEA_PROFILE_NAME": "gitea-merger",
+               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
+        with patch.dict(os.environ, env, clear=True):
+            r = gitea_merge_pr(
+                pr_number=8, confirmation=self._confirm(8),
+                expected_changed_files=["b.py", "a.py"], remote="prgs")
+        self.assertTrue(r["performed"])
+
+    # -- confirmation ---------------------------------------------------------
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_missing_confirmation_blocks_with_no_api_call(self, _auth, mock_api):
+        env = {"GITEA_PROFILE_NAME": "gitea-merger",
+               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
+        with patch.dict(os.environ, env, clear=True):
+            r = gitea_merge_pr(pr_number=8, confirmation="", remote="prgs")
+        self.assertFalse(r["performed"])
+        self.assertTrue(any("explicit confirmation required" in x for x in r["reasons"]))
+        mock_api.assert_not_called()
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_wrong_confirmation_blocks(self, _auth, mock_api):
+        env = {"GITEA_PROFILE_NAME": "gitea-merger",
+               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
+        with patch.dict(os.environ, env, clear=True):
+            r = gitea_merge_pr(pr_number=8, confirmation="MERGE PR 9", remote="prgs")
+        self.assertFalse(r["performed"])
+        mock_api.assert_not_called()
+
+    # -- identity / profile / eligibility fail-closed -------------------------
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_self_author_cannot_merge(self, _auth, mock_api):
+        mock_api.side_effect = [{"login": "jcwalker3"}, self._pr("jcwalker3")]
+        env = {"GITEA_PROFILE_NAME": "gitea-merger",
+               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
+        with patch.dict(os.environ, env, clear=True):
+            r = gitea_merge_pr(
+                pr_number=8, confirmation=self._confirm(8), remote="prgs")
+        self.assertFalse(r["performed"])
+        self.assertIn("authenticated user is PR author", r["reasons"])
+        self._assert_no_merge_call(mock_api)
+
+    @patch("mcp_server.get_auth_header", return_value=None)
+    def test_unknown_identity_blocks(self, _auth):
+        env = {"GITEA_PROFILE_NAME": "gitea-merger",
+               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
+        with patch.dict(os.environ, env, clear=True):
+            r = gitea_merge_pr(
+                pr_number=8, confirmation=self._confirm(8), remote="prgs")
+        self.assertFalse(r["performed"])
+        self.assertIsNone(r["authenticated_user"])
+        self.assertIn("authenticated identity could not be determined", r["reasons"])
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_unknown_profile_blocks(self, _auth, mock_api):
+        mock_api.side_effect = [{"login": "merger-bot"}, self._pr("author-bot")]
+        env = {}  # no GITEA_ALLOWED_OPERATIONS → empty allowed ops
+        with patch.dict(os.environ, env, clear=True):
+            r = gitea_merge_pr(
+                pr_number=8, confirmation=self._confirm(8), remote="prgs")
+        self.assertFalse(r["performed"])
+        self.assertIn(
+            "profile has no configured allowed operations (fail closed)",
+            r["reasons"])
+        self._assert_no_merge_call(mock_api)
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_profile_without_merge_permission_blocks(self, _auth, mock_api):
+        mock_api.side_effect = [{"login": "merger-bot"}, self._pr("author-bot")]
+        env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
+               "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
+        with patch.dict(os.environ, env, clear=True):
+            r = gitea_merge_pr(
+                pr_number=8, confirmation=self._confirm(8), remote="prgs")
+        self.assertFalse(r["performed"])
+        self.assertIn("profile is not allowed to merge", r["reasons"])
+        self._assert_no_merge_call(mock_api)
+
+    # -- PR state / mergeability ----------------------------------------------
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_closed_pr_blocks(self, _auth, mock_api):
+        mock_api.side_effect = [
+            {"login": "merger-bot"}, self._pr("author-bot", state="closed")]
+        env = {"GITEA_PROFILE_NAME": "gitea-merger",
+               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
+        with patch.dict(os.environ, env, clear=True):
+            r = gitea_merge_pr(
+                pr_number=8, confirmation=self._confirm(8), remote="prgs")
+        self.assertFalse(r["performed"])
+        self.assertIn("PR is not open (state=closed)", r["reasons"])
+        self._assert_no_merge_call(mock_api)
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_non_mergeable_pr_blocks(self, _auth, mock_api):
+        mock_api.side_effect = [
+            {"login": "merger-bot"}, self._pr("author-bot", mergeable=False)]
+        env = {"GITEA_PROFILE_NAME": "gitea-merger",
+               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
+        with patch.dict(os.environ, env, clear=True):
+            r = gitea_merge_pr(
+                pr_number=8, confirmation=self._confirm(8), remote="prgs")
+        self.assertFalse(r["performed"])
+        self.assertIn("PR is not mergeable", r["reasons"])
+        self._assert_no_merge_call(mock_api)
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_unknown_mergeability_fails_closed(self, _auth, mock_api):
+        mock_api.side_effect = [
+            {"login": "merger-bot"}, self._pr("author-bot", mergeable=None)]
+        env = {"GITEA_PROFILE_NAME": "gitea-merger",
+               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
+        with patch.dict(os.environ, env, clear=True):
+            r = gitea_merge_pr(
+                pr_number=8, confirmation=self._confirm(8), remote="prgs")
+        self.assertFalse(r["performed"])
+        self.assertIn("PR mergeability unknown", r["reasons"])
+        self._assert_no_merge_call(mock_api)
+
+    # -- head SHA / changed files ---------------------------------------------
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_head_sha_mismatch_blocks(self, _auth, mock_api):
+        mock_api.side_effect = [
+            {"login": "merger-bot"}, self._pr("author-bot", sha="abc123")]
+        env = {"GITEA_PROFILE_NAME": "gitea-merger",
+               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
+        with patch.dict(os.environ, env, clear=True):
+            r = gitea_merge_pr(
+                pr_number=8, confirmation=self._confirm(8),
+                expected_head_sha="deadbeef", remote="prgs")
+        self.assertFalse(r["performed"])
+        self.assertIn(
+            "expected head SHA does not match current PR head (fail closed)",
+            r["reasons"])
+        self._assert_no_merge_call(mock_api)
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_changed_files_mismatch_blocks(self, _auth, mock_api):
+        mock_api.side_effect = [
+            {"login": "merger-bot"}, self._pr("author-bot"),
+            [{"filename": "a.py"}, {"filename": "c.py"}],  # actual files
+        ]
+        env = {"GITEA_PROFILE_NAME": "gitea-merger",
+               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
+        with patch.dict(os.environ, env, clear=True):
+            r = gitea_merge_pr(
+                pr_number=8, confirmation=self._confirm(8),
+                expected_changed_files=["a.py", "b.py"], remote="prgs")
+        self.assertFalse(r["performed"])
+        self.assertIn(
+            "PR changed files do not match expected_changed_files (fail closed)",
+            r["reasons"])
+        self._assert_no_merge_call(mock_api)
+
+    # -- misc -----------------------------------------------------------------
+
+    @patch("mcp_server.api_request")
+    def test_invalid_merge_method_rejected(self, mock_api):
+        with patch.dict(os.environ, {}, clear=True):
+            r = gitea_merge_pr(
+                pr_number=8, confirmation="MERGE PR 8", do="octopus", remote="prgs")
+        self.assertFalse(r["performed"])
+        self.assertTrue(any("unknown merge method" in x for x in r["reasons"]))
+        mock_api.assert_not_called()
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_output_redacts_secrets(self, _auth, mock_api):
+        mock_api.side_effect = [
+            {"login": "merger-bot"}, self._pr("author-bot"),
+            {}, {"merged_commit_sha": "c1"},
+        ]
+        env = {"GITEA_PROFILE_NAME": "gitea-merger",
+               "GITEA_ALLOWED_OPERATIONS": "read,merge",
+               "GITEA_TOKEN": "super-secret-token"}
+        with patch.dict(os.environ, env, clear=True):
+            r = gitea_merge_pr(
+                pr_number=8, confirmation=self._confirm(8), remote="prgs")
+        blob = repr(r).lower()
+        for secret in ("super-secret-token", "authorization", "basic ", FAKE_AUTH.lower()):
+            self.assertNotIn(secret, blob)
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_merge_error_message_redacts_credential(self, _auth, mock_api):
+        mock_api.side_effect = [
+            {"login": "merger-bot"}, self._pr("author-bot"),
+            RuntimeError("HTTP 500: token abc-secret-xyz rejected"),
+        ]
+        env = {"GITEA_PROFILE_NAME": "gitea-merger",
+               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
+        with patch.dict(os.environ, env, clear=True):
+            r = gitea_merge_pr(
+                pr_number=8, confirmation=self._confirm(8), remote="prgs")
+        self.assertFalse(r["performed"])
+        blob = repr(r)
+        self.assertIn("[REDACTED]", blob)
+        self.assertNotIn("abc-secret-xyz", blob)
+
+
+class TestNoUngatedMergePath(unittest.TestCase):
+    """Prove no other exposed tool can merge (#16 surface audit)."""
+
+    def test_submit_pr_review_has_no_merge(self):
+        import inspect
+        import mcp_server
+        # No merge parameter on the review-mutation tool.
+        params = inspect.signature(mcp_server.gitea_submit_pr_review).parameters
+        self.assertNotIn("merge", params)
+        # And 'merge' is not a reviewable action.
+        self.assertNotIn("merge", mcp_server._REVIEW_ACTIONS)
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_review_pr_merge_true_makes_no_api_call(self, _auth, mock_api):
+        r = gitea_review_pr(pr_number=1, event="APPROVE", body="x", merge=True)
+        self.assertFalse(r["success"])
+        self.assertEqual(mock_api.call_count, 0)
+
+    def test_only_merge_endpoint_is_in_gated_tool(self):
+        # Source-level audit: the merge endpoint appears only inside
+        # gitea_merge_pr, which is the gated path.
+        import inspect
+        import mcp_server
+        src = inspect.getsource(mcp_server)
+        self.assertEqual(src.count("/merge\""), 1)
+        merge_src = inspect.getsource(mcp_server.gitea_merge_pr)
+        self.assertIn("/merge\"", merge_src)


 # ---------------------------------------------------------------------------