fix(reconciler): forward explicit repository through delete_branch anti-stomp preflight (Closes #733)
gitea_delete_branch accepted explicit org/repo but did not propagate them through the anti-stomp preflight. During a deletion targeting Scaled-Tech-Consulting/Gitea-Tools, preflight resolved the remote-wide prgs default (Scaled-Tech-Consulting/Timesheet) and failed closed with wrong_repo before any deletion — because verify_preflight_purity was called without org/repo and the shared #604 anti-stomp resolution fell back to the REMOTES default. Fix (delete_branch only; REMOTES untouched): 1. Forward the explicit org/repo into verify_preflight_purity so the shared #604 anti-stomp resolution validates the *targeted* repository instead of the remote-wide default. Explicit Scaled-Tech-Consulting/Gitea-Tools now propagates through role/workspace verification, verify_preflight_purity, anti-stomp repository resolution, and the final native deletion request. 2. Add a workspace-derived repository-binding gate (_delete_branch_repository_binding_block) that validates explicit coordinates against the immutable, workspace-aligned repository identity. Wrong, substituted, or unverified coordinates fail closed — independent of the anti-stomp remote/repo guard, which by the #530 contract trusts explicit caller intent. REMOTES defaults are never consulted as an authorization scope. Preserved: reconciler-only ownership of gitea.branch.delete; author/reviewer/ merger denial; protected and preservation/evidence gates; missing coordinates still fail closed via the anti-stomp default resolution. Regression matrix (tests/test_issue_733_delete_branch_repo_forwarding.py): Timesheet-default + explicit Gitea-Tools permits an eligible deletion and forwards the coordinates; wrong/substituted/unverified coordinates fail closed; author/reviewer/merger remain denied; protected and preservation branches remain blocked; anti-stomp resolution marks explicit coordinates authoritative and fails closed on the remote-wide default for omitted coordinates; the task-capability and role-routing maps stay consistent. Co-Authored-By: Claude Opus 4.8 <[email protected]> Claude-Session: https://claude.ai/code/session_01UVDxVKANhuGYzZgayDU3QS
This commit is contained in:
Claude Opus 4.8
@@ -0,0 +1,385 @@
|
||||
"""Regression matrix for Issue #733.
|
||||
|
||||
``gitea_delete_branch`` accepts explicit ``org``/``repo`` but historically did
|
||||
not propagate them through the anti-stomp preflight: preflight resolved the
|
||||
remote-wide ``REMOTES`` default (bare ``prgs`` → ``Scaled-Tech-Consulting/
|
||||
Timesheet``) and fail-closed with ``wrong_repo`` even though the caller
|
||||
explicitly targeted ``Scaled-Tech-Consulting/Gitea-Tools``.
|
||||
|
||||
The fix has two parts, both exercised here:
|
||||
|
||||
1. ``gitea_delete_branch`` forwards the explicit ``org``/``repo`` into
|
||||
``verify_preflight_purity`` so the shared #604 anti-stomp resolution
|
||||
validates the *targeted* repository instead of the remote-wide default.
|
||||
2. A workspace-derived repository-binding gate
|
||||
(``_delete_branch_repository_binding_block``) validates explicit
|
||||
coordinates against the immutable workspace identity, rejecting wrong,
|
||||
substituted, or unverified targets — independent of the anti-stomp
|
||||
remote/repo guard, which by the #530 contract trusts explicit intent.
|
||||
|
||||
Every existing gate (reconciler-only ownership; author/reviewer/merger denial;
|
||||
protected/preservation blocks) is preserved.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
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
|
||||
import remote_repo_guard
|
||||
import task_capability_map
|
||||
import role_session_router
|
||||
|
||||
FAKE_AUTH = "token fake"
|
||||
|
||||
# The workspace is bound to Gitea-Tools even though the prgs REMOTES default
|
||||
# repo is Timesheet (the exact #733 scenario).
|
||||
WORKSPACE_SLUG = "Scaled-Tech-Consulting/Gitea-Tools"
|
||||
LOCAL_GITEA_TOOLS_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
|
||||
|
||||
RECONCILER = {
|
||||
"profile_name": "prgs-reconciler",
|
||||
"role": "reconciler",
|
||||
"allowed_operations": [
|
||||
"gitea.read", "gitea.issue.comment", "gitea.pr.comment",
|
||||
"gitea.branch.delete",
|
||||
],
|
||||
"forbidden_operations": ["gitea.pr.create", "gitea.pr.merge"],
|
||||
"audit_label": "prgs-reconciler",
|
||||
}
|
||||
|
||||
AUTHOR_WITH_DELETE = {
|
||||
"profile_name": "prgs-author",
|
||||
"role": "author",
|
||||
"allowed_operations": [
|
||||
"gitea.read", "gitea.pr.create", "gitea.branch.push",
|
||||
"gitea.branch.delete",
|
||||
],
|
||||
"forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
|
||||
"audit_label": "prgs-author",
|
||||
}
|
||||
|
||||
REVIEWER_WITH_DELETE = {
|
||||
"profile_name": "prgs-reviewer",
|
||||
"role": "reviewer",
|
||||
"allowed_operations": ["gitea.read", "gitea.pr.review", "gitea.branch.delete"],
|
||||
"forbidden_operations": [],
|
||||
"audit_label": "prgs-reviewer",
|
||||
}
|
||||
|
||||
MERGER_WITH_DELETE = {
|
||||
"profile_name": "prgs-merger",
|
||||
"role": "merger",
|
||||
"allowed_operations": ["gitea.read", "gitea.pr.merge", "gitea.branch.delete"],
|
||||
"forbidden_operations": [],
|
||||
"audit_label": "prgs-merger",
|
||||
}
|
||||
|
||||
|
||||
class _DeleteBranchBase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._snap = (
|
||||
mcp_server._preflight_whoami_called,
|
||||
mcp_server._preflight_capability_called,
|
||||
)
|
||||
mcp_server._preflight_whoami_called = False
|
||||
mcp_server._preflight_capability_called = False
|
||||
# prgs default repo is Timesheet — the remote-wide default #733 must NOT
|
||||
# fall back to.
|
||||
self._remotes = patch.dict(mcp_server.REMOTES, {
|
||||
"prgs": {
|
||||
"host": "gitea.prgs.cc",
|
||||
"org": "Scaled-Tech-Consulting",
|
||||
"repo": "Timesheet",
|
||||
},
|
||||
})
|
||||
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_api.return_value = {}
|
||||
patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
|
||||
# Workspace is verifiably bound to Gitea-Tools.
|
||||
patch(
|
||||
"mcp_server._workspace_repository_slug", return_value=WORKSPACE_SLUG
|
||||
).start()
|
||||
|
||||
def tearDown(self):
|
||||
patch.stopall()
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
(
|
||||
mcp_server._preflight_whoami_called,
|
||||
mcp_server._preflight_capability_called,
|
||||
) = self._snap
|
||||
|
||||
def _bind(self, profile, role):
|
||||
patch("mcp_server.get_profile", return_value=profile).start()
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check("capability", resolved_role=role)
|
||||
|
||||
def _delete_calls(self):
|
||||
return [
|
||||
c for c in self.mock_api.call_args_list
|
||||
if c.args and c.args[0] == "DELETE"
|
||||
]
|
||||
|
||||
|
||||
class TestPositiveTimesheetDefaultExplicitGiteaTools(_DeleteBranchBase):
|
||||
"""AC: prgs default=Timesheet + explicit Gitea-Tools → preflight validates
|
||||
Gitea-Tools and permits an otherwise-eligible deletion."""
|
||||
|
||||
def test_explicit_gitea_tools_permits_delete_and_forwards_coords(self):
|
||||
self._bind(RECONCILER, "reconciler")
|
||||
with patch("mcp_server.verify_preflight_purity") as vpp:
|
||||
res = gitea_delete_branch(
|
||||
branch="feat/pr-sync-status",
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
)
|
||||
self.assertTrue(res["success"], res)
|
||||
self.assertIn("deleted", res["message"])
|
||||
self.assertTrue(self._delete_calls())
|
||||
# Forwarding: preflight received the explicit Gitea-Tools coordinates
|
||||
# and the delete_branch task (not the remote-wide Timesheet default).
|
||||
self.assertTrue(vpp.called)
|
||||
kwargs = vpp.call_args.kwargs
|
||||
self.assertEqual(kwargs.get("task"), "delete_branch")
|
||||
self.assertEqual(kwargs.get("org"), "Scaled-Tech-Consulting")
|
||||
self.assertEqual(kwargs.get("repo"), "Gitea-Tools")
|
||||
|
||||
|
||||
class TestNegativeRepositoryBinding(_DeleteBranchBase):
|
||||
"""AC negatives: wrong / substituted / unverified coordinates fail closed."""
|
||||
|
||||
def test_wrong_explicit_repo_fails_closed(self):
|
||||
self._bind(RECONCILER, "reconciler")
|
||||
with patch("mcp_server.verify_preflight_purity") as vpp:
|
||||
res = gitea_delete_branch(
|
||||
branch="feat/x",
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Timesheet", # not the workspace-bound Gitea-Tools
|
||||
)
|
||||
self.assertFalse(res["success"])
|
||||
self.assertFalse(res["performed"])
|
||||
self.assertEqual(res["blocker_kind"], "repository_binding")
|
||||
self.assertTrue(any("Timesheet" in r for r in res["reasons"]))
|
||||
self.assertFalse(self._delete_calls())
|
||||
# Fail closed BEFORE any preflight/deletion side effect.
|
||||
vpp.assert_not_called()
|
||||
|
||||
def test_repository_substitution_org_mismatch_rejected(self):
|
||||
self._bind(RECONCILER, "reconciler")
|
||||
res = gitea_delete_branch(
|
||||
branch="feat/x",
|
||||
remote="prgs",
|
||||
org="Some-Other-Org",
|
||||
repo="Gitea-Tools",
|
||||
)
|
||||
self.assertFalse(res["success"])
|
||||
self.assertEqual(res["blocker_kind"], "repository_binding")
|
||||
self.assertFalse(self._delete_calls())
|
||||
|
||||
def test_explicit_coords_without_workspace_identity_fail_closed(self):
|
||||
self._bind(RECONCILER, "reconciler")
|
||||
with patch("mcp_server._workspace_repository_slug", return_value=None):
|
||||
res = gitea_delete_branch(
|
||||
branch="feat/x",
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
)
|
||||
self.assertFalse(res["success"])
|
||||
self.assertEqual(res["blocker_kind"], "repository_binding")
|
||||
self.assertTrue(any("unverified" in r for r in res["reasons"]))
|
||||
self.assertFalse(self._delete_calls())
|
||||
|
||||
def test_matching_explicit_passes_binding(self):
|
||||
self._bind(RECONCILER, "reconciler")
|
||||
block = mcp_server._delete_branch_repository_binding_block(
|
||||
"prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools"
|
||||
)
|
||||
self.assertIsNone(block)
|
||||
|
||||
def test_omitted_coords_pass_binding_and_are_revalidated_by_preflight(self):
|
||||
# Omitted coordinates pass the binding gate (nothing to corroborate) but
|
||||
# are forwarded as None so the anti-stomp resolution fails closed on the
|
||||
# remote-wide Timesheet default (see TestAntiStompResolution below).
|
||||
self._bind(RECONCILER, "reconciler")
|
||||
self.assertIsNone(
|
||||
mcp_server._delete_branch_repository_binding_block(
|
||||
"prgs", org=None, repo=None
|
||||
)
|
||||
)
|
||||
with patch("mcp_server.verify_preflight_purity") as vpp:
|
||||
gitea_delete_branch(branch="feat/x", remote="prgs")
|
||||
self.assertTrue(vpp.called)
|
||||
kwargs = vpp.call_args.kwargs
|
||||
self.assertIsNone(kwargs.get("org"))
|
||||
self.assertIsNone(kwargs.get("repo"))
|
||||
|
||||
|
||||
class TestRoleDenials(_DeleteBranchBase):
|
||||
"""AC: author, reviewer, and merger remain denied even holding the perm."""
|
||||
|
||||
def test_author_with_delete_perm_denied(self):
|
||||
self._bind(AUTHOR_WITH_DELETE, "author")
|
||||
res = gitea_delete_branch(
|
||||
branch="feat/x", remote="prgs",
|
||||
org="Scaled-Tech-Consulting", repo="Gitea-Tools",
|
||||
)
|
||||
self.assertFalse(res["success"])
|
||||
self.assertEqual(res["required_role_kind"], "reconciler")
|
||||
self.assertEqual(res["active_role_kind"], "author")
|
||||
self.assertFalse(self._delete_calls())
|
||||
|
||||
def test_reviewer_with_delete_perm_denied(self):
|
||||
self._bind(REVIEWER_WITH_DELETE, "reviewer")
|
||||
res = gitea_delete_branch(
|
||||
branch="feat/x", remote="prgs",
|
||||
org="Scaled-Tech-Consulting", repo="Gitea-Tools",
|
||||
)
|
||||
self.assertFalse(res["success"])
|
||||
self.assertEqual(res["required_role_kind"], "reconciler")
|
||||
self.assertFalse(self._delete_calls())
|
||||
|
||||
def test_merger_with_delete_perm_denied(self):
|
||||
self._bind(MERGER_WITH_DELETE, "merger")
|
||||
res = gitea_delete_branch(
|
||||
branch="feat/x", remote="prgs",
|
||||
org="Scaled-Tech-Consulting", repo="Gitea-Tools",
|
||||
)
|
||||
self.assertFalse(res["success"])
|
||||
self.assertEqual(res["required_role_kind"], "reconciler")
|
||||
self.assertFalse(self._delete_calls())
|
||||
|
||||
|
||||
class TestProtectedAndPreservedBranches(_DeleteBranchBase):
|
||||
"""AC: protected and preservation/evidence branches remain blocked."""
|
||||
|
||||
def test_protected_branch_blocked(self):
|
||||
self._bind(RECONCILER, "reconciler")
|
||||
for branch in ("master", "main", "dev"):
|
||||
with self.subTest(branch=branch):
|
||||
res = gitea_delete_branch(
|
||||
branch=branch, remote="prgs",
|
||||
org="Scaled-Tech-Consulting", repo="Gitea-Tools",
|
||||
)
|
||||
self.assertFalse(res["success"])
|
||||
self.assertTrue(any("protected" in r for r in res["reasons"]))
|
||||
self.assertFalse(self._delete_calls())
|
||||
|
||||
def test_preservation_branch_blocked(self):
|
||||
self._bind(RECONCILER, "reconciler")
|
||||
for branch in ("chore/preserve-local-master", "feat/evidence-run"):
|
||||
with self.subTest(branch=branch):
|
||||
res = gitea_delete_branch(
|
||||
branch=branch, remote="prgs",
|
||||
org="Scaled-Tech-Consulting", repo="Gitea-Tools",
|
||||
)
|
||||
self.assertFalse(res["success"])
|
||||
self.assertTrue(
|
||||
any("preservation" in r for r in res["reasons"])
|
||||
)
|
||||
self.assertFalse(self._delete_calls())
|
||||
|
||||
|
||||
class TestAntiStompResolution(unittest.TestCase):
|
||||
"""The anti-stomp repository resolution validates the *targeted* repo and
|
||||
fails closed on the remote-wide default when coordinates are omitted."""
|
||||
|
||||
def setUp(self):
|
||||
self._remotes = patch.dict(mcp_server.REMOTES, {
|
||||
"prgs": {
|
||||
"host": "gitea.prgs.cc",
|
||||
"org": "Scaled-Tech-Consulting",
|
||||
"repo": "Timesheet",
|
||||
},
|
||||
})
|
||||
self._remotes.start()
|
||||
|
||||
def tearDown(self):
|
||||
patch.stopall()
|
||||
|
||||
def _capture_resolution(self, org, repo):
|
||||
"""Drive the live anti-stomp runner and capture the org/repo it resolved
|
||||
and passed to the pure assessor."""
|
||||
passthrough = {
|
||||
"allowed": True, "block": False, "blockers": [], "reasons": [],
|
||||
"exact_next_action": "proceed", "blocker_kind": None, "checks": {},
|
||||
}
|
||||
with patch.dict(
|
||||
os.environ, {"GITEA_TEST_FORCE_ANTI_STOMP": "1"}, clear=False
|
||||
), patch(
|
||||
"mcp_server._local_git_remote_url", return_value=LOCAL_GITEA_TOOLS_URL
|
||||
), patch(
|
||||
"mcp_server.get_profile", return_value=RECONCILER
|
||||
), patch.object(
|
||||
mcp_server.anti_stomp_preflight,
|
||||
"assess_anti_stomp_preflight",
|
||||
return_value=passthrough,
|
||||
) as m:
|
||||
mcp_server._run_anti_stomp_preflight(
|
||||
"delete_branch", remote="prgs", org=org, repo=repo
|
||||
)
|
||||
self.assertTrue(m.called)
|
||||
return m.call_args.kwargs
|
||||
|
||||
def test_explicit_gitea_tools_resolved_and_marked_explicit(self):
|
||||
kw = self._capture_resolution("Scaled-Tech-Consulting", "Gitea-Tools")
|
||||
self.assertEqual(kw["resolved_org"], "Scaled-Tech-Consulting")
|
||||
self.assertEqual(kw["resolved_repo"], "Gitea-Tools")
|
||||
self.assertTrue(kw["org_explicit"])
|
||||
self.assertTrue(kw["repo_explicit"])
|
||||
|
||||
def test_missing_coords_resolve_remote_default_and_fail_closed(self):
|
||||
# Omitted coords resolve the remote-wide Timesheet default with
|
||||
# org/repo NOT explicit; the remote/repo guard then blocks against the
|
||||
# local Gitea-Tools remote — i.e. missing coordinates fail closed.
|
||||
kw = self._capture_resolution(None, None)
|
||||
self.assertEqual(kw["resolved_repo"], "Timesheet")
|
||||
self.assertFalse(kw["org_explicit"])
|
||||
self.assertFalse(kw["repo_explicit"])
|
||||
# Compose the guard exactly as the assessor would: Timesheet default,
|
||||
# not explicit, local remote is Gitea-Tools → blocked (wrong_repo).
|
||||
assessment = remote_repo_guard.assess_remote_repo_match(
|
||||
remote="prgs",
|
||||
resolved_org=kw["resolved_org"],
|
||||
resolved_repo=kw["resolved_repo"],
|
||||
local_remote_url=LOCAL_GITEA_TOOLS_URL,
|
||||
org_explicit=kw["org_explicit"],
|
||||
repo_explicit=kw["repo_explicit"],
|
||||
)
|
||||
self.assertTrue(assessment["block"])
|
||||
|
||||
|
||||
class TestCapabilityRoleMapConsistency(unittest.TestCase):
|
||||
"""AC: task-capability and role-routing maps remain consistent (#729)."""
|
||||
|
||||
def test_delete_branch_is_reconciler_in_both_maps(self):
|
||||
self.assertEqual(
|
||||
task_capability_map.required_role("delete_branch"), "reconciler"
|
||||
)
|
||||
self.assertEqual(
|
||||
task_capability_map.required_permission("delete_branch"),
|
||||
"gitea.branch.delete",
|
||||
)
|
||||
self.assertEqual(
|
||||
role_session_router.required_role_for_task("delete_branch"),
|
||||
"reconciler",
|
||||
)
|
||||
self.assertEqual(
|
||||
task_capability_map.required_role("delete_branch"),
|
||||
role_session_router.TASK_REQUIRED_ROLE["delete_branch"],
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user