Files
Gitea-Tools/tests/test_operation_scoped_roles.py
T
sysadminandGrok 632c568865 fix(session): pin immutable MCP context; ban cross-host profile substitution (Closes #714)
Capability resolution and mutation gates no longer auto-switch profiles.
Session profile/remote/host/identity are bound after pin or first seed,
and drift or cross-host resolution fails closed before writes.

Co-Authored-By: Grok <[email protected]>
2026-07-14 23:05:41 -04:00

203 lines
8.6 KiB
Python

"""Tests for operation-scoped role selection (#228, updated by #714).
#714 removes silent automatic profile substitution. Capability resolution
and mutation gates evaluate only the active profile; explicit
``gitea_activate_profile`` is required to switch roles.
"""
import os
import sys
import json
import tempfile
import unittest
from unittest.mock import patch, MagicMock
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import gitea_config
import gitea_auth
import mcp_server
import session_context_binding as session_ctx
from reviewer_worktree import assess_author_worktree_continuity
CONFIG_TEST = {
"version": 2,
"allow_runtime_switching": True,
"contexts": {
"ctx": {
"enabled": True,
"gitea": {
"enabled": True,
"base_url": "https://gitea.example.com"
}
}
},
"profiles": {
"author-profile": {
"enabled": True,
"context": "ctx",
"role": "author",
"username": "author-user",
"base_url": "https://gitea.example.com",
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
"allowed_operations": ["gitea.read", "gitea.issue.create", "gitea.pr.create", "gitea.branch.push", "gitea.issue.comment"],
"forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
"execution_profile": "author-profile"
},
"reviewer-profile": {
"enabled": True,
"context": "ctx",
"role": "reviewer",
"username": "reviewer-user",
"base_url": "https://gitea.example.com",
"auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
"allowed_operations": ["gitea.read", "gitea.pr.review", "gitea.pr.approve", "gitea.pr.merge", "gitea.issue.comment"],
"forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
"execution_profile": "reviewer-profile"
}
}
}
class TestOperationScopedRoles(unittest.TestCase):
def setUp(self):
self._remotes_patch = patch.dict(mcp_server.REMOTES, {
"dadeschools": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"},
"prgs": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"}
})
self._remotes_patch.start()
mcp_server._IDENTITY_CACHE.clear()
gitea_config._active_profile_override = None
mcp_server._MUTATION_AUTHORITY = None
session_ctx.clear_session_context()
self._dir = tempfile.TemporaryDirectory()
self.config_path = os.path.join(self._dir.name, "profiles.json")
self._write_config(CONFIG_TEST)
def tearDown(self):
self._remotes_patch.stop()
mcp_server._IDENTITY_CACHE.clear()
gitea_config._active_profile_override = None
mcp_server._MUTATION_AUTHORITY = None
session_ctx.clear_session_context()
self._dir.cleanup()
def _write_config(self, obj):
with open(self.config_path, "w", encoding="utf-8") as fh:
fh.write(json.dumps(obj))
def _env(self, profile="author-profile", with_reviewer_token=True):
env = {
"GITEA_MCP_CONFIG": self.config_path,
"GITEA_MCP_PROFILE": profile,
"GITEA_TOKEN_AUTHOR": "author-pass",
}
if with_reviewer_token:
env["GITEA_TOKEN_REVIEWER"] = "reviewer-pass"
return env
@patch("mcp_server.api_request")
def test_no_auto_switch_to_reviewer(self, mock_api):
# #714: capability resolution must not silently switch author → reviewer
mock_api.side_effect = lambda method, url, header: (
{"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
)
with patch.dict(os.environ, self._env("author-profile")):
self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
self.assertEqual(mcp_server.get_profile()["profile_name"], "author-profile")
res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
self.assertFalse(res["allowed_in_current_session"])
self.assertFalse(res["available_in_session"])
self.assertEqual(res["active_profile"], "author-profile")
self.assertEqual(res["active_identity"], "author-user")
self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
self.assertIn("reviewer-profile", res["matching_configured_profile"])
self.assertFalse(res.get("auto_profile_substitution", True))
@patch("mcp_server.api_request")
def test_no_auto_switch_to_author(self, mock_api):
mock_api.side_effect = lambda method, url, header: (
{"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
)
with patch.dict(os.environ, self._env("reviewer-profile")):
self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
res = mcp_server.gitea_resolve_task_capability(task="create_issue", remote="prgs")
self.assertFalse(res["allowed_in_current_session"])
self.assertFalse(res["available_in_session"])
self.assertEqual(res["active_profile"], "reviewer-profile")
self.assertEqual(res["active_identity"], "reviewer-user")
self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
self.assertIn("author-profile", res["matching_configured_profile"])
@patch("mcp_server.api_request")
def test_explicit_activate_profile_still_works(self, mock_api):
mock_api.side_effect = lambda method, url, header: (
{"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
)
with patch.dict(os.environ, self._env("author-profile")):
act = mcp_server.gitea_activate_profile(
profile_name="reviewer-profile", remote="prgs"
)
self.assertTrue(act["success"])
self.assertEqual(act["after_profile"], "reviewer-profile")
res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
self.assertTrue(res["allowed_in_current_session"])
self.assertEqual(res["active_profile"], "reviewer-profile")
@patch("mcp_server.api_request")
def test_denied_when_matching_profile_token_missing(self, mock_api):
mock_api.side_effect = lambda method, url, header: {"login": "author-user"}
# launch without reviewer token in env
with patch.dict(os.environ, self._env("author-profile", with_reviewer_token=False)):
self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
self.assertFalse(res["allowed_in_current_session"])
self.assertFalse(res["available_in_session"])
self.assertTrue(res["configured"])
self.assertTrue(res["stop_required"])
self.assertEqual(res["active_profile"], "author-profile")
def test_author_continuity_dirty_worktree(self):
# author is allowed to keep dirty worktree
res = assess_author_worktree_continuity({
"task_role": "author",
"dirty_files": ["review_proofs.py"],
})
self.assertTrue(res["allowed"])
# reviewer is blocked by reviewer worktree proof
res2 = assess_author_worktree_continuity({
"task_role": "reviewer",
"worktree_path": "/repo",
"dirty_files": ["review_proofs.py"],
"pr_scope_files": ["docs/wiki/Repositories.md"],
"scratch_used": False,
})
self.assertFalse(res2["proven"])
@patch("mcp_server.api_request")
def test_mutating_actions_do_not_auto_switch(self, mock_api):
mock_api.side_effect = lambda method, url, header: (
{"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
)
with patch.dict(os.environ, self._env("author-profile")):
self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
blocked = mcp_server._profile_permission_block(
"gitea.pr.merge", remote="prgs"
)
self.assertIsNotNone(blocked)
self.assertFalse(blocked.get("success", True))
# profile must remain author — no silent substitution
self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
if __name__ == "__main__":
unittest.main()