mirror of
https://github.com/pomerium/pomerium.git
synced 2025-08-06 10:21:05 +02:00
a577f53ff4
## Summary ## Related issues Fix https://linear.app/pomerium/issue/ENG-2543/mcp-tool-ignore-non-jsonrpc-calls ## User Explanation <!-- How would you explain this change to the user? If this change doesn't create any user-facing changes, you can leave this blank. If filled out, add the `docs` label --> ## Checklist - [x] reference any related issues - [x] updated unit tests - [x] add appropriate label (`enhancement`, `bug`, `breaking`, `dependencies`, `ci`) - [x] ready for review
82 lines
2.4 KiB
Go
82 lines
2.4 KiB
Go
package criteria
|
|
|
|
import (
|
|
"testing"
|
|
|
|
"github.com/stretchr/testify/require"
|
|
|
|
"github.com/pomerium/pomerium/pkg/grpc/databroker"
|
|
)
|
|
|
|
func TestMCPTool(t *testing.T) {
|
|
t.Run("ok", func(t *testing.T) {
|
|
res, err := evaluate(t, `
|
|
allow:
|
|
and:
|
|
- mcp_tool:
|
|
is: list_tables
|
|
`, []*databroker.Record{}, Input{MCP: InputMCP{Method: "tools/call", ToolCall: &InputMCPToolCall{Name: "list_tables"}}})
|
|
require.NoError(t, err)
|
|
require.Equal(t, A{true, A{ReasonMCPToolOK}, M{}}, res["allow"])
|
|
require.Equal(t, A{false, A{}}, res["deny"])
|
|
})
|
|
|
|
t.Run("unauthorized", func(t *testing.T) {
|
|
res, err := evaluate(t, `
|
|
allow:
|
|
and:
|
|
- mcp_tool:
|
|
is: list_tables
|
|
`, []*databroker.Record{}, Input{MCP: InputMCP{Method: "tools/call", ToolCall: &InputMCPToolCall{Name: "read_table"}}})
|
|
require.NoError(t, err)
|
|
require.Equal(t, A{false, A{ReasonMCPToolUnauthorized}, M{}}, res["allow"])
|
|
require.Equal(t, A{false, A{}}, res["deny"])
|
|
})
|
|
|
|
t.Run("in list", func(t *testing.T) {
|
|
res, err := evaluate(t, `
|
|
allow:
|
|
and:
|
|
- mcp_tool:
|
|
in: ["list_tables", "read_table"]
|
|
`, []*databroker.Record{}, Input{MCP: InputMCP{Method: "tools/call", ToolCall: &InputMCPToolCall{Name: "list_tables"}}})
|
|
require.NoError(t, err)
|
|
require.Equal(t, A{true, A{ReasonMCPToolOK}, M{}}, res["allow"])
|
|
require.Equal(t, A{false, A{}}, res["deny"])
|
|
})
|
|
|
|
t.Run("not in list", func(t *testing.T) {
|
|
res, err := evaluate(t, `
|
|
allow:
|
|
and:
|
|
- mcp_tool:
|
|
in: ["list_tables", "read_table"]
|
|
`, []*databroker.Record{}, Input{MCP: InputMCP{Method: "tools/call", ToolCall: &InputMCPToolCall{Name: "delete_table"}}})
|
|
require.NoError(t, err)
|
|
require.Equal(t, A{false, A{ReasonMCPToolUnauthorized}, M{}}, res["allow"])
|
|
require.Equal(t, A{false, A{}}, res["deny"])
|
|
})
|
|
|
|
t.Run("non-tools/call method should pass", func(t *testing.T) {
|
|
res, err := evaluate(t, `
|
|
allow:
|
|
and:
|
|
- mcp_tool:
|
|
is: list_tables
|
|
`, []*databroker.Record{}, Input{MCP: InputMCP{Method: "some/other_method"}})
|
|
require.NoError(t, err)
|
|
require.Equal(t, A{true, A{ReasonMCPNotAToolCall}, M{}}, res["allow"])
|
|
require.Equal(t, A{false, A{}}, res["deny"])
|
|
})
|
|
t.Run("no method name should pass", func(t *testing.T) {
|
|
res, err := evaluate(t, `
|
|
allow:
|
|
and:
|
|
- mcp_tool:
|
|
is: list_tables
|
|
`, []*databroker.Record{}, Input{})
|
|
require.NoError(t, err)
|
|
require.Equal(t, A{true, A{ReasonMCPNotAToolCall}, M{}}, res["allow"])
|
|
require.Equal(t, A{false, A{}}, res["deny"])
|
|
})
|
|
}
|