diff --git a/docs/api.md b/docs/api.md index 5719043f4..1bd9f91c5 100644 --- a/docs/api.md +++ b/docs/api.md @@ -89,7 +89,7 @@ REST API at `/api/v1/`. Key endpoints: **Authorization is decided when a request is admitted, not re-checked at the write** (BUG-3804, ruled as policy). `RequireWorkspaceAccess` reads the caller's membership, role and guest grants once, before the handler runs. A handler's own permission and visibility checks (`requireEditPermission`, `requireItemVisible`, owner-only gates) likewise run before the store transaction that writes. As a rule, none of these is re-read inside that transaction. A few writes do re-check inside it: some app-install writes re-check the owner (`requireWorkspaceOwnerTx`), and credential mints re-check that the user is still active (`requireActiveUserTx`). Those are local hardening, not the contract. So a request already admitted when its caller is demoted, removed, restricted from a collection or loses a grant still completes, and the change applies from the next request. The window is one request's duration. Long-lived connections are the exception, because their "one request" has no end: they re-check the principal (membership, item visibility) and the credential on each revalidation tick (next paragraph). The policy covers the CALLER'S authority only. The data a write decides on (a referent's liveness, a unique holder, a guard's count, the body a patch is computed from) is not covered: a write that acts on a stale read of it is a bug, as BUG-3811 through BUG-3817 are. -**Every long-lived connection ends when the CREDENTIAL that opened it stops being valid** — both SSE endpoints and the `/api/v1/collab/{itemID}` WebSocket, checked on each connection's existing revalidation tick (BUG-3007). The invariant is deliberately wider than "the session was destroyed": a revoked PAT ends a stream too, because a revoked PAT still streaming is the same defect for a CLI or MCP caller that a destroyed session is for a browser. Before this, all three kept running indefinitely — measured at 150-180s past a logout, and 64-120s past a revocation, in both cases with `/auth/me` on that credential answering `401`. The pre-existing per-tick checks are about the PRINCIPAL (is this user still a member, is the item still visible) and none of them changes when a credential dies, which is why the two are separate questions rather than one. A request carrying NO credential AND no resolved principal — the fresh-install window, the legacy no-auth path — is unaffected: there is nothing to invalidate. A request that carries a resolved USER but no record of how the credential was established is CLOSED, not exempted: the middleware stamps that on every accept point, so its absence means an accept point recorded who without recording what, and a connection nobody can re-check is the same defect under a new name. (Both MCP accept points are in that state by design today — the PAT one records `api_token`, so it would be re-checked; the OAuth one has no liveness door for an opaque token yet, so it fails closed. Neither door is consulted yet: the MCP transport's own streams (a GET stream and `subscriptions/listen`, BUG-3835) ARE long-lived and do NOT yet re-check their credential, so a revoked PAT or OAuth token keeps one open until it ends. The keepalive (below) also keeps an idle stream past a proxy's idle timeout, which used to cut it and force a re-authenticated reconnect, so until BUG-3837 such a stream stays open as long as its client reads.) A store ERROR is not a revocation and keeps the connection until the next tick, so a database blip does not disconnect the fleet. +**Every long-lived connection ends when the CREDENTIAL that opened it stops being valid** — both SSE endpoints, the `/api/v1/collab/{itemID}` WebSocket, and the MCP transport's long-lived streams (a GET stream and a `subscriptions/listen` POST, on `/mcp` and `/mcp/chatgpt`), checked on each connection's revalidation tick (BUG-3007; MCP since BUG-3837, every ~60s, the `/events/stream` cadence). The invariant is deliberately wider than "the session was destroyed": a revoked PAT ends a stream too, because a revoked PAT still streaming is the same defect for a CLI or MCP caller that a destroyed session is for a browser. Before BUG-3007, the two SSE endpoints and the collab WebSocket kept running indefinitely — measured at 150-180s past a logout, and 64-120s past a revocation, in both cases with `/auth/me` on that credential answering `401`. The pre-existing per-tick checks are about the PRINCIPAL (is this user still a member, is the item still visible) and none of them changes when a credential dies, which is why the two are separate questions rather than one. A request carrying NO credential AND no resolved principal — the fresh-install window, the legacy no-auth path — is unaffected: there is nothing to invalidate. A request that carries a resolved USER but no record of how the credential was established is CLOSED, not exempted: the middleware stamps that on every accept point, so its absence means an accept point recorded who without recording what, and a connection nobody can re-check is the same defect under a new name. (Both MCP accept points record their kind: a PAT is re-checked as `api_token`, and an OAuth access token is re-introspected, so a revoked token, a revoked grant (a Connected Apps revoke) or an expired token ends the stream. An OAuth stream therefore ends when its access token expires, and also when the client refreshes, since a refresh rotation revokes the old access token; the new token was never presented on that stream, so the client must reconnect (and re-issue a `subscriptions/listen`) to keep receiving notifications. Nothing on the stream says why it ended: a GET stream just closes, and a listen ends with its ordinary (empty) result, so a client that wants notifications re-listens whenever a listen returns. An MCP stream also ends when its user is deleted or disabled. No revocation kicks an MCP stream at once: each ends at its next tick, so within about one interval.) A store ERROR is not a revocation and keeps the connection until the next tick, so a database blip does not disconnect the fleet. Both SSE endpoints share one admission budget, enforced **per instance**: `PAD_SSE_MAX_CONNECTIONS` (default 1000) and `PAD_SSE_MAX_PER_USER` (default 50) cover both, and the MCP transport's long-lived streams too (BUG-3835: a GET stream and a `subscriptions/listen` POST, on either MCP mount, `/mcp` or `/mcp/chatgpt`. Only a POST whose `Mcp-Protocol-Version` header is a modern version the transport serves is read to find a listen, since mcp-go serves a listen on no other; any other POST reaches the transport, so a client probing an unserved version gets its unsupported-version `400` and falls back. Pad serves only the legacy MCP protocol versions today, so no listen is counted and no POST body is read for admission. A tool call that the transport upgrades to SSE mid-call is not counted, since a refusal there would land after the call's side effects). `PAD_SSE_MAX_PER_WORKSPACE` (default 100) covers `/api/v1/events` only. Over the limit is `429` with code `sse_limit_exceeded` and a `Retry-After` header, on an MCP stream before the transport runs — clients must back off, not retry immediately (BUG-2726). The CLI does; browsers cannot, since `EventSource` exposes neither the status nor the header to the page (BUG-2733). On `/api/v1/events` only, callers with no resolved user (a legacy workspace token, or the fresh-install window before the first admin exists) are bounded per *workspace* instead of per user; `/api/v1/events/stream` has no such case — it requires a resolved user and answers `401` otherwise. diff --git a/internal/mcp/dispatch_http.go b/internal/mcp/dispatch_http.go index 707174cd3..01602b3be 100644 --- a/internal/mcp/dispatch_http.go +++ b/internal/mcp/dispatch_http.go @@ -729,12 +729,11 @@ func buildHTTPRequest(ctx context.Context, method, urlPath string, body []byte, // BUG-3007, codex round 3: that makes this request a poor fit for the // long-lived-connection liveness predicate, which re-reads the CREDENTIAL // from the request. The inbound /mcp request's `ctxAuthKind` survives into - // this context, so a synthesized request built from the MCP PAT accept - // point says `api_token` and carries no bearer to re-check — the predicate - // would close it on its first tick even though the PAT is fine. Nothing - // reaches that today: `routeTable` maps no tool to /api/v1/events, - // /api/v1/events/stream or /api/v1/collab, and a request with no auth kind - // at all is closed too, so both readings end the same way. + // this context, so a synthesized request says `api_token` or (since + // BUG-3837) `oauth_access_token`, and carries no bearer to re-check — the + // predicate would close it on its first tick even though the credential is + // fine. Nothing reaches that today: `routeTable` maps no tool to + // /api/v1/events, /api/v1/events/stream or /api/v1/collab. // // The fix when an MCP route DOES become long-lived is to forward the // credential rather than only the principal — not to widen the predicate, diff --git a/internal/server/bug3837_mcp_credential_tick_test.go b/internal/server/bug3837_mcp_credential_tick_test.go new file mode 100644 index 000000000..d1cf8d750 --- /dev/null +++ b/internal/server/bug3837_mcp_credential_tick_test.go @@ -0,0 +1,477 @@ +package server + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/PerpetualSoftware/pad/internal/models" + "github.com/PerpetualSoftware/pad/internal/oauth" + "github.com/mark3labs/mcp-go/mcp" +) + +// BUG-3837: a long-lived MCP stream ends with its credential. The transport +// here is a stub that commits an event stream and then holds it until its +// context ends, which is how mcp-go's GET and listen loops leave. + +type bug3837Fixture struct { + srv *Server + user *models.User + pat *models.APITokenWithSecret + access string + refresh string + clientID string + reqID chan string // OAuth request ids seen by the transport + // emit makes the transport write notification frames until its context + // ends, one Write per frame, as mcp-go's writeSSEEvent does. + emit atomic.Bool +} + +func newBug3837Fixture(t *testing.T, accessLifespan time.Duration) *bug3837Fixture { + t.Helper() + srv := testServer(t) + srv.SetCloudMode("test-secret") + srv.mcpCredentialRevalOverride = 5 * time.Millisecond + f := &bug3837Fixture{srv: srv, reqID: make(chan string, 16)} + srv.SetMCPTransport(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if kind, ref := MCPTokenIdentityFromContext(r.Context()); kind == "oauth" { + f.reqID <- ref + } + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + for f.emit.Load() { + select { + case <-r.Context().Done(): + return + case <-time.After(time.Millisecond): + } + fmt.Fprintf(w, "event: message\ndata: %s\n\n", `{"jsonrpc":"2.0","method":"notifications/message"}`) + w.(http.Flusher).Flush() + } + <-r.Context().Done() + }), testCanonicalAudience, testAuthServerURL, nil) + o, err := oauth.NewServer(oauth.Config{ + Store: srv.store, + HMACSecret: bytes32ForTest(), + AllowedAudience: testCanonicalAudience, + AccessTokenLifespan: accessLifespan, + }) + if err != nil { + t.Fatal(err) + } + srv.SetOAuthServer(o) + + user, sessionToken := loginTestUser(t, srv) + f.user = user + ws, err := srv.store.CreateWorkspace(models.WorkspaceCreate{Name: "BUG-3837"}) + if err != nil { + t.Fatal(err) + } + f.pat, err = srv.store.CreateAPIToken(user.ID, models.APITokenCreate{Name: "bug3837", WorkspaceID: ws.ID}, 30, 0) + if err != nil { + t.Fatal(err) + } + csrfTok := readCSRFFromCookie(t, srv, sessionToken) + clientID := registerTestClient(t, srv, "https://app.test/cb") + // Minted last, so a short lifespan is not spent on the setup above + // (hunter r3). + tokens := runAuthCodeFlow(t, srv, sessionToken, csrfTok, clientID, + "verifier-bug3837-quick-brown-fox-jumps-over-lazy-dog-12") + f.clientID = clientID + f.access, _ = tokens["access_token"].(string) + f.refresh, _ = tokens["refresh_token"].(string) + if f.access == "" || f.refresh == "" { + t.Fatalf("missing access or refresh token: %v", tokens) + } + return f +} + +// open starts a long-lived MCP request (a GET stream, or a listen POST) and +// returns a channel closed when its handler returns. +func (f *bug3837Fixture) open(t *testing.T, bearer string, listen bool) <-chan struct{} { + t.Helper() + return f.openAs(t, "Bearer", bearer, listen) +} + +func (f *bug3837Fixture) openWithScheme(t *testing.T, scheme, bearer string) <-chan struct{} { + t.Helper() + return f.openAs(t, scheme, bearer, false) +} + +func (f *bug3837Fixture) openAs(t *testing.T, scheme, bearer string, listen bool) <-chan struct{} { + t.Helper() + done, _ := f.openRecorded(t, scheme, bearer, listen) + return done +} + +// openRecorded is openAs, and also returns the response recorder. Read its +// body only after done is closed. +func (f *bug3837Fixture) openRecorded(t *testing.T, scheme, bearer string, listen bool) (<-chan struct{}, *deadlineRecorder) { + t.Helper() + rec := newDeadlineRecorder() + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + req := httptest.NewRequest("GET", "/mcp", nil) + if listen { + // A modern-era listen, the only kind admission counts (BUG-3835 r6). + req = httptest.NewRequest("POST", "/mcp", + strings.NewReader(`{"jsonrpc":"2.0","id":1,"method":"subscriptions/listen","params":{"_meta":{"`+ + mcp.MetaKeyProtocolVersion+`":"`+mcp.ProtocolVersion20260728+`"}}}`)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json, text/event-stream") + req.Header.Set(mcp.HeaderProtocolVersion, mcp.ProtocolVersion20260728) + } + req.Header.Set("Authorization", scheme+" "+bearer) + req.RemoteAddr = "192.0.2.1:1234" + req = req.WithContext(ctx) + done := make(chan struct{}) + go func() { + defer close(done) + f.srv.ServeHTTP(rec, req) + }() + deadline := time.Now().Add(5 * time.Second) + for f.srv.admission().heldTotal() == 0 { + select { + case <-done: + t.Fatal("the stream ended before it took a slot") + default: + } + if time.Now().After(deadline) { + t.Fatal("the stream never took a slot") + } + time.Sleep(time.Millisecond) + } + return done, rec +} + +func (f *bug3837Fixture) assertEnds(t *testing.T, done <-chan struct{}, why string) { + t.Helper() + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatalf("%s: the stream is still open", why) + } + if held := f.srv.admission().heldTotal(); held != 0 { + t.Fatalf("%s: slots held after the stream ended = %d, want 0", why, held) + } +} + +func (f *bug3837Fixture) assertStaysOpen(t *testing.T, done <-chan struct{}, ticks int) { + t.Helper() + select { + case <-done: + t.Fatalf("a live credential's stream ended within %d ticks", ticks) + case <-time.After(time.Duration(ticks) * f.srv.mcpCredentialRevalOverride): + } +} + +func TestMCP_BUG3837_RevokedPATEndsGETAndListen(t *testing.T) { + for _, listen := range []bool{false, true} { + name := map[bool]string{false: "GET", true: "listen"}[listen] + t.Run(name, func(t *testing.T) { + f := newBug3837Fixture(t, 0) + done := f.open(t, f.pat.Token, listen) + f.assertStaysOpen(t, done, 20) + if err := f.srv.store.DeleteAPITokenScoped(f.pat.ID, f.pat.WorkspaceID); err != nil { + t.Fatal(err) + } + f.assertEnds(t, done, "PAT revoked") + }) + } +} + +func TestMCP_BUG3837_RevokedOAuthGrantEndsGETAndListen(t *testing.T) { + for _, listen := range []bool{false, true} { + name := map[bool]string{false: "GET", true: "listen"}[listen] + t.Run(name, func(t *testing.T) { + f := newBug3837Fixture(t, 0) + done := f.open(t, f.access, listen) + // A live OAuth stream survives its ticks: without the OAuth + // door it reached the no-kind branch and closed on the first. + f.assertStaysOpen(t, done, 20) + if err := f.srv.store.RevokeOAuthGrant(<-f.reqID); err != nil { + t.Fatal(err) + } + f.assertEnds(t, done, "OAuth grant revoked") + }) + } +} + +// The lifespan leaves the fixture's setup seconds of margin before the token +// expires (hunter r1); the stream must be admitted on a live token. +func TestMCP_BUG3837_ExpiredOAuthTokenEndsTheStream(t *testing.T) { + f := newBug3837Fixture(t, 5*time.Second) + done := f.open(t, f.access, false) + f.assertStaysOpen(t, done, 20) + f.assertEnds(t, done, "OAuth access token expired") +} + +// OAuth, because the PAT door already refuses a disabled user's token and +// fosite's introspection does not look at the user at all. +func TestMCP_BUG3837_DisabledUserEndsTheStream(t *testing.T) { + f := newBug3837Fixture(t, 0) + done := f.open(t, f.access, false) + f.assertStaysOpen(t, done, 20) + // Disabled directly, with the credential left live, so only the user + // re-check can end the stream. + if _, err := f.srv.store.DB().Exec(`UPDATE users SET disabled_at = ? WHERE id = ?`, + time.Now().UTC().Format(time.RFC3339), f.user.ID); err != nil { + t.Fatal(err) + } + f.assertEnds(t, done, "user disabled") +} + +// A tick that cannot ask (a storage failure under fosite, which it wraps in +// the same outer error as a revoked token) keeps the stream, as the session +// and PAT doors do: closing every OAuth stream on a database blip would be an +// outage, not a revocation. +func TestMCP_BUG3837_OAuthStoreFailureKeepsTheStream(t *testing.T) { + f := newBug3837Fixture(t, 0) + done := f.open(t, f.access, false) + exec := func(q string) { + t.Helper() + if _, err := f.srv.store.DB().Exec(q); err != nil { + t.Fatal(err) + } + } + exec(`ALTER TABLE oauth_access_tokens RENAME TO oauth_access_tokens_bug3837`) + f.assertStaysOpen(t, done, 20) + exec(`ALTER TABLE oauth_access_tokens_bug3837 RENAME TO oauth_access_tokens`) + if err := f.srv.store.RevokeOAuthGrant(<-f.reqID); err != nil { + t.Fatal(err) + } + f.assertEnds(t, done, "OAuth grant revoked after the store recovered") +} + +// A deleted token row (fosite ErrNotFound) ends the stream. +func TestMCP_BUG3837_DeletedOAuthTokenEndsTheStream(t *testing.T) { + f := newBug3837Fixture(t, 0) + done := f.open(t, f.access, false) + f.assertStaysOpen(t, done, 20) + if _, err := f.srv.store.DB().Exec(`DELETE FROM oauth_access_tokens WHERE request_id = ?`, <-f.reqID); err != nil { + t.Fatal(err) + } + f.assertEnds(t, done, "OAuth access token row deleted") +} + +// The bearer scheme is case-insensitive (RFC 6750), and admission accepts +// "bearer ". The tick must read the token the same way (hunter r1). A +// case-sensitive read fails one of two ways: the PAT door finds no token and +// closes a live stream, and fosite rejects the whole header as a malformed +// token, which is "could not ask", so a REVOKED OAuth stream stays open. Each +// subtest therefore checks both halves: open across ticks, then ended by a +// revocation. +func TestMCP_BUG3837_LowercaseBearerIsReadLikeAdmission(t *testing.T) { + t.Run("pat", func(t *testing.T) { + f := newBug3837Fixture(t, 0) + done := f.openWithScheme(t, "bearer", f.pat.Token) + f.assertStaysOpen(t, done, 20) + if err := f.srv.store.DeleteAPITokenScoped(f.pat.ID, f.pat.WorkspaceID); err != nil { + t.Fatal(err) + } + f.assertEnds(t, done, "PAT revoked") + }) + t.Run("oauth", func(t *testing.T) { + f := newBug3837Fixture(t, 0) + done := f.openWithScheme(t, "bearer", f.access) + f.assertStaysOpen(t, done, 20) + if err := f.srv.store.RevokeOAuthGrant(<-f.reqID); err != nil { + t.Fatal(err) + } + f.assertEnds(t, done, "OAuth grant revoked") + }) +} + +// A deleted user ends the stream, not "could not ask": GetUser reports a +// missing row as (nil, nil), and the tick reads nil as gone. Driven directly, +// because deleting a user cascades to its tokens, so the credential door +// would end the stream first and hide this branch. +func TestMCP_BUG3837_DeletedUserIsNotStillValid(t *testing.T) { + f := newBug3837Fixture(t, 0) + if u, err := f.srv.store.GetUser("no-such-user"); err != nil || u != nil { + t.Fatalf("GetUser(missing) = %v, %v; want nil, nil (the premise)", u, err) + } + req := httptest.NewRequest("GET", "/mcp", nil) + req.Header.Set("Authorization", "Bearer "+f.pat.Token) + ctx := WithCurrentUser(req.Context(), &models.User{ID: "no-such-user"}) + ctx = context.WithValue(ctx, ctxAuthKind, authKindAPIToken) + req = req.WithContext(ctx) + if !f.srv.streamCredentialStillValid(req) { + t.Fatal("instrument: the PAT itself must still be live") + } + if f.srv.mcpStreamPrincipalStillValid(req) { + t.Fatal("a stream whose user no longer exists is still valid") + } +} + +// The lead's KEEP ruling on rotation (on BUG-3837): a refresh rotation flips +// the old access token inactive, so a live stream opened with it ends at its +// next tick, even though the grant is live. Pinned with its three conditions: +// it ends at about the next tick after the rotation, it ends between frames +// (the tick only cancels the context, and the transport sees that between +// writes), and the rotated-in token opens a fresh stream that stays open. +func TestMCP_BUG3837_RefreshRotationEndsTheStreamAndTheNewTokenOpensOne(t *testing.T) { + f := newBug3837Fixture(t, 0) + // A tick long enough to measure "within one" against, short enough to run. + const tick = 400 * time.Millisecond + f.srv.mcpCredentialRevalOverride = tick + f.emit.Store(true) + done, rec := f.openRecorded(t, "Bearer", f.access, false) + // Past the jittered first tick, so the next one is at most a tick away. + f.assertStaysOpen(t, done, 2) + + code, body := bug3678Refresh(f.srv, f.clientID, f.refresh) + rotated := time.Now() + if code != http.StatusOK { + t.Fatalf("refresh: status %d: %v", code, body) + } + newAccess, _ := body["access_token"].(string) + if newAccess == "" || newAccess == f.access { + t.Fatalf("refresh minted no new access token: %v", body) + } + f.assertEnds(t, done, "access token rotated out by a refresh") + // The next tick is at most one interval away; the bound is two plus + // slack, so it catches the override being ignored (the 60s production + // interval), not scheduling noise under -race (hunter r3). + if took := time.Since(rotated); took > 2*tick+2*time.Second { + t.Fatalf("the stream ended %v after the rotation, want within about one tick (%v)", took, tick) + } + + // The transport writes whole frames, one Write each, so these checks hold + // as long as ending the stream never writes to it: the tick only cancels + // the context. They are red on an end that writes to the stream (M12). + frames := rec.Body.String() + if n := strings.Count(frames, "event: message\n"); n == 0 { + t.Fatal("instrument: the transport wrote no frames before the stream ended") + } + if !strings.HasSuffix(frames, "\n\n") { + t.Fatalf("the stream ended mid-frame: body ends %q", frames[max(0, len(frames)-80):]) + } + for _, frame := range strings.Split(strings.TrimSuffix(frames, "\n\n"), "\n\n") { + if frame != `event: message`+"\n"+`data: {"jsonrpc":"2.0","method":"notifications/message"}` && + !strings.HasPrefix(frame, ":") { + t.Fatalf("a half-written or foreign frame on the stream: %q", frame) + } + } + + f.emit.Store(false) + fresh := f.open(t, newAccess, false) + f.assertStaysOpen(t, fresh, 3) +} + +// The tick runs only on the long-lived doors (where admitMCPStream took a +// slot). A tool call is an ordinary request: its credential was checked at +// admission, and it ends with its response, so a revocation mid-call must not +// cancel it (hunter r3). +func TestMCP_BUG3837_ToolCallIsNotTicked(t *testing.T) { + f := newBug3837Fixture(t, 0) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + req := httptest.NewRequest("POST", "/mcp", + strings.NewReader(`{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}`)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json, text/event-stream") + req.Header.Set("Authorization", "Bearer "+f.pat.Token) + req.RemoteAddr = "192.0.2.1:1234" + req = req.WithContext(ctx) + done := make(chan struct{}) + go func() { + defer close(done) + f.srv.ServeHTTP(newDeadlineRecorder(), req) + }() + f.assertStaysOpen(t, done, 5) + if held := f.srv.admission().heldTotal(); held != 0 { + t.Fatalf("instrument: a tool call took a stream slot (%d)", held) + } + if err := f.srv.store.DeleteAPITokenScoped(f.pat.ID, f.pat.WorkspaceID); err != nil { + t.Fatal(err) + } + f.assertStaysOpen(t, done, 20) + cancel() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("the tool call did not end with its client") + } +} + +// A user re-check that cannot ask (a storage failure reading the user) keeps +// the stream, as a credential door's store error does. OAuth, because fosite's +// introspection does not read the users table and the PAT door joins it. +func TestMCP_BUG3837_UserReadFailureKeepsTheStream(t *testing.T) { + f := newBug3837Fixture(t, 0) + done := f.open(t, f.access, false) + exec := func(q string) { + t.Helper() + if _, err := f.srv.store.DB().Exec(q); err != nil { + t.Fatal(err) + } + } + exec(`ALTER TABLE users RENAME TO users_bug3837`) + if _, err := f.srv.store.GetUser(f.user.ID); err == nil { + exec(`ALTER TABLE users_bug3837 RENAME TO users`) + t.Fatal("instrument: GetUser still succeeds with the users table renamed") + } + f.assertStaysOpen(t, done, 20) + exec(`ALTER TABLE users_bug3837 RENAME TO users`) + if err := f.srv.store.RevokeOAuthGrant(<-f.reqID); err != nil { + t.Fatal(err) + } + f.assertEnds(t, done, "OAuth grant revoked after the store recovered") +} + +// The OAuth door classifies by the error fosite returns, and fosite's +// introspection tries the access token, then falls back to reading it as a +// refresh token. In fosite v0.49.0 the fallback's error is scoped to its +// else-if (handler/oauth2/introspector.go), so the ACCESS lookup's error is +// the one returned: a store failure stays "could not ask" even when the +// fallback answers not found. The r4 decline rests on that (lead ruling), and +// a fosite bump could flip it silently, so it is pinned against real fosite. +// The control is the same fallback after a definitive access answer. +func TestMCP_BUG3837_TransientAccessErrorIsNotReadAsRevocation(t *testing.T) { + f := newBug3837Fixture(t, 0) + ctx := context.Background() + exec := func(q string) { + t.Helper() + if _, err := f.srv.store.DB().Exec(q); err != nil { + t.Fatal(err) + } + } + // The fallback reads the token with the strategy fosite uses, and must + // find nothing, in both halves. + refreshFindsNothing := func(why string) { + t.Helper() + owner, err := f.srv.oauthServer.GrantOwner(ctx, "refresh_token", f.access) + if err != nil || owner != "" { + t.Fatalf("instrument (%s): the refresh lookup of the access token = (%q, %v), want not found", why, owner, err) + } + } + if got := f.srv.oauthLiveness(ctx, f.access); got != credentialValid { + t.Fatalf("live token = %v, want valid", got) + } + + exec(`ALTER TABLE oauth_access_tokens RENAME TO oauth_access_tokens_bug3837`) + if _, err := f.srv.store.DB().Exec(`SELECT 1 FROM oauth_access_tokens`); err == nil { + t.Fatal("instrument: the access lookup still succeeds with its table renamed") + } + refreshFindsNothing("access store failing") + got := f.srv.oauthLiveness(ctx, f.access) + exec(`ALTER TABLE oauth_access_tokens_bug3837 RENAME TO oauth_access_tokens`) + if got != credentialUnknown { + t.Fatalf("access store failing, refresh fallback not found = %v, want unknown (the stream is kept)", got) + } + + // Control: the access row is gone, so the access lookup itself answers + // not found, and the same fallback ends the stream. + exec(`DELETE FROM oauth_access_tokens`) + refreshFindsNothing("access row deleted") + if got := f.srv.oauthLiveness(ctx, f.access); got != credentialInvalid { + t.Fatalf("access row deleted, refresh fallback not found = %v, want invalid (the stream ends)", got) + } +} diff --git a/internal/server/handlers_mcp.go b/internal/server/handlers_mcp.go index 05c5a3fb1..508bae969 100644 --- a/internal/server/handlers_mcp.go +++ b/internal/server/handlers_mcp.go @@ -276,6 +276,10 @@ func (s *Server) mcpBoundWrites(next http.Handler) http.Handler { case <-ctx.Done(): } }() + if release != nil { + // A long-lived door ends with its credential (BUG-3837). + s.startMCPCredentialTick(ctx, r, cancel) + } mw := newMCPStreamWriter(stream) stop := make(chan struct{}) var keepalives sync.WaitGroup diff --git a/internal/server/handlers_stream_session_liveness_test.go b/internal/server/handlers_stream_session_liveness_test.go index 09219eeec..dafde284a 100644 --- a/internal/server/handlers_stream_session_liveness_test.go +++ b/internal/server/handlers_stream_session_liveness_test.go @@ -806,7 +806,8 @@ func TestCredentialLiveness_AResolvedPrincipalWithoutAKindFailsClosed(t *testing // is a different animal wearing that branch's coat: somebody accepted a // credential and did not record which, so the predicate has nothing to // re-check and no way to notice. Both MCP accept points were in that state - // when this was written. + // when this was written; both record their kind now (OAuth since + // BUG-3837). // // The counterfactual is TestCredentialLiveness_NoCredentialIsValid above: // same empty kind, no principal, connection KEPT. The two together are what @@ -942,9 +943,8 @@ func TestMCPBearerAuth_PATPathRecordsTheAPITokenKind(t *testing.T) { // PAT in the Authorization header — so it records `api_token` and the // existing door revalidates it. // - // The OAuth branch deliberately records nothing (no liveness door for an - // opaque fosite token yet) and is covered by the fail-closed guard in - // TestCredentialLiveness_AResolvedPrincipalWithoutAKindFailsClosed. + // The OAuth branch records authKindOAuth, whose door re-introspects the + // fosite token (BUG-3837, covered by bug3837_mcp_credential_tick_test.go). srv := testServer(t) user, err := srv.store.CreateUser(models.UserCreate{ diff --git a/internal/server/mcp_stream_credential.go b/internal/server/mcp_stream_credential.go new file mode 100644 index 000000000..e426bfa5c --- /dev/null +++ b/internal/server/mcp_stream_credential.go @@ -0,0 +1,74 @@ +package server + +import ( + "context" + "log/slog" + "net/http" + "time" +) + +// BUG-3837: every long-lived connection ends with its credential (the +// BUG-3007 rule in docs/api.md). The MCP transport's two long-lived doors, +// the GET stream and a subscriptions/listen POST, never re-checked theirs: +// a signed-out session, a revoked PAT or a revoked OAuth grant kept receiving +// notifications until the client hung up. +// +// The tick runs only where admitMCPStream took a slot, at the /events/stream +// cadence, and asks two questions: is the credential that opened the stream +// still live (streamCredentialStillValid, which has an OAuth door since this +// change), and does its user still exist and stay enabled, which admission +// also checked and the credential doors do not. A "no" cancels the stream's +// context: mcp-go leaves on ctx.Done(), and mcpBoundWrites's defers stop the +// keepalive and release the slot. "Could not ask" keeps the stream, as the +// other long-lived connections do. + +// startMCPCredentialTick re-checks r's credential every interval until ctx +// ends, and calls end when it is no longer valid. The interval and the first +// delay are read here, on the handler's goroutine: tests swap both package +// variables, and the tick goroutine is not joined (hunter r1). +func (s *Server) startMCPCredentialTick(ctx context.Context, r *http.Request, end context.CancelFunc) { + interval := watchListRevalInterval + if s.mcpCredentialRevalOverride > 0 { + interval = s.mcpCredentialRevalOverride + } + go s.runMCPCredentialTick(ctx, r, end, interval, revalFirstDelay(interval)) +} + +func (s *Server) runMCPCredentialTick(ctx context.Context, r *http.Request, end context.CancelFunc, interval, first time.Duration) { + timer := time.NewTimer(first) + defer timer.Stop() + for { + select { + case <-ctx.Done(): + return + case <-timer.C: + } + if ctx.Err() != nil { + return // the stream already ended; nothing to close + } + if !s.mcpStreamPrincipalStillValid(r) { + slog.Info("MCP: credential invalidated mid-stream, closing", "method", r.Method) + end() + return + } + timer.Reset(interval) + } +} + +// mcpStreamPrincipalStillValid reports whether the stream may continue: its +// credential is live, and its user still exists and is enabled. +func (s *Server) mcpStreamPrincipalStillValid(r *http.Request) bool { + if !s.streamCredentialStillValid(r) { + return false + } + user := currentUser(r) + if user == nil { + return true + } + fresh, err := s.store.GetUser(user.ID) // (nil, nil) for a deleted user + if err != nil { + slog.Warn("MCP: stream user re-check could not complete", "error", err) + return true + } + return fresh != nil && !fresh.IsDisabled() +} diff --git a/internal/server/middleware_auth.go b/internal/server/middleware_auth.go index 375a8bb8d..1be08d4b8 100644 --- a/internal/server/middleware_auth.go +++ b/internal/server/middleware_auth.go @@ -98,6 +98,7 @@ const ( authKindSessionBearer = "session_bearer" authKindSessionCookie = "session_cookie" authKindAPIToken = "api_token" + authKindOAuth = "oauth_access_token" // MCP OAuth bearer (BUG-3837) ) // authKind returns how this request's principal was established, or "" when no diff --git a/internal/server/middleware_mcp_auth.go b/internal/server/middleware_mcp_auth.go index 23a366d33..50b68cf0c 100644 --- a/internal/server/middleware_mcp_auth.go +++ b/internal/server/middleware_mcp_auth.go @@ -192,12 +192,10 @@ func (s *Server) handleMCPPATAuth(w http.ResponseWriter, r *http.Request, token // round 2). The credential here IS a PAT in the Authorization // header, so the liveness predicate's existing api_token door // re-checks it verbatim — no new machinery, and this accept point - // stops being a hole the predicate cannot see. The OAuth branch - // below deliberately records no kind: its credential is an opaque - // fosite token whose liveness door does not exist yet, so it fails - // closed rather than exempting itself. The MCP transport's GET and - // subscriptions/listen streams ARE long-lived (BUG-3835) but do not yet - // run the re-check; wiring it, with an OAuth liveness door, is BUG-3837. + // stops being a hole the predicate cannot see. The OAuth branch below + // records authKindOAuth, whose door re-introspects the fosite token + // (BUG-3837). The MCP transport's GET and subscriptions/listen streams + // run the re-check on a tick (mcp_stream_credential.go). ctx = context.WithValue(ctx, ctxAuthKind, authKindAPIToken) // Stash the token's scopes so the in-process MCP dispatcher // (internal/mcp/dispatch_http.go) can enforce per-tool scope @@ -373,6 +371,10 @@ func (s *Server) handleMCPOAuthAuth(w http.ResponseWriter, r *http.Request, toke ctx := WithCurrentUser(r.Context(), user) ctx = context.WithValue(ctx, ctxIsAPIToken, true) + // The auth KIND, so a long-lived MCP stream's credential tick re-checks + // this token through its own door (BUG-3837) instead of reaching the + // predicate's no-kind branch, which closes a resolved principal. + ctx = context.WithValue(ctx, ctxAuthKind, authKindOAuth) // Translate fosite's space-separated scope string into the // JSON-array form the existing scope-policy check expects. // tokenScopeAllows recognizes pad:read / pad:write / pad:admin diff --git a/internal/server/server.go b/internal/server/server.go index a2e166c70..9b61f404d 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -645,6 +645,11 @@ type Server struct { // stream they want to drive — a write after setup is not observed. watchRevalTickOverride atomic.Pointer[chan time.Time] + // mcpCredentialRevalOverride is a TEST SEAM (zero in production): the + // interval of a long-lived MCP stream's credential tick (BUG-3837). + // Set before the server serves. + mcpCredentialRevalOverride time.Duration + // In-app tutorials (TASK-3452): the catalog and posters fetched from // getpad.dev on Pad Cloud, and the client tests swap in. tutorials tutorialsCache diff --git a/internal/server/stream_credential_liveness.go b/internal/server/stream_credential_liveness.go index 42a7235d6..f094bcb88 100644 --- a/internal/server/stream_credential_liveness.go +++ b/internal/server/stream_credential_liveness.go @@ -1,16 +1,21 @@ package server import ( + "context" + "errors" "log/slog" "net/http" "strings" + + "github.com/ory/fosite" ) // BUG-3007 — a long-lived connection must stop when the CREDENTIAL that opened // it stops being valid. // -// All three long-lived connections on this server revalidate something on a -// jittered ~60s tick, and none of them asked this question: +// The three long-lived connections this was written for revalidated something +// on a jittered ~60s tick, and none of them asked this question (the MCP +// transport's GET and listen streams, a fourth, ask it since BUG-3837): // // - GET /api/v1/events revalidates the connecting user's MEMBERSHIP, which a // sign-out does not change; @@ -56,7 +61,8 @@ const ( ) // streamCredentialStillValid reports whether a long-lived connection may keep -// running, and is the only thing the three tick sites call. +// running. Every tick site calls it: both SSE endpoints, collab, and the MCP +// streams, which also re-check the user (mcpStreamPrincipalStillValid). // // It KEEPS the connection on `credentialUnknown`, because a database blip is // not a revocation, and ENDS it on `credentialInvalid` and on an unrecognised @@ -107,8 +113,12 @@ func (s *Server) credentialLiveness(r *http.Request) credentialLiveness { return credentialInvalid case authKindAPIToken: - token := strings.TrimSpace(strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ")) - if token == "" { + // extractBearer, the parse both bearer accept points use: the scheme + // is case-insensitive (RFC 6750), so a case-sensitive trim here would + // read "bearer " as the whole header and close a live stream + // every tick (BUG-3837 hunter r1). + token, ok := extractBearer(r.Header.Get("Authorization")) + if !ok { return credentialInvalid } // The LIVENESS door, not the ordinary one: a revalidation tick must not @@ -123,6 +133,10 @@ func (s *Server) credentialLiveness(r *http.Request) credentialLiveness { } return credentialValid + case authKindOAuth: + token, _ := extractBearer(r.Header.Get("Authorization")) + return s.oauthLiveness(r.Context(), token) + case "": // NOTHING TO REVALIDATE, and this branch is explicit rather than a // default so it cannot absorb a future kind by accident. @@ -140,14 +154,10 @@ func (s *Server) credentialLiveness(r *http.Request) credentialLiveness { // authenticated, so a credential was accepted; an accept point that // records the user and not how it established them leaves this // predicate with nothing to re-check and no way to notice. Both MCP - // accept points were in exactly that state when this was written — the - // PAT one is fixed by recording its kind, and the OAuth one has no - // liveness door yet, so it lands here and CLOSES rather than streaming - // past a revoked connected app. No MCP request reaches this predicate - // today: internal/mcp/dispatch_http_routes.go maps nothing under - // /api/v1/events or /api/v1/collab, and the MCP transport's own - // long-lived streams (BUG-3835) do not call it yet (BUG-3837), which - // is why this is a guard rather than an incident. + // accept points were in exactly that state when this was written. Both + // now record their kind (the OAuth one since BUG-3837, with its own + // door), and an accept point added later without one lands here and + // CLOSES rather than streaming past a revoked credential. if currentUser(r) != nil { slog.Warn("stream: a principal was resolved without an auth kind, closing the connection", "path", r.URL.Path) @@ -183,3 +193,35 @@ func (s *Server) sessionLiveness(token string) credentialLiveness { } return credentialValid } + +// oauthLiveness re-introspects an MCP OAuth access token (BUG-3837). One +// fosite introspection answers all three ways the credential can end: a +// revoked token or grant (RevokeOAuthGrant flips every access token in the +// family to active=0, read back as ErrInactiveToken), a deleted row +// (ErrNotFound), and expiry (ErrTokenExpired, from the session's ExpiresAt). +// Introspection writes nothing. +// +// fosite wraps a STORAGE failure in ErrRequestUnauthorized too, so the error +// is classified by its cause, not its outer type: the named causes end the +// stream, and anything else is "could not ask", which keeps it, as the +// session and PAT doors do on a store error. +func (s *Server) oauthLiveness(ctx context.Context, token string) credentialLiveness { + if token == "" || s.oauthServer == nil { + return credentialInvalid + } + _, tokenUse, err := s.oauthServer.IntrospectToken(ctx, token) + switch { + case err == nil: + if tokenUse != fosite.AccessToken { + return credentialInvalid + } + return credentialValid + case errors.Is(err, fosite.ErrInactiveToken), + errors.Is(err, fosite.ErrNotFound), + errors.Is(err, fosite.ErrTokenExpired): + return credentialInvalid + default: + slog.Warn("stream: oauth liveness check could not complete", "error", err) + return credentialUnknown + } +} diff --git a/internal/server/task3365_kick_door_census_test.go b/internal/server/task3365_kick_door_census_test.go index 0df44a662..74c5e3115 100644 --- a/internal/server/task3365_kick_door_census_test.go +++ b/internal/server/task3365_kick_door_census_test.go @@ -38,6 +38,8 @@ import ( // by construction, and that is a review property, not a tested one. // OAuth access-token revocation is out of scope by design: an OAuth // credential cannot open any of the three long-lived routes this protects. +// The MCP transport's streams, which an OAuth token can open, are not kicked +// either: their credential tick ends them within one interval (BUG-3837). // kickWatchedStoreMethods change what a live connection may do or whether // its credential still holds.