feat: structured permission failure reports on gated tools (#142) #161

Merged
sysadmin merged 1 commits from feat/issue-142-structured-permission-failures into master 2026-07-05 10:24:33 -05:00
2 changed files with 443 additions and 9 deletions
+117 -9
View File
@@ -502,7 +502,9 @@ def gitea_check_pr_eligibility(
repo: Override the repository name.
Returns:
dict with 'eligible' (bool), the inputs inspected, and 'reasons'.
dict with 'eligible' (bool), the inputs inspected, and 'reasons';
when the block is a missing profile permission, also a structured
'permission_report' (#142).
"""
action = (action or "").strip().lower()
profile = get_profile()
@@ -623,9 +625,16 @@ def gitea_check_pr_eligibility(
# Determine missing permission
missing_perm = None
if not op_ok:
missing_perm = f"gitea.pr.{action}" if action in ("approve", "request_changes", "merge") else f"gitea.{action}"
missing_perm = (
f"gitea.pr.{action}"
if action in ("approve", "request_changes", "merge", "review")
else f"gitea.{action}")
result["missing_permission"] = missing_perm
if missing_perm:
# Structured denial guidance (#142) — read-only, never widens.
result["permission_report"] = _permission_block_report(
missing_perm, identity=auth_user)
# Determine required profile
req_profile = None
@@ -830,6 +839,8 @@ def gitea_submit_pr_review(
f"eligibility check for '{eligibility_action}' failed (fail closed)"
)
reasons.extend(elig.get("reasons", []))
if elig.get("permission_report"):
result["permission_report"] = elig["permission_report"]
return result
# Gate 3 — redundant self-approval block (belt-and-suspenders over #14).
@@ -1150,6 +1161,8 @@ def gitea_merge_pr(
if not elig.get("eligible"):
reasons.append("eligibility check for 'merge' failed (fail closed)")
reasons.extend(elig.get("reasons", []))
if elig.get("permission_report"):
result["permission_report"] = elig["permission_report"]
return result
# Gate 4 — head SHA must match if the caller pinned a reviewed SHA.
@@ -1305,7 +1318,10 @@ def gitea_review_pr(
return {"success": True, "message": f"Successfully submitted review for PR #{pr_number} with event '{event}'."}
else:
reasons = result.get("reasons", [])
return {"success": False, "message": f"Review submission failed eligibility gates: {reasons}"}
out = {"success": False, "message": f"Review submission failed eligibility gates: {reasons}"}
if result.get("permission_report"):
out["permission_report"] = result["permission_report"]
return out
@mcp.tool()
@@ -1454,6 +1470,88 @@ def gitea_view_issue(
return _with_optional_url(result, i.get("html_url"))
def _permission_block_report(required_operation: str,
identity: str | None = None) -> dict:
"""Structured, LLM-safe explanation of a permission denial (#142).
Built only after a gate has already refused; it adds guidance to the
refusal and never widens any permission, performs network I/O, or
raises (fail-soft: degrades to a minimal fail-closed report). Names
configured profiles only — never auth references, tokens, endpoint
URLs, or keychain IDs.
"""
report = {
"requested_operation": required_operation,
"missing_permission": required_operation,
"required_permission": required_operation,
"active_profile": None,
"active_identity": identity,
"active_allowed_operations": [],
"matching_configured_profiles": [],
"runtime_switching_supported": False,
"different_mcp_namespace_required": True,
"exact_safe_next_action": (
"Ask the operator to fix GITEA_MCP_CONFIG/GITEA_MCP_PROFILE; "
"the active profile could not be resolved (fail closed)."),
}
try:
profile = get_profile()
except Exception:
return report
report["active_profile"] = profile.get("profile_name")
if identity is None:
report["active_identity"] = profile.get("identity")
report["active_allowed_operations"] = (
profile.get("allowed_operations") or [])
matching = []
try:
config = gitea_config.load_config() or {}
for name, p in (config.get("profiles") or {}).items():
p_allowed = []
for op in p.get("allowed_operations") or []:
try:
p_allowed.append(gitea_config.normalize_operation(op))
except Exception:
pass
p_forbidden = []
for op in p.get("forbidden_operations") or []:
try:
p_forbidden.append(gitea_config.normalize_operation(op))
except Exception:
pass
ok, _ = gitea_config.check_operation(
required_operation, p_allowed, p_forbidden)
if ok:
matching.append(name)
except Exception:
matching = []
report["matching_configured_profiles"] = matching
try:
switching = gitea_config.is_runtime_switching_enabled()
except Exception:
switching = False
report["runtime_switching_supported"] = switching
role = "reviewer" if required_operation in (
"gitea.pr.approve", "gitea.pr.merge", "gitea.pr.request_changes",
"gitea.pr.review") else "author"
if switching and matching:
report["different_mcp_namespace_required"] = False
report["exact_safe_next_action"] = (
f"Call gitea_activate_profile with a profile that allows "
f"{required_operation} (matching configured profiles: "
f"{matching}).")
else:
report["different_mcp_namespace_required"] = True
report["exact_safe_next_action"] = (
f"Switch to the {role} MCP session (e.g. gitea-{role}), or ask "
f"the operator to set GITEA_MCP_PROFILE to a profile that "
f"allows {required_operation}.")
return report
def _issue_comment_gate(op: str) -> list[str]:
"""Profile permission check for issue-comment tools (#126).
@@ -1509,12 +1607,14 @@ def gitea_list_issue_comments(
Returns:
dict with 'success', 'issue_number', and 'comments' (each with 'id',
'author', 'created_at', 'updated_at', 'body'); on a permission block,
'success' False and 'reasons' with no API call made.
'success' False, 'reasons', and a structured 'permission_report'
(#142) with no API call made.
"""
reasons = _issue_comment_gate("gitea.read")
if reasons:
return {"success": False, "issue_number": issue_number,
"reasons": reasons}
"reasons": reasons,
"permission_report": _permission_block_report("gitea.read")}
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
api = f"{repo_api_url(h, o, r)}/issues/{issue_number}/comments"
@@ -1568,14 +1668,22 @@ def gitea_create_issue_comment(
Returns:
dict with 'success', 'comment_id', and 'issue_number' ('url' only
with the reveal opt-in); on a permission block or empty body,
'success'/'performed' False and 'reasons' with no API call made.
'success'/'performed' False and 'reasons' with no API call made
(permission blocks also carry a structured 'permission_report',
#142).
"""
reasons = _issue_comment_gate("gitea.issue.comment")
gate_reasons = _issue_comment_gate("gitea.issue.comment")
reasons = list(gate_reasons)
if not (body or "").strip():
reasons.append("comment body must be a non-empty string")
if reasons:
return {"success": False, "performed": False,
"issue_number": issue_number, "reasons": reasons}
blocked = {"success": False, "performed": False,
"issue_number": issue_number, "reasons": reasons}
if gate_reasons:
# Permission denial (not a mere input error): explain it (#142).
blocked["permission_report"] = _permission_block_report(
"gitea.issue.comment")
return blocked
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
api = f"{repo_api_url(h, o, r)}/issues/{issue_number}/comments"
+326
View File
@@ -0,0 +1,326 @@
"""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")
self.assertEqual(report["active_identity"], "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()