feat: add conflict-fix leases and stale-head protection (Closes #399 )

Introduce structured PR work leases so author conflict-fix pushes cannot race reviewer validation, approval, or merge on a moving head SHA. - pr_work_lease.py: parse/acquire leases, push and reviewer mutation gates - gitea_acquire_conflict_fix_lease, gitea_assess_conflict_fix_push MCP tools - Enforce reviewed head SHA on mark_final_review_decision, submit_pr_review, merge_pr - Final-report rules and workflow sections 20A / 26B - tests/test_pr_work_lease.py (14 cases)
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
2026-07-07 13:06:07 -04:00 · 2026-07-07 10:48:51 -05:00 · 2026-07-07 11:26:37 -04:00
3 changed files with 246 additions and 3 deletions
@@ -3581,16 +3581,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"])