Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions src/cmcp_gateway/catalog/loader.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
4 changes: 3 additions & 1 deletion src/cmcp_gateway/mcp/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down
10 changes: 10 additions & 0 deletions tests/unit/test_catalog.py
Original file line number Diff line number Diff line change
Expand Up @@ -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]))
28 changes: 28 additions & 0 deletions tests/unit/test_mcp_server_auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down
Loading