fix(pier): end only the revoked token's or client's MCP sessions and log the client address (B-WR-04, B-IN-04)
- pact.MCPSubjectChanged and pier.Event gain TokenID and ClientID; a narrowed revoked event closes only matching sessions and stateless servers - cabana fires the admin token revoke with the token row id - towel.WithClientIP/ClientIP; pier puts the address (surf applies http.trusted_proxies) on the resolve context and ResolveBackendToken logs it
This commit is contained in:
@@ -38,7 +38,8 @@ Use cases:
|
||||
- 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 or the session's record is gone.
|
||||
- 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 `<prefix>mcp:s:<id>` with `PX` TTLs, per-binding sets `<prefix>mcp:b:<binding>`, events on channel `<prefix>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.
|
||||
- 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.
|
||||
- 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. A server evicted from the LRU closes its sessions, and 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.
|
||||
- Protected resource metadata (RFC 9728, `pier.PRMScopes`): `GET /.well-known/oauth-protected-resource` and `GET /.well-known/oauth-protected-resource<mcp.path>` 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.
|
||||
@@ -107,7 +108,7 @@ An MCP client then connects to `https://example.com/mcp` with an OAuth access to
|
||||
| `pier.MaxDescriptionRunes` | The cap on a tool or prompt description (1024). |
|
||||
| `pier.Store` | The session registry and event channel: `Put`, `Get`, `Touch`, `Delete`, `Count`, `Publish`, `Subscribe`. |
|
||||
| `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.Event` | One instance-to-instance event: a kind (`pier.EventEnded` or a `pact.MCPSubjectChanged` kind), guard, user id, session id, publishing instance, and the optional token id and client id that narrow a revocation. |
|
||||
| `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`). |
|
||||
|
||||
@@ -137,7 +138,7 @@ Operational notes: a single instance (the demo deployment) runs on the memory st
|
||||
|
||||
- `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`.
|
||||
- `github.com/google/jsonschema-go` v0.4.3: input and output schema reflection and argument validation.
|
||||
- Framework: `backpack`, `pact`, `party`, `bouncer`, `cabana`, `compass`, `cooler`, `diary`.
|
||||
- Framework: `backpack`, `pact`, `party`, `bouncer`, `cabana`, `compass`, `cooler`, `diary`, `towel`.
|
||||
|
||||
Only this module (and surf's tagged `pier_on.go`) imports the SDK; `pact` stays free of it.
|
||||
|
||||
|
||||
@@ -18,7 +18,7 @@ func (p *Pier) listenEvents() {
|
||||
if ev == nil || ev.Kind == "" {
|
||||
return nil
|
||||
}
|
||||
event := Event{Kind: ev.Kind, Guard: ev.Guard, UserID: ev.UserID, Instance: p.instance}
|
||||
event := Event{Kind: ev.Kind, Guard: ev.Guard, UserID: ev.UserID, Instance: p.instance, TokenID: ev.TokenID, ClientID: ev.ClientID}
|
||||
if err := p.store.Publish(ctx, event); err != nil {
|
||||
p.handleEvent(event)
|
||||
}
|
||||
@@ -31,7 +31,9 @@ func (p *Pier) listenEvents() {
|
||||
// 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.
|
||||
// evicts the stateless servers. A revocation naming a TokenID or a ClientID
|
||||
// closes only the sessions that token or client opened (revokedMatch). A
|
||||
// link event also drops the link cache.
|
||||
func (p *Pier) handleEvent(ev Event) {
|
||||
ctx := context.Background()
|
||||
switch ev.Kind {
|
||||
@@ -40,13 +42,14 @@ func (p *Pier) handleEvent(ev Event) {
|
||||
p.sessions.closeLocal(ev.SessionID)
|
||||
}
|
||||
case pact.MCPSubjectRevoked, pact.MCPSubjectDeleted:
|
||||
for _, ls := range p.sessions.matching(ev.Guard, ev.UserID) {
|
||||
match := revokedMatch(ev)
|
||||
for _, ls := range p.sessions.matchingFunc(match) {
|
||||
_ = 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)
|
||||
p.stateless.evictFunc(match)
|
||||
}
|
||||
case pact.MCPSubjectLink, pact.MCPSubjectPermissions:
|
||||
if ev.Kind == pact.MCPSubjectLink {
|
||||
@@ -67,6 +70,28 @@ func (p *Pier) handleEvent(ev Event) {
|
||||
}
|
||||
}
|
||||
|
||||
// revokedMatch selects the identities a revoked or deleted event ends
|
||||
// (P-31). A revoked event with a TokenID matches the identities whose
|
||||
// subject under ev.Guard is ev.UserID with that TokenID and no ClientID (the
|
||||
// sessions of one personal token); one with a ClientID matches the subject
|
||||
// with that ClientID (one OAuth client); otherwise, and for every deleted
|
||||
// event, any identity that includes the subject.
|
||||
func revokedMatch(ev Event) func(Identity) bool {
|
||||
if ev.Kind == pact.MCPSubjectRevoked && (ev.TokenID != 0 || ev.ClientID != "") {
|
||||
return func(id Identity) bool {
|
||||
s := id.subjectFor(ev.Guard)
|
||||
if s == nil || s.UserID != ev.UserID {
|
||||
return false
|
||||
}
|
||||
if ev.TokenID != 0 {
|
||||
return s.TokenID == ev.TokenID && s.ClientID == ""
|
||||
}
|
||||
return s.ClientID == ev.ClientID
|
||||
}
|
||||
}
|
||||
return func(id Identity) bool { return id.has(ev.Guard, ev.UserID) }
|
||||
}
|
||||
|
||||
// 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.
|
||||
@@ -158,12 +183,17 @@ func applyDiff(srv serverFeatures, p *Pier, id Identity, tools []*tool, prompts
|
||||
|
||||
// matching lists the local sessions whose identity includes the subject.
|
||||
func (sr *sessionRouter) matching(guard string, userID uint) []*localSession {
|
||||
return sr.matchingFunc(func(id Identity) bool { return id.has(guard, userID) })
|
||||
}
|
||||
|
||||
// matchingFunc lists the local sessions whose identity satisfies match.
|
||||
func (sr *sessionRouter) matchingFunc(match func(Identity) bool) []*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)
|
||||
hit := match(ls.identity)
|
||||
ls.mu.Unlock()
|
||||
if hit {
|
||||
out = append(out, ls)
|
||||
|
||||
@@ -18,6 +18,14 @@ type Host interface {
|
||||
Owner(method, path string) (string, bool)
|
||||
}
|
||||
|
||||
// clientIPHost is optional for a Host: a host that knows the application's
|
||||
// trusted proxies answers the client address of a request, which pier puts
|
||||
// on the token resolvers' context (towel.WithClientIP). Without it pier uses
|
||||
// the host part of the request's RemoteAddr.
|
||||
type clientIPHost interface {
|
||||
ClientIP(*http.Request) string
|
||||
}
|
||||
|
||||
// RouteInfo is one registered route as the Host reports it: the HTTP method,
|
||||
// the pattern as registered (with {param} segments), the plugin that
|
||||
// registered it and its middleware names in order.
|
||||
|
||||
@@ -12,6 +12,7 @@ import (
|
||||
"crypto/rand"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"runtime/debug"
|
||||
@@ -22,6 +23,7 @@ import (
|
||||
"git.golem15.com/golem15/summercms/modules/backpack"
|
||||
"git.golem15.com/golem15/summercms/modules/cooler"
|
||||
"git.golem15.com/golem15/summercms/modules/party"
|
||||
"git.golem15.com/golem15/summercms/modules/towel"
|
||||
"github.com/modelcontextprotocol/go-sdk/auth"
|
||||
)
|
||||
|
||||
@@ -249,6 +251,7 @@ func (p *Pier) bodyLimit(next http.Handler) http.Handler {
|
||||
// protected resource metadata URL and the read scope.
|
||||
func (p *Pier) bearer(next http.Handler) http.Handler {
|
||||
verify := func(ctx context.Context, token string, r *http.Request) (*auth.TokenInfo, error) {
|
||||
ctx = towel.WithClientIP(ctx, p.clientIP(r))
|
||||
id, err := p.identities.resolve(ctx, token)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%w: %v", auth.ErrInvalidToken, err)
|
||||
@@ -269,6 +272,19 @@ func (p *Pier) bearer(next http.Handler) http.Handler {
|
||||
})
|
||||
}
|
||||
|
||||
// clientIP is the request's client address: the host's answer when it
|
||||
// implements clientIPHost (surf applies http.trusted_proxies), otherwise the
|
||||
// host part of RemoteAddr, or RemoteAddr itself when it has no port.
|
||||
func (p *Pier) clientIP(r *http.Request) string {
|
||||
if h, ok := p.host.(clientIPHost); ok {
|
||||
return h.ClientIP(r)
|
||||
}
|
||||
if host, _, err := net.SplitHostPort(r.RemoteAddr); err == nil {
|
||||
return host
|
||||
}
|
||||
return r.RemoteAddr
|
||||
}
|
||||
|
||||
// challengeWriter completes the 401 challenge with scope="read", which the
|
||||
// SDK middleware would otherwise enforce on every request.
|
||||
type challengeWriter struct {
|
||||
|
||||
@@ -198,12 +198,17 @@ func closeSessions(srv *mcp.Server) {
|
||||
|
||||
// matching lists the entries whose identity includes the subject.
|
||||
func (sp *statelessPath) matching(guard string, userID uint) []*statelessEntry {
|
||||
return sp.matchingFunc(func(id Identity) bool { return id.has(guard, userID) })
|
||||
}
|
||||
|
||||
// matchingFunc lists the entries whose identity satisfies match.
|
||||
func (sp *statelessPath) matchingFunc(match func(Identity) bool) []*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)
|
||||
hit := match(e.identity)
|
||||
e.mu.Unlock()
|
||||
if hit {
|
||||
out = append(out, e)
|
||||
@@ -215,7 +220,13 @@ func (sp *statelessPath) matching(guard string, userID uint) []*statelessEntry {
|
||||
// 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.evictFunc(func(id Identity) bool { return id.has(guard, userID) })
|
||||
}
|
||||
|
||||
// evictFunc drops the entries whose identity satisfies match and closes
|
||||
// their sessions.
|
||||
func (sp *statelessPath) evictFunc(match func(Identity) bool) {
|
||||
for _, e := range sp.matchingFunc(match) {
|
||||
sp.mu.Lock()
|
||||
if cur, ok := sp.entries[e.binding]; ok && cur == e {
|
||||
sp.order.Remove(e.elem)
|
||||
|
||||
@@ -25,6 +25,8 @@ type Event struct {
|
||||
UserID uint `json:"user_id,omitempty"`
|
||||
SessionID string `json:"session_id,omitempty"`
|
||||
Instance string `json:"instance,omitempty"`
|
||||
TokenID uint `json:"token_id,omitempty"`
|
||||
ClientID string `json:"client_id,omitempty"`
|
||||
}
|
||||
|
||||
// EventEnded is the kind of a session-ended event.
|
||||
|
||||
Reference in New Issue
Block a user