fix(mcp): surface safe diagnostics for opaque mutation internal_error

Mutation tools (create_issue, create_pr, delete_branch, issue-lock) returned reason_code=internal_error / retryable=false with no class, message, HTTP status, or actionable detail. The #699/#701 tool-error boundary intentionally collapses every non-typed exception to a fixed "Internal tool error" and drops the class and message, so a mutation failure is undiagnosable — and the #695 runtime guard blocks standalone reproduction. The true failing stage was therefore invisible by two independent mechanisms. Make the failure observable and actionable without weakening the secret-free contract: - Boundary (internal_error path ONLY): capture a safe exception class (a Python type identifier, never instance text) plus a strictly-redacted `detail` (token/Authorization/Bearer credentials, JSON/kv secret values incl. short passwords, raw URLs/hostnames, and absolute filesystem paths all stripped; whitespace collapsed; length-bounded) and an optional `mutation_stage`. Typed auth/authz/network/config/http paths are unchanged and still emit no detail. - Convert raw exceptions into typed structured errors via a safe `gitea_reason_code` attribute contract: server raisers may self-declare a known reason and the boundary emits it (fixed message) instead of an opaque internal_error. Pre-flight order violations now raise `_PreflightOrderError` (a RuntimeError subclass → preflight_order_violation). - Add a `_mutation_stage` context manager tagging escaping exceptions with the stage name (preflight_purity / resolve_auth / api_*), applied to delete_branch, create_issue, create_pr. Fail-closed identity/repo/worktree/role/permission/lock/lease/ancestry gates are untouched; only error *reporting* changed. New suite test_mutation_error_diagnostics.py (18 tests) pins the redaction, class capture, stage tagging, typed conversion, and adversarial no-leak contract. Existing boundary suite (25), core server (209), and preflight (92) suites stay green. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-17 20:42:25 -04:00
parent 39e27dc63f
commit 7e5eca08b3
3 changed files with 466 additions and 16 deletions
@@ -0,0 +1,228 @@
+"""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:[email protected]/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()