diff --git a/gitea_auth.py b/gitea_auth.py index 7c4c12c..fe2b769 100644 --- a/gitea_auth.py +++ b/gitea_auth.py @@ -245,40 +245,74 @@ def _redact(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. +# 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 a stable reason_code.""" + """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, *, reason_code=None, http_status=None): - super().__init__(message) + 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_failed" + 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 / insufficient-scope failure (typically HTTP 403 + scope).""" + """Authorization failure (HTTP 403 — scope deficiency or access denied).""" - reason_code = "authz_insufficient_scope" + 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.""" @@ -287,6 +321,13 @@ class GiteaNetworkError(GiteaClientError): 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).""" @@ -295,9 +336,40 @@ class GiteaConfigError(GiteaClientError): 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: - """True when a 403 body indicates token scope deficiency, not generic deny.""" + """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", @@ -310,27 +382,51 @@ def _looks_like_insufficient_scope(detail: str) -> bool: 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() +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: - 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, - ) + 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): - 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}") + 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): @@ -412,14 +508,17 @@ def api_request(method, url, auth_header, payload=None, *, using capped jittered exponential backoff. Successful responses are unchanged. - All failures use a clear, secret-redacted message (no raw stack traces or - credential material). Classification (#699): + 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: - 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) + - 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 → ``RuntimeError`` (not reclassified as auth) + - Malformed success JSON → plain ``RuntimeError`` (programming/protocol; + not reclassified as authentication) The ``*_func`` parameters and ``timeout`` are injection points for deterministic testing. @@ -457,24 +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() + # Classify from status (+ local body hint). Body is not embedded. 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 + 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 GiteaNetworkError( - f"network error contacting Gitea: {_redact(reason)}", - reason_code="network_error", - ) 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 769f605..ae9c408 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -1814,14 +1814,11 @@ def _auth(host: str) -> str: 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 GiteaConfigError( - f"No credentials for {host}. " - "Ensure you've logged in via HTTPS at least once.", - reason_code="config_error", - ) + raise GiteaConfigError(reason_code="config_error") return header diff --git a/mcp_tool_error_boundary.py b/mcp_tool_error_boundary.py index cb3ebae..81f6828 100644 --- a/mcp_tool_error_boundary.py +++ b/mcp_tool_error_boundary.py @@ -1,12 +1,22 @@ """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. +Known authentication / authorization / network / configuration failures leave +the tool boundary as a sanitized structured ``CallToolResult`` with +``isError=True``. Stdio transport remains connected. -Unexpected exceptions are mapped to ``internal_error`` and are never labeled -as authentication failures. +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 @@ -18,100 +28,191 @@ from typing import Any logger = logging.getLogger("gitea_mcp.tool_error_boundary") -# Stable reason codes (issue #699 AC). +# 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" -# Tokens / secret substrings that must never appear in tool error text. -_SECRET_MARKERS = ( - "token ", - "bearer ", - "basic ", - "authorization:", - "password=", - "keychain", -) +# 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 _redact_text(text: str) -> str: +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 _redact + from gitea_auth import get_profile - return _redact(text) + 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 str(text) + return None -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 _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 a structured classification for *exc*. + """Return structured classification with **fixed** messages only. - Only known auth/authz/network/config classes receive those labels. - Everything else is ``internal_error`` — never silently rebranded as auth. + 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``. """ - # Lazy import avoids circular import at module load (gitea_auth imports - # are safe; typed exceptions live there). + 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": getattr(exc, "reason_code", None) or REASON_AUTH_FAILED, + "reason_code": code, "error_class": ERROR_CLASS_AUTHENTICATION, "http_status": getattr(exc, "http_status", None) or 401, - "message": _safe_message(str(exc)), + "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": getattr(exc, "reason_code", None) - or REASON_AUTHZ_INSUFFICIENT_SCOPE, + "reason_code": code, "error_class": ERROR_CLASS_AUTHORIZATION, "http_status": getattr(exc, "http_status", None) or 403, - "message": _safe_message(str(exc)), + "message": fixed_message(code), "transport_survives": True, } if isinstance(exc, gitea_auth.GiteaNetworkError): return { - "reason_code": getattr(exc, "reason_code", None) or REASON_NETWORK_ERROR, + "reason_code": REASON_NETWORK_ERROR, "error_class": ERROR_CLASS_NETWORK, "http_status": getattr(exc, "http_status", None), - "message": _safe_message(str(exc)), + "message": fixed_message(REASON_NETWORK_ERROR), "transport_survives": True, } if isinstance(exc, gitea_auth.GiteaConfigError): return { - "reason_code": getattr(exc, "reason_code", None) or REASON_CONFIG_ERROR, + "reason_code": REASON_CONFIG_ERROR, "error_class": ERROR_CLASS_CONFIGURATION, "http_status": getattr(exc, "http_status", None), - "message": _safe_message(str(exc)), + "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, } - # gitea_config.ConfigError is configuration, not authentication. try: import gitea_config @@ -120,50 +221,23 @@ def classify_exception(exc: BaseException) -> dict[str, Any]: "reason_code": REASON_CONFIG_ERROR, "error_class": ERROR_CLASS_CONFIGURATION, "http_status": None, - "message": _safe_message(str(exc)), + "message": fixed_message(REASON_CONFIG_ERROR), "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, - } + # 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": _safe_message(str(exc) or type(exc).__name__), + "message": fixed_message(REASON_INTERNAL_ERROR), "transport_survives": True, } @@ -174,24 +248,46 @@ def build_structured_error_payload( 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 + """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( @@ -200,33 +296,43 @@ def log_sanitized_daemon_reason( tool_name: str | None = None, stream=None, ) -> None: - """Write an actionable, secret-free reason line to the daemon log.""" + """Daemon log: reason codes only — never exception text or response bodies.""" 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: + 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: - pass - logger.warning(line) + # 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( @@ -239,38 +345,62 @@ def to_call_tool_result( """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, - ) + 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 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) -> bool: + """Install a **narrow** error boundary around FastMCP ``Tool.run``. + Strategy (preserves framework semantics): -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). + * 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, "_gitea_auth_boundary_installed", False): - return + 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 @@ -281,33 +411,41 @@ def install_tool_run_boundary(Tool) -> None: convert_result: bool = False, ) -> Any: try: - result = await self.fn_metadata.call_fn_with_arg_validation( - self.fn, - self.is_async, + return await original_run( + self, arguments, - {self.context_kwarg: context} - if self.context_kwarg is not None - else None, + context=context, + convert_result=convert_result, ) - 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 + 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 - profile_name = (get_profile() or {}).get("profile_name") - except Exception: - profile_name = None + # Prefer typed cause under FastMCP ToolError wrappers. + target: BaseException = exc + if isinstance(exc, ToolError) and exc.__cause__ is not None: + target = exc.__cause__ - # Always return structured isError CallToolResult so stdio survives. + # 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( - exc, + target, 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] + 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 index bf0941e..c5b46de 100644 --- a/tests/test_structured_auth_mcp_errors.py +++ b/tests/test_structured_auth_mcp_errors.py @@ -1,21 +1,26 @@ -"""Structured MCP auth errors and stdio transport survival (#699). +"""Structured MCP auth errors and stdio transport survival (#699 / PR #701). -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 +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 @@ -24,50 +29,65 @@ 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 classification +# api_request / HTTP classification # --------------------------------------------------------------------------- -class TestApiRequestAuthClassification(unittest.TestCase): +class TestHttpClassification(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"}' - ) + 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) - 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) + 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_raises_authz(self, mock_open): + 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): [write:repository]"}', + '{"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_403_generic_not_auth(self, mock_open): + 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(RuntimeError) as ctx: + 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.assertNotIsInstance(ctx.exception, gitea_auth.GiteaAuthzError) - self.assertIn("HTTP 403", str(ctx.exception)) + self.assertNotIn("user has no permission", 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") + 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(ctx.exception.reason_code, "network_error") - self.assertIn("network error contacting Gitea", str(ctx.exception)) + 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): @@ -77,166 +97,194 @@ class TestApiRequestAuthClassification(unittest.TestCase): 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" + 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" ) - 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) + 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 + CallToolResult +# Boundary classification — no heuristics, no secret leakage # --------------------------------------------------------------------------- -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, - ) +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) - 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) + 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_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_adversarial_exception_text_not_in_payload_or_log(self): + """Poisoned exception text must never appear in result or daemon log.""" - 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") + class PoisonedAuth(gitea_auth.GiteaAuthError): + def __str__(self): + return ADVERSARIAL_BODY - 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): + exc = PoisonedAuth(reason_code="auth_invalid_token") buf = io.StringIO() - classification = { + 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": "HTTP 401: invalid username, password or token secret=abc", + "message": KEYCHAIN_BLOB, } boundary.log_sanitized_daemon_reason( - classification, tool_name="gitea_whoami", stream=buf + 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) - 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" + 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" ) - # 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"]) + 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: transport survival + second call +# Tool.run boundary — framework semantics # --------------------------------------------------------------------------- -class TestToolRunBoundaryInstall(unittest.TestCase): +class TestToolRunBoundary(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"): + def _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 test_auth_returns_is_error(self): def boom() -> dict: - raise gitea_auth.GiteaAuthError( - "HTTP 401: invalid username, password or token", - reason_code="auth_invalid_token", - http_status=401, + 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._make_tool(boom, name="gitea_whoami") - import asyncio + tool = self._tool(boom, "elicitation_tool") - result = asyncio.run(tool.run({}, convert_result=True)) - self.assertTrue(getattr(result, "isError", False)) - self.assertEqual( - result.structuredContent["reason_code"], "auth_invalid_token" - ) + async def _run(): + return await tool.run({}, convert_result=True) + + with self.assertRaises(UrlElicitationRequiredError): + asyncio.run(_run()) 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", - ) + raise gitea_auth.GiteaAuthError() return {"ok": True, "call": state["n"]} - tool = self._make_tool(flaky, name="gitea_whoami") - import asyncio + tool = self._tool(flaky, "gitea_whoami") async def _both(): first = await tool.run({}, convert_result=True) @@ -244,68 +292,84 @@ class TestToolRunBoundaryInstall(unittest.TestCase): 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 test_os_exit_not_called(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 + raise gitea_auth.GiteaAuthError() with patch("os._exit") as mock_exit: - result = asyncio.run(tool.run({}, convert_result=True)) + result = asyncio.run(self._tool(boom).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 test_internal_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)) + 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)) + # --------------------------------------------------------------------------- -# Provenance: no env flag / offline path skips structured boundary for auth +# 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_flag_cannot_disable_classification(self): - """No supported env flag turns auth failures into unlabeled exits.""" - # Even with various offline/test flags set, classification remains. + def test_env_flags_cannot_disable_classification(self): env_keys = ( "GITEA_OFFLINE", "GITEA_SKIP_AUTH_BOUNDARY", @@ -316,15 +380,12 @@ class TestNativeProvenanceNonBypass(unittest.TestCase): 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", - ) + exc = gitea_auth.GiteaAuthError() 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") + 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: @@ -332,8 +393,7 @@ class TestNativeProvenanceNonBypass(unittest.TestCase): else: os.environ[k] = v - def test_no_bypass_attribute_on_boundary(self): - """Boundary module must not expose an offline bypass switch.""" + def test_no_bypass_surface(self): for name in dir(boundary): lower = name.lower() self.assertFalse( @@ -342,5 +402,25 @@ class TestNativeProvenanceNonBypass(unittest.TestCase): ) +# --------------------------------------------------------------------------- +# 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()