Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 74 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request 'fix(reconciler): forward explicit repository through delete_branch anti-stomp preflight (Closes #733)' (#734) from fix/issue-733-delete-branch-repo-forwarding into master

Browse Source
This commit was merged in pull request #734.
This commit is contained in:
sysadmin
2026-07-17 22:03:24 -05:00
parent daf60266d0 3990fc684f
commit 11d1d2e99f
2 changed files with 467 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+82 -1
View File
@@ -8562,6 +8562,67 @@ def gitea_review_pr(
return out
def _delete_branch_repository_binding_block(
remote: str | None,
*,
org: str | None,
repo: str | None,
) -> dict | None:
"""#733: validate an explicit delete target against the workspace binding.
The trusted repository identity comes only from the verified,
workspace-aligned git remote (never ``REMOTES`` defaults, never
unvalidated caller input). Explicit ``org``/``repo`` are a *request target*
checked against that binding:
* matching or omitted coordinates pass — omitted targets are still resolved
and re-validated by the shared #604 anti-stomp preflight, which fails
closed on a remote-wide default mismatch (bare prgs → Timesheet);
* coordinates that disagree with the binding are rejected (wrong or
substituted repository);
* explicit coordinates with no establishable workspace identity fail closed
(unproven target).
Returns a structured block dict, or ``None`` when the target is authorized.
"""
workspace_slug = _workspace_repository_slug(remote)
workspace_parts = (
session_ctx.parse_repository_slug(workspace_slug) if workspace_slug else None
)
if workspace_parts:
override = session_ctx.assess_repository_override(
requested_org=org,
requested_repo=repo,
bound_org=workspace_parts[0],
bound_repo=workspace_parts[1],
)
if override.get("block"):
return {
"success": False,
"performed": False,
"required_permission": "gitea.branch.delete",
"reasons": list(override.get("reasons") or [
"delete target repository does not match the workspace "
"binding (fail closed)"
]),
"blocker_kind": "repository_binding",
}
return None
if org is not None or repo is not None:
return {
"success": False,
"performed": False,
"required_permission": "gitea.branch.delete",
"reasons": [
"repository binding unverified: no workspace repository "
"identity could be established to corroborate the explicit "
"delete target (fail closed)"
],
"blocker_kind": "repository_binding",
}
return None
@mcp.tool()
def gitea_delete_branch(
branch: str,
@@ -8652,8 +8713,28 @@ def gitea_delete_branch(
"audit_phase": audit_reconciliation_mode.current_phase(),
}
# #733: the explicitly targeted repository must agree with the immutable,
# workspace-derived repository identity before any preflight or deletion.
# This rejects wrong or substituted coordinates (e.g. an explicit Timesheet
# target while the workspace is bound to Gitea-Tools) and fails closed when
# explicit coordinates cannot be corroborated against a trusted workspace
# identity. It is independent of the anti-stomp remote/repo guard, which by
# the #530 contract trusts explicit caller intent; here explicit intent is
# validated, not trusted. REMOTES defaults are never consulted as an
# authorization scope (#530/#714).
repo_binding_block = _delete_branch_repository_binding_block(
remote, org=org, repo=repo,
)
if repo_binding_block:
return repo_binding_block
with _mutation_stage("preflight_purity"):
verify_preflight_purity(remote, task="delete_branch")
# #733: forward the explicit org/repo through verify_preflight_purity so
# the shared #604 anti-stomp resolution validates the *targeted*
# repository instead of falling back to the remote-wide REMOTES default
# (bare prgs → Timesheet), which false-positives wrong_repo and blocked
# every explicit Scaled-Tech-Consulting/Gitea-Tools deletion.
verify_preflight_purity(remote, task="delete_branch", org=org, repo=repo)
with _mutation_stage("resolve_auth"):
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
tests/test_issue_733_delete_branch_repo_forwarding.py
+385
View File
@@ -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()