diff --git a/modules/pier/README.md b/modules/pier/README.md index 4bf81297..57f53e67 100644 --- a/modules/pier/README.md +++ b/modules/pier/README.md @@ -39,7 +39,7 @@ Use cases: - Session reaper (A-CR-01): most clients exit without DELETE, so a janitor started at mount checks every local session against the store every `min(mcp.sessions.idle_ttl/2, 1m)` and closes each local copy whose record has expired (its SDK server session ends and it leaves the fan-out set), logging the `expired` step. The store's TTL decides on every instance: a session another instance keeps alive still has its record, so no instance closes it. No configuration key; the reaper runs for the life of the process. - Session stores: `memory` (one instance; records in the process) or `redis` (records `mcp:s:` with `PX` TTLs, per-binding sets `mcp:b:`, events on channel `mcp:events`, under `cooler.KeyPrefix`); Mount never waits for Redis: a background warm-up opens the client from `redis.url` with a backoff (one second doubling to thirty, each failure logged) and subscribes to the event channel, so an instance that serves only stateless clients still receives fan-out, and the first request on any path also starts the open without waiting for it; while the store is down stateful requests answer 503 and stateless requests are served. - Fan-out (D-26): every `pact.MCPSubjectChanged` fired on the application is published through the store and applied on every instance: a permission or link change recomputes the tools and prompts of the subject's sessions and stateless servers (the SDK then sends `notifications/tools/list_changed` and `notifications/prompts/list_changed`), a revocation or deletion closes those sessions, drops their records and evicts the stateless servers; a link event also drops the link cache at once. A revocation naming `TokenID` ends only the sessions whose subject under the event's guard is that user with that token id and no client id (one personal token); one naming `ClientID` ends only that OAuth client's sessions of the user; one naming neither, and every deletion, ends all sessions of the subject. -- Client address: before resolving a bearer, pier puts the client address on the resolvers' context with `towel.WithClientIP`, taken from the host when it offers one (surf applies `http.trusted_proxies`) or from the request's `RemoteAddr`, so a resolver can log a failed attempt with it. +- Client address: before resolving a bearer, pier puts the client address on the resolvers' context with `towel.WithClientIP`, taken from the host when it offers one (surf applies `http.trusted_proxies`) or from the request's `RemoteAddr`, so a resolver can log a failed attempt with it. The periodic bearer re-check of a GET or `subscriptions/listen` stream carries the address the stream was opened from. - Hybrid protocol (D-32): a request whose `Mcp-Protocol-Version` is 2026-07-28 or later, or whose first JSON-RPC method is `server/discover`, goes to the SDK's stateless handler over an in-process LRU of per-identity servers keyed by binding (capacity `mcp.sessions.max_per_binding` times 16); every other request goes to the stateful router. Both run behind the same bearer verification. Past capacity the least recently used idle server is evicted (and its sessions closed); a server that is serving a request, a `subscriptions/listen` stream included, is never evicted, so capacity never ends a quiet listener or a tool call in flight, and the LRU stays above capacity while every server is busy. A stateless `subscriptions/listen` stream re-verifies its bearer every minute and ends when it no longer resolves to its binding, as the stateful GET stream does. - Widening on return (A-WR-04): when a binding comes back with a different identity (more or fewer scopes, a backend subject added or dropped), its stateful session or stateless server is re-filtered on that request, so newly allowed tools and prompts appear at once (with `list_changed`) instead of waiting for an event. Token ids play no part in this, so an access-token rotation alone re-filters nothing. A stateless server is shared by every live token of its binding, so a return with less reach only adds there and never removes: the receiving middleware hides, per request, what that request's identity may not use, and a call of a wider token is never cut off by a narrower one. A guard provider error during this re-filtering is not a refusal: the cached server stays as it was and the next request retries (logged as `pier: widening deferred`); a fan-out recompute that meets one hides the failing items and leaves the server to be re-filtered by the binding's next request. - Protected resource metadata (RFC 9728, `pier.PRMScopes`): `GET /.well-known/oauth-protected-resource` and `GET /.well-known/oauth-protected-resource` answer the resource, the authorization server (the published verifier's resource origin, else `app.url`), the supported scopes and the bearer method; a `mcp.resource` that differs from the verifier's resource is a boot warning. diff --git a/modules/pier/session.go b/modules/pier/session.go index be6c37d0..193fe6cf 100644 --- a/modules/pier/session.go +++ b/modules/pier/session.go @@ -15,6 +15,7 @@ import ( "sync/atomic" "time" + "git.golem15.com/golem15/summercms/modules/towel" "github.com/modelcontextprotocol/go-sdk/auth" "github.com/modelcontextprotocol/go-sdk/jsonrpc" "github.com/modelcontextprotocol/go-sdk/mcp" @@ -425,12 +426,15 @@ func (sr *sessionRouter) persist(ctx context.Context, ls *localSession) { // serveStream serves the standalone GET stream and re-verifies its bearer // every recheck interval on a goroutine that holds the raw bearer only for // the stream's lifetime; a bearer that no longer resolves to the session's -// binding ends the stream (Pitfall 10). The server's base context ends +// binding ends the stream (Pitfall 10). The re-check resolves with the +// stream's client address on the context, as the bearer middleware does, so +// a failed re-check is logged with it. The server's base context ends // every stream at shutdown. func (sr *sessionRouter) serveStream(w http.ResponseWriter, r *http.Request, ls *localSession) { ctx, cancel := context.WithCancel(r.Context()) defer cancel() raw := bearerOf(r) + ip := sr.p.clientIP(r) go func() { ticker := time.NewTicker(sr.p.recheck) defer ticker.Stop() @@ -439,7 +443,7 @@ func (sr *sessionRouter) serveStream(w http.ResponseWriter, r *http.Request, ls case <-ctx.Done(): return case <-ticker.C: - id, err := sr.p.identities.resolve(ctx, raw) + id, err := sr.p.identities.resolve(towel.WithClientIP(ctx, ip), raw) if err != nil || id.Binding != ls.binding { logSession(logger(), "stream_closed", ls.id, ls.binding) cancel() diff --git a/modules/pier/stateless.go b/modules/pier/stateless.go index 87171246..72cf684b 100644 --- a/modules/pier/stateless.go +++ b/modules/pier/stateless.go @@ -11,6 +11,7 @@ import ( "sync" "time" + "git.golem15.com/golem15/summercms/modules/towel" "github.com/modelcontextprotocol/go-sdk/mcp" ) @@ -125,6 +126,7 @@ func (sp *statelessPath) ServeHTTP(w http.ResponseWriter, r *http.Request) { ctx, cancel := context.WithCancel(r.Context()) defer cancel() raw := bearerOf(r) + ip := sp.p.clientIP(r) go func() { ticker := time.NewTicker(sp.p.recheck) defer ticker.Stop() @@ -133,7 +135,7 @@ func (sp *statelessPath) ServeHTTP(w http.ResponseWriter, r *http.Request) { case <-ctx.Done(): return case <-ticker.C: - cur, err := sp.p.identities.resolve(ctx, raw) + cur, err := sp.p.identities.resolve(towel.WithClientIP(ctx, ip), raw) if err != nil || cur.Binding != id.Binding { logSession(logger(), "stream_closed", "", id.Binding) cancel() diff --git a/modules/pier/stateless_fanout_test.go b/modules/pier/stateless_fanout_test.go index e0555dbc..79fa9abc 100644 --- a/modules/pier/stateless_fanout_test.go +++ b/modules/pier/stateless_fanout_test.go @@ -12,6 +12,7 @@ import ( "time" "git.golem15.com/golem15/summercms/modules/pact" + "git.golem15.com/golem15/summercms/modules/towel" "github.com/modelcontextprotocol/go-sdk/mcp" ) @@ -182,3 +183,61 @@ func TestStatelessFanOut(t *testing.T) { waitEnded(t, ended, 5*time.Second, "revocation after the eviction") }) } + +// ipVerifier is a fakeVerifier that records the client address on the +// context of every verification. +type ipVerifier struct { + *fakeVerifier + mu sync.Mutex + ips []string +} + +func (v *ipVerifier) MCPVerifyToken(ctx context.Context, raw string) (pact.MCPGrant, error) { + ip, _ := towel.ClientIP(ctx) + v.mu.Lock() + v.ips = append(v.ips, ip) + v.mu.Unlock() + return v.fakeVerifier.MCPVerifyToken(ctx, raw) +} + +func (v *ipVerifier) seen() []string { + v.mu.Lock() + defer v.mu.Unlock() + return append([]string(nil), v.ips...) +} + +// TestStreamRecheckClientIP pins RA-IN-02: the periodic bearer re-check of +// a stateless listen stream and of a stateful GET stream resolves with the +// stream's client address on the context, as the bearer middleware does, +// so a failing re-check is logged with it. +func TestStreamRecheckClientIP(t *testing.T) { + plugin := acmePlugin{tools: []pact.MCPTool{echoTool()}} + for _, path := range []struct{ name, version string }{{"stateless listen", "newest"}, {"stateful GET", ""}} { + t.Run(path.name, func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + verifier := &ipVerifier{fakeVerifier: newFakeVerifier(nil)} + in := startInstance(t, newApp(t, appSpec{guards: []pact.MCPGuard{&acmeGuard{}}, verifier: verifier}), plugin) + t.Cleanup(in.pier().StopReaper) + in.pier().SetRecheck(30 * time.Millisecond) + if path.version == "newest" { + listenClient(t, ctx, in.url(), "oas_at_test") + } else { + sdkClient(t, ctx, in.url(), "oas_at_test", clientOpts{changed: make(chan struct{}, 1)}) + } + before := len(verifier.seen()) + deadline := time.Now().Add(5 * time.Second) + for len(verifier.seen()) < before+3 { + if time.Now().After(deadline) { + t.Fatalf("no stream re-checks within 5 s (%d verifications)", len(verifier.seen())) + } + time.Sleep(20 * time.Millisecond) + } + for i, ip := range verifier.seen() { + if ip == "" { + t.Fatalf("verification %d ran without a client address: %q", i, verifier.seen()) + } + } + }) + } +}