"""MCP tool-boundary error mapping for known Gitea client failures (#699). Known authentication / authorization / network / configuration failures leave the tool boundary as a sanitized structured ``CallToolResult`` with ``isError=True``. Stdio transport remains connected. Design constraints (reviewer-ratified, #699 / PR #701): 1. **No secret material** in tool results or daemon logs. Messages are fixed constants keyed by ``reason_code``; HTTP bodies / exception text never surface. Sanitization failure fails closed to ``internal_error``. 2. **Narrow boundary** wraps the original FastMCP ``Tool.run`` success path; re-raises framework control-flow exceptions (``UrlElicitationRequiredError``); maps only typed client failures. Installation is idempotent. 3. **No RuntimeError substring heuristics.** Only explicit typed authentication / authorization / network / configuration exceptions receive those labels. 4. **HTTP classification** lives in ``gitea_auth.classify_http_status``; every 403 is authorization-class. """ from __future__ import annotations import json import logging import sys from typing import Any logger = logging.getLogger("gitea_mcp.tool_error_boundary") # Stable reason codes (#699). REASON_AUTH_FAILED = "auth_failed" REASON_AUTH_INVALID_TOKEN = "auth_invalid_token" REASON_AUTHZ_INSUFFICIENT_SCOPE = "authz_insufficient_scope" REASON_AUTHZ_DENIED = "authz_denied" REASON_NETWORK_ERROR = "network_error" REASON_CONFIG_ERROR = "config_error" REASON_INTERNAL_ERROR = "internal_error" REASON_UPSTREAM_UNAVAILABLE = "upstream_unavailable" REASON_HTTP_ERROR = "http_error" ERROR_CLASS_AUTHENTICATION = "authentication" ERROR_CLASS_AUTHORIZATION = "authorization" ERROR_CLASS_NETWORK = "network" ERROR_CLASS_CONFIGURATION = "configuration" ERROR_CLASS_INTERNAL = "internal" ERROR_CLASS_UPSTREAM = "upstream" # Fixed, secret-free operator messages. Never interpolate HTTP bodies, # Keychain contents, tokens, or arbitrary exception text. FIXED_MESSAGES: dict[str, str] = { REASON_AUTH_FAILED: "Gitea authentication failed", REASON_AUTH_INVALID_TOKEN: ( "Gitea authentication failed: invalid or revoked credentials" ), REASON_AUTHZ_INSUFFICIENT_SCOPE: ( "Gitea authorization failed: insufficient token scope" ), REASON_AUTHZ_DENIED: "Gitea authorization failed: access denied", REASON_NETWORK_ERROR: "Network error contacting Gitea", REASON_CONFIG_ERROR: "Gitea configuration or credential resolution failed", REASON_INTERNAL_ERROR: "Internal tool error", REASON_UPSTREAM_UNAVAILABLE: "Gitea upstream unavailable", REASON_HTTP_ERROR: "Gitea HTTP request failed", } _INSTALL_FLAG = "_gitea_auth_boundary_installed" _ORIGINAL_ATTR = "_gitea_auth_boundary_original" def fixed_message(reason_code: str) -> str: """Return the fixed sanitized message for *reason_code* (fail closed).""" return FIXED_MESSAGES.get(reason_code, FIXED_MESSAGES[REASON_INTERNAL_ERROR]) def _safe_profile_name() -> str | None: try: from gitea_auth import get_profile name = (get_profile() or {}).get("profile_name") if name is None: return None text = str(name).strip() # Profile names are non-secret identifiers; still bound length. return text[:80] if text else None except Exception: return None def _is_framework_control_flow(exc: BaseException) -> bool: """True for exceptions the MCP framework must re-raise unchanged.""" try: from mcp.shared.exceptions import UrlElicitationRequiredError if isinstance(exc, UrlElicitationRequiredError): return True except Exception: pass # BaseException subclasses that must never become tool isError payloads. if isinstance(exc, (KeyboardInterrupt, SystemExit, GeneratorExit)): return True return False def is_known_client_failure(exc: BaseException) -> bool: """True only for explicit typed Gitea client / config failures. Never true for arbitrary ``RuntimeError`` (reviewer finding #3). """ try: import gitea_auth if isinstance( exc, ( gitea_auth.GiteaAuthError, gitea_auth.GiteaAuthzError, gitea_auth.GiteaNetworkError, gitea_auth.GiteaConfigError, gitea_auth.GiteaHttpError, ), ): return True except Exception: return False try: import gitea_config if isinstance(exc, gitea_config.ConfigError): return True except Exception: pass return False def classify_exception(exc: BaseException) -> dict[str, Any]: """Return structured classification with **fixed** messages only. Only typed authentication / authorization / network / configuration failures receive those labels. Unexpected programming failures are ``internal_error``. HTTP bodies and exception text are never copied into ``message``. """ try: return _classify_exception_impl(exc) except Exception: # Fail closed: sanitization / classification failure must not leak. return { "reason_code": REASON_INTERNAL_ERROR, "error_class": ERROR_CLASS_INTERNAL, "http_status": None, "message": fixed_message(REASON_INTERNAL_ERROR), "transport_survives": True, } def _classify_exception_impl(exc: BaseException) -> dict[str, Any]: import gitea_auth if isinstance(exc, gitea_auth.GiteaAuthError): code = getattr(exc, "reason_code", None) or REASON_AUTH_INVALID_TOKEN if code not in ( REASON_AUTH_FAILED, REASON_AUTH_INVALID_TOKEN, ): code = REASON_AUTH_INVALID_TOKEN return { "reason_code": code, "error_class": ERROR_CLASS_AUTHENTICATION, "http_status": getattr(exc, "http_status", None) or 401, "message": fixed_message(code), "transport_survives": True, } if isinstance(exc, gitea_auth.GiteaAuthzError): code = getattr(exc, "reason_code", None) or REASON_AUTHZ_DENIED if code not in (REASON_AUTHZ_DENIED, REASON_AUTHZ_INSUFFICIENT_SCOPE): code = REASON_AUTHZ_DENIED return { "reason_code": code, "error_class": ERROR_CLASS_AUTHORIZATION, "http_status": getattr(exc, "http_status", None) or 403, "message": fixed_message(code), "transport_survives": True, } if isinstance(exc, gitea_auth.GiteaNetworkError): return { "reason_code": REASON_NETWORK_ERROR, "error_class": ERROR_CLASS_NETWORK, "http_status": getattr(exc, "http_status", None), "message": fixed_message(REASON_NETWORK_ERROR), "transport_survives": True, } if isinstance(exc, gitea_auth.GiteaConfigError): return { "reason_code": REASON_CONFIG_ERROR, "error_class": ERROR_CLASS_CONFIGURATION, "http_status": getattr(exc, "http_status", None), "message": fixed_message(REASON_CONFIG_ERROR), "transport_survives": True, } if isinstance(exc, gitea_auth.GiteaHttpError): code = getattr(exc, "reason_code", None) or REASON_HTTP_ERROR if code == REASON_UPSTREAM_UNAVAILABLE: error_class = ERROR_CLASS_UPSTREAM else: error_class = ERROR_CLASS_INTERNAL code = REASON_HTTP_ERROR return { "reason_code": code, "error_class": error_class, "http_status": getattr(exc, "http_status", None), "message": fixed_message(code), "transport_survives": True, } try: import gitea_config if isinstance(exc, gitea_config.ConfigError): return { "reason_code": REASON_CONFIG_ERROR, "error_class": ERROR_CLASS_CONFIGURATION, "http_status": None, "message": fixed_message(REASON_CONFIG_ERROR), "transport_survives": True, } except Exception: pass # Unwrap FastMCP ToolError cause when the original was a typed failure. cause = getattr(exc, "__cause__", None) if cause is not None and cause is not exc and is_known_client_failure(cause): return _classify_exception_impl(cause) # No message-substring authentication heuristics (reviewer finding #3). return { "reason_code": REASON_INTERNAL_ERROR, "error_class": ERROR_CLASS_INTERNAL, "http_status": None, "message": fixed_message(REASON_INTERNAL_ERROR), "transport_survives": True, } def build_structured_error_payload( classification: dict[str, Any], *, tool_name: str | None = None, profile_name: str | None = None, ) -> dict[str, Any]: """LLM-safe structured payload — fixed message + typed metadata only.""" try: reason = str(classification.get("reason_code") or REASON_INTERNAL_ERROR) message = fixed_message(reason) # Refuse to emit any classification message that is not the fixed constant. if classification.get("message") != message: message = fixed_message(reason) payload: dict[str, Any] = { "success": False, "isError": True, "reason_code": reason if reason in FIXED_MESSAGES else REASON_INTERNAL_ERROR, "error_class": classification.get("error_class") or ERROR_CLASS_INTERNAL, "message": message, "transport_survives": True, "retryable": classification.get("error_class") in { ERROR_CLASS_AUTHENTICATION, ERROR_CLASS_NETWORK, ERROR_CLASS_CONFIGURATION, }, } status = classification.get("http_status") if isinstance(status, int): payload["http_status"] = status if tool_name and isinstance(tool_name, str): # Tool names are identifiers, not secrets; bound length. payload["tool"] = tool_name[:120] if profile_name and isinstance(profile_name, str): payload["profile"] = profile_name[:80] return payload except Exception: return { "success": False, "isError": True, "reason_code": REASON_INTERNAL_ERROR, "error_class": ERROR_CLASS_INTERNAL, "message": fixed_message(REASON_INTERNAL_ERROR), "transport_survives": True, "retryable": False, } def log_sanitized_daemon_reason( classification: dict[str, Any], *, tool_name: str | None = None, stream=None, ) -> None: """Daemon log: reason codes only — never exception text or response bodies.""" stream = stream if stream is not None else sys.stderr try: reason = classification.get("reason_code") or REASON_INTERNAL_ERROR error_class = classification.get("error_class") or ERROR_CLASS_INTERNAL # Only emit known tokens; never classification['message'] from callers # that might have been poisoned. if reason not in FIXED_MESSAGES: reason = REASON_INTERNAL_ERROR error_class = ERROR_CLASS_INTERNAL parts = [ "mcp_tool_error", f"reason_code={reason}", f"error_class={error_class}", ] if tool_name and isinstance(tool_name, str): parts.append(f"tool={tool_name[:120]}") status = classification.get("http_status") if isinstance(status, int): parts.append(f"http_status={status}") # Intentionally no detail= / message= field — secrets lived there. line = " ".join(parts) stream.write(line + "\n") if hasattr(stream, "flush"): stream.flush() logger.warning(line) except Exception: # Fail closed: never fall back to logging the exception. try: stream.write( "mcp_tool_error reason_code=internal_error " "error_class=internal\n" ) if hasattr(stream, "flush"): stream.flush() except Exception: pass def to_call_tool_result( exc: BaseException, *, tool_name: str | None = None, profile_name: str | None = None, log: bool = True, ) -> Any: """Build a FastMCP ``CallToolResult`` with ``isError=True`` for *exc*.""" from mcp.types import CallToolResult, TextContent try: classification = classify_exception(exc) if log: log_sanitized_daemon_reason(classification, tool_name=tool_name) payload = build_structured_error_payload( classification, tool_name=tool_name, profile_name=profile_name ) text = json.dumps(payload, indent=2, sort_keys=True) return CallToolResult( content=[TextContent(type="text", text=text)], structuredContent=payload, isError=True, ) except Exception: # Absolute fail-closed path — no exception text. fallback = { "success": False, "isError": True, "reason_code": REASON_INTERNAL_ERROR, "error_class": ERROR_CLASS_INTERNAL, "message": fixed_message(REASON_INTERNAL_ERROR), "transport_survives": True, "retryable": False, } return CallToolResult( content=[ TextContent( type="text", text=json.dumps(fallback, indent=2, sort_keys=True), ) ], structuredContent=fallback, isError=True, ) def install_tool_run_boundary(Tool) -> bool: """Install a **narrow** error boundary around FastMCP ``Tool.run``. Strategy (preserves framework semantics): * Call the **original** ``Tool.run`` for the success path (async, return types, convert_result, protocol behavior unchanged). * Re-raise ``UrlElicitationRequiredError`` and other control-flow exceptions without mapping to ``internal_error``. * Map typed Gitea client failures (and ToolError whose ``__cause__`` is typed) to structured ``CallToolResult(isError=True)``. * Map remaining unexpected tool failures to fixed ``internal_error`` isError results (transport survival) without secret-bearing text. * Idempotent: second install is a no-op and returns ``False``. """ if getattr(Tool.run, _INSTALL_FLAG, False): return False from mcp.server.fastmcp.exceptions import ToolError from mcp.shared.exceptions import UrlElicitationRequiredError original_run = Tool.run async def run_boundary( self, arguments: dict[str, Any], context=None, convert_result: bool = False, ) -> Any: try: return await original_run( self, arguments, context=context, convert_result=convert_result, ) except UrlElicitationRequiredError: # Framework control-flow — must not become internal_error. raise except BaseException as exc: if _is_framework_control_flow(exc): raise if not isinstance(exc, Exception): raise # Prefer typed cause under FastMCP ToolError wrappers. target: BaseException = exc if isinstance(exc, ToolError) and exc.__cause__ is not None: target = exc.__cause__ # Known client failures → structured isError with reason codes. # Unexpected failures → fixed internal_error isError (no secrets). profile_name = _safe_profile_name() return to_call_tool_result( target, tool_name=getattr(self, "name", None), profile_name=profile_name, ) setattr(run_boundary, _INSTALL_FLAG, True) setattr(run_boundary, _ORIGINAL_ATTR, original_run) Tool.run = run_boundary # type: ignore[method-assign] return True def boundary_is_installed(Tool) -> bool: """True when the #699 boundary is active on *Tool.run*.""" return bool(getattr(Tool.run, _INSTALL_FLAG, False))