329 lines
14 KiB
Python
329 lines
14 KiB
Python
"""Tests for structured permission failure reports on gated tools (#142).
|
|
|
|
A blocked gated call must explain itself: which operation was requested,
|
|
what permission is missing, which profile/identity is active, whether any
|
|
configured profile could perform it, whether runtime switching is
|
|
supported, whether a different MCP namespace/session is required, and the
|
|
exact safe next action. Fail-closed behavior is unchanged — the report
|
|
adds information to denials, never capability. All synthetic; no network.
|
|
"""
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
|
|
|
import gitea_config
|
|
import mcp_server
|
|
|
|
CONFIG = {
|
|
"version": 2,
|
|
"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",
|
|
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
|
|
"allowed_operations": [
|
|
"gitea.read", "gitea.pr.create", "gitea.branch.push"],
|
|
"forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
|
|
"execution_profile": "author-profile"
|
|
},
|
|
"reviewer-profile": {
|
|
"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", "gitea.issue.comment"],
|
|
"forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
|
|
"execution_profile": "reviewer-profile"
|
|
}
|
|
},
|
|
"rules": {
|
|
"allow_runtime_switching": False
|
|
}
|
|
}
|
|
|
|
CONFIG_SWITCHING = {**CONFIG, "rules": {"allow_runtime_switching": True}}
|
|
|
|
PR_PAYLOAD = {
|
|
"user": {"login": "someone-else"},
|
|
"state": "open",
|
|
"head": {"sha": "abc123"},
|
|
"mergeable": True,
|
|
}
|
|
|
|
REPORT_FIELDS = (
|
|
"requested_operation",
|
|
"missing_permission",
|
|
"required_permission",
|
|
"active_profile",
|
|
"active_identity",
|
|
"active_allowed_operations",
|
|
"matching_configured_profiles",
|
|
"runtime_switching_supported",
|
|
"different_mcp_namespace_required",
|
|
"exact_safe_next_action",
|
|
)
|
|
|
|
|
|
class PermissionReportBase(unittest.TestCase):
|
|
def setUp(self):
|
|
self._remotes = patch.dict(mcp_server.REMOTES, {
|
|
"prgs": {"host": "gitea.example.com", "org": "Example-Org",
|
|
"repo": "Example-Repo"},
|
|
})
|
|
self._remotes.start()
|
|
mcp_server._IDENTITY_CACHE.clear()
|
|
gitea_config._active_profile_override = None
|
|
self._dir = tempfile.TemporaryDirectory()
|
|
self.config_path = os.path.join(self._dir.name, "profiles.json")
|
|
self._write_config(CONFIG)
|
|
|
|
def tearDown(self):
|
|
self._remotes.stop()
|
|
mcp_server._IDENTITY_CACHE.clear()
|
|
gitea_config._active_profile_override = None
|
|
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"):
|
|
return {
|
|
"GITEA_MCP_CONFIG": self.config_path,
|
|
"GITEA_MCP_PROFILE": profile,
|
|
"GITEA_TOKEN_AUTHOR": "author-pass",
|
|
"GITEA_TOKEN_REVIEWER": "reviewer-pass",
|
|
}
|
|
|
|
def _assert_report_shape(self, report):
|
|
for field in REPORT_FIELDS:
|
|
self.assertIn(field, report, f"report missing {field!r}")
|
|
|
|
def _assert_report_safe(self, report):
|
|
blob = json.dumps(report)
|
|
for banned in ("author-pass", "reviewer-pass", "https://", "http://",
|
|
"keychain", "Authorization"):
|
|
self.assertNotIn(banned, blob, f"report leaks {banned!r}")
|
|
|
|
|
|
class TestIssueCommentDenialReport(PermissionReportBase):
|
|
def test_missing_issue_comment_permission_gets_report(self):
|
|
with patch.dict(os.environ, self._env("author-profile")):
|
|
res = mcp_server.gitea_create_issue_comment(
|
|
issue_number=7, body="hello", remote="prgs")
|
|
self.assertFalse(res["success"])
|
|
self.assertFalse(res["performed"])
|
|
self.assertTrue(res["reasons"]) # fail-closed reasons preserved
|
|
report = res["permission_report"]
|
|
self._assert_report_shape(report)
|
|
self._assert_report_safe(report)
|
|
self.assertEqual(report["missing_permission"], "gitea.issue.comment")
|
|
self.assertEqual(report["active_profile"], "author-profile")
|
|
self.assertIn("gitea.read", report["active_allowed_operations"])
|
|
self.assertEqual(
|
|
report["matching_configured_profiles"], ["reviewer-profile"])
|
|
self.assertFalse(report["runtime_switching_supported"])
|
|
self.assertTrue(report["different_mcp_namespace_required"])
|
|
self.assertTrue(report["exact_safe_next_action"])
|
|
|
|
def test_empty_body_alone_is_not_a_permission_failure(self):
|
|
# reviewer-profile HAS gitea.issue.comment; an empty body must not
|
|
# produce a permission report (it is an input error, not a denial).
|
|
with patch.dict(os.environ, self._env("reviewer-profile")):
|
|
res = mcp_server.gitea_create_issue_comment(
|
|
issue_number=7, body=" ", remote="prgs")
|
|
self.assertFalse(res["success"])
|
|
self.assertNotIn("permission_report", res)
|
|
|
|
def test_switching_enabled_changes_guidance(self):
|
|
self._write_config(CONFIG_SWITCHING)
|
|
with patch.dict(os.environ, self._env("author-profile")):
|
|
res = mcp_server.gitea_create_issue_comment(
|
|
issue_number=7, body="hello", remote="prgs")
|
|
report = res["permission_report"]
|
|
self.assertTrue(report["runtime_switching_supported"])
|
|
self.assertFalse(report["different_mcp_namespace_required"])
|
|
self.assertIn("gitea_activate_profile",
|
|
report["exact_safe_next_action"])
|
|
|
|
|
|
class TestEligibilityDenialReport(PermissionReportBase):
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
|
def test_author_attempting_review_gets_report(self, _auth, mock_api):
|
|
def fake_api(method, url, auth, payload=None):
|
|
if url.endswith("/user"):
|
|
return {"login": "author-user"}
|
|
return PR_PAYLOAD
|
|
mock_api.side_effect = fake_api
|
|
with patch.dict(os.environ, self._env("author-profile")):
|
|
res = mcp_server.gitea_check_pr_eligibility(
|
|
pr_number=42, action="approve", remote="prgs")
|
|
self.assertFalse(res["eligible"])
|
|
report = res["permission_report"]
|
|
self._assert_report_shape(report)
|
|
self._assert_report_safe(report)
|
|
self.assertEqual(report["missing_permission"], "gitea.pr.approve")
|
|
self.assertEqual(report["active_profile"], "author-profile")
|
|
# "active identity if resolved": the #164 fail-fast path performs no
|
|
# identity lookup on a pure permission block, so None is acceptable.
|
|
self.assertIn(report["active_identity"], (None, "author-user"))
|
|
self.assertEqual(
|
|
report["matching_configured_profiles"], ["reviewer-profile"])
|
|
self.assertTrue(report["different_mcp_namespace_required"])
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
|
def test_author_attempting_merge_gets_report(self, _auth, mock_api):
|
|
def fake_api(method, url, auth, payload=None):
|
|
if url.endswith("/user"):
|
|
return {"login": "author-user"}
|
|
return PR_PAYLOAD
|
|
mock_api.side_effect = fake_api
|
|
with patch.dict(os.environ, self._env("author-profile")):
|
|
res = mcp_server.gitea_check_pr_eligibility(
|
|
pr_number=42, action="merge", remote="prgs")
|
|
self.assertFalse(res["eligible"])
|
|
report = res["permission_report"]
|
|
self.assertEqual(report["missing_permission"], "gitea.pr.merge")
|
|
self._assert_report_safe(report)
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
|
def test_eligible_call_has_no_report(self, _auth, mock_api):
|
|
def fake_api(method, url, auth, payload=None):
|
|
if url.endswith("/user"):
|
|
return {"login": "reviewer-user"}
|
|
return PR_PAYLOAD
|
|
mock_api.side_effect = fake_api
|
|
with patch.dict(os.environ, self._env("reviewer-profile")):
|
|
res = mcp_server.gitea_check_pr_eligibility(
|
|
pr_number=42, action="approve", remote="prgs")
|
|
self.assertTrue(res["eligible"])
|
|
self.assertNotIn("permission_report", res)
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
|
def test_review_tool_denial_propagates_report(self, _auth, mock_api):
|
|
def fake_api(method, url, auth, payload=None):
|
|
if url.endswith("/user"):
|
|
return {"login": "author-user"}
|
|
return PR_PAYLOAD
|
|
mock_api.side_effect = fake_api
|
|
with patch.dict(os.environ, self._env("author-profile")):
|
|
res = mcp_server.gitea_submit_pr_review(
|
|
pr_number=42, action="approve", body="lgtm", remote="prgs")
|
|
self.assertFalse(res["performed"])
|
|
self.assertIn("permission_report", res)
|
|
self.assertEqual(res["permission_report"]["missing_permission"],
|
|
"gitea.pr.approve")
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
|
def test_merge_tool_denial_propagates_report(self, _auth, mock_api):
|
|
def fake_api(method, url, auth, payload=None):
|
|
if url.endswith("/user"):
|
|
return {"login": "author-user"}
|
|
return PR_PAYLOAD
|
|
mock_api.side_effect = fake_api
|
|
with patch.dict(os.environ, self._env("author-profile")):
|
|
res = mcp_server.gitea_merge_pr(
|
|
pr_number=42, confirmation="MERGE PR 42",
|
|
expected_head_sha="abc123", remote="prgs")
|
|
self.assertFalse(res["performed"])
|
|
self.assertIn("permission_report", res)
|
|
self.assertEqual(res["permission_report"]["missing_permission"],
|
|
"gitea.pr.merge")
|
|
|
|
|
|
class TestReviewCommentPathUsesCanonicalOp(PermissionReportBase):
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
|
def test_review_comment_denial_names_pr_review_op(self, _auth, mock_api):
|
|
# gitea_submit_pr_review(action="comment") maps to the 'review'
|
|
# eligibility action; the report must name the canonical
|
|
# gitea.pr.review op (never a phantom 'gitea.review'), classify the
|
|
# role as reviewer, and find the configured reviewer profile.
|
|
def fake_api(method, url, auth, payload=None):
|
|
if url.endswith("/user"):
|
|
return {"login": "author-user"}
|
|
return PR_PAYLOAD
|
|
mock_api.side_effect = fake_api
|
|
with patch.dict(os.environ, self._env("author-profile")):
|
|
res = mcp_server.gitea_submit_pr_review(
|
|
pr_number=42, action="comment", body="finding", remote="prgs")
|
|
self.assertFalse(res["performed"])
|
|
report = res["permission_report"]
|
|
self._assert_report_safe(report)
|
|
self.assertEqual(report["missing_permission"], "gitea.pr.review")
|
|
self.assertEqual(
|
|
report["matching_configured_profiles"], ["reviewer-profile"])
|
|
self.assertIn("reviewer", report["exact_safe_next_action"].lower())
|
|
|
|
|
|
class TestMatchingScanFailsClosed(PermissionReportBase):
|
|
def test_profile_with_bad_forbidden_entry_is_not_reported_matching(self):
|
|
# A profile whose forbidden list cannot be normalized would be
|
|
# refused by the real gate (invalid-forbidden-entry, fail closed);
|
|
# the report must not advertise it as a match.
|
|
config = json.loads(json.dumps(CONFIG))
|
|
config["profiles"]["reviewer-profile"]["forbidden_operations"] = [
|
|
"definitely.not.a.real.op.entry"]
|
|
self._write_config(config)
|
|
with patch.dict(os.environ, self._env("author-profile")):
|
|
report = mcp_server._permission_block_report("gitea.pr.approve")
|
|
self.assertEqual(report["matching_configured_profiles"], [])
|
|
|
|
|
|
class TestReviewerAttemptingAuthoring(PermissionReportBase):
|
|
def test_reviewer_blocked_from_authoring_op_names_author_namespace(self):
|
|
with patch.dict(os.environ, self._env("reviewer-profile")):
|
|
report = mcp_server._permission_block_report("gitea.branch.push")
|
|
self._assert_report_shape(report)
|
|
self._assert_report_safe(report)
|
|
self.assertEqual(report["missing_permission"], "gitea.branch.push")
|
|
self.assertEqual(report["active_profile"], "reviewer-profile")
|
|
self.assertEqual(
|
|
report["matching_configured_profiles"], ["author-profile"])
|
|
self.assertTrue(report["different_mcp_namespace_required"])
|
|
self.assertIn("author", report["exact_safe_next_action"].lower())
|
|
|
|
|
|
class TestUnknownProfileFailsClosed(PermissionReportBase):
|
|
def test_unknown_profile_denial_still_fails_closed_with_safe_report(self):
|
|
with patch.dict(os.environ, self._env("no-such-profile")):
|
|
res = mcp_server.gitea_create_issue_comment(
|
|
issue_number=7, body="hello", remote="prgs")
|
|
self.assertFalse(res["success"])
|
|
self.assertFalse(res["performed"])
|
|
self.assertTrue(res["reasons"])
|
|
report = res["permission_report"]
|
|
self._assert_report_shape(report)
|
|
self._assert_report_safe(report)
|
|
self.assertIsNone(report["active_profile"])
|
|
self.assertTrue(report["different_mcp_namespace_required"])
|
|
self.assertIn("operator", report["exact_safe_next_action"].lower())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|