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

Merge pull request 'fix: enforce merger role boundaries (Closes #539)' (#596) from fix/issue-539-role-boundary-hard-stops into master

Browse Source
This commit was merged in pull request #596.
This commit is contained in:
sysadmin
2026-07-09 17:07:54 -05:00
parent a32c68e24b 2941ec529c
commit 2ac7519792
4 changed files with 175 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+63 -6
View File
@@ -222,6 +222,21 @@ def _effective_workspace_role() -> str:
)
def _profile_role_kind(profile: dict) -> str:
"""Resolve a profile's declared role before inferring from permissions."""
role = (profile.get("role") or profile.get("role_kind") or "").strip()
if role:
return role
profile_name = (profile.get("profile_name") or "").strip().lower()
for candidate in ("reconciler", "merger", "reviewer", "author"):
if candidate in profile_name:
return candidate
return _role_kind(
profile.get("allowed_operations") or [],
profile.get("forbidden_operations") or [],
)
def _actual_profile_role() -> str:
"""Resolve the workspace role from the ACTIVE PROFILE alone (#540).
@@ -234,10 +249,7 @@ def _actual_profile_role() -> str:
``author`` here, so author blocking is preserved.
"""
profile = get_profile()
role = _role_kind(
profile.get("allowed_operations") or [],
profile.get("forbidden_operations") or [],
)
role = _profile_role_kind(profile)
return nwb.normalize_role_kind(
role,
profile_name=profile.get("profile_name"),
@@ -10255,6 +10267,24 @@ def gitea_resolve_task_capability(
required_permission = task_capability_map.required_permission(task)
required_role = task_capability_map.required_role(task)
role_exclusive_tasks = {
"review_pr",
"approve_pr",
"request_changes_pr",
"blind_pr_queue_review",
"pr_queue_cleanup",
"pr-queue-cleanup",
"merge_pr",
"create_branch",
"push_branch",
"create_pr",
"commit_files",
"gitea_commit_files",
"address_pr_change_requests",
"delete_branch",
"work_issue",
"work-issue",
}
infra_assessment = role_session_router.assess_infra_stop(PROJECT_ROOT)
if required_role == "reviewer":
@@ -10349,11 +10379,26 @@ def gitea_resolve_task_capability(
# Load active permissions
active_allowed = profile.get("allowed_operations") or []
active_forbidden = profile.get("forbidden_operations") or []
active_role_kind = _profile_role_kind(profile)
# Check if allowed in current session
allowed_in_current_session, _ = gitea_config.check_operation(
permission_allowed_in_current_session, _ = gitea_config.check_operation(
required_permission, active_allowed, active_forbidden
)
role_matches_current_session = (
task not in role_exclusive_tasks
or active_role_kind == required_role
)
role_mismatch_reason = None
if not role_matches_current_session:
role_mismatch_reason = (
f"Active profile role '{active_role_kind}' cannot perform "
f"{required_role} task '{task}' even if nearby permissions are "
"present (fail closed)."
)
allowed_in_current_session = (
permission_allowed_in_current_session and role_matches_current_session
)
switching = gitea_config.is_runtime_switching_enabled()
available_in_session = allowed_in_current_session
@@ -10380,7 +10425,13 @@ def gitea_resolve_task_capability(
except Exception:
pass
ok, _ = gitea_config.check_operation(required_permission, p_allowed_n, p_forbidden_n)
if ok:
p_role = (p_data.get("role") or "").strip()
p_role_ok = (
task not in role_exclusive_tasks
or not p_role
or p_role == required_role
)
if ok and p_role_ok:
matching_profiles.append(p_name)
configured = len(matching_profiles) > 0
@@ -10408,6 +10459,8 @@ def gitea_resolve_task_capability(
reason_msg = (
f"No profile configured with permission '{required_permission}'."
)
elif role_mismatch_reason:
reason_msg = role_mismatch_reason
different_namespace_required = False
next_safe_action = "None; ready for operations."
@@ -10439,6 +10492,8 @@ def gitea_resolve_task_capability(
# into author-side mutations.
stop_required = not allowed_in_current_session
task_role_guidance = []
if role_mismatch_reason:
task_role_guidance.append(f"STOP: {role_mismatch_reason}")
if required_role == "reviewer":
if allowed_in_current_session:
task_role_guidance.append(
@@ -10484,7 +10539,9 @@ def gitea_resolve_task_capability(
"required_role_kind": required_role,
"active_profile": profile["profile_name"],
"active_identity": username,
"active_role_kind": active_role_kind,
"active_profile_allowed_operations": active_allowed,
"active_profile_permission_allowed": permission_allowed_in_current_session,
"allowed_in_current_session": allowed_in_current_session,
"available_in_session": available_in_session,
"configured": configured,
role_session_router.py
+28 -2
View File
@@ -55,7 +55,6 @@ def skip_python_scan_walk_root(project_root: str, walk_root: str) -> bool:
REVIEWER_TASKS = frozenset({
"review_pr",
"merge_pr",
"blind_pr_queue_review",
"pr_queue_cleanup",
"pr-queue-cleanup",
@@ -63,6 +62,10 @@ REVIEWER_TASKS = frozenset({
"approve_pr",
})
MERGER_TASKS = frozenset({
"merge_pr",
})
AUTHOR_TASKS = frozenset({
"create_issue",
"comment_issue",
@@ -98,7 +101,7 @@ TASK_REQUIRED_ROLE = {
"address_pr_change_requests": "author",
"delete_branch": "author",
"review_pr": "reviewer",
"merge_pr": "reviewer",
"merge_pr": "merger",
"blind_pr_queue_review": "reviewer",
"pr_queue_cleanup": "reviewer",
"pr-queue-cleanup": "reviewer",
@@ -128,6 +131,10 @@ WRONG_ROLE_RECONCILER_MSG = (
"MCP namespace/profile with exact close capability."
)
WRONG_ROLE_MERGER_MSG = (
"Wrong role/session for merger task. Launch merger MCP namespace."
)
_session_last_route: dict | None = None
@@ -243,6 +250,25 @@ def route_task_session(
_record_route(result)
return result
if required_role == "merger":
result = {
"task_type": task_type,
"required_role": required_role,
"active_role": active_role_kind,
"active_profile": active_profile,
"route_result": ROUTE_WRONG_ROLE,
"downstream_allowed": False,
"reasons": [
WRONG_ROLE_MERGER_MSG,
"Merger tasks cannot run in author or reviewer sessions.",
],
"message": WRONG_ROLE_MERGER_MSG,
"runtime_switching_supported": runtime_switching_supported,
"profile_switch_blocked": not runtime_switching_supported,
}
_record_route(result)
return result
if required_role == "author":
route = ROUTE_TO_AUTHOR
message = (
tests/test_resolve_task_capability.py
+43
View File
@@ -287,6 +287,49 @@ class TestResolveTaskCapability(unittest.TestCase):
self.assertEqual(res["required_operation_permission"], "gitea.pr.merge")
self.assertEqual(res["required_role_kind"], "merger")
@patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
def test_reviewer_role_with_merge_permission_still_cannot_merge(self, _auth, _api):
# #539: exact permission alone is insufficient. The active profile
# role must also be merger for merge_pr.
config = json.loads(json.dumps(CONFIG_RESOLVER))
config["profiles"]["reviewer-with-merge"] = {
"enabled": True,
"context": "ctx",
"role": "reviewer",
"username": "reviewer-user",
"auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
"allowed_operations": [
"gitea.read", "gitea.pr.review", "gitea.pr.approve",
"gitea.pr.merge",
],
"forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
"execution_profile": "reviewer-with-merge",
}
self._write_config(config)
with patch.dict(os.environ, self._env("reviewer-with-merge")):
res = mcp_server.gitea_resolve_task_capability(
task="merge_pr", remote="prgs"
)
self.assertEqual(res["required_role_kind"], "merger")
self.assertEqual(res["active_role_kind"], "reviewer")
self.assertTrue(res["active_profile_permission_allowed"])
self.assertFalse(res["allowed_in_current_session"])
self.assertTrue(res["stop_required"])
self.assertIn("cannot perform merger task", " ".join(res["task_role_guidance"]))
@patch("mcp_server.api_request", return_value={"login": "merger-user"})
@patch("mcp_server.get_auth_header", return_value="token merger-pass")
def test_merger_profile_can_resolve_merge_task(self, _auth, _api):
with patch.dict(os.environ, self._env("merger-profile")):
res = mcp_server.gitea_resolve_task_capability(
task="merge_pr", remote="prgs"
)
self.assertEqual(res["required_role_kind"], "merger")
self.assertEqual(res["active_role_kind"], "merger")
self.assertTrue(res["allowed_in_current_session"])
self.assertFalse(res["stop_required"])
@patch("mcp_server.api_request")
def test_self_review_blocked_structured(self, mock_api):
# Test that self-approve (self-review gate) is blocked and returns structured reasons
tests/test_role_session_router.py
+41 -1
View File
@@ -52,6 +52,20 @@ CONFIG = {
],
"execution_profile": "prgs-reviewer",
},
"prgs-merger": {
"enabled": True,
"context": "ctx",
"role": "merger",
"username": "sysadmin",
"auth": {"type": "env", "name": "GITEA_TOKEN_MERGER"},
"allowed_operations": [
"gitea.read", "gitea.pr.merge", "gitea.issue.comment",
],
"forbidden_operations": [
"gitea.pr.create", "gitea.branch.push", "gitea.pr.approve",
],
"execution_profile": "prgs-merger",
},
"prgs-reconciler": {
"enabled": True,
"context": "ctx",
@@ -104,6 +118,7 @@ class TestRoleSessionRouter(unittest.TestCase):
"GITEA_MCP_PROFILE": profile,
"GITEA_TOKEN_AUTHOR": "author-pass",
"GITEA_TOKEN_REVIEWER": "reviewer-pass",
"GITEA_TOKEN_MERGER": "merger-pass",
"GITEA_TOKEN_RECONCILER": "reconciler-pass",
}
@@ -227,6 +242,31 @@ class TestRoleSessionRouter(unittest.TestCase):
self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED)
self.assertTrue(route["downstream_allowed"])
@patch("mcp_server.api_request", return_value={"login": "sysadmin"})
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
def test_merge_task_under_reviewer_profile_wrong_role_stop(self, _auth, _api):
# #539: a reviewer session must not pass the pre-task merge route
# even if its config accidentally includes gitea.pr.merge.
with patch.dict(os.environ, self._env("prgs-reviewer")):
route = mcp_server.gitea_route_task_session(
task_type="merge_pr", remote="prgs"
)
self.assertEqual(route["required_role"], "merger")
self.assertEqual(route["route_result"], role_session_router.ROUTE_WRONG_ROLE)
self.assertFalse(route["downstream_allowed"])
self.assertIn("merger", route["message"].lower())
@patch("mcp_server.api_request", return_value={"login": "sysadmin"})
@patch("mcp_server.get_auth_header", return_value="token merger-pass")
def test_merge_task_under_merger_profile_allowed(self, _auth, _api):
with patch.dict(os.environ, self._env("prgs-merger")):
route = mcp_server.gitea_route_task_session(
task_type="merge_pr", remote="prgs"
)
self.assertEqual(route["required_role"], "merger")
self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED)
self.assertTrue(route["downstream_allowed"])
@patch("mcp_server.api_request", return_value={"login": "sysadmin"})
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
def test_author_task_under_reviewer_profile_routes_to_author(self, _auth, _api):
@@ -334,4 +374,4 @@ class TestCheckMidMerge(unittest.TestCase):
if __name__ == "__main__":
unittest.main()
unittest.main()