From b4d0cb22e452a07c8e816745c704ddb0f43edf0b Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Mon, 13 Jul 2026 13:10:00 -0400 Subject: [PATCH] fix: map Gitea auth failures to structured MCP tool errors (Closes #699) HTTP 401/scope-403/network failures become typed client errors with stable reason codes. The FastMCP Tool.run boundary returns CallToolResult isError payloads so stdio transport survives; daemon logs use sanitized reason codes only. Regression tests cover author/reconciler profiles, transport survival, and non-misclassification of unexpected exceptions. Cross-links: #685, #695, #697, #698, PR #696 (scopes not absorbed). --- gitea_auth.py | 122 ++++++-- gitea_mcp_server.py | 20 +- mcp_tool_error_boundary.py | 313 ++++++++++++++++++++ tests/test_structured_auth_mcp_errors.py | 346 +++++++++++++++++++++++ 4 files changed, 782 insertions(+), 19 deletions(-) create mode 100644 mcp_tool_error_boundary.py create mode 100644 tests/test_structured_auth_mcp_errors.py diff --git a/gitea_auth.py b/gitea_auth.py index a02d92a..7c4c12c 100644 --- a/gitea_auth.py +++ b/gitea_auth.py @@ -243,6 +243,96 @@ def _redact(text): return str(text) +# ── Classified client failures (#699) ───────────────────────────────────────── +# Subclasses of RuntimeError preserve existing ``except RuntimeError`` call +# sites. The MCP tool-error boundary maps these to sanitized CallToolResult +# isError payloads so auth-class failures never terminate stdio transport. + + +class GiteaClientError(RuntimeError): + """Base for known Gitea client failures with a stable reason_code.""" + + reason_code = "client_error" + error_class = "client" + http_status = None + + def __init__(self, message, *, reason_code=None, http_status=None): + super().__init__(message) + if reason_code is not None: + self.reason_code = reason_code + if http_status is not None: + self.http_status = http_status + + +class GiteaAuthError(GiteaClientError): + """Authentication failure (invalid/revoked credentials → typically HTTP 401).""" + + reason_code = "auth_failed" + error_class = "authentication" + http_status = 401 + + +class GiteaAuthzError(GiteaClientError): + """Authorization / insufficient-scope failure (typically HTTP 403 + scope).""" + + reason_code = "authz_insufficient_scope" + error_class = "authorization" + http_status = 403 + + +class GiteaNetworkError(GiteaClientError): + """Transport / DNS / timeout failure contacting Gitea.""" + + reason_code = "network_error" + error_class = "network" + http_status = None + + +class GiteaConfigError(GiteaClientError): + """Local configuration / credential resolution failure (not HTTP auth).""" + + reason_code = "config_error" + error_class = "configuration" + http_status = None + + +def _looks_like_insufficient_scope(detail: str) -> bool: + """True when a 403 body indicates token scope deficiency, not generic deny.""" + 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 _raise_http_error(code: int, detail: str) -> None: + """Raise a classified client error for a non-retryable HTTP failure.""" + safe = _redact(detail).strip() + if code == 401: + msg = f"HTTP 401: {safe}" if safe else "HTTP 401: authentication failed" + raise GiteaAuthError( + msg, + reason_code="auth_invalid_token", + http_status=401, + ) + if code == 403 and _looks_like_insufficient_scope(safe): + msg = f"HTTP 403: {safe}" if safe else "HTTP 403: insufficient scope" + raise GiteaAuthzError( + msg, + reason_code="authz_insufficient_scope", + http_status=403, + ) + if code in (502, 503, 504): + msg = f"HTTP {code}: Gitea upstream unavailable" + raise RuntimeError(f"{msg}: {safe}" if safe else msg) + raise RuntimeError(f"HTTP {code}: {safe}" if safe else f"HTTP {code}") + + def _add_query(url, **params): """Return *url* with the given query parameters added or overridden. @@ -315,23 +405,21 @@ 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): + All failures use a clear, secret-redacted message (no raw stack traces or + credential material). Classification (#699): - - 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 with scope deficiency → :class:`GiteaAuthzError` + - Other non-429 HTTP errors → ``RuntimeError`` (502/503/504 note upstream) + - Timeouts / DNS / ``URLError`` → :class:`GiteaNetworkError` + - Malformed success JSON → ``RuntimeError`` (not reclassified as auth) The ``*_func`` parameters and ``timeout`` are injection points for deterministic testing. @@ -370,14 +458,16 @@ def api_request(method, url, auth_header, payload=None, *, 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 + try: + _raise_http_error(e.code, detail) + except Exception as mapped: + raise mapped from e + raise RuntimeError(f"HTTP {e.code}: {detail}") 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)}" + raise GiteaNetworkError( + f"network error contacting Gitea: {_redact(reason)}", + reason_code="network_error", ) from e if not body: diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 29248f3..769f605 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,12 +1809,18 @@ 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. + """ header = get_auth_header(host) if header is None: - raise RuntimeError( + raise GiteaConfigError( f"No credentials for {host}. " - "Ensure you've logged in via HTTPS at least once." + "Ensure you've logged in via HTTPS at least once.", + 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..cb3ebae --- /dev/null +++ b/mcp_tool_error_boundary.py @@ -0,0 +1,313 @@ +"""MCP tool-boundary error mapping for known Gitea client failures (#699). + +Known authentication / authorization / network / configuration failures must +leave the tool boundary as a sanitized structured ``CallToolResult`` with +``isError=True``. The stdio transport must remain connected; callers must +never observe EOF for recoverable auth-class defects. + +Unexpected exceptions are mapped to ``internal_error`` and are never labeled +as authentication failures. +""" + +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 (issue #699 AC). +REASON_AUTH_FAILED = "auth_failed" +REASON_AUTH_INVALID_TOKEN = "auth_invalid_token" +REASON_AUTHZ_INSUFFICIENT_SCOPE = "authz_insufficient_scope" +REASON_NETWORK_ERROR = "network_error" +REASON_CONFIG_ERROR = "config_error" +REASON_INTERNAL_ERROR = "internal_error" + +ERROR_CLASS_AUTHENTICATION = "authentication" +ERROR_CLASS_AUTHORIZATION = "authorization" +ERROR_CLASS_NETWORK = "network" +ERROR_CLASS_CONFIGURATION = "configuration" +ERROR_CLASS_INTERNAL = "internal" + +# Tokens / secret substrings that must never appear in tool error text. +_SECRET_MARKERS = ( + "token ", + "bearer ", + "basic ", + "authorization:", + "password=", + "keychain", +) + + +def _redact_text(text: str) -> str: + try: + from gitea_auth import _redact + + return _redact(text) + except Exception: + return str(text) + + +def _safe_message(message: str) -> str: + """Redact secrets and drop obviously sensitive fragments.""" + redacted = _redact_text(message or "") + lower = redacted.lower() + for marker in _SECRET_MARKERS: + if marker in lower and marker.strip() not in ("keychain",): + # Already redacted by gitea_auth; keep length bounded. + break + # Never echo raw multi-line bodies that might hold tokens. + one_line = " ".join(redacted.split()) + if len(one_line) > 400: + one_line = one_line[:400] + "…" + return one_line + + +def classify_exception(exc: BaseException) -> dict[str, Any]: + """Return a structured classification for *exc*. + + Only known auth/authz/network/config classes receive those labels. + Everything else is ``internal_error`` — never silently rebranded as auth. + """ + # Lazy import avoids circular import at module load (gitea_auth imports + # are safe; typed exceptions live there). + import gitea_auth + + if isinstance(exc, gitea_auth.GiteaAuthError): + return { + "reason_code": getattr(exc, "reason_code", None) or REASON_AUTH_FAILED, + "error_class": ERROR_CLASS_AUTHENTICATION, + "http_status": getattr(exc, "http_status", None) or 401, + "message": _safe_message(str(exc)), + "transport_survives": True, + } + if isinstance(exc, gitea_auth.GiteaAuthzError): + return { + "reason_code": getattr(exc, "reason_code", None) + or REASON_AUTHZ_INSUFFICIENT_SCOPE, + "error_class": ERROR_CLASS_AUTHORIZATION, + "http_status": getattr(exc, "http_status", None) or 403, + "message": _safe_message(str(exc)), + "transport_survives": True, + } + if isinstance(exc, gitea_auth.GiteaNetworkError): + return { + "reason_code": getattr(exc, "reason_code", None) or REASON_NETWORK_ERROR, + "error_class": ERROR_CLASS_NETWORK, + "http_status": getattr(exc, "http_status", None), + "message": _safe_message(str(exc)), + "transport_survives": True, + } + if isinstance(exc, gitea_auth.GiteaConfigError): + return { + "reason_code": getattr(exc, "reason_code", None) or REASON_CONFIG_ERROR, + "error_class": ERROR_CLASS_CONFIGURATION, + "http_status": getattr(exc, "http_status", None), + "message": _safe_message(str(exc)), + "transport_survives": True, + } + + # gitea_config.ConfigError is configuration, not authentication. + 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": _safe_message(str(exc)), + "transport_survives": True, + } + except Exception: + pass + + # Heuristic fallback only for already-redacted RuntimeError messages that + # historically used the plain "HTTP 401/403" form before typed exceptions. + # Never treat arbitrary RuntimeError as auth. + if isinstance(exc, RuntimeError): + text = str(exc) + lower = text.lower() + if lower.startswith("http 401") or "invalid username, password or token" in lower: + return { + "reason_code": REASON_AUTH_INVALID_TOKEN, + "error_class": ERROR_CLASS_AUTHENTICATION, + "http_status": 401, + "message": _safe_message(text), + "transport_survives": True, + } + if "insufficient scope" in lower or ( + lower.startswith("http 403") and "scope" in lower + ): + return { + "reason_code": REASON_AUTHZ_INSUFFICIENT_SCOPE, + "error_class": ERROR_CLASS_AUTHORIZATION, + "http_status": 403, + "message": _safe_message(text), + "transport_survives": True, + } + if "network error contacting gitea" in lower: + return { + "reason_code": REASON_NETWORK_ERROR, + "error_class": ERROR_CLASS_NETWORK, + "http_status": None, + "message": _safe_message(text), + "transport_survives": True, + } + + return { + "reason_code": REASON_INTERNAL_ERROR, + "error_class": ERROR_CLASS_INTERNAL, + "http_status": None, + "message": _safe_message(str(exc) or type(exc).__name__), + "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 for tool errors (no secrets).""" + payload: dict[str, Any] = { + "success": False, + "isError": True, + "reason_code": classification["reason_code"], + "error_class": classification["error_class"], + "message": classification["message"], + "transport_survives": True, + "retryable": classification["error_class"] + in {ERROR_CLASS_AUTHENTICATION, ERROR_CLASS_NETWORK, ERROR_CLASS_CONFIGURATION}, + } + if classification.get("http_status") is not None: + payload["http_status"] = classification["http_status"] + if tool_name: + payload["tool"] = tool_name + if profile_name: + payload["profile"] = profile_name + return payload + + +def log_sanitized_daemon_reason( + classification: dict[str, Any], + *, + tool_name: str | None = None, + stream=None, +) -> None: + """Write an actionable, secret-free reason line to the daemon log.""" + stream = stream if stream is not None else sys.stderr + parts = [ + "mcp_tool_error", + f"reason_code={classification.get('reason_code')}", + f"error_class={classification.get('error_class')}", + ] + if tool_name: + parts.append(f"tool={tool_name}") + status = classification.get("http_status") + if status is not None: + parts.append(f"http_status={status}") + # Message already sanitized; still scan for secret markers. + msg = _safe_message(str(classification.get("message") or "")) + for marker in ("token ", "Bearer ", "Basic ", "password="): + if marker.lower() in msg.lower(): + msg = "[redacted]" + break + parts.append(f"detail={msg}") + line = " ".join(parts) + try: + stream.write(line + "\n") + if hasattr(stream, "flush"): + stream.flush() + except Exception: + pass + logger.warning(line) + + +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 + + 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, + ) + + +def is_known_client_failure(exc: BaseException) -> bool: + """True when *exc* is a known classified client failure (not internal).""" + classification = classify_exception(exc) + return classification["error_class"] != ERROR_CLASS_INTERNAL or isinstance( + exc, RuntimeError + ) + + +def install_tool_run_boundary(Tool) -> None: + """Patch FastMCP ``Tool.run`` so failures become structured isError results. + + Auth/authz/network/config failures carry their reason codes. Unexpected + exceptions map to ``internal_error`` — never reclassified as auth. The + stdio transport receives ``CallToolResult(isError=True)`` instead of an + unhandled raise path that some hosts surface as EOF (#699). + """ + if getattr(Tool.run, "_gitea_auth_boundary_installed", False): + return + + original_run = Tool.run + + async def run_boundary( + self, + arguments: dict[str, Any], + context=None, + convert_result: bool = False, + ) -> Any: + try: + result = await self.fn_metadata.call_fn_with_arg_validation( + self.fn, + self.is_async, + arguments, + {self.context_kwarg: context} + if self.context_kwarg is not None + else None, + ) + if convert_result: + result = self.fn_metadata.convert_result(result) + return result + except Exception as exc: + profile_name = None + try: + from gitea_auth import get_profile + + profile_name = (get_profile() or {}).get("profile_name") + except Exception: + profile_name = None + + # Always return structured isError CallToolResult so stdio survives. + return to_call_tool_result( + exc, + tool_name=getattr(self, "name", None), + profile_name=profile_name, + ) + + run_boundary._gitea_auth_boundary_installed = True # type: ignore[attr-defined] + run_boundary._gitea_auth_boundary_original = original_run # type: ignore[attr-defined] + Tool.run = run_boundary # type: ignore[method-assign] diff --git a/tests/test_structured_auth_mcp_errors.py b/tests/test_structured_auth_mcp_errors.py new file mode 100644 index 0000000..bf0941e --- /dev/null +++ b/tests/test_structured_auth_mcp_errors.py @@ -0,0 +1,346 @@ +"""Structured MCP auth errors and stdio transport survival (#699). + +Acceptance criteria coverage: +- Known Gitea auth failures → sanitized structured isError CallToolResult +- Transport survives (no process exit / os._exit on auth failure) +- Subsequent tool call still returns a structured response +- Auth vs authorization vs network vs config vs internal distinction +- Unexpected exceptions are not misclassified as authentication +- Secret leakage scan of tool error text and daemon reason codes +- Author and reconciler profile labels covered in classification payloads +- Native provenance non-bypass: env flag / offline runner cannot skip the + structured boundary mapping for auth failures +""" +from __future__ import annotations + +import io +import json +import os +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 + + +# --------------------------------------------------------------------------- +# api_request classification +# --------------------------------------------------------------------------- +class TestApiRequestAuthClassification(unittest.TestCase): + @patch("gitea_auth.urllib.request.urlopen") + def test_401_raises_gitea_auth_error(self, mock_open): + mock_open.side_effect = http_error( + 401, '{"message":"invalid username, password or token"}' + ) + with self.assertRaises(gitea_auth.GiteaAuthError) as ctx: + gitea_auth.api_request("GET", URL, FAKE_AUTH) + self.assertEqual(ctx.exception.reason_code, "auth_invalid_token") + self.assertEqual(ctx.exception.http_status, 401) + self.assertEqual(ctx.exception.error_class, "authentication") + self.assertIsInstance(ctx.exception, RuntimeError) + + @patch("gitea_auth.urllib.request.urlopen") + def test_403_scope_raises_authz(self, mock_open): + mock_open.side_effect = http_error( + 403, + '{"message":"token does not have at least one of required scope(s): [write:repository]"}', + ) + 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") + + @patch("gitea_auth.urllib.request.urlopen") + def test_403_generic_not_auth(self, mock_open): + mock_open.side_effect = http_error(403, '{"message":"user has no permission"}') + with self.assertRaises(RuntimeError) as ctx: + gitea_auth.api_request("GET", URL, FAKE_AUTH) + self.assertNotIsInstance(ctx.exception, gitea_auth.GiteaAuthError) + self.assertNotIsInstance(ctx.exception, gitea_auth.GiteaAuthzError) + self.assertIn("HTTP 403", str(ctx.exception)) + + @patch("gitea_auth.urllib.request.urlopen") + def test_network_raises_gitea_network_error(self, mock_open): + mock_open.side_effect = TimeoutError("timed out") + with self.assertRaises(gitea_auth.GiteaNetworkError) as ctx: + gitea_auth.api_request("GET", URL, FAKE_AUTH) + self.assertEqual(ctx.exception.reason_code, "network_error") + self.assertIn("network error contacting Gitea", 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)) + + @patch("gitea_auth.urllib.request.urlopen") + def test_401_redacts_token_in_body(self, mock_open): + mock_open.side_effect = http_error( + 401, "rejected token supersecret123 for user" + ) + with self.assertRaises(gitea_auth.GiteaAuthError) as ctx: + gitea_auth.api_request("GET", URL, FAKE_AUTH) + msg = str(ctx.exception) + self.assertNotIn("supersecret123", msg) + self.assertNotIn(FAKE_AUTH, msg) + + +# --------------------------------------------------------------------------- +# Boundary classification + CallToolResult +# --------------------------------------------------------------------------- +class TestToolErrorBoundary(unittest.TestCase): + def test_auth_error_to_call_tool_result(self): + exc = gitea_auth.GiteaAuthError( + "HTTP 401: invalid username, password or token", + reason_code="auth_invalid_token", + http_status=401, + ) + result = boundary.to_call_tool_result( + exc, tool_name="gitea_whoami", profile_name="prgs-author", log=False + ) + self.assertTrue(result.isError) + payload = result.structuredContent + self.assertEqual(payload["reason_code"], "auth_invalid_token") + self.assertEqual(payload["error_class"], "authentication") + self.assertTrue(payload["transport_survives"]) + self.assertEqual(payload["profile"], "prgs-author") + self.assertEqual(payload["tool"], "gitea_whoami") + text = result.content[0].text + self.assertNotIn("supersecret", text) + self.assertIn("auth_invalid_token", text) + + def test_authz_distinct_from_auth(self): + exc = gitea_auth.GiteaAuthzError( + "HTTP 403: token does not have at least one of required scope(s)", + reason_code="authz_insufficient_scope", + http_status=403, + ) + c = boundary.classify_exception(exc) + self.assertEqual(c["error_class"], "authorization") + self.assertNotEqual(c["error_class"], "authentication") + self.assertEqual(c["reason_code"], "authz_insufficient_scope") + + def test_network_and_config_classes(self): + net = boundary.classify_exception( + gitea_auth.GiteaNetworkError("network error contacting Gitea: timed out") + ) + self.assertEqual(net["error_class"], "network") + cfg = boundary.classify_exception( + gitea_auth.GiteaConfigError("No credentials for gitea.example.com") + ) + self.assertEqual(cfg["error_class"], "configuration") + + def test_unexpected_exception_not_auth(self): + c = boundary.classify_exception(ValueError("something weird broke")) + self.assertEqual(c["reason_code"], "internal_error") + self.assertEqual(c["error_class"], "internal") + self.assertNotEqual(c["error_class"], "authentication") + + def test_random_runtimeerror_not_auth(self): + c = boundary.classify_exception(RuntimeError("lock file write failed")) + self.assertEqual(c["reason_code"], "internal_error") + self.assertEqual(c["error_class"], "internal") + + def test_author_and_reconciler_profiles_in_payload(self): + exc = gitea_auth.GiteaAuthError( + "HTTP 401: invalid username, password or token", + reason_code="auth_invalid_token", + ) + for profile in ("prgs-author", "prgs-reconciler"): + result = boundary.to_call_tool_result( + exc, tool_name="gitea_whoami", profile_name=profile, log=False + ) + self.assertEqual(result.structuredContent["profile"], profile) + self.assertEqual( + result.structuredContent["reason_code"], "auth_invalid_token" + ) + + def test_daemon_log_has_reason_code_no_secrets(self): + buf = io.StringIO() + classification = { + "reason_code": "auth_invalid_token", + "error_class": "authentication", + "http_status": 401, + "message": "HTTP 401: invalid username, password or token secret=abc", + } + boundary.log_sanitized_daemon_reason( + classification, tool_name="gitea_whoami", stream=buf + ) + line = buf.getvalue() + self.assertIn("reason_code=auth_invalid_token", line) + self.assertIn("tool=gitea_whoami", line) + # The message may still contain "token" as English word in Gitea messages; + # ensure raw credential material markers are not present as values. + self.assertNotIn("secret=abc", line.replace(" ", "")) + + def test_secret_markers_stripped_from_payload(self): + exc = gitea_auth.GiteaAuthError( + "HTTP 401: failed token supersecretXYZ rejected" + ) + # Simulate pre-redacted path via classify after api_request-style redact. + with patch.object( + boundary, + "_redact_text", + return_value="HTTP 401: failed token [REDACTED] rejected", + ): + c = boundary.classify_exception(exc) + self.assertNotIn("supersecretXYZ", c["message"]) + + +# --------------------------------------------------------------------------- +# Tool.run boundary: transport survival + second call +# --------------------------------------------------------------------------- +class TestToolRunBoundaryInstall(unittest.TestCase): + def setUp(self): + from mcp.server.fastmcp.tools.base import Tool + + # Re-install is a no-op when already patched by gitea_mcp_server import. + boundary.install_tool_run_boundary(Tool) + self.Tool = Tool + + def _make_tool(self, fn, name="demo_tool"): + return self.Tool.from_function(fn, name=name) + + def test_auth_failure_returns_is_error_not_raise(self): + def boom() -> dict: + raise gitea_auth.GiteaAuthError( + "HTTP 401: invalid username, password or token", + reason_code="auth_invalid_token", + http_status=401, + ) + + tool = self._make_tool(boom, name="gitea_whoami") + import asyncio + + result = asyncio.run(tool.run({}, convert_result=True)) + self.assertTrue(getattr(result, "isError", False)) + self.assertEqual( + result.structuredContent["reason_code"], "auth_invalid_token" + ) + + def test_transport_survives_second_call(self): + """After an auth failure, a subsequent call still gets a structured result.""" + state = {"n": 0} + + def flaky() -> dict: + state["n"] += 1 + if state["n"] == 1: + raise gitea_auth.GiteaAuthError( + "HTTP 401: invalid username, password or token", + reason_code="auth_invalid_token", + ) + return {"ok": True, "call": state["n"]} + + tool = self._make_tool(flaky, name="gitea_whoami") + import asyncio + + 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") + # Second call succeeds (or would return another structured error — not EOF). + self.assertFalse(getattr(second, "isError", False)) + # convert_result for dict returns content blocks / structured form + # depending on FastMCP version — assert process continued. + self.assertIsNotNone(second) + + def test_auth_failure_does_not_call_os_exit(self): + def boom() -> dict: + raise gitea_auth.GiteaAuthError( + "HTTP 401: invalid username, password or token", + reason_code="auth_invalid_token", + ) + + tool = self._make_tool(boom) + import asyncio + + with patch("os._exit") as mock_exit: + result = asyncio.run(tool.run({}, convert_result=True)) + mock_exit.assert_not_called() + self.assertTrue(result.isError) + + def test_reconciler_profile_auth_failure_structured(self): + def boom() -> dict: + raise gitea_auth.GiteaAuthError( + "HTTP 401: invalid username, password or token", + reason_code="auth_invalid_token", + ) + + tool = self._make_tool(boom, name="gitea_list_issues") + import asyncio + + with patch( + "gitea_auth.get_profile", + return_value={"profile_name": "prgs-reconciler"}, + ): + result = asyncio.run(tool.run({}, convert_result=True)) + self.assertTrue(result.isError) + self.assertEqual(result.structuredContent.get("profile"), "prgs-reconciler") + + def test_internal_exception_not_labeled_auth(self): + def boom() -> dict: + raise KeyError("unexpected internal bug") + + tool = self._make_tool(boom) + import asyncio + + result = asyncio.run(tool.run({}, convert_result=True)) + self.assertTrue(result.isError) + self.assertEqual(result.structuredContent["error_class"], "internal") + self.assertEqual(result.structuredContent["reason_code"], "internal_error") + + +# --------------------------------------------------------------------------- +# Provenance: no env flag / offline path skips structured boundary for auth +# --------------------------------------------------------------------------- +class TestNativeProvenanceNonBypass(unittest.TestCase): + def test_env_flag_cannot_disable_classification(self): + """No supported env flag turns auth failures into unlabeled exits.""" + # Even with various offline/test flags set, classification remains. + 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( + "HTTP 401: invalid username, password or token", + reason_code="auth_invalid_token", + ) + c = boundary.classify_exception(exc) + self.assertEqual(c["error_class"], "authentication") + result = boundary.to_call_tool_result(exc, log=False) + self.assertTrue(result.isError) + self.assertEqual(result.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_attribute_on_boundary(self): + """Boundary module must not expose an offline bypass switch.""" + for name in dir(boundary): + lower = name.lower() + self.assertFalse( + lower.startswith("bypass") or lower.startswith("skip_native"), + msg=f"unexpected bypass surface: {name}", + ) + + +if __name__ == "__main__": + unittest.main()