Repository navigation
fix(mcp): send mirrored headers from the stdio-to-HTTP bridge #2004
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
Merged
Changes from 16 commits
Commits
Show all changes
20 commits
Select commit
Hold shift + click to select a range
66bee8e
fix(mcp): send mirrored headers from the stdio-to-HTTP bridge
luckyPipewrench ac4bf2c
fix(mcp): keep header-contract receipts verifiable
luckyPipewrench b2fe980
fix(mcp): don't release holds on a structural-only inventory
luckyPipewrench 49e1361
docs(mcp): note bridge list exclusion without tool scanning
luckyPipewrench e8ea51a
fix(mcp): keep bridge header metadata on modern sessions
luckyPipewrench 748f124
fix(mcp): preserve session state during private discovery
luckyPipewrench e5bc2e6
fix(mcp): refuse routed notifications on modern bridge sessions
luckyPipewrench 49d38b4
docs(mcp): note the bridge refuses routed methods sent as notifications
luckyPipewrench dbae903
fix(mcp): preserve initialized bridge header contracts
luckyPipewrench e00821f
fix(mcp): stop held releases before ending the bridge session
luckyPipewrench 6793648
fix(mcp): keep the bridge's negotiated modern version
luckyPipewrench 3a5e3ab
docs(mcp): note the bridge keeps its negotiated revision
luckyPipewrench 001e730
test(mcp): assert shutdown sends no held call
luckyPipewrench 1969a71
test(mcp): check the bridge result after cancellation
luckyPipewrench da8e69d
test(mcp): cover bridge list timeout, empty tool names and spent budgets
luckyPipewrench 87f88c6
fix(mcp): refuse malformed modern protocol versions on the bridge
luckyPipewrench bdc2edd
fix(mcp): refuse any malformed per-request protocol version on the br…
luckyPipewrench aa7677c
Merge remote-tracking branch 'origin/main' into fix/mcp-bridge-param-…
luckyPipewrench 13f9c5f
fix(mcp): close a waiting tools/list and shut down cleanly on cancell…
luckyPipewrench e61d69b
test(mcp): close stdin with cancellation in the list-lock test
luckyPipewrench File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,133 @@ | ||
| // Copyright 2026 Pipelock contributors | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| package mcp | ||
|
|
||
| import ( | ||
| "encoding/hex" | ||
| "fmt" | ||
| "io" | ||
| "net/http" | ||
| "regexp" | ||
| "strings" | ||
| "sync/atomic" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/luckyPipewrench/pipelock/internal/config" | ||
| "github.com/luckyPipewrench/pipelock/internal/evidence/completeness" | ||
| "github.com/luckyPipewrench/pipelock/internal/receipt" | ||
| "github.com/luckyPipewrench/pipelock/internal/testwait" | ||
| ) | ||
|
|
||
| // openReceiptHarness starts a lifecycle-complete chain, as a running proxy | ||
| // does, so completeness analysis applies its intent/outcome pairing rules. | ||
| func openReceiptHarness(t *testing.T, opts *MCPProxyOpts, required bool) *mcpDecisionReceiptHarness { | ||
| t.Helper() | ||
| h := newMCPDecisionReceiptHarness(t) | ||
| if err := h.v1.EmitSessionOpen(); err != nil { | ||
| t.Fatal(err) | ||
| } | ||
| opts.ReceiptEmitter, opts.V2ReceiptEmitter, opts.PolicyHash, opts.RequireReceipts = h.v1, h.v2, mcpTestPolicyHash, required | ||
| return h | ||
| } | ||
|
|
||
| func requireEvidenceNotBroken(t *testing.T, label string, h *mcpDecisionReceiptHarness) { | ||
| t.Helper() | ||
| if err := h.v1.EmitSessionClose("test"); err != nil { | ||
| t.Fatal(err) | ||
| } | ||
| if err := h.rec.Close(); err != nil { | ||
| t.Fatal(err) | ||
| } | ||
| receipts := readActionReceipts(t, h.dir) | ||
| chain := receipt.VerifyChain(receipts, hex.EncodeToString(h.pub)) | ||
| if !chain.Valid { | ||
| t.Fatalf("%s chain invalid: %s", label, chain.Error) | ||
| } | ||
| report := completeness.Analyze(receipts, chain) | ||
| if report.Status == completeness.StatusBroken { | ||
| for _, r := range receipts { | ||
| a := r.ActionRecord | ||
| t.Logf("%s seq=%d action=%s phase=%q verdict=%s layer=%s", label, a.ChainSeq, a.ActionID, a.DecisionPhase, a.Verdict, a.Layer) | ||
| } | ||
| t.Fatalf("%s evidence reported broken: %s", label, report.Error) | ||
| } | ||
| } | ||
|
|
||
| func TestHTTPBridgeHeaderEvidenceCompleteness(t *testing.T) { | ||
| for _, required := range []bool{false, true} { | ||
| t.Run(fmt.Sprintf("forwarded/required=%t", required), func(t *testing.T) { | ||
| var calls atomic.Int32 | ||
| up := bridgeHeaderUpstream(t, func(w http.ResponseWriter, _ *http.Request, f MCPFrame) { | ||
| discoveryList(w, f, headerStateDefs(bridgeHeaderSchema, "Echo input"), "") | ||
| }, func(w http.ResponseWriter, _ *http.Request, f MCPFrame) { | ||
| calls.Add(1) | ||
| _, _ = fmt.Fprintf(w, `{"jsonrpc":"2.0","id":%s,"result":{"content":[]}}`, f.ID) | ||
| }) | ||
| listenerOpts := paramContractOpts(t) | ||
| listener := openReceiptHarness(t, &listenerOpts, required) | ||
| listenerURL, _ := startListenerProxyWithOpts(t, up.URL, listenerOpts) | ||
| bridgeOpts := bridgeHeaderOpts(t) | ||
| bridge := openReceiptHarness(t, &bridgeOpts, required) | ||
| headers := http.Header{"Proxy-Authorization": {"Bearer current-client-token"}} | ||
| out, logs := runBridgeHeaderInput(t, listenerURL, bridgeHeaderCall(`{"region":"body"}`), headers, bridgeOpts) | ||
| if calls.Load() != 1 || strings.Contains(out, `"error"`) { | ||
| t.Fatalf("calls=%d out=%s logs=%s", calls.Load(), out, logs) | ||
| } | ||
| requireEvidenceNotBroken(t, "bridge", bridge) | ||
| requireEvidenceNotBroken(t, "listener", listener) | ||
| }) | ||
| t.Run(fmt.Sprintf("refused/required=%t", required), func(t *testing.T) { | ||
| up := bridgeHeaderUpstream(t, func(w http.ResponseWriter, _ *http.Request, f MCPFrame) { | ||
| discoveryList(w, f, `[]`, "") | ||
| }, func(http.ResponseWriter, *http.Request, MCPFrame) { t.Error("refused call reached upstream") }) | ||
| opts := bridgeHeaderOpts(t) | ||
| h := openReceiptHarness(t, &opts, required) | ||
| out, logs := runBridgeHeaderInput(t, up.URL, bridgeHeaderCall(`{"region":"body"}`), nil, opts) | ||
| if !strings.Contains(out, `"code":-32020`) { | ||
| t.Fatalf("out=%s logs=%s", out, logs) | ||
| } | ||
| requireEvidenceNotBroken(t, "bridge", h) | ||
| }) | ||
| t.Run(fmt.Sprintf("deferred refused/required=%t", required), func(t *testing.T) { | ||
| up := bridgeHeaderUpstream(t, func(w http.ResponseWriter, _ *http.Request, f MCPFrame) { | ||
| discoveryList(w, f, `[]`, "") | ||
| }, func(http.ResponseWriter, *http.Request, MCPFrame) { t.Error("refused release reached upstream") }) | ||
| opts := bridgeHeaderOpts(t) | ||
| h := openReceiptHarness(t, &opts, required) | ||
| manager := newDeferKillManager(t) | ||
| opts.DeferManager = manager | ||
| policy := deferApprovalPolicy(config.DeferResolverProfile{}) | ||
| policy.Rules[0].ToolPattern = regexp.MustCompile(`^echo$`) | ||
| policy.Rules[0].ResolutionPolicy.ResolverProfile = "" | ||
| opts.PolicyCfg = policy | ||
| inputR, inputW := io.Pipe() | ||
| var out, logs syncBuffer | ||
| done := make(chan error, 1) | ||
| go func() { done <- RunHTTPProxy(t.Context(), inputR, &out, &logs, up.URL, nil, opts) }() | ||
| if _, err := io.WriteString(inputW, bridgeHeaderCall(`{"region":"body"}`)+"\n"); err != nil { | ||
| t.Fatal(err) | ||
| } | ||
| testwait.For(t, 5*time.Second, func() bool { return len(manager.Snapshot()) == 1 }, "held call") | ||
| if err := manager.Resolve(manager.Snapshot()[0].DeferID, config.ActionAllow, "approval"); err != nil { | ||
| t.Fatal(err) | ||
| } | ||
| if err := inputW.Close(); err != nil { | ||
| t.Fatal(err) | ||
| } | ||
| select { | ||
| case err := <-done: | ||
| if err != nil { | ||
| t.Fatal(err) | ||
| } | ||
| case <-time.After(testwait.Deadline(10 * time.Second)): | ||
| t.Fatal("bridge did not finish") | ||
| } | ||
| if !strings.Contains(out.String(), `"code":-32020`) { | ||
| t.Fatalf("out=%s logs=%s", out.String(), logs.String()) | ||
| } | ||
| requireEvidenceNotBroken(t, "bridge", h) | ||
| }) | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,109 @@ | ||
| // Copyright 2026 Pipelock contributors | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| package mcp | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "io" | ||
| "net/http" | ||
| "sync" | ||
|
|
||
| "github.com/luckyPipewrench/pipelock/internal/mcp/transport" | ||
| ) | ||
|
|
||
| // mcpBridgeHeaders joins the listener's contract acquisition/serialization and | ||
| // the bridge's per-process tool baseline. Only the contract portion of state is | ||
| // used: this bridge's fixed upstream credentials scope its lifetime. | ||
| type mcpBridgeHeaders struct { | ||
| client *transport.HTTPClient | ||
| state *mcpListenerClientState | ||
| opts MCPProxyOpts | ||
| log io.Writer | ||
|
|
||
| mu sync.Mutex | ||
| // negotiated keeps the first modern version the transport reports. The | ||
| // transport clears its copy on a new handshake and when the session ends, | ||
| // and an empty version would read as legacy and send no request metadata. | ||
| // Once modern, this bridge stays modern for its lifetime. | ||
| negotiated string | ||
| } | ||
|
|
||
| func (b *mcpBridgeHeaders) version(msg []byte) string { | ||
| // A modern per-request version takes precedence over a legacy handshake, | ||
| // matching the listener's request metadata contract. Otherwise preserve the | ||
| // transport's initialize negotiation, including its legacy fallback. A | ||
| // legacy per-request version can't downgrade a modern negotiation, which | ||
| // would otherwise drop the request's routing and parameter headers. | ||
| if version := ParseMCPFrame(msg).ProtocolVersion; version >= currentMCPVersion { | ||
| return version | ||
| } | ||
| return b.negotiatedVersion() | ||
| } | ||
|
|
||
| func (b *mcpBridgeHeaders) negotiatedVersion() string { | ||
| b.mu.Lock() | ||
| defer b.mu.Unlock() | ||
| version := b.client.ProtocolVersion() | ||
| if version >= currentMCPVersion { | ||
| b.negotiated = version | ||
| } | ||
| if b.negotiated != "" { | ||
| return b.negotiated | ||
| } | ||
| return version | ||
| } | ||
|
|
||
| func (b *mcpBridgeHeaders) prepare(ctx context.Context, msg []byte) (http.Header, error) { | ||
| version := b.version(msg) | ||
| if version < currentMCPVersion { | ||
| return nil, nil | ||
| } | ||
| if !validMCPProtocolVersion([]string{version}) { | ||
| // Lexical comparison admits malformed dates such as 2026-99-99; the | ||
| // listener accepts only canonical YYYY-MM-DD, so refuse them here | ||
| // rather than forward an unusable protocol header. | ||
| return nil, fmt.Errorf("invalid MCP protocol version") | ||
| } | ||
| frame := ParseMCPFrame(msg) | ||
| if frame.Method == "initialize" && b.negotiatedVersion() >= currentMCPVersion { | ||
| // This process already established its modern header contract. A | ||
| // second handshake would clear the transport's negotiated version, | ||
| // including while deferred calls can be released. Refuse it before | ||
| // that reset rather than let those calls take the legacy path. | ||
| return nil, fmt.Errorf("modern bridge is already initialized") | ||
| } | ||
| if isRPCNotification(frame.ID) { | ||
| switch frame.Method { | ||
| case methodToolsCall, methodResourcesRead, methodPromptsGet: | ||
| // MCP defines these methods only as requests, which must carry | ||
| // Mcp-Name and parameter mirrors. A notification form would leave | ||
| // unmirrored, so it is refused like an unavailable contract. | ||
| return nil, fmt.Errorf("%s cannot be sent as a notification", frame.Method) | ||
| } | ||
| } | ||
| if frame.Method == methodToolsCall { | ||
| if frame.RoutingNameErr != nil || frame.RoutingName == "" { | ||
| return nil, fmt.Errorf("invalid tool routing name") | ||
| } | ||
| req := &http.Request{Header: make(http.Header)} | ||
| req = req.WithContext(ctx) | ||
| req.Header.Set(listenerProtocolVersion, version) | ||
| release, err := b.state.acquireToolContract(ctx, frame.RoutingName, func() error { | ||
| return discoverMCPToolContract(mcpContractDiscoveryRequest{ | ||
| request: req, state: b.state, bridge: b.client, opts: b.opts, log: b.log, tool: frame.RoutingName, | ||
| }) | ||
| }) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("cannot establish tool header contract: %w", err) | ||
| } | ||
| defer release() | ||
| } | ||
| headers, err := synthesizeMCPRequestHeaders(frame, b.state.baseline) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| headers.Set(listenerProtocolVersion, version) | ||
| return headers, nil | ||
| } | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.