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

fix(mcp): enforce role capability invariants (Closes #723)

Browse Source
This commit is contained in:
sysadmin
2026-07-19 23:21:54 -04:00
parent fcf6981b1b
commit 4b8a9219d8
5 changed files with 561 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_issue_723_role_stamp_and_submission.py
+308
View File
@@ -0,0 +1,308 @@
"""Regression coverage for issue #723 role and capability invariants."""
from __future__ import annotations
import unittest
from unittest.mock import patch
import gitea_mcp_server as mcp_server
import task_capability_map
REVIEWER_PROFILE = {
"profile_name": "prgs-reviewer",
"role": "reviewer",
"allowed_operations": [
"gitea.read",
"gitea.pr.review",
"gitea.pr.approve",
"gitea.pr.request_changes",
"gitea.pr.comment",
"gitea.issue.comment",
],
"forbidden_operations": [
"gitea.branch.create",
"gitea.branch.push",
"gitea.repo.commit",
"gitea.pr.create",
"gitea.pr.merge",
],
}
CONFIG = {
"profiles": {
"prgs-reviewer": {
"role": "reviewer",
"allowed_operations": REVIEWER_PROFILE["allowed_operations"],
"forbidden_operations": REVIEWER_PROFILE["forbidden_operations"],
},
"prgs-merger": {
"role": "merger",
"allowed_operations": [
"gitea.read",
"gitea.pr.merge",
"gitea.pr.comment",
"gitea.issue.comment",
],
"forbidden_operations": [
"gitea.pr.approve",
"gitea.pr.review",
"gitea.pr.request_changes",
],
},
}
}
def _reset_preflight() -> None:
mcp_server._clear_preflight_capability_state()
mcp_server._preflight_whoami_called = False
mcp_server._preflight_whoami_violation = False
mcp_server.capability_stop_terminal.clear()
mcp_server.role_session_router.clear_route_state()
class _ResolveHarness(unittest.TestCase):
def setUp(self):
_reset_preflight()
def tearDown(self):
_reset_preflight()
def _resolve(
self,
task,
profile=REVIEWER_PROFILE,
required_role=None,
init_side_effect=None,
):
patches = [
patch.object(mcp_server, "get_profile", return_value=profile),
patch.object(
mcp_server.gitea_config, "load_config", return_value=CONFIG
),
patch.object(
mcp_server, "_authenticated_username", return_value="tester"
),
patch.object(
mcp_server,
"init_review_decision_lock",
return_value=None,
side_effect=init_side_effect,
),
patch.object(
mcp_server, "record_mutation_authority", return_value=None
),
patch.object(
mcp_server, "_check_mcp_runtimes_diagnostics", return_value=[]
),
]
if required_role is not None:
patches.append(
patch.object(
mcp_server.task_capability_map,
"required_role",
side_effect=lambda candidate: (
required_role
if candidate == task
else task_capability_map.TASK_CAPABILITY_MAP[candidate][
"role"
]
),
)
)
for context in patches:
context.__enter__()
try:
return mcp_server.gitea_resolve_task_capability(
task=task, remote="prgs"
)
finally:
for context in reversed(patches):
context.__exit__(None, None, None)
class TestCapabilityRoleStampSafety(_ResolveHarness):
def test_allowed_resolution_records_the_correct_stamp(self):
result = self._resolve("review_pr")
self.assertTrue(result["allowed_in_current_session"], result)
self.assertEqual(mcp_server._preflight_resolved_role, "reviewer")
self.assertEqual(mcp_server._preflight_resolved_task, "review_pr")
def test_denied_resolution_records_no_stamp(self):
with patch.object(
mcp_server,
"record_preflight_check",
wraps=mcp_server.record_preflight_check,
) as record:
result = self._resolve("review_pr", required_role="merger")
self.assertFalse(result["allowed_in_current_session"], result)
stamped_calls = [
call
for call in record.call_args_list
if len(call.args) > 1 and call.args[1] is not None
]
self.assertEqual(
stamped_calls,
[],
"a denied resolution must never transiently record a role stamp",
)
self.assertIsNone(mcp_server._preflight_resolved_role)
self.assertIsNone(mcp_server._preflight_resolved_task)
def test_denied_resolution_clears_an_existing_stamp(self):
allowed = self._resolve("review_pr")
self.assertTrue(allowed["allowed_in_current_session"], allowed)
self.assertEqual(mcp_server._preflight_resolved_role, "reviewer")
denied = self._resolve("merge_pr")
self.assertFalse(denied["allowed_in_current_session"], denied)
self.assertIsNone(mcp_server._preflight_resolved_role)
self.assertIsNone(mcp_server._preflight_resolved_task)
def test_denial_cannot_poison_a_later_allowed_task(self):
denied = self._resolve("merge_pr")
self.assertFalse(denied["allowed_in_current_session"], denied)
allowed = self._resolve("review_pr")
self.assertTrue(allowed["allowed_in_current_session"], allowed)
self.assertEqual(mcp_server._preflight_resolved_role, "reviewer")
self.assertEqual(mcp_server._preflight_resolved_task, "review_pr")
def test_unexpected_resolver_failure_leaves_no_stamp(self):
with self.assertRaisesRegex(RuntimeError, "malformed decision state"):
self._resolve(
"review_pr",
init_side_effect=RuntimeError("malformed decision state"),
)
self.assertIsNone(mcp_server._preflight_resolved_role)
self.assertIsNone(mcp_server._preflight_resolved_task)
class TestStructuredWorkspaceRoleFailures(unittest.TestCase):
def test_review_submission_returns_workspace_role_binding_failure(self):
error = RuntimeError(
"namespace workspace binding blocked: merger role in reviewer workspace"
)
with patch.object(
mcp_server, "_verify_role_mutation_workspace", side_effect=error
):
result = mcp_server._evaluate_pr_review_submission(
pr_number=721,
action="approve",
expected_head_sha="8" * 40,
remote="prgs",
live=True,
final_review_decision_ready=True,
)
self.assertFalse(result["performed"])
self.assertEqual(result["blocker_kind"], "workspace_role_binding")
self.assertTrue(
any("workspace/role binding failed" in reason for reason in result["reasons"]),
result,
)
self.assertTrue(any("merger role" in reason for reason in result["reasons"]))
def test_adopt_merger_lease_returns_workspace_role_binding_failure(self):
error = RuntimeError("merger workspace binding rejected")
with patch.object(
mcp_server, "_profile_operation_gate", return_value=[]
), patch.object(
mcp_server, "_verify_role_mutation_workspace", side_effect=error
), patch.object(mcp_server, "_resolve") as resolve:
result = mcp_server.gitea_adopt_merger_pr_lease(
pr_number=718,
worktree="branches/merge-pr-718",
expected_head_sha="7" * 40,
remote="prgs",
)
self.assertFalse(result["success"])
self.assertFalse(result["adopted"])
self.assertEqual(result["blocker_kind"], "workspace_role_binding")
self.assertEqual(result["pr_number"], 718)
self.assertEqual(result["expected_head_sha"], "7" * 40)
self.assertIsNone(result["live_head_sha"])
self.assertTrue(any("binding rejected" in reason for reason in result["reasons"]))
resolve.assert_not_called()
def test_unexpected_verifier_failure_remains_fail_closed(self):
with patch.object(
mcp_server,
"_verify_role_mutation_workspace",
side_effect=ValueError("unexpected verifier state"),
), patch.object(mcp_server, "_resolve") as resolve:
with self.assertRaisesRegex(ValueError, "unexpected verifier state"):
mcp_server._evaluate_pr_review_submission(
pr_number=721,
action="approve",
remote="prgs",
live=True,
)
resolve.assert_not_called()
class TestRuntimeCapabilityRoleFiltering(unittest.TestCase):
def test_runtime_role_filter_denies_permission_bearing_wrong_role(self):
allowed = REVIEWER_PROFILE["allowed_operations"] + ["gitea.pr.merge"]
capabilities = mcp_server._build_runtime_task_capabilities(
allowed,
[],
CONFIG,
remote="prgs",
active_role_kind="reviewer",
)
merge_entry = next(
item
for item in capabilities["task_capabilities"]
if item["task"] == "merge_pr"
)
self.assertTrue(merge_entry["role_exclusive"])
self.assertEqual(merge_entry["capability_view"], "role_filtered")
self.assertFalse(merge_entry["allowed_in_current_session"])
self.assertFalse(capabilities["can_merge_prs"])
def test_permission_only_view_is_explicit(self):
capabilities = mcp_server._build_runtime_task_capabilities(
["gitea.read", "gitea.pr.merge"],
[],
CONFIG,
active_role_kind=None,
)
merge_entry = next(
item
for item in capabilities["task_capabilities"]
if item["task"] == "merge_pr"
)
self.assertEqual(merge_entry["capability_view"], "permission_only")
self.assertTrue(merge_entry["allowed_in_current_session"])
def test_matching_profiles_honor_declared_roles(self):
capabilities = mcp_server._build_runtime_task_capabilities(
["gitea.read"],
[],
CONFIG,
active_role_kind="author",
)
review_entry = next(
item
for item in capabilities["task_capabilities"]
if item["task"] == "review_pr"
)
merge_entry = next(
item
for item in capabilities["task_capabilities"]
if item["task"] == "merge_pr"
)
self.assertEqual(
review_entry["matching_configured_profiles"], ["prgs-reviewer"]
)
self.assertEqual(
merge_entry["matching_configured_profiles"], ["prgs-merger"]
)
if __name__ == "__main__":
unittest.main()
tests/test_resolve_task_capability.py
+26
View File
@@ -242,6 +242,32 @@ class TestResolveTaskCapability(unittest.TestCase):
self.assertTrue(res.get("stop_required"))
self.assertIs(res.get("mutation_performed"), False)
@patch("mcp_server.api_request", return_value={"login": "author-user"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_denied_role_exclusive_resolution_does_not_stamp_role(
self, _auth, _api
):
with patch.dict(os.environ, self._env("author-profile")):
with patch.object(
mcp_server,
"record_preflight_check",
wraps=mcp_server.record_preflight_check,
) as record:
result = mcp_server.gitea_resolve_task_capability(
task="review_pr", remote="prgs"
)
self.assertFalse(result["allowed_in_current_session"], result)
self.assertFalse(
any(
len(call.args) > 1 and call.args[1] == "reviewer"
for call in record.call_args_list
),
"denied reviewer resolution must never record a reviewer stamp",
)
self.assertIsNone(mcp_server._preflight_resolved_role)
self.assertIsNone(mcp_server._preflight_resolved_task)
# Additional regression tests per #145 for permission boundaries and structured guidance
def test_issue_comment_does_not_imply_close(self):
# Author profile has issue.comment but not issue.close
tests/test_task_capability_role_invariants.py
+67 -1
View File
@@ -19,7 +19,12 @@ import unittest
import gitea_config
from role_session_router import MERGER_TASKS, REVIEWER_TASKS
from task_capability_map import required_permission, required_role
from task_capability_map import (
ROLE_EXCLUSIVE_TASKS,
TASK_CAPABILITY_MAP,
required_permission,
required_role,
)
# Canonical role-profile permission shape. Mirrors the configured
# author/reviewer/merger/reconciler profiles (profiles.json v2 role split):
@@ -112,6 +117,42 @@ FORMAL_REVIEW_TASKS = (
"pr-queue-cleanup",
)
# Complete resolver role-exclusive set on master when #723 was reconstructed.
# The shared constant must replace this exact inline authority without dropping
# later lease and PR-sync aliases added after the preserved source commits.
EXPECTED_ROLE_EXCLUSIVE_TASKS = frozenset(
{
"acquire_reviewer_pr_lease",
"gitea_acquire_reviewer_pr_lease",
"review_pr",
"approve_pr",
"request_changes_pr",
"blind_pr_queue_review",
"pr_queue_cleanup",
"pr-queue-cleanup",
"merge_pr",
"acquire_merger_pr_lease",
"gitea_acquire_merger_pr_lease",
"adopt_merger_pr_lease",
"gitea_adopt_merger_pr_lease",
"release_merger_pr_lease",
"gitea_release_merger_pr_lease",
"create_branch",
"push_branch",
"create_pr",
"commit_files",
"gitea_commit_files",
"address_pr_change_requests",
"update_pr_branch_by_merge",
"gitea_update_pr_branch_by_merge",
"delete_branch",
"cleanup_merged_pr_branch",
"reconciliation_cleanup",
"work_issue",
"work-issue",
}
)
def _profile_satisfies(role_name, task):
"""True when the canonical *role_name* profile can perform *task*."""
@@ -201,5 +242,30 @@ class TestMergerBoundary(unittest.TestCase):
)
class TestRoleExclusiveSetIntegrity(unittest.TestCase):
"""#723: the shared set is complete, mapped, and role-satisfiable."""
def test_complete_current_role_exclusive_set(self):
self.assertEqual(ROLE_EXCLUSIVE_TASKS, EXPECTED_ROLE_EXCLUSIVE_TASKS)
def test_every_role_exclusive_task_exists_in_capability_map(self):
for task in sorted(ROLE_EXCLUSIVE_TASKS):
with self.subTest(task=task):
self.assertIn(task, TASK_CAPABILITY_MAP)
def test_formal_review_tasks_are_role_exclusive(self):
self.assertTrue(set(FORMAL_REVIEW_TASKS) <= ROLE_EXCLUSIVE_TASKS)
def test_every_role_exclusive_task_has_a_satisfying_profile(self):
for task in sorted(ROLE_EXCLUSIVE_TASKS):
with self.subTest(task=task):
role = required_role(task)
self.assertIn(role, CANONICAL_ROLE_PROFILES)
self.assertTrue(
_profile_satisfies(role, task),
f"canonical {role!r} profile cannot satisfy {task!r}",
)
if __name__ == "__main__":
unittest.main()