99 lines
3.4 KiB
Python
99 lines
3.4 KiB
Python
"""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",
|
|
}
|