diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index dbe1c2e..9913535 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -478,6 +478,7 @@ import agent_temp_artifacts import issue_lock_worktree # noqa: E402 import issue_claim_heartbeat # noqa: E402 import merged_cleanup_reconcile # noqa: E402 +import native_mcp_preference # noqa: E402 # Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue, @@ -4083,6 +4084,14 @@ _GUIDE_RULES = { "inventory, summarize) needs no explicit authorization. Enforced " "fail closed via subagent_gate.assess_subagent_delegation and " "validate_subagent_report (#266)."), + "native_mcp_preference": ( + "Gitea mutations must use native MCP tools first. When MCP is " + "available, block shell scripts, direct API calls, WebFetch, " + "Playwright, helper encoders, and profile-secret reads unless " + "explicit recovery-mode proof is complete. If MCP transport is " + "broken or the shell circuit breaker trips, emit a terminal recovery " + "report instead of improvising unsafe fallback. Reviewers must not " + "touch or restart MCP server files/processes (#270)."), } _COMMON_WORKFLOWS = [ @@ -4787,6 +4796,69 @@ def _build_runtime_task_capabilities( } +@mcp.tool() +@mcp.tool() +def gitea_assess_gitea_operation_path( + task: str, + path_kind: str | None = None, + command_or_detail: str | None = None, + mcp_available: bool = True, + mcp_tool_visible: bool = True, + recovery_mode: bool = False, + recovery_proof_complete: bool = False, + remote: str = "dadeschools", + host: str | None = None, +) -> dict: + """Read-only: assess whether a proposed Gitea mutation path is allowed (#270). + + Native MCP must be preferred when available. Shell scripts, direct API + calls, WebFetch, Playwright, and helper encoders are blocked unless + explicit recovery-mode proof is supplied. + """ + profile = get_profile() + role = _role_kind( + profile.get("allowed_operations") or [], + profile.get("forbidden_operations") or [], + ) + return native_mcp_preference.assess_gitea_operation_path( + task=task, + path_kind=path_kind, + command_or_detail=command_or_detail, + mcp_available=mcp_available, + mcp_tool_visible=mcp_tool_visible, + recovery_mode=recovery_mode, + recovery_proof_complete=recovery_proof_complete, + role=role, + active_profile=profile.get("profile_name"), + ) + + +@mcp.tool() +def gitea_record_shell_spawn_outcome( + exit_code: int | None = None, + stdout: str = "", + stderr: str = "", + spawn_failure: bool | None = None, + probe_attempted: bool = False, + probe_succeeded: bool | None = None, +) -> dict: + """Record shell spawn outcome and update the session circuit breaker (#270).""" + return native_mcp_preference.record_shell_spawn_outcome( + exit_code=exit_code, + stdout=stdout, + stderr=stderr, + spawn_failure=spawn_failure, + probe_attempted=probe_attempted, + probe_succeeded=probe_succeeded, + ) + + +@mcp.tool() +def gitea_get_shell_health() -> dict: + """Read-only: return shell spawn circuit-breaker state for this MCP session (#270).""" + return native_mcp_preference.shell_health_status() + + @mcp.tool() def gitea_get_runtime_context( remote: str = "dadeschools", @@ -4905,6 +4977,7 @@ def gitea_get_runtime_context( "preflight_block_reasons": preflight["preflight_block_reasons"], "preflight_workspace": preflight.get("preflight_workspace"), "session_capabilities": session_capabilities, + "shell_health": native_mcp_preference.shell_health_status(), } if reveal and h: diff --git a/native_mcp_preference.py b/native_mcp_preference.py new file mode 100644 index 0000000..872b800 --- /dev/null +++ b/native_mcp_preference.py @@ -0,0 +1,369 @@ +"""Native MCP preference gate and shell health circuit breaker (#270). + +Gitea mutations must use native MCP tools first. Shell scripts, direct API +calls, browser helpers, and improvised encoders are blocked when MCP is +available unless explicit recovery-mode proof is supplied. +""" + +from __future__ import annotations + +import re +from typing import Any + +from reviewer_fallback import LOCAL_GITEA_SCRIPT_NAMES + +# Tasks that must prefer native MCP over shell/API/helper fallbacks. +GITEA_MUTATION_TASKS = frozenset({ + "comment_issue", + "mark_issue", + "lock_issue", + "set_issue_labels", + "create_issue", + "close_issue", + "create_branch", + "push_branch", + "create_pr", + "close_pr", + "comment_pr", + "review_pr", + "merge_pr", + "approve_pr", + "request_changes_pr", + "commit_files", + "gitea_commit_files", + "delete_branch", + "address_pr_change_requests", +}) + +ALLOWED_PATH_KINDS = frozenset({ + "mcp_native", + "recovery_fallback", +}) + +BLOCKED_PATH_KINDS = frozenset({ + "shell_script", + "direct_api", + "webfetch", + "playwright", + "helper_script", + "unsafe_helper", + "mcp_server_touch", +}) + +SHELL_SPAWN_FAILURE_THRESHOLD = 2 + +TERMINAL_REPORT_HEADING = ( + "MCP transport unavailable or shell circuit breaker tripped. " + "No unsafe Gitea fallback performed." +) + +_RECOVERY_MODE_RE = re.compile( + r"\b(?:recovery mode|explicit recovery|mcp unavailable|mcp not available|" + r"mcp tools unavailable|no mcp path)\b", + re.IGNORECASE, +) +_LOCAL_SCRIPT_RE = re.compile( + r"(?:^|[\s\"'`/])(?:python3?|bash|sh)?\s*(?:" + + "|".join(re.escape(name) for name in LOCAL_GITEA_SCRIPT_NAMES) + + r")", + re.IGNORECASE, +) +_WEBFETCH_RE = re.compile(r"\b(?:webfetch|mcp_web_fetch|fetch\s+url)\b", re.IGNORECASE) +_PLAYWRIGHT_RE = re.compile(r"\b(?:playwright|browser_navigate|browser_click)\b", re.IGNORECASE) +_HELPER_SCRIPT_RE = re.compile( + r"(?:^|[\s\"'`/])(?:_encode_|_emit_|_inline_)[\w-]+\.py", + re.IGNORECASE, +) +_MANUAL_BASE64_RE = re.compile( + r"\b(?:manual\s+base64|llm-generated\s+base64|base64-encode\s+in\s+chat)\b", + re.IGNORECASE, +) +_DIRECT_API_RE = re.compile( + r"\b(?:api_request|urllib\.request|requests\.(?:post|patch|put)|curl\s+-X\s+POST)\b", + re.IGNORECASE, +) +_MCP_SERVER_TOUCH_RE = re.compile( + r"\b(?:kill|pkill|restart|reload|touch|edit|modify|write)\b.{0,40}\b(?:mcp_server|mcp-server|gitea_mcp_server)\b", + re.IGNORECASE, +) +_CLI_AUTH_DIVERGENCE_RE = re.compile( + r"GITEA_MCP_PROFILE\s*=\s*['\"]?([^\s'\";]+)", + re.IGNORECASE, +) + +_session_shell: dict[str, Any] = { + "consecutive_spawn_failures": 0, + "shell_unavailable": False, + "hard_stopped": False, + "last_exit_code": None, +} + + +def _clean(value: str | None) -> str: + return (value or "").strip() + + +def is_spawn_failure( + *, + exit_code: int | None = None, + stdout: str | None = None, + stderr: str | None = None, + spawn_failure: bool | None = None, +) -> bool: + """True for executor spawn failures (exit_code -1, empty output).""" + if spawn_failure is True: + return True + if spawn_failure is False: + return False + if exit_code != -1: + return False + return not (_clean(stdout) or _clean(stderr)) + + +def record_shell_spawn_outcome( + *, + exit_code: int | None = None, + stdout: str | None = None, + stderr: str | None = None, + spawn_failure: bool | None = None, + probe_attempted: bool = False, + probe_succeeded: bool | None = None, +) -> dict[str, Any]: + """Track shell spawn outcomes and trip the session circuit breaker (#270 AC4).""" + global _session_shell + failed = is_spawn_failure( + exit_code=exit_code, + stdout=stdout, + stderr=stderr, + spawn_failure=spawn_failure, + ) + + if failed: + _session_shell["consecutive_spawn_failures"] = ( + int(_session_shell.get("consecutive_spawn_failures") or 0) + 1 + ) + _session_shell["last_exit_code"] = exit_code + if probe_attempted and probe_succeeded is False: + _session_shell["shell_unavailable"] = True + else: + _session_shell["consecutive_spawn_failures"] = 0 + _session_shell["shell_unavailable"] = False + _session_shell["hard_stopped"] = False + _session_shell["last_exit_code"] = exit_code + + failures = int(_session_shell["consecutive_spawn_failures"]) + if failures >= SHELL_SPAWN_FAILURE_THRESHOLD: + _session_shell["shell_unavailable"] = True + _session_shell["hard_stopped"] = True + + return shell_health_status() + + +def shell_health_status() -> dict[str, Any]: + """Return the current shell health / circuit-breaker state.""" + failures = int(_session_shell.get("consecutive_spawn_failures") or 0) + hard_stopped = bool(_session_shell.get("hard_stopped")) + shell_unavailable = bool(_session_shell.get("shell_unavailable")) + return { + "consecutive_spawn_failures": failures, + "shell_unavailable": shell_unavailable, + "hard_stopped": hard_stopped, + "threshold": SHELL_SPAWN_FAILURE_THRESHOLD, + "shell_use_allowed": not hard_stopped, + "last_exit_code": _session_shell.get("last_exit_code"), + "safe_next_action": ( + "emit terminal recovery report; prefer native MCP for remaining Gitea mutations" + if hard_stopped + else ( + "probe shell once with echo/pwd; if probe fails mark shell unavailable" + if failures == 1 + else "shell healthy" + ) + ), + } + + +def clear_shell_health_for_tests() -> None: + """Reset session shell state (tests only).""" + global _session_shell + _session_shell = { + "consecutive_spawn_failures": 0, + "shell_unavailable": False, + "hard_stopped": False, + "last_exit_code": None, + } + + +def classify_command_path(command_or_detail: str | None) -> str: + """Classify a proposed command/detail into a path kind.""" + text = command_or_detail or "" + if _MCP_SERVER_TOUCH_RE.search(text): + return "mcp_server_touch" + if _WEBFETCH_RE.search(text): + return "webfetch" + if _PLAYWRIGHT_RE.search(text): + return "playwright" + if _HELPER_SCRIPT_RE.search(text) or _MANUAL_BASE64_RE.search(text): + return "unsafe_helper" + if _LOCAL_SCRIPT_RE.search(text): + return "shell_script" + if _DIRECT_API_RE.search(text): + return "direct_api" + return "mcp_native" + + +def detect_cli_auth_divergence( + command_or_detail: str | None, + *, + active_profile: str | None, +) -> list[str]: + """Flag shell commands that override GITEA_MCP_PROFILE away from the session.""" + reasons: list[str] = [] + text = command_or_detail or "" + active = _clean(active_profile) + match = _CLI_AUTH_DIVERGENCE_RE.search(text) + if not match or not active: + return reasons + requested = _clean(match.group(1)) + if requested and requested != active: + reasons.append( + f"CLI auth divergence: command sets GITEA_MCP_PROFILE='{requested}' " + f"but active session profile is '{active}' (fail closed)" + ) + return reasons + + +def format_mcp_unavailable_terminal_report( + *, + task: str | None = None, + reasons: list[str] | None = None, +) -> str: + """Terminal report when MCP is broken and unsafe recovery is forbidden (#270 AC3).""" + lines = [TERMINAL_REPORT_HEADING, ""] + if task: + lines.append(f"Blocked task: {task}") + if reasons: + lines.extend(["", "Reasons:"]) + lines.extend(f"- {reason}" for reason in reasons) + lines.extend([ + "", + "Required recovery (no improvised fallback):", + "- Restart the MCP session", + "- Kill hung background terminals holding the shell executor", + "- Retry the native MCP tool once after reconnect", + "- If MCP remains unavailable, stop and hand off to the operator", + "- Do not run local Gitea scripts, WebFetch, Playwright, or manual base64 encoders", + ]) + return "\n".join(lines) + + +def assess_gitea_operation_path( + *, + task: str, + path_kind: str | None = None, + command_or_detail: str | None = None, + mcp_available: bool = True, + mcp_tool_visible: bool = True, + recovery_mode: bool = False, + recovery_proof_complete: bool = False, + role: str | None = None, + active_profile: str | None = None, +) -> dict[str, Any]: + """Fail closed when a Gitea mutation would bypass native MCP (#270).""" + task_name = _clean(task) + resolved_kind = _clean(path_kind) or classify_command_path(command_or_detail) + reasons: list[str] = [] + + if task_name and task_name not in GITEA_MUTATION_TASKS: + reasons.append( + f"unknown or non-mutation Gitea task '{task_name}'; " + "native MCP preference gate applies only to Gitea mutations" + ) + + shell = shell_health_status() + if shell["hard_stopped"] and resolved_kind != "mcp_native": + reasons.append( + "shell circuit breaker is hard-stopped after consecutive spawn failures; " + "use native MCP or emit terminal recovery report" + ) + + recovery_declared = recovery_mode or bool( + _RECOVERY_MODE_RE.search(command_or_detail or "") + ) + recovery_allowed = ( + recovery_declared + and recovery_proof_complete + and not mcp_available + and resolved_kind in BLOCKED_PATH_KINDS - {"mcp_server_touch"} + ) + + if resolved_kind in BLOCKED_PATH_KINDS and not recovery_allowed: + reasons.append(f"path kind '{resolved_kind}' is not an approved Gitea mutation path") + + if resolved_kind == "mcp_server_touch": + if _clean(role) == "reviewer" or ( + active_profile and "reviewer" in active_profile.lower() + ): + reasons.append( + "reviewer workflows must not touch, restart, or kill MCP server files/processes" + ) + else: + reasons.append( + "MCP server files/processes must not be modified during normal Gitea workflows" + ) + + reasons.extend( + detect_cli_auth_divergence(command_or_detail, active_profile=active_profile) + ) + + if ( + mcp_available + and mcp_tool_visible + and resolved_kind != "mcp_native" + and not recovery_allowed + ): + if not recovery_declared: + reasons.append( + "native MCP tools are available; shell/API/helper fallback is forbidden" + ) + elif not recovery_proof_complete: + reasons.append( + "recovery-mode fallback requires complete recovery proof before proceeding" + ) + + if not mcp_available and resolved_kind != "mcp_native" and not recovery_allowed: + if not recovery_declared: + reasons.append( + "MCP transport unavailable; produce terminal recovery report instead of " + "improvising unsafe fallback" + ) + + block = bool(reasons) + terminal_report = None + if block and (not mcp_available or shell["hard_stopped"]): + terminal_report = format_mcp_unavailable_terminal_report( + task=task_name or None, + reasons=reasons, + ) + + return { + "task": task_name or None, + "path_kind": resolved_kind, + "mcp_available": mcp_available, + "mcp_tool_visible": mcp_tool_visible, + "recovery_mode": recovery_declared, + "shell_health": shell, + "block": block, + "allowed": not block, + "reasons": reasons, + "terminal_report": terminal_report, + "safe_next_action": ( + "use the native MCP tool for this task" + if mcp_available and mcp_tool_visible and block + else ( + terminal_report or "proceed with native MCP" + if block + else "proceed with native MCP" + ) + ), + } \ No newline at end of file diff --git a/tests/test_native_mcp_preference.py b/tests/test_native_mcp_preference.py new file mode 100644 index 0000000..7607814 --- /dev/null +++ b/tests/test_native_mcp_preference.py @@ -0,0 +1,121 @@ +"""Tests for native MCP preference gate and shell circuit breaker (#270).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import native_mcp_preference as nmp # noqa: E402 + + +class TestShellHealthCircuitBreaker(unittest.TestCase): + def setUp(self): + nmp.clear_shell_health_for_tests() + + def test_two_spawn_failures_hard_stop(self): + nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="") + status = nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="") + self.assertTrue(status["hard_stopped"]) + self.assertFalse(status["shell_use_allowed"]) + self.assertEqual(status["consecutive_spawn_failures"], 2) + + def test_success_resets_breaker(self): + nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="") + status = nmp.record_shell_spawn_outcome(exit_code=0, stdout="ok", stderr="") + self.assertFalse(status["hard_stopped"]) + self.assertEqual(status["consecutive_spawn_failures"], 0) + + +class TestNativeMcpPreferenceGate(unittest.TestCase): + def setUp(self): + nmp.clear_shell_health_for_tests() + + def test_mcp_available_blocks_local_script(self): + result = nmp.assess_gitea_operation_path( + task="create_pr", + command_or_detail="python create_pr.py --remote prgs --title T --head h", + mcp_available=True, + mcp_tool_visible=True, + ) + self.assertTrue(result["block"]) + self.assertEqual(result["path_kind"], "shell_script") + + def test_mcp_native_path_allowed(self): + result = nmp.assess_gitea_operation_path( + task="comment_issue", + path_kind="mcp_native", + mcp_available=True, + mcp_tool_visible=True, + ) + self.assertFalse(result["block"]) + self.assertTrue(result["allowed"]) + + def test_mcp_unavailable_requires_terminal_report(self): + result = nmp.assess_gitea_operation_path( + task="merge_pr", + path_kind="shell_script", + mcp_available=False, + mcp_tool_visible=False, + ) + self.assertTrue(result["block"]) + self.assertTrue( + any("terminal recovery report" in r for r in result["reasons"]) + ) + self.assertIn(nmp.TERMINAL_REPORT_HEADING, result["terminal_report"]) + + def test_recovery_fallback_allowed_with_proof(self): + result = nmp.assess_gitea_operation_path( + task="merge_pr", + path_kind="shell_script", + command_or_detail="Recovery mode: MCP tools unavailable in this session.", + mcp_available=False, + recovery_mode=True, + recovery_proof_complete=True, + ) + self.assertFalse(result["block"]) + + def test_cli_auth_divergence_blocked(self): + result = nmp.assess_gitea_operation_path( + task="create_pr", + command_or_detail="GITEA_MCP_PROFILE=prgs-reviewer python create_pr.py", + mcp_available=True, + active_profile="prgs-author", + ) + self.assertTrue(result["block"]) + self.assertTrue(any("CLI auth divergence" in r for r in result["reasons"])) + + def test_unsafe_helper_fallback_denied(self): + result = nmp.assess_gitea_operation_path( + task="commit_files", + command_or_detail="python _encode_payload.py", + mcp_available=True, + mcp_tool_visible=True, + ) + self.assertTrue(result["block"]) + self.assertEqual(result["path_kind"], "unsafe_helper") + + def test_reviewer_mcp_server_touch_blocked(self): + result = nmp.assess_gitea_operation_path( + task="review_pr", + command_or_detail="pkill -f gitea_mcp_server.py", + mcp_available=True, + role="reviewer", + active_profile="prgs-reviewer", + ) + self.assertTrue(result["block"]) + self.assertEqual(result["path_kind"], "mcp_server_touch") + + def test_hard_stopped_shell_blocks_non_mcp_path(self): + nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="") + nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="") + result = nmp.assess_gitea_operation_path( + task="lock_issue", + path_kind="shell_script", + mcp_available=True, + ) + self.assertTrue(result["block"]) + self.assertTrue(any("circuit breaker" in r for r in result["reasons"])) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file