From a2673f0839cddc23fd105da504f93c09f8ab52d8 Mon Sep 17 00:00:00 2001
From: Imran Siddique <imran.siddique@opaque.co>
Date: Sat, 6 Jun 2026 16:19:49 -0700
Subject: [PATCH] fix(server,catalog): canonicalize tool names at ingress;
 enforce lowercase in catalog (POLICY-002)

Tool names are now lowercased in _handle_tool_call() before reaching
the proxy, so Cedar policy, catalog, and request always use the same
case. Catalog loading now rejects any entry whose tool_name is not
already lowercase, preventing inconsistency that could allow a
case-variant to bypass deny rules.

Closes #159

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
---
 src/cmcp_gateway/catalog/loader.py |  7 +++++++
 src/cmcp_gateway/mcp/server.py     |  4 +++-
 tests/unit/test_catalog.py         | 10 ++++++++++
 tests/unit/test_mcp_server_auth.py | 28 ++++++++++++++++++++++++++++
 4 files changed, 48 insertions(+), 1 deletion(-)

diff --git a/src/cmcp_gateway/catalog/loader.py b/src/cmcp_gateway/catalog/loader.py
index c11391cd..916ce825 100644
--- a/src/cmcp_gateway/catalog/loader.py
+++ b/src/cmcp_gateway/catalog/loader.py
@@ -127,6 +127,13 @@ def load_catalog(catalog_path: str, expected_hash: str | None = None) -> ToolCat
         _validate_entry(raw, entry_schema)
 
         tool_name: str = raw["tool_name"]
+        # POLICY-002: enforce lowercase-only tool names so ingress canonicalization
+        # (server.py .lower()) never silently renames a correctly-spelled tool.
+        if tool_name != tool_name.lower():
+            raise ConfigError(
+                f"Catalog entry tool_name '{tool_name}' must be lowercase; "
+                "use lowercase in catalog and Cedar policy documents"
+            )
         if tool_name in entries:
             raise CatalogToolNameCollision(
                 f"Duplicate tool_name '{tool_name}' — gateway will not start",
diff --git a/src/cmcp_gateway/mcp/server.py b/src/cmcp_gateway/mcp/server.py
index fc831b40..4509037a 100644
--- a/src/cmcp_gateway/mcp/server.py
+++ b/src/cmcp_gateway/mcp/server.py
@@ -214,7 +214,9 @@ async def _handle_mcp(self, request: Request) -> Response:
 
     async def _handle_tool_call(self, rpc_id: Any, params: dict[str, Any]) -> Response:
         """Route a tools/call request through the proxy."""
-        tool_name: str = params.get("name", "")
+        # POLICY-002: canonicalize tool names at ingress so Cedar policy, catalog, and
+        # request all use the same case — prevents case-variant bypass of deny rules.
+        tool_name: str = params.get("name", "").lower()
         arguments: dict[str, Any] = params.get("arguments", {})
         call_id = str(uuid.uuid4())
         workflow_id: str | None = params.get("_cmcp", {}).get("workflow_id")
diff --git a/tests/unit/test_catalog.py b/tests/unit/test_catalog.py
index 6bc96601..2d6ecff6 100644
--- a/tests/unit/test_catalog.py
+++ b/tests/unit/test_catalog.py
@@ -150,3 +150,13 @@ def test_catalog_hash_changes_when_entry_changes(catalog_file):
     modified["approved_by"] = "changed@example.com"
     c2 = load_catalog(catalog_file([modified]))
     assert c1.catalog_hash != c2.catalog_hash
+
+
+# ── POLICY-002: tool name must be lowercase ───────────────────────────────────
+
+def test_uppercase_tool_name_is_rejected(catalog_file):
+    """POLICY-002 — mixed-case tool names must be rejected at load time."""
+    entry = dict(ENTRY_1)
+    entry["tool_name"] = "CRM.Query"
+    with pytest.raises(ConfigError, match="lowercase"):
+        load_catalog(catalog_file([entry]))
diff --git a/tests/unit/test_mcp_server_auth.py b/tests/unit/test_mcp_server_auth.py
index f65653da..77e4033a 100644
--- a/tests/unit/test_mcp_server_auth.py
+++ b/tests/unit/test_mcp_server_auth.py
@@ -178,6 +178,34 @@ def test_unhandled_exception_returns_generic_500():
     assert body.get("error_code") == "INTERNAL_ERROR"
 
 
+# ── POLICY-002: ingress tool name canonicalized to lowercase ─────────────────
+
+def test_tool_name_is_lowercased_at_ingress():
+    """POLICY-002 — tool name from MCP request must be lowercased before catalog lookup."""
+    received_names: list[str] = []
+
+    async def _capture(call_id, tool_name, arguments, **kwargs):
+        received_names.append(tool_name)
+        return MagicMock(
+            allowed=True, deny_reason=None, response="ok",
+            audit_entry_hash="sha256:" + "0" * 64,
+            would_have_denied=False, latency_us=100,
+        )
+
+    proxy = MagicMock()
+    proxy._catalog = MagicMock()
+    proxy._catalog.entries = {}
+    proxy.call_tool = _capture
+    with patch("cmcp_gateway.mcp.server.StatelessKernel"):
+        server = MCPServer(proxy)
+    client = TestClient(server.app, raise_server_exceptions=False)
+    client.post(
+        "/mcp",
+        json={"jsonrpc": "2.0", "method": "tools/call", "params": {"name": "UPPER_TOOL", "arguments": {}}, "id": 1},
+    )
+    assert received_names == ["upper_tool"]
+
+
 def test_deny_response_does_not_include_internal_reason():
     """INJECT-003 — internal deny_reason must not appear in 403 response body."""
     proxy = MagicMock()