From 3cb05284b35bff23b8a5c68a7e67bd57f01ed5fa Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Wed, 8 Jul 2026 04:18:09 -0400 Subject: [PATCH] feat: guard merged branch cleanup path Closes #514 --- branch_cleanup_guard.py | 98 +++++++++ final_report_validator.py | 15 ++ gitea_mcp_server.py | 144 ++++++++++++++ role_session_router.py | 4 +- .../workflows/review-merge-pr.md | 9 + task_capability_map.py | 4 + tests/test_branch_cleanup_guard.py | 187 ++++++++++++++++++ 7 files changed, 460 insertions(+), 1 deletion(-) create mode 100644 branch_cleanup_guard.py create mode 100644 tests/test_branch_cleanup_guard.py 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 85d7b4e..62f95a7 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_lock_provenance from post_merge_cleanup_proof import assess_post_merge_cleanup_proof from review_proofs import ( @@ -1024,6 +1025,19 @@ 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_review_mutation( report_text: str, *, @@ -1061,6 +1075,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, ) _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 411a80b..d8db5f6 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -615,6 +615,7 @@ import reconciliation_workflow # noqa: E402 import review_merge_state_machine # noqa: E402 import pr_work_lease # noqa: E402 import native_mcp_preference # noqa: E402 +import branch_cleanup_guard # noqa: E402 # Keyed issue-lock storage (#443): per remote/org/repo/issue files under @@ -4116,6 +4117,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 76a7000..e035fd4 100644 --- a/skills/llm-project-workflow/workflows/review-merge-pr.md +++ b/skills/llm-project-workflow/workflows/review-merge-pr.md @@ -871,6 +871,15 @@ 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. + Do not delete or mutate unrelated branches/worktrees. Do not touch the main checkout except to update the stable branch after merge if explicitly allowed by the workflow. diff --git a/task_capability_map.py b/task_capability_map.py index 1d40a78..c158cee 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -92,6 +92,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()