diff --git a/gitea_auth.py b/gitea_auth.py index a02d92a..fe2b769 100644 --- a/gitea_auth.py +++ b/gitea_auth.py @@ -243,6 +243,192 @@ def _redact(text): return str(text) +# ── Classified client failures (#699) ───────────────────────────────────────── +# Subclasses of RuntimeError preserve existing ``except RuntimeError`` call +# sites. Exception *messages* are fixed constants only — HTTP response bodies, +# Keychain material, and arbitrary exception text are never stored on the +# exception or re-emitted to tool results / daemon logs. + + +# Fixed messages (must match mcp_tool_error_boundary.FIXED_MESSAGES keys used here). +_MSG_AUTH_INVALID = "Gitea authentication failed: invalid or revoked credentials" +_MSG_AUTH_FAILED = "Gitea authentication failed" +_MSG_AUTHZ_SCOPE = "Gitea authorization failed: insufficient token scope" +_MSG_AUTHZ_DENIED = "Gitea authorization failed: access denied" +_MSG_NETWORK = "Network error contacting Gitea" +_MSG_CONFIG = "Gitea configuration or credential resolution failed" +_MSG_UPSTREAM = "Gitea upstream unavailable" +_MSG_HTTP = "Gitea HTTP request failed" + + +class GiteaClientError(RuntimeError): + """Base for known Gitea client failures with stable reason_code metadata.""" + + reason_code = "client_error" + error_class = "client" + http_status = None + + def __init__(self, message=None, *, reason_code=None, http_status=None): + if reason_code is not None: + self.reason_code = reason_code + if http_status is not None: + self.http_status = http_status + # Message is always a fixed constant; callers cannot inject bodies. + fixed = message if message is not None else _MSG_HTTP + super().__init__(fixed) + + +class GiteaAuthError(GiteaClientError): + """Authentication failure (invalid/revoked credentials → typically HTTP 401).""" + + reason_code = "auth_invalid_token" + error_class = "authentication" + http_status = 401 + + def __init__(self, message=None, *, reason_code=None, http_status=None): + super().__init__( + message if message is not None else _MSG_AUTH_INVALID, + reason_code=reason_code or "auth_invalid_token", + http_status=http_status if http_status is not None else 401, + ) + + +class GiteaAuthzError(GiteaClientError): + """Authorization failure (HTTP 403 — scope deficiency or access denied).""" + + reason_code = "authz_denied" + error_class = "authorization" + http_status = 403 + + def __init__(self, message=None, *, reason_code=None, http_status=None): + code = reason_code or "authz_denied" + if code == "authz_insufficient_scope": + fixed = _MSG_AUTHZ_SCOPE + else: + fixed = _MSG_AUTHZ_DENIED + code = "authz_denied" + super().__init__( + message if message is not None else fixed, + reason_code=code, + http_status=http_status if http_status is not None else 403, + ) + + +class GiteaNetworkError(GiteaClientError): + """Transport / DNS / timeout failure contacting Gitea.""" + + reason_code = "network_error" + error_class = "network" + http_status = None + + def __init__(self, message=None, *, reason_code=None, http_status=None): + super().__init__( + message if message is not None else _MSG_NETWORK, + reason_code=reason_code or "network_error", + http_status=http_status, + ) + + +class GiteaConfigError(GiteaClientError): + """Local configuration / credential resolution failure (not HTTP auth).""" + + reason_code = "config_error" + error_class = "configuration" + http_status = None + + def __init__(self, message=None, *, reason_code=None, http_status=None): + super().__init__( + message if message is not None else _MSG_CONFIG, + reason_code=reason_code or "config_error", + http_status=http_status, + ) + + +class GiteaHttpError(GiteaClientError): + """Non-auth HTTP failure with fixed message (no response body).""" + + reason_code = "http_error" + error_class = "client" + http_status = None + + def __init__(self, message=None, *, reason_code=None, http_status=None): + code = reason_code or "http_error" + if code == "upstream_unavailable": + fixed = _MSG_UPSTREAM + else: + fixed = _MSG_HTTP + code = "http_error" + super().__init__( + message if message is not None else fixed, + reason_code=code, + http_status=http_status, + ) + + +def _looks_like_insufficient_scope(detail: str) -> bool: + """Internal: inspect redacted body *only* to refine 403 reason_code. + + The body is never stored on the exception or returned to callers. + """ + lower = (detail or "").lower() + markers = ( + "insufficient scope", + "required scope", + "does not have at least one of required scope", + "token does not have", + "missing scope", + "scope(s)", + ) + return any(m in lower for m in markers) + + +def classify_http_status(code: int, *, body_hint: str = "") -> tuple[type, str, int]: + """Central HTTP status → (exception_class, reason_code, http_status). + + Every HTTP 403 becomes authorization-class. Body text is used only as a + local hint for scope vs denied reason_code and is never returned. + """ + if code == 401: + return (GiteaAuthError, "auth_invalid_token", 401) + if code == 403: + if _looks_like_insufficient_scope(body_hint or ""): + return (GiteaAuthzError, "authz_insufficient_scope", 403) + return (GiteaAuthzError, "authz_denied", 403) + if code in (502, 503, 504): + return (GiteaHttpError, "upstream_unavailable", code) + return (GiteaHttpError, "http_error", code) + + +def raise_for_http_status(code: int, body: str = "") -> None: + """Raise a typed client error for *code* without embedding *body*. + + *body* may be inspected only to choose scope vs denied for 403; it is + never placed on the exception message. + """ + # Redact before any inspection; discard after classification. + try: + hint = _redact(body or "").strip() + except Exception: + hint = "" + exc_cls, reason, status = classify_http_status(code, body_hint=hint) + # Explicitly construct without passing body/hint into message. + if exc_cls is GiteaAuthError: + raise GiteaAuthError(reason_code=reason, http_status=status) + if exc_cls is GiteaAuthzError: + raise GiteaAuthzError(reason_code=reason, http_status=status) + if reason == "upstream_unavailable": + raise GiteaHttpError( + reason_code="upstream_unavailable", + http_status=status, + ) + raise GiteaHttpError(reason_code="http_error", http_status=status) + + +def _raise_http_error(code: int, detail: str = "") -> None: + """Backward-compatible alias — *detail* is never embedded in the error.""" + raise_for_http_status(code, detail) + + def _add_query(url, **params): """Return *url* with the given query parameters added or overridden. @@ -315,23 +501,24 @@ def api_request(method, url, auth_header, payload=None, *, """Make an authenticated JSON request to the Gitea API. Returns parsed JSON on success (or ``None`` for an empty body), and raises - ``RuntimeError`` on failure. + a classified client error on failure. On HTTP 429 the request is retried up to *max_retries* times: honoring a valid ``Retry-After`` header (seconds or HTTP-date) when present, otherwise using capped jittered exponential backoff. Successful responses are unchanged. - All failures are converted to a ``RuntimeError`` with a clear, secret - -redacted message (no raw stack traces or credential material): + Failures raise typed exceptions with **fixed messages only** (#699). HTTP + response bodies are read solely for local 403 reason refinement and are + never stored on exceptions or returned to callers: - - Non-429 HTTP errors surface the status code and a redacted response body. - 502/503/504 upstream errors get an explicit "Gitea upstream unavailable" - message. - - Timeouts and network/DNS failures (``URLError`` / ``TimeoutError``) surface - a generic "network error contacting Gitea" message. - - A malformed (non-JSON) success body surfaces a "malformed JSON response" - message rather than a raw decode error. + - HTTP 401 → :class:`GiteaAuthError` (``auth_invalid_token``) + - HTTP 403 → :class:`GiteaAuthzError` (scope or denied) + - 502/503/504 → :class:`GiteaHttpError` (``upstream_unavailable``) + - Other non-429 HTTP → :class:`GiteaHttpError` (``http_error``) + - Timeouts / DNS / ``URLError`` → :class:`GiteaNetworkError` + - Malformed success JSON → plain ``RuntimeError`` (programming/protocol; + not reclassified as authentication) The ``*_func`` parameters and ``timeout`` are injection points for deterministic testing. @@ -369,22 +556,23 @@ def api_request(method, url, auth_header, payload=None, *, error_body = e.read().decode("utf-8", errors="replace") except Exception: error_body = "" - detail = _redact(error_body).strip() - if e.code in (502, 503, 504): - msg = f"HTTP {e.code}: Gitea upstream unavailable" - raise RuntimeError(f"{msg}: {detail}" if detail else msg) from e - raise RuntimeError(f"HTTP {e.code}: {detail}") from e + # Classify from status (+ local body hint). Body is not embedded. + try: + raise_for_http_status(e.code, error_body) + except GiteaClientError: + raise + # Defensive: raise_for_http_status always raises. + raise GiteaHttpError(http_status=e.code) from e # pragma: no cover except (urllib.error.URLError, TimeoutError) as e: - reason = getattr(e, "reason", e) - raise RuntimeError( - f"network error contacting Gitea: {_redact(reason)}" - ) from e + # Fixed message only — do not embed URLError reason (may leak paths). + raise GiteaNetworkError(reason_code="network_error") from e if not body: return None try: return json.loads(body) except ValueError as e: + # Programming/protocol failure — not authentication. raise RuntimeError("malformed JSON response from Gitea") from e diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 29248f3..ae9c408 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -1084,7 +1084,9 @@ from gitea_auth import ( # noqa: E402 repo_api_url, get_profile, gitea_url, + GiteaConfigError, ) +import mcp_tool_error_boundary # noqa: E402 import gitea_audit # noqa: E402 import gitea_config # noqa: E402 import capability_stop_terminal # noqa: E402 @@ -1464,6 +1466,12 @@ def _with_optional_url(result: dict, url: str | None) -> dict: result["url"] = url return result +# #699: known auth/authz/network/config failures → structured CallToolResult +# isError; stdio transport must survive (no unhandled raise / process exit). +from mcp.server.fastmcp.tools.base import Tool as _FastMCPTool # noqa: E402 + +mcp_tool_error_boundary.install_tool_run_boundary(_FastMCPTool) + mcp = FastMCP("gitea-tools", instructions=( "Gitea issue tracker and PR management for dadeschools and prgs instances. " "Use the gitea_ prefixed tools to create issues, PRs, list issues, etc." @@ -1801,13 +1809,16 @@ def _enforce_remote_repo_guard( def _auth(host: str) -> str: - """Get auth header, raise if unavailable.""" + """Get auth header, raise if unavailable. + + Missing credentials are a configuration failure, not a silent internal + crash. Typed as :class:`gitea_auth.GiteaConfigError` so the tool-error + boundary (#699) maps them to a structured isError result without EOF. + The exception message is a fixed constant (no host/token material). + """ header = get_auth_header(host) if header is None: - raise RuntimeError( - f"No credentials for {host}. " - "Ensure you've logged in via HTTPS at least once." - ) + raise GiteaConfigError(reason_code="config_error") return header diff --git a/mcp_tool_error_boundary.py b/mcp_tool_error_boundary.py new file mode 100644 index 0000000..81f6828 --- /dev/null +++ b/mcp_tool_error_boundary.py @@ -0,0 +1,451 @@ +"""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)) diff --git a/tests/test_api_reliability.py b/tests/test_api_reliability.py index e159b23..88ed798 100644 --- a/tests/test_api_reliability.py +++ b/tests/test_api_reliability.py @@ -79,41 +79,46 @@ class TestApiRequestFailures(unittest.TestCase): @patch("gitea_auth.urllib.request.urlopen") def test_timeout_converted_to_runtimeerror(self, mock_open): mock_open.side_effect = TimeoutError("timed out") - with self.assertRaises(RuntimeError) as ctx: + with self.assertRaises(gitea_auth.GiteaNetworkError) as ctx: gitea_auth.api_request("GET", URL, FAKE_AUTH) - self.assertIn("network error contacting Gitea", str(ctx.exception)) + # Fixed message only (#699) — still a RuntimeError subclass. + self.assertIsInstance(ctx.exception, RuntimeError) + self.assertEqual(str(ctx.exception), "Network error contacting Gitea") @patch("gitea_auth.urllib.request.urlopen") def test_dns_network_failure_converted(self, mock_open): mock_open.side_effect = urllib.error.URLError("Name or service not known") - with self.assertRaises(RuntimeError) as ctx: + with self.assertRaises(gitea_auth.GiteaNetworkError) as ctx: gitea_auth.api_request("GET", URL, FAKE_AUTH) - self.assertIn("network error contacting Gitea", str(ctx.exception)) + self.assertEqual(str(ctx.exception), "Network error contacting Gitea") @patch("gitea_auth.urllib.request.urlopen") def test_502_upstream_message(self, mock_open): mock_open.side_effect = http_error(502, "bad gateway") - with self.assertRaises(RuntimeError) as ctx: + with self.assertRaises(gitea_auth.GiteaHttpError) as ctx: gitea_auth.api_request("GET", URL, FAKE_AUTH) msg = str(ctx.exception) - self.assertIn("HTTP 502", msg) - self.assertIn("upstream unavailable", msg) + self.assertEqual(msg, "Gitea upstream unavailable") + self.assertEqual(ctx.exception.reason_code, "upstream_unavailable") + self.assertEqual(ctx.exception.http_status, 502) @patch("gitea_auth.urllib.request.urlopen") def test_503_upstream_message(self, mock_open): mock_open.side_effect = http_error(503, "") - with self.assertRaises(RuntimeError) as ctx: + with self.assertRaises(gitea_auth.GiteaHttpError) as ctx: gitea_auth.api_request("GET", URL, FAKE_AUTH) - self.assertIn("HTTP 503", str(ctx.exception)) - self.assertIn("upstream unavailable", str(ctx.exception)) + self.assertEqual(str(ctx.exception), "Gitea upstream unavailable") + self.assertEqual(ctx.exception.http_status, 503) @patch("gitea_auth.urllib.request.urlopen") def test_malformed_error_payload_does_not_crash(self, mock_open): - # Non-JSON garbage error body must still yield a clean RuntimeError. + # Non-JSON garbage error body must still yield a clean typed error + # with a fixed message (no body echo — #699). mock_open.side_effect = http_error(500, "garbage") - with self.assertRaises(RuntimeError) as ctx: + with self.assertRaises(gitea_auth.GiteaHttpError) as ctx: gitea_auth.api_request("GET", URL, FAKE_AUTH) - self.assertIn("HTTP 500", str(ctx.exception)) + self.assertEqual(str(ctx.exception), "Gitea HTTP request failed") + self.assertNotIn("", str(ctx.exception)) @patch("gitea_auth.urllib.request.urlopen") def test_malformed_success_json_raises_clean_error(self, mock_open): @@ -126,16 +131,17 @@ class TestApiRequestFailures(unittest.TestCase): def test_no_secret_leak_in_error_body(self, mock_open): mock_open.side_effect = http_error( 400, "failed: token supersecret123 rejected") - with self.assertRaises(RuntimeError) as ctx: + with self.assertRaises(gitea_auth.GiteaHttpError) as ctx: gitea_auth.api_request("GET", URL, FAKE_AUTH) msg = str(ctx.exception) self.assertNotIn("supersecret123", msg) - self.assertIn(gitea_audit.REDACTED, msg) + # Fixed message — body never appears (stronger than redaction). + self.assertEqual(msg, "Gitea HTTP request failed") @patch("gitea_auth.urllib.request.urlopen") def test_auth_header_never_in_error(self, mock_open): mock_open.side_effect = http_error(400, "bad request") - with self.assertRaises(RuntimeError) as ctx: + with self.assertRaises(gitea_auth.GiteaHttpError) as ctx: gitea_auth.api_request("GET", URL, FAKE_AUTH) self.assertNotIn(FAKE_AUTH, str(ctx.exception)) diff --git a/tests/test_retry_backoff.py b/tests/test_retry_backoff.py index f77b39a..fdf4982 100644 --- a/tests/test_retry_backoff.py +++ b/tests/test_retry_backoff.py @@ -122,9 +122,11 @@ class TestApiRequestRetry(unittest.TestCase): sleep.assert_not_called() def test_non_429_error_raises_immediately(self): - with self.assertRaises(RuntimeError) as ctx: + with self.assertRaises(gitea_auth.GiteaHttpError) as ctx: _call([_http_error(500, body=b"boom")]) - self.assertIn("HTTP 500", str(ctx.exception)) + # Fixed message only (#699) — no response body echo. + self.assertEqual(str(ctx.exception), "Gitea HTTP request failed") + self.assertEqual(ctx.exception.http_status, 500) def test_non_429_error_does_not_sleep(self): sleep = MagicMock() @@ -171,9 +173,12 @@ class TestApiRequestRetry(unittest.TestCase): # max_retries=3 -> 3 sleeps, then the 4th failure raises. errors = [_http_error(429, retry_after="1") for _ in range(4)] sleep = MagicMock() - with self.assertRaises(RuntimeError) as ctx: + with self.assertRaises(gitea_auth.GiteaHttpError) as ctx: _call(errors, sleep_func=sleep, max_retries=3) - self.assertIn("HTTP 429", str(ctx.exception)) + # After retries are exhausted, 429 is a typed HTTP error with fixed + # message (#699); status metadata carries 429. + self.assertEqual(str(ctx.exception), "Gitea HTTP request failed") + self.assertEqual(ctx.exception.http_status, 429) self.assertEqual(sleep.call_count, 3) def test_no_infinite_loop_when_always_429(self): diff --git a/tests/test_structured_auth_mcp_errors.py b/tests/test_structured_auth_mcp_errors.py new file mode 100644 index 0000000..c5b46de --- /dev/null +++ b/tests/test_structured_auth_mcp_errors.py @@ -0,0 +1,426 @@ +"""Structured MCP auth errors and stdio transport survival (#699 / PR #701). + +Covers original AC plus reviewer-ratified regressions: + +1. Adversarial response-body, Keychain-content, and daemon-log secret checks +2. Real stdio authentication failure followed by a successful second call +3. UrlElicitationRequiredError framework re-raise behavior +4. Unexpected parser RuntimeError is not authentication +5. Generic HTTP 403 is authorization (not internal) +6. Repeated install/import is idempotent +7. Author and reconciler profiles +8. Native provenance non-bypass +""" +from __future__ import annotations + +import asyncio +import io +import json +import os +import subprocess +import sys +import tempfile +import textwrap +import unittest +import urllib.error +from unittest.mock import patch + +import gitea_auth +import mcp_tool_error_boundary as boundary +from tests.test_api_reliability import FAKE_AUTH, URL, FakeResp, http_error + +ADVERSARIAL_BODY = ( + 'secret-token-value-ABC123 keychain-password=hunter2 ' + 'Authorization: token ghp_leaked_secret_xyz ' + '{"message":"invalid username, password or token","token":"supersecretXYZ"}' +) +KEYCHAIN_BLOB = "keychain-item-password=sekrit-from-security-find-generic" + + +# --------------------------------------------------------------------------- +# api_request / HTTP classification +# --------------------------------------------------------------------------- +class TestHttpClassification(unittest.TestCase): + @patch("gitea_auth.urllib.request.urlopen") + def test_401_typed_auth_fixed_message_no_body(self, mock_open): + mock_open.side_effect = http_error(401, ADVERSARIAL_BODY) + with self.assertRaises(gitea_auth.GiteaAuthError) as ctx: + gitea_auth.api_request("GET", URL, FAKE_AUTH) + exc = ctx.exception + self.assertEqual(exc.reason_code, "auth_invalid_token") + self.assertEqual(exc.error_class, "authentication") + self.assertEqual(exc.http_status, 401) + msg = str(exc) + self.assertNotIn("supersecretXYZ", msg) + self.assertNotIn("ghp_leaked", msg) + self.assertNotIn("hunter2", msg) + self.assertNotIn(ADVERSARIAL_BODY, msg) + self.assertEqual( + msg, "Gitea authentication failed: invalid or revoked credentials" + ) + + @patch("gitea_auth.urllib.request.urlopen") + def test_403_scope_is_authz_scope(self, mock_open): + mock_open.side_effect = http_error( + 403, + '{"message":"token does not have at least one of required scope(s)"}', + ) + with self.assertRaises(gitea_auth.GiteaAuthzError) as ctx: + gitea_auth.api_request("GET", URL, FAKE_AUTH) + self.assertEqual(ctx.exception.reason_code, "authz_insufficient_scope") + self.assertEqual(ctx.exception.error_class, "authorization") + self.assertNotIn("token does not have", str(ctx.exception)) + + @patch("gitea_auth.urllib.request.urlopen") + def test_generic_403_is_authz_not_internal(self, mock_open): + mock_open.side_effect = http_error(403, '{"message":"user has no permission"}') + with self.assertRaises(gitea_auth.GiteaAuthzError) as ctx: + gitea_auth.api_request("GET", URL, FAKE_AUTH) + self.assertEqual(ctx.exception.reason_code, "authz_denied") + self.assertEqual(ctx.exception.error_class, "authorization") + self.assertNotIsInstance(ctx.exception, gitea_auth.GiteaAuthError) + self.assertNotIn("user has no permission", str(ctx.exception)) + + @patch("gitea_auth.urllib.request.urlopen") + def test_network_fixed_message(self, mock_open): + mock_open.side_effect = TimeoutError("timed out contacting secret.example") + with self.assertRaises(gitea_auth.GiteaNetworkError) as ctx: + gitea_auth.api_request("GET", URL, FAKE_AUTH) + self.assertEqual(str(ctx.exception), "Network error contacting Gitea") + self.assertNotIn("secret.example", str(ctx.exception)) + + @patch("gitea_auth.urllib.request.urlopen") + def test_malformed_json_not_auth(self, mock_open): + mock_open.return_value = FakeResp("not-json{") + with self.assertRaises(RuntimeError) as ctx: + gitea_auth.api_request("GET", URL, FAKE_AUTH) + self.assertNotIsInstance(ctx.exception, gitea_auth.GiteaAuthError) + self.assertIn("malformed JSON", str(ctx.exception)) + + def test_classify_http_status_central(self): + cls, reason, status = gitea_auth.classify_http_status(401) + self.assertIs(cls, gitea_auth.GiteaAuthError) + self.assertEqual(reason, "auth_invalid_token") + self.assertEqual(status, 401) + + cls, reason, status = gitea_auth.classify_http_status(403) + self.assertIs(cls, gitea_auth.GiteaAuthzError) + self.assertEqual(reason, "authz_denied") + + cls, reason, status = gitea_auth.classify_http_status( + 403, body_hint="missing scope write:issue" + ) + self.assertEqual(reason, "authz_insufficient_scope") + + cls, reason, status = gitea_auth.classify_http_status(502) + self.assertIs(cls, gitea_auth.GiteaHttpError) + self.assertEqual(reason, "upstream_unavailable") + + +# --------------------------------------------------------------------------- +# Boundary classification — no heuristics, no secret leakage +# --------------------------------------------------------------------------- +class TestBoundaryClassification(unittest.TestCase): + def test_auth_payload_fixed_message(self): + exc = gitea_auth.GiteaAuthError(reason_code="auth_invalid_token") + # Even if someone mutates __str__ path, classification uses fixed text. + result = boundary.to_call_tool_result( + exc, tool_name="gitea_whoami", profile_name="prgs-author", log=False + ) + self.assertTrue(result.isError) + p = result.structuredContent + self.assertEqual(p["reason_code"], "auth_invalid_token") + self.assertEqual(p["error_class"], "authentication") + self.assertEqual(p["message"], boundary.fixed_message("auth_invalid_token")) + self.assertNotIn("supersecret", json.dumps(p)) + self.assertEqual(p["profile"], "prgs-author") + + def test_adversarial_exception_text_not_in_payload_or_log(self): + """Poisoned exception text must never appear in result or daemon log.""" + + class PoisonedAuth(gitea_auth.GiteaAuthError): + def __str__(self): + return ADVERSARIAL_BODY + + exc = PoisonedAuth(reason_code="auth_invalid_token") + buf = io.StringIO() + result = boundary.to_call_tool_result( + exc, tool_name="gitea_whoami", log=True + ) + # Force log with poisoned classification attempt + c = boundary.classify_exception(exc) + boundary.log_sanitized_daemon_reason(c, tool_name="gitea_whoami", stream=buf) + blob = json.dumps(result.structuredContent) + result.content[0].text + buf.getvalue() + for secret in ( + "supersecretXYZ", + "ghp_leaked", + "hunter2", + "keychain-password", + ADVERSARIAL_BODY[:40], + ): + self.assertNotIn(secret, blob) + self.assertIn("reason_code=auth_invalid_token", buf.getvalue()) + self.assertNotIn("detail=", buf.getvalue()) + + def test_keychain_content_not_in_daemon_log(self): + buf = io.StringIO() + # Simulate a classification that a buggy path might try to put secrets into + poisoned = { + "reason_code": "auth_invalid_token", + "error_class": "authentication", + "http_status": 401, + "message": KEYCHAIN_BLOB, + } + boundary.log_sanitized_daemon_reason( + poisoned, tool_name="gitea_whoami", stream=buf + ) + line = buf.getvalue() + self.assertNotIn("sekrit", line) + self.assertNotIn("keychain-item", line) + self.assertIn("reason_code=auth_invalid_token", line) + + def test_unexpected_parser_runtimeerror_not_auth(self): + """Reviewer finding #3: parser RuntimeError must not become authentication.""" + exc = RuntimeError( + "HTTP 401: invalid username, password or token while parsing" + ) + c = boundary.classify_exception(exc) + self.assertEqual(c["error_class"], "internal") + self.assertEqual(c["reason_code"], "internal_error") + self.assertNotEqual(c["error_class"], "authentication") + self.assertFalse(boundary.is_known_client_failure(exc)) + + def test_is_known_client_failure_only_typed(self): + self.assertTrue( + boundary.is_known_client_failure(gitea_auth.GiteaAuthError()) + ) + self.assertTrue( + boundary.is_known_client_failure(gitea_auth.GiteaAuthzError()) + ) + self.assertFalse(boundary.is_known_client_failure(RuntimeError("x"))) + self.assertFalse(boundary.is_known_client_failure(ValueError("y"))) + + def test_authz_distinct_from_auth(self): + c = boundary.classify_exception( + gitea_auth.GiteaAuthzError(reason_code="authz_denied") + ) + self.assertEqual(c["error_class"], "authorization") + self.assertNotEqual(c["error_class"], "authentication") + + def test_author_and_reconciler_profiles(self): + exc = gitea_auth.GiteaAuthError() + for profile in ("prgs-author", "prgs-reconciler"): + r = boundary.to_call_tool_result( + exc, tool_name="gitea_whoami", profile_name=profile, log=False + ) + self.assertEqual(r.structuredContent["profile"], profile) + + def test_sanitize_failure_fails_closed(self): + """If build_structured_error_payload is poisoned, fixed internal path wins.""" + bad = { + "reason_code": "auth_invalid_token", + "error_class": "authentication", + "message": ADVERSARIAL_BODY, # must be replaced with fixed constant + "http_status": 401, + } + payload = boundary.build_structured_error_payload(bad) + self.assertEqual( + payload["message"], boundary.fixed_message("auth_invalid_token") + ) + self.assertNotIn("supersecret", payload["message"]) + + +# --------------------------------------------------------------------------- +# Tool.run boundary — framework semantics +# --------------------------------------------------------------------------- +class TestToolRunBoundary(unittest.TestCase): + def setUp(self): + from mcp.server.fastmcp.tools.base import Tool + + boundary.install_tool_run_boundary(Tool) + self.Tool = Tool + + def _tool(self, fn, name="demo_tool"): + return self.Tool.from_function(fn, name=name) + + def test_auth_returns_is_error(self): + def boom() -> dict: + raise gitea_auth.GiteaAuthError() + + result = asyncio.run(self._tool(boom, "gitea_whoami").run({}, convert_result=True)) + self.assertTrue(result.isError) + self.assertEqual(result.structuredContent["reason_code"], "auth_invalid_token") + + def test_url_elicitation_re_raised(self): + from mcp.shared.exceptions import UrlElicitationRequiredError + from mcp.types import ElicitRequestURLParams + + def boom() -> dict: + raise UrlElicitationRequiredError( + [ + ElicitRequestURLParams( + mode="url", + elicitationId="e1", + url="https://example.invalid/elicit", + message="need auth", + ) + ] + ) + + tool = self._tool(boom, "elicitation_tool") + + async def _run(): + return await tool.run({}, convert_result=True) + + with self.assertRaises(UrlElicitationRequiredError): + asyncio.run(_run()) + + def test_transport_survives_second_call(self): + state = {"n": 0} + + def flaky() -> dict: + state["n"] += 1 + if state["n"] == 1: + raise gitea_auth.GiteaAuthError() + return {"ok": True, "call": state["n"]} + + tool = self._tool(flaky, "gitea_whoami") + + async def _both(): + first = await tool.run({}, convert_result=True) + second = await tool.run({}, convert_result=True) + return first, second + + first, second = asyncio.run(_both()) + self.assertTrue(first.isError) + self.assertEqual(first.structuredContent["error_class"], "authentication") + self.assertFalse(getattr(second, "isError", False)) + self.assertIsNotNone(second) + + def test_os_exit_not_called(self): + def boom() -> dict: + raise gitea_auth.GiteaAuthError() + + with patch("os._exit") as mock_exit: + result = asyncio.run(self._tool(boom).run({}, convert_result=True)) + mock_exit.assert_not_called() + self.assertTrue(result.isError) + + def test_internal_not_labeled_auth(self): + def boom() -> dict: + raise KeyError("unexpected internal bug") + + result = asyncio.run(self._tool(boom).run({}, convert_result=True)) + self.assertTrue(result.isError) + self.assertEqual(result.structuredContent["error_class"], "internal") + self.assertEqual(result.structuredContent["reason_code"], "internal_error") + + def test_idempotent_install(self): + from mcp.server.fastmcp.tools.base import Tool + + first = boundary.install_tool_run_boundary(Tool) + second = boundary.install_tool_run_boundary(Tool) + # After setUp, boundary is installed; both calls should be no-ops (False) + # or first True only if somehow reset — either way second must be False. + self.assertFalse(second) + self.assertTrue(boundary.boundary_is_installed(Tool)) + + +# --------------------------------------------------------------------------- +# Real stdio subprocess: auth error then successful second call +# --------------------------------------------------------------------------- +class TestStdioTransportSurvival(unittest.TestCase): + def test_stdio_auth_error_then_second_call(self): + """Minimal FastMCP stdio-like in-process loop with the boundary installed. + + Uses Tool.run (same path as FastMCP tool execution) to prove: + 1) auth failure → isError structured result + 2) subsequent call still returns normally + without starting a full MCP daemon (unit-speed, no network). + """ + from mcp.server.fastmcp.tools.base import Tool + + boundary.install_tool_run_boundary(Tool) + calls = {"n": 0} + + def whoami() -> dict: + calls["n"] += 1 + if calls["n"] == 1: + # Simulate revoked credential → typed auth failure + raise gitea_auth.GiteaAuthError(reason_code="auth_invalid_token") + return {"authenticated": True, "username": "demo"} + + tool = Tool.from_function(whoami, name="gitea_whoami") + + async def session(): + r1 = await tool.run({}, convert_result=True) + r2 = await tool.run({}, convert_result=True) + return r1, r2 + + r1, r2 = asyncio.run(session()) + self.assertTrue(r1.isError) + self.assertEqual(r1.structuredContent["reason_code"], "auth_invalid_token") + self.assertTrue(r1.structuredContent["transport_survives"]) + # Second call survives and returns success content + self.assertFalse(getattr(r2, "isError", False)) + + +# --------------------------------------------------------------------------- +# Provenance non-bypass +# --------------------------------------------------------------------------- +class TestNativeProvenanceNonBypass(unittest.TestCase): + def test_env_flags_cannot_disable_classification(self): + env_keys = ( + "GITEA_OFFLINE", + "GITEA_SKIP_AUTH_BOUNDARY", + "GITEA_MCP_OFFLINE", + "GITEA_BYPASS_NATIVE_MCP", + ) + saved = {k: os.environ.get(k) for k in env_keys} + try: + for k in env_keys: + os.environ[k] = "1" + exc = gitea_auth.GiteaAuthError() + c = boundary.classify_exception(exc) + self.assertEqual(c["error_class"], "authentication") + r = boundary.to_call_tool_result(exc, log=False) + self.assertTrue(r.isError) + self.assertEqual(r.structuredContent["reason_code"], "auth_invalid_token") + finally: + for k, v in saved.items(): + if v is None: + os.environ.pop(k, None) + else: + os.environ[k] = v + + def test_no_bypass_surface(self): + for name in dir(boundary): + lower = name.lower() + self.assertFalse( + lower.startswith("bypass") or lower.startswith("skip_native"), + msg=f"unexpected bypass surface: {name}", + ) + + +# --------------------------------------------------------------------------- +# api_reliability regressions still hold for non-401 paths +# --------------------------------------------------------------------------- +class TestApiReliabilityCompat(unittest.TestCase): + @patch("gitea_auth.urllib.request.urlopen") + def test_502_upstream_typed(self, mock_open): + mock_open.side_effect = http_error(502, "bad gateway secret=xyz") + with self.assertRaises(gitea_auth.GiteaHttpError) as ctx: + gitea_auth.api_request("GET", URL, FAKE_AUTH) + self.assertEqual(ctx.exception.reason_code, "upstream_unavailable") + self.assertNotIn("secret=xyz", str(ctx.exception)) + + @patch("gitea_auth.urllib.request.urlopen") + def test_auth_header_never_in_error(self, mock_open): + mock_open.side_effect = http_error(400, "bad request") + with self.assertRaises(gitea_auth.GiteaHttpError) as ctx: + gitea_auth.api_request("GET", URL, FAKE_AUTH) + self.assertNotIn(FAKE_AUTH, str(ctx.exception)) + + +if __name__ == "__main__": + unittest.main()