diff --git a/branch_cleanup_guard.py b/branch_cleanup_guard.py new file mode 100644 index 0000000..66d2708 --- /dev/null +++ b/branch_cleanup_guard.py @@ -0,0 +1,98 @@ +"""Guards for merged-PR branch cleanup and raw git delete bypasses (#514).""" + +from __future__ import annotations + +import re +from typing import Any + +PROTECTED_BRANCHES = frozenset({"master", "main", "dev"}) + +_RAW_BRANCH_DELETE_PATTERNS = ( + re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+branch\s+-[dD]\b[^\n\r]*", re.I), + re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+push\b[^\n\r]*\s--delete\b[^\n\r]*", re.I), + re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+push\b[^\n\r]*\s:[^\s`]+", re.I), +) + + +def raw_branch_delete_commands(text: str | None) -> list[str]: + """Return raw git branch-delete commands cited in *text*.""" + if not text: + return [] + commands: list[str] = [] + for pattern in _RAW_BRANCH_DELETE_PATTERNS: + commands.extend(match.group(0).strip("` ") for match in pattern.finditer(text)) + return list(dict.fromkeys(commands)) + + +def assess_raw_branch_delete_report(text: str | None) -> dict[str, Any]: + """Fail closed when a report uses raw git branch deletion as cleanup proof.""" + commands = raw_branch_delete_commands(text) + reasons = [ + ( + "raw git branch deletion bypasses MCP branch.delete cleanup gates: " + f"{command}" + ) + for command in commands + ] + return { + "proven": not reasons, + "block": bool(reasons), + "commands": commands, + "reasons": reasons, + "safe_next_action": ( + "use gitea_cleanup_merged_pr_branch or another approved cleanup " + "helper with explicit branch.delete capability" + if reasons + else "proceed" + ), + } + + +def assess_merged_pr_branch_cleanup( + *, + pr_number: int, + head_branch: str, + merged: bool, + remote_branch_exists: bool, + open_pr_heads: set[str], + head_on_target: bool | None, + delete_capability_allowed: bool, + confirmation: str | None, + protected_branches: frozenset[str] | None = None, +) -> dict[str, Any]: + """Assess whether a merged PR source branch can be deleted via MCP.""" + protected = protected_branches or PROTECTED_BRANCHES + expected_confirmation = f"CLEANUP MERGED PR {pr_number} BRANCH {head_branch}" + reasons: list[str] = [] + if not merged: + reasons.append("PR is not merged") + if not remote_branch_exists: + reasons.append("remote branch already absent") + if not head_branch: + reasons.append("PR head branch is missing") + if head_branch in protected: + reasons.append(f"branch '{head_branch}' is protected") + if head_branch in open_pr_heads: + reasons.append("an open PR still references this head branch") + if head_on_target is False: + reasons.append("PR head is not an ancestor of the target branch") + if head_on_target is None: + reasons.append("PR head ancestry could not be proven") + if not delete_capability_allowed: + reasons.append("gitea.branch.delete capability is not allowed") + if confirmation != expected_confirmation: + reasons.append( + "confirmation must equal " + f"'{expected_confirmation}' for branch cleanup" + ) + + safe = not reasons + return { + "pr_number": pr_number, + "head_branch": head_branch, + "expected_confirmation": expected_confirmation, + "remote_branch_exists": remote_branch_exists, + "safe_to_delete": safe, + "block_reasons": reasons, + "recommended_action": "delete_remote_branch" if safe else "keep_remote_branch", + } diff --git a/final_report_validator.py b/final_report_validator.py index a16b4a0..4a5aa0b 100644 --- a/final_report_validator.py +++ b/final_report_validator.py @@ -11,6 +11,7 @@ import inspect import re from typing import Any, Callable +import branch_cleanup_guard import issue_acceptance_gate import issue_lock_provenance import reviewer_handoff_consistency @@ -1294,6 +1295,17 @@ def _rule_shared_author_reviewer_same_run(report_text: str) -> list[dict[str, st ) +def _rule_shared_raw_branch_delete_bypass(report_text: str) -> list[dict[str, str]]: + result = branch_cleanup_guard.assess_raw_branch_delete_report(report_text) + if not result["block"]: + return [] + return _findings_from_reasons( + "shared.raw_branch_delete_bypass", + result["reasons"], + field="Cleanup mutations", + severity="block", + safe_next_action=result["safe_next_action"], + ) def _rule_reviewer_workflow_load_boundary(report_text: str) -> list[dict[str, str]]: """#403: require structured workflow-load helper result, not file-view narrative.""" if not report_text.strip(): @@ -1448,6 +1460,7 @@ _SHARED_ISSUE_LOCK_RULES = ( _rule_shared_issue_lock_external_state, _rule_shared_manual_lock_pr_override, _rule_shared_author_reviewer_same_run, + _rule_shared_raw_branch_delete_bypass, _rule_shared_canonical_state_update, ) diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 241b278..49460df 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -832,6 +832,7 @@ import pr_work_lease # noqa: E402 import workflow_skill # noqa: E402 import conflict_fix_classification # noqa: E402 import native_mcp_preference # noqa: E402 +import branch_cleanup_guard # noqa: E402 import thread_state_ledger_validator # noqa: E402 import master_parity_gate # noqa: E402 @@ -4852,6 +4853,149 @@ def gitea_delete_branch( return {"success": True, "message": f"Remote branch '{branch}' deleted."} +@mcp.tool() +def gitea_cleanup_merged_pr_branch( + pr_number: int, + confirmation: str, + branch: str | None = None, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, + worktree_path: str | None = None, +) -> dict: + """Delete a merged PR source branch through the guarded MCP path (#514).""" + 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"), + } + + profile = get_profile() + active_role = _role_kind( + profile.get("allowed_operations", []), + profile.get("forbidden_operations", []), + ) + if active_role == "reviewer": + return { + "success": False, + "performed": False, + "required_permission": "gitea.branch.delete", + "reasons": [ + "reviewer profile is not authorized for merged branch cleanup " + "(fail closed)" + ], + "permission_report": _permission_block_report("gitea.branch.delete"), + } + + if worktree_path is None or "/branches/" not in os.path.realpath(worktree_path): + return { + "success": False, + "performed": False, + "required_permission": "gitea.branch.delete", + "reasons": [ + "merged branch cleanup requires an explicit branches/ worktree " + "path; root checkout branch ref mutation is blocked (fail closed)" + ], + } + + verify_preflight_purity( + remote, + worktree_path=worktree_path, + task="cleanup_merged_pr_branch", + ) + + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) + base = repo_api_url(h, o, r) + pr = api_request("GET", f"{base}/pulls/{pr_number}", auth) + pr_head = pr.get("head") or {} + head_branch = branch or pr_head.get("ref") or "" + head_sha = pr_head.get("sha") + target_branch = (pr.get("base") or {}).get("ref") or "master" + + if branch and branch != pr_head.get("ref"): + return { + "success": False, + "performed": False, + "pr_number": pr_number, + "branch": branch, + "reasons": [ + f"requested branch '{branch}' does not match PR head " + f"'{pr_head.get('ref')}'" + ], + } + + open_prs = api_get_all(f"{base}/pulls?state=open", auth) + open_heads = { + str((open_pr.get("head") or {}).get("ref")) + for open_pr in open_prs + if (open_pr.get("head") or {}).get("ref") + } + remote_exists = _remote_branch_exists(h, o, r, auth, head_branch) + target_ref = ( + f"{remote}/{target_branch}" + if remote in REMOTES + else f"origin/{target_branch}" + ) + head_on_target = merged_cleanup_reconcile.is_head_ancestor_of_ref( + PROJECT_ROOT, + head_sha, + target_ref, + ) + assessment = branch_cleanup_guard.assess_merged_pr_branch_cleanup( + pr_number=pr_number, + head_branch=head_branch, + merged=bool(pr.get("merged") or pr.get("merged_at")), + remote_branch_exists=remote_exists, + open_pr_heads=open_heads, + head_on_target=head_on_target, + delete_capability_allowed=True, + confirmation=confirmation, + ) + if not assessment["safe_to_delete"]: + return { + "success": False, + "performed": False, + "pr_number": pr_number, + "branch": head_branch, + "assessment": assessment, + "reasons": assessment["block_reasons"], + } + + import urllib.parse + + encoded_branch = urllib.parse.quote(head_branch, safe="") + url = f"{base}/branches/{encoded_branch}" + with _audited( + "cleanup_merged_pr_branch", + host=h, + remote=remote, + org=o, + repo=r, + pr_number=pr_number, + target_branch=head_branch, + request_metadata={ + "branch": head_branch, + "required_permission": "gitea.branch.delete", + "cleanup_path": "gitea_cleanup_merged_pr_branch", + }, + ): + api_request("DELETE", url, auth) + return { + "success": True, + "performed": True, + "pr_number": pr_number, + "branch": head_branch, + "message": f"Merged PR #{pr_number} source branch '{head_branch}' deleted.", + "assessment": assessment, + } + + def _remote_branch_exists(h: str, o: str, r: str, auth: str, branch: str) -> bool: import urllib.parse diff --git a/role_session_router.py b/role_session_router.py index 8f30dc5..48dbdd2 100644 --- a/role_session_router.py +++ b/role_session_router.py @@ -80,6 +80,7 @@ AUTHOR_TASKS = frozenset({ }) RECONCILER_TASKS = frozenset({ + "cleanup_merged_pr_branch", "reconcile_already_landed_pr", "reconcile_already_landed", "reconcile-landed-pr", @@ -109,6 +110,7 @@ TASK_REQUIRED_ROLE = { "reconcile_already_landed_pr": "reconciler", "reconcile_already_landed": "reconciler", "reconcile-landed-pr": "reconciler", + "cleanup_merged_pr_branch": "reconciler", # #309: reconciler tasks close already-landed PRs/issues only. "reconcile_close_landed_pr": "reconciler", "reconcile_close_landed_issue": "reconciler", @@ -406,4 +408,4 @@ def assess_infra_stop(project_root: str | None = None) -> dict: def check_mid_merge(project_root: str | None = None) -> bool: """Return True if the repository is mid-merge, mid-rebase, or has conflict markers.""" - return assess_infra_stop(project_root)["infra_stop"] \ No newline at end of file + return assess_infra_stop(project_root)["infra_stop"] diff --git a/skills/llm-project-workflow/workflows/review-merge-pr.md b/skills/llm-project-workflow/workflows/review-merge-pr.md index 1ccbedb..6c35ce9 100644 --- a/skills/llm-project-workflow/workflows/review-merge-pr.md +++ b/skills/llm-project-workflow/workflows/review-merge-pr.md @@ -936,6 +936,14 @@ Confirm: Clean only the session-owned `branches/` review worktree if the project workflow explicitly allows cleanup. +Do not delete source branches from reviewer mode with raw git commands. Commands +such as `git branch -d ` and `git push --delete ` are +not proof of authorized cleanup and bypass MCP role/capability gates. Merged PR +source branch cleanup must use an explicit MCP cleanup tool such as +`gitea_cleanup_merged_pr_branch` or another approved cleanup helper with exact +`gitea.branch.delete` authority, merged-PR proof, no open PR using the branch, +target-branch ancestry proof, a `branches/` worktree path, and cleanup mutations +reported separately from review mutations. Review, baseline, and merge-simulation worktrees created during this run are transient and are removed automatically at successful completion once they are clean, carry no open PR, and hold no active lease (#401). Use diff --git a/task_capability_map.py b/task_capability_map.py index fb68bed..d32d315 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -100,6 +100,10 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.branch.delete", "role": "author", }, + "cleanup_merged_pr_branch": { + "permission": "gitea.branch.delete", + "role": "reconciler", + }, "commit_files": { "permission": "gitea.repo.commit", "role": "author", diff --git a/tests/test_branch_cleanup_guard.py b/tests/test_branch_cleanup_guard.py new file mode 100644 index 0000000..dd4053f --- /dev/null +++ b/tests/test_branch_cleanup_guard.py @@ -0,0 +1,187 @@ +import sys +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +import branch_cleanup_guard as guard # noqa: E402 +import mcp_server # noqa: E402 +import task_capability_map # noqa: E402 +from final_report_validator import assess_final_report_validator # noqa: E402 +from mcp_server import gitea_cleanup_merged_pr_branch # noqa: E402 + +FAKE_AUTH = "token fake" + + +class TestRawBranchDeleteGuard(unittest.TestCase): + def test_detects_local_and_remote_raw_git_delete_commands(self): + text = """ + Cleanup mutations: + - git branch -d feat/issue-485-lease-comments-non-list-guard + - git push prgs --delete feat/issue-485-lease-comments-non-list-guard + """ + commands = guard.raw_branch_delete_commands(text) + self.assertEqual(len(commands), 2) + self.assertIn("git branch -d", commands[0]) + self.assertIn("git push prgs --delete", commands[1]) + + def test_final_report_blocks_raw_delete_as_cleanup_proof(self): + report = """ + ## Controller Handoff + - Task: review_pr + - Cleanup mutations: git push prgs --delete feat/branch + """ + result = assess_final_report_validator(report, "review_pr") + self.assertTrue(result["blocked"]) + self.assertTrue( + any("shared.raw_branch_delete_bypass" in r for r in result["reasons"]) + ) + + def test_cleanup_task_requires_branch_delete_capability(self): + self.assertEqual( + task_capability_map.required_permission("cleanup_merged_pr_branch"), + "gitea.branch.delete", + ) + self.assertEqual( + task_capability_map.required_role("cleanup_merged_pr_branch"), + "reconciler", + ) + + +class TestMergedPrBranchCleanupTool(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() + patch("gitea_audit.audit_enabled", return_value=False).start() + self.mock_api = patch("mcp_server.api_request").start() + self.mock_all = patch("mcp_server.api_get_all", return_value=[]).start() + patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start() + patch( + "mcp_server.merged_cleanup_reconcile.is_head_ancestor_of_ref", + return_value=True, + ).start() + + def tearDown(self): + patch.stopall() + + def test_reviewer_without_delete_authority_fails_closed(self): + patch( + "mcp_server.get_profile", + return_value={ + "profile_name": "reviewer", + "allowed_operations": ["gitea.read", "gitea.pr.review"], + "forbidden_operations": ["gitea.branch.delete"], + }, + ).start() + res = gitea_cleanup_merged_pr_branch( + pr_number=487, + confirmation="CLEANUP MERGED PR 487 BRANCH feat/branch", + branch="feat/branch", + remote="prgs", + worktree_path="/tmp/repo/branches/cleanup", + ) + self.assertFalse(res["performed"]) + self.assertEqual(res["required_permission"], "gitea.branch.delete") + self.mock_api.assert_not_called() + + def test_root_checkout_cleanup_fails_closed(self): + patch( + "mcp_server.get_profile", + return_value={ + "profile_name": "branch-cleanup", + "allowed_operations": ["gitea.read", "gitea.branch.delete"], + "forbidden_operations": [], + }, + ).start() + res = gitea_cleanup_merged_pr_branch( + pr_number=487, + confirmation="CLEANUP MERGED PR 487 BRANCH feat/branch", + branch="feat/branch", + remote="prgs", + worktree_path="/tmp/repo/Gitea-Tools", + ) + self.assertFalse(res["performed"]) + self.assertIn("root checkout", " ".join(res["reasons"])) + self.mock_api.assert_not_called() + + def test_authorized_cleanup_deletes_through_api_path(self): + branch = "feat/issue-485-lease-comments-non-list-guard" + patch( + "mcp_server.get_profile", + return_value={ + "profile_name": "branch-cleanup", + "allowed_operations": ["gitea.read", "gitea.branch.delete"], + "forbidden_operations": [], + }, + ).start() + self.mock_api.side_effect = [ + { + "number": 487, + "merged": True, + "merged_at": "2026-07-08T01:00:00Z", + "head": {"ref": branch, "sha": "a" * 40}, + "base": {"ref": "master"}, + }, + {}, + {}, + ] + res = gitea_cleanup_merged_pr_branch( + pr_number=487, + confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}", + branch=branch, + remote="prgs", + worktree_path="/tmp/repo/branches/cleanup", + ) + self.assertTrue(res["performed"]) + delete_calls = [ + call for call in self.mock_api.call_args_list if call.args[0] == "DELETE" + ] + self.assertEqual(len(delete_calls), 1) + self.assertIn("branches/feat%2Fissue-485", delete_calls[0].args[1]) + + def test_confirmation_required_before_delete(self): + branch = "feat/issue-485-lease-comments-non-list-guard" + patch( + "mcp_server.get_profile", + return_value={ + "profile_name": "branch-cleanup", + "allowed_operations": ["gitea.read", "gitea.branch.delete"], + "forbidden_operations": [], + }, + ).start() + self.mock_api.side_effect = [ + { + "number": 487, + "merged": True, + "merged_at": "2026-07-08T01:00:00Z", + "head": {"ref": branch, "sha": "a" * 40}, + "base": {"ref": "master"}, + }, + {}, + ] + res = gitea_cleanup_merged_pr_branch( + pr_number=487, + confirmation="wrong", + branch=branch, + remote="prgs", + worktree_path="/tmp/repo/branches/cleanup", + ) + self.assertFalse(res["performed"]) + self.assertIn("confirmation", " ".join(res["reasons"])) + delete_calls = [ + call for call in self.mock_api.call_args_list if call.args[0] == "DELETE" + ] + self.assertFalse(delete_calls) + + +if __name__ == "__main__": + unittest.main()