From 806e87c510dac5b2708c9a3775dad547a6c80ef3 Mon Sep 17 00:00:00 2001 From: Jakub Zych Date: Thu, 22 Oct 2026 18:23:43 +0200 Subject: [PATCH] feat(19.7-02): Redis sessions, event fan-out, stateless protocol path and resource metadata - store_redis.go: records with PX TTLs, per-binding sets, pub/sub events, opened lazily from redis.url under cooler's prefix (nothing dials at mount) - events.go: pact.MCPSubjectChanged republished through the store; every instance recomputes the subject's sessions and stateless servers (list_changed) or ends them on revocation; the link cache is dropped - stateless.go: the SDK's stateless handler over per-identity servers for 2026-07-28 clients, routed by the protocol header or server/discover - prm.go: RFC 9728 metadata at both well-known paths; long GET streams re-verify their bearer every minute - TestPierSessionsSmoke, TestPierFanOutSmoke, TestPierStatelessSmoke on miniredis; README features, configuration and operational notes --- modules/pier/README.md | 15 +- modules/pier/config.go | 12 +- modules/pier/events.go | 184 ++++++++++++++++++++ modules/pier/pier.go | 52 +++++- modules/pier/prm.go | 59 +++++++ modules/pier/session.go | 39 ++++- modules/pier/sessions_smoke_test.go | 247 +++++++++++++++++++++++++++ modules/pier/stateless.go | 174 +++++++++++++++++++ modules/pier/stateless_smoke_test.go | 80 +++++++++ modules/pier/store_redis.go | 247 +++++++++++++++++++++++++++ modules/pier/tracer_smoke_test.go | 56 ++++-- 11 files changed, 1133 insertions(+), 32 deletions(-) create mode 100644 modules/pier/events.go create mode 100644 modules/pier/prm.go create mode 100644 modules/pier/sessions_smoke_test.go create mode 100644 modules/pier/stateless.go create mode 100644 modules/pier/stateless_smoke_test.go create mode 100644 modules/pier/store_redis.go diff --git a/modules/pier/README.md b/modules/pier/README.md index c38860a3..802d9b01 100644 --- a/modules/pier/README.md +++ b/modules/pier/README.md @@ -12,7 +12,7 @@ pier mounts a streamable HTTP MCP endpoint (POST, GET and DELETE at `mcp.path`, Every request to the endpoint is authorized afresh from its bearer: an OAuth access token through the `pact.MCPTokenVerifier` the OAuth server plugin publishes, or, behind `mcp.allow_personal_tokens`, a personal API token (an admin `g15_` token through cabana, a frontend token through the published `pact.MCPPersonalTokenResolver`). The resulting `pier.Identity` names a frontend subject and, when the grant carries the `backend` scope and the published `pact.MCPLinkedBackendResolver` confirms a link, a backend subject. `tools/list` is computed for that identity: a tool needs a subject under its guard, a grant without `write` sees only read-only tools, and the guard provider's own check (the same logic its route middleware enforces) decides the rest. A `tools/call` runs the bound route's compiled middleware chain in-process with the `bouncer.Dispatch` marker carrying the principal the guard provider resolved; the inner request never carries an `Authorization` or `Cookie` header. A 2xx answer becomes text plus structured content, anything else a tool error carrying the route's own body. -Sessions (`Mcp-Session-Id`) are bound to the identity that opened them (client and user, or the personal token), live in a store with a sliding idle TTL, and are served by a per-session SDK server on whichever instance receives the request, rehydrated from the stored session state when it is not local. The memory store serves one instance; the Redis store (next task) shares sessions and events across instances. +Sessions (`Mcp-Session-Id`) are bound to the identity that opened them (client and user, or the personal token), live in a store with a sliding idle TTL, and are served by a per-session SDK server on whichever instance receives the request, rehydrated from the stored session state when it is not local. The memory store serves one instance; the Redis store shares sessions across instances and fans subject events out to every instance, so a permission or link change pushes `notifications/tools/list_changed` to the sessions it concerns and a revocation ends them everywhere. Clients of protocol 2026-07-28 and later, which have no sessions, are served statelessly from the same path by the SDK's stateless handler over per-identity servers, sharing identity, dispatch and filtering with the stateful path. ## Role in the framework @@ -35,7 +35,11 @@ Use cases: - Built-in `backend` guard provider (`pier.BackendGuard`): routes behind cabana's `backend` middleware, listing through `cabana.Allows` on the administrator's principal from `cabana.BackendPrincipal`, an admin personal token's scope limit computed per dispatched route method with `cabana.ScopeLimitFor`; a plugin that registers its own `backend` provider wins. - Identity per request (`pier.Identity`): `oas_at_` tokens through the published verifier, personal tokens only with `mcp.allow_personal_tokens`, the backend subject through the published link resolver with an in-process cache of at most `pier.LinkCacheTTL`; the binding key `oauth::`, `pat:user:` or `pat:backend:` survives access-token rotation. - Per-identity SDK server plus a receiving middleware that re-filters `tools/list` and `prompts/list` and refuses `tools/call` and `prompts/get` for the identity of the current request, so a permission change takes effect on the next request (D-08, D-25). -- Stateful sessions: `initialize` opens a session (429 above `mcp.sessions.max_per_binding`), a request with an unknown id answers 404 (the client re-initialises), another identity presenting the id answers 403 `session_binding`, DELETE ends the session everywhere, the record's TTL slides on every request, a session initialised on one instance is rehydrated on another from the stored `ServerSessionState`. +- Stateful sessions (D-25, D-27): `initialize` opens a session (429 above `mcp.sessions.max_per_binding`), a request with an unknown id answers 404 (the client re-initialises), another identity presenting the id answers 403 `session_binding`, DELETE ends the session everywhere, the record's TTL slides on every request, a session initialised on one instance is rehydrated on another from the stored `ServerSessionState`. A standalone GET stream re-verifies its bearer every minute and ends when it no longer resolves to the session's binding. +- 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`); the Redis client is opened on the first request from `redis.url`, never at mount, so `route:list` and tests never dial, and a missing URL is a 503 logged once. +- 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. +- 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. +- 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. - Prompts (D-28): prefixed and filtered like tools, required arguments checked before `Render` runs; a prompt never dispatches a route. - A 401 carries `WWW-Authenticate: Bearer resource_metadata="/.well-known/oauth-protected-resource", scope="read"`; a browser request whose `Origin` is not in `mcp.allowed_origins` is refused with 403 before any body is read. - One audit line per tool call (`pier: tool call`: tool, guard, user id, admin id, client id, token kind, session prefix, status, duration) and one per session step (`pier: session`), never arguments, bodies or tokens. @@ -103,6 +107,7 @@ An MCP client then connects to `https://example.com/mcp` with an OAuth access to | `pier.Record` | One stored session: binding, session state JSON and creation time. | | `pier.Event` | One instance-to-instance event: a kind (`pier.EventEnded` or a `pact.MCPSubjectChanged` kind), guard, user id, session id and publishing instance. | | `pier.LogComponent` | `pier`, the diary component of the audit lines. | +| `pier.PRMScopes` | The `scopes_supported` list of the protected resource metadata (`read`, `write`, `ai`, `offline_access`, `backend`). | Configuration keys are exported as constants: `pier.EnabledKey`, `pier.PathKey`, `pier.AllowPersonalTokensKey`, `pier.SessionStoreKey`, `pier.SessionIdleTTLKey`, `pier.SessionMaxPerBindingKey`, `pier.ResourceKey`, `pier.ThrottleKey`, `pier.AllowedOriginsKey`, `pier.CallTimeoutKey`, `pier.MaxResultBytesKey`, with defaults `pier.DefaultPath`, `pier.DefaultSessionStore`, `pier.DefaultIdleTTL`, `pier.DefaultMaxPerBinding`, `pier.DefaultThrottle`, `pier.DefaultCallTimeout` and `pier.DefaultMaxResultBytes`, and store names `pier.StoreMemory` and `pier.StoreRedis`. @@ -113,7 +118,7 @@ Configuration keys are exported as constants: `pier.EnabledKey`, `pier.PathKey`, | `mcp.enabled` | `false` | Mounts the endpoint. `true` in a binary built without the `summercms_mcp` tag fails boot naming the tag. | | `mcp.path` | `/mcp` | The endpoint path: an absolute URL path without scheme, host, query, wildcard or dot segments. A route a plugin registered there fails boot naming the plugin. | | `mcp.allow_personal_tokens` | `false` | Lets personal API tokens open sessions: an admin `g15_` token yields backend tools only, a frontend token (through the published resolver) frontend tools only, each capped by the token's own scope. OAuth tokens are always accepted. | -| `mcp.sessions.store` | `memory` | `memory` (one instance; sessions end with the process) or `redis` (shared through `redis.url`; the Redis store arrives with the next task). | +| `mcp.sessions.store` | `memory` | `memory` (one instance; sessions end with the process and events reach only its own sessions) or `redis` (sessions and events shared by every instance through `redis.url`, opened on the first request). | | `mcp.sessions.idle_ttl` | `30m` | Sliding idle TTL of a session record. | | `mcp.sessions.max_per_binding` | `20` | Open sessions one binding (client and user, or personal token) may hold; the next `initialize` answers 429. | | `mcp.resource` | `app.url` + `mcp.path` | The resource indicator the endpoint identifies itself by; must equal the OAuth server's MCP resource. Required (or `app.url`) when enabled. | @@ -124,6 +129,8 @@ Configuration keys are exported as constants: `pier.EnabledKey`, `pier.PathKey`, Environment: `SUMMER_MCP__ENABLED`, `SUMMER_MCP__PATH`, `SUMMER_MCP__SESSIONS__STORE` and so on. +Operational notes: a single instance (the demo deployment) runs on the memory store; several instances behind one load balancer need `mcp.sessions.store: redis` so a session opened on one instance is served by another and a permission change reaches every session. A reverse proxy must route `mcp.path` and both `/.well-known/oauth-protected-resource` paths to the binary and must not buffer the endpoint's responses (server-sent events). Every tool call leaves one `pier: tool call` line and every session step one `pier: session` line under the `pier` component. + ## Dependencies - `github.com/modelcontextprotocol/go-sdk` v1.8.0 (`mcp`, `auth`, `oauthex`, `jsonrpc`): the server, sessions, streamable HTTP transport, bearer middleware and metadata types; decided in `.planning/notes/2026-10-22-mcp-go-sdk.md`. @@ -134,4 +141,4 @@ Only this module (and surf's tagged `pier_on.go`) imports the SDK; `pact` stays ## Testing -`go test -short ./modules/pier/...` runs without Docker, Redis or a build tag: `TestPierTracerSmoke` builds a real surf router with a fake plugin, mounts pier through an in-test `pier.Host` adapter, connects the SDK client over `httptest` and proves `tools/list`, `tools/call` through the compiled guard chain with the dispatch marker (no `Authorization` or `Cookie` reaches the route), the 401 challenge, and the boot errors for a duplicate name and an unknown route. `go test -short -tags summercms_mcp ./modules/pier/... ./modules/surf/...` runs the same tests with the SDK linked into surf. +`go test -short ./modules/pier/...` runs without Docker, a real Redis or a build tag: `TestPierTracerSmoke` builds a real surf router with a fake plugin, mounts pier through an in-test `pier.Host` adapter, connects the SDK client over `httptest` and proves `tools/list`, `tools/call` through the compiled guard chain with the dispatch marker (no `Authorization` or `Cookie` reaches the route), the 401 challenge, and the boot errors for a duplicate name and an unknown route; `TestPierSessionsSmoke` runs two instances on one miniredis (rehydration, the binding refusal, DELETE, the idle TTL, the per-binding cap); `TestPierFanOutSmoke` receives `list_changed` on one instance for an event fired on the other and sees a revocation end the session; `TestPierStatelessSmoke` connects the newest SDK client statelessly next to a 2025-11-25 client with a session. `go test -short -tags summercms_mcp ./modules/pier/... ./modules/surf/...` runs the same tests with the SDK linked into surf. diff --git a/modules/pier/config.go b/modules/pier/config.go index cce6b27f..2d704198 100644 --- a/modules/pier/config.go +++ b/modules/pier/config.go @@ -49,12 +49,12 @@ const ( // Defaults of the mcp.* keys. const ( - DefaultPath = "/mcp" - DefaultSessionStore = "memory" - DefaultIdleTTL = 30 * time.Minute - DefaultMaxPerBinding = 20 - DefaultThrottle = "120,1" - DefaultCallTimeout = 30 * time.Second + DefaultPath = "/mcp" + DefaultSessionStore = "memory" + DefaultIdleTTL = 30 * time.Minute + DefaultMaxPerBinding = 20 + DefaultThrottle = "120,1" + DefaultCallTimeout = 30 * time.Second DefaultMaxResultBytes = int64(1 << 20) ) diff --git a/modules/pier/events.go b/modules/pier/events.go new file mode 100644 index 00000000..0b688f8c --- /dev/null +++ b/modules/pier/events.go @@ -0,0 +1,184 @@ +package pier + +import ( + "context" + + "git.golem15.com/golem15/summercms/modules/pact" +) + +// listenEvents republishes every pact.MCPSubjectChanged fired on the +// application through the store, so every instance (this one included) +// applies it in handleEvent; when publishing fails the event is applied +// locally at least. +func (p *Pier) listenEvents() { + if p.app == nil || p.app.Events == nil { + return + } + p.app.Events.Listen(PluginID, func(ctx context.Context, ev *pact.MCPSubjectChanged) error { + if ev == nil || ev.Kind == "" { + return nil + } + event := Event{Kind: ev.Kind, Guard: ev.Guard, UserID: ev.UserID, Instance: p.instance} + if err := p.store.Publish(ctx, event); err != nil { + p.handleEvent(event) + } + return nil + }) +} + +// handleEvent applies one event on this instance (D-26, D-27): a session +// ended elsewhere is closed here; a permission or link change recomputes +// the tools and prompts of every local session and stateless server of +// that subject, which makes the SDK send the list_changed notifications; a +// revocation or deletion closes those sessions, drops their records and +// evicts the stateless servers. A link event also drops the link cache. +func (p *Pier) handleEvent(ev Event) { + ctx := context.Background() + switch ev.Kind { + case EventEnded: + if ev.SessionID != "" && ev.Instance != p.instance { + p.sessions.closeLocal(ev.SessionID) + } + case pact.MCPSubjectRevoked, pact.MCPSubjectDeleted: + for _, ls := range p.sessions.matching(ev.Guard, ev.UserID) { + _ = p.store.Delete(ctx, ls.id) + p.sessions.closeLocal(ls.id) + logSession(logger(), "revoked", ls.id, ls.binding) + } + if p.stateless != nil { + p.stateless.evict(ev.Guard, ev.UserID) + } + case pact.MCPSubjectLink, pact.MCPSubjectPermissions: + if ev.Kind == pact.MCPSubjectLink { + if ev.Guard == BackendGuard { + p.identities.forgetAllLinks() + } else { + p.identities.forgetLink(ev.UserID) + } + } + for _, ls := range p.sessions.matching(ev.Guard, ev.UserID) { + p.recomputeSession(ctx, ls, ev.Kind == pact.MCPSubjectLink) + } + if p.stateless != nil { + for _, e := range p.stateless.matching(ev.Guard, ev.UserID) { + p.recomputeStateless(ctx, e, ev.Kind == pact.MCPSubjectLink) + } + } + } +} + +// recomputeSession re-filters a local session's identity (re-resolving +// the backend link after a link event) and applies the difference with +// AddTool, RemoveTools, AddPrompt and RemovePrompts. +func (p *Pier) recomputeSession(ctx context.Context, ls *localSession, relink bool) { + ls.mu.Lock() + id := ls.identity + if relink { + p.identities.refreshBackend(ctx, &id) + ls.identity = id + } + tools, prompts := p.catalog.allowed(ctx, id) + applyDiff(ls.server, p, id, tools, prompts, ls.tools, ls.prompts) + ls.mu.Unlock() +} + +// recomputeStateless does the same for one per-identity stateless server. +func (p *Pier) recomputeStateless(ctx context.Context, e *statelessEntry, relink bool) { + e.mu.Lock() + id := e.identity + if relink { + p.identities.refreshBackend(ctx, &id) + e.identity = id + } + tools, prompts := p.catalog.allowed(ctx, id) + applyDiff(e.server, p, id, tools, prompts, e.tools, e.prompts) + e.mu.Unlock() +} + +// applyDiff brings a server's tool and prompt sets to the allowed ones, +// updating the tracking maps in place. +func applyDiff(srv serverFeatures, p *Pier, id Identity, tools []*tool, prompts []*prompt, haveTools, havePrompts map[string]bool) { + wantTools := map[string]bool{} + for _, t := range tools { + wantTools[t.name] = true + if !haveTools[t.name] { + srv.AddTool(t.desc, p.toolHandler(t, id)) + haveTools[t.name] = true + } + } + var dropTools []string + for name := range haveTools { + if !wantTools[name] { + dropTools = append(dropTools, name) + delete(haveTools, name) + } + } + if len(dropTools) > 0 { + srv.RemoveTools(dropTools...) + } + wantPrompts := map[string]bool{} + for _, pr := range prompts { + wantPrompts[pr.name] = true + if !havePrompts[pr.name] { + srv.AddPrompt(pr.desc, p.promptHandler(pr)) + havePrompts[pr.name] = true + } + } + var dropPrompts []string + for name := range havePrompts { + if !wantPrompts[name] { + dropPrompts = append(dropPrompts, name) + delete(havePrompts, name) + } + } + if len(dropPrompts) > 0 { + srv.RemovePrompts(dropPrompts...) + } +} + +// matching lists the local sessions whose identity includes the subject. +func (sr *sessionRouter) matching(guard string, userID uint) []*localSession { + sr.mu.Lock() + defer sr.mu.Unlock() + var out []*localSession + for _, ls := range sr.local { + ls.mu.Lock() + hit := ls.identity.has(guard, userID) + ls.mu.Unlock() + if hit { + out = append(out, ls) + } + } + return out +} + +// refreshBackend re-resolves an OAuth identity's backend subject through +// the link resolver (after a link event), adding or dropping it. +func (r *identityResolver) refreshBackend(ctx context.Context, id *Identity) { + if id.TokenKind != TokenKindOAuth || id.Frontend == nil { + return + } + id.Backend = nil + for _, s := range id.Scopes { + if s != "backend" { + continue + } + if link, ok := r.linkedBackend(ctx, id.Frontend.UserID); ok { + id.Backend = &pact.MCPSubject{ + Guard: BackendGuard, + UserID: link.ID, + Scopes: id.Scopes, + ClientID: id.Frontend.ClientID, + TokenID: id.Frontend.TokenID, + } + } + return + } +} + +// forgetAllLinks drops the whole link cache. +func (r *identityResolver) forgetAllLinks() { + r.mu.Lock() + r.links = map[uint]linkEntry{} + r.mu.Unlock() +} diff --git a/modules/pier/pier.go b/modules/pier/pier.go index 12f15d45..0a9cb5b1 100644 --- a/modules/pier/pier.go +++ b/modules/pier/pier.go @@ -17,11 +17,18 @@ import ( "runtime/debug" "strings" + "time" + "git.golem15.com/golem15/summercms/modules/backpack" + "git.golem15.com/golem15/summercms/modules/cooler" "git.golem15.com/golem15/summercms/modules/party" "github.com/modelcontextprotocol/go-sdk/auth" ) +// streamRecheckInterval is how often a long-lived GET stream re-verifies +// its bearer (Pitfall 10); a failing bearer closes the stream. +const streamRecheckInterval = time.Minute + // frameworkModule is the module path whose version names the server. const frameworkModule = "git.golem15.com/golem15/summercms" @@ -43,7 +50,9 @@ type Pier struct { dispatch *dispatcher store Store sessions *sessionRouter + stateless *statelessPath handler http.Handler + recheck time.Duration prmURL string appName string version string @@ -89,16 +98,18 @@ func Mount(host Host, app *backpack.App, plugins []party.Plugin) error { app: app, host: host, catalog: cat, - store: newMemoryStore(), appName: appName(app), version: frameworkVersion(), instance: rand.Text(), + recheck: streamRecheckInterval, } p.identities = newIdentityResolver(app, cfg, frontend) p.dispatch = &dispatcher{host: host, app: app, cfg: cfg} p.sessions = newSessionRouter(p) + p.stateless = newStatelessPath(p) + p.store = p.newStore() p.prmURL = p.metadataURL() - p.handler = p.originCheck(p.bodyLimit(p.bearer(p.sessions))) + p.handler = p.originCheck(p.bodyLimit(p.bearer(http.HandlerFunc(p.route)))) methods := []string{http.MethodPost, http.MethodGet, http.MethodDelete} for _, m := range methods { @@ -111,9 +122,46 @@ func Mount(host Host, app *backpack.App, plugins []party.Plugin) error { return err } } + if err := p.mountPRM(); err != nil { + return err + } + p.listenEvents() return nil } +// newStore builds the session store mcp.sessions.store selects: the memory +// store, subscribed to its own events at once, or a Redis store opened on +// first use from redis.url under cooler's key prefix (nothing dials at +// mount; a missing URL is a 503 on the first request, logged once). +func (p *Pier) newStore() Store { + if p.cfg.SessionStore != StoreRedis { + ms := newMemoryStore() + _ = ms.Subscribe(context.Background(), p.handleEvent) + return ms + } + url := "" + prefix := cooler.KeyPrefix(nil) + if p.app.Config != nil { + url = strings.TrimSpace(p.app.Config.String(cooler.RedisURLKey)) + prefix = cooler.KeyPrefix(p.app.Config) + } + return &lazyStore{open: func(ctx context.Context) (Store, error) { + if err := redisURLError(url); err != nil { + return nil, err + } + client, err := cooler.OpenRedis(ctx, url) + if err != nil { + return nil, err + } + rs := newRedisStore(client, prefix) + if err := rs.Subscribe(context.Background(), p.handleEvent); err != nil { + _ = client.Close() + return nil, err + } + return rs, nil + }} +} + // Handler returns the endpoint's handler (the Origin check, the body limit, // the bearer verification and the session router), for tests and hosts // that serve it outside the route table. diff --git a/modules/pier/prm.go b/modules/pier/prm.go new file mode 100644 index 00000000..b34f64b5 --- /dev/null +++ b/modules/pier/prm.go @@ -0,0 +1,59 @@ +package pier + +import ( + "net/http" + "net/url" + + "git.golem15.com/golem15/summercms/modules/pact" + "github.com/modelcontextprotocol/go-sdk/auth" + "github.com/modelcontextprotocol/go-sdk/oauthex" +) + +// PRMScopes is the scopes_supported list of the protected resource +// metadata: the OAuth server's MCP scopes plus backend (D-21). +var PRMScopes = []string{"read", "write", "ai", "offline_access", "backend"} + +// mountPRM serves the RFC 9728 protected resource metadata at +// /.well-known/oauth-protected-resource and at +// /.well-known/oauth-protected-resource, through the host's +// plain GET route kind. The authorization +// server is the published verifier's resource origin when a verifier is +// published, else app.url. A mcp.resource that differs from the verifier's +// resource is logged as a warning at boot (Pitfall 8). +func (p *Pier) mountPRM() error { + issuer := p.cfg.AppURL + if verifier, ok := p.app.Lookup[pact.MCPTokenVerifier](); ok && verifier != nil { + if res := verifier.MCPResource(); res != "" { + if u, err := url.Parse(res); err == nil && u.Scheme != "" && u.Host != "" { + issuer = u.Scheme + "://" + u.Host + } + if res != p.cfg.Resource { + logger().Warn("pier: mcp.resource differs from the verifier's resource; clients that send the verifier's resource will be refused", + "mcp_resource", p.cfg.Resource, "verifier_resource", res) + } + } + } + meta := &oauthex.ProtectedResourceMetadata{ + Resource: p.cfg.Resource, + ScopesSupported: append([]string(nil), PRMScopes...), + BearerMethodsSupported: []string{"header"}, + } + if issuer != "" { + meta.AuthorizationServers = []string{issuer} + } + h := nosniff(auth.ProtectedResourceMetadataHandler(meta)) + for _, path := range []string{wellKnownPRM, wellKnownPRM + p.cfg.Path} { + if err := p.host.MountJSON(http.MethodGet, path, h); err != nil { + return err + } + } + return nil +} + +// nosniff adds X-Content-Type-Options to a JSON document route. +func nosniff(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("X-Content-Type-Options", "nosniff") + next.ServeHTTP(w, r) + }) +} diff --git a/modules/pier/session.go b/modules/pier/session.go index 200a5484..a1b8aca7 100644 --- a/modules/pier/session.go +++ b/modules/pier/session.go @@ -10,6 +10,7 @@ import ( "io" "net/http" "slices" + "strings" "sync" "sync/atomic" "time" @@ -346,9 +347,43 @@ func (sr *sessionRouter) persist(ctx context.Context, ls *localSession) { } } -// serveStream serves the standalone GET stream. +// 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 +// every stream at shutdown. func (sr *sessionRouter) serveStream(w http.ResponseWriter, r *http.Request, ls *localSession) { - ls.transport.ServeHTTP(w, r) + ctx, cancel := context.WithCancel(r.Context()) + defer cancel() + raw := bearerOf(r) + go func() { + ticker := time.NewTicker(sr.p.recheck) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + id, err := sr.p.identities.resolve(ctx, raw) + if err != nil || id.Binding != ls.binding { + logSession(logger(), "stream_closed", ls.id, ls.binding) + cancel() + return + } + _ = sr.p.store.Touch(ctx, ls.id, sr.p.cfg.IdleTTL) + } + } + }() + ls.transport.ServeHTTP(w, r.WithContext(ctx)) +} + +// bearerOf reads the bearer of a request's Authorization header. +func bearerOf(r *http.Request) string { + fields := strings.Fields(r.Header.Get("Authorization")) + if len(fields) != 2 || !strings.EqualFold(fields[0], "bearer") { + return "" + } + return fields[1] } // closeLocal closes and forgets this instance's copy of a session. diff --git a/modules/pier/sessions_smoke_test.go b/modules/pier/sessions_smoke_test.go new file mode 100644 index 00000000..2192573a --- /dev/null +++ b/modules/pier/sessions_smoke_test.go @@ -0,0 +1,247 @@ +package pier_test + +import ( + "bufio" + "bytes" + "context" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "git.golem15.com/golem15/summercms/modules/backpack" + "git.golem15.com/golem15/summercms/modules/pact" + "git.golem15.com/golem15/summercms/modules/party" + "github.com/alicebob/miniredis/v2" + "github.com/modelcontextprotocol/go-sdk/mcp" +) + +// Smoke tests of phase 19.7 plan 02 Task 3: Redis sessions shared by two +// pier instances on one miniredis (rehydration, binding, DELETE, idle TTL, +// the per-binding cap) and the pub/sub fan-out of subject events. + +// instance is one mounted pier with its app and test server. +type instance struct { + app *backpack.App + host *testHost + srv *httptest.Server +} + +// newInstance mounts pier on a fresh app sharing the miniredis and guard. +func newInstance(t *testing.T, mr *miniredis.Miniredis, mcpYAML string, guard *acmeGuard) *instance { + t.Helper() + files := map[string]string{"mcp.yaml": mcpYAML} + if mr != nil { + files["redis.yaml"] = "url: redis://" + mr.Addr() + "/0\n" + } + app := pierAppWith(t, files, guard) + host, err := mountPier(t, app, []party.Plugin{acmePlugin{tools: []pact.MCPTool{echoTool()}}}) + if err != nil { + t.Fatal(err) + } + srv := httptest.NewServer(host) + t.Cleanup(srv.Close) + return &instance{app: app, host: host, srv: srv} +} + +// rpcReply is one raw JSON-RPC exchange: the status, the session header +// and the first message (an SSE data line or the JSON body). +type rpcReply struct { + status int + session string + body string + header http.Header +} + +// rpc posts one JSON-RPC message with a bearer and an optional session id. +func rpc(t *testing.T, ctx context.Context, url, token, session, body string) rpcReply { + t.Helper() + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewBufferString(body)) + if err != nil { + t.Fatal(err) + } + req.Header.Set("Authorization", "Bearer "+token) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json, text/event-stream") + req.Header.Set("Mcp-Protocol-Version", "2025-11-25") + if session != "" { + req.Header.Set("Mcp-Session-Id", session) + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + reply := rpcReply{status: resp.StatusCode, session: resp.Header.Get("Mcp-Session-Id"), header: resp.Header} + if strings.HasPrefix(resp.Header.Get("Content-Type"), "text/event-stream") { + scanner := bufio.NewScanner(resp.Body) + scanner.Buffer(make([]byte, 1<<20), 1<<20) + for scanner.Scan() { + line := scanner.Text() + if strings.HasPrefix(line, "data: ") { + reply.body = strings.TrimPrefix(line, "data: ") + break + } + } + return reply + } + raw, _ := io.ReadAll(resp.Body) + reply.body = strings.TrimSpace(string(raw)) + return reply +} + +const initializeBody = `{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-11-25","capabilities":{},"clientInfo":{"name":"raw","version":"1"}}}` +const initializedBody = `{"jsonrpc":"2.0","method":"notifications/initialized"}` +const callBody = `{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"acme_echo","arguments":{"id":9,"text":"hey"}}}` + +// openSession initialises a session on an instance and completes the +// handshake, returning the session id. +func openSession(t *testing.T, ctx context.Context, in *instance, token string) string { + t.Helper() + init := rpc(t, ctx, in.srv.URL+"/mcp", token, "", initializeBody) + if init.status != http.StatusOK || init.session == "" || !strings.Contains(init.body, `"protocolVersion":"2025-11-25"`) { + t.Fatalf("initialize = %d session %q body %s", init.status, init.session, init.body) + } + if done := rpc(t, ctx, in.srv.URL+"/mcp", token, init.session, initializedBody); done.status != http.StatusAccepted { + t.Fatalf("initialized = %d %s", done.status, done.body) + } + return init.session +} + +func TestPierSessionsSmoke(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + mr := miniredis.RunT(t) + cfg := "enabled: false\nsessions:\n store: redis\n idle_ttl: 30m\n max_per_binding: 2\n" + a := newInstance(t, mr, cfg, &acmeGuard{}) + b := newInstance(t, mr, cfg, &acmeGuard{}) + + // Initialise on A, call on B: B rehydrates the session from Redis. + sid := openSession(t, ctx, a, "oas_at_test") + call := rpc(t, ctx, b.srv.URL+"/mcp", "oas_at_test", sid, callBody) + if call.status != http.StatusOK || !strings.Contains(call.body, `\"principal\":7`) || strings.Contains(call.body, `"isError":true`) { + t.Fatalf("call on B = %d %s", call.status, call.body) + } + keys := mr.Keys() + if !containsPrefix(keys, "acme:mcp:s:") || !containsPrefix(keys, "acme:mcp:b:oauth:c1:7") { + t.Fatalf("redis keys = %v, want acme:mcp:s: and acme:mcp:b:oauth:c1:7", keys) + } + + // Another binding presenting the same id is refused. + other := rpc(t, ctx, b.srv.URL+"/mcp", "oas_at_other", sid, callBody) + if other.status != http.StatusForbidden || !strings.Contains(other.body, "session_binding") { + t.Fatalf("other binding = %d %s", other.status, other.body) + } + + // DELETE on B ends the session everywhere: A answers 404 next. + del, err := http.NewRequestWithContext(ctx, http.MethodDelete, b.srv.URL+"/mcp", nil) + if err != nil { + t.Fatal(err) + } + del.Header.Set("Authorization", "Bearer oas_at_test") + del.Header.Set("Mcp-Session-Id", sid) + resp, err := http.DefaultClient.Do(del) + if err != nil { + t.Fatal(err) + } + _ = resp.Body.Close() + if resp.StatusCode != http.StatusNoContent { + t.Fatalf("DELETE on B = %d", resp.StatusCode) + } + if gone := rpc(t, ctx, a.srv.URL+"/mcp", "oas_at_test", sid, callBody); gone.status != http.StatusNotFound { + t.Fatalf("call on A after DELETE = %d %s", gone.status, gone.body) + } + + // An idle session expires: after the TTL the id is unknown. + expiring := openSession(t, ctx, a, "oas_at_test") + mr.FastForward(31 * time.Minute) + if expired := rpc(t, ctx, a.srv.URL+"/mcp", "oas_at_test", expiring, callBody); expired.status != http.StatusNotFound { + t.Fatalf("call after the idle TTL = %d %s", expired.status, expired.body) + } + + // max_per_binding + 1 initialisations answer 429. + openSession(t, ctx, a, "oas_at_test") + openSession(t, ctx, b, "oas_at_test") + third := rpc(t, ctx, a.srv.URL+"/mcp", "oas_at_test", "", initializeBody) + if third.status != http.StatusTooManyRequests || !strings.Contains(third.body, "too_many_sessions") { + t.Fatalf("third session = %d %s", third.status, third.body) + } +} + +func containsPrefix(keys []string, prefix string) bool { + for _, k := range keys { + if strings.HasPrefix(k, prefix) { + return true + } + } + return false +} + +func TestPierFanOutSmoke(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + mr := miniredis.RunT(t) + guard := &acmeGuard{} + cfg := "enabled: false\nsessions:\n store: redis\n" + a := newInstance(t, mr, cfg, guard) + b := newInstance(t, mr, cfg, guard) + + changed := make(chan struct{}, 4) + client := mcp.NewClient(&mcp.Implementation{Name: "acme-test", Version: "1"}, &mcp.ClientOptions{ + ToolListChangedHandler: func(context.Context, *mcp.ToolListChangedRequest) { + select { + case changed <- struct{}{}: + default: + } + }, + }) + transport := &mcp.StreamableClientTransport{ + Endpoint: b.srv.URL + "/mcp", + HTTPClient: &http.Client{Transport: bearerTransport{token: "oas_at_test"}}, + } + cs, err := client.Connect(ctx, transport, &mcp.ClientSessionOptions{ProtocolVersion: "2025-11-25"}) + if err != nil { + t.Fatalf("connect: %v", err) + } + defer cs.Close() + list, err := cs.ListTools(ctx, nil) + if err != nil || len(list.Tools) != 1 { + t.Fatalf("tools/list = %+v, %v", list, err) + } + + // The guard now denies the tool; a permissions event fired on A reaches + // the session on B as notifications/tools/list_changed. + guard.deny.Store(true) + if err := a.app.Events.Fire(ctx, &pact.MCPSubjectChanged{Guard: "acme", UserID: 7, Kind: pact.MCPSubjectPermissions}); err != nil { + t.Fatal(err) + } + select { + case <-changed: + case <-time.After(10 * time.Second): + t.Fatal("no tools/list_changed within 10 s") + } + list, err = cs.ListTools(ctx, nil) + if err != nil || len(list.Tools) != 0 { + t.Fatalf("tools/list after the permission change = %+v, %v", list, err) + } + + // A revocation event ends the session: the next call is refused. + if err := a.app.Events.Fire(ctx, &pact.MCPSubjectChanged{Guard: "acme", UserID: 7, Kind: pact.MCPSubjectRevoked}); err != nil { + t.Fatal(err) + } + deadline := time.Now().Add(10 * time.Second) + for { + if _, err := cs.ListTools(ctx, nil); err != nil { + break + } + if time.Now().After(deadline) { + t.Fatal("the session survived the revocation event") + } + time.Sleep(50 * time.Millisecond) + } + if raw := rpc(t, ctx, b.srv.URL+"/mcp", "oas_at_test", cs.ID(), callBody); raw.status != http.StatusNotFound { + t.Fatalf("call after revocation = %d %s", raw.status, raw.body) + } +} diff --git a/modules/pier/stateless.go b/modules/pier/stateless.go new file mode 100644 index 00000000..c020ad65 --- /dev/null +++ b/modules/pier/stateless.go @@ -0,0 +1,174 @@ +package pier + +import ( + "bytes" + "container/list" + "errors" + "io" + "net/http" + "slices" + "sync" + + "github.com/modelcontextprotocol/go-sdk/mcp" +) + +// serverFeatures is what applyDiff needs from an SDK server. +type serverFeatures interface { + AddTool(t *mcp.Tool, h mcp.ToolHandler) + RemoveTools(names ...string) + AddPrompt(p *mcp.Prompt, h mcp.PromptHandler) + RemovePrompts(names ...string) +} + +// statelessEntry is one per-identity server of the stateless path, keyed +// by binding, with the names it holds for fan-out diffs. +type statelessEntry struct { + binding string + server *mcp.Server + elem *list.Element + + mu sync.Mutex + identity Identity + tools map[string]bool + prompts map[string]bool +} + +// statelessPath serves clients of protocol 2026-07-28 and later (D-32, +// P-08): the SDK's stateless handler over an in-process LRU of +// per-identity servers (capacity mcp.sessions.max_per_binding * 16), +// evicted on revocation, mutated on fan-out events so subscriptions/listen +// clients receive the list changes. Every request is authorized afresh by +// the bearer middleware in front of it. +type statelessPath struct { + p *Pier + handler http.Handler + capacity int + + mu sync.Mutex + entries map[string]*statelessEntry + order *list.List // front = most recently used +} + +// statelessCapacityFactor multiplies mcp.sessions.max_per_binding into the +// LRU capacity. +const statelessCapacityFactor = 16 + +func newStatelessPath(p *Pier) *statelessPath { + sp := &statelessPath{ + p: p, + capacity: p.cfg.MaxPerBinding * statelessCapacityFactor, + entries: map[string]*statelessEntry{}, + order: list.New(), + } + sp.handler = mcp.NewStreamableHTTPHandler(sp.getServer, &mcp.StreamableHTTPOptions{ + Stateless: true, + // pier runs behind a reverse proxy and checks Origin itself; the + // SDK's loopback rule would refuse every proxied request of a + // binary listening on 127.0.0.1. + DisableLocalhostProtection: true, + MaxRequestBodyBytes: maxRequestBody, + }) + return sp +} + +func (sp *statelessPath) ServeHTTP(w http.ResponseWriter, r *http.Request) { + sp.handler.ServeHTTP(w, r) +} + +// getServer returns the per-identity server of the request's binding, +// building it on a miss and refreshing the identity on a hit. +func (sp *statelessPath) getServer(r *http.Request) *mcp.Server { + id, ok := identityFromRequest(r) + if !ok { + return nil + } + sp.mu.Lock() + defer sp.mu.Unlock() + if e, ok := sp.entries[id.Binding]; ok { + sp.order.MoveToFront(e.elem) + e.mu.Lock() + e.identity = id + e.mu.Unlock() + return e.server + } + srv, tools, prompts := sp.p.newServer(r.Context(), id, nil) + e := &statelessEntry{binding: id.Binding, server: srv, identity: id, tools: tools, prompts: prompts} + e.elem = sp.order.PushFront(e) + sp.entries[id.Binding] = e + for sp.order.Len() > sp.capacity { + last := sp.order.Back() + if last == nil { + break + } + old := last.Value.(*statelessEntry) + sp.order.Remove(last) + delete(sp.entries, old.binding) + } + return srv +} + +// matching lists the entries whose identity includes the subject. +func (sp *statelessPath) matching(guard string, userID uint) []*statelessEntry { + sp.mu.Lock() + defer sp.mu.Unlock() + var out []*statelessEntry + for _, e := range sp.entries { + e.mu.Lock() + hit := e.identity.has(guard, userID) + e.mu.Unlock() + if hit { + out = append(out, e) + } + } + return out +} + +// evict drops the entries of a subject (a revocation or deletion) and +// closes their sessions, so a subscriptions/listen stream ends. +func (sp *statelessPath) evict(guard string, userID uint) { + for _, e := range sp.matching(guard, userID) { + sp.mu.Lock() + if cur, ok := sp.entries[e.binding]; ok && cur == e { + sp.order.Remove(e.elem) + delete(sp.entries, e.binding) + } + sp.mu.Unlock() + for ss := range e.server.Sessions() { + _ = ss.Close() + } + } +} + +// route sends a request to the stateless path when its protocol version +// header is 2026-07-28 or later, or when the first JSON-RPC method of a +// POST is server/discover; everything else goes to the stateful router +// (P-08). +func (p *Pier) route(w http.ResponseWriter, r *http.Request) { + if p.stateless == nil { + p.sessions.ServeHTTP(w, r) + return + } + if pv := r.Header.Get(protocolVersionHeader); pv >= protocolVersion202607 { + p.stateless.ServeHTTP(w, r) + return + } + if r.Method == http.MethodPost && r.Body != nil { + body, err := io.ReadAll(r.Body) + if err != nil { + var mbe *http.MaxBytesError + if errors.As(err, &mbe) { + http.Error(w, "request body too large", http.StatusRequestEntityTooLarge) + return + } + http.Error(w, "failed to read body", http.StatusBadRequest) + return + } + _ = r.Body.Close() + r.Body = io.NopCloser(bytes.NewReader(body)) + if slices.Contains(peekMethods(body), methodDiscover) { + p.stateless.ServeHTTP(w, r) + return + } + } + p.sessions.ServeHTTP(w, r) +} diff --git a/modules/pier/stateless_smoke_test.go b/modules/pier/stateless_smoke_test.go new file mode 100644 index 00000000..e62e8ca2 --- /dev/null +++ b/modules/pier/stateless_smoke_test.go @@ -0,0 +1,80 @@ +package pier_test + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + "time" + + "git.golem15.com/golem15/summercms/modules/pact" + "git.golem15.com/golem15/summercms/modules/party" + "github.com/modelcontextprotocol/go-sdk/mcp" +) + +// Smoke test of phase 19.7 plan 02 Task 3: the hybrid protocol (D-32). A +// client at the SDK's newest protocol negotiates 2026-07-28 statelessly +// (no Mcp-Session-Id) and calls the tool; a client pinned to 2025-11-25 +// gets a session on the stateful path; both see the same per-identity list. +func TestPierStatelessSmoke(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + app := pierApp(t, "enabled: false\n") + host, err := mountPier(t, app, []party.Plugin{acmePlugin{tools: []pact.MCPTool{echoTool()}}}) + if err != nil { + t.Fatal(err) + } + srv := httptest.NewServer(host) + defer srv.Close() + transport := func() *mcp.StreamableClientTransport { + return &mcp.StreamableClientTransport{ + Endpoint: srv.URL + "/mcp", + HTTPClient: &http.Client{Transport: bearerTransport{token: "oas_at_test"}}, + } + } + + newest := mcp.NewClient(&mcp.Implementation{Name: "newest", Version: "1"}, nil) + stateless, err := newest.Connect(ctx, transport(), nil) + if err != nil { + t.Fatalf("stateless connect: %v", err) + } + defer stateless.Close() + if stateless.ID() != "" { + t.Fatalf("the newest client got a session id %q, want none", stateless.ID()) + } + if v := stateless.InitializeResult().ProtocolVersion; v != "2026-07-28" { + t.Fatalf("negotiated %q, want 2026-07-28", v) + } + res, err := stateless.CallTool(ctx, &mcp.CallToolParams{Name: "acme_echo", Arguments: map[string]any{"id": 3, "text": "yo"}}) + if err != nil || res.IsError { + t.Fatalf("stateless tools/call = %+v, %v", res, err) + } + if structured, _ := res.StructuredContent.(map[string]any); structured["principal"] != 7.0 || structured["id"] != 3.0 { + t.Fatalf("stateless structuredContent = %+v", res.StructuredContent) + } + + legacy := mcp.NewClient(&mcp.Implementation{Name: "legacy", Version: "1"}, nil) + stateful, err := legacy.Connect(ctx, transport(), &mcp.ClientSessionOptions{ProtocolVersion: "2025-11-25"}) + if err != nil { + t.Fatalf("stateful connect: %v", err) + } + defer stateful.Close() + if stateful.ID() == "" { + t.Fatal("the 2025-11-25 client got no session id") + } + if v := stateful.InitializeResult().ProtocolVersion; v != "2025-11-25" { + t.Fatalf("negotiated %q, want 2025-11-25", v) + } + + a, err := stateless.ListTools(ctx, nil) + if err != nil { + t.Fatal(err) + } + b, err := stateful.ListTools(ctx, nil) + if err != nil { + t.Fatal(err) + } + if len(a.Tools) != 1 || len(b.Tools) != 1 || a.Tools[0].Name != b.Tools[0].Name || a.Tools[0].Name != "acme_echo" { + t.Fatalf("lists differ: stateless %+v, stateful %+v", a.Tools, b.Tools) + } +} diff --git a/modules/pier/store_redis.go b/modules/pier/store_redis.go new file mode 100644 index 00000000..bc03150f --- /dev/null +++ b/modules/pier/store_redis.go @@ -0,0 +1,247 @@ +package pier + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/redis/go-redis/v9" +) + +// Redis key and channel layout under cooler's application prefix. +const ( + redisSessionKey = "mcp:s:" // mcp:s: holds the JSON record + redisBindingKey = "mcp:b:" // mcp:b: is the set of its session ids + redisEventsKey = "mcp:events" // mcp:events carries JSON events +) + +// redisStore keeps session records in Redis with PX TTLs and fans events +// out through pub/sub, so several instances share sessions and react to +// each other's events. +type redisStore struct { + client *redis.Client + prefix string +} + +func newRedisStore(client *redis.Client, prefix string) *redisStore { + return &redisStore{client: client, prefix: prefix} +} + +func (s *redisStore) sessionKey(id string) string { return s.prefix + redisSessionKey + id } + +func (s *redisStore) bindingKey(binding string) string { return s.prefix + redisBindingKey + binding } + +func (s *redisStore) channel() string { return s.prefix + redisEventsKey } + +func (s *redisStore) Put(ctx context.Context, id string, rec Record, ttl time.Duration) error { + raw, err := json.Marshal(rec) + if err != nil { + return err + } + pipe := s.client.TxPipeline() + pipe.Set(ctx, s.sessionKey(id), raw, ttl) + pipe.SAdd(ctx, s.bindingKey(rec.Binding), id) + pipe.PExpire(ctx, s.bindingKey(rec.Binding), ttl) + _, err = pipe.Exec(ctx) + return err +} + +func (s *redisStore) Get(ctx context.Context, id string) (Record, bool, error) { + raw, err := s.client.Get(ctx, s.sessionKey(id)).Bytes() + if errors.Is(err, redis.Nil) { + return Record{}, false, nil + } + if err != nil { + return Record{}, false, err + } + var rec Record + if err := json.Unmarshal(raw, &rec); err != nil { + return Record{}, false, err + } + return rec, true, nil +} + +// Touch slides the record's TTL and its binding set's with it. +func (s *redisStore) Touch(ctx context.Context, id string, ttl time.Duration) error { + rec, ok, err := s.Get(ctx, id) + if err != nil || !ok { + return err + } + pipe := s.client.TxPipeline() + pipe.PExpire(ctx, s.sessionKey(id), ttl) + pipe.PExpire(ctx, s.bindingKey(rec.Binding), ttl) + _, err = pipe.Exec(ctx) + return err +} + +func (s *redisStore) Delete(ctx context.Context, id string) error { + rec, ok, err := s.Get(ctx, id) + if err != nil { + return err + } + pipe := s.client.TxPipeline() + pipe.Del(ctx, s.sessionKey(id)) + if ok { + pipe.SRem(ctx, s.bindingKey(rec.Binding), id) + } + _, err = pipe.Exec(ctx) + return err +} + +// Count answers the live sessions of a binding, pruning members whose +// record has expired from the set. +func (s *redisStore) Count(ctx context.Context, binding string) (int, error) { + members, err := s.client.SMembers(ctx, s.bindingKey(binding)).Result() + if err != nil { + return 0, err + } + n := 0 + for _, id := range members { + exists, err := s.client.Exists(ctx, s.sessionKey(id)).Result() + if err != nil { + return 0, err + } + if exists > 0 { + n++ + continue + } + _ = s.client.SRem(ctx, s.bindingKey(binding), id).Err() + } + return n, nil +} + +func (s *redisStore) Publish(ctx context.Context, ev Event) error { + raw, err := json.Marshal(ev) + if err != nil { + return err + } + return s.client.Publish(ctx, s.channel(), raw).Err() +} + +// Subscribe delivers every event published on the channel to fn, on its +// own goroutine, until ctx ends. +func (s *redisStore) Subscribe(ctx context.Context, fn func(Event)) error { + sub := s.client.Subscribe(ctx, s.channel()) + if _, err := sub.Receive(ctx); err != nil { + _ = sub.Close() + return err + } + go func() { + defer sub.Close() + ch := sub.Channel() + for { + select { + case <-ctx.Done(): + return + case msg, ok := <-ch: + if !ok { + return + } + var ev Event + if err := json.Unmarshal([]byte(msg.Payload), &ev); err != nil { + continue + } + fn(ev) + } + } + }() + return nil +} + +// lazyStore opens the Redis store on first use, so nothing dials at mount +// (route:list and tests never dial) and a missing redis.url is a 503 on the +// first request, logged once, not a boot failure (cooler's rule). +type lazyStore struct { + open func(ctx context.Context) (Store, error) + mu sync.Mutex + store Store + warned atomic.Bool +} + +// get returns the store, opening it on the first call and retrying on a +// later call after a failed open. +func (l *lazyStore) get(ctx context.Context) (Store, error) { + l.mu.Lock() + defer l.mu.Unlock() + if l.store != nil { + return l.store, nil + } + s, err := l.open(ctx) + if err != nil { + if !l.warned.Swap(true) { + logger().Error("pier: session store could not be opened", "error", err) + } + return nil, err + } + l.store = s + return s, nil +} + +func (l *lazyStore) Put(ctx context.Context, id string, rec Record, ttl time.Duration) error { + s, err := l.get(ctx) + if err != nil { + return err + } + return s.Put(ctx, id, rec, ttl) +} + +func (l *lazyStore) Get(ctx context.Context, id string) (Record, bool, error) { + s, err := l.get(ctx) + if err != nil { + return Record{}, false, err + } + return s.Get(ctx, id) +} + +func (l *lazyStore) Touch(ctx context.Context, id string, ttl time.Duration) error { + s, err := l.get(ctx) + if err != nil { + return err + } + return s.Touch(ctx, id, ttl) +} + +func (l *lazyStore) Delete(ctx context.Context, id string) error { + s, err := l.get(ctx) + if err != nil { + return err + } + return s.Delete(ctx, id) +} + +func (l *lazyStore) Count(ctx context.Context, binding string) (int, error) { + s, err := l.get(ctx) + if err != nil { + return 0, err + } + return s.Count(ctx, binding) +} + +func (l *lazyStore) Publish(ctx context.Context, ev Event) error { + s, err := l.get(ctx) + if err != nil { + return err + } + return s.Publish(ctx, ev) +} + +func (l *lazyStore) Subscribe(ctx context.Context, fn func(Event)) error { + s, err := l.get(ctx) + if err != nil { + return err + } + return s.Subscribe(ctx, fn) +} + +// redisURLError names the key a redis store needs. +func redisURLError(url string) error { + if strings.TrimSpace(url) == "" { + return fmt.Errorf("pier: redis.url is required when %s is %s (set it in config/redis.yaml or SUMMER_REDIS__URL)", SessionStoreKey, StoreRedis) + } + return nil +} diff --git a/modules/pier/tracer_smoke_test.go b/modules/pier/tracer_smoke_test.go index aabd8777..d23c22d9 100644 --- a/modules/pier/tracer_smoke_test.go +++ b/modules/pier/tracer_smoke_test.go @@ -12,6 +12,7 @@ import ( "path/filepath" "strconv" "strings" + "sync/atomic" "testing" "time" @@ -106,11 +107,14 @@ func (p acmePlugin) MCP() pact.MCP { } // acmeGuard is the fake guard provider for "acme": token-guarded by -// acme.auth, allows everything, dispatches as the subject's user id. -type acmeGuard struct{} +// acme.auth, allows everything unless deny is set, dispatches as the +// subject's user id. +type acmeGuard struct { + deny atomic.Bool +} -func (acmeGuard) MCPGuardName() string { return "acme" } -func (acmeGuard) MCPTokenGuarded(mw []string) bool { +func (*acmeGuard) MCPGuardName() string { return "acme" } +func (*acmeGuard) MCPTokenGuarded(mw []string) bool { for _, m := range mw { if m == "acme.auth" { return true @@ -118,24 +122,28 @@ func (acmeGuard) MCPTokenGuarded(mw []string) bool { } return false } -func (acmeGuard) MCPRequirements([]string) (pact.MCPRequirements, error) { +func (*acmeGuard) MCPRequirements([]string) (pact.MCPRequirements, error) { return pact.MCPRequirements{}, nil } -func (acmeGuard) MCPAllows(context.Context, pact.MCPSubject, pact.MCPRequirements, []string) (bool, error) { - return true, nil +func (g *acmeGuard) MCPAllows(context.Context, pact.MCPSubject, pact.MCPRequirements, []string) (bool, error) { + return !g.deny.Load(), nil } -func (acmeGuard) MCPDispatch(_ context.Context, subject pact.MCPSubject, _ string) (pact.MCPDispatch, error) { +func (*acmeGuard) MCPDispatch(_ context.Context, subject pact.MCPSubject, _ string) (pact.MCPDispatch, error) { return pact.MCPDispatch{Principal: &bouncer.Principal{ID: subject.UserID}, Guards: []string{"acme"}}, nil } -// acmeVerifier accepts oas_at_test for user 7, client c1, read and write. +// acmeVerifier accepts oas_at_test for user 7, client c1, read and write, +// and oas_at_other for user 8, client c2 (another binding). type acmeVerifier struct{} func (acmeVerifier) MCPVerifyToken(_ context.Context, raw string) (pact.MCPGrant, error) { - if raw != "oas_at_test" { - return pact.MCPGrant{}, pact.ErrMCPTokenInvalid + switch raw { + case "oas_at_test": + return pact.MCPGrant{UserID: 7, ClientID: "c1", TokenID: 1, Scopes: []string{"read", "write"}, ExpiresAt: time.Now().Add(time.Hour)}, nil + case "oas_at_other": + return pact.MCPGrant{UserID: 8, ClientID: "c2", TokenID: 2, Scopes: []string{"read", "write"}, ExpiresAt: time.Now().Add(time.Hour)}, nil } - return pact.MCPGrant{UserID: 7, ClientID: "c1", TokenID: 1, Scopes: []string{"read", "write"}, ExpiresAt: time.Now().Add(time.Hour)}, nil + return pact.MCPGrant{}, pact.ErrMCPTokenInvalid } func (acmeVerifier) MCPResource() string { return "http://acme.test/mcp" } @@ -196,16 +204,26 @@ func (h *testHost) ServeHTTP(w http.ResponseWriter, r *http.Request) { http.NotFound(w, r) } -// pierApp builds an app with mcp.yaml from the given body, publishes the -// fake verifier and guard registry, builds the router with the MCP flag -// off (so no tag is needed) and then enables it for pier.Mount. +// pierApp builds an app with mcp.yaml from the given body and publishes +// the fake verifier and a guard registry with a fresh acme guard. func pierApp(t *testing.T, mcpYAML string) *backpack.App { + t.Helper() + return pierAppWith(t, map[string]string{"mcp.yaml": mcpYAML}, &acmeGuard{}) +} + +// pierAppWith builds an app from extra config files (app.yaml and +// http.yaml have defaults) and publishes the fake verifier and the given +// guard provider; the router is built with the MCP flag off (so no tag is +// needed) and the flag is enabled for pier.Mount afterwards. +func pierAppWith(t *testing.T, extra map[string]string, guard *acmeGuard) *backpack.App { t.Helper() dir := t.TempDir() files := map[string]string{ "app.yaml": "name: acme\nurl: http://acme.test\n", "http.yaml": "body_limits:\n default_bytes: 1048576\n upload_bytes: 1048576\n", - "mcp.yaml": mcpYAML, + } + for name, body := range extra { + files[name] = body } for name, body := range files { if err := os.WriteFile(filepath.Join(dir, name), []byte(body), 0o644); err != nil { @@ -221,7 +239,7 @@ func pierApp(t *testing.T, mcpYAML string) *backpack.App { t.Fatal(err) } reg := pact.NewMCPGuards() - if err := reg.Register(acmeGuard{}); err != nil { + if err := reg.Register(guard); err != nil { t.Fatal(err) } if err := app.Publish(reg); err != nil { @@ -290,7 +308,9 @@ func TestPierTracerSmoke(t *testing.T) { Endpoint: srv.URL + "/mcp", HTTPClient: &http.Client{Transport: bearerTransport{token: "oas_at_test"}}, } - cs, err := client.Connect(ctx, transport, nil) + // Pinned to 2025-11-25: the stateful path with sessions (the newest + // protocol is served statelessly, see TestPierStatelessSmoke). + cs, err := client.Connect(ctx, transport, &mcp.ClientSessionOptions{ProtocolVersion: "2025-11-25"}) if err != nil { t.Fatalf("connect: %v", err) }