"""Structured MCP auth errors and stdio transport survival (#699 / PR #701). Covers original AC plus reviewer-ratified regressions: 1. Adversarial response-body, Keychain-content, and daemon-log secret checks 2. Real stdio authentication failure followed by a successful second call 3. UrlElicitationRequiredError framework re-raise behavior 4. Unexpected parser RuntimeError is not authentication 5. Generic HTTP 403 is authorization (not internal) 6. Repeated install/import is idempotent 7. Author and reconciler profiles 8. Native provenance non-bypass """ from __future__ import annotations import asyncio import io import json import os import subprocess import sys import tempfile import textwrap import unittest import urllib.error from unittest.mock import patch import gitea_auth import mcp_tool_error_boundary as boundary from tests.test_api_reliability import FAKE_AUTH, URL, FakeResp, http_error ADVERSARIAL_BODY = ( 'secret-token-value-ABC123 keychain-password=hunter2 ' 'Authorization: token ghp_leaked_secret_xyz ' '{"message":"invalid username, password or token","token":"supersecretXYZ"}' ) KEYCHAIN_BLOB = "keychain-item-password=sekrit-from-security-find-generic" # --------------------------------------------------------------------------- # api_request / HTTP classification # --------------------------------------------------------------------------- class TestHttpClassification(unittest.TestCase): @patch("gitea_auth.urllib.request.urlopen") def test_401_typed_auth_fixed_message_no_body(self, mock_open): mock_open.side_effect = http_error(401, ADVERSARIAL_BODY) with self.assertRaises(gitea_auth.GiteaAuthError) as ctx: gitea_auth.api_request("GET", URL, FAKE_AUTH) exc = ctx.exception self.assertEqual(exc.reason_code, "auth_invalid_token") self.assertEqual(exc.error_class, "authentication") self.assertEqual(exc.http_status, 401) msg = str(exc) self.assertNotIn("supersecretXYZ", msg) self.assertNotIn("ghp_leaked", msg) self.assertNotIn("hunter2", msg) self.assertNotIn(ADVERSARIAL_BODY, msg) self.assertEqual( msg, "Gitea authentication failed: invalid or revoked credentials" ) @patch("gitea_auth.urllib.request.urlopen") def test_403_scope_is_authz_scope(self, mock_open): mock_open.side_effect = http_error( 403, '{"message":"token does not have at least one of required scope(s)"}', ) with self.assertRaises(gitea_auth.GiteaAuthzError) as ctx: gitea_auth.api_request("GET", URL, FAKE_AUTH) self.assertEqual(ctx.exception.reason_code, "authz_insufficient_scope") self.assertEqual(ctx.exception.error_class, "authorization") self.assertNotIn("token does not have", str(ctx.exception)) @patch("gitea_auth.urllib.request.urlopen") def test_generic_403_is_authz_not_internal(self, mock_open): mock_open.side_effect = http_error(403, '{"message":"user has no permission"}') with self.assertRaises(gitea_auth.GiteaAuthzError) as ctx: gitea_auth.api_request("GET", URL, FAKE_AUTH) self.assertEqual(ctx.exception.reason_code, "authz_denied") self.assertEqual(ctx.exception.error_class, "authorization") self.assertNotIsInstance(ctx.exception, gitea_auth.GiteaAuthError) self.assertNotIn("user has no permission", str(ctx.exception)) @patch("gitea_auth.urllib.request.urlopen") def test_network_fixed_message(self, mock_open): mock_open.side_effect = TimeoutError("timed out contacting secret.example") with self.assertRaises(gitea_auth.GiteaNetworkError) as ctx: gitea_auth.api_request("GET", URL, FAKE_AUTH) self.assertEqual(str(ctx.exception), "Network error contacting Gitea") self.assertNotIn("secret.example", str(ctx.exception)) @patch("gitea_auth.urllib.request.urlopen") def test_malformed_json_not_auth(self, mock_open): mock_open.return_value = FakeResp("not-json{") with self.assertRaises(RuntimeError) as ctx: gitea_auth.api_request("GET", URL, FAKE_AUTH) self.assertNotIsInstance(ctx.exception, gitea_auth.GiteaAuthError) self.assertIn("malformed JSON", str(ctx.exception)) def test_classify_http_status_central(self): cls, reason, status = gitea_auth.classify_http_status(401) self.assertIs(cls, gitea_auth.GiteaAuthError) self.assertEqual(reason, "auth_invalid_token") self.assertEqual(status, 401) cls, reason, status = gitea_auth.classify_http_status(403) self.assertIs(cls, gitea_auth.GiteaAuthzError) self.assertEqual(reason, "authz_denied") cls, reason, status = gitea_auth.classify_http_status( 403, body_hint="missing scope write:issue" ) self.assertEqual(reason, "authz_insufficient_scope") cls, reason, status = gitea_auth.classify_http_status(502) self.assertIs(cls, gitea_auth.GiteaHttpError) self.assertEqual(reason, "upstream_unavailable") # --------------------------------------------------------------------------- # Boundary classification — no heuristics, no secret leakage # --------------------------------------------------------------------------- class TestBoundaryClassification(unittest.TestCase): def test_auth_payload_fixed_message(self): exc = gitea_auth.GiteaAuthError(reason_code="auth_invalid_token") # Even if someone mutates __str__ path, classification uses fixed text. result = boundary.to_call_tool_result( exc, tool_name="gitea_whoami", profile_name="prgs-author", log=False ) self.assertTrue(result.isError) p = result.structuredContent self.assertEqual(p["reason_code"], "auth_invalid_token") self.assertEqual(p["error_class"], "authentication") self.assertEqual(p["message"], boundary.fixed_message("auth_invalid_token")) self.assertNotIn("supersecret", json.dumps(p)) self.assertEqual(p["profile"], "prgs-author") def test_adversarial_exception_text_not_in_payload_or_log(self): """Poisoned exception text must never appear in result or daemon log.""" class PoisonedAuth(gitea_auth.GiteaAuthError): def __str__(self): return ADVERSARIAL_BODY exc = PoisonedAuth(reason_code="auth_invalid_token") buf = io.StringIO() result = boundary.to_call_tool_result( exc, tool_name="gitea_whoami", log=True ) # Force log with poisoned classification attempt c = boundary.classify_exception(exc) boundary.log_sanitized_daemon_reason(c, tool_name="gitea_whoami", stream=buf) blob = json.dumps(result.structuredContent) + result.content[0].text + buf.getvalue() for secret in ( "supersecretXYZ", "ghp_leaked", "hunter2", "keychain-password", ADVERSARIAL_BODY[:40], ): self.assertNotIn(secret, blob) self.assertIn("reason_code=auth_invalid_token", buf.getvalue()) self.assertNotIn("detail=", buf.getvalue()) def test_keychain_content_not_in_daemon_log(self): buf = io.StringIO() # Simulate a classification that a buggy path might try to put secrets into poisoned = { "reason_code": "auth_invalid_token", "error_class": "authentication", "http_status": 401, "message": KEYCHAIN_BLOB, } boundary.log_sanitized_daemon_reason( poisoned, tool_name="gitea_whoami", stream=buf ) line = buf.getvalue() self.assertNotIn("sekrit", line) self.assertNotIn("keychain-item", line) self.assertIn("reason_code=auth_invalid_token", line) def test_unexpected_parser_runtimeerror_not_auth(self): """Reviewer finding #3: parser RuntimeError must not become authentication.""" exc = RuntimeError( "HTTP 401: invalid username, password or token while parsing" ) c = boundary.classify_exception(exc) self.assertEqual(c["error_class"], "internal") self.assertEqual(c["reason_code"], "internal_error") self.assertNotEqual(c["error_class"], "authentication") self.assertFalse(boundary.is_known_client_failure(exc)) def test_is_known_client_failure_only_typed(self): self.assertTrue( boundary.is_known_client_failure(gitea_auth.GiteaAuthError()) ) self.assertTrue( boundary.is_known_client_failure(gitea_auth.GiteaAuthzError()) ) self.assertFalse(boundary.is_known_client_failure(RuntimeError("x"))) self.assertFalse(boundary.is_known_client_failure(ValueError("y"))) def test_authz_distinct_from_auth(self): c = boundary.classify_exception( gitea_auth.GiteaAuthzError(reason_code="authz_denied") ) self.assertEqual(c["error_class"], "authorization") self.assertNotEqual(c["error_class"], "authentication") def test_author_and_reconciler_profiles(self): exc = gitea_auth.GiteaAuthError() for profile in ("prgs-author", "prgs-reconciler"): r = boundary.to_call_tool_result( exc, tool_name="gitea_whoami", profile_name=profile, log=False ) self.assertEqual(r.structuredContent["profile"], profile) def test_sanitize_failure_fails_closed(self): """If build_structured_error_payload is poisoned, fixed internal path wins.""" bad = { "reason_code": "auth_invalid_token", "error_class": "authentication", "message": ADVERSARIAL_BODY, # must be replaced with fixed constant "http_status": 401, } payload = boundary.build_structured_error_payload(bad) self.assertEqual( payload["message"], boundary.fixed_message("auth_invalid_token") ) self.assertNotIn("supersecret", payload["message"]) # --------------------------------------------------------------------------- # Tool.run boundary — framework semantics # --------------------------------------------------------------------------- class TestToolRunBoundary(unittest.TestCase): def setUp(self): from mcp.server.fastmcp.tools.base import Tool boundary.install_tool_run_boundary(Tool) self.Tool = Tool def _tool(self, fn, name="demo_tool"): return self.Tool.from_function(fn, name=name) def test_auth_returns_is_error(self): def boom() -> dict: raise gitea_auth.GiteaAuthError() result = asyncio.run(self._tool(boom, "gitea_whoami").run({}, convert_result=True)) self.assertTrue(result.isError) self.assertEqual(result.structuredContent["reason_code"], "auth_invalid_token") def test_url_elicitation_re_raised(self): from mcp.shared.exceptions import UrlElicitationRequiredError from mcp.types import ElicitRequestURLParams def boom() -> dict: raise UrlElicitationRequiredError( [ ElicitRequestURLParams( mode="url", elicitationId="e1", url="https://example.invalid/elicit", message="need auth", ) ] ) tool = self._tool(boom, "elicitation_tool") async def _run(): return await tool.run({}, convert_result=True) with self.assertRaises(UrlElicitationRequiredError): asyncio.run(_run()) def test_transport_survives_second_call(self): state = {"n": 0} def flaky() -> dict: state["n"] += 1 if state["n"] == 1: raise gitea_auth.GiteaAuthError() return {"ok": True, "call": state["n"]} tool = self._tool(flaky, "gitea_whoami") async def _both(): first = await tool.run({}, convert_result=True) second = await tool.run({}, convert_result=True) return first, second first, second = asyncio.run(_both()) self.assertTrue(first.isError) self.assertEqual(first.structuredContent["error_class"], "authentication") self.assertFalse(getattr(second, "isError", False)) self.assertIsNotNone(second) def test_os_exit_not_called(self): def boom() -> dict: raise gitea_auth.GiteaAuthError() with patch("os._exit") as mock_exit: result = asyncio.run(self._tool(boom).run({}, convert_result=True)) mock_exit.assert_not_called() self.assertTrue(result.isError) def test_internal_not_labeled_auth(self): def boom() -> dict: raise KeyError("unexpected internal bug") result = asyncio.run(self._tool(boom).run({}, convert_result=True)) self.assertTrue(result.isError) self.assertEqual(result.structuredContent["error_class"], "internal") self.assertEqual(result.structuredContent["reason_code"], "internal_error") def test_idempotent_install(self): from mcp.server.fastmcp.tools.base import Tool first = boundary.install_tool_run_boundary(Tool) second = boundary.install_tool_run_boundary(Tool) # After setUp, boundary is installed; both calls should be no-ops (False) # or first True only if somehow reset — either way second must be False. self.assertFalse(second) self.assertTrue(boundary.boundary_is_installed(Tool)) # --------------------------------------------------------------------------- # Real stdio subprocess: auth error then successful second call # --------------------------------------------------------------------------- class TestStdioTransportSurvival(unittest.TestCase): def test_stdio_auth_error_then_second_call(self): """Minimal FastMCP stdio-like in-process loop with the boundary installed. Uses Tool.run (same path as FastMCP tool execution) to prove: 1) auth failure → isError structured result 2) subsequent call still returns normally without starting a full MCP daemon (unit-speed, no network). """ from mcp.server.fastmcp.tools.base import Tool boundary.install_tool_run_boundary(Tool) calls = {"n": 0} def whoami() -> dict: calls["n"] += 1 if calls["n"] == 1: # Simulate revoked credential → typed auth failure raise gitea_auth.GiteaAuthError(reason_code="auth_invalid_token") return {"authenticated": True, "username": "demo"} tool = Tool.from_function(whoami, name="gitea_whoami") async def session(): r1 = await tool.run({}, convert_result=True) r2 = await tool.run({}, convert_result=True) return r1, r2 r1, r2 = asyncio.run(session()) self.assertTrue(r1.isError) self.assertEqual(r1.structuredContent["reason_code"], "auth_invalid_token") self.assertTrue(r1.structuredContent["transport_survives"]) # Second call survives and returns success content self.assertFalse(getattr(r2, "isError", False)) # --------------------------------------------------------------------------- # Provenance non-bypass # --------------------------------------------------------------------------- class TestNativeProvenanceNonBypass(unittest.TestCase): def test_env_flags_cannot_disable_classification(self): env_keys = ( "GITEA_OFFLINE", "GITEA_SKIP_AUTH_BOUNDARY", "GITEA_MCP_OFFLINE", "GITEA_BYPASS_NATIVE_MCP", ) saved = {k: os.environ.get(k) for k in env_keys} try: for k in env_keys: os.environ[k] = "1" exc = gitea_auth.GiteaAuthError() c = boundary.classify_exception(exc) self.assertEqual(c["error_class"], "authentication") r = boundary.to_call_tool_result(exc, log=False) self.assertTrue(r.isError) self.assertEqual(r.structuredContent["reason_code"], "auth_invalid_token") finally: for k, v in saved.items(): if v is None: os.environ.pop(k, None) else: os.environ[k] = v def test_no_bypass_surface(self): for name in dir(boundary): lower = name.lower() self.assertFalse( lower.startswith("bypass") or lower.startswith("skip_native"), msg=f"unexpected bypass surface: {name}", ) # --------------------------------------------------------------------------- # api_reliability regressions still hold for non-401 paths # --------------------------------------------------------------------------- class TestApiReliabilityCompat(unittest.TestCase): @patch("gitea_auth.urllib.request.urlopen") def test_502_upstream_typed(self, mock_open): mock_open.side_effect = http_error(502, "bad gateway secret=xyz") with self.assertRaises(gitea_auth.GiteaHttpError) as ctx: gitea_auth.api_request("GET", URL, FAKE_AUTH) self.assertEqual(ctx.exception.reason_code, "upstream_unavailable") self.assertNotIn("secret=xyz", str(ctx.exception)) @patch("gitea_auth.urllib.request.urlopen") def test_auth_header_never_in_error(self, mock_open): mock_open.side_effect = http_error(400, "bad request") with self.assertRaises(gitea_auth.GiteaHttpError) as ctx: gitea_auth.api_request("GET", URL, FAKE_AUTH) self.assertNotIn(FAKE_AUTH, str(ctx.exception)) if __name__ == "__main__": unittest.main()