diff --git a/mcp_server.py b/mcp_server.py index 6c8ede6..1795407 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -503,7 +503,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() @@ -624,9 +626,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 @@ -831,6 +840,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). @@ -1151,6 +1162,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. @@ -1306,7 +1319,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() @@ -1455,6 +1471,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). @@ -1510,12 +1608,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" @@ -1569,14 +1669,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" diff --git a/tests/test_permission_reports.py b/tests/test_permission_reports.py new file mode 100644 index 0000000..b2959ef --- /dev/null +++ b/tests/test_permission_reports.py @@ -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()