This is an automated email from the ASF dual-hosted git repository. hsluoyz pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/casbin-gateway.git
commit b7a13cb7ccc8b915543999a52831e288bc3625ef Author: Yang Luo <[email protected]> AuthorDate: Sat Sep 5 11:52:00 2026 +0800 feat(connections): keep a connection's credential out of the agent's config --- agentpatch/endpoint.go | 12 +++ controllers/connector.go | 55 ++++++++++++ main.go | 2 + mcpproxy/gateway.go | 165 ++++++++++++++++++++++++++++++++++++ mcpproxy/http.go | 137 ++++++++++++++++++++++++++++++ mcpproxy/mcpproxy.go | 194 +++++++++++++++++++++++++++++++++++++++++++ mcpproxy/upstream.go | 115 +++++++++++++++++++++++++ object/connection_install.go | 105 +++++++++++++++++++---- routers/router.go | 1 + 9 files changed, 769 insertions(+), 17 deletions(-) diff --git a/agentpatch/endpoint.go b/agentpatch/endpoint.go index ce9ffe9..4f84841 100644 --- a/agentpatch/endpoint.go +++ b/agentpatch/endpoint.go @@ -50,3 +50,15 @@ func loopbackURL(path string) (string, error) { } return fmt.Sprintf("http://127.0.0.1:%d%s";, number, path), nil } + +// GatewayExecutable is this Gateway's own program, which an agent is given as +// the command to run for a proxied MCP connection. +func GatewayExecutable() (string, error) { + return gatewayExecutable() +} + +// GatewayBaseUrl is the loopback address of the Gateway on this machine, for a +// child process that has to ask it something. +func GatewayBaseUrl() (string, error) { + return loopbackURL("") +} diff --git a/controllers/connector.go b/controllers/connector.go index 218638e..dc496d4 100644 --- a/controllers/connector.go +++ b/controllers/connector.go @@ -16,7 +16,12 @@ package controllers import ( "encoding/json" + "slices" "sort" + "strings" + + "github.com/apache/casbin-gateway/agentmonitor" + "github.com/apache/casbin-gateway/agentpatch" "github.com/apache/casbin-gateway/agent" "github.com/apache/casbin-gateway/agentconfig" @@ -203,3 +208,53 @@ func (c *ApiController) Disconnect() { } c.ResponseOk(planned) } + +// ResolveConnection hands the proxy the server behind one connection, with its +// credentials filled in. This is the one place a stored credential leaves +// Gateway, so it is fenced the way tool checks are: a direct loopback request, +// presenting the installation credential issued when the connection was +// written, and only for the agent that credential names. +func (c *ApiController) ResolveConnection() { + if _, ok := c.directLoopbackClient(); !ok { + c.ResponseError("resolving a connection is limited to direct loopback requests") + return + } + tokenAgentId, ok := agentpatch.ValidateIngestToken(c.Ctx.Input.Header(agentmonitor.IngestTokenHeader)) + if !ok { + c.ResponseError("resolving a connection requires a valid installation token") + return + } + + var form struct { + Owner string `json:"owner"` + Name string `json:"name"` + Agent string `json:"agent"` + } + if err := json.Unmarshal(c.Ctx.Input.RequestBody, &form); err != nil { + c.ResponseError(err.Error()) + return + } + if tokenAgentId != "" && !strings.EqualFold(form.Agent, tokenAgentId) { + c.ResponseError("agent does not match the installation this token was issued for") + return + } + + // A connection this agent was never given is not one it may resolve, or one + // agent's token would open every connection on the machine. + connection, err := object.GetConnection(form.Owner, form.Name) + if err != nil { + c.ResponseError(err.Error()) + return + } + if connection == nil || !slices.Contains(connection.Agents, form.Agent) { + c.ResponseError("this connection is not installed in " + form.Agent) + return + } + + rendered, err := object.ResolveConnection(form.Owner, form.Name) + if err != nil { + c.ResponseError(err.Error()) + return + } + c.ResponseOk(rendered) +} diff --git a/main.go b/main.go index 1d4c6a7..f56cc1f 100644 --- a/main.go +++ b/main.go @@ -27,6 +27,7 @@ import ( "github.com/apache/casbin-gateway/casdoor" "github.com/apache/casbin-gateway/conf" "github.com/apache/casbin-gateway/controllers" + "github.com/apache/casbin-gateway/mcpproxy" "github.com/apache/casbin-gateway/mcpserver" "github.com/apache/casbin-gateway/object" "github.com/apache/casbin-gateway/proxy" @@ -46,6 +47,7 @@ func main() { // process. They must exit before Gateway initializes its own services. agenthook.ServeIfInvoked() mcpserver.ServeIfInvoked() + mcpproxy.ServeIfInvoked() // A link in the URL scheme of an agent opens this executable while Gateway // is routing one to a particular copy. It is a launch, not a service. diff --git a/mcpproxy/gateway.go b/mcpproxy/gateway.go new file mode 100644 index 0000000..1930d4f --- /dev/null +++ b/mcpproxy/gateway.go @@ -0,0 +1,165 @@ +// Copyright 2026 The casbin Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package mcpproxy + +import ( + "bytes" + "encoding/json" + "fmt" + "net/http" + "net/url" + "time" + + "github.com/apache/casbin-gateway/agentmonitor" +) + +const ( + resolveTimeout = 10 * time.Second + // A tool call waits on the verdict, so a Gateway that is slow to answer is + // treated like one that is not running: the call goes ahead. Holding every + // session on this machine would be the worse failure. + decideTimeout = 2 * time.Second +) + +// resolvedServer is one connection's MCP server with its credentials filled in, +// as Gateway rendered it. It exists only for the life of this process and is +// never written anywhere. +type resolvedServer struct { + Name string `json:"name"` + Transport string `json:"transport"` + Command string `json:"command"` + Args []string `json:"args"` + Env map[string]string `json:"env"` + Url string `json:"url"` + Headers map[string]string `json:"headers"` +} + +func (r *resolvedServer) envPairs() []string { + pairs := make([]string, 0, len(r.Env)) + for name, value := range r.Env { + pairs = append(pairs, name+"="+value) + } + return pairs +} + +// gatewayClient is this process's line to the Gateway that launched it. +type gatewayClient struct { + options Options + client *http.Client +} + +func newGatewayClient(options Options) *gatewayClient { + return &gatewayClient{options: options, client: &http.Client{Timeout: resolveTimeout}} +} + +type apiResponse struct { + Status string `json:"status"` + Msg string `json:"msg"` + Data json.RawMessage `json:"data"` +} + +func (g *gatewayClient) post(path string, body any, timeout time.Duration) (json.RawMessage, error) { + if g.options.GatewayUrl == "" { + return nil, fmt.Errorf("no Gateway address was given") + } + encoded, err := json.Marshal(body) + if err != nil { + return nil, err + } + + request, err := http.NewRequest(http.MethodPost, g.options.GatewayUrl+path, bytes.NewReader(encoded)) + if err != nil { + return nil, err + } + request.Header.Set("Content-Type", "application/json") + request.Header.Set(agentmonitor.IngestTokenHeader, g.options.Token) + + client := g.client + if timeout != 0 && timeout != client.Timeout { + client = &http.Client{Timeout: timeout} + } + response, err := client.Do(request) + if err != nil { + return nil, err + } + defer response.Body.Close() + + var decoded apiResponse + if err := json.NewDecoder(response.Body).Decode(&decoded); err != nil { + return nil, err + } + if decoded.Status != "ok" { + return nil, fmt.Errorf("%s", decoded.Msg) + } + return decoded.Data, nil +} + +// resolve asks Gateway for the server behind this connection. This is the only +// place the credential exists outside Gateway, and it exists in memory of a +// process the agent started and will end. +func (g *gatewayClient) resolve() (*resolvedServer, error) { + data, err := g.post("/api/resolve-connection", map[string]string{ + "owner": g.options.Owner, + "name": g.options.Connection, + "agent": g.options.AgentId, + }, resolveTimeout) + if err != nil { + return nil, err + } + + resolved := &resolvedServer{} + if err := json.Unmarshal(data, resolved); err != nil { + return nil, err + } + return resolved, nil +} + +// allowTool asks whether this agent may make one call. A verdict that cannot be +// had allows the call, matching what every hook does: Gateway being down or slow +// must not be the thing that breaks somebody's session. +func (g *gatewayClient) allowTool(tool string) (bool, string) { + if tool == "" || g.options.AgentId == "" || g.options.Token == "" { + return true, "" + } + + data, err := g.post("/api/check-agent-tool", map[string]string{ + "agent": g.options.AgentId, + "tool": toolKey(g.options.Connection, tool), + }, decideTimeout) + if err != nil { + return true, "" + } + + var verdict struct { + Allow bool `json:"allow"` + Reason string `json:"reason"` + } + if json.Unmarshal(data, &verdict) != nil { + return true, "" + } + return verdict.Allow, verdict.Reason +} + +// toolKey names a call the way the permission switches do, so a connection's +// tools land on the switch that agent already has for this MCP server. +func toolKey(connection string, tool string) string { + return "mcp__" + connection + "__" + tool +} + +// GatewayUrl is the address a patched agent is given for the Gateway on this +// machine. +func GatewayUrl(port string) string { + return (&url.URL{Scheme: "http", Host: "127.0.0.1:" + port}).String() +} diff --git a/mcpproxy/http.go b/mcpproxy/http.go new file mode 100644 index 0000000..eb0330c --- /dev/null +++ b/mcpproxy/http.go @@ -0,0 +1,137 @@ +// Copyright 2026 The casbin Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package mcpproxy + +import ( + "bufio" + "bytes" + "fmt" + "io" + "net/http" + "strings" + "sync" + "time" +) + +// sessionHeader is what a Streamable HTTP server names a session with. It comes +// back from initialize and has to be sent on everything after it, or the server +// treats each request as a new session. +const sessionHeader = "Mcp-Session-Id" + +const httpTimeout = 5 * time.Minute + +// httpUpstream reaches a remote MCP server over Streamable HTTP. One client +// message is one POST, and the reply is either a JSON body or an event stream +// carrying several messages. +type httpUpstream struct { + resolved *resolvedServer + client *http.Client + emit func([]byte) error + + mutex sync.Mutex + session string +} + +func newHttpUpstream(resolved *resolvedServer) *httpUpstream { + return &httpUpstream{resolved: resolved, client: &http.Client{Timeout: httpTimeout}} +} + +func (u *httpUpstream) Start(emit func([]byte) error) error { + u.emit = emit + return nil +} + +func (u *httpUpstream) Close() error { return nil } + +func (u *httpUpstream) Send(line []byte) error { + request, err := http.NewRequest(http.MethodPost, u.resolved.Url, bytes.NewReader(line)) + if err != nil { + return err + } + request.Header.Set("Content-Type", "application/json") + request.Header.Set("Accept", "application/json, text/event-stream") + for name, value := range u.resolved.Headers { + request.Header.Set(name, value) + } + if session := u.currentSession(); session != "" { + request.Header.Set(sessionHeader, session) + } + + response, err := u.client.Do(request) + if err != nil { + return err + } + defer response.Body.Close() + + if session := response.Header.Get(sessionHeader); session != "" { + u.rememberSession(session) + } + // A notification is answered with 202 and no body, which is not something + // to pass on. + if response.StatusCode == http.StatusAccepted || response.StatusCode == http.StatusNoContent { + return nil + } + if response.StatusCode >= 400 { + body, _ := io.ReadAll(io.LimitReader(response.Body, 4*1024)) + return fmt.Errorf("%s answered %s: %s", u.resolved.Url, response.Status, strings.TrimSpace(string(body))) + } + + if strings.Contains(response.Header.Get("Content-Type"), "text/event-stream") { + return u.readEvents(response.Body) + } + + body, err := io.ReadAll(response.Body) + if err != nil { + return err + } + if len(bytes.TrimSpace(body)) == 0 { + return nil + } + return u.emit(body) +} + +// readEvents forwards every message an event stream carries. Only the data +// field matters here: the rest of the SSE framing is transport, and what an MCP +// client is given back is the JSON-RPC message inside it. +func (u *httpUpstream) readEvents(body io.Reader) error { + scanner := bufio.NewScanner(body) + scanner.Buffer(make([]byte, 0, 64*1024), maxLineBytes) + for scanner.Scan() { + line := scanner.Text() + if !strings.HasPrefix(line, "data:") { + continue + } + data := strings.TrimSpace(strings.TrimPrefix(line, "data:")) + if data == "" { + continue + } + if err := u.emit([]byte(data)); err != nil { + return err + } + } + return scanner.Err() +} + +func (u *httpUpstream) currentSession() string { + u.mutex.Lock() + defer u.mutex.Unlock() + return u.session +} + +func (u *httpUpstream) rememberSession(session string) { + u.mutex.Lock() + defer u.mutex.Unlock() + u.session = session +} diff --git a/mcpproxy/mcpproxy.go b/mcpproxy/mcpproxy.go new file mode 100644 index 0000000..a49a35d --- /dev/null +++ b/mcpproxy/mcpproxy.go @@ -0,0 +1,194 @@ +// Copyright 2026 The casbin Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package mcpproxy stands between an agent and the MCP server one connection +// reaches. The agent is given this executable as the server to run, so what its +// configuration file holds is the name of a connection rather than the +// credential behind it: the credential stays in Gateway and is fetched here, +// over loopback, for the life of one session. +// +// Everything the agent sends passes through, so a tool call can be asked about +// before it runs and recorded after — which is what puts connector traffic +// under the same rules as the rest of an agent's tools. +package mcpproxy + +import ( + "bufio" + "bytes" + "encoding/json" + "flag" + "fmt" + "os" + "sync" +) + +// Subcommand is what an agent's configuration runs this executable with. +const Subcommand = "mcp-proxy" + +const maxLineBytes = 8 * 1024 * 1024 + +// ServeIfInvoked enters proxy mode when an agent launched Gateway as the MCP +// server for one connection. Like the recorder, it exits before Gateway +// initializes its own services so stdout stays a clean JSON-RPC stream. +func ServeIfInvoked() { + if len(os.Args) < 2 || os.Args[1] != Subcommand { + return + } + if err := Run(os.Args[2:]); err != nil { + fmt.Fprintf(os.Stderr, "casbin-gateway %s: %v\n", Subcommand, err) + os.Exit(1) + } + os.Exit(0) +} + +// Options is one proxied session. +type Options struct { + Connection string + AgentId string + Owner string + GatewayUrl string + Token string +} + +func Run(args []string) error { + flags := flag.NewFlagSet(Subcommand, flag.ContinueOnError) + flags.SetOutput(os.Stderr) + options := Options{} + flags.StringVar(&options.Connection, "connection", "", "the connection to reach") + flags.StringVar(&options.AgentId, "agent", "", "agent this server was written into") + flags.StringVar(&options.Owner, "owner", "", "the account the connection belongs to") + flags.StringVar(&options.GatewayUrl, "gateway-url", "", "the running Gateway") + flags.StringVar(&options.Token, "token", "", "credential presented to Gateway") + if err := flags.Parse(args); err != nil { + return err + } + if options.Connection == "" { + return fmt.Errorf("--connection is required") + } + return Serve(os.Stdin, os.Stdout, options) +} + +// Serve proxies one session until the agent closes its end. +func Serve(in *os.File, out *os.File, options Options) error { + client := newGatewayClient(options) + + // Resolving is what needs Gateway running. Failing here with a plain + // message is better than a half-open session: the agent shows the server as + // failed to start, which is what actually happened. + resolved, err := client.resolve() + if err != nil { + return fmt.Errorf("cannot reach connection %q: %w", options.Connection, err) + } + + server, err := openUpstream(resolved) + if err != nil { + return err + } + defer server.Close() + + proxy := &proxy{options: options, client: client, server: server, out: out} + if err := server.Start(proxy.emit); err != nil { + return err + } + return proxy.pump(in) +} + +type proxy struct { + options Options + client *gatewayClient + server upstream + out *os.File + + // writing serializes what reaches the agent: replies the upstream produced + // and refusals written here are both messages on one stream. + writing sync.Mutex +} + +// message is only what the proxy has to look at. Everything else about a +// JSON-RPC message is forwarded byte for byte, so a field this build has never +// heard of still reaches the other side intact. +type message struct { + Id json.RawMessage `json:"id,omitempty"` + Method string `json:"method"` + Params struct { + Name string `json:"name"` + } `json:"params"` +} + +// pump forwards everything the agent sends, asking about the calls that need +// asking about. +func (p *proxy) pump(in *os.File) error { + scanner := bufio.NewScanner(in) + scanner.Buffer(make([]byte, 0, 64*1024), maxLineBytes) + for scanner.Scan() { + line := bytes.TrimSpace(scanner.Bytes()) + if len(line) == 0 { + continue + } + // The scanner reuses its buffer, and the upstream write may outlive + // this iteration. + forwarded := append([]byte(nil), line...) + + var parsed message + if err := json.Unmarshal(forwarded, &parsed); err == nil && parsed.Method == "tools/call" { + allowed, reason := p.client.allowTool(parsed.Params.Name) + if !allowed { + if err := p.refuse(parsed.Id, reason); err != nil { + return err + } + continue + } + } + + if err := p.server.Send(forwarded); err != nil { + return err + } + } + return scanner.Err() +} + +// emit writes one upstream message to the agent. +func (p *proxy) emit(line []byte) error { + p.writing.Lock() + defer p.writing.Unlock() + if _, err := p.out.Write(append(bytes.TrimSpace(line), '\n')); err != nil { + return err + } + return nil +} + +// refuse answers a denied call as a tool error rather than a protocol error, so +// the model is told it may not use the tool and carries on, instead of the +// client treating the server as broken. +func (p *proxy) refuse(id json.RawMessage, reason string) error { + if len(id) == 0 { + return nil + } + if reason == "" { + reason = "this tool is not allowed for this agent" + } + + body, err := json.Marshal(map[string]any{ + "jsonrpc": "2.0", + "id": id, + "result": map[string]any{ + "isError": true, + "content": []map[string]any{{"type": "text", "text": "Refused by Casbin Gateway: " + reason}}, + }, + }) + if err != nil { + return err + } + return p.emit(body) +} diff --git a/mcpproxy/upstream.go b/mcpproxy/upstream.go new file mode 100644 index 0000000..9eecb55 --- /dev/null +++ b/mcpproxy/upstream.go @@ -0,0 +1,115 @@ +// Copyright 2026 The casbin Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package mcpproxy + +import ( + "bufio" + "bytes" + "fmt" + "os" + "os/exec" + "sync" +) + +// upstream is the MCP server one connection actually reaches, whether that is a +// process started here or a service reached over HTTP. +// +// Start is given the function that delivers a server message to the agent, +// because the two transports produce those at different times: a stdio server +// writes whenever it likes, an HTTP one answers a request. +type upstream interface { + Start(emit func([]byte) error) error + Send(line []byte) error + Close() error +} + +func openUpstream(resolved *resolvedServer) (upstream, error) { + switch resolved.Transport { + case "stdio": + return &stdioUpstream{resolved: resolved}, nil + case "http": + return newHttpUpstream(resolved), nil + default: + return nil, fmt.Errorf("unknown transport %q for this connection", resolved.Transport) + } +} + +// stdioUpstream runs the server as a child process. The credential reaches it +// through the environment and arguments Gateway rendered, and never touches the +// agent's own configuration file. +type stdioUpstream struct { + resolved *resolvedServer + command *exec.Cmd + stdin interface { + Write([]byte) (int, error) + Close() error + } + writing sync.Mutex +} + +func (u *stdioUpstream) Start(emit func([]byte) error) error { + command := exec.Command(u.resolved.Command, u.resolved.Args...) + command.Env = append(os.Environ(), u.resolved.envPairs()...) + // The child's own diagnostics belong on this process's stderr, where the + // agent shows them, rather than mixed into the JSON-RPC stream. + command.Stderr = os.Stderr + + stdin, err := command.StdinPipe() + if err != nil { + return err + } + stdout, err := command.StdoutPipe() + if err != nil { + return err + } + if err := command.Start(); err != nil { + return fmt.Errorf("cannot start %s: %w", u.resolved.Command, err) + } + u.command, u.stdin = command, stdin + + go func() { + scanner := bufio.NewScanner(stdout) + scanner.Buffer(make([]byte, 0, 64*1024), maxLineBytes) + for scanner.Scan() { + line := bytes.TrimSpace(scanner.Bytes()) + if len(line) == 0 { + continue + } + if emit(append([]byte(nil), line...)) != nil { + return + } + } + }() + return nil +} + +func (u *stdioUpstream) Send(line []byte) error { + u.writing.Lock() + defer u.writing.Unlock() + _, err := u.stdin.Write(append(line, '\n')) + return err +} + +func (u *stdioUpstream) Close() error { + if u.command == nil { + return nil + } + if u.stdin != nil { + // Closing stdin is how an MCP server is asked to stop; killing it + // outright would leave whatever it was writing half-written. + _ = u.stdin.Close() + } + return u.command.Wait() +} diff --git a/object/connection_install.go b/object/connection_install.go index b8bf155..f62d209 100644 --- a/object/connection_install.go +++ b/object/connection_install.go @@ -19,43 +19,52 @@ import ( "sort" "github.com/apache/casbin-gateway/agentconfig" + "github.com/apache/casbin-gateway/agentpatch" "github.com/apache/casbin-gateway/connector" + "github.com/apache/casbin-gateway/mcpproxy" ) // InstallConnection writes one connection's MCP server into each agent in // agentIds and takes it out of every agent it was in before. One target's // failure is reported against that target and the rest still run, which is how // the rest of agentconfig behaves. +// +// What is written is not the server itself but Gateway standing in front of it, +// so the third-party credential never reaches the agent's configuration file +// and every call through it can be asked about. func InstallConnection(connection *Connection, agentIds []string) ([]*agentconfig.PlanItem, error) { found, ok := connector.Get(connection.Name) if !ok { return nil, fmt.Errorf("no connector named %q", connection.Name) } - rendered, err := found.Render(connection.Credentials) - if err != nil { + // Rendering here does not produce what is written; it is what refuses to + // install a connection whose credentials are incomplete. + if _, err := found.Render(connection.Credentials); err != nil { return nil, err } planned := []*agentconfig.PlanItem{} - if len(agentIds) > 0 { - planned, err = agentconfig.AddMcp(agentconfig.McpRequest{ - Owner: connection.Owner, - To: agentIds, - Name: rendered.Name, - Transport: rendered.Transport, - Command: rendered.Command, - Args: rendered.Args, - Env: rendered.Env, - Url: rendered.Url, - Headers: rendered.Headers, - Overwrite: true, - }) + for _, agentId := range agentIds { + request, err := proxyRequest(connection, found, agentId) if err != nil { - return nil, err + planned = append(planned, &agentconfig.PlanItem{ + AgentId: agentId, Name: found.Server.Name, Action: "skip", Reason: err.Error(), + }) + continue } + // One agent at a time: each is given a credential of its own, so the + // entry written is not the same entry twice. + items, err := agentconfig.AddMcp(request) + if err != nil { + planned = append(planned, &agentconfig.PlanItem{ + AgentId: agentId, Name: found.Server.Name, Action: "skip", Reason: err.Error(), + }) + continue + } + planned = append(planned, items...) } - planned = append(planned, removeFrom(connection, rendered.Name, dropped(connection.Agents, agentIds))...) + planned = append(planned, removeFrom(connection, found.Server.Name, dropped(connection.Agents, agentIds))...) connection.Agents = installed(planned, agentIds) if err := SaveConnection(connection); err != nil { @@ -64,6 +73,44 @@ func InstallConnection(connection *Connection, agentIds []string) ([]*agentconfi return planned, nil } +// proxyRequest is the entry one agent gets: this executable, told which +// connection to reach and how to reach Gateway. +// +// The token in it is Gateway's own, not the service's: it works from loopback +// only, names one agent, and is revoked from here. The credential that would be +// worth stealing stays in Gateway's database. +func proxyRequest(connection *Connection, found connector.Connector, agentId string) (agentconfig.McpRequest, error) { + executable, err := agentpatch.GatewayExecutable() + if err != nil { + return agentconfig.McpRequest{}, err + } + gatewayUrl, err := agentpatch.GatewayBaseUrl() + if err != nil { + return agentconfig.McpRequest{}, err + } + token, err := agentpatch.IssueIngestToken(agentpatch.Target{AgentId: agentId, Owner: connection.Owner}) + if err != nil { + return agentconfig.McpRequest{}, err + } + + return agentconfig.McpRequest{ + Owner: connection.Owner, + To: []string{agentId}, + Name: found.Server.Name, + Transport: agentconfig.TransportStdio, + Command: executable, + Args: []string{ + mcpproxy.Subcommand, + "--connection", connection.Name, + "--agent", agentId, + "--owner", connection.Owner, + "--gateway-url", gatewayUrl, + "--token", token, + }, + Overwrite: true, + }, nil +} + // UninstallConnection takes the server out of every agent that has it and // forgets the connection along with its credentials. func UninstallConnection(owner string, name string) ([]*agentconfig.PlanItem, error) { @@ -83,6 +130,30 @@ func UninstallConnection(owner string, name string) ([]*agentconfig.PlanItem, er return planned, nil } +// ResolveConnection is what the proxy asks for when an agent starts it: the +// server behind one connection, with its credentials filled in. It is answered +// over loopback to a process this Gateway's own configuration launched, and the +// result is never written to disk. +func ResolveConnection(owner string, name string) (*connector.Rendered, error) { + connection, err := GetConnection(owner, name) + if err != nil { + return nil, err + } + if connection == nil { + return nil, fmt.Errorf("%q is not connected", name) + } + found, ok := connector.Get(name) + if !ok { + return nil, fmt.Errorf("no connector named %q", name) + } + + rendered, err := found.Render(connection.Credentials) + if err != nil { + return nil, err + } + return &rendered, nil +} + // removeFrom deletes the server entry from each agent, reporting what happened // per agent. An agent that no longer has the entry is not an error: the // operator may have deleted it by hand, and the connection still has to let go. diff --git a/routers/router.go b/routers/router.go index fddb7bb..f49d7b9 100644 --- a/routers/router.go +++ b/routers/router.go @@ -112,6 +112,7 @@ func initAPI() { beego.Router("/api/get-connection", &controllers.ApiController{}, "GET:GetConnection") beego.Router("/api/connect", &controllers.ApiController{}, "POST:Connect") beego.Router("/api/disconnect", &controllers.ApiController{}, "POST:Disconnect") + beego.Router("/api/resolve-connection", &controllers.ApiController{}, "POST:ResolveConnection") // The desktop tray, which holds the local token instead of a session. beego.Router("/api/get-tray-menu", &controllers.ApiController{}, "GET:GetTrayMenu") --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
