"""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()