"""Safe mutation-error diagnostics for the tool boundary. The #699/#701 boundary intentionally collapses every non-typed exception to an opaque ``internal_error`` with the class and message discarded. That makes a mutation failure (create_issue / create_pr / delete_branch / issue-lock) undiagnosable: reason_code=internal_error, retryable=false, no actionable detail. This suite pins the additive contract: 1. For the ``internal_error`` fallback ONLY, capture a **safe exception class** (a Python type identifier — never secrets) and a **strictly-redacted** detail (no tokens, Authorization headers, raw URLs/hosts, absolute paths, response bodies, env dumps), plus an optional ``mutation_stage`` tag. 2. Typed classifications (auth/authz/network/config/http) are unchanged and still emit fixed messages with NO detail in the daemon log. 3. Any exception may self-declare a safe ``gitea_reason_code`` to be converted into a typed structured error instead of an opaque internal_error. """ from __future__ import annotations import io import json import unittest import gitea_auth import mcp_tool_error_boundary as boundary # Adversarial text a poisoned internal exception might carry. POISON = ( "boom at /Users/jasonwalker/secret/creds.json " "Authorization: token ghp_leaked_secret_xyz " "https://sysadmin:hunter2@gitea.prgs.cc/api/v1/x?token=supersecretXYZ " 'Bearer abc.def.ghi {"token":"supersecretXYZ","password":"hunter2"}' ) SECRETS = ( "ghp_leaked_secret_xyz", "supersecretXYZ", "hunter2", "/Users/jasonwalker/secret", "gitea.prgs.cc", ) class TestRedactDetail(unittest.TestCase): def test_strips_secrets_paths_urls_and_truncates(self): out = boundary._redact_detail(POISON) for s in SECRETS: self.assertNotIn(s, out) self.assertLessEqual(len(out), boundary._DETAIL_LIMIT) def test_empty_is_safe(self): self.assertEqual(boundary._redact_detail(""), "") self.assertEqual(boundary._redact_detail(None), "") class TestSafeExceptionClass(unittest.TestCase): def test_fully_qualified_type_identifier(self): self.assertEqual( boundary._safe_exception_class(RuntimeError("x")), "builtins.RuntimeError" ) self.assertEqual( boundary._safe_exception_class(KeyError("x")), "builtins.KeyError" ) def test_class_name_never_carries_instance_text(self): # The class identifier must not include the (possibly secret) message. cls = boundary._safe_exception_class(RuntimeError(POISON)) for s in SECRETS: self.assertNotIn(s, cls) class TestInternalErrorGainsDiagnostics(unittest.TestCase): def test_classify_internal_has_class_and_redacted_detail(self): c = boundary.classify_exception(RuntimeError(POISON)) self.assertEqual(c["reason_code"], "internal_error") self.assertEqual(c["error_class"], "internal") self.assertEqual(c["exception_class"], "builtins.RuntimeError") self.assertIn("detail", c) for s in SECRETS: self.assertNotIn(s, json.dumps(c)) def test_payload_surfaces_class_and_detail_but_fixed_message(self): p = boundary.build_structured_error_payload( boundary.classify_exception(RuntimeError(POISON)), tool_name="gitea_delete_branch", ) self.assertEqual(p["reason_code"], "internal_error") self.assertEqual(p["exception_class"], "builtins.RuntimeError") self.assertIn("detail", p) self.assertEqual(p["message"], boundary.fixed_message("internal_error")) for s in SECRETS: self.assertNotIn(s, json.dumps(p)) def test_daemon_log_internal_includes_class_and_detail_no_secrets(self): buf = io.StringIO() c = boundary.classify_exception(RuntimeError(POISON)) boundary.log_sanitized_daemon_reason( c, tool_name="gitea_delete_branch", stream=buf ) line = buf.getvalue() self.assertIn("reason_code=internal_error", line) self.assertIn("exception_class=builtins.RuntimeError", line) self.assertIn("detail=", line) for s in SECRETS: self.assertNotIn(s, line) class TestTypedPathsUnchanged(unittest.TestCase): def test_auth_payload_has_no_detail_or_class(self): p = boundary.build_structured_error_payload( boundary.classify_exception(gitea_auth.GiteaAuthError()) ) self.assertNotIn("detail", p) self.assertNotIn("exception_class", p) def test_auth_daemon_log_has_no_detail(self): buf = io.StringIO() boundary.log_sanitized_daemon_reason( boundary.classify_exception(gitea_auth.GiteaAuthError()), tool_name="gitea_whoami", stream=buf, ) self.assertNotIn("detail=", buf.getvalue()) self.assertNotIn("exception_class=", buf.getvalue()) class TestMutationStageTag(unittest.TestCase): def test_stage_flows_to_payload_and_log(self): exc = RuntimeError("boom") setattr(exc, "_gitea_mutation_stage", "preflight_purity") c = boundary.classify_exception(exc) self.assertEqual(c["mutation_stage"], "preflight_purity") p = boundary.build_structured_error_payload(c, tool_name="gitea_create_pr") self.assertEqual(p["mutation_stage"], "preflight_purity") buf = io.StringIO() boundary.log_sanitized_daemon_reason(c, tool_name="gitea_create_pr", stream=buf) self.assertIn("mutation_stage=preflight_purity", buf.getvalue()) def test_stage_value_is_bounded_and_safe(self): exc = RuntimeError("boom") setattr(exc, "_gitea_mutation_stage", "x" * 500 + POISON) c = boundary.classify_exception(exc) for s in SECRETS: self.assertNotIn(s, c["mutation_stage"]) self.assertLessEqual(len(c["mutation_stage"]), 64) class TestGiteaReasonCodeConversion(unittest.TestCase): def test_self_declared_reason_becomes_typed(self): exc = RuntimeError("Pre-flight order violation: whoami missing") setattr(exc, "gitea_reason_code", "preflight_order_violation") c = boundary.classify_exception(exc) self.assertEqual(c["reason_code"], "preflight_order_violation") self.assertNotEqual(c["reason_code"], "internal_error") p = boundary.build_structured_error_payload(c, tool_name="gitea_delete_branch") self.assertEqual(p["reason_code"], "preflight_order_violation") self.assertEqual( p["message"], boundary.fixed_message("preflight_order_violation") ) def test_unknown_self_declared_reason_ignored(self): exc = RuntimeError("boom") setattr(exc, "gitea_reason_code", "not_a_real_reason") c = boundary.classify_exception(exc) self.assertEqual(c["reason_code"], "internal_error") def test_preflight_reason_is_registered(self): self.assertIn("preflight_order_violation", boundary.FIXED_MESSAGES) class TestAdversarialInternalNoLeak(unittest.TestCase): def test_poisoned_internal_never_leaks_in_result_or_log(self): exc = RuntimeError(POISON) buf = io.StringIO() result = boundary.to_call_tool_result( exc, tool_name="gitea_delete_branch", log=False ) c = boundary.classify_exception(exc) boundary.log_sanitized_daemon_reason( c, tool_name="gitea_delete_branch", stream=buf ) blob = ( json.dumps(result.structuredContent) + result.content[0].text + buf.getvalue() ) for s in SECRETS: self.assertNotIn(s, blob) class TestServerSideTypedAndStage(unittest.TestCase): """Server raisers are typed and stage-taggable, end-to-end into the boundary.""" def test_preflight_order_error_is_typed_runtimeerror(self): import gitea_mcp_server as s exc = s._PreflightOrderError("whoami missing") self.assertIsInstance(exc, RuntimeError) # legacy except RuntimeError self.assertEqual(exc.gitea_reason_code, "preflight_order_violation") c = boundary.classify_exception(exc) self.assertEqual(c["reason_code"], "preflight_order_violation") self.assertNotEqual(c["reason_code"], "internal_error") def test_mutation_stage_tags_and_reraises(self): import gitea_mcp_server as s with self.assertRaises(RuntimeError) as ctx: with s._mutation_stage("preflight_purity"): raise RuntimeError("boom") self.assertEqual( getattr(ctx.exception, "_gitea_mutation_stage", None), "preflight_purity" ) def test_mutation_stage_inner_wins(self): import gitea_mcp_server as s with self.assertRaises(RuntimeError) as ctx: with s._mutation_stage("outer"): with s._mutation_stage("inner"): raise RuntimeError("boom") self.assertEqual( getattr(ctx.exception, "_gitea_mutation_stage", None), "inner" ) if __name__ == "__main__": unittest.main()