fix: preserve actual reconciler role for #274/#475 role exemptions during comment_issue preflight (Closes #540)
resolve_task_capability("comment_issue") stamps _preflight_resolved_role
= "author" because comment_issue.required_role_kind = author.
_effective_workspace_role() prefers that stamp, so a genuine
prgs-reconciler session was classified as an author inside the #274
branch-only mutation guard and the #475 root checkout guard, blocking
gitea_create_issue_comment from the control checkout.
Fix keys the role exemptions off the actual active profile role as well
as the effective workspace role:
- Add _actual_profile_role(): derives the workspace role from the active
profile alone, never from _preflight_resolved_role.
- _enforce_branches_only_author_mutation: exempt when EITHER the
effective role OR the actual profile role is a non-author role.
- _enforce_root_checkout_guard: pass actual_role; assess_root_checkout_guard
now exempts a reconciler by resolved OR actual role.
Author blocking is preserved: an actual author profile classifies as
author in both signals, so it stays blocked on a contaminated control
checkout. Merger strictness stays keyed on the resolved task role so a
merger operating from its clean branches/ workspace under a non-merge
task is not over-tightened.
Regression tests (tests/test_issue_540_comment_role_poison.py) cover the
reconciler comment path, author-blocking path, reviewer/merger/reconciler
role classification under a poisoned author stamp, and the root guard
actual_role exemption.
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
+35
-2
@@ -211,6 +211,28 @@ def _effective_workspace_role() -> str:
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _actual_profile_role() -> str:
|
||||||
|
"""Resolve the workspace role from the ACTIVE PROFILE alone (#540).
|
||||||
|
|
||||||
|
Unlike :func:`_effective_workspace_role`, this never consults
|
||||||
|
``_preflight_resolved_role``. A task capability whose ``required_role_kind``
|
||||||
|
is ``author`` (e.g. ``comment_issue``) stamps ``_preflight_resolved_role =
|
||||||
|
"author"``; the #274/#475 role exemptions must key off the real profile
|
||||||
|
identity so that stamp cannot poison a genuine reviewer/merger/reconciler
|
||||||
|
session into being treated as an author. Author profiles still classify as
|
||||||
|
``author`` here, so author blocking is preserved.
|
||||||
|
"""
|
||||||
|
profile = get_profile()
|
||||||
|
role = _role_kind(
|
||||||
|
profile.get("allowed_operations") or [],
|
||||||
|
profile.get("forbidden_operations") or [],
|
||||||
|
)
|
||||||
|
return nwb.normalize_role_kind(
|
||||||
|
role,
|
||||||
|
profile_name=profile.get("profile_name"),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
|
def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
|
||||||
"""Resolve the namespace-scoped workspace root inspected by pre-flight guards."""
|
"""Resolve the namespace-scoped workspace root inspected by pre-flight guards."""
|
||||||
role = _effective_workspace_role()
|
role = _effective_workspace_role()
|
||||||
@@ -541,9 +563,19 @@ def _enforce_branches_only_author_mutation(worktree_path: str | None = None) ->
|
|||||||
Reviewer, merger, and reconciler roles are exempt: reconciler ``close_pr``
|
Reviewer, merger, and reconciler roles are exempt: reconciler ``close_pr``
|
||||||
is a Gitea metadata mutation and must not require ``GITEA_AUTHOR_WORKTREE``
|
is a Gitea metadata mutation and must not require ``GITEA_AUTHOR_WORKTREE``
|
||||||
(#468). Non-author namespaces use dedicated workspace env vars (#510).
|
(#468). Non-author namespaces use dedicated workspace env vars (#510).
|
||||||
|
|
||||||
|
The exemption honours BOTH the effective workspace role and the actual
|
||||||
|
profile role (#540). ``comment_issue`` preflight stamps
|
||||||
|
``_preflight_resolved_role = "author"`` (its ``required_role_kind``), which
|
||||||
|
would otherwise poison :func:`_effective_workspace_role` into classifying a
|
||||||
|
genuine reconciler as an author and defeat this exemption. Keying off the
|
||||||
|
real profile role as well preserves the exemption without weakening author
|
||||||
|
blocking — an actual author profile classifies as ``author`` in both.
|
||||||
"""
|
"""
|
||||||
role = _effective_workspace_role()
|
if (
|
||||||
if role in nwb.NON_AUTHOR_ROLES:
|
_effective_workspace_role() in nwb.NON_AUTHOR_ROLES
|
||||||
|
or _actual_profile_role() in nwb.NON_AUTHOR_ROLES
|
||||||
|
):
|
||||||
return
|
return
|
||||||
ctx = _resolve_namespace_mutation_context(worktree_path)
|
ctx = _resolve_namespace_mutation_context(worktree_path)
|
||||||
workspace = ctx["workspace_path"]
|
workspace = ctx["workspace_path"]
|
||||||
@@ -710,6 +742,7 @@ def _enforce_root_checkout_guard(worktree_path: str | None = None) -> None:
|
|||||||
porcelain_status=git_state.get("porcelain_status") or "",
|
porcelain_status=git_state.get("porcelain_status") or "",
|
||||||
remote_master_sha=remote_master_sha,
|
remote_master_sha=remote_master_sha,
|
||||||
resolved_role=_preflight_resolved_role,
|
resolved_role=_preflight_resolved_role,
|
||||||
|
actual_role=_actual_profile_role(),
|
||||||
)
|
)
|
||||||
if assessment["block"]:
|
if assessment["block"]:
|
||||||
raise RuntimeError(root_checkout_guard.format_root_checkout_guard_error(assessment))
|
raise RuntimeError(root_checkout_guard.format_root_checkout_guard_error(assessment))
|
||||||
|
|||||||
+17
-2
@@ -58,15 +58,30 @@ def assess_root_checkout_guard(
|
|||||||
porcelain_status: str,
|
porcelain_status: str,
|
||||||
remote_master_sha: str | None,
|
remote_master_sha: str | None,
|
||||||
resolved_role: str | None = None,
|
resolved_role: str | None = None,
|
||||||
|
actual_role: str | None = None,
|
||||||
) -> dict:
|
) -> dict:
|
||||||
"""Fail closed when the control checkout is not clean master/prgs/master."""
|
"""Fail closed when the control checkout is not clean master/prgs/master.
|
||||||
|
|
||||||
|
``resolved_role`` is the preflight-resolved *task* role and ``actual_role``
|
||||||
|
is the *active profile* role (#540). The reconciler exemption honours either
|
||||||
|
signal so a ``comment_issue`` preflight (which stamps the task role as
|
||||||
|
``author``) cannot strip a genuine reconciler of its exemption. An actual
|
||||||
|
author profile classifies as ``author`` in both signals, so author blocking
|
||||||
|
on a contaminated control checkout is preserved.
|
||||||
|
|
||||||
|
Merger *strictness* (a merger must not be auto-exempted by working from a
|
||||||
|
``branches/`` worktree) stays keyed on the resolved task role: merge
|
||||||
|
operations resolve their own task role, and widening the merger test with
|
||||||
|
``actual_role`` would wrongly subject a merger operating from its clean
|
||||||
|
workspace under a non-merge task to full control-checkout checks.
|
||||||
|
"""
|
||||||
reasons: list[str] = []
|
reasons: list[str] = []
|
||||||
root = os.path.realpath(canonical_repo_root)
|
root = os.path.realpath(canonical_repo_root)
|
||||||
workspace = os.path.realpath(workspace_path)
|
workspace = os.path.realpath(workspace_path)
|
||||||
branch = (current_branch or "").strip()
|
branch = (current_branch or "").strip()
|
||||||
dirty_files = parse_dirty_tracked_files(porcelain_status)
|
dirty_files = parse_dirty_tracked_files(porcelain_status)
|
||||||
|
|
||||||
if resolved_role == "reconciler":
|
if resolved_role == "reconciler" or actual_role == "reconciler":
|
||||||
return _assessment(True, [], root, workspace, branch, head_sha, dirty_files)
|
return _assessment(True, [], root, workspace, branch, head_sha, dirty_files)
|
||||||
|
|
||||||
if resolved_role != "merger" and is_path_under_branches(workspace, root):
|
if resolved_role != "merger" and is_path_under_branches(workspace, root):
|
||||||
|
|||||||
@@ -0,0 +1,271 @@
|
|||||||
|
"""Regression tests for #540: comment_issue preflight must not poison the
|
||||||
|
actual reconciler role for #274 branch-only / #475 root-checkout exemptions.
|
||||||
|
|
||||||
|
`resolve_task_capability("comment_issue")` stamps
|
||||||
|
``_preflight_resolved_role = "author"`` because ``comment_issue`` has
|
||||||
|
``required_role_kind = author``. Before the fix, ``_effective_workspace_role``
|
||||||
|
preferred that stamp and a genuine ``prgs-reconciler`` session was treated as an
|
||||||
|
author inside the #274 branch-only mutation guard and the #475 root checkout
|
||||||
|
guard, blocking ``gitea_create_issue_comment`` from the control checkout.
|
||||||
|
|
||||||
|
The fix keys the role exemptions off the *actual profile role* as well, so the
|
||||||
|
exemption survives the poisoned task role while author profiles stay blocked.
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
import unittest
|
||||||
|
from pathlib import Path
|
||||||
|
from unittest.mock import patch
|
||||||
|
|
||||||
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||||
|
|
||||||
|
import gitea_mcp_server as srv # noqa: E402
|
||||||
|
import root_checkout_guard as rcg # noqa: E402
|
||||||
|
|
||||||
|
FAKE_AUTH = "token test"
|
||||||
|
CONTROL_CHECKOUT_ROOT = str(Path(__file__).resolve().parents[3])
|
||||||
|
MASTER_SHA = "a" * 40
|
||||||
|
OTHER_SHA = "b" * 40
|
||||||
|
|
||||||
|
RECONCILER_PROFILE = {
|
||||||
|
"profile_name": "prgs-reconciler",
|
||||||
|
"allowed_operations": [
|
||||||
|
"gitea.read",
|
||||||
|
"gitea.pr.close",
|
||||||
|
"gitea.pr.comment",
|
||||||
|
"gitea.issue.comment",
|
||||||
|
],
|
||||||
|
"forbidden_operations": [
|
||||||
|
"gitea.pr.approve",
|
||||||
|
"gitea.pr.merge",
|
||||||
|
"gitea.pr.create",
|
||||||
|
"gitea.branch.push",
|
||||||
|
"gitea.repo.commit",
|
||||||
|
],
|
||||||
|
"audit_label": "prgs-reconciler",
|
||||||
|
}
|
||||||
|
|
||||||
|
AUTHOR_PROFILE = {
|
||||||
|
"profile_name": "prgs-author",
|
||||||
|
"allowed_operations": [
|
||||||
|
"gitea.read",
|
||||||
|
"gitea.issue.comment",
|
||||||
|
"gitea.issue.create",
|
||||||
|
"gitea.pr.create",
|
||||||
|
"gitea.branch.push",
|
||||||
|
"gitea.repo.commit",
|
||||||
|
],
|
||||||
|
"forbidden_operations": [
|
||||||
|
"gitea.pr.approve",
|
||||||
|
"gitea.pr.merge",
|
||||||
|
],
|
||||||
|
"audit_label": "prgs-author",
|
||||||
|
}
|
||||||
|
|
||||||
|
REVIEWER_PROFILE = {
|
||||||
|
"profile_name": "prgs-reviewer",
|
||||||
|
"allowed_operations": ["gitea.read", "gitea.pr.approve", "gitea.pr.merge"],
|
||||||
|
"forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
|
||||||
|
"audit_label": "prgs-reviewer",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
class TestActualProfileRole(unittest.TestCase):
|
||||||
|
"""`_actual_profile_role` ignores the poisoned preflight task role (#540)."""
|
||||||
|
|
||||||
|
def tearDown(self):
|
||||||
|
srv._preflight_resolved_role = None
|
||||||
|
|
||||||
|
@patch("gitea_mcp_server.get_profile", return_value=RECONCILER_PROFILE)
|
||||||
|
def test_reconciler_profile_role_survives_author_task_stamp(self, _profile):
|
||||||
|
srv._preflight_resolved_role = "author" # comment_issue poison
|
||||||
|
self.assertEqual(srv._actual_profile_role(), "reconciler")
|
||||||
|
# _effective_workspace_role is still poisoned to author (unchanged #510)...
|
||||||
|
self.assertEqual(srv._effective_workspace_role(), "author")
|
||||||
|
|
||||||
|
@patch("gitea_mcp_server.get_profile", return_value=AUTHOR_PROFILE)
|
||||||
|
def test_author_profile_role_is_author(self, _profile):
|
||||||
|
srv._preflight_resolved_role = "author"
|
||||||
|
self.assertEqual(srv._actual_profile_role(), "author")
|
||||||
|
|
||||||
|
@patch("gitea_mcp_server.get_profile", return_value=REVIEWER_PROFILE)
|
||||||
|
def test_reviewer_profile_role_is_reviewer(self, _profile):
|
||||||
|
srv._preflight_resolved_role = "author"
|
||||||
|
self.assertEqual(srv._actual_profile_role(), "reviewer")
|
||||||
|
|
||||||
|
|
||||||
|
class TestBranchesOnlyExemptionRealRole(unittest.TestCase):
|
||||||
|
"""#274 branches-only exemption keys off the actual profile role (#540)."""
|
||||||
|
|
||||||
|
def tearDown(self):
|
||||||
|
srv._preflight_resolved_role = None
|
||||||
|
|
||||||
|
@patch("gitea_mcp_server.get_profile", return_value=RECONCILER_PROFILE)
|
||||||
|
def test_reconciler_exempt_despite_author_task_stamp(self, _profile):
|
||||||
|
srv._preflight_resolved_role = "author" # poison
|
||||||
|
# Must return without raising and without resolving an author worktree.
|
||||||
|
with patch("gitea_mcp_server._resolve_namespace_mutation_context") as ctx:
|
||||||
|
srv._enforce_branches_only_author_mutation()
|
||||||
|
ctx.assert_not_called()
|
||||||
|
|
||||||
|
@patch("gitea_mcp_server.get_profile", return_value=AUTHOR_PROFILE)
|
||||||
|
def test_author_not_exempt(self, _profile):
|
||||||
|
srv._preflight_resolved_role = "author"
|
||||||
|
sentinel = RuntimeError("author-mutation-guard-reached")
|
||||||
|
|
||||||
|
def _blow_up(*_a, **_k):
|
||||||
|
raise sentinel
|
||||||
|
|
||||||
|
# Author is not exempt: the guard proceeds to resolve/assess the
|
||||||
|
# workspace (proven by reaching the patched context resolver).
|
||||||
|
with patch(
|
||||||
|
"gitea_mcp_server._resolve_namespace_mutation_context",
|
||||||
|
side_effect=_blow_up,
|
||||||
|
):
|
||||||
|
with self.assertRaises(RuntimeError) as raised:
|
||||||
|
srv._enforce_branches_only_author_mutation()
|
||||||
|
self.assertIs(raised.exception, sentinel)
|
||||||
|
|
||||||
|
|
||||||
|
class TestRootCheckoutGuardRealRole(unittest.TestCase):
|
||||||
|
"""#475 root guard honours the actual profile role too (#540)."""
|
||||||
|
|
||||||
|
def _assess(self, **kwargs):
|
||||||
|
defaults = {
|
||||||
|
"workspace_path": CONTROL_CHECKOUT_ROOT,
|
||||||
|
"canonical_repo_root": CONTROL_CHECKOUT_ROOT,
|
||||||
|
"current_branch": "feat/some-branch",
|
||||||
|
"head_sha": OTHER_SHA,
|
||||||
|
"porcelain_status": " M gitea_mcp_server.py\n",
|
||||||
|
"remote_master_sha": MASTER_SHA,
|
||||||
|
"resolved_role": "author", # poisoned task role
|
||||||
|
}
|
||||||
|
defaults.update(kwargs)
|
||||||
|
return rcg.assess_root_checkout_guard(**defaults)
|
||||||
|
|
||||||
|
def test_actual_reconciler_exempt_despite_poisoned_resolved_author(self):
|
||||||
|
result = self._assess(actual_role="reconciler")
|
||||||
|
self.assertTrue(result["proven"])
|
||||||
|
self.assertFalse(result["block"])
|
||||||
|
|
||||||
|
def test_actual_author_still_blocked_on_contaminated_root(self):
|
||||||
|
result = self._assess(actual_role="author")
|
||||||
|
self.assertTrue(result["block"])
|
||||||
|
|
||||||
|
def test_merger_strictness_stays_keyed_on_resolved_task_role(self):
|
||||||
|
# When the merge task resolves the merger role, the branches/ auto
|
||||||
|
# exemption is denied and a clean control checkout is required.
|
||||||
|
blocked = self._assess(
|
||||||
|
workspace_path=f"{CONTROL_CHECKOUT_ROOT}/branches/review-pr-1",
|
||||||
|
current_branch="review-pr-1",
|
||||||
|
porcelain_status="",
|
||||||
|
head_sha=OTHER_SHA,
|
||||||
|
resolved_role="merger",
|
||||||
|
)
|
||||||
|
self.assertTrue(blocked["block"])
|
||||||
|
|
||||||
|
def test_actual_merger_does_not_over_tighten_non_merge_task(self):
|
||||||
|
# A merger profile whose current task did NOT resolve to merger keeps
|
||||||
|
# the branches/ workspace exemption (regression guard for #540): the
|
||||||
|
# actual_role signal must not force merger strictness here.
|
||||||
|
result = self._assess(
|
||||||
|
workspace_path=f"{CONTROL_CHECKOUT_ROOT}/branches/review-pr-1",
|
||||||
|
current_branch="review-pr-1",
|
||||||
|
porcelain_status="",
|
||||||
|
head_sha=OTHER_SHA,
|
||||||
|
resolved_role="reviewer",
|
||||||
|
actual_role="merger",
|
||||||
|
)
|
||||||
|
self.assertTrue(result["proven"])
|
||||||
|
self.assertFalse(result["block"])
|
||||||
|
|
||||||
|
def test_backward_compatible_without_actual_role(self):
|
||||||
|
# No actual_role supplied: behaviour falls back to resolved_role.
|
||||||
|
result = self._assess(resolved_role="reconciler")
|
||||||
|
self.assertTrue(result["proven"])
|
||||||
|
|
||||||
|
|
||||||
|
class TestReconcilerCommentThroughCanonicalPath(unittest.TestCase):
|
||||||
|
"""Integration: reconciler comment survives the poisoned author task role."""
|
||||||
|
|
||||||
|
def setUp(self):
|
||||||
|
srv._preflight_whoami_called = True
|
||||||
|
srv._preflight_capability_called = True
|
||||||
|
srv._preflight_whoami_violation = False
|
||||||
|
srv._preflight_capability_violation = False
|
||||||
|
srv._preflight_capability_baseline_porcelain = ""
|
||||||
|
self._orig_in_test = srv._preflight_in_test_mode
|
||||||
|
srv._preflight_in_test_mode = lambda: False
|
||||||
|
|
||||||
|
def tearDown(self):
|
||||||
|
srv._preflight_in_test_mode = self._orig_in_test
|
||||||
|
srv._preflight_resolved_role = None
|
||||||
|
|
||||||
|
@patch("gitea_mcp_server._get_workspace_porcelain", return_value="")
|
||||||
|
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
|
||||||
|
@patch("gitea_mcp_server.api_request")
|
||||||
|
@patch(
|
||||||
|
"gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha",
|
||||||
|
return_value=MASTER_SHA,
|
||||||
|
)
|
||||||
|
@patch(
|
||||||
|
"gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
|
||||||
|
return_value={
|
||||||
|
"current_branch": "master",
|
||||||
|
"head_sha": MASTER_SHA,
|
||||||
|
"porcelain_status": "",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
@patch("gitea_mcp_server.get_profile", return_value=RECONCILER_PROFILE)
|
||||||
|
def test_reconciler_comment_from_control_checkout_succeeds(
|
||||||
|
self, _profile, _git, _remote_sha, mock_api, _auth, _porcelain
|
||||||
|
):
|
||||||
|
srv._preflight_resolved_role = "author" # comment_issue poison
|
||||||
|
mock_api.return_value = {"id": 9001, "html_url": "https://x/y"}
|
||||||
|
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
|
||||||
|
with patch.dict(os.environ, {}, clear=False):
|
||||||
|
os.environ.pop("GITEA_AUTHOR_WORKTREE", None)
|
||||||
|
os.environ.pop("GITEA_ACTIVE_WORKTREE", None)
|
||||||
|
os.environ.pop("GITEA_RECONCILER_WORKTREE", None)
|
||||||
|
result = srv.gitea_create_issue_comment(
|
||||||
|
515, "canonical reconciler audit", remote="prgs"
|
||||||
|
)
|
||||||
|
self.assertTrue(result["success"])
|
||||||
|
self.assertEqual(result["comment_id"], 9001)
|
||||||
|
mock_api.assert_called_once()
|
||||||
|
|
||||||
|
@patch("gitea_mcp_server._get_workspace_porcelain", return_value="")
|
||||||
|
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
|
||||||
|
@patch("gitea_mcp_server.api_request")
|
||||||
|
@patch(
|
||||||
|
"gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha",
|
||||||
|
return_value=MASTER_SHA,
|
||||||
|
)
|
||||||
|
@patch(
|
||||||
|
"gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
|
||||||
|
return_value={
|
||||||
|
"current_branch": "master",
|
||||||
|
"head_sha": MASTER_SHA,
|
||||||
|
"porcelain_status": "",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
@patch("gitea_mcp_server.get_profile", return_value=AUTHOR_PROFILE)
|
||||||
|
def test_author_comment_from_control_checkout_blocked(
|
||||||
|
self, _profile, _git, _remote_sha, mock_api, _auth, _porcelain
|
||||||
|
):
|
||||||
|
srv._preflight_resolved_role = "author"
|
||||||
|
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
|
||||||
|
with patch.dict(os.environ, {}, clear=False):
|
||||||
|
os.environ.pop("GITEA_AUTHOR_WORKTREE", None)
|
||||||
|
os.environ.pop("GITEA_ACTIVE_WORKTREE", None)
|
||||||
|
with self.assertRaises(RuntimeError):
|
||||||
|
srv.gitea_create_issue_comment(
|
||||||
|
515, "author note", remote="prgs"
|
||||||
|
)
|
||||||
|
mock_api.assert_not_called()
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
Reference in New Issue
Block a user